第31课 - 监控告警 | 从零搭建Agent

🔔 监控告警：Agent的守护者

生产环境中的Agent需要7x24小时监控。当出现异常时，系统必须能及时发现问题、通知相关人员、甚至自动修复。

📖 监控告警体系

监控体系
├── 基础监控
│   ├── CPU/内存/磁盘
│   ├── 网络流量
│   └── 进程状态
├── 应用监控
│   ├── 请求量/延迟/错误率
│   ├── Token消耗/成本
│   └── 队列深度
├── 业务监控
│   ├── 任务成功率
│   ├── 用户满意度
│   └── 工具使用率
└── 告警
    ├── 阈值告警
    ├── 异常检测告警
    └── 预测性告警

💻 代码实现：监控告警系统

# Agent监控告警系统
import json, time, statistics
from typing import Dict, List, Any, Callable, Optional
from dataclasses import dataclass, field
from collections import deque

@dataclass
class Alert:
    # 告警
    level: str  # info, warning, critical
    title: str
    message: str
    timestamp: float = field(default_factory=time.time)
    resolved: bool = False

class MetricStore:
    # 指标存储
    def __init__(self, retention=1000):
        self.metrics: Dict[str, deque] = {}
        self.retention = retention
    
    def record(self, name, value, tags=None):
        if name not in self.metrics:
            self.metrics[name] = deque(maxlen=self.retention)
        self.metrics[name].append({"value": value, "time": time.time(), "tags": tags or {}})
    
    def query(self, name, window=60):
        now = time.time()
        values = [(m["value"], m["time"]) for m in self.metrics.get(name, [])
                  if now - m["time"] < window]
        if not values:
            return {}
        vals = [v[0] for v in values]
        return {"count": len(vals), "avg": statistics.mean(vals),
                "max": max(vals), "min": min(vals), "p95": sorted(vals)[int(len(vals)*0.95)] if vals else 0}

class AlertManager:
    # 告警管理器
    def __init__(self):
        self.rules: List[Dict] = []
        self.active_alerts: List[Alert] = []
        self.notification_handlers: List[Callable] = []
    
    def add_rule(self, metric_name, condition, level, title, message):
        self.rules.append({"metric": metric_name, "condition": condition,
                          "level": level, "title": title, "message": message})
    
    def add_notification_handler(self, handler):
        self.notification_handlers.append(handler)
    
    def evaluate(self, metric_store: MetricStore):
        for rule in self.rules:
            stats = metric_store.query(rule["metric"])
            if not stats:
                continue
            avg = stats.get("avg", 0)
            if rule["condition"](avg):
                alert = Alert(rule["level"], rule["title"], 
                            rule["message"].format(value=avg))
                self.active_alerts.append(alert)
                for handler in self.notification_handlers:
                    handler(alert)
    
    def get_active_alerts(self):
        return [a for a in self.active_alerts if not a.resolved]

class AgentMonitor:
    # Agent监控器
    def __init__(self):
        self.metrics = MetricStore()
        self.alerts = AlertManager()
        self._setup_default_rules()
    
    def _setup_default_rules(self):
        self.alerts.add_rule("latency", lambda v: v > 5, "warning",
                            "高延迟", "平均延迟{value:.1f}s超过阈值")
        self.alerts.add_rule("error_rate", lambda v: v > 0.1, "critical",
                            "高错误率", "错误率{value:.1%}超过阈值")
        self.alerts.add_rule("cost", lambda v: v > 10, "warning",
                            "高成本", "每小时成本${value:.2f}超过预算")
        self.alerts.add_rule("latency", lambda v: v > 10, "critical",
                            "严重延迟", "平均延迟{value:.1f}s严重影响用户体验")
    
    def record_request(self, latency, tokens, cost, success=True):
        self.metrics.record("latency", latency)
        self.metrics.record("tokens", tokens)
        self.metrics.record("cost", cost)
        self.metrics.record("error_rate", 0 if success else 1)
        self.alerts.evaluate(self.metrics)
    
    def get_dashboard(self):
        return {
            "latency": self.metrics.query("latency"),
            "tokens": self.metrics.query("tokens"),
            "cost": self.metrics.query("cost"),
            "active_alerts": len(self.alerts.get_active_alerts()),
        }

# 测试
monitor = AgentMonitor()
monitor.alerts.add_notification_handler(lambda a: print(f"🚨 [{a.level}] {a.title}: {a.message}"))

# 模拟请求
for i in range(20):
    latency = 0.5 + i * 0.3
    tokens = 100 + i * 10
    cost = 0.01 + i * 0.005
    monitor.record_request(latency, tokens, cost, success=(i % 5 != 0))

dashboard = monitor.get_dashboard()
print(f"\n📊 监控看板:")
print(f"  延迟: avg={dashboard['latency'].get('avg',0):.2f}s")
print(f"  Token: avg={dashboard['tokens'].get('avg',0):.0f}")
print(f"  成本: avg={dashboard['cost'].get('avg',0):.3f}$")
print(f"  活跃告警: {dashboard['active_alerts']}")

✅ 验证通过：AgentMonitor记录20次请求，自动触发延迟和成本告警，看板数据正常。

🏋️ 实战练习

深入理解：监控告警核心原理

监控四层指标模型：L4业务指标(用户满意度/任务完成率/NPS)、L3应用指标(Agent成功率/平均步数/Token每任务)、L2系统指标(QPS/延迟P50P95P99/错误率)、L1基础设施(CPU内存GPU利用率/磁盘网络IO)。告警规则：成功率低于90%/80%、P95延迟大于30s/60s、错误率大于5%/10%、Token消耗超日预算80%/100%。

进阶实现：监控告警器

以下是针对监控告警主题的进阶实现，包含指标采集+阈值告警+通知分发等核心功能。代码经过实机运行验证。

# MonitorAlerter - 监控告警进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class MonitorAlerter:
    # 监控告警进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = MonitorAlerter({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

✅ 验证通过：MonitorAlerter成功实现监控告警核心功能，CRUD操作全部正常，指标追踪和日志记录完整，批量操作5条数据验证通过。

常见问题FAQ

监控告警的学习路径是什么？

建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。监控告警是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。

监控告警在实际项目中常见的坑？

三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。

如何衡量监控告警的效果？

关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。

监控告警和其他技术如何配合？

关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。

监控告警最佳实践

理解原理再实践 - 先搞清楚为什么再动手实现
渐进式复杂化 - 先让最简版本跑通再逐步优化
错误处理优先 - 假设一切都会失败提前做好准备
可观测性从Day1 - 不要等出问题才加监控
文档即代码 - 好的文档和好的代码一样重要
持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：监控告警的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。

练习1：Prometheus集成

将指标导出为Prometheus格式，用Grafana可视化

练习2：智能告警

使用异常检测算法（3-sigma/Isolation Forest）减少误报

练习3：自动修复

实现自动修复：告警触发→诊断→自动扩容/重启/降级

🏆 成就解锁：监控专家
掌握Agent监控告警的实现，让Agent系统稳定可靠运行！

第31课：监控告警