成本统计漏算了子 Agent 用量

月底账单显示 OpenAI API 消耗了 3000 美元，但你的成本仪表盘只记录了 1200 美元——差额的 1800 美元来自子 Agent 和工具调用，但这些调用没有被归因到任何任务 ID。或者在 CrewAI 里，你能看到主 Crew 的 token 使用，但每个 Agent 独立调用 LLM 的消耗被聚合成了一个「misc」桶，无法区分是 Research Agent 还是Writer Agent 在贡献大头。在 AutoGen 的多 Agent 对话里，嵌套的 agent 调用会生成独立的 API 请求，如果每个请求都用不同的 API key 或没有 correlation_id，成本就无法汇总。

常见原因

1. 子 Agent 用独立的 API 客户端实例，没有传递追踪元数据

主 Agent 创建了一个 anthropic.Anthropic() 客户端并注入了 x-trace-id header，但子 Agent 各自 new Anthropic() 了一个新实例，没有继承追踪元数据，导致它们的调用在账单里是独立的「无归因」请求。

怎么判断：在 Anthropic Console 或 OpenAI 账单里，查看有多少调用没有关联的 metadata 标签（如 task_id、workflow_id）。如果超过 20%，就是归因断层。

2. 工具调用内部的 LLM 调用没有被 token 计数器覆盖

某个工具（如「内容摘要」工具）内部调用了 LLM，但工具被注册为「外部工具」，其内部的 API 调用没有被追踪中间件拦截。

怎么判断：检查所有工具的实现，列出哪些工具内部调用了 LLM API，确认这些调用是否有 usage tracking。

3. 使用了多个不同的 API key，成本分散在不同账户

开发环境用个人 API key，生产环境用公司 API key，某些子 Agent 用了 OpenRouter 的代理 key。总成本分散在 3 个不同的地方，没有汇总视图。

怎么判断：列出代码库里所有 OPENAI_API_KEY、ANTHROPIC_API_KEY、OPENROUTER_API_KEY 等环境变量的使用位置，以及它们对应的账单账户。

4. LangChain 的 callback 只注册在顶层 chain 上，子 chain 的调用不在追踪范围

get_openai_callback() 的上下文管理器只追踪在其 with 块内直接调用的 LLM，动态创建的子 chain（如 create_stuff_documents_chain）如果在 with 块外初始化，它的调用不被计入。

怎么判断：对比 with get_openai_callback() as cb 结束后的 cb.total_tokens 与 OpenAI 账单上同一时间段的 token 数。差值越大，漏算越严重。

5. Token 计数只看 API 响应里的 usage 字段，不看实际账单

Anthropic 的 usage.input_tokens 不包含 prompt caching 的 cache_read_input_tokens，只看 input_tokens 会低估缓存命中场景下的实际成本（虽然命中缓存更便宜，但仍然有成本）。

怎么判断：对比代码里统计的 total_cost 与账单里显示的实际扣费，如果后者系统性地高出 10-20%，就是 token 类型计算不全。

6. 批处理 API 调用的成本被延迟记账，实时成本追踪看不到

OpenAI Batch API 的调用在任务完成后才产生账单记录，但实时成本仪表盘假设每次 API 调用立即记账，导致批处理任务的成本在仪表盘里「消失」，直到月底对账才出现。

怎么判断：检查代码里是否使用了 Batch API（client.batches.create(...)），以及成本仪表盘是否有专门处理批处理延迟记账的逻辑。

最短修复路径

Step 1：创建单例 API 客户端工厂，注入全局追踪元数据

import os
from anthropic import Anthropic
from contextvars import ContextVar

# 全局 workflow 上下文
_current_workflow_id: ContextVar[str] = ContextVar("workflow_id", default="unknown")
_current_task_id: ContextVar[str] = ContextVar("task_id", default="unknown")

class TrackedAnthropicClient:
    """带追踪元数据的 Anthropic 客户端包装，所有 Agent 共用同一个工厂。"""
    
    def __init__(self):
        self._client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.total_cost_usd = 0.0
    
    def create_message(self, **kwargs) -> "Message":
        # 注入追踪标签
        kwargs.setdefault("metadata", {})
        # Anthropic 的 user metadata（不影响计费，用于账单归因）
        response = self._client.messages.create(**kwargs)
        
        # 记录 usage（包含所有 token 类型）
        usage = response.usage
        self._record_usage(
            input_tokens=usage.input_tokens,
            output_tokens=usage.output_tokens,
            cache_read=getattr(usage, "cache_read_input_tokens", 0),
            cache_write=getattr(usage, "cache_creation_input_tokens", 0),
            workflow_id=_current_workflow_id.get(),
            task_id=_current_task_id.get()
        )
        return response
    
    def _record_usage(self, input_tokens, output_tokens, cache_read, cache_write, workflow_id, task_id):
        # claude-sonnet-4-6 定价（截至 2026-05）
        PRICE = {
            "input": 3e-6,
            "output": 15e-6,
            "cache_read": 0.3e-6,     # cache read 比 input 便宜 90%
            "cache_write": 3.75e-6    # cache write 比 input 贵 25%
        }
        cost = (input_tokens * PRICE["input"] +
                output_tokens * PRICE["output"] +
                cache_read * PRICE["cache_read"] +
                cache_write * PRICE["cache_write"])
        
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        self.total_cost_usd += cost
        
        # 发到成本仪表盘
        metrics.record("llm_cost", cost, tags={"workflow_id": workflow_id, "task_id": task_id})

# 全局单例，所有 Agent 通过工厂获取，不自己 new
llm_client = TrackedAnthropicClient()

Step 2：在 LangChain 里把 callback 注入到所有子 chain

from langchain_community.callbacks import get_openai_callback
from langchain_core.callbacks import BaseCallbackHandler

class GlobalCostTracker(BaseCallbackHandler):
    """注入到所有 LLM 调用的全局 callback，不只是顶层 chain。"""
    
    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {})
        total_tokens = usage.get("total_tokens", 0)
        metrics.increment("llm_tokens_total", total_tokens)

tracker = GlobalCostTracker()

# 在所有 LLM 实例创建时注入，而不只是在 chain 的 with 块里
llm = ChatOpenAI(
    model="gpt-4o",
    callbacks=[tracker]     # 这个 LLM 实例的所有调用都被追踪
)

Step 3：统一 API key 管理，通过 proxy 集中追踪

# 方案：通过 LiteLLM proxy 统一出口，所有 Agent 调用同一个 proxy
# proxy 记录所有调用的 metadata 和 cost
litellm --config litellm_config.yaml --port 4000

# 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 tag 分组
spend_logs:
  enabled: true
  group_by: ["metadata.workflow_id", "metadata.task_id"]

Step 4：对账脚本：每日对比代码侧统计与账单数据

import anthropic

def daily_cost_reconciliation(date: str) -> dict:
    """每日对账：对比内部统计与 Anthropic 账单。"""
    # 内部统计
    internal_cost = metrics.query(
        "sum(llm_cost)",
        filters={"date": date}
    )
    
    # Anthropic 账单 API（如果有）
    # 或者读取账单 CSV 导出
    
    discrepancy = abs(internal_cost - billing_cost)
    discrepancy_pct = discrepancy / billing_cost if billing_cost > 0 else 0
    
    if discrepancy_pct > 0.05:  # 超过 5% 差异触发告警
        alert(f"成本对账差异 {discrepancy_pct:.1%}：内部统计 ${internal_cost:.2f}，账单 ${billing_cost:.2f}")
    
    return {"internal": internal_cost, "billing": billing_cost, "discrepancy_pct": discrepancy_pct}

预防建议

• 所有 LLM 客户端通过单例工厂创建，工厂自动注入 workflow_id 和 task_id 追踪元数据
• token 统计必须包含所有类型：input、output、cache_read、cache_creation，不能只看 input + output
• 统一 API key，通过 LiteLLM proxy 或 Portkey 等代理集中出口，避免成本分散在多个账户
• 每日自动对账：对比应用侧统计与账单数据，差异超过 5% 时告警
• 为每条 workflow 执行记录预估成本（启动前）和实际成本（完成后），差异大于 2x 时告警
• Batch API 调用记录提交时间，在成本仪表盘里按「预期完成时间」而不是「提交时间」归因
• 每月做 top-N 成本分析：按 workflow_id 排列，找出贡献最多成本的工作流，审查是否可优化

常见问答 (FAQ)

Q: Langfuse 的成本追踪是否能自动覆盖子 Agent 的调用？ A: 如果子 Agent 也用了 @observe 装饰器或显式创建了 Langfuse trace，并且 trace context 被正确传播，Langfuse 会把子 Agent 的 token 成本归入父级 trace。但如果子 Agent 用了独立的 Langfuse 实例或者没有传播 trace context，就会产生孤儿 trace，成本无法归因。

Q: LiteLLM proxy 会增加延迟吗？ A: 在同机房部署时，proxy 添加的额外延迟通常在 2-5ms，对大多数场景可忽略。如果对延迟极度敏感，可以只用 LiteLLM 的 SDK 层（不走 proxy），SDK 层的 callback 也能追踪成本，只是没有中心化的查询界面。

Q: 如何追踪用了 Embedding API 或 Image API 的成本，这些 API 的计价方式不同？ A: 用 LiteLLM 的 completion_cost() 函数，它内置了所有 OpenAI 和 Anthropic API（包括 Embedding、Image、Audio）的计价规则，传入 response 对象自动计算成本。对于不在 LiteLLM 支持列表里的模型，手动在 custom_pricing 字典里注册价格。

Q: 成本追踪代码本身会不会影响 Agent 性能？ A: 同步写入成本记录会增加延迟（尤其是写数据库），推荐异步写入：把成本记录放入内存队列，后台线程批量写入数据库。LiteLLM proxy 和 Langfuse SDK 默认都是异步写入的，不阻塞 LLM 调用路径。

相关阅读

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

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

相关文章

常见问题解决库

Agent 预算在任务中途被吃光

Agent 执行长任务时 token 或费用预算耗尽，任务强制中止且无中间结果可用。本文分析预算失控的根因并给出限额与断点续传方案。

2026/05/25 #AI 编程 #Agents

常见问题解决库

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

Agent 从 checkpoint 恢复时，加载出来的状态字段缺失、类型错误或值异常，导致 workflow 以错误状态继续执行，产生难以追踪的下游错误。本文分析 checkpoint 损坏的根因并给出写入和恢复的防御方案。

2026/05/25 #AI 编程 #Agents

常见问题解决库

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

多 Agent 系统中，Agent 之间的调用关系形成了有向环，导致任务无限循环、token 持续消耗，直到预算耗尽才暴露问题。本文分析循环产生的原因并给出静态检测和运行时防护方案。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 交接时上下文丢失了

多 Agent 流水线中，子 Agent 接手任务后丢失关键背景，导致重复询问或输出偏离预期。本文逐步拆解丢失原因并给出可落地的修复方案。

2026/05/25 #AI 编程 #Agents

常见问题解决库

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

多 Agent 系统中两个或多个 Agent 互相等待对方的输出，整个流水线挂起不前进。本文分析死锁成因并提供超时、优先级和拓扑检测方案。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 输出里夹带了 secret

Agent 在生成代码、文档或日志时，把 API key、密码、token 或其他敏感凭证输出到了对话记录、trace 日志或下游系统里，造成安全风险。本文分析 secret 泄漏的触发路径并给出检测和阻断方案。

2026/05/25 #AI 编程 #Agents