引言

Agent系统的日志不仅是排障工具，更是质量改进和安全审计的数据基础。一次Agent对话可能涉及路由决策、记忆检索、工具调用、LLM推理等多个步骤，跨越多个微服务。如何在分布式环境中建立完整的日志链路，是Agent系统可观测性的核心挑战。

日志架构全景

┌──────────────────────────────────────────────────────┐
│                    日志数据流                          │
│                                                      │
│  Agent Services                                      │
│  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐                   │
│  │Router│ │Tool │ │LLM  │ │Memory│                   │
│  └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘                   │
│     │       │       │       │                        │
│     ▼       ▼       ▼       ▼                        │
│  ┌──────────────────────────────┐                    │
│  │    Fluent Bit (采集)          │                    │
│  └──────────┬───────────────────┘                    │
│             │                                         │
│     ┌───────┼───────┐                                │
│     ▼               ▼                                │
│  ┌──────┐    ┌──────────┐                           │
│  │ Loki │    │Elasticsearch│                          │
│  │(日志) │    │  (全文搜索) │                          │
│  └──────┘    └──────────┘                           │
│     │               │                                │
│     └───────┬───────┘                                │
│             ▼                                        │
│  ┌──────────────┐                                   │
│  │   Grafana    │                                   │
│  │  (可视化)     │                                   │
│  └──────────────┘                                   │
└──────────────────────────────────────────────────────┘

结构化日志标准

import structlog
from datetime import datetime
import uuid

# 结构化日志配置
structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),
)

logger = structlog.get_logger()

class AgentLogger:
    """Agent专用日志器"""
    
    @staticmethod
    def log_request(
        session_id: str,
        request_id: str,
        user_input: str,
        route_decision: dict
    ):
        """记录请求日志"""
        logger.info(
            "agent_request_received",
            session_id=session_id,
            request_id=request_id,
            user_input_length=len(user_input),
            user_input_preview=user_input[:100],
            route_model=route_decision.get("model"),
            route_tools=route_decision.get("tools"),
            timestamp=datetime.now().isoformat()
        )
    
    @staticmethod
    def log_tool_call(
        session_id: str,
        request_id: str,
        tool_name: str,
        params: dict,
        result: dict,
        latency_ms: float,
        success: bool
    ):
        """记录工具调用日志"""
        logger.info(
            "tool_call_completed",
            session_id=session_id,
            request_id=request_id,
            tool_name=tool_name,
            params_hash=hash(str(sorted(params.items()))),
            result_size=len(str(result)),
            latency_ms=latency_ms,
            success=success,
            error=result.get("error") if not success else None
        )
    
    @staticmethod
    def log_llm_call(
        session_id: str,
        request_id: str,
        model: str,
        prompt_tokens: int,
        completion_tokens: int,
        latency_ms: float,
        quality_score: float = None
    ):
        """记录LLM调用日志"""
        logger.info(
            "llm_call_completed",
            session_id=session_id,
            request_id=request_id,
            model=model,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
            latency_ms=latency_ms,
            quality_score=quality_score
        )
    
    @staticmethod
    def log_agent_decision(
        session_id: str,
        decision_type: str,
        reasoning: str,
        action: str,
        confidence: float
    ):
        """记录Agent决策日志——用于审计和改进"""
        logger.info(
            "agent_decision",
            session_id=session_id,
            decision_type=decision_type,  # route, tool_select, terminate
            reasoning=reasoning[:500],    # 截断推理过程
            action=action,
            confidence=confidence
        )

Trace ID传播

from opentelemetry import trace
from opentelemetry.propagate import inject, extract

class TracingMiddleware:
    """分布式追踪中间件"""
    
    async def __call__(self, request, call_next):
        # 提取或生成trace context
        context = extract(request.headers)
        
        tracer = trace.get_tracer(__name__)
        
        with tracer.start_as_current_span(
            "agent_request",
            context=context,
            attributes={
                "session_id": request.session_id,
                "user_id": request.user_id,
            }
        ) as span:
            # 在请求上下文中注入trace信息
            headers = {}
            inject(headers)
            request.trace_context = headers
            
            try:
                response = await call_next(request)
                span.set_attribute("response.status", "success")
                span.set_attribute(
                    "response.latency_ms",
                    response.latency_ms
                )
                return response
            except Exception as e:
                span.record_exception(e)
                span.set_status(trace.Status(trace.StatusCode.ERROR))
                raise


# 在服务间调用时传播Trace ID
class ServiceClient:
    """带Trace传播的服务客户端"""
    
    async def call_service(
        self,
        service: str,
        method: str,
        data: dict,
        trace_context: dict = None
    ) -> dict:
        """调用其他微服务"""
        headers = {
            "Content-Type": "application/json",
        }
        
        # 注入trace context
        if trace_context:
            headers.update(trace_context)
        else:
            inject(headers)  # 从当前context注入
        
        # 记录出站调用
        tracer = trace.get_tracer(__name__)
        with tracer.start_as_current_span(
            f"call_{service}",
            attributes={
                "peer.service": service,
                "http.method": method,
            }
        ) as span:
            response = await self.http_client.post(
                f"http://{service}/{method}",
                json=data,
                headers=headers
            )
            span.set_attribute("http.status_code", response.status_code)
            return response.json()

日志关联查询

class LogCorrelator:
    """日志关联查询器"""
    
    async def get_session_timeline(
        self,
        session_id: str
    ) -> list:
        """获取会话的完整事件时间线"""
        
        # 从多个数据源查询
        logs = await self.loki.query(
            f'{{session_id="{session_id}"}} | json'
        )
        traces = await self.jaeger.get_traces(
            tags={"session_id": session_id}
        )
        metrics = await self.prometheus.query_range(
            query=f'agent_session_metrics{{session_id="{session_id}"}}',
            start=..., end=...
        )
        
        # 合并并按时间排序
        events = []
        for log in logs:
            events.append({
                "type": "log",
                "timestamp": log["timestamp"],
                "level": log["level"],
                "message": log["message"],
                **log
            })
        
        for trace in traces:
            for span in trace.spans:
                events.append({
                    "type": "trace",
                    "timestamp": span.start_time,
                    "span_name": span.name,
                    "duration_ms": span.duration_ms,
                    "service": span.service,
                    **span.tags
                })
        
        events.sort(key=lambda e: e["timestamp"])
        return events

日志采样策略

class LogSampler:
    """日志采样器——在保证可观测性的前提下控制日志量"""
    
    SAMPLING_RULES = {
        # 正常请求：10%采样
        "normal": {"rate": 0.1, "level": "INFO"},
        
        # 错误请求：100%记录
        "error": {"rate": 1.0, "level": "ERROR"},
        
        # 慢请求（>5s）：100%记录
        "slow": {"rate": 1.0, "level": "INFO", "min_latency_ms": 5000},
        
        # 工具调用失败：100%记录
        "tool_failure": {"rate": 1.0, "level": "WARN"},
        
        # 安全相关：100%记录
        "security": {"rate": 1.0, "level": "INFO"},
        
        # 高价值用户：50%采样
        "enterprise": {"rate": 0.5, "level": "INFO"},
    }
    
    def should_log(
        self,
        log_type: str,
        request: dict,
        response: dict = None
    ) -> tuple:
        """判断是否需要记录日志"""
        
        # 优先级判断
        if response and response.get("error"):
            rule = self.SAMPLING_RULES["error"]
        elif response and response.get("latency_ms", 0) > 5000:
            rule = self.SAMPLING_RULES["slow"]
        elif request.get("user_tier") == "enterprise":
            rule = self.SAMPLING_RULES["enterprise"]
        else:
            rule = self.SAMPLING_RULES["normal"]
        
        import random
        if random.random() < rule["rate"]:
            return True, rule["level"]
        return False, None

日志分析

class LogAnalyzer:
    """日志分析器"""
    
    async def analyze_session(self, session_id: str) -> dict:
        """分析单个会话日志"""
        events = await self.log_store.get_session_events(session_id)
        
        analysis = {
            "session_id": session_id,
            "total_steps": len(events),
            "tool_calls": [],
            "llm_calls": [],
            "errors": [],
            "total_tokens": 0,
            "total_latency_ms": 0,
            "quality_indicators": {},
        }
        
        for event in events:
            if event["type"] == "tool_call":
                analysis["tool_calls"].append({
                    "tool": event["tool_name"],
                    "latency_ms": event["latency_ms"],
                    "success": event["success"]
                })
                if not event["success"]:
                    analysis["errors"].append(event)
            
            elif event["type"] == "llm_call":
                analysis["llm_calls"].append({
                    "model": event["model"],
                    "tokens": event["total_tokens"],
                    "latency_ms": event["latency_ms"]
                })
                analysis["total_tokens"] += event["total_tokens"]
            
            analysis["total_latency_ms"] += event.get("latency_ms", 0)
        
        return analysis
    
    async def detect_anomalies(
        self,
        time_window_hours: int = 1
    ) -> list:
        """检测日志异常模式"""
        anomalies = []
        
        # 1. 突发错误聚集
        error_clusters = await self._find_error_clusters(time_window_hours)
        for cluster in error_clusters:
            anomalies.append({
                "type": "error_cluster",
                "service": cluster["service"],
                "error_count": cluster["count"],
                "time_range": cluster["range"]
            })
        
        # 2. 异常Token消耗
        token_outliers = await self._find_token_outliers(time_window_hours)
        for outlier in token_outliers:
            anomalies.append({
                "type": "token_anomaly",
                "session_id": outlier["session_id"],
                "tokens": outlier["tokens"],
                "expected": outlier["expected"]
            })
        
        # 3. 工具调用模式异常
        tool_anomalies = await self._find_tool_anomalies(time_window_hours)
        anomalies.extend(tool_anomalies)
        
        return anomalies

日志保留策略

日志类型热存储（SSD）温存储（HDD）冷存储（S3）
错误日志7天30天180天
安全审计30天180天2年
正常请求3天14天90天
Trace数据3天7天30天
指标数据7天90天365天

总结

Agent系统的日志架构需要在"详细度"和"成本"之间取得平衡。结构化日志是一切的基础——没有结构化，就无法进行有效的查询和分析。Trace ID的跨服务传播让分布式追踪成为可能。智能采样策略确保在控制成本的同时不丢失关键信息。

核心原则：日志的价值不在于"记录了什么"，而在于"能找到什么"。好的日志架构让排障时间从小时级降到分钟级，让系统改进从"凭感觉"变成"看数据"。

加入讨论

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

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