引言 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的跨服务传播让分布式追踪成为可能。智能采样策略确保在控制成本的同时不丢失关键信息。
...
