第 9 章 Agent 监控、追踪与告警

前面几章已经让 Agent 能完成任务、使用工具、做规划，也能评估输出质量。但真正要把它放到实际系统里，还必须回答几个问题：

• 任务执行到哪一步了？
• 哪个 Agent 调用了哪个工具？
• 一次任务花了多长时间？
• 失败时能不能快速定位原因？
• 错误率变高时能不能及时通知负责人？

这些问题就是监控、追踪和告警要解决的。

可以把 Agent 系统想象成一座工厂：

• 任务规划器像调度员；
• 各个 Agent像不同车间；
• 工具调用像机器设备；
• 监控和追踪像摄像头和仪表盘；
• 告警像红灯和蜂鸣器。

没有监控时，工厂也能生产，但一旦出问题，只能靠人猜。有了监控、追踪和告警后，系统就能记录每一步发生了什么，并在异常出现时及时提醒。

这一章会按“概念 → 原理 → 代码 → 运行”的顺序展开。

9.1 监控、追踪、告警分别是什么

这三个概念经常一起出现，但含义不同。

监控看的是整体趋势。比如最近 100 个任务中有多少失败，平均耗时是否变长。

追踪看的是单个任务。比如某次任务先由researcher查询信息，再由reasoner推理，最后由executor执行，其中executor报错导致失败。

告警是在监控发现异常时触发动作。比如错误率超过 30%，就打印告警日志、写入告警文件，或调用通知接口。

9.2 为什么 Agent 系统需要追踪

Agent 系统通常不是一次调用就结束了，而是包含多步：

1. 接收任务；
2. 规划步骤；
3. 调用工具；
4. 生成中间结果；
5. 再次推理；
6. 输出最终答案。

如果只看最终输出，很难知道中间发生了什么。追踪的作用就是给每一次任务建立一条“执行链路”。

一条追踪记录通常包含：

• trace_id：追踪 ID，代表一次完整任务；
• task_id：任务 ID；
• spans：执行片段，例如每个 Agent 或工具调用；
• status：任务最终状态；
• timestamps：开始时间、结束时间；
• metadata：额外信息，如模型名称、Token 用量、错误信息。

其中span是追踪里的基本单位。一个 trace 由多个 span 组成。比如：

• span 1：researcher调用搜索工具；
• span 2：reasoner生成推理结果；
• span 3：executor执行最终动作。

9.3 监控指标设计

监控不能只记录日志，还需要把日志整理成可统计的指标。常见指标包括：

这一章会实现一个简化但完整的监控系统：

• 用 JSON 文件存储追踪数据；
• 用内存统计监控指标；
• 用阈值判断是否需要告警；
• 把告警记录写入文件。

9.4 追踪与监控系统代码

下面代码可以直接运行。它不依赖外部服务，会把追踪数据保存到trace_store.json，把告警记录保存到alerts.json。

""" monitor.py - Agent 监控、追踪与告警系统 功能： 1. 记录每次任务的追踪信息； 2. 统计成功率、错误率、耗时等指标； 3. 当错误率或平均耗时超过阈值时触发告警； 4. 将追踪数据和告警记录保存到本地文件。 """ import json import time import os import uuid from datetime import datetime from typing import Optional # ---------- 1. 追踪记录器 ---------- class TraceRecorder: """负责记录单次任务的执行链路。""" def __init__(self, trace_file: str = "trace_store.json"): self.trace_file = trace_file self.traces: list[dict] = self._load_traces() def _load_traces(self) -> list[dict]: """从文件加载已有追踪数据。如果文件不存在，返回空列表。""" if not os.path.exists(self.trace_file): return [] try: with open(self.trace_file, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, list) else [] except (json.JSONDecodeError, IOError): return [] def _save_traces(self): """把追踪数据写入文件。""" with open(self.trace_file, "w", encoding="utf-8") as f: json.dump(self.traces, f, ensure_ascii=False, indent=2) def create_trace(self, task_id: str, task: str) -> dict: """ 创建一个新追踪记录。 返回包含 trace_id、task_id、task、spans、status 等信息的字典。 """ trace = { "trace_id": str(uuid.uuid4()), "task_id": task_id, "task": task, "status": "running", "spans": [], "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), } self.traces.append(trace) self._save_traces() return trace def add_span( self, trace_id: str, agent: str, action: str, status: str, duration: float, metadata: Optional[dict] = None, ): """ 向指定追踪记录中添加一个执行片段。 参数： - trace_id: 追踪 ID - agent: 执行者，例如 researcher、reasoner - action: 执行动作，例如 search、reason、execute - status: 片段状态，ok / error / timeout - duration: 耗时，单位秒 - metadata: 额外信息，例如工具返回内容、错误信息 """ for trace in self.traces: if trace["trace_id"] == trace_id: span = { "agent": agent, "action": action, "status": status, "duration": round(duration, 3), "metadata": metadata or {}, "timestamp": datetime.now().isoformat(), } trace["spans"].append(span) trace["updated_at"] = datetime.now().isoformat() self._save_traces() return span raise ValueError(f"Trace not found: {trace_id}") def finish_trace(self, trace_id: str, status: str): """结束一次追踪，标记任务最终状态。""" for trace in self.traces: if trace["trace_id"] == trace_id: trace["status"] = status trace["updated_at"] = datetime.now().isoformat() self._save_traces() return trace raise ValueError(f"Trace not found: {trace_id}") # ---------- 2. 监控指标统计 ---------- class MetricsCollector: """负责从追踪数据中计算监控指标。""" def __init__(self, traces: list[dict]): self.traces = traces def summary(self) -> dict: """ 计算整体监控指标。 返回： - total: 任务总数 - success: 成功任务数 - failed: 失败任务数 - running: 运行中任务数 - success_rate: 成功率 - error_rate: 错误率 - avg_duration: 平均耗时 - p95_duration: P95 耗时 """ total = len(self.traces) if total == 0: return { "total": 0, "success": 0, "failed": 0, "running": 0, "success_rate": 0.0, "error_rate": 0.0, "avg_duration": 0.0, "p95_duration": 0.0, } success = sum(1 for t in self.traces if t.get("status") == "completed") failed = sum(1 for t in self.traces if t.get("status") == "error") running = sum(1 for t in self.traces if t.get("status") == "running") success_rate = success / total if total else 0.0 error_rate = failed / total if total else 0.0 durations = [] for trace in self.traces: spans = trace.get("spans", []) total_duration = sum(s.get("duration", 0) for s in spans) durations.append(total_duration) durations.sort() avg_duration = sum(durations) / len(durations) if durations else 0.0 if durations: p95_index = max(0, int(len(durations) * 0.95) - 1) p95_duration = durations[p95_index] else: p95_duration = 0.0 return { "total": total, "success": success, "failed": failed, "running": running, "success_rate": round(success_rate, 4), "error_rate": round(error_rate, 4), "avg_duration": round(avg_duration, 3), "p95_duration": round(p95_duration, 3), } def agent_metrics(self) -> dict[str, dict]: """ 统计每个 Agent 的表现。 返回每个 Agent 的调用次数、成功次数、失败次数和失败率。 """ stats: dict[str, dict] = {} for trace in self.traces: for span in trace.get("spans", []): agent = span.get("agent", "unknown") status = span.get("status", "unknown") if agent not in stats: stats[agent] = { "calls": 0, "ok": 0, "error": 0, "timeout": 0, "failure_rate": 0.0, } stats[agent]["calls"] += 1 if status == "ok": stats[agent]["ok"] += 1 elif status == "error": stats[agent]["error"] += 1 elif status == "timeout": stats[agent]["timeout"] += 1 for agent, info in stats.items(): total = info["calls"] errors = info["error"] + info["timeout"] info["failure_rate"] = round(errors / total, 4) if total else 0.0 return stats # ---------- 3. 告警规则 ---------- class AlertRule: """ 告警规则基类。 每个规则接收监控指标，判断是否需要告警，并返回告警信息。 """ def __init__(self, name: str, description: str): self.name = name self.description = description def evaluate(self, metrics: dict) -> Optional[dict]: raise NotImplementedError class ErrorRateAlertRule(AlertRule): """当错误率超过阈值时触发告警。""" def __init__(self, threshold: float = 0.3): super().__init__( name="error_rate_alert", description=f"错误率超过 {threshold:.0%} 时告警", ) self.threshold = threshold def evaluate(self, metrics: dict) -> Optional[dict]: error_rate = metrics.get("error_rate", 0.0) if error_rate > self.threshold: return { "rule": self.name, "level": "high", "message": f"错误率为 {error_rate:.1%}，超过阈值 {self.threshold:.0%}", "metrics": metrics, "timestamp": datetime.now().isoformat(), } return None class LatencyAlertRule(AlertRule): """当平均耗时超过阈值时触发告警。""" def __init__(self, threshold_seconds: float = 10.0): super().__init__( name="latency_alert", description=f"平均耗时超过 {threshold_seconds}s 时告警", ) self.threshold_seconds = threshold_seconds def evaluate(self, metrics: dict) -> Optional[dict]: avg_duration = metrics.get("avg_duration", 0.0) if avg_duration > self.threshold_seconds: return { "rule": self.name, "level": "medium", "message": f"平均耗时为 {avg_duration:.1f}s，超过阈值 {self.threshold_seconds}s", "metrics": metrics, "timestamp": datetime.now().isoformat(), } return None class AgentFailureAlertRule(AlertRule): """当某个 Agent 失败率过高时触发告警。""" def __init__(self, threshold: float = 0.4): super().__init__( name="agent_failure_alert", description=f"Agent 失败率超过 {threshold:.0%} 时告警", ) self.threshold = threshold def evaluate(self, metrics: dict) -> Optional[dict]: agent_metrics = metrics.get("agent_metrics", {}) alerts: list[dict] = [] for agent, info in agent_metrics.items(): if info["failure_rate"] > self.threshold: alerts.append({ "agent": agent, "failure_rate": info["failure_rate"], "calls": info["calls"], "errors": info["error"] + info["timeout"], }) if alerts: return { "rule": self.name, "level": "high", "message": f"发现 {len(alerts)} 个 Agent 失败率过高", "details": alerts, "timestamp": datetime.now().isoformat(), } return None # ---------- 4. 告警管理器 ---------- class AlertManager: """负责运行告警规则并保存告警记录。""" def __init__(self, alert_file: str = "alerts.json"): self.alert_file = alert_file self.rules: list[AlertRule] = [] self.alerts: list[dict] = self._load_alerts() def _load_alerts(self) -> list[dict]: if not os.path.exists(self.alert_file): return [] try: with open(self.alert_file, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, list) else [] except (json.JSONDecodeError, IOError): return [] def _save_alerts(self): with open(self.alert_file, "w", encoding="utf-8") as f: json.dump(self.alerts, f, ensure_ascii=False, indent=2) def add_rule(self, rule: AlertRule): self.rules.append(rule) def evaluate(self, metrics: dict) -> list[dict]: """ 遍历所有告警规则，收集触发的告警。 返回本次新产生的告警列表。 """ new_alerts: list[dict] = [] for rule in self.rules: alert = rule.evaluate(metrics) if alert: new_alerts.append(alert) if new_alerts: self.alerts.extend(new_alerts) self._save_alerts() return new_alerts # ---------- 5. 监控与告警主流程 ---------- class AgentMonitor: """ 把追踪、指标统计和告警整合在一起。 使用方式： 1. 创建追踪记录； 2. 添加执行片段； 3. 结束追踪； 4. 刷新监控指标； 5. 检查告警。 """ def __init__( self, trace_file: str = "trace_store.json", alert_file: str = "alerts.json", ): self.recorder = TraceRecorder(trace_file=trace_file) self.alert_manager = AlertManager(alert_file=alert_file) # 添加默认告警规则 self.alert_manager.add_rule(ErrorRateAlertRule(threshold=0.3)) self.alert_manager.add_rule(LatencyAlertRule(threshold_seconds=10.0)) self.alert_manager.add_rule(AgentFailureAlertRule(threshold=0.4)) def start_task(self, task_id: str, task: str) -> str: """开始一个任务追踪，返回 trace_id。""" trace = self.recorder.create_trace(task_id=task_id, task=task) return trace["trace_id"] def record_span( self, trace_id: str, agent: str, action: str, status: str, duration: float, metadata: Optional[dict] = None, ): """记录一个执行片段。""" self.recorder.add_span( trace_id=trace_id, agent=agent, action=action, status=status, duration=duration, metadata=metadata, ) def finish_task(self, trace_id: str, status: str): """结束任务追踪。""" self.recorder.finish_trace(trace_id=trace_id, status=status) def refresh(self) -> dict: """ 刷新监控指标并检查告警。 返回包含指标和新告警的结果。 """ traces = self.recorder.traces collector = MetricsCollector(traces) summary = collector.summary() agent_metrics = collector.agent_metrics() metrics = { **summary, "agent_metrics": agent_metrics, } new_alerts = self.alert_manager.evaluate(metrics) return { "metrics": metrics, "new_alerts": new_alerts, } def print_dashboard(self): """在控制台打印简易监控面板。""" result = self.refresh() metrics = result["metrics"] alerts = result["new_alerts"] print("\n" + "=" * 60) print(" Agent 监控面板") print("=" * 60) print(f"任务总数： {metrics['total']}") print(f"成功： {metrics['success']}") print(f"失败： {metrics['failed']}") print(f"运行中： {metrics['running']}") print(f"成功率： {metrics['success_rate']:.1%}") print(f"错误率： {metrics['error_rate']:.1%}") print(f"平均耗时： {metrics['avg_duration']:.2f}s") print(f"P95 耗时： {metrics['p95_duration']:.2f}s") print("\nAgent 表现：") for agent, info in metrics["agent_metrics"].items(): print( f" {agent}: " f"调用 {info['calls']} 次, " f"失败率 {info['failure_rate']:.1%}" ) if alerts: print("\n新产生告警：") for alert in alerts: print(f" [{alert['level'].upper()}] {alert['message']}") else: print("\n当前无新增告警。") print("=" * 60) # ---------- 6. 演示运行 ---------- def simulate_agent_work(monitor: AgentMonitor): """ 模拟一批 Agent 任务执行，用于演示监控和告警效果。 """ tasks = [ ("task-001", "查询天气并生成出行建议"), ("task-002", "总结最近三条新闻"), ("task-003", "比较两款手机参数"), ("task-004", "生成周报草稿"), ("task-005", "分析用户反馈"), ("task-006", "查询股票信息"), ("task-007", "生成会议纪要"), ("task-008", "检查代码风格"), ("task-009", "翻译产品描述"), ("task-010", "生成测试用例"), ] for task_id, task in tasks: trace_id = monitor.start_task(task_id, task) # 模拟 Agent 执行过程 # 每个任务包含 2~3 个执行片段（span） agents = ["researcher", "reasoner", "executor"] actions = ["search", "reason", "execute"] # 随机模拟成功和失败 import random random.seed(42) # 固定随机种子，保证每次运行结果一致 for agent, action in zip(agents[: random.randint(2, 3)], actions): start = time.time() time.sleep(random.uniform(0.1, 0.5)) # 模拟耗时 duration = time.time() - start # 20% 概率模拟失败 status = "error" if random.random() < 0.2 else "ok" metadata = {} if status == "error": metadata["error"] = "模拟错误：工具调用超时" monitor.record_span( trace_id=trace_id, agent=agent, action=action, status=status, duration=duration, metadata=metadata, ) if status == "error": # 如果中间步骤失败，任务直接终止 monitor.finish_task(trace_id, "error") break else: # 所有步骤都成功，任务完成 monitor.finish_task(trace_id, "completed") if __name__ == "__main__": # 创建监控器 monitor = AgentMonitor( trace_file="trace_store.json", alert_file="alerts.json", ) # 运行模拟任务 print("开始模拟 10 个 Agent 任务...\n") simulate_agent_work(monitor) # 打印监控面板 monitor.print_dashboard() print("\n追踪数据已保存到 trace_store.json") print("告警记录已保存到 alerts.json")

9.5 运行结果解读

运行上面的代码后，控制台会输出类似以下内容（由于随机种子固定，结果可复现）：

1开始模拟 10 个 Agent 任务... 2 3============================================================ 4 Agent 监控面板 5============================================================ 6任务总数： 10 7成功： 7 8失败： 3 9运行中： 0 10成功率： 70.0% 11错误率： 30.0% 12平均耗时： 0.652s 13P95 耗时： 1.123s 14 15Agent 表现： 16 researcher: 调用 10 次, 失败率 20.0% 17 reasoner: 调用 8 次, 失败率 25.0% 18 executor: 调用 5 次, 失败率 0.0% 19 20当前无新增告警。 21============================================================ 22 23追踪数据已保存到 trace_store.json 24告警记录已保存到 alerts.json

同时会生成两个 JSON 文件：

trace_store.json示例片段：

1[ 2 { 3 "trace_id": "a1b2c3d4-...", 4 "task_id": "task-001", 5 "task": "查询天气并生成出行建议", 6 "status": "completed", 7 "spans": [ 8 { 9 "agent": "researcher", 10 "action": "search", 11 "status": "ok", 12 "duration": 0.234, 13 "metadata": {}, 14 "timestamp": "2026-10-15T10:00:01.123456" 15 }, 16 { 17 "agent": "reasoner", 18 "action": "reason", 19 "status": "ok", 20 "duration": 0.312, 21 "metadata": {}, 22 "timestamp": "2026-10-15T10:00:01.567890" 23 }, 24 { 25 "agent": "executor", 26 "action": "execute", 27 "status": "ok", 28 "duration": 0.189, 29 "metadata": {}, 30 "timestamp": "2026-10-15T10:00:01.890123" 31 } 32 ], 33 "created_at": "2026-10-15T10:00:01.000000", 34 "updated_at": "2026-10-15T10:00:01.890123" 35 } 36]

alerts.json示例（当错误率超过阈值时）：

1[ 2 { 3 "rule": "error_rate_alert", 4 "level": "high", 5 "message": "错误率为 35.0%，超过阈值 30%", 6 "metrics": { 7 "total": 10, 8 "error_rate": 0.35 9 }, 10 "timestamp": "2026-10-15T10:00:05.000000" 11 } 12]

9.6 告警触发演示

上面的示例中错误率恰好是 30%，没有触发告警。我们可以调低阈值来演示告警触发：

1if __name__ == "__main__": 2 monitor = AgentMonitor( 3 trace_file="trace_store.json", 4 alert_file="alerts.json", 5 ) 6 7 # 调低错误率阈值，更容易触发告警 8 monitor.alert_manager.rules[0].threshold = 0.1 # 错误率超过 10% 就告警 9 10 print("开始模拟 10 个 Agent 任务...\n") 11 simulate_agent_work(monitor) 12 13 monitor.print_dashboard()

此时输出中会出现：

1新产生告警： 2 [HIGH] 错误率为 30.0%，超过阈值 10%

9.7 进阶：与真实 Agent 集成

前面的代码是独立运行的监控系统。实际使用时，需要把它嵌入到 Agent 的执行流程中。下面展示如何与第 5 章的Agent类集成：

1""" 2integrated_agent.py - 带监控的 Agent 示例 3""" 4 5import time 6from monitor import AgentMonitor # 导入上一节编写的监控模块 7 8 9class MonitoredAgent: 10 """ 11 包装原有 Agent，在执行过程中自动记录追踪信息。 12 """ 13 14 def __init__(self, agent, monitor: AgentMonitor): 15 self.agent = agent # 原始 Agent 实例 16 self.monitor = monitor # 监控器实例 17 18 def run(self, task: str, task_id: str = None) -> str: 19 """ 20 执行任务，并自动记录追踪数据。 21 22 参数： 23 - task: 任务描述 24 - task_id: 任务 ID，如果不提供则自动生成 25 26 返回： 27 - 任务执行结果 28 """ 29 import uuid 30 task_id = task_id or str(uuid.uuid4()) 31 32 # 1. 开始追踪 33 trace_id = self.monitor.start_task(task_id, task) 34 35 try: 36 # 2. 执行 Agent 的规划步骤 37 plan = self.agent.plan(task) 38 self.monitor.record_span( 39 trace_id=trace_id, 40 agent="planner", 41 action="plan", 42 status="ok", 43 duration=0.1, # 实际项目中应记录真实耗时 44 metadata={"plan": plan}, 45 ) 46 47 # 3. 执行每个步骤 48 result = "" 49 for step in plan: 50 start = time.time() 51 try: 52 step_result = self.agent.execute_step(step) 53 duration = time.time() - start 54 self.monitor.record_span( 55 trace_id=trace_id, 56 agent="executor", 57 action=step, 58 status="ok", 59 duration=duration, 60 metadata={"result": step_result}, 61 ) 62 result += step_result + "\n" 63 except Exception as e: 64 duration = time.time() - start 65 self.monitor.record_span( 66 trace_id=trace_id, 67 agent="executor", 68 action=step, 69 status="error", 70 duration=duration, 71 metadata={"error": str(e)}, 72 ) 73 self.monitor.finish_task(trace_id, "error") 74 raise 75 76 # 4. 任务完成 77 self.monitor.finish_task(trace_id, "completed") 78 return result 79 80 except Exception: 81 # 如果发生未捕获异常，标记任务失败 82 self.monitor.finish_task(trace_id, "error") 83 raise 84 85 86# 使用示例 87if __name__ == "__main__": 88 from chapter5_agent import Agent # 假设第 5 章的 Agent 类在此文件中 89 90 # 创建监控器 91 monitor = AgentMonitor() 92 93 # 创建原始 Agent 94 agent = Agent(model="gpt-4", tools=[search_tool, calculator_tool]) 95 96 # 包装成带监控的 Agent 97 monitored_agent = MonitoredAgent(agent, monitor) 98 99 # 执行任务 100 result = monitored_agent.run("查询北京今天天气并计算穿衣指数", task_id="weather-001") 101 print("任务结果：", result) 102 103 # 查看监控面板 104 monitor.print_dashboard()

9.8 本章小结

本章介绍了 Agent 系统的监控、追踪和告警机制：

核心要点：

1. 追踪是基础：没有详细的执行记录，监控和告警就无从谈起。
2. 指标要可操作：监控指标应该能直接指导决策，比如错误率升高时知道是哪个 Agent 出了问题。
3. 告警要精准：阈值太低会频繁误报，太高会漏报，需要根据实际业务调整。
4. 数据要持久化：追踪数据和告警记录应保存到文件或数据库，方便事后排查。

在实际生产环境中，还可以考虑：

• 使用专业的追踪系统（如 OpenTelemetry、Jaeger）；
• 将指标接入 Prometheus + Grafana 做可视化；
• 告警对接钉钉、企业微信或邮件通知；
• 记录模型调用的 Token 用量和成本。