引言

Agent 的执行过程是黑盒——你看到输入和输出，但中间发生了什么？调用了什么工具？为什么选择这个路径？Token 花在哪里？结构化日志是打开这个黑盒的钥匙。2026年，随着 Agent 系统复杂度增长，日志不再是"给人看的文本"，而是"给系统查询的数据"。

一、Agent 日志设计原则

传统日志 vs Agent 日志

维度传统日志Agent 日志
格式半结构化文本全结构化 JSON
粒度请求级步骤级（每轮迭代）
关联request_idtrace_id + session_id + step_id
内容状态和错误决策推理、工具调用、Token消耗
用途故障排查故障排查 + 质量分析 + 成本归因
查询grep/正则结构化查询 + 聚合分析

设计原则

  1. 一切皆结构化：每条日志都是可查询的 JSON
  2. 因果链完整：从输入到输出的每一步都可追溯
  3. 上下文丰富：每条日志携带足够的上下文独立理解
  4. 成本感知：Token 和费用信息嵌入每条日志
  5. 隐私安全：PII 自动脱敏

二、日志数据模型

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import uuid

class LogLevel(Enum):
    DEBUG = "debug"
    INFO = "info"
    WARN = "warn"
    ERROR = "error"
    CRITICAL = "critical"

class EventType(Enum):
    # Agent 生命周期
    AGENT_START = "agent.start"
    AGENT_END = "agent.end"
    AGENT_INTERRUPT = "agent.interrupt"
    
    # LLM 交互
    LLM_REQUEST = "llm.request"
    LLM_RESPONSE = "llm.response"
    LLM_ERROR = "llm.error"
    LLM_RETRY = "llm.retry"
    
    # 工具调用
    TOOL_DECISION = "tool.decision"
    TOOL_CALL_START = "tool.call.start"
    TOOL_CALL_END = "tool.call.end"
    TOOL_ERROR = "tool.error"
    
    # 决策推理
    REASONING = "reasoning"
    PLANNING = "planning"
    REFLECTION = "reflection"
    
    # 状态变更
    STATE_UPDATE = "state.update"
    CONTEXT_PRUNED = "context.pruned"
    
    # 错误恢复
    ERROR_RECOVERY = "error.recovery"
    FALLBACK_TRIGGERED = "fallback.triggered"

@dataclass
class AgentLogEntry:
    """Agent 结构化日志条目"""
    
    # 标识
    log_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    
    # 关联
    trace_id: str = ""        # 贯穿整个请求
    session_id: str = ""      # 会话ID
    step_id: str = ""         # 当前步骤ID
    parent_step_id: str = ""  # 父步骤（用于嵌套）
    
    # 事件
    event_type: EventType = EventType.AGENT_START
    level: LogLevel = LogLevel.INFO
    
    # Agent 信息
    agent_name: str = ""
    agent_version: str = ""
    
    # 内容
    message: str = ""
    data: dict = field(default_factory=dict)
    
    # 性能
    duration_ms: float = 0
    
    # 成本
    tokens_in: int = 0
    tokens_out: int = 0
    cost_usd: float = 0
    
    # 上下文
    user_id: str = ""
    tenant_id: str = ""
    environment: str = "production"

三、结构化日志实现

3.1 日志记录器

import structlog
from structlog.contextvars import bind_contextvars, clear_contextvars

class AgentLogger:
    """Agent 结构化日志记录器"""
    
    def __init__(self):
        self.logger = structlog.get_logger("agent")
    
    def bind_request_context(
        self,
        trace_id: str,
        session_id: str,
        user_id: str,
        agent_name: str,
        agent_version: str
    ):
        """绑定请求级上下文"""
        bind_contextvars(
            trace_id=trace_id,
            session_id=session_id,
            user_id=user_id,
            agent_name=agent_name,
            agent_version=agent_version,
        )
    
    def log_agent_start(
        self,
        query: str,
        available_tools: list[str],
        max_iterations: int
    ):
        """记录 Agent 启动"""
        self.logger.info(
            "agent.start",
            event_type=EventType.AGENT_START.value,
            query_preview=query[:200],
            query_length=len(query),
            available_tools=available_tools,
            max_iterations=max_iterations,
        )
    
    def log_llm_call(
        self,
        model: str,
        messages_count: int,
        input_tokens: int,
        temperature: float,
        tools_provided: bool
    ):
        """记录 LLM 调用"""
        self.logger.info(
            "llm.request",
            event_type=EventType.LLM_REQUEST.value,
            model=model,
            messages_count=messages_count,
            input_tokens=input_tokens,
            temperature=temperature,
            tools_provided=tools_provided,
        )
    
    def log_llm_response(
        self,
        model: str,
        output_tokens: int,
        duration_ms: float,
        cost_usd: float,
        tool_calls: list[dict] | None,
        finish_reason: str
    ):
        """记录 LLM 响应"""
        self.logger.info(
            "llm.response",
            event_type=EventType.LLM_RESPONSE.value,
            model=model,
            output_tokens=output_tokens,
            duration_ms=round(duration_ms, 2),
            cost_usd=round(cost_usd, 6),
            tool_calls_count=len(tool_calls) if tool_calls else 0,
            tool_calls=[
                {"tool": tc["function"]["name"], 
                 "args_preview": str(tc["function"]["arguments"])[:100]}
                for tc in (tool_calls or [])
            ],
            finish_reason=finish_reason,
        )
    
    def log_tool_decision(
        self,
        selected_tool: str,
        available_tools: list[str],
        reasoning: str,
        confidence: float | None = None
    ):
        """记录工具选择决策"""
        self.logger.debug(
            "tool.decision",
            event_type=EventType.TOOL_DECISION.value,
            selected_tool=selected_tool,
            available_tools=available_tools,
            reasoning=reasoning,
            confidence=confidence,
        )
    
    def log_tool_execution(
        self,
        tool_name: str,
        args: dict,
        result: any,
        duration_ms: float,
        success: bool,
        error: str | None = None
    ):
        """记录工具执行"""
        log_data = {
            "event_type": EventType.TOOL_CALL_END.value,
            "tool": tool_name,
            "args_preview": self._truncate_args(args),
            "duration_ms": round(duration_ms, 2),
            "success": success,
        }
        
        if success:
            log_data["result_preview"] = str(result)[:500]
            log_data["result_size"] = len(str(result))
        else:
            log_data["error"] = error
        
        self.logger.info("tool.call.end", **log_data)
    
    def log_reasoning(
        self,
        step: int,
        thought: str,
        action: str,
        observation: str
    ):
        """记录 ReAct 推理过程"""
        self.logger.debug(
            "reasoning",
            event_type=EventType.REASONING.value,
            step=step,
            thought=thought[:500],
            action=action,
            observation=observation[:500],
        )
    
    def log_agent_end(
        self,
        total_iterations: int,
        total_tokens_in: int,
        total_tokens_out: int,
        total_cost: float,
        total_duration_ms: float,
        tools_used: list[str],
        status: str
    ):
        """记录 Agent 结束"""
        self.logger.info(
            "agent.end",
            event_type=EventType.AGENT_END.value,
            total_iterations=total_iterations,
            total_tokens_in=total_tokens_in,
            total_tokens_out=total_tokens_out,
            total_cost_usd=round(total_cost, 6),
            total_duration_ms=round(total_duration_ms, 2),
            tools_used=tools_used,
            status=status,
        )
    
    def _truncate_args(self, args: dict, max_len: int = 200) -> dict:
        """截断过长的参数"""
        result = {}
        for k, v in args.items():
            s = str(v)
            result[k] = s[:max_len] + "..." if len(s) > max_len else v
        return result
    
    def clear_context(self):
        """清理上下文"""
        clear_contextvars()

3.2 日志中间件

class AgentLoggingMiddleware:
    """Agent 日志中间件——自动记录"""
    
    def __init__(self, logger: AgentLogger):
        self.logger = logger
    
    async def wrap_agent(
        self,
        agent: Agent,
        request: Request
    ) -> Response:
        """包装 Agent 执行，自动记录日志"""
        
        trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4()))
        
        # 绑定上下文
        self.logger.bind_request_context(
            trace_id=trace_id,
            session_id=request.session_id,
            user_id=request.user_id,
            agent_name=agent.name,
            agent_version=agent.version
        )
        
        start_time = time.time()
        
        try:
            # 记录启动
            self.logger.log_agent_start(
                query=request.query,
                available_tools=agent.get_tool_names(),
                max_iterations=agent.max_iterations
            )
            
            # 执行 Agent（内部会通过回调记录各步骤）
            response = await agent.run(request.query)
            
            # 记录结束
            self.logger.log_agent_end(
                total_iterations=agent.iteration_count,
                total_tokens_in=agent.total_input_tokens,
                total_tokens_out=agent.total_output_tokens,
                total_cost=agent.total_cost,
                total_duration_ms=(time.time() - start_time) * 1000,
                tools_used=agent.tools_used,
                status="success"
            )
            
            return response
            
        except Exception as e:
            self.logger.logger.error(
                "agent.error",
                event_type="agent.error",
                error_type=type(e).__name__,
                error_message=str(e),
                duration_ms=(time.time() - start_time) * 1000,
            )
            raise
        finally:
            self.logger.clear_context()

四、日志输出配置

import structlog
import logging
import sys

def configure_logging(environment: str = "production"):
    """配置结构化日志"""
    
    if environment == "production":
        # 生产环境：JSON 输出到 stdout
        structlog.configure(
            processors=[
                structlog.contextvars.merge_contextvars,
                structlog.processors.add_log_level,
                structlog.processors.TimeStamper(fmt="iso"),
                _add_server_info(),
                _pii_redactor(),           # PII 脱敏
                structlog.processors.JSONRenderer(ensure_ascii=False),
            ],
            wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
            logger_factory=structlog.PrintLoggerFactory(),
        )
    
    elif environment == "development":
        # 开发环境：彩色控制台输出
        structlog.configure(
            processors=[
                structlog.contextvars.merge_contextvars,
                structlog.processors.add_log_level,
                structlog.dev.ConsoleRenderer(colors=True),
            ],
        )
    
    # 同时写入文件（轮转）
    file_handler = logging.handlers.RotatingFileHandler(
        "/var/log/agent/agent.log",
        maxBytes=100_000_000,  # 100MB
        backupCount=10,
    )
    file_handler.setFormatter(
        logging.Formatter('{"message": "%(message)s"}')
    )


def _add_server_info():
    """添加服务器信息处理器"""
    def processor(logger, method_name, event_dict):
        event_dict["hostname"] = socket.gethostname()
        event_dict["pid"] = os.getpid()
        return event_dict
    return processor


def _pii_redactor():
    """PII 脱敏处理器"""
    patterns = {
        "email": (r'[\w.-]+@[\w.-]+\.\w+', '[REDACTED_EMAIL]'),
        "phone": (r'\b1[3-9]\d{9}\b', '[REDACTED_PHONE]'),
        "id_card": (r'\b\d{17}[\dXx]\b', '[REDACTED_ID]'),
    }
    
    def redact(text: str) -> str:
        for pattern, replacement in patterns.values():
            text = re.sub(pattern, replacement, text)
        return text
    
    def processor(logger, method_name, event_dict):
        for key, value in event_dict.items():
            if isinstance(value, str):
                event_dict[key] = redact(value)
            elif isinstance(value, dict):
                event_dict[key] = {
                    k: redact(v) if isinstance(v, str) else v
                    for k, v in value.items()
                }
        return event_dict
    
    return processor

五、日志查询与分析

5.1 日志查询接口

class AgentLogQuery:
    """Agent 日志查询接口"""
    
    async def get_trace(self, trace_id: str) -> list[dict]:
        """获取完整执行链路"""
        return await self.elasticsearch.search(
            index="agent-logs-*",
            body={
                "query": {"term": {"trace_id": trace_id}},
                "sort": [{"timestamp": "asc"}]
            }
        )
    
    async def find_slow_agents(
        self,
        threshold_ms: float = 30000,
        time_range: str = "1h"
    ) -> list[dict]:
        """查找慢 Agent 执行"""
        return await self.elasticsearch.search(
            index="agent-logs-*",
            body={
                "query": {
                    "bool": {
                        "filter": [
                            {"term": {"event_type": "agent.end"}},
                            {"range": {
                                "total_duration_ms": {"gte": threshold_ms}
                            }},
                            {"range": {
                                "timestamp": {"gte": f"now-{time_range}"}
                            }}
                        ]
                    }
                },
                "sort": [{"total_duration_ms": "desc"}],
                "size": 50
            }
        )
    
    async def find_expensive_sessions(
        self,
        min_cost: float = 0.10,
        time_range: str = "24h"
    ) -> list[dict]:
        """查找高成本会话"""
        return await self.elasticsearch.search(
            index="agent-logs-*",
            body={
                "query": {
                    "bool": {
                        "filter": [
                            {"term": {"event_type": "agent.end"}},
                            {"range": {"total_cost_usd": {"gte": min_cost}}},
                            {"range": {"timestamp": {"gte": f"now-{time_range}"}}}
                        ]
                    }
                },
                "sort": [{"total_cost_usd": "desc"}]
            }
        )
    
    async def get_tool_failure_rate(
        self,
        time_range: str = "1h"
    ) -> dict:
        """工具失败率统计"""
        result = await self.elasticsearch.search(
            index="agent-logs-*",
            body={
                "size": 0,
                "query": {
                    "bool": {
                        "filter": [
                            {"term": {"event_type": "tool.call.end"}},
                            {"range": {"timestamp": {"gte": f"now-{time_range}"}}}
                        ]
                    }
                },
                "aggs": {
                    "by_tool": {
                        "terms": {"field": "tool"},
                        "aggs": {
                            "success_count": {
                                "filter": {"term": {"success": True}}
                            },
                            "failure_count": {
                                "filter": {"term": {"success": False}}
                            }
                        }
                    }
                }
            }
        )
        
        return {
            bucket["key"]: {
                "total": bucket["doc_count"],
                "success": bucket["success_count"]["doc_count"],
                "failure": bucket["failure_count"]["doc_count"],
                "failure_rate": bucket["failure_count"]["doc_count"] / bucket["doc_count"]
            }
            for bucket in result["aggregations"]["by_tool"]["buckets"]
        }

5.2 执行链路回放

class TraceReplay:
    """Agent 执行链路回放"""
    
    async def replay(self, trace_id: str) -> str:
        """生成可读的执行链路报告"""
        events = await self.query.get_trace(trace_id)
        
        if not events:
            return f"No trace found for {trace_id}"
        
        report = []
        report.append(f"=== Agent Trace Replay: {trace_id} ===\n")
        
        total_tokens = 0
        total_cost = 0
        total_duration = 0
        
        for event in events:
            ts = event["timestamp"]
            event_type = event["event_type"]
            data = event.get("data", {})
            
            if event_type == "agent.start":
                report.append(f"[{ts}] 🚀 Agent started")
                report.append(f"  Query: {data.get('query_preview', '')[:100]}")
                report.append(f"  Tools: {data.get('available_tools', [])}")
            
            elif event_type == "llm.request":
                report.append(f"[{ts}] 📤 LLM call → {data.get('model')}")
                report.append(f"  Input tokens: {data.get('input_tokens', 0)}")
                total_tokens += data.get('input_tokens', 0)
            
            elif event_type == "llm.response":
                report.append(f"[{ts}] 📥 LLM response ← {data.get('model')}")
                report.append(f"  Output tokens: {data.get('output_tokens', 0)}")
                report.append(f"  Duration: {data.get('duration_ms', 0):.0f}ms")
                report.append(f"  Cost: ${data.get('cost_usd', 0):.6f}")
                if data.get('tool_calls_count', 0) > 0:
                    report.append(f"  Tool calls: {data['tool_calls']}")
                total_tokens += data.get('output_tokens', 0)
                total_cost += data.get('cost_usd', 0)
                total_duration += data.get('duration_ms', 0)
            
            elif event_type == "tool.call.end":
                status = "✅" if data.get('success') else "❌"
                report.append(f"[{ts}] 🔧 {status} {data.get('tool')}")
                report.append(f"  Duration: {data.get('duration_ms', 0):.0f}ms")
                if not data.get('success'):
                    report.append(f"  Error: {data.get('error', '')}")
                total_duration += data.get('duration_ms', 0)
            
            elif event_type == "agent.end":
                report.append(f"\n[{ts}] 🏁 Agent finished")
                report.append(f"  Status: {data.get('status')}")
                report.append(f"  Total iterations: {data.get('total_iterations')}")
        
        report.append(f"\n=== Summary ===")
        report.append(f"Total tokens: {total_tokens}")
        report.append(f"Total cost: ${total_cost:.6f}")
        report.append(f"Total duration: {total_duration:.0f}ms")
        
        return "\n".join(report)

输出示例：

=== Agent Trace Replay: a1b2c3d4 ===

[2026-06-28T10:00:01] 🚀 Agent started
  Query: 分析2026年AI市场趋势
  Tools: ['web_search', 'data_analyzer', 'chart_generator']
[2026-06-28T10:00:01] 📤 LLM call → gpt-5
  Input tokens: 1523
[2026-06-28T10:00:03] 📥 LLM response ← gpt-5
  Output tokens: 456
  Duration: 1820ms
  Cost: $0.013728
  Tool calls: [{'tool': 'web_search', 'args_preview': '{"q": "2026 AI market trends"}'}]
[2026-06-28T10:00:03] 🔧 ✅ web_search
  Duration: 340ms
[2026-06-28T10:00:04] 📤 LLM call → gpt-5
  Input tokens: 3245
[2026-06-28T10:00:06] 📥 LLM response ← gpt-5
  Output tokens: 890
  Duration: 2100ms
  Cost: $0.030950

[2026-06-28T10:00:06] 🏁 Agent finished
  Status: success
  Total iterations: 2

=== Summary ===
Total tokens: 6114
Total cost: $0.044678
Total duration: 4260ms

六、日志存储策略

日志类型存储介质保留期查询需求
实时日志Redis Stream1小时实时监控
近期日志Elasticsearch30天搜索分析
归档日志S3/OSS1年合规审计
聚合指标Prometheus90天趋势分析

七、日志设计 Checklist

□ 所有日志 JSON 结构化
□ trace_id 贯穿请求全链路
□ 每步记录 Token 和成本
□ PII 自动脱敏
□ 日志按级别过滤（生产 INFO+）
□ 文件轮转防止磁盘满
□ 日志索引支持快速检索
□ 执行链路可回放
□ 慢/贵/失败请求可快速查找
□ 日志导出支持合规审计

结语

结构化日志是 Agent 可观测性的数据基础。当你能查询"过去24小时所有使用 web_search 工具且耗时超过30秒的 Agent 会话"时，你就拥有了理解和优化 Agent 的能力。好的日志不是事后补救的调试工具，而是架构设计的一部分——在设计 Agent 时就设计好要记录什么。让每一步都可追溯，让每个决策都可理解，让每次故障都可回放。

加入讨论

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

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