AI Agent调试方法论：日志、追踪与评估闭环

第一层：日志（Logs）—— 发生了什么
第二层：追踪（Traces）—— 为什么发生
第三层：评估（Evaluation）—— 发生得对不对

import json
import time
import uuid
from enum import Enum
from datetime import datetime

class LogLevel(Enum):
    DEBUG = "DEBUG"
    INFO = "INFO"
    WARN = "WARN"
    ERROR = "ERROR"

class AgentLogger:
    """生产级Agent结构化日志"""
    
    def __init__(self, agent_name):
        self.agent_name = agent_name
    
    def log(self, level, event, **fields):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": level.value,
            "agent": self.agent_name,
            "event": event,
            "trace_id": fields.get("trace_id"),
            "span_id": fields.get("span_id"),
            **fields
        }
        print(json.dumps(entry, ensure_ascii=False, default=str))

# 使用示例
logger = AgentLogger("research_agent")
logger.log(LogLevel.INFO, "tool_call",
    trace_id="tr_abc123",
    span_id="sp_001",
    tool_name="web_search",
    tool_input={"query": "2026 AI芯片市场"},
    tool_output={"results_count": 5},
    duration_ms=1200,
    tokens_used=150
)

class AgentEventTypes:
    """Agent生命周期中的关键事件"""
    
    # 规划阶段
    PLAN_CREATED = "plan_created"           # Agent制定了执行计划
    PLAN_REVISED = "plan_revised"           # 计划被修改
    GOAL_DECOMPOSED = "goal_decomposed"     # 目标被分解
    
    # 执行阶段
    TOOL_CALL_START = "tool_call_start"     # 工具调用开始
    TOOL_CALL_END = "tool_call_end"         # 工具调用结束
    TOOL_CALL_ERROR = "tool_call_error"     # 工具调用失败
    TOOL_CALL_RETRY = "tool_call_retry"     # 工具调用重试
    
    # 推理阶段
    LLM_CALL_START = "llm_call_start"       # LLM调用开始
    LLM_CALL_END = "llm_call_end"           # LLM调用结束
    REASONING_STEP = "reasoning_step"       # 推理步骤
    DECISION_MADE = "decision_made"         # 做出决策
    
    # 状态管理
    CONTEXT_UPDATED = "context_updated"     # 上下文更新
    MEMORY_READ = "memory_read"             # 读取记忆
    MEMORY_WRITE = "memory_write"           # 写入记忆
    
    # 错误与异常
    HALLUCINATION_DETECTED = "hallucination_detected"
    LOOP_DETECTED = "loop_detected"         # 检测到循环
    BUDGET_EXCEEDED = "budget_exceeded"     # 预算超限
    MAX_STEPS_REACHED = "max_steps_reached" # 达到最大步数

def analyze_agent_logs(logs):
    """分析Agent日志，识别常见问题模式"""
    
    patterns = {
        # 模式1：工具调用循环
        "tool_loop": detect_tool_loops(logs),
        
        # 模式2：LLM调用失败率
        "llm_failure_rate": calculate_llm_failure_rate(logs),
        
        # 模式3：Token消耗异常
        "token_anomaly": detect_token_anomalies(logs),
        
        # 模式4：延迟热点
        "latency_hotspots": find_latency_hotspots(logs),
        
        # 模式5：幻觉信号
        "hallucination_signals": detect_hallucination_signals(logs),
    }
    
    return patterns

def detect_tool_loops(logs):
    """检测工具调用循环——Agent反复调用同一工具"""
    tool_calls = [l for l in logs if l["event"] == "tool_call_end"]
    
    loops = []
    window = 5  # 检查窗口
    for i in range(len(tool_calls) - window):
        window_calls = tool_calls[i:i+window]
        tool_names = [c["tool_name"] for c in window_calls]
        
        # 如果同一工具在窗口内被调用3次以上
        from collections import Counter
        counts = Counter(tool_names)
        for tool, count in counts.items():
            if count >= 3:
                loops.append({
                    "tool": tool,
                    "count": count,
                    "window_start": window_calls[0]["timestamp"],
                    "severity": "high" if count >= 4 else "medium",
                })
    
    return loops

class AgentTracer:
    """Agent全链路追踪系统"""
    
    def __init__(self):
        self.spans = []
    
    def start_trace(self, agent_name, user_input, context=None):
        """开始一个新的追踪"""
        trace_id = f"tr_{uuid.uuid4().hex[:12]}"
        return {
            "trace_id": trace_id,
            "agent_name": agent_name,
            "user_input": user_input,
            "context": context,
            "start_time": time.time(),
            "spans": [],
        }
    
    def start_span(self, trace, span_name, span_type, parent_id=None):
        """开始一个Span"""
        span_id = f"sp_{uuid.uuid4().hex[:8]}"
        span = {
            "span_id": span_id,
            "parent_id": parent_id,
            "name": span_name,
            "type": span_type,  # llm, tool, reasoning, memory
            "start_time": time.time(),
            "inputs": None,
            "outputs": None,
            "error": None,
        }
        trace["spans"].append(span)
        return span_id
    
    def end_span(self, trace, span_id, outputs=None, error=None):
        """结束一个Span"""
        for span in trace["spans"]:
            if span["span_id"] == span_id:
                span["end_time"] = time.time()
                span["duration_ms"] = (span["end_time"] - span["start_time"]) * 1000
                span["outputs"] = outputs
                span["error"] = error
                break

# 追踪使用示例
tracer = AgentTracer()

trace = tracer.start_trace("research_agent", "分析2026年AI芯片市场")
root_span = tracer.start_span(trace, "research_task", "root")

# 规划阶段
plan_span = tracer.start_span(trace, "planning", "reasoning", root_span)
plan = agent.plan("分析2026年AI芯片市场")
tracer.end_span(trace, plan_span, outputs={"plan": plan})

# 执行阶段
for step in plan.steps:
    if step.type == "tool_call":
        tool_span = tracer.start_span(trace, f"tool:{step.tool}", "tool", root_span)
        result = agent.call_tool(step.tool, step.params)
        tracer.end_span(trace, tool_span, outputs=result)

tracer.end_span(trace, root_span, outputs={"answer": agent.final_answer})

Trace: tr_abc123 | research_agent | 总耗时: 12.3s
│
├── [reasoning] planning (0.8s)
│   └── 计划: 1.搜索数据 2.分析数据 3.生成报告
│
├── [tool] web_search (1.2s)
│   ├── query: "2026 AI芯片市场份额"
│   └── results: 5条
│
├── [tool] web_search (1.1s)
│   ├── query: "NVIDIA Blackwell vs 国产芯片"
│   └── results: 8条
│
├── [llm] data_analysis (3.5s)
│   ├── input_tokens: 2048
│   ├── output_tokens: 1024
│   └── cost: $0.03
│
├── [tool] write_file (0.3s)
│   └── output: report.md
│
└── [llm] final_answer (5.4s)
    ├── input_tokens: 4096
    ├── output_tokens: 2048
    └── cost: $0.06

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

class OTelAgentTracer:
    """与OpenTelemetry集成的Agent追踪"""
    
    def __init__(self, service_name="ai_agent"):
        provider = TracerProvider()
        trace.set_tracer_provider(provider)
        self.tracer = trace.get_tracer(service_name)
    
    async def trace_agent_run(self, agent, user_input):
        """追踪完整的Agent执行"""
        with self.tracer.start_as_current_span("agent_run") as root:
            root.set_attribute("agent.name", agent.name)
            root.set_attribute("agent.input", user_input)
            
            async for event in agent.run(user_input):
                if event.type == "tool_call":
                    with self.tracer.start_as_current_span(
                        f"tool:{event.tool_name}"
                    ) as tool_span:
                        tool_span.set_attribute("tool.input", str(event.tool_input))
                        tool_span.set_attribute("tool.output", str(event.tool_output))
                        tool_span.set_attribute("tool.duration_ms", event.duration_ms)
                
                elif event.type == "llm_call":
                    with self.tracer.start_as_current_span(
                        f"llm:{event.purpose}"
                    ) as llm_span:
                        llm_span.set_attribute("llm.input_tokens", event.input_tokens)
                        llm_span.set_attribute("llm.output_tokens", event.output_tokens)
                        llm_span.set_attribute("llm.model", event.model)

class AgentEvaluator:
    """Agent质量评估框架"""
    
    def evaluate(self, agent, test_cases):
        """
        在测试集上评估Agent表现
        test_cases: [{"input": "...", "expected": "...", "criteria": [...]}]
        """
        results = []
        
        for case in test_cases:
            # 运行Agent
            agent_response = agent.run(case["input"])
            
            # 多维度评估
            scores = {
                "任务完成度": self._eval_task_completion(
                    agent_response, case["expected"]
                ),
                "工具使用效率": self._eval_tool_efficiency(
                    agent_response.trace
                ),
                "推理质量": self._eval_reasoning_quality(
                    agent_response.trace, case["input"]
                ),
                "安全性": self._eval_safety(agent_response),
                "延迟": self._eval_latency(agent_response.trace),
                "成本": self._eval_cost(agent_response.trace),
            }
            
            # 综合评分
            scores["总分"] = sum(scores.values()) / len(scores)
            results.append({
                "input": case["input"],
                "response": agent_response,
                "scores": scores,
            })
        
        return self._generate_report(results)
    
    def _eval_tool_efficiency(self, trace):
        """评估工具使用效率：是否用了不必要的工具？"""
        tool_calls = [s for s in trace.spans if s.type == "tool"]
        
        # 评分标准
        if len(tool_calls) == 0:
            return 0.0  # 没用任何工具（可能直接回答了）
        
        # 检查是否有冗余调用
        tool_names = [c.name for c in tool_calls]
        unique_tools = set(tool_names)
        redundancy_ratio = 1 - len(unique_tools) / len(tool_names)
        
        # 检查是否有失败重试
        failed_calls = [c for c in tool_calls if c.error]
        retry_penalty = len(failed_calls) * 0.1
        
        score = 1.0 - redundancy_ratio - retry_penalty
        return max(0.0, min(1.0, score))
    
    def _eval_reasoning_quality(self, trace, user_input):
        """评估推理质量：决策是否合理？"""
        reasoning_steps = [s for s in trace.spans if s.type == "reasoning"]
        
        # 使用LLM-as-Judge评估推理质量
        judge_prompt = f"""
        用户目标：{user_input}
        
        Agent的推理步骤：
        {self._format_reasoning_steps(reasoning_steps)}
        
        请评估Agent的推理质量（0-10分）：
        1. 目标理解是否准确？
        2. 计划是否合理？
        3. 决策是否有依据？
        4. 是否有明显的逻辑错误？
        """
        
        score = llm_judge.evaluate(judge_prompt)
        return score / 10.0

class OnlineAgentMonitor:
    """Agent在线质量监控"""
    
    def __init__(self):
        self.metrics = {
            "success_rate": SlidingWindowMetric(window=100),
            "avg_steps": SlidingWindowMetric(window=100),
            "avg_cost": SlidingWindowMetric(window=100),
            "avg_latency": SlidingWindowMetric(window=100),
            "tool_error_rate": SlidingWindowMetric(window=100),
            "user_satisfaction": SlidingWindowMetric(window=50),
        }
    
    def record_execution(self, result):
        """记录每次Agent执行"""
        self.metrics["success_rate"].update(result.success)
        self.metrics["avg_steps"].update(result.step_count)
        self.metrics["avg_cost"].update(result.total_cost)
        self.metrics["avg_latency"].update(result.total_duration_ms)
        self.metrics["tool_error_rate"].update(result.has_tool_error)
        
        # 检查告警条件
        self._check_alerts()
    
    def _check_alerts(self):
        """告警检查"""
        alerts = []
        
        if self.metrics["success_rate"].value() < 0.85:
            alerts.append({
                "level": "critical",
                "message": "Agent成功率低于85%",
                "value": self.metrics["success_rate"].value(),
            })
        
        if self.metrics["avg_steps"].value() > 15:
            alerts.append({
                "level": "warning",
                "message": "平均执行步数过多，可能存在循环",
                "value": self.metrics["avg_steps"].value(),
            })
        
        if self.metrics["tool_error_rate"].value() > 0.2:
            alerts.append({
                "level": "warning",
                "message": "工具调用错误率超过20%",
                "value": self.metrics["tool_error_rate"].value(),
            })
        
        for alert in alerts:
            self._send_alert(alert)

def diagnose_agent_loop(trace):
    """诊断Agent循环问题"""
    steps = trace.spans
    
    # 检测重复模式
    for window_size in [3, 5, 7]:
        for i in range(len(steps) - window_size * 2):
            pattern1 = [s.name for s in steps[i:i+window_size]]
            pattern2 = [s.name for s in steps[i+window_size:i+window_size*2]]
            
            if pattern1 == pattern2:
                return {
                    "diagnosis": "agent_loop",
                    "pattern": pattern1,
                    "repeated_count": len(steps[i:]) // window_size,
                    "root_cause": analyze_loop_cause(steps[i:i+window_size]),
                    "fix": "增加循环检测机制或调整Agent的stop条件",
                }
    
    return {"diagnosis": "no_loop_detected"}

症状：Agent选择了不合适的工具
诊断：
  1. 检查工具描述是否清晰
  2. 检查工具选择时的Prompt
  3. 检查是否有工具描述冲突
修复：
  1. 优化工具描述，增加使用场景说明
  2. 在系统Prompt中加入工具选择示例
  3. 实现工具选择验证层

症状：任务未完成但Agent停止执行
诊断：
  1. 检查stop条件是否触发
  2. 检查LLM是否生成了停止信号
  3. 检查是否达到token限制
修复：
  1. 调整stop条件
  2. 增加完成度检查
  3. 实现任务完成验证

def diagnose_context_loss(trace):
    """诊断Agent上下文丢失问题"""
    context_lengths = []
    for span in trace.spans:
        if span.type == "llm":
            context_lengths.append({
                "step": span.name,
                "input_tokens": span.inputs.get("token_count", 0),
                "context_included": span.inputs.get("includes_context", False),
            })
    
    # 检查上下文是否被截断
    if len(context_lengths) > 3:
        early_context = context_lengths[0]["input_tokens"]
        late_context = context_lengths[-1]["input_tokens"]
        
        if late_context < early_context * 0.3:
            return {
                "diagnosis": "context_truncation",
                "early_tokens": early_context,
                "late_tokens": late_context,
                "fix": "实现上下文压缩或摘要策略，保留关键信息",
            }
    
    return {"diagnosis": "no_context_loss"}