1. 多 Agent 协作下链路断点排查为什么你的分布式追踪看不到失败节点多 Agent 协作系统跑起来之后最让人头疼的不是单个 Agent 报错而是整条链路跑完了结果不对但你不知道是哪一步开始歪的。我试过在一个三 Agent 协作的代码审查场景里Planner 拆任务、Coder 写代码、Reviewer 审代码三个 Agent 通过消息队列串联。某天开始Reviewer 频繁给出“无法理解上下文”的结论但单独测 Reviewer 又完全正常。这就是典型的链路断点问题故障不在单个节点而在节点之间的上下文传递。传统 APM 只能看到 HTTP 200 和 1.2s 延迟看不到 Planner 传给 Coder 的 task 描述里丢了关键约束也看不到 Coder 传给 Reviewer 的 diff 里少了文件路径。要解决这个问题分布式追踪的 Span 结构必须从“扁平请求”升级为“树状执行轨迹”。一次用户请求进来根 Span 是 Agent Session下面挂载 Memory Recall、ReAct Turn 1、Tool Call、ReAct Turn 2、Final Synthesis 等子 Span。每个子 Span 记录自己的输入输出、Token 消耗、耗时和状态码。关键点在于跨 Agent 调用时Trace ID 必须透传。Planner 生成的任务 ID 要作为 Span attribute 写入Coder 处理时读取同一个 Trace ID 并创建子 SpanReviewer 再挂载到 Coder 的 Span 下面。这样整条链路就是一棵完整的树而不是三个孤立的请求。我踩过的坑是一开始只在每个 Agent 内部埋点Agent 之间通过消息队列通信时没有传递 trace context。结果就是三个独立的 Trace根本串不起来。后来在消息体里强制加入trace_id和parent_span_id字段才把链路打通。链路诊断的核心查询逻辑是给定一个失败的 Trace ID按parent_span_id递归展开所有子 Span找到第一个statuserror或output_tokens0的节点。这个节点就是断点。如果断点节点的input_tokens异常大说明上下文注入出了问题如果duration_ms异常高说明工具调用或模型推理卡住了。对于多 Agent 协作场景还需要额外记录 Agent 之间的“交接协议”是否完整。比如 Planner 输出的 JSON schema 是否被 Coder 正确解析Coder 输出的 diff 是否包含 Reviewer 需要的所有字段。这些可以在 Span attribute 里用handoff_valid: true/false标记诊断时直接过滤。2. TaoToken 前置统一模型接入层与可观测性数据源配置多 Agent 系统里每个 Agent 可能调用不同的模型有的用推理型有的用代码型有的用轻量型。如果每个 Agent 各自直连模型厂商追踪数据就散落在各处Token 成本也无法统一核算。所以第一步是建立一个统一的模型接入层所有 Agent 的 LLM 调用都走同一个入口。TaoToken 在这里的角色是提供统一的 API 入口和用量数据。你可以在官网了解整体能力API 地址是https://taotoken.net/api。所有 Agent 的模型调用都指向这个 Base URL用同一个 API Key 认证。这样带来的好处是追踪系统只需要在一个地方采集 Token 用量成本核算也只需要汇总一个数据源。配置方式很简单以 Python 的 OpenAI SDK 为例设置环境变量export OPENAI_BASE_URLhttps://taotoken.net/api export OPENAI_API_KEYsk-your-key-here然后在代码里正常初始化客户端from openai import OpenAI client OpenAI( base_urlhttps://taotoken.net/api, api_keysk-your-key-here ) response client.chat.completions.create( modelclaude-sonnet-4-20250514, messages[{role: user, content: 分析这段代码的潜在问题}], temperature0.2 )这里的关键是 Model ID 要写对。不同 Agent 用不同模型时Model ID 就是成本核算的维度之一。比如 Planner 用claude-sonnet-4-20250514Coder 用gpt-4oReviewer 用claude-haiku-3-5。每个模型的输入输出单价不同核算时要分开统计。如果你用 Claude Code 做开发辅助可以在 settings 里配置{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-your-key-here, ANTHROPIC_MODEL: claude-sonnet-4-20250514 } }这样 Claude Code 的所有请求也走统一入口用量数据自动汇总。对于 Cline 或 Roo Code 这类插件在 MCP 配置里填 Base URL、API Key 和 Model ID 三件套即可。统一接入层之后可观测性系统就有了稳定的数据源。每次 LLM 调用返回的usage字段包含prompt_tokens、completion_tokens和total_tokens这些数据直接写入对应的 Span。如果响应头里带了x-request-id也可以作为 Span attribute 记录方便和上游日志关联。需要注意的是统一接入层不改变你的业务逻辑只是把模型调用的出口收敛到一个地方。追踪埋点还是在你的 Agent 代码里做只是采集到的 Token 数据更完整、更一致。3. 可复制配置OpenTelemetry 追踪埋点与成本核算脚本这一节给出可以直接复制运行的配置和代码。目标是每次 Agent 执行生成一棵完整的 Trace 树每个 Span 记录耗时、Token 和成本最后导出结构化数据供诊断和核算使用。先安装依赖pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp pydantic然后创建追踪器模块agent_tracer.pyimport time import uuid from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class SpanRecord(BaseModel): span_id: str Field(default_factorylambda: str(uuid.uuid4())[:8]) parent_id: Optional[str] None name: str agent_name: str start_time: float 0.0 end_time: Optional[float] None duration_ms: Optional[float] None input_tokens: int 0 output_tokens: int 0 model_id: str estimated_cost_usd: float 0.0 status: str ok attributes: Dict[str, Any] Field(default_factorydict) class AgentTraceContext: def __init__(self, trace_id: str, tenant_id: str, user_id: str, task_id: str): self.trace_id trace_id self.tenant_id tenant_id self.user_id user_id self.task_id task_id self.spans: List[SpanRecord] [] self._stack: List[SpanRecord] [] def start_span(self, name: str, agent_name: str , attributes: Optional[Dict] None) - SpanRecord: parent_id self._stack[-1].span_id if self._stack else None span SpanRecord( parent_idparent_id, namename, agent_nameagent_name, start_timetime.time(), attributesattributes or {} ) self.spans.append(span) self._stack.append(span) return span def end_span(self, input_tokens: int 0, output_tokens: int 0, model_id: str , input_price: float 2.5, output_price: float 10.0, status: str ok, extra: Optional[Dict] None) - None: if not self._stack: return span self._stack.pop() span.end_time time.time() span.duration_ms round((span.end_time - span.start_time) * 1000.0, 2) span.input_tokens input_tokens span.output_tokens output_tokens span.model_id model_id span.status status cost (input_tokens / 1_000_000 * input_price) (output_tokens / 1_000_000 * output_price) span.estimated_cost_usd round(cost, 6) if extra: span.attributes.update(extra) def export(self) - Dict[str, Any]: total_tokens sum(s.input_tokens s.output_tokens for s in self.spans) total_cost sum(s.estimated_cost_usd for s in self.spans) return { trace_id: self.trace_id, tenant_id: self.tenant_id, user_id: self.user_id, task_id: self.task_id, total_spans: len(self.spans), total_tokens: total_tokens, total_cost_usd: round(total_cost, 4), spans: [s.model_dump() for s in self.spans] }使用方式是在 Agent 执行入口创建 context每个步骤 start/end spanctx AgentTraceContext( trace_idtrace_abc123, tenant_idteam_alpha, user_iduser_001, task_idtask_code_review_42 ) # Planner 阶段 ctx.start_span(planner_reasoning, agent_namePlanner) # ... 调用 LLM ... ctx.end_span(input_tokens1200, output_tokens350, model_idclaude-sonnet-4-20250514, input_price3.0, output_price15.0) # Coder 阶段 ctx.start_span(coder_generation, agent_nameCoder) # ... 调用 LLM 和工具 ... ctx.end_span(input_tokens2800, output_tokens900, model_idgpt-4o, input_price2.5, output_price10.0) # Reviewer 阶段 ctx.start_span(reviewer_check, agent_nameReviewer) # ... 调用 LLM ... ctx.end_span(input_tokens1500, output_tokens200, model_idclaude-haiku-3-5, input_price0.8, output_price4.0) result ctx.export() print(result[total_tokens], result[total_cost_usd])对于跨 Agent 的消息传递在消息体里带上trace_id和parent_span_idmessage { trace_id: ctx.trace_id, parent_span_id: ctx.spans[-1].span_id, task: review the following diff, payload: diff_content }接收方 Agent 用同一个 trace_id 创建新的 context并把 parent_span_id 作为根 Span 的 parent_id。这样整条链路就是一棵完整的树。成本核算脚本可以按 tenant、user、task、model 四个维度聚合from collections import defaultdict def aggregate_cost(traces: list) - dict: by_tenant defaultdict(float) by_model defaultdict(float) by_task defaultdict(float) for t in traces: for s in t[spans]: by_tenant[t[tenant_id]] s[estimated_cost_usd] by_model[s[model_id]] s[estimated_cost_usd] by_task[t[task_id]] s[estimated_cost_usd] return { by_tenant: dict(by_tenant), by_model: dict(by_model), by_task: dict(by_task) }这套配置跑通后你就能回答“哪个部门的 Agent 最烧钱”“哪个模型单价最高”“哪个任务 Token 消耗异常”这些问题。4. 验证请求与成功结果从 Trace 导出到成本报表配置完成后需要验证整条链路是否正常工作。验证分三步单次请求追踪、跨 Agent 链路串联、成本报表生成。第一步跑一个最简单的单 Agent 请求确认 Span 能正常记录。执行上面的示例代码打印ctx.export()的结果。你应该看到类似这样的输出{ trace_id: trace_abc123, total_spans: 3, total_tokens: 6950, total_cost_usd: 0.0234, spans: [ { span_id: a1b2c3d4, parent_id: null, name: planner_reasoning, agent_name: Planner, duration_ms: 1820.5, input_tokens: 1200, output_tokens: 350, model_id: claude-sonnet-4-20250514, estimated_cost_usd: 0.00885, status: ok } ] }关键检查点parent_id为 null 的是根 Span其他 Span 的parent_id应该指向上一层的span_id。duration_ms应该和实际耗时吻合。estimated_cost_usd按单价换算后应该合理。第二步验证跨 Agent 链路。启动两个 AgentPlanner 和 Coder通过消息队列传递 trace context。在 Coder 的日志里打印接收到的trace_id和parent_span_id确认和 Planner 发出的一致。然后导出 Coder 的 Trace检查它的根 Span 的parent_id是否等于 Planner 最后一个 Span 的span_id。如果是说明链路串联成功。第三步生成成本报表。把多次请求的 Trace 数据收集起来跑聚合脚本。你应该得到按租户、按模型、按任务的成本分布。比如{ by_tenant: {team_alpha: 0.0234, team_beta: 0.0567}, by_model: {claude-sonnet-4-20250514: 0.00885, gpt-4o: 0.0145}, by_task: {task_code_review_42: 0.0234} }如果某个租户的成本突然飙升可以下钻到具体 Trace看是哪个 Span 的 Token 消耗异常。常见原因是上下文注入过多导致 input_tokens 暴涨或者某个工具调用返回了超大结果被塞进 Prompt。验证通过后你可以把 Trace 数据导出到 OTLP 兼容的后端比如 Jaeger 或 Grafana Tempo做可视化展示。也可以直接存到数据库用 SQL 做更灵活的查询。对于模型对话的快速验证可以直接在模型对话页面测试不同模型的响应和用量确认 Model ID 和单价配置正确。长期跑 Agent 任务的话Coding Plan 提供了更稳定的配额和成本控制。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错对照接入和追踪过程中最常见的报错集中在认证、网络和响应解析三个环节。下面按真实报错信息逐一排查。401 UnauthorizedAPI Key 无效或未正确传递。检查环境变量OPENAI_API_KEY或ANTHROPIC_API_KEY是否设置值是否以sk-开头。如果用的是 Claude Code检查 settings.json 里的ANTHROPIC_API_KEY字段。另外确认 Base URL 没有多余斜杠正确写法是https://taotoken.net/api不要写成https://taotoken.net/api/。local proxy failed本地代理配置冲突。如果你之前设置过HTTP_PROXY或HTTPS_PROXY环境变量SDK 会尝试走代理导致连接失败。解决方法是清空这些变量unset HTTP_PROXY unset HTTPS_PROXY unset ALL_PROXY然后在代码里显式指定base_url不走系统代理。reading choices 报错通常是响应结构不符合预期。比如你用的是 OpenAI SDK但模型返回的是 Anthropic 格式。检查 Model ID 和 SDK 是否匹配。用 OpenAI SDK 时Model ID 应该是gpt-4o这类用 Anthropic SDK 时Model ID 是claude-sonnet-4-20250514。如果混用就会在解析choices字段时报错。OAuth 相关报错Claude Code 或某些 CLI 工具默认走 OAuth 登录如果你配置了 API Key 但工具还在尝试 OAuth会报 token 无效。解决方法是在 settings 里显式关闭 OAuth只保留 API Key 认证。对于 Claude Code确认ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY都正确设置并且没有残留的 OAuth token 文件。Token 用量为 0追踪脚本里end_span没有传入input_tokens和output_tokens。检查 LLM 调用的响应对象usage字段是否被正确读取。有些 SDK 返回的是response.usage.prompt_tokens有些是response.usage.input_tokens需要按实际 SDK 调整。Span 树断裂跨 Agent 传递时parent_span_id丢失。检查消息体里是否真的带上了这个字段接收方是否用它创建了根 Span。如果接收方重新生成了 trace_id链路就会断成两棵树。成本核算偏差大单价配置错误。不同模型的输入输出单价不同而且可能随时调整。建议把单价配置抽成独立的字典方便统一修改MODEL_PRICING { claude-sonnet-4-20250514: {input: 3.0, output: 15.0}, gpt-4o: {input: 2.5, output: 10.0}, claude-haiku-3-5: {input: 0.8, output: 4.0} }排查时优先看错误信息里的关键词401 查 Keyproxy 查网络choices 查 SDK 匹配OAuth 查认证方式。大部分问题都能在五分钟内定位。6. 从 Trace 到成本看板把可观测性变成日常工具追踪和核算跑通之后下一步是把它变成团队日常用的工具。我的做法是每天定时跑一次聚合脚本把前一天的 Trace 数据汇总成报表推送到团队频道。报表包含三个核心指标总 Token 消耗、总成本、Top 5 高消耗任务。对于异常检测设置简单的阈值规则单个 Trace 的 Token 消耗超过 10000 就告警单个任务的成本超过 0.5 美元就标记。这样能及时发现上下文注入过多或工具返回超大结果的问题。链路诊断的日常用法是用户反馈某个任务结果不对时直接拿 trace_id 查完整链路。按parent_span_id展开树看每个节点的输入输出。通常问题出在某个 Agent 的输入被截断或者工具返回了错误格式的数据导致后续 Agent 理解偏差。成本优化的切入点也在 Trace 里。如果发现某个 Agent 的 input_tokens 远大于 output_tokens说明上下文注入过多可以考虑压缩历史消息或改用更小的模型做预处理。如果某个工具调用的 duration_ms 特别高说明外部接口慢可以考虑加缓存或异步化。把这套体系跑顺之后多 Agent 系统就不再是黑盒。每次执行都有完整的轨迹每个 Token 都有归属每个失败都有据可查。这才是可观测性真正的价值不是事后追责而是让系统在运行中就能被理解。