Agent Observability：当 AI 执行了100次工具调用，我们如何知道它在干什么？
从 Trace、Span 到生产级 Agent 可观测性：深入理解 OpenTelemetry、跨 Agent 调用链、Token 成本、延迟分析、隐私安全和持续质量评测
一、一个难以回答的问题：Agent 为什么花了3分钟才完成任务？
假设我们开发了一个 AI 行业研究 Agent。
用户要求：
“研究过去24小时 OpenAI、Anthropic 和 Google 的重要动态，核实官方来源，生成一份中文研究报告。”
Agent 开始工作。
它进行了12次模型推理、46次网页搜索、28次页面读取、9次文档解析和5次子 Agent 调用。
最终，它执行了100次工具调用。
三分钟后，报告生成。
用户反馈：
“结果还不错，但是太慢了。”
开发者打开应用日志，只看到：
09:00:00 Task started
09:03:12 Task completed
Status: success问题来了：
为什么需要192秒？
是模型生成太慢？
还是搜索工具超时？
是否有 Agent 重复研究同一条新闻？
是否反复打开了相同网页？
某个子 Agent 是否阻塞了整个任务？
一共消耗多少 Token？
最贵的模型调用发生在哪里？
如果最终报告引用错误，应该从哪一步开始调查？
这些问题，单靠一条“任务成功”的日志根本无法回答。
我们真正需要的是：
能够还原 Agent 整个执行过程的可观测系统。
这就是 Agent Observability。
二、什么是 Agent Observability？
在传统软件工程中，Observability 通常建立在三个基础信号之上：
-
Logs：日志。
-
Metrics：指标。
-
Traces：分布式调用链。
对于普通Web应用，这些工具主要帮助开发者诊断请求耗时、服务错误和系统资源问题。
但 Agent 是一种具有动态决策能力的软件系统。
它可能根据中间结果改变下一步计划。
它可能选择不同工具。
它可能重复搜索、发生 Handoff、等待人工审批、暂停并恢复。
因此，Agent Observability 不仅需要知道“哪个HTTP请求很慢”，还需要回答：
Agent 选择了什么行动、实际执行了什么、产生了什么结果，以及这些行为为什么影响最终质量。
我建议将 Agent Observability 定义为：
通过结构化记录 Agent 的执行轨迹、工具调用、运行状态、资源消耗、业务结果与安全决策，使开发者能够诊断问题、衡量性能、控制成本并持续验证系统质量的一套工程能力。
这里需要明确一个边界：
Observability 记录的是系统可观测到的执行行为，不等于读取模型不可见的内部推理状态。
我们可以记录模型返回的工具调用、结构化决策和最终输出。
但不应该把这些外部记录描述成模型隐藏思维链的完整还原。
三、Logs、Metrics、Traces：为什么三者缺一不可？
1. Logs：发生了什么事件？
例如：
09:00:01 agent.started
09:00:04 tool.search.started
09:00:07 tool.search.failed
09:00:08 tool.search.retry
09:00:11 tool.search.succeededLogs 适合记录离散事件。
例如：
-
任务开始和结束。
-
工具失败。
-
用户审批。
-
沙箱退出。
-
任务重试。
-
安全策略拒绝。
但是单独的日志存在问题。
假设10个用户同时执行任务，日志消息可能交织在一起。
如果没有统一的关联标识，就很难判断哪些日志属于同一个任务。
2. Metrics：系统整体表现怎样？
例如：
agent_task_success_rate = 0.92
agent_p95_latency = 48s
agent_avg_tool_calls = 16
agent_cost_per_success = $0.08Metrics 适合回答趋势问题：
-
最近一小时的失败率是否上升？
-
新版本是否比旧版更慢？
-
哪个工具的超时率最高？
-
每成功任务成本是否增加？
-
P95 延迟是否超过SLO？
但 Metrics 无法直接告诉开发者某一次失败究竟发生在哪一步。
3. Traces：这一次任务是怎样执行的？
Trace 负责将一个任务中的多个操作连接起来。
例如：
Agent Task
│
├── Planning 0.8s
│
├── Research 8.2s
│ ├── Search OpenAI 2.1s
│ ├── Search Anthropic 3.4s
│ └── Search Google 2.7s
│
├── Verification 4.3s
│ ├── Fetch Source A 1.2s
│ ├── Fetch Source B 1.8s
│ └── Validate Claims 1.3s
│
└── Report Generation 5.6s通过 Trace，可以看到一次任务的调用关系、执行顺序、重试、失败和耗时分布。
因此：
Logs告诉我们发生了什么。Metrics告诉我们问题有多严重。Traces帮助我们定位问题发生在哪里。
对于 Agent，三者还应该与业务结果和 Evals 建立关联。
四、理解 Trace 与 Span：Agent 可观测性的基本数据结构
OpenTelemetry 将 Span 定义为一次操作的记录。
Span 通常包含：
-
Trace ID。
-
Span ID。
-
Parent Span ID。
-
Operation Name。
-
开始时间。
-
结束时间。
-
状态。
-
Attributes。
-
Events。
-
Links。
其中：
Trace ID 标识同一条完整调用链。
Span ID 标识其中一次操作。
Parent Span ID 表示父子关系。
例如：
Trace ID: 8a92...
Root Span
└── agent.run
├── llm.plan
├── tool.search
├── tool.fetch
├── llm.analyze
└── llm.summarize所有 Span 使用同一个 Trace ID。
每个子操作通过 Parent Span ID 关联到上一层。
一个简化的 Span 示例
{
"trace_id": "8a92...",
"span_id": "d714...",
"parent_span_id": "f025...",
"name": "execute_tool search_web",
"status": "OK",
"duration_ms": 348,
"attributes": {
"gen_ai.operation.name": "execute_tool",
"gen_ai.tool.name": "search_web",
"app.tool.retry_count": 0
}
}这个结构只包含执行元数据，没有记录网页正文、用户Prompt或工具的完整参数。
这是我推荐的默认方向。
先收集可用于运维的元数据，再按实际需要、授权和数据策略决定是否采集内容。
五、生产级 Agent 应该怎样划分 Span？
一个常见错误是，只在最外层创建一个 Span：
agent.run: 120 seconds这只能证明任务运行了120秒。
几乎没有诊断价值。
另一个极端是，每一行代码都建立一个 Span。
这样又会产生大量噪声。
我建议根据业务语义和执行边界划分。
| 层级 | 建议的操作 | 核心信息 |
|---|---|---|
| Workflow | 完整用户任务 | 任务类别、版本、状态 |
| Agent | Agent运行或子任务 | Agent名称、角色 |
| LLM | 一次模型请求 | 模型、Token、延迟 |
| Tool | 一次工具执行 | 工具名称、状态、耗时 |
| Handoff | Agent之间交接 | 来源角色、目标角色 |
| Retrieval | 检索动作 | 检索类型、命中数 |
| Approval | 人工审批 | 风险类别、结果、等待时间 |
| Sandbox | 受限代码执行 | 执行环境、退出状态 |
| Verification | 结果验证 | 校验类型、成功状态 |
需要注意，不是所有这些名称都是 OpenTelemetry 已经固定的标准 Span 类型。
它们是推荐的业务建模层级。
实际实现时，应将能够对应到 OpenTelemetry GenAI 语义约定的字段进行标准化，其余字段保留清晰命名空间。
例如：
gen_ai.request.model
gen_ai.usage.input_tokens
gen_ai.usage.output_tokens
是 GenAI 语义约定中使用的字段。
而：
app.workflow.version
app.agent.role
app.tool.retry_count
则可以作为应用自定义属性。
不要把自定义属性误认为 OpenTelemetry 官方固定标准。
六、一套生产级 Agent Observability 架构
如果今天要从零设计一个 Agent 系统的可观测性，我会采用下面的结构。
User / API
│
▼
Agent Runtime
│
┌──────────┼──────────┐
▼ ▼ ▼
LLM Tools Subagents
│ │ │
└──────────┼──────────┘
│
▼
OpenTelemetry SDK
│
▼
Local Collector
│
┌────────────┼────────────┐
▼ ▼ ▼
Allowlist Sampling Batching
/ Redaction
│ │ │
└────────────┼────────────┘
│
▼
Telemetry Pipeline
│
┌────────────┼────────────┐
▼ ▼ ▼
Trace Store Metrics Store Logs
│ │ │
└────────────┼────────────┘
│
▼
Observability UI
│
┌──────────┼──────────┐
▼ ▼ ▼
Debugging Cost Reliability
│
▼
Evals这套架构包含几个关键原则。
原则一：Agent不直接负责遥测导出
模型只负责调用工具和完成任务。
真正的 Trace 采集应该由可信 Runtime、SDK或者工具适配层完成。
不能让模型决定：
“这一次危险工具调用不需要记录。”
原则二：重要安全决策独立记录
例如：
-
拒绝越权。
-
人工审批通过或拒绝。
-
沙箱策略触发。
-
敏感操作执行。
这些事件需要可靠审计。
但审计日志和普通性能 Trace 不应完全混为一谈。
因为普通 Trace 可能采样、过期或被丢弃。
高价值安全审计通常需要独立保留策略和更严格的完整性保证。
原则三：可观测性平台不是敏感数据仓库
完整Prompt、工具结果和文件内容可能包含非常敏感的数据。
生产环境应优先选择元数据采集。
需要详细内容时，才针对受控测试环境或经过授权的诊断会话开启有限采集。
原则四：把质量结果与 Trace 关联
可观测性不应该止步于性能。
例如：
某次任务成功生成报告，但引用校验失败。
如果 Eval 系统能够保存对应的 Trace ID，开发者就可以直接定位：
是检索阶段漏掉了来源？
还是汇总阶段丢失了引用？
这才形成 Observability 与 Evals 的闭环。
七、OpenAI：Agents SDK 的原生 Tracing
OpenAI Agents SDK 提供内置的 Tracing 能力。
官方文档描述了对以下操作的记录：
-
Agent运行。
-
模型生成。
-
工具调用。
-
Handoff。
-
Guardrails。
-
自定义Span。
这些记录可以在对应的 Traces 视图中用于调试和分析。
对于复杂工作流，开发者还可以显式创建 Trace，把多个 Agent Run 关联到同一工作流。
OpenAI 官方实践最值得借鉴的三个点
第一，执行轨迹不局限于模型调用。
工具、Handoff、Guardrail和应用自定义阶段，都可以成为观测对象。
第二，可以为不同运行粒度配置追踪。
例如，一个用户任务包含多个连续的 Agent Run。
开发者可以使用同一工作流追踪上下文来组织这些执行记录。
第三，Tracing 可以连接 Evals。
OpenAI 的 Trace Grading 支持对执行轨迹进行结构化评分。
这让开发者不只是看到工具调用数量，还可以检查执行路径是否违反约束、是否出现不必要的重复动作。
一个重要的隐私细节
OpenAI Agents SDK 文档明确提示，某些模型与函数 Span 可能包含输入和输出内容。
Python SDK 的 trace_include_sensitive_data 默认值为 True。
因此，如果业务涉及金融信息、医疗数据、企业内部文件或个人隐私，不应该无检查地开启默认数据采集并直接向外导出。
可以通过运行配置关闭敏感内容记录。
如果需要自行控制导出目的地，还可以使用自定义 Trace Processor 和 Exporter。
但要注意：
添加一个脱敏 Processor，并不意味着此前已经注册的其他 Exporter 就一定不会收到原始数据。
对于严格的数据控制需求，应该在 Trace 创建之前配置唯一可信的导出路径，并保证脱敏失败时停止导出。
另外，某些具有特定数据保留约束的账户或部署模式，可能不支持默认Tracing。实际应以当前产品和合同条件为准。
八、Anthropic：从 Claude Code 遥测到 OpenTelemetry Agent Tracing
Anthropic 在2026年10月14日公开了 Claude Platform 的 OpenTelemetry Agent Tracing 支持。
其目标是让开发者能够把 Claude SDK、Messages API 和 Managed Agents 的执行记录接入已有可观测性平台。
这包括：
-
模型请求。
-
工具调用。
-
托管 Agent 工作流。
-
Token 使用量。
-
执行耗时。
这一变化的重要意义在于：
开发者不一定需要为每家模型供应商建立完全独立的监控系统。
当多个供应商提供标准化 Trace 时，可以通过同一 OpenTelemetry 基础设施汇总。
但还需要注意两个工程问题。
1. SDK Trace 与服务端 Trace 可能重复
Anthropic 文档明确提醒，在某些配置下，SDK与API可能同时记录同一次请求。
如果不区分 Span 来源，可能发生重复计数。
例如，同一个模型调用在客户端和提供方服务端各出现一次。
如果直接把两边的 Token 相加，就会高估实际费用。
2. 流式请求必须正确结束
对于 Streaming 模式，Span 生命周期可能覆盖整个响应流的消费过程。
如果程序没有正常关闭流，Trace 的结束时间或输出记录可能不完整。
因此，模型调用监控不能只在开始请求时打点。
还需要考虑流结束、取消、超时和连接中断。
这是 Agent Runtime 和 Observability 必须协同处理的问题。
九、Google：将 Agent Trace 接入 Cloud Trace
Google Cloud 的 Agent 可观测性实践同样以 OpenTelemetry 为重要基础。
其文档介绍了如何对 Agent Development Kit（ADK）等框架进行遥测采集，并在 Cloud Trace 中分析：
-
模型交互。
-
工具调用。
-
多 Agent 协作。
-
执行耗时。
-
错误。
-
Token使用情况。
这与 OpenAI、Anthropic 的技术方向基本一致。
但 Google 有一个细节非常值得重视：
基础遥测采集与完整Prompt/Response内容采集，是两种不同的隐私级别。
在部分 Agent Runtime 配置中，开启遥测并不自动意味着收集原始Prompt。
如果另外开启内容采集，则可能包含用户输入、模型输出及用户标识。
这意味着我们应该把遥测采集分成不同层级。
Level 1：基本运行指标
只记录状态、耗时、工具名称、Token数量。
Level 2：结构化执行轨迹
增加任务分支、Handoff、失败类型、结果校验状态。
Level 3：受控内容诊断
在明确授权和数据保护条件下，有限收集经过筛选的输入输出样本。
对于个人金融、医疗和企业机密场景，Level 1和Level 2通常更适合作为默认配置。
不应该为了更直观地调试，就无限期采集完整对话和业务内容。
十、跨 Agent 调用链如何追踪？
上一篇讨论 Multi-Agent 时，我们强调了 Supervisor、Handoff、Parallel Workflow。
它们带来一个新问题：
多个 Agent可能运行在不同进程、容器甚至不同服务器上。
怎么知道它们属于同一个任务？
答案是 Context Propagation。
OpenTelemetry 使用 Trace Context 在服务之间传播调用关联信息。
典型场景下，HTTP 请求可以通过 W3C traceparent 等标准头传递 Trace ID 和父 Span 信息。
例如：
User Request
│
▼
Supervisor Trace
│
├──── Research Agent A
│ │
│ └──── Search API
│
├──── Research Agent B
│ │
│ └──── Document Reader
│
└──── Verification Agent
│
└──── Fact Checker不同服务通过统一的追踪上下文，形成可关联的执行链。
Handoff 需要传递什么？
我建议至少包括：
-
Trace Context。
-
可信任务标识。
-
父任务标识。
-
Agent角色。
-
工作流版本。
-
交接事件类型。
但不要把用户的原始邮件、完整Prompt或访问Token塞入Trace元数据。
对于跨服务调用，追踪上下文必须通过受控通道传播。
从不可信外部请求接收到的追踪标识也需要合理的信任与关联策略，不能把它当作身份凭据。
异步任务怎么办？
如果一个 Agent 把任务放入队列，后续由另一个Worker执行，通常会出现时间跨度较长的异步关系。
这时可根据执行语义使用父子 Span 或 Span Links 来表达关联。
重要的是区分：
执行上的父子关系 与 通过队列建立的因果关联。
不要仅靠业务任务ID拼接日志，就把它误称为符合标准的分布式 Trace。
十一、Agent 延迟分析：为什么不能简单相加 Span 耗时？
假设一次研究任务包含三个并行子 Agent：
-
Agent A：10秒。
-
Agent B：15秒。
-
Agent C：20秒。
如果三个 Agent 同时启动，那么整体研究阶段的耗时不应该简单计算为：
10 + 15 + 20 = 45秒。
理想情况下，它更接近20秒。
因为三个分支存在时间重叠。
这也是为什么 Trace Waterfall 比简单日志更有价值。
它可以展示并行执行的时间线。
时间轴 →
0s 10s 20s 30s
Agent A ██████████
Agent B ███████████████
Agent C ████████████████████
Synthesis ███████整体延迟由关键路径决定。
而不是所有 Span 持续时间之和。
推荐关注四类延迟指标
End-to-end Latency
从用户提交任务到最终完成的总耗时。
TTFT / TTFC
Time to First Token：首个生成Token的延迟。
Time to First Chunk：首个流式响应块的延迟。
两者需要根据供应商和流协议明确测量口径，不能一概混用。
Tool Latency
单个工具操作的耗时。
例如搜索、读取网页、数据库查询或文件操作。
Critical Path Latency
决定整个任务结束时间的关键执行链路。
它帮助判断应该优化哪个阶段。
例如一个 Agent 有100次工具调用，但其中80次都在并行执行，真正阻塞结果的可能只是某个耗时30秒的文档读取操作。
因此，工具调用次数本身不等于延迟问题。
十二、Token 成本分析：不能只看总Token数
生产级Agent的成本分析，至少需要分清不同模型、不同Token类型和不同调用来源。
一个基础成本模型为：
[
C =
\sum_m
\left(
\frac{T_{in,m}}{10^6} P_{in,m}
+
\frac{T_{out,m}}{10^6} P_{out,m}
\right)
+
C_{tools}
+
C_{runtime}
]
其中：
-
(T_{in,m})：模型m的输入Token。
-
(T_{out,m})：模型m的输出Token。
-
(P_{in,m})：每百万输入Token价格。
-
(P_{out,m})：每百万输出Token价格。
-
(C_{tools})：付费工具和第三方API费用。
-
(C_{runtime})：沙箱、容器和计算资源成本。
但这仍然只是简化模型。
真实产品可能还有：
-
缓存读取输入Token优惠。
-
缓存写入费用。
-
模型特定的推理Token计费规则。
-
图像、音频等多模态费用。
-
供应商调用附加费用。
-
批处理或不同服务等级价格。
对于有缓存的模型，应该按照供应商公布的账单口径分别计算，而不是把缓存Token简单重复计费。
一个容易造成成本翻倍的错误
假设某次任务中：
Supervisor调用Research Agent。
Research Agent消耗了10,000 Token。
如果 Supervisor 汇总统计已经包含所有子 Agent 消耗，而开发者又把 Research Agent 的Token单独加一次，那么就可能重复计数。
同理，SDK层与服务端模型层可能同时产生遥测。
因此，我建议：
成本统计只从定义明确的实际模型请求记录或供应商用量记录进行归集。
Agent父Span主要用于任务结构，不应直接与所有子Span的Token再次相加。
更值得监控的指标
我认为应该重点看：
\frac{\text{Total Cost}}
{\text{Successful Tasks}}
]
以及：
\frac{\text{Total Cost}}
{\text{Validated Outputs}}
]
例如，Research Agent生成100份报告。
其中90份通过事实核查。
即使模型总成本很低，也应该按90份有效结果分析成本，而不是按100次执行计算。
这会直接连接到第四篇 Agent Evals 的质量评测体系。
十三、动手实现：可运行的 Python OpenTelemetry Agent Demo
下面实现一个最小但结构完整的 Agent Observability 示例。
它不需要调用真实大模型API，也不需要把数据上传到云端。
我们模拟一个研究Agent：
-
规划任务。
-
执行两个并行搜索工具。
-
模拟模型汇总。
-
记录 Token 使用量。
-
计算示例成本。
-
在本地导出结构化Trace。
-
输出汇总指标。
代码使用真实 OpenTelemetry Python SDK，而不是自己伪造 Trace ID。
1. 安装依赖
python -m pip install "opentelemetry-api>=1.25" "opentelemetry-sdk>=1.25"建议 Python 3.10 及以上版本。
正式项目应该锁定经过测试的具体依赖版本。
2. 完整示例
将下列代码保存为 agent_observability.py：
import asyncio
import time
from decimal import Decimal
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
SimpleSpanProcessor,
InMemorySpanExporter,
)
from opentelemetry.trace import SpanKind, Status, StatusCode
# 仅用于演示的假设价格，非实时供应商报价。
INPUT_PRICE_PER_M = Decimal("1.00")
OUTPUT_PRICE_PER_M = Decimal("4.00")
def calculate_cost(input_tokens, output_tokens):
return (
Decimal(input_tokens) * INPUT_PRICE_PER_M
+ Decimal(output_tokens) * OUTPUT_PRICE_PER_M
) / Decimal(1_000_000)
# 生产系统应在应用启动时初始化一次 Provider。
provider = TracerProvider(
resource=Resource.create({
"service.name": "research-agent-demo",
"service.version": "1.0.0",
})
)
# 本地内存导出，不上传任何数据。
exporter = InMemorySpanExporter()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("research-agent-demo")
async def search_tool(tool_name, delay):
with tracer.start_as_current_span(
"execute_tool " + tool_name,
kind=SpanKind.INTERNAL,
attributes={
"gen_ai.operation.name": "execute_tool",
"gen_ai.tool.name": tool_name,
}
) as span:
await asyncio.sleep(delay)
# 演示只记录结果数量，不记录搜索内容。
span.set_attribute("app.tool.result_count", 3)
return {
"tool": tool_name,
"result_count": 3,
}
async def model_call():
with tracer.start_as_current_span(
"chat demo-model",
kind=SpanKind.CLIENT,
attributes={
"gen_ai.operation.name": "chat",
"gen_ai.provider.name": "demo",
"gen_ai.request.model": "demo-model",
}
) as span:
await asyncio.sleep(0.12)
input_tokens = 1200
output_tokens = 400
span.set_attribute(
"gen_ai.usage.input_tokens",
input_tokens
)
span.set_attribute(
"gen_ai.usage.output_tokens",
output_tokens
)
cost = calculate_cost(
input_tokens,
output_tokens
)
# 演示应用自定义价格字段。
span.set_attribute(
"app.cost.estimated_usd",
float(cost)
)
return {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"estimated_cost_usd": cost,
}
async def run_agent():
started = time.perf_counter()
with tracer.start_as_current_span(
"invoke_workflow research",
attributes={
"gen_ai.operation.name": "invoke_workflow",
"app.workflow.version": "v1",
"app.workflow.type": "research",
}
) as root:
try:
with tracer.start_as_current_span(
"invoke_agent researcher",
attributes={
"gen_ai.operation.name": "invoke_agent",
"gen_ai.agent.name": "researcher",
}
):
# 并行执行两个工具。
tools = await asyncio.gather(
search_tool("search_openai", 0.08),
search_tool("search_anthropic", 0.11)
)
generation = await model_call()
# 用结构化结果验证任务完成情况。
success = (
len(tools) == 2
and all(
t["result_count"] > 0
for t in tools
)
)
root.set_attribute(
"app.task.success",
success
)
if not success:
root.set_status(
Status(StatusCode.ERROR)
)
except Exception:
# 不把可能含敏感信息的异常正文写进Trace。
root.set_status(Status(StatusCode.ERROR))
root.set_attribute(
"app.error.category",
"workflow_failure"
)
raise
elapsed = time.perf_counter() - started
return {
"success": success,
"elapsed_seconds": elapsed,
"generation": generation,
}
def print_trace_summary():
spans = exporter.get_finished_spans()
print("\n--- TRACE SUMMARY ---")
for span in sorted(
spans,
key=lambda s: s.start_time
):
elapsed_ms = (
span.end_time - span.start_time
) / 1_000_000
print(
f"{span.name:32} "
f"{elapsed_ms:8.1f} ms"
)
# Token只从真实模型调用Span层归集，
# 不再加上Agent父Span的聚合值。
model_spans = [
span for span in spans
if span.attributes.get(
"gen_ai.operation.name"
) == "chat"
]
input_tokens = sum(
s.attributes.get(
"gen_ai.usage.input_tokens", 0
)
for s in model_spans
)
output_tokens = sum(
s.attributes.get(
"gen_ai.usage.output_tokens", 0
)
for s in model_spans
)
print("Input tokens:", input_tokens)
print("Output tokens:", output_tokens)
print(
"Estimated cost:",
calculate_cost(input_tokens, output_tokens)
)
tool_spans = [
span for span in spans
if span.attributes.get(
"gen_ai.operation.name"
) == "execute_tool"
]
print("Tool calls:", len(tool_spans))
print("Recorded spans:", len(spans))
async def main():
result = await run_agent()
print("Task success:", result["success"])
print(
"End-to-end latency:",
round(result["elapsed_seconds"], 3),
"seconds"
)
print_trace_summary()
provider.force_flush()
if __name__ == "__main__":
asyncio.run(main())3. 运行
python agent_observability.py可能得到类似输出：
Task success: True
End-to-end latency: 0.232 seconds
--- TRACE SUMMARY ---
invoke_workflow research 232.0 ms
invoke_agent researcher 231.8 ms
execute_tool search_openai 80.3 ms
execute_tool search_anthropic 110.5 ms
chat demo-model 120.3 ms
Input tokens: 1200
Output tokens: 400
Estimated cost: 0.0028
Tool calls: 2
Recorded spans: 5以上是按模拟等待时间给出的预期示例，不是生产模型测量结果；实际输出取决于调度和运行环境。
4. 这段代码证明了什么？
首先，整个工作流拥有统一的Trace。
其次，工具调用与模型调用具有父子关系。
第三，并行执行的两个搜索工具会在时间上重叠。
第四，Token只在模型调用层统计，避免重复计算。
第五，成本通过明确的模拟单价计算，而不是把总Token直接当作费用。
第六，没有收集完整Prompt、工具输入或敏感数据。
5. 为什么这是教学版，而不是完整生产实现？
因为它目前只把Span放在本地内存里。
没有连接真正的OpenTelemetry Collector。
没有实现跨进程上下文传播。
没有生产级采样、队列背压或数据保留机制。
也没有自动从真实供应商API读取Token用量。
但它已经具备真实OpenTelemetry的核心机制，可以作为后续生产化改造的起点。
需要注意，代码中的自定义 app.* 字段和部分Agent层级字段，是演示应用自己的语义建模。正式上线应对照选定版本的 GenAI 语义约定进行适配。
十四、从本地 Demo 升级到生产级 OpenTelemetry
当Agent应用部署到真实服务器后，可以让遥测通过 OTLP 协议进入 OpenTelemetry Collector。
基本结构如下：
Agent App
│
▼
OpenTelemetry SDK
│
▼
OTLP Exporter
│
▼
OpenTelemetry Collector
│
├── Memory Limiter
├── Attribute Filtering
├── Sampling
└── Batch Export
│
▼
Observability Backend为什么建议增加Collector？
因为它可以统一管理：
-
数据接收。
-
属性筛选。
-
批处理。
-
采样。
-
导出目的地。
-
部署和运维配置。
但这里要明确一点：
Collector不是敏感数据的万能防线。
如果应用已经将完整银行邮件、密码或API Key写进Span，那么这些数据在到达Collector之前，就已经进入了遥测处理链路。
因此，更好的策略是源头不采集敏感信息。
Collector上的脱敏应当作为第二层保护，而不是第一层。
一个 Collector 配置示意
下面示例用于说明处理管线结构：
receivers:
otlp:
protocols:
grpc:
endpoint: 127.0.0.1:4317
processors:
memory_limiter:
check_interval: 1s
limit_mib: 256
spike_limit_mib: 64
attributes/privacy:
actions:
- key: gen_ai.input.messages
action: delete
- key: gen_ai.output.messages
action: delete
- key: gen_ai.tool.call.arguments
action: delete
- key: gen_ai.tool.call.result
action: delete
batch: {}
exporters:
debug:
verbosity: basic
service:
pipelines:
traces:
receivers: [otlp]
processors:
- memory_limiter
- attributes/privacy
- batch
exporters: [debug]这是一份用于本地验证的配置示意。
真实环境中还需要：
-
固定Collector发行版与版本。
-
验证处理器是否包含在当前发行版中。
-
配置认证与TLS。
-
检查Span Events、日志及其他字段是否携带敏感内容。
-
配置后端导出器。
-
限制Collector访问权限。
-
明确保留期和删除机制。
如果需要严格保证脱敏失败不导出，应该在应用内部就实施白名单式采集，并验证出口数据结构。
另外，采样可以降低遥测成本，但不能保证所有重要安全事件都会被保留。
安全审计不应该完全依赖采样Trace。
十五、如何建立 Agent Observability Dashboard？
一个真正有用的看板，不应该只是显示每秒请求数。
我建议至少包含四个部分。
第一部分：业务质量
-
Task Success Rate。
-
Validated Success Rate。
-
人工接管率。
-
任务中断率。
-
Eval通过率。
第二部分：性能
-
P50/P95/P99任务延迟。
-
模型请求延迟。
-
工具请求延迟。
-
关键路径耗时。
-
重试与等待时间。
第三部分：成本
-
每任务模型费用。
-
每成功任务费用。
-
每模型Token消耗。
-
缓存Token比例。
-
付费工具费用。
-
不同业务类别成本。
第四部分：安全和运行可靠性
-
越权请求拒绝次数。
-
工具审批次数与拒绝比例。
-
沙箱异常退出率。
-
工具错误率。
-
失控循环或调用预算耗尽。
-
任务恢复失败率。
这里有一个重要的产品判断：
不要让一张大盘代替所有分析。
例如，总成功率95%可能看起来不错。
但如果账单解析任务成功率99.9%，外部转账相关任务成功率只有80%，风险显然完全不同。
因此，指标至少应该按任务类型、Agent版本、模型版本、工具和风险等级切分。
对于用户ID、邮箱地址等高基数字段，不应直接作为Metrics标签。
否则会造成成本膨胀，也可能增加隐私暴露。
十六、怎样通过 Trace 定位生产故障？
假设研究Agent在某次执行中出现：
Task duration: 182s
Tool calls: 104
Status: completed如果只看总体指标，我们可能认为模型太慢。
但查看Trace后发现：
Planning 4s
Search 36s
Fetch 22s
Verification 11s
Retry Loop 92s
Report Generation 17s问题真正出在重试循环。
进一步展开：
fetch_page
├── attempt 1 -> timeout
├── attempt 2 -> timeout
├── attempt 3 -> timeout
├── attempt 4 -> timeout
└── attempt 5 -> success这时正确的优化方向就不是更换更强的模型。
而是：
-
增加工具级超时。
-
限制重试次数。
-
引入退避策略。
-
改善失败恢复。
-
对重复URL进行缓存。
-
对不可恢复错误快速失败。
这体现了 Agent Observability 的核心价值：
把模糊的“Agent不够好”，转化为可以定位和修复的工程问题。
另外，不能仅凭工具调用次数判断问题。
比如100次调用中，有80次是合理的并行检索，这可能是正常的。
真正需要评测的是：
这些调用是否必要？
是否遵守预算？
是否导致重复工作？
是否改善最终任务成功率？
十七、隐私安全：Tracing 可能成为新的数据泄露入口
对于能够处理真实业务数据的Agent，Observability 本身也是一个潜在攻击面。
例如一个账单Agent：
从QQ邮箱读取信用卡账单。
调用解析器提取金额。
把结果写入数据库。
如果我们把完整工具输入输出记录到Trace：
tool.fetch_email
arguments: 邮箱地址与邮件ID
result: 完整银行账单正文那么即使业务数据库采用严格的加密与访问控制，账单内容仍然可能通过日志或Tracing后端泄露。
这是一种非常典型的“旁路数据泄露”。
我的建议：默认只记录元数据
对于邮件读取操作，记录：
{
"tool": "fetch_bill_email",
"status": "success",
"latency_ms": 285,
"message_count": 1,
"parser_version": "v3"
}不要默认记录：
{
"email": "用户完整邮箱地址",
"body": "完整账单正文",
"authorization_code": "邮箱授权码"
}五条安全要求
第一，最小化采集。
只记录真正用于调试、监控和评测的字段。
第二，源头脱敏。
不要依赖下游平台自动识别所有敏感信息。
第三，控制访问。
不同开发、运营和安全角色，应拥有不同的Trace查看权限。
第四，设置保留期。
原始Trace、聚合Metrics和安全审计可以采用不同的数据保留策略。
第五，建立删除与审计机制。
用户数据删除请求应考虑遥测副本和备份，而不是只删除业务数据库。
对于受监管行业，还需要根据适用法规、合同和企业政策设计具体的数据保留方案。
十八、Observability 与 Agent Evals 有什么区别？
这是前八篇之间最重要的一条联系。
Observability 回答：
“Agent实际上执行了什么？”
Evals回答：
“Agent执行得是否符合预期？”
两者不能相互替代。
例如：
Trace显示 Agent 调用了正确的搜索工具。
并不能证明它找到的事实一定正确。
Trace显示所有工具调用成功。
也不能证明最终报告没有引用错误。
反过来，Eval只给出一个“失败”结果，也未必能告诉我们错误发生在哪一环。
因此，应该建立：
Production Agent
│
▼
Trace
│
┌──────┴──────┐
▼ ▼
Metrics Eval Runner
│ │
▼ ▼
Monitoring Quality Score
│ │
└──────┬──────┘
▼
Failure Analysis
│
▼
Regression Test
│
▼
Agent Improvement例如：
某类研究任务的引用准确率突然下降。
Eval系统发现失败率上升。
Observability系统进一步定位到某次搜索工具改版后，返回结果缺少来源URL。
开发者修复工具适配层，再运行回归评测。
这才是从生产监控到持续改进的完整闭环。
建议建立的指标关联
| 评测指标 | 需要关联的Trace信息 |
|---|---|
| 任务成功率 | Workflow结果和验证结果 |
| 工具选择正确率 | Tool Call序列 |
| 重复工作率 | 相同工具与目标的重复调用 |
| 引用准确率 | 检索和来源验证记录 |
| 成本效率 | 模型调用Token和费用 |
| 失败恢复能力 | Retry、Checkpoint、恢复事件 |
| 安全约束遵守率 | 工具授权与审批记录 |
对于高风险业务，最终是否成功仍然需要依据可信的业务状态，而不是依据模型的自我声明。
十九、生产级 Agent 可观测性最常见的六个错误
错误一：只记录最终回答
最终回答只能反映Agent输出，无法还原中间执行过程。
错误二：记录所有Prompt和工具内容
这可能让Tracing系统成为一个不受控制的敏感数据仓库。
错误三：把所有Span耗时相加
并行任务存在时间重叠。
整体延迟应分析真实时间线与关键路径。
错误四：重复统计Token
SDK、模型提供方和Agent父Span的用量记录可能重叠。
必须确定唯一计费归集口径。
错误五：只关注平均值
平均延迟无法反映尾部任务的严重卡顿。
应该同时分析P95/P99和任务分布。
错误六：把可观测性当作安全保证
Trace能够帮助发现异常，但不能代替权限校验、沙箱隔离和安全审批。
如果一个Agent已经越权执行，事后记录其行为并不能撤销风险。
Observability 是验证和诊断能力，不是执行权限边界本身。
二十、我建议的落地路线
如果正在开发自己的Agent系统，不需要第一天就部署庞大的可观测性平台。
可以按三个阶段推进。
第一阶段：最小可观测性
适合早期MVP。
实现：
-
每次任务的Trace ID。
-
模型与工具调用Span。
-
错误状态。
-
Token使用量。
-
总延迟。
-
基础成功率。
此时可以先使用本地OpenTelemetry Exporter。
第二阶段：生产监控
适合正式服务。
增加：
-
OpenTelemetry Collector。
-
Trace存储与检索。
-
P95/P99延迟。
-
每成功任务成本。
-
工具失败率。
-
告警。
-
数据脱敏与保留策略。
第三阶段：质量闭环
适合复杂Agent和多Agent系统。
增加：
-
Trace与Eval关联。
-
自动回归测试。
-
跨Agent调用链。
-
关键路径分析。
-
异常执行模式识别。
-
版本比较与灰度发布。
-
安全审计与任务恢复分析。
这里最重要的不是使用哪家商业监控平台。
而是先设计出可靠、可解释、隐私可控的数据模型。
二十一、回顾整个系列：Agent 工程化走到了哪一步？
到目前为止，我们已经讨论了八个核心主题。
第一篇：Harness Engineering。
解决Agent如何组织执行。
第二篇：Context Engineering。
解决Agent如何获得正确的上下文。
第三篇：Memory Engineering。
解决Agent如何跨会话积累和维护知识。
第四篇：Agent Evals。
解决怎样证明Agent真的变好了。
第五篇：Tool Calling与MCP。
解决Agent怎样安全调用外部工具。
第六篇：Runtime与Sandbox。
解决Agent如何在受限环境中执行代码。
第七篇：Multi-Agent。
解决多个Agent如何合理协作。
今天的第八篇：Observability。
解决Agent在真实生产环境中做了什么、为什么失败、成本从哪里产生，以及如何持续改进。
组合起来，就是：
User Task
│
▼
Agent Harness
│
┌─────────┴─────────┐
▼ ▼
Context Memory
│ │
└─────────┬─────────┘
│
▼
LLM
│
▼
Agent Orchestration
│
▼
Tool Gateway
│
▼
Runtime & Sandbox
│
▼
Real-world State
│
┌─────────┴─────────┐
▼ ▼
Observability Evals
│ │
└─────────┬─────────┘
│
▼
Improvement Loop这已经不再是简单的Prompt工程。
它是一套完整的软件工程体系。
结语：真正可靠的Agent，必须能够解释它做过什么
AI Agent的发展正在改变传统软件的执行方式。
过去，应用程序的大部分控制流程由开发者预先编写。
现在，部分控制流程由模型动态决定。
因此，开发者不能只观察输入和输出。
还必须能够了解：
Agent选择了什么工具？
使用了哪个模型？
每一步耗时多久？
哪些操作失败并重试？
哪些子Agent参与协作？
执行是否遵守权限？
任务结果是否经过验证？
Token和成本消耗在哪里？
如果系统突然变慢、变贵或者变得不可靠，能否通过证据定位原因？
这些问题最终都指向 Observability。
但可观测性真正的价值，不是将所有执行细节无限制记录下来。
而是以合适的数据粒度，建立可追溯、可衡量、可诊断的执行证据。
我们不需要看见模型的每一个内部想法，但必须能够追踪它在真实系统中采取的每一个关键行动。
当 Agent 拥有越来越多自主权限时，可观测性将从开发辅助工具，逐渐成为生产级 Agent 的基础设施。
未来优秀的 Agent 系统，不仅要能够完成任务，还必须能够清楚地说明：
它实际做了什么、付出了多少成本，以及我们凭什么相信它完成得正确。
- 点赞
- 收藏
- 关注作者
评论（0）