SRE

引言 Agent系统的故障排查比传统应用复杂得多——一个"回复质量下降"的问题可能涉及Prompt变化、模型版本更新、工具API变更、记忆检索质量下降等多个因素。没有系统化的排查方法论，工程师可能花费数小时甚至数天才能定位根因。 本文基于大量实战经验，提供一套系统化的Agent故障排查方法论。 故障排查金字塔 ┌─────────────┐ │ 用户反馈 │ │ "回复不对" │ └──────┬──────┘ │ ┌──────▼──────┐ │ 指标异常 │ │ 质量评分下降 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 链路追踪 │ │ 定位慢步骤 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 日志分析 │ │ 找到错误日志 │ └──────┬──────┘ │ ┌──────▼──────┐ │ 根因定位 │ │ Prompt/模型 │ │ /数据/代码 │ └─────────────┘ 故障分类 Agent系统故障分类 │ ├── 功能故障 │ ├── 响应错误（幻觉、曲解意图） │ ├── 工具调用失败 │ ├── 路由错误 │ └── 超时/死循环 │ ├── 性能故障 │ ├── 响应变慢 │ ├── 吞吐量下降 │ └── 资源耗尽 │ ├── 质量故障 │ ├── 回复质量下降 │ ├── 用户满意度降低 │ └── Token消耗异常 │ └── 成本故障 ├── Token消耗突增 ├── 基础设施成本异常 └── API调用费用超标 排查流程 Step 1：收集症状 class SymptomCollector: """症状收集器""" async def collect(self, incident_id: str) -> dict: """收集故障症状""" symptoms = { "incident_id": incident_id, "reported_at": datetime.now().isoformat(), "reported_by": None, "description": None, "affected_users": [], "affected_services": [], "start_time": None, "error_rate": None, "quality_score": None, "recent_changes": [], } # 从多个数据源收集 symptoms["recent_deployments"] = await self._get_recent_deployments() symptoms["recent_config_changes"] = await self._get_recent_config_changes() symptoms["recent_model_updates"] = await self._get_recent_model_updates() symptoms["error_logs"] = await self._get_recent_errors() symptoms["metrics_anomalies"] = await self._get_metrics_anomalies() return symptoms Step 2：复现问题 class IssueReproducer: """问题复现器""" async def reproduce( self, session_id: str, user_input: str ) -> dict: """复现问题""" reproduction = { "original_session": session_id, "input": user_input, "attempts": [], "reproduced": False, "reproduction_rate": 0, } # 尝试复现3次 for i in range(3): try: result = await self.agent.process(user_input) reproduction["attempts"].append({ "attempt": i + 1, "response": result["response"], "quality_score": result.get("quality_score"), "full_trace": result.get("trace"), }) # 判断是否复现 if self._is_similar_issue(result, user_input): reproduction["reproduced"] = True except Exception as e: reproduction["attempts"].append({ "attempt": i + 1, "error": str(e), }) reproduction["reproduction_rate"] = sum( 1 for a in reproduction["attempts"] if a.get("error") or self._is_similar_issue(a, user_input) ) / len(reproduction["attempts"]) return reproduction Step 3：分析日志 class LogAnalyzer: """日志分析器""" async def analyze_for_incident( self, incident: dict, time_window_minutes: int = 60 ) -> dict: """分析故障相关日志""" analysis = { "error_patterns": [], "timeline": [], "affected_components": set(), "suspected_root_cause": None, } # 获取时间窗口内的日志 start = incident["start_time"] - timedelta(minutes=10) end = incident["start_time"] + timedelta(minutes=time_window_minutes) logs = await self.log_store.query( time_range=(start, end), filters={ "level": ["ERROR", "CRITICAL"], "service": incident.get("affected_services", []) } ) # 分析错误模式 error_counts = {} for log in logs: error_key = f"{log['service']}:{log['message'][:50]}" error_counts[error_key] = error_counts.get(error_key, 0) + 1 analysis["affected_components"].add(log["service"]) analysis["timeline"].append({ "timestamp": log["timestamp"], "service": log["service"], "message": log["message"], }) # 排序错误频率 analysis["error_patterns"] = sorted( [{"pattern": k, "count": v} for k, v in error_counts.items()], key=lambda x: x["count"], reverse=True ) # 推断根因 if analysis["error_patterns"]: top_error = analysis["error_patterns"][0] analysis["suspected_root_cause"] = { "type": "error_pattern", "pattern": top_error["pattern"], "frequency": top_error["count"], "confidence": min(top_error["count"] / len(logs), 1.0) } return analysis Step 4：链路追踪分析 class TraceAnalyzer: """Trace分析器""" async def find_slow_spans( self, trace_id: str, threshold_ms: int = 1000 ) -> list: """找到慢Span""" trace = await self.jaeger.get_trace(trace_id) spans = self._flatten_spans(trace) slow_spans = [ { "operation": span["operationName"], "service": span["process"]["serviceName"], "duration_ms": span["duration"], "tags": span.get("tags", {}), } for span in spans if span["duration"] > threshold_ms * 1000 # 转换为微秒 ] return sorted(slow_spans, key=lambda s: s["duration_ms"], reverse=True) async def find_error_spans(self, trace_id: str) -> list: """找到错误Span""" trace = await self.jaeger.get_trace(trace_id) spans = self._flatten_spans(trace) error_spans = [ { "operation": span["operationName"], "service": span["process"]["serviceName"], "error": span.get("tags", {}).get("error", True), "logs": span.get("logs", []), } for span in spans if span.get("tags", {}).get("error") ] return error_spans Step 5：根因分析 class RootCauseAnalyzer: """根因分析器""" async def analyze(self, incident: dict) -> dict: """分析根因""" hypotheses = [] # 假设1：最近的部署导致 recent_deploy = await self._check_recent_deployment(incident) if recent_deploy: hypotheses.append({ "hypothesis": "Recent deployment caused the issue", "evidence": recent_deploy, "confidence": 0.8, "verify_command": f"kubectl rollback deployment {recent_deploy['deployment']}" }) # 假设2：模型行为变化 model_change = await self._check_model_change(incident) if model_change: hypotheses.append({ "hypothesis": "Model behavior changed", "evidence": model_change, "confidence": 0.7, "verify_command": f"Compare outputs with previous model version" }) # 假设3：Prompt被修改 prompt_change = await self._check_prompt_change(incident) if prompt_change: hypotheses.append({ "hypothesis": "Prompt template was modified", "evidence": prompt_change, "confidence": 0.9, "verify_command": f"git diff {prompt_change['commit']} prompts/" }) # 假设4：工具API变更 tool_change = await self._check_tool_api_change(incident) if tool_change: hypotheses.append({ "hypothesis": "Tool API behavior changed", "evidence": tool_change, "confidence": 0.6, "verify_command": f"Test tool {tool_change['tool']} with previous inputs" }) # 按置信度排序 hypotheses.sort(key=lambda h: h["confidence"], reverse=True) return { "incident_id": incident["id"], "hypotheses": hypotheses, "recommended_first_check": hypotheses[0] if hypotheses else None, } 常见故障排查 故障1：响应质量下降 class QualityDropTroubleshooter: """回复质量下降排查""" CHECKLIST = [ { "name": "Check model version", "command": "kubectl get configmap agent-config -o jsonpath='{.data.model_version}'", "fix": "Rollback model version if recently changed" }, { "name": "Check prompt template", "command": "git log --oneline -10 prompts/", "fix": "Revert prompt changes if quality dropped after commit" }, { "name": "Check tool success rate", "command": "rate(agent_tool_calls_total{status='success'}[1h])", "fix": "Debug failing tools, may affect response quality" }, { "name": "Check memory retrieval quality", "command": "Analyze recall@5 for recent queries", "fix": "Reindex vector database if recall dropped" }, { "name": "Check for model drift", "command": "Compare quality scores across time", "fix": "Consider model retraining or fine-tuning" }, ] 故障2：Token消耗突增 class TokenSpikeTroubleshooter: """Token消耗突增排查""" async def diagnose(self) -> dict: diagnosis = { "possible_causes": [], "recommendations": [], } # 检查是否有循环 cycles = await self._check_for_cycles() if cycles: diagnosis["possible_causes"].append("Agent entering tool call loops") diagnosis["recommendations"].append("Enable cycle detection and break logic") # 检查Prompt是否变长 prompt_length = await self._get_avg_prompt_length() if prompt_length > self.baseline_prompt_length * 1.5: diagnosis["possible_causes"].append("Prompt length increased significantly") diagnosis["recommendations"].append("Optimize prompt template, use summarization") # 检查是否启用了缓存 cache_hit_rate = await self._get_cache_hit_rate() if cache_hit_rate < 0.3: diagnosis["possible_causes"].append("Cache hit rate too low") diagnosis["recommendations"].append("Investigate cache configuration") return diagnosis 排查工具箱 class TroubleshootingToolkit: """排障工具箱""" TOOLS = { "log_query": { "description": "Query logs with Loki", "example": 'logcli query \'{service="agent"} |~ "ERROR"\' --since=1h', }, "trace_query": { "description": "Find traces with Jaeger", "example": 'jaeger query --service=agent --lookback=1h --minDuration=1s', }, "metrics_query": { "description": "Query metrics with PromQL", "example": 'rate(agent_requests_total[5m])', }, "config_check": { "description": "Check K8s config", "example": 'kubectl get configmap agent-config -o yaml', }, "rollback": { "description": "Rollback deployment", "example": 'kubectl rollout undo deployment agent-service', }, "compare_versions": { "description": "Compare model/prompt versions", "example": 'ab compare --model-a=gpt-4o --model-b=gpt-4o-mini --samples=100', } } 总结 Agent系统故障排查需要系统化的方法论——从收集症状开始，通过复现问题、分析日志、追踪链路，最终定位根因。关键在于将模糊的"回复不对"拆解成可度量的指标异常，再通过对指标的分析定位到具体的组件和行为。 ...

引言 Agent系统的运维复杂度远超传统Web应用——LLM推理服务的GPU故障、工具调用的外部依赖故障、Token消耗突增导致的成本爆炸，这些都需要自动化运维系统能够及时发现并自动处理。2026年，成熟的Agent系统已经实现了从"手动运维"到"自愈+自动扩缩容"的跨越，将人工干预频率降低了90%以上。 自动化运维架构 ┌──────────────────────────────────────────────────────────┐ │ Agent自动化运维体系 │ │ │ │ ┌────────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 健康检测 │ │ 告警 │ │ 自愈 │ │ │ │ Health │ │ Alerts │ │ Self- │ │ │ │ Check │ │ │ │ healing │ │ │ └─────┬──────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────┐ │ │ │ 决策引擎 (Decision Engine) │ │ │ │ - 规则引擎 │ │ │ │ - ML预测模型 │ │ │ │ - 执行计划生成 │ │ │ └─────────────────┬───────────────────────────┘ │ │ │ │ │ ┌───────────┼───────────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌──────────┐ ┌────────────┐ │ │ │Auto- │ │Auto- │ │Auto- │ │ │ │Scaling │ │Remediation│ │Configuration│ │ │ │扩缩容 │ │修复 │ │配置 │ │ │ └─────────┘ └──────────┘ └────────────┘ │ └──────────────────────────────────────────────────────────┘ 自愈机制 健康检查 class HealthChecker: """Agent系统健康检查""" CHECKS = { "llm_service": { "endpoint": "http://llm-service:8080/health", "timeout": 5, "expected_status": 200, "degraded_threshold_ms": 500, }, "tool_services": { "endpoint": "http://tool-service:8080/health", "timeout": 3, "expected_status": 200, }, "vector_db": { "endpoint": "http://qdrant:6333/health", "timeout": 5, }, "redis": { "command": "PING", "expected_response": "PONG", } } async def run_all_checks(self) -> dict: """运行所有健康检查""" results = {} for name, check in self.CHECKS.items(): try: result = await self._run_check(name, check) results[name] = result except Exception as e: results[name] = { "status": "unhealthy", "error": str(e), "timestamp": datetime.now().isoformat() } # 计算整体健康分 healthy_count = sum( 1 for r in results.values() if r["status"] == "healthy" ) health_score = healthy_count / len(results) return { "overall_health": health_score, "status": "healthy" if health_score > 0.8 else "degraded", "checks": results, "timestamp": datetime.now().isoformat() } async def _run_check(self, name: str, check: dict) -> dict: """运行单项检查""" start = time.monotonic() if "endpoint" in check: async with httpx.AsyncClient(timeout=check["timeout"]) as client: response = await client.get(check["endpoint"]) latency_ms = (time.monotonic() - start) * 1000 if response.status_code != check["expected_status"]: return { "status": "unhealthy", "actual_status": response.status_code, "expected_status": check["expected_status"] } status = "healthy" if latency_ms > check.get("degraded_threshold_ms", 1000): status = "degraded" return { "status": status, "latency_ms": latency_ms, "status_code": response.status_code } 自愈动作 class SelfHealingActions: """自愈动作库""" async def restart_service(self, service_name: str) -> dict: """重启服务""" logger.warning(f"Self-healing: restarting {service_name}") try: # 通过K8s API重启 await self.k8s_client.restart_deployment(service_name) # 等待就绪 await self._wait_for_ready(service_name, timeout=120) return { "action": "restart_service", "service": service_name, "success": True } except Exception as e: logger.error(f"Self-healing failed for {service_name}: {e}") return { "action": "restart_service", "service": service_name, "success": False, "error": str(e) } async def clear_cache(self, cache_type: str) -> dict: """清理缓存""" if cache_type == "redis": await self.redis_client.flushdb() elif cache_type == "vector": await self.vector_db.clear_cache() return {"action": "clear_cache", "type": cache_type, "success": True} async def scale_service( self, service_name: str, replicas: int ) -> dict: """扩缩容服务""" current = await self.k8s_client.get_deployment_replicas(service_name) if current == replicas: return {"action": "scale", "changed": False} await self.k8s_client.scale_deployment(service_name, replicas) return { "action": "scale", "service": service_name, "from": current, "to": replicas } async def failover_to_backup(self, service_name: str) -> dict: """故障转移到备用服务""" backup_config = self.backup_configs.get(service_name) if not backup_config: raise ValueError(f"No backup configured for {service_name}") # 切换流量到备用 await self.traffic_router.switch_to_backup( service_name, backup_config["endpoint"] ) return { "action": "failover", "service": service_name, "backup": backup_config["name"] } 自动扩缩容 预测性扩缩容 class PredictiveAutoScaler: """预测性自动扩缩容""" def __init__(self, k8s_client, metrics_client): self.k8s = k8s_client self.metrics = metrics_client self.prediction_model = None # 加载训练好的预测模型 async def run_scaling_loop(self): """扩缩容主循环""" while True: try: # 1. 收集当前指标 current_metrics = await self._collect_metrics() # 2. 预测未来负载 predicted_load = await self._predict_load( current_metrics, horizon_minutes=15 ) # 3. 计算目标副本数 target_replicas = self._calculate_target_replicas( predicted_load ) # 4. 执行扩缩容 await self._apply_scaling(target_replicas) # 5. 等待下一次评估 await asyncio.sleep(60) # 每分钟评估一次 except Exception as e: logger.error(f"Auto-scaling error: {e}") await asyncio.sleep(60) async def _predict_load(self, current: dict, horizon_minutes: int) -> dict: """预测未来负载""" # 使用历史数据 + 当前趋势预测 if self.prediction_model: features = self._extract_prediction_features(current) prediction = await self.prediction_model.predict(features) return prediction else: # 简单线性回归预测 trend = self._calculate_trend(current) predicted_qps = current["qps"] * (1 + trend * horizon_minutes / 60) return {"predicted_qps": max(0, predicted_qps)} def _calculate_target_replicas(self, predicted_load: dict) -> dict: """计算目标副本数""" targets = {} for service, config in self.service_configs.items(): predicted_qps = predicted_load.get("predicted_qps", 0) # 每个副本能处理的QPS qps_per_replica = config.get("qps_per_replica", 10) # 目标副本数（加20%安全余量） target = int(predicted_qps / qps_per_replica * 1.2) # 应用上下限 target = max(config["min_replicas"], target) target = min(config["max_replicas"], target) targets[service] = target return targets 扩缩容策略 class ScalingStrategy: """扩缩容策略""" STRATEGIES = { "conservative": { "scale_up_step": 1, # 每次只加1个副本 "scale_down_step": 1, # 每次只减1个副本 "scale_up_cooldown": 120, # 扩容冷却2分钟 "scale_down_cooldown": 300, # 缩容冷却5分钟 }, "aggressive": { "scale_up_step": 5, "scale_down_step": 2, "scale_up_cooldown": 30, "scale_down_cooldown": 180, }, "predictive": { "lookahead_minutes": 15, "safety_margin": 0.3, # 30%安全余量 } } async def execute_scaling( self, service: str, current: int, target: int, strategy: str = "conservative" ): """执行扩缩容""" config = self.STRATEGIES[strategy] if target > current: # 扩容 step = config["scale_up_step"] new_replicas = min(current + step, target) logger.info( f"Scaling up {service}: {current} -> {new_replicas}" ) await self.k8s.scale_deployment(service, new_replicas) elif target < current: # 缩容——更保守 step = config["scale_down_step"] new_replicas = max(current - step, target) # 检查是否有正在处理的请求 active_requests = await self._get_active_requests(service) if active_requests > 0: logger.info( f"Postponing scale down for {service}: " f"{active_requests} active requests" ) return logger.info( f"Scaling down {service}: {current} -> {new_replicas}" ) await self.k8s.scale_deployment(service, new_replicas) 故障预测 class FailurePredictor: """故障预测器""" async def predict_failures(self) -> list: """预测可能发生的故障""" predictions = [] # 1. 基于指标的预测 metrics_anomalies = await self._detect_metric_anomalies() for anomaly in metrics_anomalies: predictions.append({ "type": "metric_anomaly", "service": anomaly["service"], "probability": anomaly["probability"], "description": anomaly["description"], "recommended_action": anomaly["action"] }) # 2. 基于日志的预测 log_patterns = await self._analyze_log_patterns() for pattern in log_patterns: if pattern["risk_score"] > 0.7: predictions.append({ "type": "log_pattern", "pattern": pattern["pattern"], "probability": pattern["risk_score"], "description": f"Detected pattern: {pattern['description']}" }) # 3. 基于依赖健康的预测 dependency_health = await self._check_dependency_health() for dep in dependency_health: if dep["health_score"] < 0.5: predictions.append({ "type": "dependency", "dependency": dep["name"], "probability": 1 - dep["health_score"], "description": f"Dependency {dep['name']} is unhealthy" }) return sorted(predictions, key=lambda p: p["probability"], reverse=True) AIOps实践 class AIOpsEngine: """AIOps引擎""" async def analyze_incident(self, incident: dict) -> dict: """AI辅助事故分析""" # 1. 收集相关日志/指标/Trace context = await self._gather_incident_context(incident) # 2. 使用LLM分析根因 analysis = await self.llm.analyze( prompt=f""" Analyze this incident and identify the root cause: Incident: {incident} Context: {context} Provide: 1. Most likely root cause 2. Contributing factors 3. Recommended actions 4. Prevention measures """ ) # 3. 从历史事故中找相似案例 similar_incidents = await self._find_similar_incidents(incident) return { "incident_id": incident["id"], "root_cause_analysis": analysis["root_cause"], "recommended_actions": analysis["actions"], "similar_incidents": similar_incidents, "confidence": analysis["confidence"] } async def generate_runbook(self, incident_type: str) -> str: """自动生成Runbook""" return await self.llm.generate( prompt=f""" Generate a detailed runbook for handling {incident_type} incidents. Include: 1. Detection steps 2. Diagnosis procedures 3. Resolution steps 4. Verification steps 5. Post-mortem template """ ) 总结 Agent自动化运维的核心目标是"让系统自己管理自己"。自愈机制通过健康检查+修复动作的组合，能够自动处理80%以上的常见故障。预测性扩缩容通过提前预判负载变化，避免了响应式扩缩容的滞后性。AIOps则通过AI辅助事故分析和Runbook生成，显著提升了排障效率。 ...

引言 Agent系统的监控告警比传统应用复杂一个量级——除了需要监控CPU、内存、延迟等基础设施指标外，还需要监控Token消耗、工具成功率、幻觉率、安全违规率等Agent特有指标。一个没有完善监控的Agent系统就像盲飞——出了问题不知道哪里出了问题，没出问题不知道什么时候会出问题。 2026年，Prometheus + Grafana + AlertManager已成为Agent监控告警的事实标准，但Agent系统需要在此基础上构建专门的监控体系。 指标体系设计 四层指标架构 ┌─────────────────────────────────────────────────────┐ │ Layer 4: Business Metrics │ │ 用户满意度、任务完成率、对话质量评分 │ ├─────────────────────────────────────────────────────┤ │ Layer 3: Agent Metrics │ │ Token消耗、工具调用成功率、幻觉率、安全违规率 │ ├─────────────────────────────────────────────────────┤ │ Layer 2: Application Metrics │ │ 请求QPS、响应延迟、错误率、并发会话数 │ ├─────────────────────────────────────────────────────┤ │ Layer 1: Infrastructure Metrics │ │ CPU、内存、GPU利用率、磁盘IO、网络流量 │ └─────────────────────────────────────────────────────┘ Agent核心指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # ===== Layer 3: Agent特有指标 ===== # Token消耗 token_usage = Counter( "agent_token_total", "Total tokens consumed", ["tenant_id", "model", "type"] # type: input/output ) # 工具调用 tool_calls = Counter( "agent_tool_calls_total", "Total tool calls", ["tool_name", "status"] # status: success/failed/timeout ) tool_latency = Histogram( "agent_tool_latency_seconds", "Tool execution latency", ["tool_name"], buckets=[0.1, 0.5, 1, 5, 10, 30, 60] ) # Agent质量 hallucination_rate = Gauge( "agent_hallucination_rate", "Hallucination rate (rolling 1h)", ["model"] ) safety_violations = Counter( "agent_safety_violations_total", "Safety violations detected", ["type", "severity"] ) # 循环检测 cycle_detections = Counter( "agent_cycle_detections_total", "Cycle detections", ["cycle_type", "resolution"] # resolution: broken/escalated ) # 会话指标 active_sessions = Gauge( "agent_active_sessions", "Active sessions", ["tenant_id"] ) session_duration = Histogram( "agent_session_duration_seconds", "Session duration", buckets=[10, 30, 60, 120, 300, 600, 1200] ) # 路由决策 routing_decisions = Counter( "agent_routing_decisions_total", "Routing decisions", ["source_model", "target_model", "reason"] ) Prometheus配置 # prometheus.yml global: scrape_interval: 15s evaluation_interval: 15s rule_files: - "agent_alerts.yml" scrape_configs: # Agent应用指标 - job_name: "agent-service" metrics_path: /metrics static_configs: - targets: ["agent-service:9090"] labels: service: "agent" # LLM推理服务 - job_name: "llm-inference" metrics_path: /metrics static_configs: - targets: ["llm-inference:9090"] # 工具执行服务 - job_name: "tool-executor" metrics_path: /metrics static_configs: - targets: ["tool-executor-0:9090", "tool-executor-1:9090"] 告警规则 # agent_alerts.yml groups: - name: agent_infra_alerts rules: # 高错误率 - alert: AgentHighErrorRate expr: | sum(rate(agent_requests_total{status="error"}[5m])) by (service) / sum(rate(agent_requests_total[5m])) by (service) > 0.05 for: 2m labels: severity: critical team: agent-platform annotations: summary: "Agent error rate > 5%" description: "{{ $labels.service }} error rate is {{ $value | humanizePercentage }}" # P99延迟过高 - alert: AgentHighLatency expr: | histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m]) ) > 5 for: 5m labels: severity: warning annotations: summary: "Agent P99 latency > 5s" - name: agent_quality_alerts rules: # 幻觉率过高 - alert: AgentHighHallucination expr: agent_hallucination_rate > 0.05 for: 10m labels: severity: warning team: agent-quality annotations: summary: "Hallucination rate > 5% for {{ $labels.model }}" # 安全违规 - alert: AgentSafetyViolation expr: increase(agent_safety_violations_total[1h]) > 0 labels: severity: critical team: security annotations: summary: "Safety violation detected" # 循环检测频繁 - alert: AgentFrequentCycles expr: | increase(agent_cycle_detections_total[1h]) > 10 for: 5m labels: severity: warning annotations: summary: "Frequent cycle detections (>10/hour)" - name: agent_cost_alerts rules: # Token消耗异常 - alert: AgentTokenSpike expr: | rate(agent_token_total[5m]) > 2 * avg_over_time(rate(agent_token_total[5m])[1h:5m]) for: 10m labels: severity: warning team: agent-platform annotations: summary: "Token consumption spiked 2x above average" # 工具调用失败率 - alert: ToolFailureRate expr: | sum(rate(agent_tool_calls_total{status="failed"}[5m])) by (tool_name) / sum(rate(agent_tool_calls_total[5m])) by (tool_name) > 0.1 for: 5m labels: severity: warning annotations: summary: "Tool {{ $labels.tool_name }} failure rate > 10%" 告警路由与通知 class AlertRouter: """告警路由器""" ROUTING_RULES = { "critical": { "channels": ["pagerduty", "slack:#oncall", "sms"], "escalation_delay_min": 5, "escalation_target": "team-lead", }, "warning": { "channels": ["slack:#alerts"], "escalation_delay_min": 30, "escalation_target": "secondary-oncall", }, "info": { "channels": ["slack:#monitoring"], "escalation_delay_min": None, "escalation_target": None, } } async def handle_alert(self, alert: dict): """处理告警""" severity = alert["labels"]["severity"] rule = self.ROUTING_RULES[severity] # 告警去重 if await self._is_duplicate(alert): logger.debug(f"Duplicate alert suppressed: {alert['fingerprint']}") return # 告警分组 group_key = self._get_group_key(alert) group = await self._get_or_create_group(group_key) group.add_alert(alert) # 发送通知 for channel in rule["channels"]: await self._send_notification(channel, group) # 设置升级定时器 if rule["escalation_delay_min"]: asyncio.create_task( self._schedule_escalation( group, rule["escalation_delay_min"], rule["escalation_target"] ) ) 告警治理 class AlertGovernance: """告警治理——防止告警风暴""" def __init__(self): self.suppression_rules = [] self.rate_limits = {} def should_send(self, alert: dict) -> bool: """判断告警是否应该发送""" # 1. 维护窗口抑制 if self._in_maintenance_window(alert): return False # 2. 依赖抑制——如果上游告警活跃，抑制下游 if self._suppressed_by_dependency(alert): return False # 3. 频率限制——同一告警5分钟内只发一次 alert_key = alert["fingerprint"] if alert_key in self.rate_limits: last_sent = self.rate_limits[alert_key] if (datetime.now() - last_sent).total_seconds() < 300: return False # 4. 告警噪音评分 noise_score = self._calculate_noise_score(alert) if noise_score < 0.3: return False self.rate_limits[alert_key] = datetime.now() return True Grafana仪表板 { "dashboard": { "title": "Agent System Overview", "panels": [ { "title": "Request Rate & Error Rate", "targets": [ { "expr": "sum(rate(agent_requests_total[5m]))", "legendFormat": "QPS" }, { "expr": "sum(rate(agent_requests_total{status=\"error\"}[5m])) / sum(rate(agent_requests_total[5m]))", "legendFormat": "Error Rate" } ] }, { "title": "Token Consumption by Model", "targets": [ { "expr": "sum(rate(agent_token_total[5m])) by (model)", "legendFormat": "{{model}}" } ] }, { "title": "Tool Success Rate", "targets": [ { "expr": "sum(rate(agent_tool_calls_total{status=\"success\"}[5m])) by (tool_name) / sum(rate(agent_tool_calls_total[5m])) by (tool_name)", "legendFormat": "{{tool_name}}" } ] }, { "title": "Active Sessions & Concurrency", "targets": [ { "expr": "agent_active_sessions", "legendFormat": "{{tenant_id}}" } ] } ] } } SLI/SLO定义 # Agent系统SLO定义 slo: availability: target: 99.9% window: 30d query: | 1 - (sum(rate(agent_requests_total{status="error"}[5m])) / sum(rate(agent_requests_total[5m]))) latency_p99: target: 2000ms window: 7d query: | histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m])) quality_score: target: 0.85 window: 7d query: | avg(agent_response_quality_score) safety: target: 99.99% window: 30d query: | 1 - (increase(agent_safety_violations_total[30d]) / sum(increase(agent_requests_total[30d]))) 总结 Agent系统的监控告警需要覆盖从基础设施到业务质量的四个层次。指标设计要全面但不过载，告警规则要精准且有层次，通知路由要高效且不产生噪音。告警治理是长期工作——定期回顾告警有效性，淘汰无用告警，优化有用告警。 ...

引言 2026年，AIOps（AI for IT Operations）已从概念走向规模化落地。根据Gartner统计，全球财富500强中62%的企业已部署至少一个AIOps场景，平均MTTR（平均故障恢复时间）降低55%。随着LLM与运维场景的深度融合，AIOps正从"规则引擎+机器学习"进化为"LLM驱动的智能运维Agent"。本文将系统介绍AIOps的实践路径。 一、AIOps架构 1.1 总体架构 ┌─────────────────────────────────────────┐ │ 交互层（ChatOps/可视化） │ ├─────────────────────────────────────────┤ │ 智能决策层（LLM Agent） │ │ ┌─────────┐ ┌──────────┐ ┌──────────┐ │ │ │根因分析 │ │自动修复 │ │容量规划 │ │ │ └─────────┘ └──────────┘ └──────────┘ │ ├─────────────────────────────────────────┤ │ 分析层（ML/DL模型） │ │ ┌─────────┐ ┌──────────┐ ┌──────────┐ │ │ │异常检测 │ │日志分析 │ │关联分析 │ │ │ └─────────┘ └──────────┘ └──────────┘ │ ├─────────────────────────────────────────┤ │ 数据层（数据湖/流处理） │ │ Metrics │ Logs │ Traces │ Events │ Topology│ └─────────────────────────────────────────┘ 1.2 核心组件 组件 功能 技术选型 数据采集 指标/日志/链路数据 Prometheus, Fluentd, OpenTelemetry 数据存储 时序+全文+图 VictoriaMetrics, Elasticsearch, Neo4j 实时处理 流式计算 Flink, Kafka Streams ML平台 模型训练/推理 MLflow, KServe LLM引擎 自然语言理解/推理 GPT-4o / Llama 3.1 405B 可视化 仪表板/告警 Grafana, Kibana ChatOps 交互入口 Slack/飞书/钉钉 Bot 二、智能监控 2.1 传统监控 vs AI监控 维度 传统监控 AI监控 告警规则 人工设定阈值 自适应基线+动态阈值 告警粒度 单指标告警 多指标关联+场景告警 误报率 30-50% 5-10% 告警量 高（告警风暴） 低（智能合并） 响应速度 人工判断 秒级自动响应 2.2 异常检测实践 2026年主流的异常检测方案采用多模型融合策略： ...