1. 多智能体调用 LLM 时,为什么链路追踪总是断在中间
做多智能体(Multi-Agent)项目的人,大概率都遇到过这种场景:本地跑一个 Planner + Executor + Reviewer 的三段式 Agent,日志里只看到「Planner 完成」「Executor 完成」,但中间某一步 LLM 调用超时、返回空、或者工具参数拼错,你根本不知道是哪一层出的问题。传统 APM 能告诉你 HTTP 200、延迟 800ms,但它回答不了「这次调用里 prompt 注入了什么 context」「模型为什么选了那个工具」「第几轮开始 context 膨胀」。
这就是 AI Agent 可观测性要解决的核心问题:把一次智能体运行拆成可回溯的调用链,让每一次 LLM 请求、每一次工具执行、每一次上下文拼接都有迹可循。适合谁?适合正在本地开发或测试环境里调试多智能体流程的工程师,尤其是用 Python/Node 写 Agent、又不想一上来就搭一整套重型监控栈的人。
我试过在三个 Agent 项目里分别用裸日志、OpenTelemetry、以及统一 Key 通道 + 埋点的方式做追踪,最后发现最省事的路径是:先把所有 LLM 调用收敛到一个统一的 API 通道,再在这个通道上做埋点。原因很直接——多智能体项目里最容易失控的不是代码逻辑,而是 Key 分散在多个 Agent、多个环境变量、多个 SDK 配置里,导致你连「这次请求到底走了哪个模型」都说不清。TaoToken 在这里的角色就是统一 Key/API 通道:所有 Agent 的 LLM 请求都指向同一个 base_url,用同一套 Key,链路追踪的入口就唯一了。
下面我会给出可复制的config.toml和settings.json骨架,把 TaoToken 作为统一通道接入监控埋点,然后走三步验证:发起一次 Agent 调用、查看请求日志、确认异常可回溯。全程面向本地开发与测试环境,不涉及生产库直连。
2. 前置准备:把 TaoToken 作为统一 Key 通道接进来
在动手写埋点之前,先把通道统一。多智能体项目常见的坑是:Planner 用一份 Key,Executor 用另一份,Reviewer 又读环境变量,结果 trace 里三个 Agent 的请求散落在不同 provider 下,根本串不起来。统一到 TaoToken 之后,所有 Agent 共享一个 base_url 和一套 Key,trace_id 才能跨 Agent 传递。
你需要先拿到 Key。访问 API Keys 管理页创建:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite
创建后你会得到形如sk-xxxx的 Key。注意两点:一是本地开发建议单独建一个测试用 Key,方便按 Key 维度过滤日志;二是不要把 Key 硬编码进config.toml提交到仓库,用环境变量注入。
TaoToken 的 API 入口是https://taotoken.net/api,兼容 OpenAI 风格的/v1/chat/completions,所以现有用 openai SDK 的 Agent 代码基本不用改,只改base_url和api_key即可。这一点对可观测性很关键:你不需要为每个 Agent 写不同的适配层,埋点可以统一加在 SDK 客户端初始化处。
如果你还没决定用哪个模型跑 Agent,可以先去模型对话页试一下不同模型在多轮工具调用下的表现:https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
对于长期跑编码类 Agent 的场景,Coding Plan 会更划算,后面第 6 节会提:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
3. 可复制配置:config.toml 与 settings.json 骨架
这一节给两份可直接抄的配置。config.toml用于 Python 侧 Agent(读取通道、模型、埋点开关),settings.json用于 Node 侧或需要 JSON 配置的工具链。两份配置里的base_url都指向 TaoToken,api_key从环境变量读。
先看config.toml:
# config.toml —— 多智能体统一通道与埋点配置 [llm] provider = "taotoken" base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" # 从环境变量读取,不写死 default_model = "gpt-4o-mini" timeout_seconds = 60 max_retries = 2 [observability] enabled = true trace_exporter = "console" # 本地开发先用 console,接 Jaeger 改 otlp service_name = "multi-agent-local" log_prompt_hash = true # 只记 prompt 哈希,不落原文 log_token_usage = true log_latency = true sample_rate = 1.0 # 本地全采样 [agents] planner_model = "gpt-4o-mini" executor_model = "gpt-4o-mini" reviewer_model = "gpt-4o-mini" max_turns = 6再看settings.json,给 Node 侧或统一读取 JSON 的埋点脚本用:
{ "llm": { "provider": "taotoken", "baseUrl": "https://taotoken.net/api", "apiKeyEnv": "TAOTOKEN_API_KEY", "defaultModel": "gpt-4o-mini", "timeoutMs": 60000 }, "observability": { "enabled": true, "exporter": "console", "serviceName": "multi-agent-local", "logPromptHash": true, "logTokenUsage": true, "logLatency": true, "sampleRate": 1.0 }, "agents": { "planner": { "model": "gpt-4o-mini", "maxTurns": 3 }, "executor": { "model": "gpt-4o-mini", "maxTurns": 6 }, "reviewer": { "model": "gpt-4o-mini", "maxTurns": 2 } } }两份配置的字段是对齐的,方便你在 Python 和 Node 混合的 Agent 项目里共用同一套语义。几个参数说明一下:trace_exporter本地先用console,把 span 打到终端,确认链路通了再换成otlp发到 Jaeger 或 Tempo;log_prompt_hash打开后只记录 prompt 的哈希值,避免把长文本和潜在敏感内容写进日志;sample_rate本地设 1.0 全采样,生产再降。
环境变量这样注入:
export TAOTOKEN_API_KEY="sk-你的测试Key"然后写一个最小的埋点初始化脚本,把配置读进来并初始化 tracer:
# observability.py import os import hashlib import time import tomllib from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, ConsoleSpanExporter, ) from opentelemetry.sdk.resources import Resource def load_config(path: str = "config.toml") -> dict: with open(path, "rb") as f: return tomllib.load(f) def init_tracer(cfg: dict): resource = Resource.create({ "service.name": cfg["observability"]["service_name"], }) provider = TracerProvider(resource=resource) if cfg["observability"]["trace_exporter"] == "console": provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) trace.set_tracer_provider(provider) return trace.get_tracer("multi-agent") def prompt_hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]这段代码做了三件事:读配置、按配置初始化 exporter、提供一个 prompt 哈希函数。prompt_hash是后面日志里关联「同一次请求」的关键,因为你不落原文,只落哈希,靠哈希去重和回溯。
4. 三步验证:发起调用、看日志、确认异常可回溯
配置就绪后,用三步验证链路是否真的打通。这三步是递进的:第一步确认请求能发出去,第二步确认埋点有输出,第三步确认异常能被定位。
4.1 第一步:发起一次带 trace 的 Agent 调用
写一个最小 Agent,把 TaoToken 作为通道,并在每次 LLM 调用外层包一个 span:
# agent_demo.py import os import time from openai import OpenAI from observability import load_config, init_tracer, prompt_hash cfg = load_config() tracer = init_tracer(cfg) client = OpenAI( base_url=cfg["llm"]["base_url"], api_key=os.environ[cfg["llm"]["api_key_env"]], ) def call_llm(messages, model, turn): with tracer.start_as_current_span(f"llm.call.turn_{turn}") as span: span.set_attribute("llm.model", model) span.set_attribute("llm.messages_count", len(messages)) span.set_attribute("llm.prompt_hash", prompt_hash(str(messages))) start = time.monotonic() resp = client.chat.completions.create( model=model, messages=messages, temperature=0.7, ) elapsed = time.monotonic() - start usage = resp.usage span.set_attribute("llm.usage.prompt_tokens", usage.prompt_tokens) span.set_attribute("llm.usage.completion_tokens", usage.completion_tokens) span.set_attribute("llm.latency_ms", round(elapsed * 1000, 2)) span.set_attribute("llm.finish_reason", resp.choices[0].finish_reason) return resp.choices[0].message.content def run_agent(user_input: str): with tracer.start_as_current_span("agent.run") as root: root.set_attribute("agent.input_hash", prompt_hash(user_input)) messages = [ {"role": "system", "content": "你是一个数据分析助手,先规划再执行。"}, {"role": "user", "content": user_input}, ] final = None for turn in range(1, cfg["agents"]["max_turns"] + 1): content = call_llm(messages, cfg["agents"]["planner_model"], turn) messages.append({"role": "assistant", "content": content}) if "[DONE]" in content: final = content break messages.append({"role": "user", "content": "继续执行下一步。"}) root.set_attribute("agent.total_turns", turn) return final if __name__ == "__main__": result = run_agent("帮我规划一个三步的数据清洗流程") print(result)运行python agent_demo.py,你会看到 ConsoleSpanExporter 把每个 span 打到终端,包含llm.model、llm.usage.prompt_tokens、llm.latency_ms等属性。这一步成功意味着:请求走了 TaoToken 通道,埋点也生效了。
4.2 第二步:查看请求日志,确认字段齐全
把 span 输出重定向到文件,方便过滤:
python agent_demo.py 2>&1 | tee agent_trace.log然后检查关键字段是否都在:
grep -E "llm.model|llm.usage|llm.latency_ms|llm.prompt_hash" agent_trace.log你应该能看到类似这样的输出(字段名以实际 exporter 为准):
"llm.model": "gpt-4o-mini" "llm.usage.prompt_tokens": 128 "llm.usage.completion_tokens": 64 "llm.latency_ms": 842.31 "llm.prompt_hash": "a1b2c3d4e5f6a7b8"如果llm.usage缺失,说明响应里没有 usage 字段,检查是不是用了流式但没开stream_options;如果llm.latency_ms异常大,先看是不是网络问题,再看 prompt 是不是太长。这一步的核心是确认「每次 LLM 调用都有独立的 span 和 token 记录」,而不是只有一个笼统的 agent.run。
4.3 第三步:制造一次异常,确认可回溯
可观测性的价值在异常时才体现。故意把模型名改成一个不存在的值,或者把 Key 换成错的,再跑一次:
# 临时改 config.toml 里 default_model = "gpt-not-exist"运行后你会看到 span 上出现 error 状态,并且llm.finish_reason或异常信息被记录。此时用prompt_hash去日志里反查:
grep "a1b2c3d4e5f6a7b8" agent_trace.log能定位到具体是哪一轮、哪个 Agent、哪次调用出的问题。这就是「异常可回溯」:不是靠翻全量日志,而是靠 trace_id + prompt_hash 精确定位。如果你把 exporter 换成 OTLP 发到 Jaeger,这一步就是在 Jaeger UI 里按 trace_id 搜索,效果更直观。
5. 本篇常见错排查
5.1 报错401 Unauthorized或invalid api key
最常见的原因是环境变量没生效。检查echo $TAOTOKEN_API_KEY是否有值,以及config.toml里的api_key_env名字是否和实际导出的变量名一致。另一个原因是 Key 复制时带了空格或换行,重新从 API Keys 页面复制一次。注意本地测试 Key 和正式 Key 不要混用,否则日志里按 Key 过滤会乱。
5.2 span 打出来了,但 token 用量全是 0
通常是响应对象里没有usage字段。如果你用了流式调用,需要在请求里加stream_options={"include_usage": True},否则最后一个 chunk 才带 usage,而你可能提前 break 了。非流式调用一般都有 usage,如果没有,检查是不是中间层做了转发丢字段。
5.3 trace 里多个 Agent 的 span 串不起来
根因是每个 Agent 各自初始化了 TracerProvider,导致 trace_id 不共享。正确做法是全局只初始化一次 provider,所有 Agent 从同一个 tracer 取 span。如果你是多进程部署,需要把 trace context 通过消息头传递,本地开发阶段可以先单进程跑通。
5.4config.toml读取报tomllib不存在
tomllib是 Python 3.11 才进标准库的。如果你用 3.10 或更早,装tomli并改成import tomli as tomllib。或者干脆把配置换成settings.json,用json.load读,兼容性更好。
5.5 日志里 prompt 原文泄露
如果你不小心把log_prompt_hash关了又直接记了messages,长 prompt 会写进日志。回到配置把log_prompt_hash = true打开,并且代码里只记哈希不记原文。需要看原文时,用 trace_id 去专门的调试存储查,不要混在常规日志里。
6. 把统一通道用在长期编码 Agent 上
本地验证跑通后,如果你要把这套多智能体流程长期用于编码类任务(比如自动改代码、跑测试、生成 PR 描述),单次调用成本会累积得很快。这时候可以考虑 Coding Plan,它面向长期编码场景,配合统一 Key 通道能让 trace 和成本统计都收敛到一处:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
接入文档在这里,里面有各语言 SDK 的 base_url 配置示例,照着改就能把现有 Agent 迁过来:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
如果你用的是 Claude Code 这类工具做 Agent 开发,Anthropic 兼容通道的配置也在文档里:https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite
最后提醒一句:可观测性不是等出问题才补的。我踩过的坑是,早期为了快,Agent 里直接print日志,结果多轮调用一多,终端刷屏根本找不到哪次是哪次。后来把 trace 下沉到 turn 级别、prompt 只记哈希、token 和延迟都进 span,排查效率才上来。你可以先从config.toml的 console exporter 跑通三步验证,再逐步换成 OTLP 接 Jaeger,链路就稳了。