可观测性

引言 Agent 的执行过程是黑盒——你看到输入和输出，但中间发生了什么？调用了什么工具？为什么选择这个路径？Token 花在哪里？结构化日志是打开这个黑盒的钥匙。2026年，随着 Agent 系统复杂度增长，日志不再是"给人看的文本"，而是"给系统查询的数据"。 一、Agent 日志设计原则 传统日志 vs Agent 日志 维度 传统日志 Agent 日志 格式 半结构化文本 全结构化 JSON 粒度 请求级 步骤级（每轮迭代） 关联 request_id trace_id + session_id + step_id 内容 状态和错误 决策推理、工具调用、Token消耗 用途 故障排查 故障排查 + 质量分析 + 成本归因 查询 grep/正则 结构化查询 + 聚合分析 设计原则 一切皆结构化：每条日志都是可查询的 JSON 因果链完整：从输入到输出的每一步都可追溯 上下文丰富：每条日志携带足够的上下文独立理解 成本感知：Token 和费用信息嵌入每条日志 隐私安全：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) 输出示例： ...

LangChain 2026：从链到图的范式跃迁 2024 年初，LangChain 创始人 Harrison Chase 做了一个大胆的决定：将框架核心从"链"（Chain）转向"图"（Graph）。两年后的今天，这个决定被证明是极具前瞻性的。LangChain 2026 生态已经发展为一个包含 LangGraph、LangSmith、LangServe、LangChain CLI 在内的完整工具矩阵，月活开发者超过 120 万，成为 Agent 开发领域事实上的标准。 核心组件架构 LangGraph：状态机驱动的 Agent 编排 LangGraph 是 2026 年 LangChain 生态中最核心的组件。与传统的链式调用不同，LangGraph 采用有向有环图（Directed Cyclic Graph）来建模 Agent 工作流，原生支持循环、分支、并行和人工干预。 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] context: dict iteration: int def research_node(state: AgentState): # 调用搜索工具获取信息 result = search_tool.invoke(state["messages"][-1].content) return {"messages": [{"role": "tool", "content": result}]} def reasoning_node(state: AgentState): # LLM 推理节点 response = llm.invoke(state["messages"]) return {"messages": [response]} def should_continue(state: AgentState) -> str: last_msg = state["messages"][-1] if state["iteration"] >= 5: return END if last_msg.tool_calls: return "tools" return END # 构建工作流图 workflow = StateGraph(AgentState) workflow.add_node("research", research_node) workflow.add_node("reasoning", reasoning_node) workflow.add_node("tools", tool_executor) workflow.set_entry_point("reasoning") workflow.add_conditional_edges("reasoning", should_continue) workflow.add_edge("tools", "reasoning") app = workflow.compile(checkpointer=MemorySaver()) LangGraph 2026 关键特性 特性 2024 版本 2026 版本 状态管理 基础状态字典 类型安全的 TypedDict + Reducer 并行执行 不支持 原生 Fan-out/Fan-in 持久化 基础 Memory SQLite/PostgreSQL/Redis 多后端 人工干预 基础中断 细粒度审批流 + 超时机制 流式输出 仅支持文本 事件流 + Token 级流式 子图嵌套 不支持 多级子图 + 状态隔离 时间旅行 不支持 完整状态回放 + 分支执行 LangSmith：全链路可观测性平台 LangSmith 在 2026 年已经从单纯的调试工具进化为完整的 LLM 可观测性平台。它提供： ...

引言 Agent 系统的"黑盒"问题是生产化的最大障碍之一。一个 Agent 调用了 3 个工具、经过 5 轮推理、消耗了 8000 tokens，但出了问题你却不知道在哪一步。2026年，OpenTelemetry 社区正式发布了 SemConv for GenAI 规范，为 LLM 可观测性提供了标准化方案。 一、为什么 Agent 可观测性不同于传统应用 传统微服务的可观测性关注：请求路径、延迟分布、错误率。Agent 系统增加了三个新维度： Token 维度：每次调用消耗多少 Token？成本如何分摊？ 推理维度：模型"想"了什么？为什么选择这个工具？为什么跳过某步？ 非确定性维度：相同输入可能产生不同输出，仅靠日志无法复现 二、OpenTelemetry GenAI 语义规范 2026年正式定稿的 GenAI SemConv 定义了以下核心 Attributes： # GenAI 基础属性 gen_ai.system: "openai" # 提供商 gen_ai.request.model: "gpt-5" # 模型名称 gen_ai.request.temperature: 0.7 # 采样温度 gen_ai.request.max_tokens: 4096 # 最大 Token # Token 使用 gen_ai.usage.input_tokens: 1523 # 输入 Token gen_ai.usage.output_tokens: 876 # 输出 Token gen_ai.usage.cost: 0.0234 # 本次调用成本（美元） # Agent 特有 gen_ai.agent.name: "research-agent" gen_ai.agent.tool.name: "web_search" gen_ai.agent.tool.result.quality: 0.85 gen_ai.agent.iteration: 3 # 第几轮迭代 # 工具调用 gen_ai.tool.name: "calculator" gen_ai.tool.input: '{"expr": "2+2"}' gen_ai.tool.output: '{"result": 4}' gen_ai.tool.duration_ms: 45 三、全链路追踪实现 架构概览 ┌─────────────────────────────────────────────────────┐ │ Agent Application │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │Step1│──►│Step2│──►│Step3│──►│Step4│ │ │ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ ┌──▼─────────▼─────────▼─────────▼──┐ │ │ │ OpenTelemetry SDK │ │ │ │ (Auto-instrumentation + Custom) │ │ │ └──────────────┬────────────────────┘ │ └─────────────────┼───────────────────────────────────┘ │ OTLP/gRPC ┌────────────▼────────────┐ │ OTel Collector │ │ (处理/采样/导出) │ └──┬─────┬─────┬──────────┘ │ │ │ ┌────▼┐ ┌─▼──┐ ┌▼─────┐ │Jaeger│ │Prom│ │Loki │ │(Trace)│ │(Met)│ │(Log)│ └─────┘ └────┘ └──────┘ Python 实现 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.instrumentation.openai import OpenAIInstrumentor # 1. 初始化 OTel provider = TracerProvider() processor = BatchSpanProcessor( OTLPSpanExporter(endpoint="http://otel-collector:4317") ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. 自动注入 OpenAI 调用的 Span OpenAIInstrumentor().instrument() # 3. Agent 自定义 Span tracer = trace.get_tracer("agent-system") class ObservabilityMiddleware: """Agent 可观测性中间件""" def __init__(self): self.tracer = trace.get_tracer("agent") async def on_agent_start(self, agent_name: str, input_data: dict): """Agent 启动时创建 Root Span""" self.root_span = self.tracer.start_span( f"agent.{agent_name}", attributes={ "gen_ai.agent.name": agent_name, "agent.input.size": len(str(input_data)), } ) async def on_llm_call(self, model: str, messages: list, **kwargs): """LLM 调用前记录""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"llm.{model}", context=ctx, attributes={ "gen_ai.request.model": model, "gen_ai.request.message_count": len(messages), "gen_ai.request.temperature": kwargs.get("temperature", 1.0), } ) return span async def on_llm_end(self, span, response): """LLM 调用后记录 Token 使用""" usage = response.usage span.set_attributes({ "gen_ai.usage.input_tokens": usage.prompt_tokens, "gen_ai.usage.output_tokens": usage.completion_tokens, "gen_ai.usage.total_tokens": usage.total_tokens, "gen_ai.usage.cost": calculate_cost( usage.prompt_tokens, usage.completion_tokens, response.model ), }) span.end() async def on_tool_call(self, tool_name: str, tool_input: dict): """工具调用追踪""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"tool.{tool_name}", context=ctx, attributes={ "gen_ai.tool.name": tool_name, "gen_ai.tool.input": json.dumps(tool_input)[:500], } ) return span async def on_agent_end(self, output: str): """Agent 结束""" self.root_span.set_attributes({ "agent.output.size": len(output), "agent.status": "success", }) self.root_span.end() Trace 可视化示例 在 Jaeger 中看到的典型 Agent Trace： ...

概述 Agent可观测性架构是AI智能体领域中Agent可观测性架构的重要主题。本文将从多个角度深入分析这一话题，为读者提供系统性的认知框架和实践参考。 核心概念 基本定义 在深入讨论之前，我们需要明确几个核心概念。AI智能体是指能够感知环境、理解指令、规划行动并调用工具完成任务的AI系统。与传统的聊天机器人不同，智能体具有自主性、目标导向性和工具使用能力。 Agent可观测性架构涉及的关键技术包括： 大语言模型：作为智能体的认知引擎，负责理解、推理和生成 工具调用：通过Function Calling或MCP协议与外部系统交互 记忆系统：短期记忆处理当前对话，长期记忆存储历史经验 规划引擎：将复杂任务分解为可执行的子步骤 技术原理 从技术层面看，Agent可观测性架构的核心在于如何让AI系统更好地理解和执行人类意图。这涉及多个技术环节的协同： 首先是感知层，智能体需要准确理解用户的自然语言指令，提取关键信息和约束条件。其次是规划层，将高层目标分解为具体的执行步骤。然后是执行层，调用合适的工具完成每个步骤。最后是反馈层，根据执行结果调整后续策略。 实践分析 当前现状 在架构设计领域，当前的技术实践呈现出几个明显特征： 工程化程度提升：从实验室原型到生产级系统，工程能力成为关键差异化因素 评估体系完善：越来越多标准化的评测基准被提出，帮助开发者量化能力边界 开源生态繁荣：开源框架和工具链的成熟降低了开发门槛 安全意识增强：对AI安全和对齐问题的重视程度显著提升 关键挑战 尽管进展显著，Agent可观测性架构仍面临几个核心挑战： 技术挑战： 大模型的幻觉问题在智能体场景下被放大，因为智能体需要做出实际决策 多步推理中的错误累积效应导致长程任务成功率下降 工具调用的可靠性受外部API稳定性影响 工程挑战： 智能体的可观测性不足，调试和排错困难 成本控制与性能优化的平衡 从单机到分布式部署的架构复杂性 安全挑战： Prompt注入等攻击手段不断进化 智能体权限管理需要更精细化的控制 数据隐私保护在多Agent协作场景下更加复杂 优化策略 针对上述挑战，以下是几个关键优化方向： 技术优化 分而治之：将复杂任务分解为可独立验证的子任务，降低单步错误影响 多路投票：对关键决策使用多次采样投票机制，提高可靠性 渐进式信任：智能体权限从最小化开始，根据表现逐步扩展 人在回路：高风险决策保留人工审核环节 工程优化 可观测性优先：建立完善的日志、指标和追踪体系 灰度发布：新版本智能体先在小流量环境验证 自动化测试：构建端到端测试套件，防止回归 成本监控：实时追踪Token消耗和API调用成本 案例研究 为了更具体地说明Agent可观测性架构的实践价值，我们来看一个典型场景： 某科技公司在内部IT运维中部署了AI智能体，负责处理员工的工单请求。智能体需要理解员工的自然语言描述，判断问题类型，查询知识库，执行修复操作或转接人工。 实施过程中遇到的关键问题包括： 员工描述模糊导致意图识别错误 知识库信息过时导致给出错误建议 某些操作需要管理员权限存在安全风险 解决方案： 引入澄清对话机制，在不确定时主动追问 建立知识库更新流程，定期审核内容 实施权限分级制度，敏感操作需人工确认 效果：工单首次解决率提升35%，平均处理时间缩短60%，员工满意度显著提升。 未来趋势 Agent可观测性架构的发展趋势值得关注： 标准化：MCP等开放协议将推动工具接口标准化，降低集成成本 垂直化：针对特定行业和场景的专用智能体将大量涌现 协作化：多智能体协作将成为复杂任务的标准解决方案 自主化：智能体的自主决策能力将持续提升，但需要配套的安全机制 结论 Agent可观测性架构是AI智能体技术发展中的重要一环。无论是技术原理的深入理解，还是实践中的工程优化，都需要系统性思维。对于开发者和企业而言，关键在于： 理解技术能力和边界，避免过度期待 建立系统化的评估和监控体系 在创新和安全之间找到平衡 持续学习和适应快速变化的技术生态 硅基AGI探索者将持续关注架构设计领域的最新进展，为读者提供深度分析和实践指导。— ...

为什么智能体需要专属的可观测性？ 传统软件的可观测性聚焦于 CPU、内存、延迟等系统指标。但 AI 智能体引入了全新的可观测维度： Token 消耗：每次 LLM 调用都有成本，需要精确追踪 推理链路：Agent 可能经历多轮 Thought → Action → Observation 循环，需要完整 Trace 工具调用质量：工具是否被正确调用？返回结果是否有效？ Prompt 效果：不同 Prompt 版本对输出质量的影响如何？ 幻觉检测：模型输出是否与事实相符？ 缺乏可观测性的智能体就像一个黑盒——你不知道它在想什么，也不知道它为什么出错。本文将带你从零搭建一套生产级的智能体可观测性平台。 可观测性三支柱在 AI 场景下的重构 传统可观测性的三支柱是 Metrics、Logs、Traces。在智能体场景下，我们需要将其扩展为五支柱： 支柱 传统场景 智能体场景 Traces 请求链路追踪 Thought-Action-Observation 链路 Metrics QPS、延迟、错误率 Token 用量、工具调用成功率、幻觉率 Logs 结构化日志 Prompt/Completion 完整记录 Evaluations N/A 输出质量自动评估 Cost N/A Token 成本与预算控制 架构设计 ┌────────────────────────────────────────────────────────┐ │ Agent Application │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────────┐ │ │ │ LLM Call │ │ Tool Call │ │ Observation Callback │ │ │ └─────┬─────┘ └─────┬────┘ └──────────┬───────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Observability SDK │ │ │ │ (OpenTelemetry + Custom Spans + Token Counter) │ │ │ └──────────────────────┬──────────────────────────┘ │ └─────────────────────────┼──────────────────────────────┘ │ ┌───────────────┼───────────────┐ ▼ ▼ ▼ ┌─────────────┐ ┌─────────────┐ ┌───────────┐ │ Jaeger │ │ Prometheus │ │ Postgres │ │ (Traces) │ │ (Metrics) │ │ (Logs) │ └─────────────┘ └─────────────┘ └───────────┘ │ │ │ └───────────────┼───────────────┘ ▼ ┌─────────────┐ │ Grafana │ │ (Dashboard) │ └─────────────┘ 核心组件实现 1. Trace 追踪：基于 OpenTelemetry 扩展 智能体的 Trace 与传统微服务不同，需要记录 LLM 特有的 Span 属性： ...

1. 为什么 Agent 需要专门的可观测性 传统微服务可观测性关注请求-响应链路。Agent 系统则复杂得多： 多步骤推理：一个用户请求可能触发 10-50 次 LLM 调用 不确定输出：相同输入可能产生不同输出，难以复现问题 工具调用链：Agent → tool1 → Agent → tool2 → … 形成深层调用链 成本高昂：每次 LLM 调用都有实际成本，需要精准归因 幻觉风险：输出质量难以量化监控 普通监控工具（Prometheus/Grafana）只覆盖指标，无法完整追踪 Agent 行为。需要专门的 Agent 可观测性方案。 2. 可观测性三大支柱 ┌─────────────────────────────────────────────────┐ │ Agent Observability │ ├────────────┬──────────────┬────────────────────┤ │ Traces │ Metrics │ Logs │ │ (追踪) │ (指标) │ (日志) │ │ │ │ │ │ ┌────────┐ │ ┌────────┐ │ ┌──────────────┐ │ │ │Span │ │ │Counter │ │ │ Struct- │ │ │ │Tree │ │ │Gauge │ │ │ ured Log │ │ │ │Timeline│ │ │Histog. │ │ │ Event Log │ │ │ └────────┘ │ └────────┘ │ └──────────────┘ │ │ │ │ │ │ "发生了 │ "发生了几次" │ "发生了什么" │ │ 什么" │ "耗时多久" │ "为什么失败" │ └────────────┴──────────────┴────────────────────┘ 统一关联：TraceID + SpanID + LogID 3. 分布式追踪设计 3.1 Agent Span 模型 from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Optional import uuid class SpanKind(str, Enum): LLM_CALL = "llm_call" TOOL_CALL = "tool_call" AGENT_STEP = "agent_step" RETRIEVAL = "retrieval" USER_INPUT = "user_input" SYSTEM = "system" @dataclass class SpanContext: trace_id: str span_id: str parent_span_id: Optional[str] = None @dataclass class SpanEvent: name: str timestamp: datetime attributes: dict = field(default_factory=dict) @dataclass class AgentSpan: """Agent 追踪 Span - 兼容 OpenTelemetry""" trace_id: str span_id: str parent_span_id: Optional[str] kind: SpanKind name: str agent_id: str session_id: str # 时间 start_time: datetime = field(default_factory=datetime.now) end_time: Optional[datetime] = None duration_ms: Optional[float] = None # 输入/输出 input: dict = field(default_factory=dict) output: dict = field(default_factory=dict) # 状态 status: str = "ok" # ok, error, timeout error_message: Optional[str] = None # 成本 & 性能 token_usage: dict = field(default_factory=dict) # {input: N, output: N} cost_usd: float = 0.0 model: Optional[str] = None # 扩展属性 attributes: dict = field(default_factory=dict) events: list[SpanEvent] = field(default_factory=list) child_spans: list[str] = field(default_factory=list) # child span_ids def finish(self, status: str = "ok"): self.end_time = datetime.now() self.duration_ms = (self.end_time - self.start_time).total_seconds() * 1000 self.status = status def add_event(self, name: str, attributes: dict = None): self.events.append(SpanEvent( name=name, timestamp=datetime.now(), attributes=attributes or {} )) 3.2 追踪器实现 class AgentTracer: """Agent 分布式追踪器""" def __init__(self, exporter: "TraceExporter"): self.exporter = exporter self._active_spans: dict[str, AgentSpan] = {} # span_id -> span self._trace_tree: dict[str, list[str]] = {} # trace_id -> [span_ids] def start_trace(self, session_id: str, agent_id: str, name: str = "agent_session") -> tuple[str, str]: """开始一次 Agent 会话追踪""" trace_id = str(uuid.uuid4()) root_span_id = self._new_span_id() span = AgentSpan( trace_id=trace_id, span_id=root_span_id, parent_span_id=None, kind=SpanKind.AGENT_STEP, name=name, agent_id=agent_id, session_id=session_id, ) self._active_spans[root_span_id] = span self._trace_tree[trace_id] = [root_span_id] return trace_id, root_span_id def start_span(self, trace_id: str, parent_span_id: str, kind: SpanKind, name: str, agent_id: str, session_id: str, input_data: dict = None) -> str: span_id = self._new_span_id() span = AgentSpan( trace_id=trace_id, span_id=span_id, parent_span_id=parent_span_id, kind=kind, name=name, agent_id=agent_id, session_id=session_id, input=input_data or {}, ) self._active_spans[span_id] = span self._trace_tree.setdefault(trace_id, []).append(span_id) # 更新父 span 的 child 列表 if parent_span_id in self._active_spans: self._active_spans[parent_span_id].child_spans.append(span_id) return span_id def finish_span(self, span_id: str, output_data: dict = None, status: str = "ok", error: str = None): if span_id not in self._active_spans: return span = self._active_spans[span_id] span.finish(status) if output_data: span.output = output_data if error: span.error_message = error # 异步导出 asyncio.create_task(self.exporter.export_span(span)) def get_trace(self, trace_id: str) -> list[AgentSpan]: span_ids = self._trace_tree.get(trace_id, []) return [self._active_spans[sid] for sid in span_ids if sid in self._active_spans] @staticmethod def _new_span_id() -> str: return str(uuid.uuid4())[:16] # 短 ID，便于展示 3.3 OpenTelemetry 集成 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter class OTelIntegration: """将 Agent Span 桥接到 OpenTelemetry""" def __init__(self, endpoint: str = "localhost:4317"): self.provider = TracerProvider() self.exporter = OTLPSpanExporter(endpoint=endpoint) self.processor = BatchSpanProcessor(self.exporter) self.provider.add_span_processor(self.processor) trace.set_tracer_provider(self.provider) self.tracer = trace.get_tracer("agent-system") def to_otel_span(self, agent_span: AgentSpan): """将 AgentSpan 转换为 OTel Span""" with self.tracer.start_as_current_span( agent_span.name, context=self._make_context(agent_span), kind=self._map_kind(agent_span.kind), ) as otel_span: # 设置属性 otel_span.set_attribute("agent.id", agent_span.agent_id) otel_span.set_attribute("session.id", agent_span.session_id) otel_span.set_attribute("llm.model", agent_span.model or "") otel_span.set_attribute("cost.usd", agent_span.cost_usd) for k, v in agent_span.token_usage.items(): otel_span.set_attribute(f"token.{k}", v) # 设置状态 if agent_span.status == "error": otel_span.set_status(trace.Status(trace.StatusCode.ERROR, agent_span.error_message)) # 添加事件 for evt in agent_span.events: otel_span.add_event(evt.name, evt.attributes) def _map_kind(self, kind: SpanKind): mapping = { SpanKind.LLM_CALL: trace.SpanKind.CLIENT, SpanKind.TOOL_CALL: trace.SpanKind.INTERNAL, SpanKind.AGENT_STEP: trace.SpanKind.INTERNAL, SpanKind.RETRIEVAL: trace.SpanKind.CLIENT, } return mapping.get(kind, trace.SpanKind.INTERNAL) 4. 指标采集 4.1 关键指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary import time class AgentMetrics: """Agent 系统关键指标""" def __init__(self, namespace: str = "agent"): # LLM 调用指标 self.llm_requests = Counter( f"{namespace}_llm_requests_total", "Total LLM requests", ["model", "agent_id", "status"] ) self.llm_latency = Histogram( f"{namespace}_llm_latency_seconds", "LLM request latency", ["model", "agent_id"], buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0] ) self.llm_tokens = Counter( f"{namespace}_llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) self.llm_cost = Counter( f"{namespace}_llm_cost_usd_total", "Total LLM cost in USD", ["model", "agent_id"] ) # Agent 行为指标 self.agent_steps = Counter( f"{namespace}_agent_steps_total", "Total agent reasoning steps", ["agent_id", "step_type"] ) self.agent_session_duration = Histogram( f"{namespace}_agent_session_duration_seconds", "Agent session duration", ["agent_id", "outcome"] # outcome: success/failure/timeout ) self.agent_tool_calls = Counter( f"{namespace}_agent_tool_calls_total", "Total tool calls", ["agent_id", "tool_name", "status"] ) # 系统指标 self.active_sessions = Gauge( f"{namespace}_active_sessions", "Current active sessions", ["agent_id"] ) self.queue_depth = Gauge( f"{namespace}_queue_depth", "Current queue depth", ["queue_name"] ) self.error_rate = Summary( f"{namespace}_error_rate", "Error rate over 5m window", ["agent_id", "error_type"] ) def record_llm_call(self, model: str, agent_id: str, duration: float, input_tokens: int, output_tokens: int, cost: float, status: str): self.llm_requests.labels(model=model, agent_id=agent_id, status=status).inc() self.llm_latency.labels(model=model, agent_id=agent_id).observe(duration) self.llm_tokens.labels(model=model, direction="input").inc(input_tokens) self.llm_tokens.labels(model=model, direction="output").inc(output_tokens) self.llm_cost.labels(model=model, agent_id=agent_id).inc(cost) def record_agent_step(self, agent_id: str, step_type: str): self.agent_steps.labels(agent_id=agent_id, step_type=step_type).inc() def record_tool_call(self, agent_id: str, tool_name: str, status: str): self.agent_tool_calls.labels(agent_id=agent_id, tool_name=tool_name, status=status).inc() 4.2 成本归因指标 from collections import defaultdict class CostAttribution: """成本归因：按用户/会话/功能分解成本""" def __init__(self, redis_client): self.redis = redis_client async def record_cost(self, trace_id: str, user_id: str, feature: str, model: str, cost: float, tokens: int): pipe = self.redis.pipeline() # 按用户聚合 pipe.incrbyfloat(f"cost:user:{user_id}:total", cost) pipe.incrby(f"cost:user:{user_id}:tokens", tokens) # 按功能聚合 pipe.incrbyfloat(f"cost:feature:{feature}:total", cost) # 按模型聚合 pipe.incrbyfloat(f"cost:model:{model}:total", cost) # 按 trace 明细 pipe.hset(f"cost:trace:{trace_id}", mapping={ "user_id": user_id, "feature": feature, "model": model, "cost": str(cost), "tokens": str(tokens), "timestamp": str(time.time()) }) await pipe.execute() async def get_user_cost_report(self, user_id: str, days: int = 7) -> dict: return { "total_cost_usd": float(await self.redis.get(f"cost:user:{user_id}:total") or 0), "total_tokens": int(await self.redis.get(f"cost:user:{user_id}:tokens") or 0), "by_feature": await self._get_by_dimension(f"cost:feature", user_id, days), "by_model": await self._get_by_dimension(f"cost:model", user_id, days), } 5. 结构化日志 5.1 日志格式规范 import json import logging import sys from datetime import datetime from typing import Any class StructuredLogger: """结构化日志：JSON 格式，便于 ELK/Loki 检索""" def __init__(self, service_name: str, level: str = "INFO"): self.service = service_name self.logger = logging.getLogger(service_name) self.logger.setLevel(getattr(logging, level)) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(self) self.logger.addHandler(handler) def format(self, record: logging.LogRecord) -> str: log_entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "level": record.levelname, "service": self.service, "message": record.getMessage(), "trace_id": getattr(record, "trace_id", None), "span_id": getattr(record, "span_id", None), "session_id": getattr(record, "session_id", None), "agent_id": getattr(record, "agent_id", None), "user_id": getattr(record, "user_id", None), } # 添加额外字段 if hasattr(record, "extra_fields"): log_entry.update(record.extra_fields) return json.dumps(log_entry, ensure_ascii=False) def bind(self, **kwargs) -> "BoundLogger": return BoundLogger(self, kwargs) class BoundLogger: """绑定上下文的 Logger""" def __init__(self, logger: StructuredLogger, bound: dict): self._logger = logger self._bound = bound def _log(self, level: str, msg: str, **kwargs): extra = {**self._bound, **kwargs} record = self._logger.logger.makeRecord( self._logger.logger.name, getattr(logging, level), "(unknown)", 0, msg, [], None ) record.extra_fields = extra for k, v in extra.items(): setattr(record, k, v) self._logger.logger.handle(record) def info(self, msg: str, **kwargs): self._log("INFO", msg, **kwargs) def warning(self, msg: str, **kwargs): self._log("WARNING", msg, **kwargs) def error(self, msg: str, **kwargs): self._log("ERROR", msg, **kwargs) def debug(self, msg: str, **kwargs): self._log("DEBUG", msg, **kwargs) 5.2 日志事件类型 class AgentLogEvents: """Agent 关键生命周期事件""" @staticmethod def session_started(logger: BoundLogger, session_id: str, user_id: str): logger.info("agent.session.started", session_id=session_id, user_id=user_id, event_type="session_lifecycle") @staticmethod def llm_call_started(logger: BoundLogger, model: str, prompt_length: int): logger.info("llm.call.started", model=model, prompt_length=prompt_length, event_type="llm_call") @staticmethod def llm_call_completed(logger: BoundLogger, model: str, latency_ms: float, input_tokens: int, output_tokens: int, cost: float): logger.info("llm.call.completed", model=model, latency_ms=latency_ms, input_tokens=input_tokens, output_tokens=output_tokens, cost_usd=cost, event_type="llm_call") @staticmethod def tool_call_started(logger: BoundLogger, tool_name: str, args: dict): logger.info("tool.call.started", tool_name=tool_name, tool_args=args, event_type="tool_call") @staticmethod def tool_call_completed(logger: BoundLogger, tool_name: str, success: bool, result_summary: str): logger.info("tool.call.completed", tool_name=tool_name, success=success, result_summary=result_summary[:200], event_type="tool_call") @staticmethod def hallucination_detected(logger: BoundLogger, claim: str, confidence: float): logger.warning("agent.hallucination.detected", claim=claim[:100], confidence=confidence, event_type="quality") @staticmethod def session_completed(logger: BoundLogger, session_id: str, total_steps: int, total_cost: float, outcome: str): logger.info("agent.session.completed", session_id=session_id, total_steps=total_steps, total_cost_usd=total_cost, outcome=outcome, event_type="session_lifecycle") 6. 完整可观测性集成 class ObservableAgent: """集成可观测性的 Agent 基类""" def __init__(self, agent_id: str, tracer: AgentTracer, metrics: AgentMetrics, logger: StructuredLogger): self.agent_id = agent_id self.tracer = tracer self.metrics = metrics self.logger = logger.bind(agent_id=agent_id) self._current_trace_id: str | None = None self._current_span_id: str | None = None async def run(self, user_query: str, user_id: str, session_id: str) -> str: # 开始追踪 trace_id, root_span_id = self.tracer.start_trace(session_id, self.agent_id) self._current_trace_id = trace_id self._current_span_id = root_span_id # 绑定 trace 上下文到日志 self.logger = self.logger.bind(trace_id=trace_id, session_id=session_id, user_id=user_id) session_start = time.time() try: self.logger.info("agent.session.started", query=user_query[:100]) self.metrics.active_sessions.labels(agent_id=self.agent_id).inc() # 执行 Agent 逻辑（带追踪） result = await self._execute_with_tracing(user_query) duration = time.time() - session_start self.metrics.agent_session_duration.labels( agent_id=self.agent_id, outcome="success" ).observe(duration) self.logger.info("agent.session.completed", duration_seconds=duration, outcome="success") self.tracer.finish_span(root_span_id, {"result": result}, "ok") return result except Exception as e: duration = time.time() - session_start self.metrics.agent_session_duration.labels( agent_id=self.agent_id, outcome="failure" ).observe(duration) self.logger.error("agent.session.failed", error=str(e), exc_info=True) self.tracer.finish_span(root_span_id, status="error", error=str(e)) raise finally: self.metrics.active_sessions.labels(agent_id=self.agent_id).dec() async def _execute_with_tracing(self, query: str) -> str: # LLM 调用 llm_span_id = self.tracer.start_span( self._current_trace_id, self._current_span_id, SpanKind.LLM_CALL, "llm_chat", self.agent_id, self._get_session_id(), {"query": query} ) llm_start = time.time() try: self.logger.info("llm.call.started") response = await self._call_llm(query) latency = time.time() - llm_start self.metrics.record_llm_call( model="gpt-4o", agent_id=self.agent_id, duration=latency, input_tokens=100, output_tokens=50, cost=0.003, status="ok" ) self.tracer.finish_span(llm_span_id, {"response": response}, "ok") return response except Exception as e: self.tracer.finish_span(llm_span_id, status="error", error=str(e)) raise def _get_session_id(self) -> str: return self.logger._bound.get("session_id", "") 7. 可观测性数据流水线 Agent 运行时 │ ├── Traces ──→ OTel Collector ──→ Jaeger/Tempo ──→ Grafana │ ├── Metrics ──→ Prometheus ──────────────────────→ Grafana │ └── Logs ────→ Loki/ELK ───────────────────────→ Grafana │ └──→ 告警规则 → AlertManager → Slack/PagerDuty 7.1 OpenTelemetry Collector 配置 # otel-collector-config.yaml receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 processors: batch: timeout: 5s send_batch_size: 100 memory_limiter: check_interval: 5s limit_mib: 1000 attributes: actions: - key: agent.id action: insert value: "unknown" resource: attributes: - key: deployment.environment value: "production" action: insert exporters: otlp/jaeger: endpoint: jaeger:4317 tls: insecure: true prometheus: endpoint: 0.0.0.0:8889 loki: endpoint: http://loki:3100/loki/api/v1/push service: pipelines: traces: receivers: [otlp] processors: [batch, memory_limiter, attributes] exporters: [otlp/jaeger] metrics: receivers: [otlp] processors: [batch, memory_limiter] exporters: [prometheus] logs: receivers: [otlp] processors: [batch, memory_limiter] exporters: [loki] 8. 告警规则 # prometheus-alerts.yaml groups: - name: agent_alerts rules: - alert: AgentHighErrorRate expr: | sum(rate(agent_agent_steps_total{status="error"}[5m])) by (agent_id) / sum(rate(agent_agent_steps_total[5m])) by (agent_id) > 0.05 for: 2m labels: severity: warning annotations: summary: "Agent {{ $labels.agent_id }} 错误率超过 5%" - alert: AgentHighLatency expr: | histogram_quantile(0.95, sum(rate(agent_llm_latency_seconds_bucket[5m])) by (le, agent_id) ) > 5 for: 2m labels: severity: warning annotations: summary: "Agent {{ $labels.agent_id }} P95 延迟超过 5 秒" - alert: AgentHighCost expr: | rate(agent_llm_cost_usd_total[1h]) > 10 for: 5m labels: severity: critical annotations: summary: "Agent 每小时成本超过 $10" - alert: AgentHallucinationDetected expr: | increase(agent_quality_hallucination_total[10m]) > 5 labels: severity: warning annotations: summary: "检测到 {{ $value }} 次幻觉" 9. 总结 Agent 可观测性的核心在于全链路关联： ...

为什么 LLM 需要专项可观测性？ 传统 APM 不够：LLM 有 Token 计费、Prompt 变体、模型路由、工具调用链等特有维度。一个请求可能涉及 3 个模型 + 5 个工具调用 + 2 次检索，没有 Tracing 根本无法定位问题。 三位一体架构 ┌──────────────────────────────────────────────────┐ │ 用户请求 │ │ trace_id = xxx │ └──────────────────────┬───────────────────────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Logs │ │ Traces │ │ Metrics │ │ 结构化 │ │ 链路 │ │ 聚合 │ │ 日志 │ │ 追踪 │ │ 指标 │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ ELK / │ │ Jaeger /│ │Prometheus│ │ Loki │ │ Langfuse│ │ + Grafana│ └─────────┘ └─────────┘ └─────────┘ │ │ │ └──────────────┼──────────────┘ ▼ ┌─────────────────┐ │ AlertManager │ │ 告警 + 通知 │ └─────────────────┘ 一、结构化日志 import structlog import json # 配置 structlog structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ], ) logger = structlog.get_logger() class LLMLogger: """LLM 专用结构化日志""" def log_request(self, trace_id: str, user_id: str, model: str, prompt: str, **kwargs): logger.info("llm_request", trace_id=trace_id, user_id=user_id, model=model, prompt_length=len(prompt), prompt_tokens=kwargs.get("input_tokens"), max_tokens=kwargs.get("max_tokens"), temperature=kwargs.get("temperature", 1.0), tools=kwargs.get("tools"), timestamp=datetime.utcnow().isoformat(), ) def log_response(self, trace_id: str, response: str, input_tokens: int, output_tokens: int, latency_ms: float, model: str, **kwargs): logger.info("llm_response", trace_id=trace_id, model=model, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, latency_ms=latency_ms, finish_reason=kwargs.get("finish_reason"), cost_usd=self._calc_cost(model, input_tokens, output_tokens), ) def log_tool_call(self, trace_id: str, tool_name: str, params: dict, result: dict, latency_ms: float): logger.info("tool_call", trace_id=trace_id, tool=tool_name, params_keys=list(params.keys()), result_status="success" if result.get("success") else "failed", latency_ms=latency_ms, ) def log_error(self, trace_id: str, error: Exception, context: dict): logger.error("llm_error", trace_id=trace_id, error_type=type(error).__name__, error_message=str(error), context=context, ) 日志查询示例 # ELK / Loki 查询：查找高延迟请求 # Kibana KQL: # llm_response AND latency_ms > 5000 AND model: "gpt-4" # Grafana Loki LogQL: # {app="llm-service"} |= "llm_response" | json | latency_ms > 5000 二、分布式链路追踪 LLM 请求的典型链路：API → Router → Cache → Model → Tool → Model → Response ...

LLM 应用的可观测性困境 当你的 LLM 应用从 Demo 走向生产，一系列问题随之而来：某个请求为什么返回了错误结果？Prompt 改动后效果到底是变好还是变差？不同模型的延迟和成本如何对比？传统 APM 工具（如 Datadog）对 LLM 场景缺乏原生支持。Langfuse 作为开源的 LLM 可观测性平台，正填补了这一空白。 Langfuse 核心功能 功能矩阵 功能模块 说明 Tracing 请求级全链路追踪，含 LLM 调用、工具调用、检索流程 Prompt Management Prompt 版本管理、AB 测试、在线编辑 Analytics 延迟、Token 用量、成本统计 Evaluation 人工标注 + 自动评估（基于规则或模型裁判） User Feedback 用户点赞/点踩采集与关联 Model Usage 多模型/多项目用量看板 Tracing：全链路追踪 集成方式 Python SDK（推荐） from langfuse import Langfuse from langfuse.decorators import observe langfuse = Langfuse() @observe() # 自动创建 trace def chat_bot(user_message: str): # 嵌套的 @observe 会自动建立父子关系 retrieved_docs = retrieve(user_message) # 被 trace answer = generate_response(user_message, retrieved_docs) # 被 trace return answer @observe(as_type="generation") def generate_response(question, docs): response = openai.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": f"基于以下内容回答：{docs}\n问题：{question}"}], ) return response.choices[0].message.content @observe(as_type="span") def retrieve(query): # 检索逻辑 return vector_db.search(query) OpenAI SDK 集成（Drop-in） from langfuse.openai import openai # 只需替换 import，所有 OpenAI 调用自动被 trace response = openai.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Hello"}], metadata={"user_id": "user_123"}, # 附加用户信息 ) LangChain 集成 from langfuse.callback import CallbackHandler handler = CallbackHandler() chain.invoke( {"input": "什么是量子计算？"}, config={"callbacks": [handler]} ) Trace 结构 Trace (chat_bot 调用) ├── Span (retrieve - 向量检索) │ └── Generation (embedding 调用) ├── Generation (generate_response - LLM 调用) │ ├── input: "基于以下内容..." │ ├── output: "量子计算是..." │ ├── model: gpt-4o │ ├── tokens: {input: 350, output: 280} │ └── cost: $0.0091 └── Span (后处理) 自定义 Span from langfuse import Langfuse langfuse = Langfuse() trace = langfuse.trace(name="rag-pipeline", user_id="user_123") span = trace.span(name="document-retrieval", input={"query": "量子计算"}) # ... 执行检索 ... span.end(output={"docs": [...]}) gen = trace.generation( name="answer-generation", model="gpt-4o", input={"messages": [...]}, output={"content": "..."}, usage={"prompt_tokens": 350, "completion_tokens": 280}, ) Prompt 管理 版本化 Prompt from langfuse import Langfuse langfuse = Langfuse() # 获取生产环境的 Prompt（自动缓存） prompt = langfuse.get_prompt( name="summarizer", version=2, # 指定版本，或省略获取最新 ) # 编译 Prompt（变量替换） compiled = prompt.compile( text="一段需要总结的长文本...", max_length="200" ) # 直接用于 LLM 调用 response = openai.chat.completions.create( model=prompt.config["model"], messages=[{"role": "user", "content": compiled}], ) Prompt AB 测试 import random # 50% 用户用 v1，50% 用 v2 version = random.choice(["v1", "v2"]) prompt = langfuse.get_prompt(name="customer-support", version=version) # Langfuse 自动记录每个请求使用了哪个版本 trace = langfuse.trace(name="support-chat") trace.update(metadata={"prompt_version": version}) 在线编辑 Langfuse Web UI 提供 Prompt 在线编辑器： ...

为什么 AI 应用需要可观测性 传统应用监控关注 CPU、内存、QPS。AI 应用需要关注：LLM 调用了几次？每次延迟多少？Token 消耗多少？检索结果是否相关？Agent 的推理链路是什么？没有可观测性，AI 应用就是黑盒，出问题只能猜。 三支柱模型 支柱 传统应用 AI 应用 工具 Tracing 请求链路 LLM 调用链 + Agent 推理步骤 + 检索过程 LangSmith/Langfuse Logging 日志 Prompt/Response/中间状态 结构化日志 Metrics QPS/延迟 Token消耗/检索召回率/幻觉率 Prometheus+Grafana 关键指标定义 from dataclasses import dataclass from typing import Optional @dataclass class LLMCallMetrics: # 延迟指标 time_to_first_token: float # 首 token 延迟(ms) total_latency: float # 总延迟(ms) # 成本指标 prompt_tokens: int completion_tokens: int total_cost: float # 质量指标 model_name: str temperature: float finish_reason: str # 检索指标(RAG场景) retrieval_count: Optional[int] = None retrieval_relevance: Optional[float] = None context_utilization: Optional[float] = None @dataclass class AgentTrace: trace_id: str spans: list[dict] # 每个步骤一个 span # span 结构: name, start, end, input, output, metadata Langfuse 集成实现 from langfuse import Langfuse from langfuse.decorators import langfuse_context, observe import functools, time langfuse = Langfuse( public_key="pk-lf-xxx", secret_key="sk-lf-xxx", host="https://your-langfuse.com" ) def trace_llm(name="llm_call"): def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): start = time.time() span = langfuse_context.start_span(name=name, input=kwargs) try: result = await func(*args, **kwargs) ttft = time.time() - start span.end( output=result, metadata={"ttft_ms": ttft * 1000, "model": kwargs.get("model", "unknown")} ) langfuse_context.update_current_observation( usage={ "prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0), "completion_tokens": result.get("usage", {}).get("completion_tokens", 0), "total_tokens": result.get("usage", {}).get("total_tokens", 0), }, metadata={"latency_ms": (time.time() - start) * 1000} ) return result except Exception as e: span.end(level="ERROR", status_message=str(e)) raise return wrapper return decorator @observe(name="rag_pipeline") async def rag_pipeline(query: str): docs = await retrieve(query) answer = await generate(query, docs) return answer @observe(name="retrieve") async def retrieve(query: str): results = await vector_search(query) return {"query": query, "docs": results} @trace_llm(name="generate") async def generate(query: str, docs: list): resp = await llm_client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": f"Q: {query}\nContext: {docs}"}] ) return resp 工具对比 工具 部署方式 核心优势 适用场景 价格 LangSmith SaaS LangChain 深度集成 LangChain 用户 $39/月起 Langfuse 自部署/SaaS 开源，框架无关 通用 AI 应用 免费/自部署 Phoenix 自部署 Arize 出品，评估强 需要评估体系 开源 OpenTelemetry 自部署 标准化，通用 已有 OTel 体系 免费 选型建议：不用 LangChain 选 Langfuse（开源、框架无关），用 LangChain 选 LangSmith（集成最好），需要深度评估选 Phoenix。 ...