引言 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系统故障排查需要系统化的方法论——从收集症状开始，通过复现问题、分析日志、追踪链路，最终定位根因。关键在于将模糊的"回复不对"拆解成可度量的指标异常，再通过对指标的分析定位到具体的组件和行为。
...
