引言

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

核心原则：好的排障不是"猜对了"，而是"推理对了"。每一次排障都应该记录假设、证据和验证过程，形成知识库，让下一次排障更快。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。