AI Agent 可观测性:看清重复调用、成本失控与决策链路

2026-08-04 43 预计阅读时间: 1 分钟
来源: cncf.io AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:10 分钟

传统 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 是否达到最大步数,而不是自然完成。
  • 最终状态是 completeddegradedbudget_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 做了什么,为什么这么做,以及这条决策链付出了多少代价。


相关推荐