1. 从 CobbleDB 热存储读取切入:先把 Computer 智能体的 Token 曲线接上 TaoToken
最近 Perplexity 公开了自研键值数据库 CobbleDB 用于网页抓取热存储,讨论焦点之一是热数据批次读取与 Computer 智能体流水线成本。准备批量抓取前,先到 TaoToken 官网领取 Key:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_intro ,并把 Base URL 设为 https://taotoken.net/api。本文不重复热点新闻,而是从 Claude Code 的settings.json、Codex 的config.toml、CC Switch 三件套开始,把冷热分层指标、Token 曲线命令和 DynamoDB 对比数据做成可复现产出。
如果你正在跑 Computer 智能体做网页抓取、正文抽取、摘要、去重和结构化入库,常见卡点不是“模型完全不可用”,而是:热存储批次读取看起来不慢,但每个网页仍要消耗 prompt token;冷数据回表时又触发二次模型调用;并发一上来,重试 token 把曲线拉高。CobbleDB 的启发是把最热的内容留在快速读取层,把历史快照和低频内容下推。TaoToken 的观察角度则是:在冷热分层之外,把每次模型调用的 token 曲线也当成基础设施指标。
本文的目标很具体:你在本地或测试环境执行命令后,应得到三份文件:
hot_cold_metrics.csv:冷热分层指标;token_usage.csv:按请求记录的 token 曲线;dynamodb_vs_local.csv:DynamoDB 公开样本与本地热批次读取的对比数据。
下面的配置和命令都在读者本地终端执行,不连接生产库,也不通过 MCP 或 Agent 直接操作在线数据库。
2. 冷热分层指标:把 CobbleDB 的热存储思路映射到抓取智能体
CobbleDB 被讨论的核心场景是快速网页内容抓取。把它抽象成智能体流水线,可以拆成两层:
- 热层:最近抓取、待摘要、待去重、待向量化、被高频访问的网页正文、指纹、元数据;
- 冷层:已处理完成、访问频率低、只需要归档或偶尔回查的 HTML 快照、旧版本内容、历史任务产物。
如果只盯数据库 P50,你可能会漏掉模型侧成本。真正影响 Computer 智能体批量抓取效率的指标,至少包括以下几类。
| 指标 | 含义 | 建议采集方式 |
|---|---|---|
| 热命中率 | 请求在热层直接命中的比例 | 本地批处理日志 |
| 热批次读取 P50/P95 | 一个批次读取热数据的延迟分布 | 本地压测脚本 |
| 冷回表次数 | 热层未命中后回查冷层的次数 | 任务日志打点 |
| 冷回表 token | 回表后重新调用模型的 token 消耗 | API usage 字段 |
| 每页 prompt tokens | 每个网页送入模型的输入 token | 请求日志 |
| 每页 completion tokens | 每个网页模型输出的 token | 响应 usage |
| 重试 token 占比 | 失败重试产生的 token / 总 token | 请求状态与 usage |
| 缓存节省 token | 命中模板或结果缓存后减少的 token | 与未缓存基线对比 |
| 每千页总 token | 规模化的成本观察单位 | 聚合 CSV |
| 每千页成本 | 按你的实际单价换算 | 本地账单模型 |
下面这段 Python 可以在本地读取两个 JSONL:local_batches.jsonl记录批次读取,token_usage.jsonl记录模型调用。它会输出hot_cold_metrics.csv。
import json import csv import statistics from collections import defaultdict def percentile(values, p): if not values: return 0.0 values = sorted(values) k = (len(values) - 1) * p f = int(k) c = min(f + 1, len(values) - 1) if f == c: return float(values[f]) return values[f] + (values[c] - values[f]) * (k - f) batches = [] with open("local_batches.jsonl", "r", encoding="utf-8") as f: for line in f: if line.strip(): batches.append(json.loads(line)) tokens = [] with open("token_usage.jsonl", "r", encoding="utf-8") as f: for line in f: if line.strip(): tokens.append(json.loads(line)) hot_reads = [b["read_ms"] for b in batches if b.get("tier") == "hot"] cold_reads = [b["read_ms"] for b in batches if b.get("tier") == "cold"] hot_hits = sum(1 for b in batches if b.get("hit") is True) total_requests = max(len(batches), 1) prompt_tokens = [t.get("prompt_tokens", 0) for t in tokens] completion_tokens = [t.get("completion_tokens", 0) for t in tokens] retry_tokens = [t.get("retry_tokens", 0) for t in tokens] total_tokens = [t.get("total_tokens", 0) for t in tokens] pages = max(len(tokens), 1) result = { "hot_hit_rate": round(hot_hits / total_requests, 4), "hot_read_p50_ms": round(percentile(hot_reads, 0.50), 2), "hot_read_p95_ms": round(percentile(hot_reads, 0.95), 2), "cold_read_p50_ms": round(percentile(cold_reads, 0.50), 2), "cold_backfill_count": sum(1 for b in batches if b.get("hit") is False), "prompt_per_page": round(sum(prompt_tokens) / pages, 2), "completion_per_page": round(sum(completion_tokens) / pages, 2), "retry_token_ratio": round(sum(retry_tokens) / max(sum(total_tokens), 1), 4), "total_tokens_per_1k_pages": round(sum(total_tokens) / pages * 1000, 2), } with open("hot_cold_metrics.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=list(result.keys())) writer.writeheader() writer.writerow(result) print(result)你不需要一次接入真实数据库。先跑本地样本,观察热命中率和冷回表次数。若冷回表次数高,优先调整热层保留窗口、批次大小和指纹策略;若每页 prompt token 高,优先压缩模板、只送正文片段和必要元数据,不要把整页 HTML 直接塞进模型。
3. TaoToken 接入配置:Claude Code、Codex、CC Switch 三件套
在 Computer 智能体批量抓取前,建议先到 TaoToken 官网把 Key 和 Base URL 准备好:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_config 。打开后按控制台提示创建 Key,占位符统一写成YOUR_API_KEY。工具配置里的 Base URL 固定为:
https://taotoken.net/api注意:Base URL 不要额外拼 UTM,UTM 只用于官网和 CTA 链接。
3.1 Claude Code 用 settings.json 和 ANTHROPIC_*
Claude Code 建议在settings.json中通过环境变量接入。下面是一个最小示例,模型名请以 TaoToken 控制台实际可用模型为准:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "YOUR_CLAUDE_MODEL" } }如果你已有 team 级或用户级settings.json,不要整文件覆盖,只把env里的三个键合并进去。Claude Code 相关能力、模型映射和权限配置,可以再看官方文档:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_claude_hint 。配置完成后,在项目根目录启动 Claude Code,先执行一个只读任务,例如让它列出仓库结构,确认没有 401 或 404。
3.2 Codex 用 config.toml,不要让 ANTHROPIC_* 混进来
Codex 的配置模型与 Claude Code 不同。不要把ANTHROPIC_*变量塞到 Codex 配置里,否则排障会被带偏。Codex 使用config.toml,示例如下:
model = "YOUR_CODEX_MODEL" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"然后在本地终端设置环境变量:
export TAOTOKEN_API_KEY="YOUR_API_KEY"Windows PowerShell 可以用:
$env:TAOTOKEN_API_KEY="YOUR_API_KEY"Codex 的供应商名可以自定,但base_url和env_key必须与你的实际配置一致。不要把 Claude Code 的ANTHROPIC_AUTH_TOKEN复制到 Codex 的env_key里。
3.3 CC Switch 三件套:供应商、Claude Code Profile、Codex Profile
如果你用 CC Switch 管理多套工具配置,建议建立三件套,而不是把所有变量堆在一个全局配置里:
- 供应商条目:名称写 TaoToken,Base URL 写
https://taotoken.net/api,密钥写YOUR_API_KEY; - Claude Code Profile:引用供应商条目,并生成
ANTHROPIC_BASE_URL、ANTHROPIC_AUTH_TOKEN、ANTHROPIC_MODEL; - Codex Profile:引用供应商条目,并生成
config.toml中的model_provider、base_url、env_key。
切换前先确认当前 Profile 属于哪个工具。常见错误是:在 Codex 终端里加载了 Claude Code Profile,或者反过来。表现通常是变量存在但请求路径不匹配,最后得到 401 或 404。
3.4 接入排障速查
| 现象 | 可能原因 | 处理方式 |
|---|---|---|
| 401 Unauthorized | Key 为空、占位符未替换、Authorization 格式不对 | 检查YOUR_API_KEY是否替换;Claude Code 看ANTHROPIC_AUTH_TOKEN,Codex 看TAOTOKEN_API_KEY |
| 404 Not Found | Base URL 多写或少写路径 | 工具配置用https://taotoken.net/api;OpenAI 兼容请求再按客户端要求拼/v1/chat/completions |
| 429 Too Many Requests | 并发过高、缺少退避 | 降低抓取并发,加入指数退避和幂等键 |
| 请求超时 | 批次过大、单页内容过长 | 减小批次,正文抽取后再送模型 |
| Token 曲线突然抬高 | 模板膨胀、输出格式失控、重试过多 | 检查prompt_per_page、completion_per_page、retry_token_ratio |
4. Token 曲线命令:把每次抓取调用的 usage 落成 CSV
冷热分层解决“数据放在哪里读”,Token 曲线解决“模型调用花了多少”。建议先用 TaoToken 模型对话页验证 Key 和模型是否可用:https://taotoken.net/models/detail/chat?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_model_chat 。验证通过后,再进入批量任务。
下面命令在你本地终端执行。它会调用 OpenAI 兼容的 chat completions,并把原始响应追加到token_raw.jsonl。
export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_API_KEY="YOUR_API_KEY" curl -sS "$TAOTOKEN_BASE_URL/v1/chat/completions" \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "YOUR_MODEL", "messages": [ { "role": "user", "content": "请只输出一行 JSON,字段为 title 和 summary。网页正文:<这里放抽取后的正文片段>" } ], "temperature": 0 }' | tee -a token_raw.jsonl然后提取 usage 字段,生成token_usage.jsonl。如果你的响应结构包含usage.prompt_tokens、usage.completion_tokens、usage.total_tokens,可以用 jq 处理:
jq -c '{ ts: now, model: .model, prompt_tokens: (.usage.prompt_tokens // 0), completion_tokens: (.usage.completion_tokens // 0), total_tokens: (.usage.total_tokens // 0), retry_tokens: 0 }' token_raw.jsonl > token_usage.jsonl批量抓取时,建议在业务代码里直接记录每次请求的usage。重试请求不要丢掉,可以把retry_tokens单独打点,这样后面才能算重试 token 占比。
用 Python 聚合每日 Token 曲线:
import json import csv import statistics from collections import defaultdict from datetime import datetime daily = defaultdict(list) with open("token_usage.jsonl", "r", encoding="utf-8") as f: for line in f: if not line.strip(): continue item = json.loads(line) day = datetime.fromtimestamp(item.get("ts", 0)).strftime("%Y-%m-%d") daily[day].append(item) rows = [] for day, items in sorted(daily.items()): prompts = [i.get("prompt_tokens", 0) for i in items] completions = [i.get("completion_tokens", 0) for i in items] totals = [i.get("total_tokens", 0) for i in items] retries = [i.get("retry_tokens", 0) for i in items] rows.append({ "date": day, "requests": len(items), "prompt_tokens": sum(prompts), "completion_tokens": sum(completions), "total_tokens": sum(totals), "retry_token_ratio": round(sum(retries) / max(sum(totals), 1), 4), "avg_total_per_request": round(statistics.mean(totals), 2) if totals else 0, }) with open("token_curve_daily.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()) if rows else []) writer.writeheader() writer.writerows(rows) print("written token_curve_daily.csv")看曲线时,重点看四个信号:
prompt_tokens随网页数线性增长,说明模板和上下文没有压缩;completion_tokens突然变高,说明输出格式约束失效,需要强制 JSON schema 或字段白名单;retry_token_ratio升高,说明限流、超时或解析失败在重复消耗;avg_total_per_request下不来,但热命中率已经很高,说明瓶颈在模型调用侧,而不是热存储读取侧。
5. DynamoDB 对比数据与本地批次读取压测
CobbleDB 被拿来和 DynamoDB 对比,是因为网页抓取的热存储读取需要更低、更稳定的批次延迟。公开分享中给出的样本参考是:DynamoDB 全分布 P50 约 31.4 ms,CobbleDB 热存储批次读取 P50 约 5.60 ms。这个数字只适合当参考,不适合直接当成你的生产基准。你真正需要的是在自己批次大小、并发数、单页正文长度下复测。
可以先建立一张对比表:
| 指标 | 公开样本参考 | 本地复测字段 | 解读 |
|---|---|---|---|
| 热存储批次读取 P50 | DynamoDB 全分布约 31.4 ms;CobbleDB 热存储约 5.60 ms | hot_read_p50_ms | 仅作参考,必须本地复测 |
| 热存储批次读取 P95 | 需本地测量 | hot_read_p95_ms | 尾延迟决定并发上限 |
| 冷回表次数 | 需本地测量 | cold_backfill_count | 高则扩大热层或合并批次 |
| 每页 prompt tokens | 需本地测量 | prompt_per_page | 高则压缩模板 |
| 每页 completion tokens | 需本地测量 | completion_per_page | 高则限制输出字段 |
| 重试 token 占比 | 需本地测量 | retry_token_ratio | 高则加退避与幂等 |
下面 SQL 只在本地 SQLite 执行,不要连接生产库。
-- 在你本地 SQLite 中执行 CREATE TABLE IF NOT EXISTS page_batches ( batch_id TEXT, page_id TEXT, tier TEXT, read_ms REAL, prompt_tokens INTEGER, completion_tokens INTEGER, retry_tokens INTEGER, created_at TEXT ); -- 示例查询:按热/冷层统计平均读取延迟 SELECT tier, COUNT(*) AS batches, AVG(read_ms) AS avg_ms, MAX(read_ms) AS max_ms FROM page_batches GROUP BY tier;SQLite 没有内置百分位函数,可以用 Python 计算 P50/P95,并输出dynamodb_vs_local.csv:
import sqlite3 import csv def percentile(values, p): values = sorted(values) if not values: return 0.0 k = (len(values) - 1) * p f = int(k) c = min(f + 1, len(values) - 1) if f == c: return float(values[f]) return values[f] + (values[c] - values[f]) * (k - f) conn = sqlite3.connect("local_metrics.db") cur = conn.cursor() cur.execute("SELECT read_ms FROM page_batches WHERE tier = 'hot'") hot = [row[0] for row in cur.fetchall()] cur.execute("SELECT read_ms FROM page_batches WHERE tier = 'cold'") cold = [row[0] for row in cur.fetchall()] cur.execute("SELECT COUNT(*) FROM page_batches WHERE tier = 'cold'") cold_count = cur.fetchone()[0] rows = [ { "item": "hot_read_p50_ms", "public_reference": "DynamoDB full distribution: 31.4; CobbleDB hot batch: 5.60", "local_value": round(percentile(hot, 0.50), 2), }, { "item": "hot_read_p95_ms", "public_reference": "not provided", "local_value": round(percentile(hot, 0.95), 2), }, { "item": "cold_read_p50_ms", "public_reference": "not provided", "local_value": round(percentile(cold, 0.50), 2), }, { "item": "cold_backfill_count", "public_reference": "not provided", "local_value": cold_count, }, ] with open("dynamodb_vs_local.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["item", "public_reference", "local_value"]) writer.writeheader() writer.writerows(rows) conn.close() print("written dynamodb_vs_local.csv")压测时不要一上来跑全量。先用 100 到 1000 个网页样本,比较不同批次大小:8、16、32、64。每次只改一个变量,记录热读取 P50/P95、冷回表次数、每页 prompt token、每页 completion token、重试 token 占比。若热层 P50 已经稳定,但总 token 曲线仍上升,问题大概率在 prompt 模板、去重策略或输出字段约束。
6. 批量抓取前的检查清单与 CTA:模型对话 → Coding Plan → 创建 Key → Claude Code 文档
在 Computer 智能体开始批量抓取前,建议按下面清单过一遍:
- 已到 TaoToken 官网确认 Key、Base URL 和模型可用,官网入口:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_checklist ;
- Claude Code 使用
settings.json与ANTHROPIC_*,Codex 使用config.toml,两者没有混用变量; - CC Switch 三件套已经区分供应商、Claude Code Profile、Codex Profile;
- 热层保留窗口、批次大小、去重指纹策略已经明确;
token_usage.jsonl已开始记录,retry_tokens单独打点;- 本地 SQLite 压测和
dynamodb_vs_local.csv已跑通; - 重试策略有指数退避和幂等键,不会因为 429 或超时无限放大 token;
- 输出格式有字段白名单,避免 completion token 失控。
高转化路径建议按顺序走:
- 先用模型对话页验证 Key、模型和响应格式:https://taotoken.net/models/detail/chat?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_model_chat ;
- 如果 Computer 智能体要持续批量运行,查看 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_coding_plan ;
- 到控制台创建或管理 API Key:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_api_keys ;
- 最后按 Claude Code 文档完成
settings.json配置:https://taotoken.net/doc/ClaudeCodeAnthropic?utm_source=taotoken_aicg_blog_end&utm_content=cobbledb_claude_doc 。
CobbleDB 的热存储读取给了我们一个很好的观察视角:先把热数据读快,再看模型侧成本。对 Computer 智能体而言,冷热分层指标和 Token 曲线不是两套报表,而是同一条流水线的两端。把 TaoToken 的 Base URL 设为https://taotoken.net/api,用YOUR_API_KEY跑通最小请求,再逐步放大批次和并发,你才能同时看到 P50、P95、每页 token 和重试 token 占比的真实变化。