传统 APM 擅长回答“接口为什么慢”“哪个服务报错”,却很难解释另一类生产问题:为什么 Agent 对同一个问题调用了三次模型、为什么一次任务的成本突然翻倍、又为什么它在几个工具之间来回循环。Agent 的故障往往不是单点异常,而是一条仍然返回成功状态的错误决策链。
要理解这类系统,仅记录延迟和 HTTP 状态码还不够。团队需要把一次 Agent 运行拆成可追踪的步骤,并同时观察提示词、模型调用、工具参数、Token、成本、重试与最终结果。
从请求链路转向决策链路
普通 Web 请求通常有相对稳定的调用图:网关进入服务,服务查询数据库,再返回响应。Agent 的执行图是动态的。模型可能根据上下文选择搜索、数据库或内部 API,也可能发现信息不足后重新规划。
因此,一次 Agent 运行至少应有以下层级:
run:用户任务从进入系统到产生最终答案的完整生命周期。step:规划、模型推理、工具调用、结果校验等独立步骤。model_call:模型、Token、延迟、重试次数和估算成本。tool_call:工具名称、脱敏后的参数、返回状态和结果摘要。decision:模型为什么进入下一步,以及是否触发回退或重新规划。
这些层级最好用同一个 trace_id 串起来。否则,监控系统虽然能显示三次模型请求,却无法判断它们属于三个用户,还是同一个 Agent 在原地打转。
除了链路关系,还要为每次运行记录几个 Agent 特有的指标:
| 指标 | 用途 |
|---|---|
| 每次运行的模型调用数 | 发现循环、过度规划和异常重试 |
| 输入与输出 Token | 定位上下文膨胀和成本变化 |
| 每步估算成本 | 找出最昂贵的模型或步骤 |
| 工具调用成功率 | 区分模型决策错误与外部系统故障 |
| 重复提示词或参数的哈希 | 发现语义近似的重复调用 |
| 任务完成状态 | 避免只用 HTTP 200 判断成功 |
“成功”不等于任务完成
Agent 最危险的异常未必会产生错误码。例如,模型连续三次生成相同查询,每次调用都返回 200 OK,整个任务最终也可能给出答案。对 APM 来说,这是一条延迟偏高但成功的请求;对业务来说,它浪费了成本并暴露出循环控制缺陷。
告警规则因此不能只围绕错误率设计。可以同时检查:
- 单次运行的模型调用数是否超过预算。
- 相同
prompt_hash是否在短时间内重复出现。 - 工具调用参数是否连续不变。
- Token 或成本是否显著偏离该任务类型的基线。
- Agent 是否达到最大步数,而不是自然完成。
- 最终状态是
completed、degraded、budget_exceeded还是loop_detected。
这里要保留原始事件与聚合指标。指标适合告警,Trace 适合还原执行顺序,日志则适合查看脱敏后的输入、输出摘要和错误细节。只保留其中一种,排障时通常会缺少关键上下文。
可以这样实践:记录一条可查询的 Agent Trace
下面是一个只依赖 Python 标准库的最小示例。它不绑定具体模型 SDK,而是模拟模型调用,并把每个 Span 写入 agent-trace.jsonl。接入真实 Agent 时,将 fake_model_call() 替换成实际 SDK 调用,并按供应商价格更新成本计算即可。
import hashlib
import json
import time
import uuid
from contextlib import contextmanager
from pathlib import Path
TRACE_FILE = Path("agent-trace.jsonl")
def stable_hash(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
@contextmanager
def span(trace_id: str, name: str, attributes: dict):
started = time.time()
event = {
"trace_id": trace_id,
"span_id": uuid.uuid4().hex[:16],
"name": name,
"started_at": started,
"attributes": attributes,
}
try:
yield event
event["status"] = "ok"
except Exception as exc:
event["status"] = "error"
event["error_type"] = type(exc).__name__
raise
finally:
event["duration_ms"] = round((time.time() - started) * 1000, 2)
with TRACE_FILE.open("a", encoding="utf-8") as file:
file.write(json.dumps(event, ensure_ascii=False) + "\n")
def fake_model_call(prompt: str) -> dict:
time.sleep(0.05)
return {
"text": "调用 knowledge_search 查询退款规则",
"input_tokens": len(prompt) // 4 + 1,
"output_tokens": 12,
}
def run_agent(question: str) -> None:
trace_id = uuid.uuid4().hex
seen_prompts: dict[str, int] = {}
total_cost_usd = 0.0
for step in range(1, 6):
prompt = f"用户问题:{question}\n请决定下一步。"
prompt_hash = stable_hash(prompt)
seen_prompts[prompt_hash] = seen_prompts.get(prompt_hash, 0) + 1
if seen_prompts[prompt_hash] > 2:
with span(trace_id, "loop_guard", {
"step": step,
"prompt_hash": prompt_hash,
"reason": "same_prompt_seen_more_than_twice",
}):
pass
print(f"停止运行:检测到重复调用,trace_id={trace_id}")
return
with span(trace_id, "model_call", {
"step": step,
"model": "replace-with-your-model",
"prompt_hash": prompt_hash,
}) as current_span:
result = fake_model_call(prompt)
cost = result["input_tokens"] * 0.000001 + result["output_tokens"] * 0.000002
total_cost_usd += cost
current_span["attributes"].update({
"input_tokens": result["input_tokens"],
"output_tokens": result["output_tokens"],
"estimated_cost_usd": round(cost, 8),
})
print(f"运行结束:trace_id={trace_id}, cost=${total_cost_usd:.6f}")
if __name__ == "__main__":
TRACE_FILE.unlink(missing_ok=True)
run_agent("退款需要多长时间?")
print(TRACE_FILE.read_text(encoding="utf-8"))
将代码保存为 agent_trace.py 后可以直接运行:
python agent_trace.py
这个示例故意使用不变的提示词,因此第三次执行前会触发循环保护。生产实现还应考虑“文本不同但语义相同”的情况。可以结合归一化参数、工具名和任务阶段生成指纹;语义相似度检测则需要设置阈值,避免把合理的迭代误判为循环。
数据越详细,治理要求越高
记录提示词和工具参数能显著缩短排障时间,也会带来隐私与安全风险。用户输入可能包含个人信息、访问令牌或内部文档内容。建议默认执行以下控制:
- 在进入遥测管道前脱敏,而不是事后清理。
- 默认记录哈希、长度、Token 数和摘要,仅在受控环境保存全文。
- 对 Trace 设置访问控制、保留期限和审计日志。
- 不记录密钥、认证头和完整数据库结果。
- 对采样策略分级:异常运行全量保留,正常运行按比例采样。
此外,成本估算只是近似值。模型供应商的缓存、批处理、区域和计费规则都可能影响实际账单,应定期用账单数据校准遥测中的价格表。
上线前的检查清单
一套实用的 Agent 可观测方案,不需要一开始就记录所有内容,但应覆盖最短闭环:每次运行有唯一 Trace,模型与工具调用能按顺序还原,Token 和成本可以聚合,循环与预算超限能够主动终止,敏感数据在采集前完成脱敏。
落地时可先选择一个高价值 Agent,建立“每类任务的正常调用次数和成本”基线,再增加重复调用、最大步数和预算告警。可观测性的目标不是制造更多日志,而是让工程师能够回答三个具体问题:Agent 做了什么,为什么这么做,以及这条决策链付出了多少代价。