1. 从一次上千子代理的 Python 重构说起:为什么计量要压在 Key 上
最近 AI 工程圈在讨论一次规模相当夸张的自动化重构:一个 Agent harness 驱动上千个子代理并行处理一个约百万行的 Python 代码库,把清理、拆分、去重、接口收敛这类工作拆成大量原子任务分发下去。Elvis Saravia 转发了这件事并给出两点判断:一是自进化技能能不能沉淀成真正的工程复利,二是这种打法未必能平移到其他 harness,他甚至反过来问——如果用更少的子代理,是不是能更省地完成同样的活。这两个问题都很关键,但作为要动手的人,我更关心第三个问题:这么多子代理跑完,token 账单到底怎么算清楚。
先说结论:子代理数量越多,计量越不该分散在业务代码里,而应该收敛到统一的出口上。原因很朴素——每个子代理最终都要发一次模型请求,只要所有这些请求走同一个 Base URL、带同一个 Key,计量点就是天然唯一的。这也是我把 TaoToken 放在这套流程最前面的原因:先到 TaoToken 官网 拿一个 Key,再把所有子代理客户端的 Base URL 统一设为https://taotoken.net/api,后面无论是 10 个子代理还是 1000 个子代理,账单都从同一个入口出。
本文不复述那条行业新闻,而是把它拆成一份可跟做的工程流程:怎么拿到 Key、怎么把 Hermes Agent 这类 harness 的客户端接上去、怎么把每次子代理调用的usage字段落成本地 JSONL、怎么用一段脚本聚合出「按子代理 / 按文件 / 按任务」三层的 token 账单,以及并发跑起来之后最常见的几类报错怎么排。全程你只需要在本地执行命令,不需要把任何 Agent 指到线上业务库。
2. 拿 Key 与首次握手:确认https://taotoken.net/api真的通
第一步不是在 harness 里改配置,而是先用最笨的 curl 把链路打通。顺序上我建议:访问 TaoToken 官网,进控制台创建 API Key,模型 ID 从控制台的模型列表里挑,不要凭记忆瞎填。
Key 到手之后,先在 shell 里把两个变量立起来,注意不要写进任何会被提交的脚本:
# 不建议写进 .bashrc,子代理批量拉起时会继承到子进程环境 export TAOTOKEN_API_KEY="YOUR_API_KEY" export TAOTOKEN_BASE_URL="https://taotoken.net/api"然后打一次最小的请求,重点是看响应体里有没有usage字段——它是后面所有计量工作的原子数据:
curl -sS "${TAOTOKEN_BASE_URL}/v1/chat/completions" \ -H "Authorization: Bearer ${TAOTOKEN_API_KEY}" \ -H "Content-Type: application/json" \ -d '{ "model": "REPLACE_WITH_MODEL_ID", "messages": [{"role": "user", "content": "reply with the single word: pong"}], "max_tokens": 16 }' | jq '{model, usage, finish_reason: .choices[0].finish_reason}'如果客户端要求 OpenAI 兼容路径,注意 Base URL 和端点路径的拼接关系:Base URL 是用https://taotoken.net/api,具体端点按客户端文档要求补/v1/...。这一步确认无误之后,再往 harness 里塞配置,否则你会在一个上千并发的场景里同时面对「Key 不对」「路径不对」「模型名不对」三类问题,排查成本会翻好几倍。
还有一个容易被忽略的点:先记录你这次请求返回的usage结构长什么样。不同供应商对缓存命中、推理 token 的字段命名并不一致,有的放在prompt_tokens_details.cached_tokens,有的直接给cached_tokens,有的用input_tokens/output_tokens。你的聚合脚本要按实际返回的字段来写,而不是按你记忆里的字段来写。
3. 把 Hermes Agent 这类 harness 接到 TaoToken:Claude Code / Codex / CC Switch 三件套
真正的子代理重构任务,通常不会手写 curl,而是跑在 Claude Code、Codex 这类命令行 Agent 上,由 harness 负责把任务拆给子代理。这三种接入方式的环境变量和配置文件格式完全不同,混用会直接导致请求打到错误的地方。
3.1 Claude Code:用settings.json承载ANTHROPIC_*
Claude Code 读的是settings.json,推荐放在项目级或用户级配置里,避免依赖 shell 里的临时 export:
{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "YOUR_API_KEY", "ANTHROPIC_MODEL": "REPLACE_WITH_MODEL_ID", "ANTHROPIC_SMALL_FAST_MODEL": "REPLACE_WITH_SMALL_MODEL_ID" } }两个实践建议。第一,ANTHROPIC_SMALL_FAST_MODEL一定要显式指定,因为子代理跑重构时会有大量「读文件、判类型、生成摘要」的轻量调用,如果不指定,这些调用可能落到主力模型上,账单会莫名其妙地膨胀。第二,settings.json属于敏感文件,别提交到仓库,本地权限收到 600。
如果你同时跑多个 harness,用 TaoToken 官网 控制台里的 Key 做区分会更省事:重构任务用一个专用 Key,日常问答用另一个,这样后面看账单时能直接按 Key 分账,不用再靠日志猜。
3.2 Codex:用config.toml,不要套ANTHROPIC_*
Codex 走的是完全不同的配置体系。它不认ANTHROPIC_*系列变量,你把 Claude Code 的环境变量复制过去只会得到连接失败。正确姿势是写~/.codex/config.toml,用 provider 段声明一个自定义供应商:
# ~/.codex/config.toml model = "REPLACE_WITH_MODEL_ID" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" # 具体 wire 协议按客户端文档选择,如 responses / chat wire_api = "responses"配合在 shell 里导出TAOTOKEN_API_KEY,Codex 启动时会把 Key 注入到 provider 配置声明的那个环境变量里。切换模型时只改model一行,不要去动base_url——Base URL 一旦被打散到多处,你就再也无法用一个出口统一计量了。
3.3 CC Switch 三件套:Base URL、API Key、模型名
如果你用 CC Switch 这类供应商切换工具管理多套配置,本质上要填的就三样东西,我把它叫「三件套」:
| 字段 | 值 | 说明 |
|---|---|---|
| Base URL | https://taotoken.net/api | 所有供应商配置里的这一项必须一致 |
| API Key | YOUR_API_KEY | 重构任务建议单独建一个 Key |
| 默认模型 | 从控制台模型列表选取 | 主力模型与轻量模型分开配 |
三件套里最容易出错的是第一项:很多人会在切换工具里给不同模型配不同 Base URL,结果请求被分到两三个不同的出口上。一旦出口分散,你在第 4 节做的本地计量就只能覆盖其中一部分,账单永远对不上。所以从接入的第一天起,就把「Base URL 全局唯一」当成一条硬性规则。
4. 子代理并发下的计量:从usage字段到分代理账单
这一节是本文的核心。目标很明确:任务跑完之后,你能回答三个问题——哪个子代理最费 token、哪类文件最费 token、整次重构一共花了多少。
4.1 计量口径设计:三层聚合
不要一上来就统计总量,总量没有优化价值。建议按三层设计:
第一层是请求级,每次模型调用记一条记录,字段包括时间戳、子代理 ID、模型名、输入 token、输出 token、缓存命中 token。第二层是子代理级,把同一个子代理 ID 的所有请求加总,用来看哪个角色最贵。第三层是任务级,按你给子代理分配的文件路径或模块前缀聚合,用来看哪块代码最难啃。
这里有一个关键设计:子代理 ID 必须由 harness 侧传入,而不是从返回体里猜。多数 API 响应里并不包含「这是哪个子代理发的请求」,这个信息只有你自己的调度代码知道。所以每条记录都要带上它。
4.2 采集:把每次调用追加成 JSONL
如果你的 harness 支持自定义 HTTP 层,最简单的做法是在中间加一层薄包装,把响应里的usage落盘。用 curl 演示就是这样:
#!/usr/bin/env bash # meter_call.sh —— 单次调用并落盘一条计量记录 set -euo pipefail SUBAGENT_ID="${1:?usage: meter_call.sh <subagent_id> <payload.json>}" PAYLOAD="${2:?usage: meter_call.sh <subagent_id> <payload.json>}" LOG="subagent_calls.jsonl" resp="$(curl -sS "${TAOTOKEN_BASE_URL}/v1/chat/completions" \ -H "Authorization: Bearer ${TAOTOKEN_API_KEY}" \ -H "Content-Type: application/json" \ -d @"${PAYLOAD}")" echo "${resp}" | jq -c \ --arg sid "${SUBAGENT_ID}" \ --arg ts "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \ '{ts: $ts, subagent_id: $sid, model: .model, usage: .usage}' \ >> "${LOG}"注意这里有个坑:如果响应体不是合法 JSON(比如被网关拦截返回了 HTML),jq会报错,而set -e会让脚本直接退出,导致整批子代理中断。生产里更稳的写法是先判断响应是否可解析,解析失败时把原始响应单独写到一个errors.jsonl,而不是让采集脚本崩掉:
if echo "${resp}" | jq -e . >/dev/null 2>&1; then echo "${resp}" | jq -c --arg sid "${SUBAGENT_ID}" \ '{ts: now | todate, subagent_id: $sid, model: .model, usage: .usage}' >> "${LOG}" else echo "${resp}" | jq -Rs --arg sid "${SUBAGENT_ID}" \ '{subagent_id: $sid, raw: .}' >> errors.jsonl fi4.3 聚合脚本:把 JSONL 变成分代理账单
拿到subagent_calls.jsonl之后,下面这段脚本就能产出按子代理聚合的账单。单价留成常量,你自己从控制台页面抄进来,不要在脚本里硬编码来源不明的价格:
#!/usr/bin/env python3 """meter_subagents.py —— 从 subagent_calls.jsonl 聚合成分代理 token 账单""" import json import sys from collections import defaultdict from pathlib import Path # 单价请以控制台实际计费口径为准,单位:每 100 万 token PRICE_IN = 0.0 PRICE_OUT = 0.0 def iter_records(path: Path): with path.open("r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError: continue def pick_usage(usage: dict) -> tuple[int, int, int]: """兼容不同字段命名,返回 (输入, 输出, 缓存命中)""" u = usage or {} tin = u.get("prompt_tokens", u.get("input_tokens", 0)) or 0 tout = u.get("completion_tokens", u.get("output_tokens", 0)) or 0 details = u.get("prompt_tokens_details") or {} cached = details.get("cached_tokens", u.get("cache_read_input_tokens", 0)) or 0 return int(tin), int(tout), int(cached) def main(log_path: str) -> None: agg = defaultdict(lambda: {"calls": 0, "tin": 0, "tout": 0, "cached": 0}) for rec in iter_records(Path(log_path)): sid = rec.get("subagent_id") or "unknown" tin, tout, cached = pick_usage(rec.get("usage")) row = agg[sid] row["calls"] += 1 row["tin"] += tin row["tout"] += tout row["cached"] += cached total_in = sum(r["tin"] for r in agg.values()) total_out = sum(r["tout"] for r in agg.values()) total_cached = sum(r["cached"] for r in agg.values()) total_calls = sum(r["calls"] for r in agg.values()) cost = total_in / 1_000_000 * PRICE_IN + total_out / 1_000_000 * PRICE_OUT print(f"{'subagent':<28}{'calls':>8}{'in':>12}{'out':>12}{'cached':>12}") for sid, r in sorted(agg.items(), key=lambda kv: -(kv[1]["tin"] + kv[1]["tout"])): print(f"{sid:<28}{r['calls']:>8}{r['tin']:>12}{r['tout']:>12}{r['cached']:>12}") print("-" * 72) print(f"{'TOTAL':<28}{total_calls:>8}{total_in:>12}{total_out:>12}{total_cached:>12}") ratio = total_cached / total_in if total_in else 0.0 print(f"cache_hit_ratio={ratio:.2%} est_cost={cost:.4f}") if __name__ == "__main__": main(sys.argv[1] if len(sys.argv) > 1 else "subagent_calls.jsonl")跑起来就是一行命令:
python3 meter_subagents.py subagent_calls.jsonl输出会是一张按 token 消耗降序排列的表。这张表直接回答了我们最开始的问题:上千个子代理并不是均匀烧钱的,通常前几个「擅长读大文件」的子代理就会吃掉相当比例的成本。
4.4 按文件维度再切一刀
子代理级账单告诉你「谁在花钱」,文件级账单告诉你「花在哪」。如果你的 harness 在 payload 里带了目标文件路径,可以在采集阶段把它一并记下来:
jq -c --arg sid "${SUBAGENT_ID}" --arg file "${TARGET_FILE}" \ '{ts: now | todate, subagent_id: $sid, target_file: $file, model: .model, usage: .usage}' >> subagent_calls.jsonl然后在聚合脚本里把聚合键从subagent_id换成target_file,就能看出哪些模块的输入 token 明显偏高。偏高的原因通常有三类:文件本身太长、子代理反复重读同一段代码、任务边界切得太粗。这三类问题的解法完全不同,前者要拆文件,中者要加缓存,后者要改调度策略——所以这一刀必须切。
5. 一次重构任务的 token 账单模板
把上面的采集和聚合跑通之后,你得到的产物应该长这样。下面这份模板可以直接抄成 CSV,作为每次重构任务的交付物之一:
subagent_role,requests,input_tokens,output_tokens,cached_tokens,notes scan-inventory,1,0,0,0,只做本地扫描不调模型 split-module,42,0,0,0,按依赖图切分子任务 refactor-core,188,0,0,0,核心逻辑重写 refactor-tests,96,0,0,0,同步更新测试 dedup-imports,57,0,0,0,轻量任务,走小模型 verify-and-fix,73,0,0,0,回归失败修复 summary-report,1,0,0,0,汇总产出这份模板有两个刻意的设计。第一,scan-inventory和split-module这类真正「不花钱」的阶段也保留在表里,请求数为 0 时它提醒你:重构任务的成本大头往往不在扫描,而在反复的验证与修复循环。第二,notes列必须写,因为一个季度之后你回看这张表时,唯一能让你想起当时为什么这么切分的就是这一列。
成本估算的公式很简单,但有一个前提必须说清楚:
任务总成本 ≈ (输入 token / 1e6) × 输入单价 + (输出 token / 1e6) × 输出单价缓存命中的部分是否单独计价、按什么折扣计价,以控制台展示的计费口径为准,不要按经验值反推。另外,输出 token 的波动通常比输入 token 大得多——同一类重构任务,模型多说几句解释,输出量就可能翻倍。所以如果你的目标是控制成本,优先盯输出侧。
6. 排障清单:并发跑起来之后最常撞到的四件事
上千个子代理并发时,问题不是「会不会出错」,而是「同时出几种错」。下面四类是实测高频的。
第一类:429 限流。症状是部分子代理请求被拒,采集日志里出现空usage。处理原则是给调度层加指数退避,而不是无限重试。重试次数建议封顶在小个位数,因为一次失败的重试本身就是一条新的 token 消耗记录,无限重试会把排障变成烧钱。
第二类:超时与半截响应。大文件重构任务里,输出被截断是很常见的。判断方法看finish_reason,如果是长度截断,要改的是任务切分粒度而不是超时时间。把超时从 60 秒调到 300 秒,通常只能让你更晚地发现输出还是被截断。
第三类:上下文漂移。同一个子代理连续处理多个文件时,前面文件的结论会污染后面的判断。表现是「明明已经改过的模式,它在下一个文件里又改回去了」。解法是让子代理尽量无状态——一个子代理只负责一个明确边界内的任务,需要跨文件的知识用显式输入传进去,而不是指望它记得。
第四类:计量本身出错。最常见的两种:一是采集脚本因为响应不是合法 JSON 而整批中断,二是不同客户端的usage字段命名不一致,导致你只统计到了一半流量。第一种用第 4.2 节的容错写法解决;第二种的排查方法是把聚合脚本对每个子代理的请求数和 harness 侧的任务派发数做一次对账,两者不一致就说明有流量没被采集到。
7. 复现清单与下一步
如果你打算把这套流程搬到自己项目里,按这个顺序走一遍最省时间:
- 先做一次单请求握手,确认返回体里有
usage,并且结构和你预期的字段对得上。 - 把 Claude Code 或 Codex 的配置按第 3 节改好,Base URL 全局只留一个值。
- 给重构任务单独建一个 Key,让账单天然按 Key 分账。
- 把采集逻辑加到 HTTP 层,先跑 10 个任务验证 JSONL 格式正确,再放大到全量。
- 用聚合脚本产出三层账单,把 CSV 和任务产出一起归档。
需要动手的时候,这几个入口按顺序用就够了:
- 先到 模型对话 手动发一次请求,核对响应里的
usage字段结构; - 如果是要长期跑子代理 harness,看 Coding Plan 更适合这种高频、长周期的调用模式;
- 准备正式开跑前,去 API Keys 把重构任务的专用 Key 建出来,避免和日常 Key 混用;
- Claude Code 的具体接法参考 Claude Code 文档,里面有
settings.json的完整字段说明; - 官网入口在这里:TaoToken。
最后回到那位点评者提出的问题:用更少的子代理能不能更省地完成。这件事没法靠直觉回答,只能靠计量数据回答。当你手里有了分代理、分文件、分任务三层账单,「把 20 个子代理砍到 8 个」这种调整才有可比较的基线——否则你只是在换一种方式猜。