第 9 章 Agent 监控、追踪与告警-北京尧图网络

前面几章已经让 Agent 能完成任务、使用工具、做规划也能评估输出质量。但真正要把它放到实际系统里还必须回答几个问题任务执行到哪一步了哪个 Agent 调用了哪个工具一次任务花了多长时间失败时能不能快速定位原因错误率变高时能不能及时通知负责人这些问题就是监控、追踪和告警要解决的。可以把 Agent 系统想象成一座工厂任务规划器像调度员各个 Agent像不同车间工具调用像机器设备监控和追踪像摄像头和仪表盘告警像红灯和蜂鸣器。没有监控时工厂也能生产但一旦出问题只能靠人猜。有了监控、追踪和告警后系统就能记录每一步发生了什么并在异常出现时及时提醒。这一章会按“概念 → 原理 → 代码 → 运行”的顺序展开。9.1 监控、追踪、告警分别是什么这三个概念经常一起出现但含义不同。概念关注什么典型问题监控系统当前状态现在是否正常错误率是否升高追踪单次任务的执行链路这个任务经过了哪些步骤哪里失败了告警异常提醒错误率超过阈值时通知负责人监控看的是整体趋势。比如最近 100 个任务中有多少失败平均耗时是否变长。追踪看的是单个任务。比如某次任务先由researcher查询信息再由reasoner推理最后由executor执行其中executor报错导致失败。告警是在监控发现异常时触发动作。比如错误率超过 30%就打印告警日志、写入告警文件或调用通知接口。9.2 为什么 Agent 系统需要追踪Agent 系统通常不是一次调用就结束了而是包含多步接收任务规划步骤调用工具生成中间结果再次推理输出最终答案。如果只看最终输出很难知道中间发生了什么。追踪的作用就是给每一次任务建立一条“执行链路”。一条追踪记录通常包含trace_id追踪 ID代表一次完整任务task_id任务 IDspans执行片段例如每个 Agent 或工具调用status任务最终状态timestamps开始时间、结束时间metadata额外信息如模型名称、Token 用量、错误信息。其中span是追踪里的基本单位。一个 trace 由多个 span 组成。比如span 1researcher调用搜索工具span 2reasoner生成推理结果span 3executor执行最终动作。9.3 监控指标设计监控不能只记录日志还需要把日志整理成可统计的指标。常见指标包括指标含义用途任务总数已执行任务数量了解系统负载成功率成功完成任务的比例判断系统是否稳定错误率失败任务比例触发告警的重要依据平均耗时每次任务平均用时发现性能退化P95 耗时95% 任务的耗时上限发现长尾慢请求Agent 失败率单个 Agent 的失败比例定位问题 Agent工具错误率工具调用失败比例判断工具是否可靠这一章会实现一个简化但完整的监控系统用 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, encodingutf-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, encodingutf-8) as f: json.dump(self.traces, f, ensure_asciiFalse, indent2) 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(fTrace 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(fTrace 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__( nameerror_rate_alert, descriptionf错误率超过 {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__( namelatency_alert, descriptionf平均耗时超过 {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__( nameagent_failure_alert, descriptionfAgent 失败率超过 {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, encodingutf-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, encodingutf-8) as f: json.dump(self.alerts, f, ensure_asciiFalse, indent2) 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_filetrace_file) self.alert_manager AlertManager(alert_filealert_file) # 添加默认告警规则 self.alert_manager.add_rule(ErrorRateAlertRule(threshold0.3)) self.alert_manager.add_rule(LatencyAlertRule(threshold_seconds10.0)) self.alert_manager.add_rule(AgentFailureAlertRule(threshold0.4)) def start_task(self, task_id: str, task: str) - str: 开始一个任务追踪返回 trace_id。 trace self.recorder.create_trace(task_idtask_id, tasktask) 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_idtrace_id, agentagent, actionaction, statusstatus, durationduration, metadatametadata, ) def finish_task(self, trace_id: str, status: str): 结束任务追踪。 self.recorder.finish_trace(trace_idtrace_id, statusstatus) 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(fP95 耗时 {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_idtrace_id, agentagent, actionaction, statusstatus, durationduration, metadatametadata, ) if status error: # 如果中间步骤失败任务直接终止 monitor.finish_task(trace_id, error) break else: # 所有步骤都成功任务完成 monitor.finish_task(trace_id, completed) if __name__ __main__: # 创建监控器 monitor AgentMonitor( trace_filetrace_store.json, alert_filealerts.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_filetrace_store.json, 4 alert_filealerts.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_idtrace_id, 40 agentplanner, 41 actionplan, 42 statusok, 43 duration0.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_idtrace_id, 56 agentexecutor, 57 actionstep, 58 statusok, 59 durationduration, 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_idtrace_id, 67 agentexecutor, 68 actionstep, 69 statuserror, 70 durationduration, 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(modelgpt-4, tools[search_tool, calculator_tool]) 95 96 # 包装成带监控的 Agent 97 monitored_agent MonitoredAgent(agent, monitor) 98 99 # 执行任务 100 result monitored_agent.run(查询北京今天天气并计算穿衣指数, task_idweather-001) 101 print(任务结果, result) 102 103 # 查看监控面板 104 monitor.print_dashboard()9.8 本章小结本章介绍了 Agent 系统的监控、追踪和告警机制概念作用实现方式追踪记录单次任务的完整执行链路TraceRecorder记录 trace 和 span监控统计系统整体运行指标MetricsCollector计算成功率、耗时等告警异常时及时通知AlertManager按规则触发告警核心要点追踪是基础没有详细的执行记录监控和告警就无从谈起。指标要可操作监控指标应该能直接指导决策比如错误率升高时知道是哪个 Agent 出了问题。告警要精准阈值太低会频繁误报太高会漏报需要根据实际业务调整。数据要持久化追踪数据和告警记录应保存到文件或数据库方便事后排查。在实际生产环境中还可以考虑使用专业的追踪系统如 OpenTelemetry、Jaeger将指标接入 Prometheus Grafana 做可视化告警对接钉钉、企业微信或邮件通知记录模型调用的 Token 用量和成本。

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

相关推荐

2026年别再自建网关：企业级AI大模型API聚合平台深度测评

PWM控制本质解析：从占空比、频率到嵌入式实操配置

实验室用等离子表面处理设备处理包装薄膜的一点记录

更多资讯

DeepSeek本地部署实战：Ollama + Open WebUI完整部署指南

MCP与A2A如何选？从JSON-RPC到Agent协议分层解析

Seedance 2.0角色一致性视频生成全解析：从原理到实操

C语言指针与数组交互：内存地址、类型退化与安全修改

JavaScript刷算法：V8引擎约束下的最优解法

STM32裸机驱动OV2640输出JPEG流实战指南

今日更新

NetSpot 实战：用信道分析与频宽优化解决 Wi-Fi 卡顿

C++ split() 字符串分割的五种实现与性能实测

R语言坐标轴范围设置：从xlim/ylim到coord_cartesian的完整指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？