阅读时间：约 13 分钟
前置知识：D01 监控

---

D01 讲了监控：你知道 Agent 出问题了。然后呢？

你接到告警：错误率飙到 15%。你打开系统，看着"500 Internal Server Error"。用户说的是"Agent 回复有问题"。你不知道从哪查起。

日志不是"把 print 全开着"。日志是你排查问题的唯一线索。这线索怎么布，决定了你排查问题花 5 分钟还是 5 天。

---

前言：日志的代价

两个极端：

• 日志太少：出问题后一脸茫然，连用户问了什么都不知道
• 日志太多：每天几 GB 日志，排查问题时翻了三页就放弃了

好日志的标准：出问题后的第一个 5 分钟，能从日志里定位到根因。

📌 本章核心：日志分层（请求日志、执行日志、LLM 日志）→ 结构化 → 链路追踪 → 排查流程。目标是出问题 5 分钟定位。

---

第一部分：三层日志模型

请求日志（用户视角）
├── 用户问了什么
├── Agent 返回了什么
├── 花了多长时间
└── 成功还是失败

执行日志（Agent 内部视角）
├── 调了哪个工具
├── 工具返回了什么
├── 推理过程
└── 哪个环节卡住了

LLM 日志（模型视角）
├── Prompt 发了什么
├── LLM 返回了什么
├── Token 消耗
└── 重试了几次

1.1 请求日志

import json
import time
import uuid
from datetime import datetime
from typing import Optional, Dict, Any

class RequestLogger:
    """
    请求日志：记录每次用户请求的完整生命周期
    """
    
    def __init__(self, log_file: str = "agent_requests.log"):
        self.log_file = log_file
    
    def log_request(
        self,
        request_id: str,
        user_id: str,
        query: str,
        response: str,
        latency_ms: float,
        success: bool,
        error_msg: Optional[str] = None
    ):
        """记录一次请求"""
        entry = {
            "type": "request",
            "request_id": request_id,
            "user_id": user_id,
            "timestamp": datetime.now().isoformat(),
            "query": query[:500],  # 截断，防止过长
            "response": response[:2000],
            "latency_ms": round(latency_ms, 1),
            "success": success,
            "error_msg": error_msg
        }
        
        self._write(entry)
    
    def _write(self, entry: Dict):
        """写日志（一条 JSON 一行，方便 grep）"""
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

1.2 执行日志

class ExecutionLogger:
    """
    执行日志：记录 Agent 内部每一步操作
    """
    
    def __init__(self, log_file: str = "agent_execution.log"):
        self.log_file = log_file
    
    def log_tool_call(
        self,
        request_id: str,
        tool_name: str,
        tool_input: Dict,
        tool_output: Any,
        latency_ms: float,
        success: bool
    ):
        """记录一次工具调用"""
        entry = {
            "type": "tool_call",
            "request_id": request_id,
            "timestamp": datetime.now().isoformat(),
            "tool": tool_name,
            "input": str(tool_input)[:1000],
            "output": str(tool_output)[:2000],
            "latency_ms": round(latency_ms, 1),
            "success": success
        }
        
        self._write(entry)
    
    def log_reasoning(self, request_id: str, step: int, thought: str):
        """记录推理步骤"""
        entry = {
            "type": "reasoning",
            "request_id": request_id,
            "timestamp": datetime.now().isoformat(),
            "step": step,
            "thought": thought[:2000]
        }
        
        self._write(entry)
    
    def log_error(
        self,
        request_id: str,
        error_type: str,
        error_msg: str,
        context: Optional[Dict] = None
    ):
        """记录错误"""
        entry = {
            "type": "error",
            "request_id": request_id,
            "timestamp": datetime.now().isoformat(),
            "error_type": error_type,
            "error_msg": error_msg,
            "context": context or {}
        }
        
        self._write(entry)
    
    def _write(self, entry: Dict):
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

1.3 LLM 日志

class LLMLogger:
    """
    LLM 日志：记录每次大模型调用
    """
    
    def __init__(self, log_file: str = "agent_llm.log"):
        self.log_file = log_file
    
    def log_llm_call(
        self,
        request_id: str,
        model: str,
        prompt_tokens: int,
        completion_tokens: int,
        latency_ms: float,
        retry_count: int,
        error_msg: Optional[str] = None
    ):
        """记录一次 LLM 调用"""
        entry = {
            "type": "llm_call",
            "request_id": request_id,
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
            "latency_ms": round(latency_ms, 1),
            "retry_count": retry_count,
            "error_msg": error_msg
        }
        
        self._write(entry)
    
    def _write(self, entry: Dict):
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

📌 本章要点：三层日志分开记，一行一条 JSON。出问题时按 request_id 串起来，从请求层到执行层到 LLM 层，逐层排查。

---

第二部分：结构化日志

2.1 为什么一行一条 JSON

// ❌ 传统日志（人读方便，机器难处理）
[2026-07-10 14:23:01] INFO  User 123 asked: "什么是 RAG"
[2026-07-10 14:23:02] DEBUG Tool search_docs called with query="RAG 定义"
[2026-07-10 14:23:03] DEBUG Tool search_docs returned 5 results in 1.2s
[2026-07-10 14:23:05] INFO  Response sent (234 tokens, 4.5s total)

// ✅ 结构化日志（机器好处理，人也能读）
{"type":"request","request_id":"req_abc","query":"什么是 RAG","latency_ms":4500}
{"type":"tool_call","request_id":"req_abc","tool":"search_docs","latency_ms":1200}
{"type":"llm_call","request_id":"req_abc","model":"deepseek-v4-pro","total_tokens":234}

一行一条 JSON 的好处：

• grep request_id 直接拉出一次请求的所有日志
• jq 可以按字段过滤统计
• 日志平台（ELK、Loki）天然支持

2.2 结构化日志的统一接口

class StructuredLogger:
    """
    统一结构化日志接口
    所有日志通过这个类输出，保证格式一致
    """
    
    def __init__(self, base_dir: str = "./logs"):
        import os
        os.makedirs(base_dir, exist_ok=True)
        
        self.base_dir = base_dir
        self.request_logger = RequestLogger(f"{base_dir}/requests.log")
        self.exec_logger = ExecutionLogger(f"{base_dir}/execution.log")
        self.llm_logger = LLMLogger(f"{base_dir}/llm.log")
    
    def log_request(self, request_id: str, **kwargs):
        self.request_logger.log_request(request_id=request_id, **kwargs)
    
    def log_tool_call(self, request_id: str, **kwargs):
        self.exec_logger.log_tool_call(request_id=request_id, **kwargs)
    
    def log_llm_call(self, request_id: str, **kwargs):
        self.llm_logger.log_llm_call(request_id=request_id, **kwargs)
    
    def log_error(self, request_id: str, error_type: str, error_msg: str, **kwargs):
        self.exec_logger.log_error(
            request_id=request_id,
            error_type=error_type,
            error_msg=error_msg,
            context=kwargs
        )
        # 严重错误同步记到请求日志
        if error_type in ("timeout", "auth_failure", "model_error"):
            self.request_logger.log_request(
                request_id=request_id,
                user_id=kwargs.get("user_id", "unknown"),
                query=kwargs.get("query", ""),
                response="",
                latency_ms=0,
                success=False,
                error_msg=f"{error_type}: {error_msg}"
            )

2.3 日志查询工具

class LogQuery:
    """
    日志查询：按 request_id 串起整个请求的日志链路
    """
    
    def __init__(self, base_dir: str = "./logs"):
        self.base_dir = base_dir
    
    def trace_request(self, request_id: str) -> Dict:
        """
        按 request_id 追踪一次请求的完整日志
        """
        result = {
            "request_id": request_id,
            "timeline": []
        }
        
        # 从三个日志文件中搜索
        log_files = [
            f"{self.base_dir}/requests.log",
            f"{self.base_dir}/execution.log",
            f"{self.base_dir}/llm.log"
        ]
        
        for log_file in log_files:
            try:
                with open(log_file, "r", encoding="utf-8") as f:
                    for line in f:
                        if request_id in line:
                            entry = json.loads(line)
                            result["timeline"].append(entry)
            except FileNotFoundError:
                continue
        
        # 按时间排序
        result["timeline"].sort(key=lambda x: x.get("timestamp", ""))
        
        return result
    
    def print_trace(self, request_id: str):
        """打印请求追踪信息"""
        trace = self.trace_request(request_id)
        
        print(f"\n=== 请求追踪: {request_id} ===")
        print(f"日志条数: {len(trace['timeline'])}")
        
        for entry in trace["timeline"]:
            ts = entry.get("timestamp", "")[11:19]  # 只取时间部分
            etype = entry.get("type", "unknown")
            
            if etype == "request":
                status = "✓" if entry.get("success") else "✗"
                print(f"  {ts} [{etype}] {status} 耗时 {entry.get('latency_ms', 0):.0f}ms")
            elif etype == "tool_call":
                status = "✓" if entry.get("success") else "✗"
                print(f"  {ts} [{etype}] {status} {entry.get('tool', '')} ({entry.get('latency_ms', 0):.0f}ms)")
            elif etype == "llm_call":
                print(f"  {ts} [{etype}] 模型 {entry.get('model', '')} Token {entry.get('total_tokens', 0)}")
            elif etype == "error":
                print(f"  {ts} [{etype}] ⚠ {entry.get('error_type', '')}: {entry.get('error_msg', '')[:100]}")
        
        print("=== 追踪结束 ===\n")

📌 本章要点：结构化日志的核心是一行一条 JSON + 统一的 request_id。出问题时 grep request_id 就能拉出完整链路。

---

第三部分：链路追踪

3.1 request_id 的全链路传递

import uuid

class TraceContext:
    """
    链路追踪上下文
    确保 request_id 在整个调用链路中传递
    """
    
    _context = {}
    
    @classmethod
    def new_request(cls, user_id: str, query: str) -> str:
        """创建新的请求上下文"""
        request_id = f"req_{uuid.uuid4().hex[:12]}"
        cls._context = {
            "request_id": request_id,
            "user_id": user_id,
            "query": query[:100],
            "start_time": time.time()
        }
        return request_id
    
    @classmethod
    def get_request_id(cls) -> Optional[str]:
        """获取当前 request_id"""
        return cls._context.get("request_id")
    
    @classmethod
    def get_context(cls) -> Dict:
        """获取当前上下文"""
        return cls._context.copy()


class TracedAgent:
    """
    带链路追踪的 Agent
    每次操作自动带上 request_id
    """
    
    def __init__(self, logger: StructuredLogger):
        self.logger = logger
    
    def process(self, user_id: str, query: str) -> str:
        """处理用户请求（带全链路追踪）"""
        request_id = TraceContext.new_request(user_id, query)
        start_time = time.time()
        
        try:
            # 执行 Agent 逻辑
            response = self._execute(query)
            latency_ms = (time.time() - start_time) * 1000
            
            self.logger.log_request(
                request_id=request_id,
                user_id=user_id,
                query=query,
                response=response,
                latency_ms=latency_ms,
                success=True
            )
            
            return response
            
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            
            self.logger.log_error(
                request_id=request_id,
                error_type=type(e).__name__,
                error_msg=str(e),
                user_id=user_id,
                query=query
            )
            
            self.logger.log_request(
                request_id=request_id,
                user_id=user_id,
                query=query,
                response="",
                latency_ms=latency_ms,
                success=False,
                error_msg=str(e)
            )
            
            return "抱歉，处理请求时出错了。"
    
    def _execute(self, query: str) -> str:
        """Agent 执行逻辑（带工具调用追踪）"""
        request_id = TraceContext.get_request_id()
        
        # 工具调用
        tool_start = time.time()
        try:
            tool_result = self._call_tool("search", {"query": query})
            tool_latency = (time.time() - tool_start) * 1000
            
            self.logger.log_tool_call(
                request_id=request_id,
                tool_name="search",
                tool_input={"query": query},
                tool_output=tool_result,
                latency_ms=tool_latency,
                success=True
            )
        except Exception as e:
            self.logger.log_tool_call(
                request_id=request_id,
                tool_name="search",
                tool_input={"query": query},
                tool_output=str(e),
                latency_ms=(time.time() - tool_start) * 1000,
                success=False
            )
            raise
        
        # LLM 调用
        llm_start = time.time()
        try:
            response = self._call_llm(query, tool_result)
            llm_latency = (time.time() - llm_start) * 1000
            
            self.logger.log_llm_call(
                request_id=request_id,
                model="deepseek-v4-pro",
                prompt_tokens=len(query),
                completion_tokens=len(response),
                latency_ms=llm_latency,
                retry_count=0
            )
            
            return response
        except Exception as e:
            self.logger.log_error(
                request_id=request_id,
                error_type="llm_error",
                error_msg=str(e)
            )
            raise
    
    def _call_tool(self, tool_name: str, tool_input: Dict) -> str:
        """模拟工具调用"""
        return f"[模拟]{tool_name} 的结果"
    
    def _call_llm(self, query: str, context: str) -> str:
        """模拟 LLM 调用"""
        return f"[模拟]LLM 对 '{query}' 的回答"

3.2 排查链路

排查步骤（出问题后的标准流程）：

1. grep "error" → 找到错误 request_id
2. trace request_id → 看到完整链路
3. 找到第一个 fail 的步骤 → 定位根因
4. 看前后步骤 → 理解上下文

class TroubleshootingGuide:
    """
    常见问题排查指南
    """
    
    def __init__(self, log_query: LogQuery):
        self.log_query = log_query
    
    def diagnose(self, request_id: str) -> Dict:
        """
        自动诊断一个请求
        """
        trace = self.log_query.trace_request(request_id)
        timeline = trace.get("timeline", [])
        
        diagnosis = {
            "request_id": request_id,
            "status": "unknown",
            "root_cause": None,
            "suggestions": []
        }
        
        # 找到第一个错误
        errors = [e for e in timeline if e.get("type") == "error"]
        
        if not errors:
            diagnosis["status"] = "no_errors_found"
            diagnosis["suggestions"].append("没有找到错误，检查是否日志级别不够")
            return diagnosis
        
        first_error = errors[0]
        error_type = first_error.get("error_type", "")
        
        if "timeout" in error_type.lower():
            diagnosis["status"] = "timeout"
            diagnosis["root_cause"] = "请求超时"
            diagnosis["suggestions"] = [
                "检查下游服务是否响应",
                "检查网络延迟",
                "考虑增加超时时间"
            ]
        elif "auth" in error_type.lower():
            diagnosis["status"] = "auth_error"
            diagnosis["root_cause"] = "认证失败"
            diagnosis["suggestions"] = [
                "检查 API Key 是否过期",
                "检查权限配置"
            ]
        elif "rate" in error_type.lower():
            diagnosis["status"] = "rate_limit"
            diagnosis["root_cause"] = "频率限制"
            diagnosis["suggestions"] = [
                "检查当前 QPS",
                "考虑增加限流阈值",
                "检查是否有异常流量"
            ]
        else:
            diagnosis["status"] = "unknown_error"
            diagnosis["root_cause"] = error_type
            diagnosis["suggestions"] = [
                f"查看完整错误信息: {first_error.get('error_msg', '')[:200]}",
                "查看发生错误前的最后几步操作"
            ]
        
        return diagnosis

📌 本章要点：链路追踪的核心是 request_id 全链路传递。排查流程：找错误 → 追踪链路 → 定位第一次失败 → 看上下文。

---

第四部分：日志分级

4.1 日志级别

import logging

class LogLevel:
    """
    日志级别定义
    """
    DEBUG = 10    # 调试信息（开发环境）
    INFO = 20     # 一般信息（生产默认）
    WARNING = 30  # 警告（需要注意但不影响功能）
    ERROR = 40    # 错误（影响功能）
    CRITICAL = 50 # 严重错误（需要立即处理）


class LevelBasedLogger:
    """
    分级日志：不同环境用不同级别
    """
    
    def __init__(self, level: int = LogLevel.INFO):
        self.level = level
    
    def debug(self, msg: str):
        """调试信息：只在开发环境输出"""
        if self.level <= LogLevel.DEBUG:
            self._output("DEBUG", msg)
    
    def info(self, msg: str):
        """一般信息：生产环境默认级别"""
        if self.level <= LogLevel.INFO:
            self._output("INFO", msg)
    
    def warning(self, msg: str):
        """警告信息：可以保留但不影响功能"""
        if self.level <= LogLevel.WARNING:
            self._output("WARNING", msg)
    
    def error(self, msg: str):
        """错误信息：影响功能的错误"""
        if self.level <= LogLevel.ERROR:
            self._output("ERROR", msg)
    
    def critical(self, msg: str):
        """严重错误：需要立即处理"""
        if self.level <= LogLevel.CRITICAL:
            self._output("CRITICAL", msg)
    
    def _output(self, level: str, msg: str):
        print(f"[{level}] {datetime.now().isoformat()} {msg}")

4.2 各级别日志的内容

级别	记录内容	保留时长
DEBUG	函数入参出参、变量值、中间状态	1 天
INFO	请求记录、工具调用、LLM 调用	7 天
WARNING	重试、降级、缓存过期	30 天
ERROR	调用失败、超时、返回异常	90 天
CRITICAL	服务宕机、数据丢失	永久

class LogRetention:
    """
    日志保留策略
    """
    
    def __init__(self, base_dir: str = "./logs"):
        self.base_dir = base_dir
        self.retention = {
            "DEBUG": 1,      # 天
            "INFO": 7,
            "WARNING": 30,
            "ERROR": 90,
            "CRITICAL": 365
        }
    
    def cleanup(self):
        """清理过期日志"""
        import os
        now = time.time()
        
        for filename in os.listdir(self.base_dir):
            if not filename.endswith(".log"):
                continue
            
            filepath = os.path.join(self.base_dir, filename)
            
            # 获取文件修改时间
            mtime = os.path.getmtime(filepath)
            age_days = (now - mtime) / 86400
            
            # 检查所有级别的保留时间
            for level, days in self.retention.items():
                log_file = f"{self.base_dir}/{level.lower()}.log"
                if filepath == log_file and age_days > days:
                    os.remove(filepath)
                    print(f"清理过期日志: {filename} (已保留 {age_days:.0f} 天)")

📌 本章要点：日志分级让生产环境不输出调试信息，减少 I/O 开销。不同级别不同保留时长，平衡存储成本。

---

第五部分：日志分析

5.1 快速分析脚本

class LogAnalyzer:
    """
    日志分析：快速统计常见问题
    """
    
    def __init__(self, log_file: str):
        self.log_file = log_file
    
    def count_errors(self) -> Dict:
        """统计错误分布"""
        error_counts = {}
        total_lines = 0
        
        with open(self.log_file, "r", encoding="utf-8") as f:
            for line in f:
                total_lines += 1
                try:
                    entry = json.loads(line)
                    if entry.get("type") == "error":
                        error_type = entry.get("error_type", "unknown")
                        error_counts[error_type] = error_counts.get(error_type, 0) + 1
                except json.JSONDecodeError:
                    continue
        
        return {
            "total_lines": total_lines,
            "error_distribution": dict(
                sorted(error_counts.items(), key=lambda x: x[1], reverse=True)
            )
        }
    
    def top_slow_requests(self, n: int = 10) -> list:
        """最慢的请求 Top-N"""
        slow_requests = []
        
        with open(self.log_file, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    entry = json.loads(line)
                    if entry.get("type") == "request":
                        slow_requests.append({
                            "request_id": entry.get("request_id"),
                            "latency_ms": entry.get("latency_ms", 0),
                            "query": entry.get("query", "")[:50]
                        })
                except json.JSONDecodeError:
                    continue
        
        slow_requests.sort(key=lambda x: x["latency_ms"], reverse=True)
        return slow_requests[:n]
    
    def time_range_summary(self, start_hour: int, end_hour: int) -> Dict:
        """某个时间段的请求总结"""
        request_count = 0
        error_count = 0
        total_latency = 0
        
        with open(self.log_file, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    entry = json.loads(line)
                    ts = entry.get("timestamp", "")
                    if not ts:
                        continue
                    
                    hour = int(ts[11:13])
                    if start_hour <= hour < end_hour:
                        if entry.get("type") == "request":
                            request_count += 1
                            total_latency += entry.get("latency_ms", 0)
                            if not entry.get("success", True):
                                error_count += 1
                except (json.JSONDecodeError, ValueError):
                    continue
        
        return {
            "time_range": f"{start_hour:02d}:00 - {end_hour:02d}:00",
            "total_requests": request_count,
            "error_count": error_count,
            "error_rate": round(error_count / request_count, 3) if request_count > 0 else 0,
            "avg_latency_ms": round(total_latency / request_count, 1) if request_count > 0 else 0
        }

5.2 排查示例

场景：用户反馈"Agent 回复特别慢"

排查步骤：
1. 查慢请求 Top-10 → 发现 5 个请求延迟超过 15s
2. trace 其中一个 request_id → 看到工具调用花了 12s
3. 查工具日志 → 发现向量搜索超时
4. 查向量库 → 索引大小从 10GB 涨到了 80GB
5. 结论：向量库性能下降，需要优化索引

全程不超过 5 分钟，因为有结构化日志和链路追踪。

📌 本章要点：日志分析不是翻文件，是写脚本自动统计。错误分布、慢请求 Top-N、时段总结，三个脚本覆盖 80% 的排查场景。

---

总结

1. 三层日志：请求日志（用户视角）+ 执行日志（Agent 内部）+ LLM 日志（模型视角）
2. 结构化：一行一条 JSON，统一 request_id，方便 grep 和 jq
3. 链路追踪：request_id 全链路传递，出问题按 ID 串起所有日志
4. 日志分级：DEBUG/INFO/WARNING/ERROR/CRITICAL，不同级别不同保留策略
5. 自动分析：错误分布、慢请求、时段统计，三个脚本覆盖大部分排查场景

🤔 试试看：你的 Agent 现在有结构化日志吗？随便挑一个最近的 request_id，能在 1 分钟内拉出完整执行链路吗？

---

思维导图

• 日志
  • 三层模型
    • 请求日志（用户视角）
    • 执行日志（Agent 内部）
    • LLM 日志（模型视角）
  • 结构化日志
    • 一行一条 JSON
    • 统一 request_id
    • 日志查询工具
  • 链路追踪
    • request_id 全链路传递
    • 排查流程
    • 常见问题诊断
  • 日志分级
    • DEBUG/INFO/WARNING/ERROR/CRITICAL
    • 不同级别不同内容
    • 日志保留策略
  • 日志分析
    • 错误分布统计
    • 慢请求 Top-N
    • 时段总结