AI Agent是天然的"黑盒"——它做了什么、为什么这么做、为什么出错了,这些问题在生产环境中极难回答。一个完善的日志与可观测性体系,是把黑盒变白盒的关键。本文将系统介绍AI Agent的日志设计与故障排查方法论。

一、Agent可观测性的特殊挑战

1.1 与传统服务日志的区别

传统服务的日志是线性的:请求A → 处理 → 响应A。但Agent的执行是非线性的:

用户输入 → 意图理解 → 规划 → 工具调用1 → 工具调用2 → 反思 → 
         → 修正规划 → 工具调用3 → 总结 → 输出

每一步都可能分叉、回退、重试。传统的"一条请求一条日志"模式无法捕捉这种复杂流程。

1.2 核心观测维度

L1: 基础设施层 — GPU利用率、内存、网络
L2: API服务层 — 请求量、延迟、错误率
L3: Agent逻辑层 — 意图、规划、工具调用、反思
L4: LLM推理层 — prompt内容、生成内容、token消耗
L5: 业务效果层 — 任务完成率、用户满意度

大部分团队只关注L1和L2,但Agent故障的根因往往在L3和L4。

二、结构化日志设计

2.1 Trace-Tree模型

Agent的执行过程天然是树状结构,应当用Trace-Tree而非线性日志来记录:

@dataclass
class AgentTrace:
    trace_id: str           # 全局追踪ID
    session_id: str         # 会话ID
    root_span: AgentSpan    # 根span

@dataclass  
class AgentSpan:
    span_id: str
    parent_id: str
    name: str               # e.g., "intent_understanding", "tool_call"
    span_type: str          # think / act / observe / reflect
    input: dict
    output: dict
    start_time: float
    end_time: float
    status: str             # success / error / timeout
    metadata: dict          # 额外信息
    children: List[AgentSpan]

2.2 关键Span类型

class SpanTypes:
    INTENT = "intent"           # 意图理解
    PLANNING = "planning"       # 规划
    TOOL_CALL = "tool_call"     # 工具调用
    LLM_CALL = "llm_call"       # LLM推理
    REFLECTION = "reflection"   # 反思
    DELEGATION = "delegation"   # 委托子Agent
    OUTPUT = "output"           # 最终输出

2.3 日志记录实现

class AgentLogger:
    def __init__(self):
        self.tracer = DistributedTracer()
    
    @contextmanager
    def span(self, name, span_type, parent_id=None):
        span = AgentSpan(
            span_id=generate_id(),
            parent_id=parent_id,
            name=name,
            span_type=span_type,
            start_time=time.time(),
            input={},
            output={},
            status="running",
            metadata={},
            children=[]
        )
        try:
            yield span
            span.status = "success"
        except Exception as e:
            span.status = "error"
            span.metadata["error"] = str(e)
            span.metadata["traceback"] = traceback.format_exc()
            raise
        finally:
            span.end_time = time.time()
            self.tracer.report(span)
    
    def log_llm_call(self, span, prompt, response, model, tokens):
        """记录LLM调用的详细信息"""
        span.metadata["llm"] = {
            "model": model,
            "prompt_tokens": tokens["prompt"],
            "completion_tokens": tokens["completion"],
            "prompt_hash": hash(prompt[:100]),  # 隐私保护
            "response_length": len(response),
            "latency_ms": span.duration_ms
        }
    
    def log_tool_call(self, span, tool_name, args, result, success):
        """记录工具调用"""
        span.metadata["tool"] = {
            "name": tool_name,
            "args_hash": hash(str(args)),  # 参数指纹
            "result_size": len(str(result)),
            "success": success
        }

2.4 完整Trace示例

{
  "trace_id": "trace_abc123",
  "session_id": "sess_xyz",
  "duration_ms": 4500,
  "status": "success",
  "spans": [
    {
      "name": "intent_understanding",
      "type": "intent",
      "duration_ms": 320,
      "input": {"user_message": "帮我查下最近的报销进度"},
      "output": {"intent": "query_reimbursement", "entities": {}},
      "children": [
        {
          "name": "llm_call",
          "type": "llm_call",
          "duration_ms": 310,
          "metadata": {
            "model": "gpt-4-turbo",
            "prompt_tokens": 850,
            "completion_tokens": 45
          }
        }
      ]
    },
    {
      "name": "planning",
      "type": "planning",
      "duration_ms": 280,
      "output": {"plan": ["call_finance_api", "summarize_result"]}
    },
    {
      "name": "tool_call:finance_api",
      "type": "tool_call",
      "duration_ms": 1200,
      "metadata": {
        "tool": "finance_api",
        "args": {"user_id": "***", "date_range": "30d"},
        "success": true
      }
    },
    {
      "name": "llm_call:summarize",
      "type": "llm_call", 
      "duration_ms": 890,
      "metadata": {
        "model": "gpt-4-turbo",
        "prompt_tokens": 1200,
        "completion_tokens": 180
      }
    }
  ]
}

三、常见故障模式与排查

3.1 意图误判

症状:Agent执行了正确的工具但回答了错误的问题

排查路径

1. 查看intent span的output → Agent理解的意图是什么?
2. 对比用户原始输入 → 意图理解是否正确?
3. 查看LLM call的完整prompt → 上下文是否足够?
4. 检查是否有歧义 → 用户表达是否模糊?

常见根因

  • System prompt中缺少该意图的描述
  • 多轮对话中上下文丢失
  • 用户使用了非标准表达

3.2 工具调用循环

症状:Agent反复调用同一工具,无法退出

排查路径

1. 统计同一工具的调用次数 → 是否超过阈值?
2. 查看每次工具调用的输入输出 → 参数是否在变化?
3. 查看反思span → Agent是否意识到自己在循环?
4. 查看LLM的完整输出 → 是否有"break"指令但被忽略?

修复方案

class LoopDetector:
    def __init__(self, max_repeat=3):
        self.max_repeat = max_repeat
    
    def check(self, trace: AgentTrace):
        tool_calls = [s for s in trace.all_spans() if s.span_type == "tool_call"]
        
        # 检测重复工具调用
        tool_counts = Counter(s.name for s in tool_calls)
        for tool, count in tool_counts.items():
            if count > self.max_repeat:
                # 检查输入是否相同
                inputs = [s.input for s in tool_calls if s.name == tool]
                if len(set(str(i) for i in inputs)) == 1:
                    return LoopDetected(
                        tool=tool, count=count,
                        suggestion="相同输入重复调用,可能是LLM未理解工具返回值"
                    )
        return None

3.3 幻觉输出

症状:Agent输出了不基于工具返回结果的内容

排查路径

1. 对比output span和最后一个tool_call span的输出
2. Agent是否引用了不存在的数据
3. 查看LLM prompt中是否包含了工具返回结果
4. 模型是否"创造性"地补充了未提供的信息?

3.4 超时故障

症状:请求处理时间超过阈值

排查路径

1. 查看Trace Tree中哪个span耗时最长
2. LLM call慢LLM API问题)还是tool call慢(外部服务问题)?
3. 是否有retry导致的重复调用
4. 是否有不必要的规划轮次?

四、日志分析工具链

4.1 实时分析

class AgentLogAnalyzer:
    def __init__(self):
        self.es = Elasticsearch()  # 日志存储
        self.grafana = GrafanaAPI()
    
    async def realtime_stats(self):
        """实时统计关键指标"""
        return {
            "active_traces": await self.count_active_traces(),
            "avg_duration": await self.avg_duration(window="5m"),
            "error_rate": await self.error_rate(window="5m"),
            "tool_failure_rate": await self.tool_failure_rate(),
            "llm_token_consumption": await self.token_consumption("1h"),
            "top_errors": await self.top_errors(limit=10)
        }
    
    async def slow_traces(self, threshold_ms=5000, limit=20):
        """找出慢请求"""
        return await self.es.query(
            filter={"duration_ms": {"gt": threshold_ms}},
            sort="-duration_ms",
            limit=limit
        )

4.2 可视化Trace树

将Trace Tree可视化为流程图,是调试Agent最直观的方式:

[Intent] ──320ms──→ "query_reimbursement" ✅
    └── [LLM] ──310ms──→ GPT-4 (850→45 tokens) ✅
[Planning] ──280ms──→ ["call_finance_api", "summarize"] ✅
    └── [LLM] ──270ms──→ GPT-4 (850→30 tokens) ✅
[Tool: finance_api] ──1200ms──→ ⚠️ SLOW (threshold: 1000ms)
[LLM: summarize] ──890ms──→ 180 tokens output ✅
[Output] ──50ms──→ "您的报销审批中..." ✅

Total: 4500ms | Status: SUCCESS (with warnings)

4.3 异常检测

class AnomalyDetector:
    async def detect(self, trace: AgentTrace):
        anomalies = []
        
        # 1. 异常长的LLM调用
        for span in trace.llm_spans():
            if span.duration_ms > 5000:
                anomalies.append(f"LLM调用异常慢: {span.duration_ms}ms")
        
        # 2. 工具调用失败率异常
        tool_spans = trace.tool_spans()
        if tool_spans:
            failure_rate = sum(1 for s in tool_spans if s.status == "error") / len(tool_spans)
            if failure_rate > 0.5:
                anomalies.append(f"工具调用失败率过高: {failure_rate:.0%}")
        
        # 3. 规划步骤过多
        planning_spans = [s for s in trace.all_spans() if s.span_type == "planning"]
        if len(planning_spans) > 3:
            anomalies.append(f"规划次数过多: {len(planning_spans)},可能有循环")
        
        # 4. Token消耗异常
        total_tokens = sum(s.metadata.get("llm", {}).get("prompt_tokens", 0) 
                          for s in trace.llm_spans())
        if total_tokens > 50000:
            anomalies.append(f"Token消耗异常高: {total_tokens}")
        
        return anomalies

五、隐私与安全

5.1 敏感信息脱敏

Agent处理的用户输入可能包含敏感信息,日志记录前必须脱敏:

class LogSanitizer:
    PATTERNS = {
        "phone": (r'\b1[3-9]\d{9}\b', '[PHONE]'),
        "email": (r'\b[\w.-]+@[\w.-]+\.\w+\b', '[EMAIL]'),
        "id_card": (r'\b\d{17}[\dXx]\b', '[ID_CARD]'),
        "bank_card": (r'\b\d{16,19}\b', '[BANK_CARD]'),
    }
    
    def sanitize(self, text):
        for name, (pattern, replacement) in self.PATTERNS.items():
            text = re.sub(pattern, replacement, text)
        return text

5.2 日志访问控制

  • Trace日志包含用户对话内容,属于敏感数据
  • 访问需要权限审批
  • 日志保留期限合规(如90天后自动删除)

六、从日志到改进

6.1 故障复盘流程

1. 发现故障(告警/用户反馈)
2. 拉取Trace → 还原执行流程
3. 定位异常Span → 缩小排查范围
4. 分析根因 → LLM输出问题?工具问题?Prompt问题?
5. 修复 → 修改Prompt/代码/配置
6. 添加回归测试 → 防止复发
7. 更新Runbook → 积累排查经验

6.2 持续优化

定期分析日志数据,发现系统性问题:

  • 哪些意图的误判率最高?→ 优化对应prompt
  • 哪些工具的失败率最高?→ 优化工具或增加重试
  • 哪些场景的token消耗最大?→ 优化上下文管理
  • 哪些时间段的延迟最高?→ 优化资源调度

结语

可观测性不是事后补救的附加功能,而是Agent系统的核心设计要素。一个好的日志体系,能让Agent从"出了问题不知道为什么"变成"每个决策都可追溯"。这不仅是运维的福音,更是产品迭代的基石——只有看见Agent在做什么,才能让它做得更好。

本文同步发布于 硅基AGI论坛