Vibe Coding 与 AI Agent 全景知识库

主题

8.12 可观测性与性能评估：给 Agent 装上仪表盘与体检报告 ​

“赛车手没有仪表盘，就看不出转速与油耗，也就不可能调校出冠军圈速——Agent 也一样：没有可观测性，你永远只能靠猜来优化它。”

🕵️ 黑盒 Agent 的三大困惑：为什么必须可观测？ ​

经过 8.1 - 8.10 的逐个攻坚，我们已把 Agent 的每个器官都搓好了；8.11 的会话持久化让它学会了存档，8.13 的综合实战将把所有器官组装成完整的个人助手。而在正式组装之前，我们先给它装上一套"仪表盘"——否则一旦跑起来，它几乎是一个黑盒：

  1. 它到底在干嘛？ —— 任务卡住了，它是在思考、在等工具返回、还是陷入了死循环？你只能干瞪眼；
  2. 花了多少钱？ —— 跑了十分钟，Token 烧了多少？账单上的钱花得值不值？
  3. 跑得多快？ —— 平均一步多久？哪个工具调用最慢？哪里是性能瓶颈？

如果这些问题答不上来，任何“优化”都只是拍脑袋。这正是**可观测性（Observability）**要解决的命题。工业界为此沉淀了两条关键经验，本节我们逐一复刻：

经验来源核心思想我们本节实现
hello-agents 第12章「智能体性能评估」定义评估指标、跑基准测试、用评估框架量化 Agent 质量EvalSuite 评估套件：成功率 / 平均耗时 / 平均 Token
Pi (earendil-works/pi) 的「事件驱动」设计事件联合体 + subscribe 订阅机制，让外部实时感知 Agent 全过程EventBus 事件总线 + AgentEvent 事件联合体 + TokenCostAudit 费用审计

⚡ 事件驱动 vs 事后打印：从“马后炮”到“实时直播” ​

最朴素的可观测方式是事后打印：在 Agent 每个关键节点 print 一行日志。它能工作，但毛病不少：

  • 耦合：日志逻辑硬编码在 Agent 主流程里，加一个“费用统计”就要改 Agent 源码；
  • 无结构：打印出的是一堆字符串，无法被程序化消费（统计、过滤、回放）；
  • 不可旁观：外部模块（Web UI、监控面板）无法在运行中实时订阅到内部状态。

事件驱动（Event-Driven）则彻底换了一种姿势：Agent 不关心谁在看，它只负责把每一次动作“广播”成一条事件；任何关心者都可以通过 subscribe 订阅。这正是 Pi 项目的核心设计——事件联合体 + 订阅发布。

⚡ 事件驱动 vs 事后打印：从“马后炮”到“实时直播”

维度事后打印 (print)事件驱动 (EventBus)
日志与主流程耦合🔴 硬编码在 Agent 里🟢 完全解耦，Agent 只 emit
数据结构化🔴 一串字符串🟢 AgentEvent.to_dict() 结构化字典
实时旁观🔴 只能事后翻终端🟢 运行中即可订阅、可 * 通配监听
扩展新能力🔴 改 Agent 源码🟢 新增一个订阅者即可

💻 源码实现：给 Agent 装上仪表盘与体检报告 ​

完整代码见 code/s12_observability.py，共 4 个核心组件。

1️⃣ AgentEvent：可观测事件联合体 ​

Pi 项目用“事件联合体”描述一切运行时动作。我们用 @dataclass 实现一个轻量版：事件类型 + 工具名 + 内容 + Token + 耗时 + 时间戳，再提供 to_dict() 便于 JSON 化：

python
@dataclass
class AgentEvent:
    """🚀 一条可观测事件（Pi 风格事件联合体）：记录 Agent 运行中的一次瞬时动作"""
    event_type: str      # agent_start / llm_call / tool_call / tool_result / compact / error / finish
    tool_name: str = ""  # 涉及的工具名（非工具事件为空）
    content: str = ""    # 事件附带文本（工具结果 / 回复 / 错误信息）
    tokens: int = 0      # 本次 LLM 调用消耗的总 token 数
    latency_ms: float = 0.0   # 本次动作耗时（毫秒）
    timestamp: float = field(default_factory=time.time)  # 事件发生时间戳

    def to_dict(self) -> Dict[str, Any]:
        """将事件序列化为字典，便于 JSON 展示与持久化"""
        return {"event_type": self.event_type, "tool_name": self.tool_name, "content": self.content,
                "tokens": self.tokens, "latency_ms": round(self.latency_ms, 2), "timestamp": round(self.timestamp, 3)}

event_type 枚举了 Agent 的完整生命周期：任务启动 → LLM 调用 → 工具执行 → 工具结果 → （上下文压缩）→ 错误 → 完成。这 7 种事件足以描述 Agent 的每一个关键时刻。

2️⃣ EventBus：Pi 风格的事件总线（subscribe / emit） ​

事件总线是订阅发布的调度中枢：subscribe 注册监听器，emit 广播事件。我们额外支持 '*' 通配订阅（Pi 风格），让关心“一切”的观察者（比如费用审计、轨迹回放）一次订阅即可收到所有事件；同时用 self.history 保留完整事件历史，方便事后审计与回放：

python
class EventBus:
    """📡 事件总线：让外部模块实时感知 Agent 的完整运行过程"""
    def __init__(self) -> None:
        self._subscribers: Dict[str, List[Callable[[AgentEvent], None]]] = {}
        self.history: List[AgentEvent] = []   # 📼 完整事件历史（审计 / 回放用）

    def subscribe(self, event_type: str, callback: Callable[[AgentEvent], None]) -> None:
        """订阅某类事件；传 '*' 表示通配订阅所有事件（Pi 风格）"""
        self._subscribers.setdefault(event_type, []).append(callback)

    def emit(self, event: AgentEvent) -> None:
        """广播一条事件：分发给精确匹配者 + 通配订阅者，并写入历史"""
        self.history.append(event)
        for cb in list(self._subscribers.get(event.event_type, [])) + self._subscribers.get("*", []):
            cb(event)

💡 设计亮点：Agent 主流程从此只需要关心“我发生了什么”，而“谁关心、关心什么”全部由外部订阅者决定——这就是关注点分离。

3️⃣ TokenCostAudit：不乱猜价格的 Token 审计员 ​

Token 审计用 tiktoken 估算文本 Token 数。模型价格经常变化，因此组件不内置假价格：调用方必须从供应商官方定价页确认后显式传入。没配置价格时仍统计用量，但成本栏显示“未配置当前官方价格”。这就像收银员可以数清商品，却不能自己编价签。

python
class TokenCostAudit:
    """Token 审计员：价格必须由调用方按官方当前价格显式传入。"""
    def __init__(self, prices=None):
        self.prices = prices or {}
        self._records = []

    def record_usage(self, model_name: str, input_tokens: int, output_tokens: int,
                     latency_ms: float = 0.0) -> None:
        """记录一次 LLM 调用用量：输入/输出 token 与耗时，并累计成本"""
        price = self.prices.get(model_name)
        cost = None if not price else (
            input_tokens / 1e6 * price["input"] + output_tokens / 1e6 * price["output"]
        )
        self._records.append({"model": model_name, "input_tokens": input_tokens, "output_tokens": output_tokens,
                              "cost": cost, "latency_ms": latency_ms})

summary() 会把累计统计整理成人类可读文本。只有每条记录都有显式价格时才求和，否则成本显示“未配置当前官方价格”：

python
    def summary(self) -> str:
        """输出人类可读的费用审计账单文本"""
        n = len(self._records)
        recs = self._records
        total_in = sum(r["input_tokens"] for r in recs); total_out = sum(r["output_tokens"] for r in recs)
        priced = [r["cost"] for r in recs if r["cost"] is not None]
        cost_text = f"¥ {sum(priced):.4f}" if priced and len(priced) == n else "未配置当前官方价格"
        return (f"📊 ====== Token 费用审计账单 ======\n"
                f"🔢 调用次数      : {n}\n"
                f"📥 输入 Tokens   : {total_in:,}\n"
                f"📤 输出 Tokens   : {total_out:,}\n"
                f"🧮 总 Tokens     : {total_in + total_out:,}\n"
                f"💴 预估成本      : {cost_text}\n")

这里的 Token 通常优先读取模型响应中的真实 usage；供应商没有返回用量时才用 tokenizer 估算。估算值和真实账单可能不同，所以报告要明确数据来源。

4️⃣ EvalSuite：评估套件（hello-agents 第12章思想） ​

EvalSuite 对一组任务各跑 trials 次，每次用独立 EventBus 收集轨迹。每条 EvalCase 都带验证器，必须检查答案或执行产物是否真的正确；模型自己说“我完成了”不算证据。没有验证器的开放问题会标为 unverified，不计算成功率。

python
@dataclass
class EvalCase:
    name: str
    prompt: str
    validator: Callable[[str], bool]

case = EvalCase(
    name="算术结果",
    prompt="计算 25 * 4 + 10",
    validator=lambda output: "110" in output,
)

suite = EvalSuite(client=client, tasks=[case], trials=3)

💡 关键理解：run_eval 接受一个 agent_runner_callable(bus, task) —— 你的 Agent 被包装成“接收事件总线、执行任务、返回最终文本”的函数。评估套件完全不关心 Agent 内部实现，只要它愿意 emit 事件即可被统计。这正体现了事件驱动的开闭原则。

🧭 契约：如何把任意 Agent 包装成可评估的 runner？ ​

评估套件对 runner 只有一条要求：签名为 (bus, task) -> str，并在关键节点主动 emit 事件。以 8.13 收官章节的 MiniAgent（code/s13_mini_agent.py）为例，只需在它的主循环里插入几行"广播"即可无缝接入：

python
def agent_runner(bus: EventBus, task: str) -> str:
    """🧭 把任意 Agent 包装成 (bus, task) -> 最终文本 的可评估函数"""
    audit = TokenCostAudit()
    bus.emit(AgentEvent("agent_start", content=f"🎬 开始任务：{task}"))
    reply = your_agent.run(task)                      # 🏃 你的 Agent 主循环（跑完并返回最终答复）
    bus.emit(AgentEvent("llm_call", content=reply,
                        tokens=audit.estimate_tokens(reply), latency_ms=random.uniform(300, 900)))
    if "失败" in reply:                               # 出错时额外广播 error 事件
        bus.emit(AgentEvent("error", content="捕获到模型报错，正在记录。"))
    bus.emit(AgentEvent("finish", content=reply))
    return reply

接入成本极低：Agent 主流程几乎不改，只多几行 emit；而费用统计、轨迹回放、成功率计算这些“体检项目”全部由订阅者完成。

📊 实战演示：一张评估报告 JSON ​

在 __main__ 自测中，两个固定 Mock 任务分别用“答案包含 110”和“回答包含 404 与路由”验证结果，全程本地、不联网。普通字符串任务则会显示 verification: unverified 和 success_rate: null。

json
{
  "summary": {
    "total_tasks": 2,
    "total_runs": 6,
    "verified_runs": 6,
    "overall_success_rate": 1.0,
    "avg_latency_ms": 112.9,
    "avg_tokens": 53.0,
    "cost_audit": "...预估成本: 未配置当前官方价格..."
  },
  "tasks": {
    "算术结果": { "success_rate": 1.0, "verification": "validator" },
    "404 诊断": { "success_rate": 1.0, "verification": "validator" }
  }
}

这份 JSON 就是 Agent 的“体检报告”：哪个任务稳定、哪个任务必挂、平均花多少钱、平均跑多慢，一目了然。你还可以从 traces 里逐条回放某次运行的完整事件轨迹：

[agent_start] 🎬 开始任务：写一个计算器程序   (tokens=0,   0ms)
[  llm_call] ✅ 任务执行成功，已顺利完成全部步骤！ (tokens=34, 888ms)
[    finish] ✅ 任务执行成功，已顺利完成全部步骤！ (tokens=0,   0ms)

🕹️ 在 Gradio 中动手体验 ​

在 code/app.py 中切换到 8.12 可观测性与性能评估 标签页：

  1. 点击 🎯 本地 Mock 评估（2 任务 × 3 次）：全程不联网、不调模型，用固定返回值验证事件轨迹、验证器和 Token 统计管道；
  2. 或点击 🚀 真实引擎联动评估（s01+s02 ReActAgent）：真实调用 ZhipuGLMClient 和 ReActAgent，需要配置 ZHIPU_API_KEY；
  3. 观察三块面板：
    • 📊 评估报告：成功率 / 平均耗时 / 平均 Token 汇总表；
    • 💰 费用账单：Token 审计员输出的成本明细；
    • 📜 轨迹回放：某次运行的事件流水线（agent_start → llm_call → tool_call → … → finish）。

评估应围绕任务结果 ​

Token、耗时和调用次数只能描述运行成本，不能单独证明任务质量。评估集应同时记录结果是否正确、工具是否按权限使用、关键证据是否完整，以及人工需要返工多久。对于无法自动判断的任务，可以用明确量表做人工抽检。

比较两个版本时，应固定任务集、初始状态和外部依赖，并保存失败轨迹。只展示平均数容易掩盖偶发的严重错误，因此还要关注最慢请求、连续失败和高风险工具的误调用。价格换算应注明来源和日期，无法确认时保留 Token 数量即可。

📝 本节小结 ​

  • 事件驱动是可观测性的基石：Agent 只负责 emit，外部用 subscribe 各取所需（参考 Pi 的事件驱动订阅机制），彻底解耦日志与主流程；
  • 费用审计不编价格：优先记录真实 usage；只有显式配置当前官方价格时才换算成本，否则清楚标成未知；
  • 评估套件让优化有据可依：成功率由任务验证器判断，没有验证器就标记为 unverified，同时保留时延、Token 和完整轨迹；
  • 下一步：本节我们把"过程"看得清清楚楚。现在万事俱备，是时候把所有器官——思考、规划、工具、权限、记忆、子代理、会话、观测——一次性组装成一台 会联网搜索、会深度思考的个人 Mini-Agent！请进入收官章节——8.13 综合实战：打造个人 Mini-Agent！