成本统计漏算了子 Agent 用量

LangGraph 流水线跑完，LangSmith 报告本次运行总成本 $2.10。月底 Anthropic 账单却是 $380，差不多是预估的 20 倍。这个缺口几乎总是出在子 Agent 调用上：编排器（orchestrator）派生出一个研究子 Agent，它又派生出一个网页搜索子 Agent，后者还会自己发起 LLM 调用。每一层用的是不同的 API key，或者发出的调用根本没有归因回根运行（root run）。你的成本统计器只看得到编排器直接发出的调用，整棵子 Agent 树是隐形的，它的成本也跟着隐形。

**最快修复：**从 API 响应的 usage 字段读取 token 数（绝不要自己估算），把 output 和 cache token 算进公式，并把同一个根 run ID 透传给每个子 Agent，让所有调用都汇总到它名下。如果不想自己写这套管道，就让所有 Agent 都走同一个 LiteLLM proxy，然后读 spend_logs。详见下文。

先判断你属于哪种情况

症状	可能原因	跳转
账单远高于统计值，且你用了不止一个 API key	有未追踪的 key	原因 1
子 Agent 跑在独立的服务/容器里	没有共享 callback	原因 2
LangSmith trace 树里节点数比 Agent 数少	父子关系断裂	原因 3
统计值固定偏低约 80%（output 占大头的运行）	漏算 output/cache token	原因 4
偏低约 20-40%，非英文或工具密集调用更严重	用了客户端 token 估算	原因 5
流式调用贡献了 0 个 token	流式 usage 被丢弃	原因 6

常见原因

1. 子 Agent 用独立的 API key，没有和父级关联

编排器用的是 API key sk-orch。它派生子 Agent 时，子 Agent 从自己的环境加载了另一个 key sk-subagent。成本统计的 callback 只挂在 sk-orch 的调用上，子 Agent 的成本记到了 sk-subagent 名下，从未汇总进父级运行的报告。

怎么判断：列出整条流水线里用到的所有 API key（grep 一下 ANTHROPIC_API_KEY、OPENAI_API_KEY、OPENROUTER_API_KEY 以及各服务自带的 secret）。只要不止一个，就检查每个 key 的用量是否都汇总进了同一个统计器。任何没被追踪的 key = 隐形成本。

2. 子 Agent 是独立进程或服务，没有共享 callback

编排器通过 HTTP（POST /run-agent）调用子 Agent，而子 Agent 是个独立微服务，自己发起 LLM 调用。编排器的 LangSmith 或 OpenAI callback 只追踪编排器进程内的调用。子 Agent 服务要么没有 callback，要么上报到了另一个 project。

怎么判断：检查是否有子 Agent 跑在独立的进程、Docker 容器或服务里。任何不在统计器同一进程内、又没有跨边界透传 trace 上下文的子 Agent，都是隐形的。

3. LangSmith trace 层级断裂：子运行没挂到根运行下

LangSmith 用 run_id 和 parent_run_id 构建 trace 树。trace 树会聚合整条 trace 的 token 用量和成本，并把每个子运行的明细向上汇总到对应的父级。如果调用子 Agent 时没有传 parent_run_id，它就会生成一个没有父级的根级 trace，其成本永远不会并进主运行的总额。

怎么判断：在 LangSmith 里打开根运行，展开 trace 树。如果看到的节点数比实际 Agent 数少，说明有子运行没被挂上。也留意那些和主运行用了相同 session 或 thread ID、却孤悬在根级的运行。

4. 聚合只算 input token，漏掉 output 和 cache token

你的成本公式是 cost = input_tokens * price_per_input_token，漏掉了 output token 成本、缓存写入（cache write）成本和图像 token。截至 2026 年 6 月，Claude Sonnet 4.6 的 output 单价是 input 的 5 倍（$15 vs $3 / MTok），Opus 4.7 同样是 5 倍（$25 vs $5）。output 占大头的 Agent 运行里，output 可能占到总成本的 80%，只算 input 的公式大概只报出真实数字的五分之一。

怎么判断：把你的公式和厂商定价页逐项对照。如果没有把 output token、cache-write token 和 tool-use token 分开计入，那就是在低估。

5. token 数是估算的，不是从 API 响应读取的

成本统计器用 tiktoken.encode(prompt) 之类的客户端估算器来数 token，但厂商是按自己的 tokenizer 计费，数法不一样。对同一段文本，Claude 的 tokenizer 往往比 tiktoken 数出更多 token，而且在代码、函数调用负载和非英文文本上偏差最大。对于这些输入，厂商数出来的 token 经常比客户端估算高 20-40%。

怎么判断：拿同一次调用，把你估算的 token 数和 API 响应里的 usage 字段对比。如果厂商的 usage.input_tokens 和你的估算差超过 10%，就全面改用 API 返回的数值。

6. 流式响应把 usage 数据丢了

用流式（streaming）时，不同厂商的 usage 处理方式不一样。OpenAI 默认不在流式响应里返回 usage，除非你传 stream_options={"include_usage": True}；忘了传，这些调用就贡献 0 个 token。Anthropic 流式时是会返回 usage 的（input token 在 message_start 事件里送达，output token 累积统计，总量放在 stream.get_final_message().usage 里），但前提是你的代码真的去读了最终消息，而不是在最后一个文本块之后就把 stream 丢掉。

怎么判断：给每个流式响应都记录一下 usage 是否存在。流式调用上缺失或为 0 的 usage 会造成系统性少算。

最短修复路径

Step 1：建一个按 run ID 归集的中心化成本累加器

要追踪全部四类 token，而不只是 input。下面的定价表截至 2026 年 6 月（cache write = 5 分钟 TTL 下 1.25 倍基础 input 单价，cache read = 0.1 倍基础 input 单价）。

from collections import defaultdict
from dataclasses import dataclass
from threading import Lock

# 每 MTok 的美元单价，2026 年 6 月。cache_write 取 5 分钟 TTL 单价（1.25 倍 input）。
MODEL_PRICING = {
    "claude-sonnet-4-6": {"input": 3.0, "output": 15.0, "cache_write": 3.75, "cache_read": 0.30},
    "claude-opus-4-7":   {"input": 5.0, "output": 25.0, "cache_write": 6.25, "cache_read": 0.50},
}

@dataclass
class RunCost:
    input_tokens: int = 0
    output_tokens: int = 0
    cache_write_tokens: int = 0
    cache_read_tokens: int = 0

    def total_cost_usd(self, model: str) -> float:
        p = MODEL_PRICING[model]
        return (
            self.input_tokens / 1_000_000 * p["input"]
            + self.output_tokens / 1_000_000 * p["output"]
            + self.cache_write_tokens / 1_000_000 * p.get("cache_write", 0)
            + self.cache_read_tokens / 1_000_000 * p.get("cache_read", 0)
        )

_costs: dict[str, RunCost] = defaultdict(RunCost)
_lock = Lock()

def record_usage(run_id: str, usage: dict, model: str):
    with _lock:
        c = _costs[run_id]
        c.input_tokens += usage.get("input_tokens", 0)
        c.output_tokens += usage.get("output_tokens", 0)
        c.cache_write_tokens += usage.get("cache_creation_input_tokens", 0)
        c.cache_read_tokens += usage.get("cache_read_input_tokens", 0)

Step 2：把根 run ID 传给子 Agent，并透传到每一次调用

import uuid

# 编排器一侧
root_run_id = str(uuid.uuid4())

def invoke_sub_agent(task: str, parent_run_id: str) -> str:
    resp = sub_agent_client.post(
        "/run",
        json={"task": task},
        headers={"X-Run-Id": parent_run_id},  # 跨边界透传 ID
    )
    return resp.json()["result"]

# 子 Agent 服务一侧
@app.post("/run")
def run_agent(request: Request, body: AgentRequest):
    run_id = request.headers.get("X-Run-Id", str(uuid.uuid4()))
    # 本服务内所有 LLM 调用都记到 run_id 名下
    result = execute(body.task, run_id=run_id)
    return {"result": result}

具体到 LangSmith，要传 parent_run_id（或使用其 trace 上下文透传 header / OpenTelemetry），这样分布式的子运行才能挂到根 trace 上，成本自动向上汇总。

Step 3：永远从 API 响应读 usage，别用客户端估算

# Anthropic
response = client.messages.create(
    model="claude-sonnet-4-6",
    messages=messages,
    max_tokens=1024,
)
usage = {
    "input_tokens": response.usage.input_tokens,
    "output_tokens": response.usage.output_tokens,
    "cache_creation_input_tokens": getattr(response.usage, "cache_creation_input_tokens", 0),
    "cache_read_input_tokens": getattr(response.usage, "cache_read_input_tokens", 0),
}
record_usage(current_run_id, usage, model="claude-sonnet-4-6")

Step 4：从流式响应里抓取 usage

# Anthropic：usage 挂在最终累积出来的消息上
with client.messages.stream(
    model="claude-sonnet-4-6",
    messages=messages,
    max_tokens=1024,
) as stream:
    for _ in stream.text_stream:
        pass  # 边到边渲染 token

    final = stream.get_final_message()  # 阻塞直到流读完
    record_usage(run_id, {
        "input_tokens": final.usage.input_tokens,
        "output_tokens": final.usage.output_tokens,
        "cache_creation_input_tokens": getattr(final.usage, "cache_creation_input_tokens", 0),
        "cache_read_input_tokens": getattr(final.usage, "cache_read_input_tokens", 0),
    }, model="claude-sonnet-4-6")

OpenAI 的对应做法是加上 stream_options={"include_usage": True}，usage 对象随后会出现在最后一个 chunk 里。

Step 5：每周用厂商账单对账内部统计

import logging
logger = logging.getLogger(__name__)

def reconcile_costs(internal_usd: float, invoice_usd: float) -> None:
    discrepancy_pct = abs(internal_usd - invoice_usd) / invoice_usd * 100
    logger.info(
        "Cost reconciliation: internal=%.2f invoice=%.2f discrepancy=%.1f%%",
        internal_usd, invoice_usd, discrepancy_pct,
    )
    if discrepancy_pct > 10:
        alert(f"Cost tracking discrepancy {discrepancy_pct:.1f}% — investigate sub-agent attribution")

每个计费周期跑一次。差异超过 10% 就说明还有没被追踪的调用路径。

可选：用 LiteLLM proxy 省掉管道

如果你不想往每个服务里塞 callback，就让所有 Agent 都指向同一个 LiteLLM proxy。不管调用来自哪个服务，它都会记录成本，你按 tag 查询 spend_logs 即可。

# litellm_config.yaml
model_list:
  - model_name: claude-sonnet
    litellm_params:
      model: anthropic/claude-sonnet-4-6
      api_key: os.environ/ANTHROPIC_API_KEY

general_settings:
  master_key: sk-your-proxy-key
  database_url: postgresql://...   # 存储每次调用的成本记录

每个请求带上 metadata.tags（以及一个 per-agent 的 trace ID），再按 tag 拆分成本。LiteLLM 还提供 max_budget_per_session 和按 tag 设置的预算，这样某个失控的子 Agent 会先触发预算上限，而不是变成一张惊人的账单。

如何确认已修复

1. 完整跑一遍流水线（含所有子 Agent），记下统计器对这个 run ID 的总额。
2. 等调用出现在厂商控制台里（Anthropic Console → Usage，或 OpenAI → Usage），按同一时间窗口过滤。
3. 两个数字应当相差在约 5% 以内。如果统计器仍然偏低，缺失的那一块就指向某个具体原因：未追踪的 key（原因 1）、没做透传的子 Agent 服务（原因 2）、孤儿 LangSmith trace（原因 3），或只算 input 的公式（原因 4）。
4. 确认 trace 树的节点数等于你的 Agent 数，且每个节点的 output token 都不为 0。

预防建议

• 整条流水线和子 Agent 树尽量用同一个 API key；若必须隔离，就把所有 key 汇总进一个统计器（或统一走一个 LiteLLM proxy）。
• 通过 header 或上下文变量把根 run ID 透传给每个子 Agent 调用，每次 LLM 调用都记到它名下。
• token 用量永远从 API 响应的 usage 字段读取，绝不客户端估算。
• 公式里要把 output token、cache-write token、cache-read token 和 input token 一起算。
• 流式响应要显式抓取 usage：Anthropic 读最终消息，OpenAI 设 include_usage。
• 每月用厂商账单对账内部统计；差异持续扩大就意味着冒出了新的未追踪调用路径。
• 给每次运行加一个成本上限，超过预期成本 2 倍时触发告警（而不是硬中断）。
• 把子 Agent 的花费显著地放进仪表盘；只要它隐形，规划时就总会低估。

常见问答 (FAQ)

Q：怎么跨多家厂商（Anthropic + OpenAI + Gemini）统计成本？ A：用一套厂商无关的结构（input token、output token、model、provider）加一张按模型的定价表。litellm 自带跨厂商的成本追踪和统一接口，也内置了定价表，completion_cost(response) 能直接返回美元金额，不用你手动维护单价。厂商一改价就更新这张表。

Q：LangSmith 能聚合分布式子 Agent 的成本吗？ A：能。传 parent_run_id（或跨服务边界透传 trace 上下文 / OpenTelemetry），让每个子运行挂到根上，LangSmith 就会把树里所有运行的成本汇总到根。硬性要求是：无论编排器还是子 Agent，每次 LLM 调用都要带上正确的父级记录。

Q：为什么 cache-read 成本和 input 成本差这么多？ A：prompt caching 有自己的单价。截至 2026 年 6 月，cache read 是基础 input 单价的 0.1 倍，5 分钟的 cache write 是 1.25 倍（1 小时的 write 是 2 倍）。在 Sonnet 4.6 上就是 read $0.30/MTok 对比 input $3。如果你的公式把缓存命中的 input 当成全价 input，按命中率不同会高估或低估，所以要把 cache_creation_input_tokens 和 cache_read_input_tokens 当成两条独立明细来记。

Q：我们的子 Agent 跑在不同团队的基础设施上，怎么归因成本？ A：给每次 LLM 调用打上 team_id 和 pipeline_id 的 metadata 标签（LiteLLM、LangSmith 以及各家原生 SDK 都支持自定义标签）。再做一份按团队和流水线拆分成本的周报。

Q：一个多 Agent 代码评审流水线，每次运行多少钱算合理？ A：评审单个 PR（5-10 个文件、3 轮 Agent）在 Sonnet 4.6 上通常落在 $0.05-$0.30，取决于文件大小和上下文长度。每个 PR 超过 $1.00 就值得排查 token 膨胀、整段上下文重发，或意外的重试循环。

相关阅读

• Agent 预算在任务中途被吃光
• 一个 Agent 触发 rate limit 拖垮整链
• 不稳定 tool 触发 Agent 重试风暴
• Trace 里看不到关键 tool call
• Agent 在步骤之间丢失上下文
• Claude Code 在用量限制处停止

标签: #AI 编程 #Agents #排查

相关文章

常见问题解决库

Agent 预算在任务中途被吃光

Agent 执行长任务时 token 或费用预算耗尽，任务半途中止、半成品代码留在磁盘上。本文教你定位预算去向、抢救中间结果，并从断点续跑。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Checkpoint 恢复出来的状态是损坏的

Agent 从 checkpoint 恢复后状态字段缺失、类型错乱或值异常。本文教你定位损坏、回滚到完好的历史版本，并写出真正抗崩溃的 checkpoint。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 调用图出现循环但没人发现

多 Agent 之间互相交接，形成永不终止的环，因为没有任何东西在检测循环。本文教你抓出这个环、配好框架自带的上限、并把循环彻底限定住。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 交接时上下文丢失了

下游 Agent 重复询问已答过的问题，或输出与前几步的决策相矛盾。本文帮你定位丢失上下文的交接边界，并在一小时内接上结构化、可持久的状态传递。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 编排器互相等待出现死锁

两个 Agent 互相等待对方的输出，整条流水线永远挂起。本文教你找到环、加超时、几分钟内打破死锁，并给出 LangGraph、Temporal、AutoGen 的具体做法。

2026/05/25 #AI 编程 #Agents

常见问题解决库

修复：Agent 把 secret 写进了日志和 git

AI Agent 把真实的 API key、token 或密码写进了日志、trace 或已提交的文件。这里给出最快的止血路径，以及如何脱敏输出、阻止它再次发生。

2026/05/25 #AI 编程 #Agents