企业官网建设流程全解析

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 并创建子 Span，Reviewer 再挂载到 Coder 的 Span 下面。这样整条链路就是一棵完整的树，而不是三个孤立的请求。

我踩过的坑是：一开始只在每个 Agent 内部埋点，Agent 之间通过消息队列通信时没有传递 trace context。结果就是三个独立的 Trace，根本串不起来。后来在消息体里强制加入trace_id和parent_span_id字段，才把链路打通。

链路诊断的核心查询逻辑是：给定一个失败的 Trace ID，按parent_span_id递归展开所有子 Span，找到第一个status=error或output_tokens=0的节点。这个节点就是断点。如果断点节点的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_URL="https://taotoken.net/api" export OPENAI_API_KEY="sk-your-key-here"

然后在代码里正常初始化客户端：

from openai import OpenAI client = OpenAI( base_url="https://taotoken.net/api", api_key="sk-your-key-here" ) response = client.chat.completions.create( model="claude-sonnet-4-20250514", messages=[{"role": "user", "content": "分析这段代码的潜在问题"}], temperature=0.2 )

这里的关键是 Model ID 要写对。不同 Agent 用不同模型时，Model ID 就是成本核算的维度之一。比如 Planner 用claude-sonnet-4-20250514，Coder 用gpt-4o，Reviewer 用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.py：

import time import uuid from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class SpanRecord(BaseModel): span_id: str = Field(default_factory=lambda: 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_factory=dict) 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_id=parent_id, name=name, agent_name=agent_name, start_time=time.time(), attributes=attributes 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 span：

ctx = AgentTraceContext( trace_id="trace_abc123", tenant_id="team_alpha", user_id="user_001", task_id="task_code_review_42" ) # Planner 阶段 ctx.start_span("planner_reasoning", agent_name="Planner") # ... 调用 LLM ... ctx.end_span(input_tokens=1200, output_tokens=350, model_id="claude-sonnet-4-20250514", input_price=3.0, output_price=15.0) # Coder 阶段 ctx.start_span("coder_generation", agent_name="Coder") # ... 调用 LLM 和工具 ... ctx.end_span(input_tokens=2800, output_tokens=900, model_id="gpt-4o", input_price=2.5, output_price=10.0) # Reviewer 阶段 ctx.start_span("reviewer_check", agent_name="Reviewer") # ... 调用 LLM ... ctx.end_span(input_tokens=1500, output_tokens=200, model_id="claude-haiku-3-5", input_price=0.8, output_price=4.0) result = ctx.export() print(result["total_tokens"], result["total_cost_usd"])

对于跨 Agent 的消息传递，在消息体里带上trace_id和parent_span_id：

message = { "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 链路。启动两个 Agent，Planner 和 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 Unauthorized：API 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 查 Key，proxy 查网络，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 都有归属，每个失败都有据可查。这才是可观测性真正的价值：不是事后追责，而是让系统在运行中就能被理解。