在构建企业级AI应用时很多开发者都经历过这样的困境基于Langchain快速搭建的原型Agent在本地测试时运行良好一旦部署到生产环境面对复杂的用户请求和并发场景就变成了一个“黑盒”。Agent内部如何决策为什么这次调用耗时长达10秒用户反馈的“胡言乱语”背后是哪一步的Prompt出了问题缺乏有效的可观测性Observability和日志追踪让问题排查和性能优化变得异常困难。本文将系统性地解决这一问题。我们将从Langchain的基础概念出发逐步深入到企业级AI Agent的构建并重点探讨如何为其注入强大的可观测性与日志追踪能力。通过一个完整的实战案例你将掌握从零搭建一个具备完整监控、链路追踪和日志聚合能力的AI Agent系统的全流程。无论你是希望将个人项目升级为健壮服务还是需要在企业环境中落地AI应用本文提供的架构整合方案和代码都能直接复用。1. 背景与核心概念为什么企业级AI Agent需要可观测性在深入实战之前我们有必要厘清几个核心概念并理解为什么可观测性对于AI Agent至关重要。AI Agent通常指能够感知环境、进行决策并执行行动以达成目标的智能体。在大模型时代一个典型的AI Agent架构通常包含以下组件大脑LLM负责理解、推理和生成。工具Tools扩展Agent能力如搜索、计算、调用API。记忆Memory存储对话历史、知识或状态。规划器Planner拆解复杂任务为可执行的子步骤。执行器Executor协调工具调用和LLM交互。Langchain是一个流行的框架它通过提供标准化的接口和丰富的组件库极大地简化了构建此类Agent的过程。然而Langchain默认提供的日志信息较为基础难以满足生产环境的需求。可观测性Observability是一个源于运维领域的术语指通过系统外部输出的数据如日志、指标、追踪来理解其内部状态的能力。对于AI Agent可观测性意味着我们能清晰地看到请求全链路一个用户问题触发了哪些工具调用LLM被调用了多少次性能指标每一步的耗时是多少Token消耗情况如何决策依据Agent选择某个工具的原因是什么LLM的思考过程Chain-of-Thought是怎样的错误与异常失败发生在哪个环节是工具API超时还是LLM返回了无法解析的内容没有可观测性AI Agent的运维将如同“盲人摸象”。而日志追踪是实现可观测性的关键技术手段它通过为每个请求分配唯一的追踪IDTrace ID将分散在各个模块和微服务中的日志串联起来形成完整的“故事线”。2. 环境准备与版本说明本教程将使用 Python 作为开发语言构建一个具备可观测性的问答型AI Agent。请确保你的环境满足以下要求操作系统Windows 10/11, macOS 或 Linux (Ubuntu 20.04)。本文命令以Linux/macOS为例。Python版本3.9 或 3.10。推荐使用3.10以获得最佳兼容性。核心框架与库langchain0.1.0 构建Agent的核心框架。注意Langchain版本迭代较快本文基于0.1.x版本编写核心概念相通。langchain-openai0.0.5 OpenAI模型集成。openai1.0.0 OpenAI官方SDK。可观测性组件opentelemetry-sdk1.20.0 OpenTelemetry SDK用于生成追踪和指标。opentelemetry-exporter-otlp1.20.0 将遥测数据导出到后端如Jaeger。loguru0.7.0 更友好、功能更强大的日志库。辅助工具pydantic2.0.0 用于数据验证和设置管理。python-dotenv1.0.0 管理环境变量。版本管理建议在实际企业项目中强烈建议使用requirements.txt或pyproject.toml文件精确锁定所有依赖的版本以避免因依赖更新导致的不兼容问题。项目结构预览observable_ai_agent/ ├── .env # 环境变量API密钥等 ├── requirements.txt # 项目依赖 ├── config/ │ └── settings.py # 应用配置 ├── core/ │ ├── agent.py # Agent核心逻辑 │ ├── tracing.py # 可观测性初始化与配置 │ └── logging.py # 日志配置 ├── tools/ │ └── custom_tools.py # 自定义工具 ├── main.py # 应用入口 └── docker-compose.yml # 用于启动可观测性后端如Jaeger接下来我们将从零开始搭建这个项目。3. 核心配置与原理拆解3.1 使用Pydantic管理配置将配置如API密钥、模型名称、追踪端点集中管理是工程化的第一步。我们使用Pydantic的BaseSettings它能自动从环境变量加载配置。# config/settings.py from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): 应用配置优先从环境变量读取 # OpenAI配置 openai_api_key: str Field(..., descriptionOpenAI API密钥) openai_model: str Field(defaultgpt-3.5-turbo-0125, description使用的模型名称) # 可观测性后端配置 (以Jaeger为例) tracing_enabled: bool Field(defaultTrue, description是否启用分布式追踪) otlp_endpoint: str Field(defaulthttp://localhost:4317, descriptionOTLP接收端地址) service_name: str Field(defaultai-agent-service, description服务名称用于追踪) # 日志配置 log_level: str Field(defaultINFO, description日志级别) log_file: str Field(default./logs/agent.log, description日志文件路径) class Config: env_file .env # 从.env文件加载 env_file_encoding utf-8 # 创建全局配置实例 settings Settings()对应的.env文件# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_MODELgpt-4-turbo-preview TRACING_ENABLEDtrue OTLP_ENDPOINThttp://localhost:4317 SERVICE_NAMEproduction-ai-agent LOG_LEVELDEBUG为什么这么做将配置外部化避免了将敏感信息硬编码在代码中方便不同环境开发、测试、生产的切换也符合十二要素应用原则。3.2 配置结构化日志与LoguruPython标准库的logging模块功能强大但配置繁琐。Loguru提供了更简洁易用的API并支持结构化日志JSON格式便于后续被日志收集系统如ELK解析。# core/logging.py import sys from loguru import logger import json from datetime import datetime from core.config import settings def setup_logging(): 配置应用日志 # 移除默认处理器 logger.remove() # 定义日志格式控制台使用友好格式文件使用JSON格式 console_format green{time:YYYY-MM-DD HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level json_format lambda record: json.dumps({ timestamp: datetime.utcfromtimestamp(record[time].timestamp()).isoformat() Z, level: record[level].name, service: settings.service_name, logger: record[name], function: record[function], line: record[line], message: record[message], # 可以自动注入追踪ID需要与Tracing集成 trace_id: record[extra].get(trace_id, ) if hasattr(record[extra], get) else , span_id: record[extra].get(span_id, ) if hasattr(record[extra], get) else , }) # 添加控制台处理器 logger.add( sys.stderr, formatconsole_format, levelsettings.log_level, colorizeTrue, ) # 添加文件处理器JSON格式 logger.add( settings.log_file, formatjson_format, levelsettings.log_level, rotation10 MB, # 日志轮转每10MB一个新文件 retention30 days, # 保留30天 compressionzip, serializeTrue, # 输出为JSON字符串 ) # 将logger实例导出方便其他模块导入使用 return logger # 初始化并导出logger app_logger setup_logging()关键点serializeTrue确保日志以JSON格式写入文件便于后续用Filebeat等工具采集。rotation和retention参数实现了日志的生命周期管理避免磁盘被撑满。我们在JSON格式中预留了trace_id和span_id字段这将与分布式追踪系统联动。3.3 集成OpenTelemetry实现分布式追踪OpenTelemetry (OTel) 是CNCF孵化的项目提供了统一的API来收集遥测数据追踪、指标、日志。我们将使用它为Agent的每次执行创建追踪链路。# core/tracing.py from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource, SERVICE_NAME import functools from core.config import settings def setup_tracing(): 设置OpenTelemetry追踪 if not settings.tracing_enabled: # 如果不启用追踪设置一个无操作的TracerProvider trace.set_tracer_provider(trace.NoOpTracerProvider()) return trace.get_tracer(__name__) # 1. 创建TracerProvider并指定服务资源信息 resource Resource(attributes{ SERVICE_NAME: settings.service_name, deployment.environment: production, }) tracer_provider TracerProvider(resourceresource) # 2. 创建Span处理器导出器 span_processors [] # 2.1 控制台导出器用于本地调试 if settings.log_level DEBUG: console_exporter ConsoleSpanExporter() span_processors.append(BatchSpanProcessor(console_exporter)) # 2.2 OTLP导出器导出到Jaeger、Tempo等后端 otlp_exporter OTLPSpanExporter(endpointsettings.otlp_endpoint, insecureTrue) span_processors.append(BatchSpanProcessor(otlp_exporter)) # 3. 将处理器添加到TracerProvider for processor in span_processors: tracer_provider.add_span_processor(processor) # 4. 设置为全局TracerProvider trace.set_tracer_provider(tracer_provider) # 5. 获取一个本模块的Tracer并返回 return trace.get_tracer(__name__) # 初始化Tracer tracer setup_tracing() def trace_agent_execution(func): 一个装饰器用于自动追踪Agent执行函数 functools.wraps(func) def wrapper(*args, **kwargs): # 使用函数名作为Span的名称 with tracer.start_as_current_span(func.__name__) as span: # 可以在这里向Span添加一些自定义属性 # span.set_attribute(agent.type, react) try: result func(*args, **kwargs) span.set_status(trace.Status(trace.StatusCode.OK)) return result except Exception as e: # 记录异常信息到Span span.record_exception(e) span.set_status(trace.Status(trace.StatusCode.ERROR, str(e))) raise return wrapper原理说明TracerProvider 追踪器的工厂负责创建Tracer和管理配置。Tracer 用于创建Span。Span 代表一个工作单元如一次LLM调用、一次工具执行是追踪链路中的基本节点。Span之间通过Trace ID关联通过Parent Span ID形成树形结构。SpanProcessor 负责处理导出已完成的Span。BatchSpanProcessor会批量处理提高效率。OTLP OpenTelemetry Protocol是OTel定义的统一数据输出协议兼容Jaeger、Zipkin、Prometheus Tempo等多种后端。这个装饰器trace_agent_execution可以轻松地包装任何函数自动为其创建追踪Span。4. 完整实战案例构建可观测的天气查询Agent现在我们将综合运用以上组件构建一个能够查询天气的AI Agent。这个Agent会先判断用户意图如果需要查询天气则调用一个模拟的天气工具。4.1 创建项目结构与依赖首先创建项目目录并安装依赖。# 创建项目目录 mkdir observable_ai_agent cd observable_ai_agent mkdir -p config core tools logs # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 创建requirements.txt cat requirements.txt EOF langchain0.1.0 langchain-openai0.0.5 openai1.0.0 pydantic2.0.0 pydantic-settings2.0.0 python-dotenv1.0.0 loguru0.7.0 opentelemetry-sdk1.20.0 opentelemetry-exporter-otlp1.20.0 opentelemetry-instrumentation0.40b0 EOF # 安装依赖 pip install -r requirements.txt将前面章节的config/settings.py,core/logging.py,core/tracing.py文件创建好。4.2 实现自定义工具并注入追踪工具是Agent能力的延伸。我们实现一个模拟的天气查询工具并在其中集成日志和追踪。# tools/custom_tools.py from langchain.tools import BaseTool from pydantic import Field from typing import Optional, Type from core.logging import app_logger from core.tracing import tracer import random import time class WeatherQueryTool(BaseTool): 一个模拟的天气查询工具。 name: str get_weather description: str 根据城市名称查询该城市的当前天气情况。输入应为城市名例如北京。 city: str Field(..., description要查询天气的城市名称) # 为了简化我们重写 _run 方法。更标准的做法是使用 args_schema。 def _run(self, city: str) - str: 执行工具的主逻辑。 # 为工具执行创建一个独立的Span with tracer.start_as_current_span(tool.get_weather) as span: span.set_attribute(tool.input.city, city) app_logger.info(f开始查询天气城市: {city}, extra{tool: self.name}) # 模拟网络延迟 time.sleep(random.uniform(0.1, 0.5)) # 模拟根据城市返回天气这里只是随机生成 weather_conditions [晴, 多云, 阴, 小雨, 中雨, 大雪] temperatures range(-10, 35) result_weather random.choice(weather_conditions) result_temp random.choice(temperatures) result f{city}的天气是{result_weather}气温{result_temp}摄氏度。 span.set_attribute(tool.output.weather, result_weather) span.set_attribute(tool.output.temperature, result_temp) app_logger.info(f天气查询完成结果: {result}, extra{tool: self.name}) return result # 为了支持异步调用如果需要 async def _arun(self, city: str) - str: 异步执行。本例中与同步相同。 return self._run(city)关键设计继承BaseTool 这是Langchain工具的标准基类。清晰的name和description 这至关重要因为LLM会根据描述来决定是否以及如何使用该工具。在_run内部创建Span 这样每次工具调用都会在追踪链路中生成一个子Span清晰地展示其在整体请求中的位置和耗时。结构化日志 使用app_logger.info并添加extra参数可以丰富日志的上下文。4.3 构建可观测的Agent接下来我们创建Agent本身并用我们之前编写的装饰器来追踪其核心执行方法。# core/agent.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.tools import Tool from tools.custom_tools import WeatherQueryTool from core.config import settings from core.logging import app_logger from core.tracing import trace_agent_execution, tracer class ObservableAgent: 具备可观测性的AI Agent def __init__(self): # 1. 初始化LLM self.llm ChatOpenAI( modelsettings.openai_model, openai_api_keysettings.openai_api_key, temperature0, # 降低随机性使Agent行为更确定 ) app_logger.debug(fLLM初始化完成模型: {settings.openai_model}) # 2. 初始化工具列表 weather_tool_instance WeatherQueryTool() self.tools [ Tool( nameweather_tool_instance.name, funcweather_tool_instance._run, descriptionweather_tool_instance.description, args_schemaweather_tool_instance.args_schema if hasattr(weather_tool_instance, args_schema) else None, ) ] app_logger.info(fAgent工具加载完成共{len(self.tools)}个工具) # 3. 定义Prompt模板ReAct框架典型模板 self.prompt PromptTemplate.from_template( 你是一个乐于助人的助手可以回答问题并使用工具。 如果你需要查询天气请使用工具。 工具列表 {tools} 使用以下格式 问题用户输入的问题 思考你需要思考做什么是否需要使用工具 行动要使用的工具名称必须是[{tool_names}]中的一个 行动输入工具的输入 观察工具返回的结果 ... (这个思考/行动/观察循环可以重复多次) 最终答案根据观察得出的最终答案 开始 问题{input} 思考{agent_scratchpad} ) # 4. 创建Agent和Executor self.agent create_react_agent(llmself.llm, toolsself.tools, promptself.prompt) self.agent_executor AgentExecutor( agentself.agent, toolsself.tools, verboseFalse, # 我们用自己的日志所以关闭Langchain的verbose handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate, # 提前停止策略 ) app_logger.info(ObservableAgent 初始化成功) trace_agent_execution def run(self, user_input: str) - str: 执行Agent的主入口该方法已被追踪装饰器包装 app_logger.info(f收到用户请求: {user_input}) # 在Span中添加用户输入作为属性 current_span trace.get_current_span() if current_span: current_span.set_attribute(user.input, user_input) try: # 调用Langchain Agent执行 result self.agent_executor.invoke({input: user_input}) final_answer result.get(output, 抱歉我没有得到答案。) app_logger.info(f请求处理完成最终答案: {final_answer}) # 记录本次执行的元数据到Span if current_span: current_span.set_attribute(agent.output, final_answer) current_span.set_attribute(agent.iteration_count, result.get(intermediate_steps, [])) return final_answer except Exception as e: app_logger.error(fAgent执行过程中发生异常: {e}, exc_infoTrue) # 异常会被装饰器自动记录到Span中 return f处理您的请求时出现错误{str(e)}核心要点trace_agent_execution装饰器 这是关键它自动为每次run方法调用创建了一个顶层的Span所有后续的工具调用Span都会成为它的子Span。关闭verbose Langchain自带的verbose输出虽然详细但格式杂乱不利于收集。我们使用结构化的日志替代它。错误处理handle_parsing_errorsTrue能防止因为LLM输出格式错误导致整个Agent崩溃。资源限制max_iterations防止Agent陷入思考循环消耗过多Token和时间。4.4 应用入口与可观测性后端启动最后我们创建应用入口并提供一个docker-compose文件来一键启动追踪后端Jaeger。# main.py import asyncio from core.agent import ObservableAgent from core.config import settings from core.logging import app_logger def main(): 主函数 app_logger.info(f启动可观测AI Agent服务服务名: {settings.service_name}) # 初始化Agent agent ObservableAgent() # 模拟处理一些请求 test_queries [ 你好今天天气怎么样, 请问北京和上海的天气分别如何, 你能做什么, 帮我查一下巴黎的天气。, ] for query in test_queries: print(f\n用户: {query}) response agent.run(query) print(fAgent: {response}) # 添加短暂间隔方便观察 asyncio.sleep(0.5) app_logger.info(演示请求处理完毕。) if __name__ __main__: main()为了可视化追踪数据我们需要一个后端。使用Docker Compose可以轻松启动Jaeger。# docker-compose.yml version: 3.8 services: jaeger-all-in-one: image: jaegertracing/all-in-one:latest container_name: jaeger ports: - 16686:16686 # Jaeger UI - 4317:4317 # OTLP gRPC接收端口 - 4318:4318 # OTLP HTTP接收端口 environment: - COLLECTOR_OTLP_ENABLEDtrue networks: - observability-net networks: observability-net: driver: bridge运行步骤启动可观测性后端docker-compose up -d访问Jaeger UIhttp://localhost:16686在项目根目录运行AgentOPENAI_API_KEYyour_key python main.py4.5 结果验证与可视化运行main.py后你将在控制台看到结构化的日志输出。同时所有追踪数据已通过OTLP协议发送到Jaeger。日志分析 查看./logs/agent.log文件里面是JSON格式的日志包含了时间、级别、服务名、函数、行号以及我们注入的trace_id。链路追踪可视化 打开Jaeger UI (http://localhost:16686)在服务下拉框中选择production-ai-agent或你在配置中设置的服务名点击Find Traces。你将看到每次Agent执行的追踪链路图。点击一个Trace可以看到树状结构最顶层是runSpan其下可能有多个tool.get_weatherSpan如果一次查询涉及多个城市。每个Span都显示了耗时、属性如user.input,tool.input.city和状态。这直观地展示了Agent内部的工作流程和性能瓶颈。5. 常见问题与排查思路在企业级部署中你可能会遇到以下问题问题现象可能原因排查思路与解决方案Jaeger UI中看不到Trace1. OTLP导出配置错误地址、端口2. 网络策略阻止3. Agent未成功发送数据1. 检查docker-compose端口映射和settings.py中的OTLP_ENDPOINT。2. 使用curl测试端点连通性curl http://localhost:4318/。3. 启用控制台导出器(ConsoleSpanExporter)确认Span是否生成。日志文件未生成或为空1. 目录权限不足2.loguru配置路径错误3. 日志级别设置过高1. 检查logs/目录是否存在且可写。2. 检查settings.log_file路径是否为绝对路径或相对路径正确。3. 将LOG_LEVEL设置为DEBUG查看控制台是否有输出。Agent执行缓慢1. LLM API调用延迟高2. 工具调用如网络请求超时3. Agent陷入循环max_iterations过大1. 在Jaeger中查看runSpan下各子Span的耗时定位瓶颈。2. 为工具调用设置超时timeout参数。3. 检查日志中agent.iteration_count适当调低max_iterations。LLM不调用工具1. 工具描述(description)不清晰2. Prompt模板未优化3. LLM温度(temperature)过高输出不稳定1. 精炼工具描述确保LLM能理解其用途和输入格式。2. 使用Langchain的AgentType如ZERO_SHOT_REACT_DESCRIPTION或微调Prompt。3. 将temperature设为0或更低值。内存消耗持续增长1. 对话历史Memory未限制长度2. 产生了内存泄漏1. 使用ConversationBufferWindowMemory等有长度限制的Memory。2. 使用如tracemalloc等工具进行内存分析检查工具或自定义组件。6. 最佳实践与工程建议将AI Agent投入生产环境除了可观测性还需考虑以下工程实践配置管理进阶多环境配置 使用不同的.env文件如.env.production或配置中心如Apollo, Consul管理环境变量。密钥安全 永远不要将API密钥提交到代码仓库。使用云厂商的密钥管理服务如AWS KMS, GCP Secret Manager或专业的密钥管理工具如HashiCorp Vault。增强的日志与追踪关联日志与追踪 将Trace ID和Span ID注入到每一条业务日志中如通过loguru的filter或patcher实现通过一个ID查询全链路日志。业务指标埋点 使用OpenTelemetry Metrics API记录业务指标如“每日请求量”、“各工具调用次数”、“平均响应时间”、“Token消耗分布”。LLM特定监控 监控LLM调用的延迟、速率限制、Token使用成本和特定错误如上下文过长。Agent架构优化超时与重试 为所有外部调用LLM、工具API设置合理的超时和重试策略如指数退避。限流与熔断 使用如tenacity进行重试使用circuitbreaker实现熔断防止下游故障拖垮整个Agent。异步化 如果工具是I/O密集型如网络请求考虑使用Langchain的异步接口(ainvoke,_arun)提升并发性能。测试与验证单元测试 为每个工具和关键的Agent逻辑编写单元测试。集成测试 模拟端到端流程验证Agent在给定输入下能否产生预期输出和正确的工具调用序列。评估与基准测试 建立测试数据集定期运行评估Agent回答的准确性和可靠性监控性能指标变化。部署与运维容器化 使用Docker将Agent及其依赖打包确保环境一致性。健康检查 为Agent服务添加/health端点检查LLM连通性、工具可用性等。版本化与回滚 Agent的Prompt、工具集、模型版本都应进行版本控制并具备快速回滚能力。通过本文的实战你已经掌握了为Langchain AI Agent构建企业级可观测性的核心方法。从基础的结构化日志和分布式追踪入手逐步扩展到配置管理、错误处理和性能优化这套方法论是构建可靠、可维护AI应用系统的基石。真正的价值在于当线上出现问题时你能快速定位根因而不是在庞杂的日志中大海捞针。下一步你可以尝试将追踪数据导出到PrometheusGrafana栈进行指标监控或者探索LangGraph等更复杂的Agent编排框架下的可观测性实践。 企业级AI Agent可观测性实战：基于Langchain与OpenTelemetry的日志追踪架构
在构建企业级AI应用时很多开发者都经历过这样的困境基于Langchain快速搭建的原型Agent在本地测试时运行良好一旦部署到生产环境面对复杂的用户请求和并发场景就变成了一个“黑盒”。Agent内部如何决策为什么这次调用耗时长达10秒用户反馈的“胡言乱语”背后是哪一步的Prompt出了问题缺乏有效的可观测性Observability和日志追踪让问题排查和性能优化变得异常困难。本文将系统性地解决这一问题。我们将从Langchain的基础概念出发逐步深入到企业级AI Agent的构建并重点探讨如何为其注入强大的可观测性与日志追踪能力。通过一个完整的实战案例你将掌握从零搭建一个具备完整监控、链路追踪和日志聚合能力的AI Agent系统的全流程。无论你是希望将个人项目升级为健壮服务还是需要在企业环境中落地AI应用本文提供的架构整合方案和代码都能直接复用。1. 背景与核心概念为什么企业级AI Agent需要可观测性在深入实战之前我们有必要厘清几个核心概念并理解为什么可观测性对于AI Agent至关重要。AI Agent通常指能够感知环境、进行决策并执行行动以达成目标的智能体。在大模型时代一个典型的AI Agent架构通常包含以下组件大脑LLM负责理解、推理和生成。工具Tools扩展Agent能力如搜索、计算、调用API。记忆Memory存储对话历史、知识或状态。规划器Planner拆解复杂任务为可执行的子步骤。执行器Executor协调工具调用和LLM交互。Langchain是一个流行的框架它通过提供标准化的接口和丰富的组件库极大地简化了构建此类Agent的过程。然而Langchain默认提供的日志信息较为基础难以满足生产环境的需求。可观测性Observability是一个源于运维领域的术语指通过系统外部输出的数据如日志、指标、追踪来理解其内部状态的能力。对于AI Agent可观测性意味着我们能清晰地看到请求全链路一个用户问题触发了哪些工具调用LLM被调用了多少次性能指标每一步的耗时是多少Token消耗情况如何决策依据Agent选择某个工具的原因是什么LLM的思考过程Chain-of-Thought是怎样的错误与异常失败发生在哪个环节是工具API超时还是LLM返回了无法解析的内容没有可观测性AI Agent的运维将如同“盲人摸象”。而日志追踪是实现可观测性的关键技术手段它通过为每个请求分配唯一的追踪IDTrace ID将分散在各个模块和微服务中的日志串联起来形成完整的“故事线”。2. 环境准备与版本说明本教程将使用 Python 作为开发语言构建一个具备可观测性的问答型AI Agent。请确保你的环境满足以下要求操作系统Windows 10/11, macOS 或 Linux (Ubuntu 20.04)。本文命令以Linux/macOS为例。Python版本3.9 或 3.10。推荐使用3.10以获得最佳兼容性。核心框架与库langchain0.1.0 构建Agent的核心框架。注意Langchain版本迭代较快本文基于0.1.x版本编写核心概念相通。langchain-openai0.0.5 OpenAI模型集成。openai1.0.0 OpenAI官方SDK。可观测性组件opentelemetry-sdk1.20.0 OpenTelemetry SDK用于生成追踪和指标。opentelemetry-exporter-otlp1.20.0 将遥测数据导出到后端如Jaeger。loguru0.7.0 更友好、功能更强大的日志库。辅助工具pydantic2.0.0 用于数据验证和设置管理。python-dotenv1.0.0 管理环境变量。版本管理建议在实际企业项目中强烈建议使用requirements.txt或pyproject.toml文件精确锁定所有依赖的版本以避免因依赖更新导致的不兼容问题。项目结构预览observable_ai_agent/ ├── .env # 环境变量API密钥等 ├── requirements.txt # 项目依赖 ├── config/ │ └── settings.py # 应用配置 ├── core/ │ ├── agent.py # Agent核心逻辑 │ ├── tracing.py # 可观测性初始化与配置 │ └── logging.py # 日志配置 ├── tools/ │ └── custom_tools.py # 自定义工具 ├── main.py # 应用入口 └── docker-compose.yml # 用于启动可观测性后端如Jaeger接下来我们将从零开始搭建这个项目。3. 核心配置与原理拆解3.1 使用Pydantic管理配置将配置如API密钥、模型名称、追踪端点集中管理是工程化的第一步。我们使用Pydantic的BaseSettings它能自动从环境变量加载配置。# config/settings.py from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): 应用配置优先从环境变量读取 # OpenAI配置 openai_api_key: str Field(..., descriptionOpenAI API密钥) openai_model: str Field(defaultgpt-3.5-turbo-0125, description使用的模型名称) # 可观测性后端配置 (以Jaeger为例) tracing_enabled: bool Field(defaultTrue, description是否启用分布式追踪) otlp_endpoint: str Field(defaulthttp://localhost:4317, descriptionOTLP接收端地址) service_name: str Field(defaultai-agent-service, description服务名称用于追踪) # 日志配置 log_level: str Field(defaultINFO, description日志级别) log_file: str Field(default./logs/agent.log, description日志文件路径) class Config: env_file .env # 从.env文件加载 env_file_encoding utf-8 # 创建全局配置实例 settings Settings()对应的.env文件# .env OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_MODELgpt-4-turbo-preview TRACING_ENABLEDtrue OTLP_ENDPOINThttp://localhost:4317 SERVICE_NAMEproduction-ai-agent LOG_LEVELDEBUG为什么这么做将配置外部化避免了将敏感信息硬编码在代码中方便不同环境开发、测试、生产的切换也符合十二要素应用原则。3.2 配置结构化日志与LoguruPython标准库的logging模块功能强大但配置繁琐。Loguru提供了更简洁易用的API并支持结构化日志JSON格式便于后续被日志收集系统如ELK解析。# core/logging.py import sys from loguru import logger import json from datetime import datetime from core.config import settings def setup_logging(): 配置应用日志 # 移除默认处理器 logger.remove() # 定义日志格式控制台使用友好格式文件使用JSON格式 console_format green{time:YYYY-MM-DD HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level json_format lambda record: json.dumps({ timestamp: datetime.utcfromtimestamp(record[time].timestamp()).isoformat() Z, level: record[level].name, service: settings.service_name, logger: record[name], function: record[function], line: record[line], message: record[message], # 可以自动注入追踪ID需要与Tracing集成 trace_id: record[extra].get(trace_id, ) if hasattr(record[extra], get) else , span_id: record[extra].get(span_id, ) if hasattr(record[extra], get) else , }) # 添加控制台处理器 logger.add( sys.stderr, formatconsole_format, levelsettings.log_level, colorizeTrue, ) # 添加文件处理器JSON格式 logger.add( settings.log_file, formatjson_format, levelsettings.log_level, rotation10 MB, # 日志轮转每10MB一个新文件 retention30 days, # 保留30天 compressionzip, serializeTrue, # 输出为JSON字符串 ) # 将logger实例导出方便其他模块导入使用 return logger # 初始化并导出logger app_logger setup_logging()关键点serializeTrue确保日志以JSON格式写入文件便于后续用Filebeat等工具采集。rotation和retention参数实现了日志的生命周期管理避免磁盘被撑满。我们在JSON格式中预留了trace_id和span_id字段这将与分布式追踪系统联动。3.3 集成OpenTelemetry实现分布式追踪OpenTelemetry (OTel) 是CNCF孵化的项目提供了统一的API来收集遥测数据追踪、指标、日志。我们将使用它为Agent的每次执行创建追踪链路。# core/tracing.py from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource, SERVICE_NAME import functools from core.config import settings def setup_tracing(): 设置OpenTelemetry追踪 if not settings.tracing_enabled: # 如果不启用追踪设置一个无操作的TracerProvider trace.set_tracer_provider(trace.NoOpTracerProvider()) return trace.get_tracer(__name__) # 1. 创建TracerProvider并指定服务资源信息 resource Resource(attributes{ SERVICE_NAME: settings.service_name, deployment.environment: production, }) tracer_provider TracerProvider(resourceresource) # 2. 创建Span处理器导出器 span_processors [] # 2.1 控制台导出器用于本地调试 if settings.log_level DEBUG: console_exporter ConsoleSpanExporter() span_processors.append(BatchSpanProcessor(console_exporter)) # 2.2 OTLP导出器导出到Jaeger、Tempo等后端 otlp_exporter OTLPSpanExporter(endpointsettings.otlp_endpoint, insecureTrue) span_processors.append(BatchSpanProcessor(otlp_exporter)) # 3. 将处理器添加到TracerProvider for processor in span_processors: tracer_provider.add_span_processor(processor) # 4. 设置为全局TracerProvider trace.set_tracer_provider(tracer_provider) # 5. 获取一个本模块的Tracer并返回 return trace.get_tracer(__name__) # 初始化Tracer tracer setup_tracing() def trace_agent_execution(func): 一个装饰器用于自动追踪Agent执行函数 functools.wraps(func) def wrapper(*args, **kwargs): # 使用函数名作为Span的名称 with tracer.start_as_current_span(func.__name__) as span: # 可以在这里向Span添加一些自定义属性 # span.set_attribute(agent.type, react) try: result func(*args, **kwargs) span.set_status(trace.Status(trace.StatusCode.OK)) return result except Exception as e: # 记录异常信息到Span span.record_exception(e) span.set_status(trace.Status(trace.StatusCode.ERROR, str(e))) raise return wrapper原理说明TracerProvider 追踪器的工厂负责创建Tracer和管理配置。Tracer 用于创建Span。Span 代表一个工作单元如一次LLM调用、一次工具执行是追踪链路中的基本节点。Span之间通过Trace ID关联通过Parent Span ID形成树形结构。SpanProcessor 负责处理导出已完成的Span。BatchSpanProcessor会批量处理提高效率。OTLP OpenTelemetry Protocol是OTel定义的统一数据输出协议兼容Jaeger、Zipkin、Prometheus Tempo等多种后端。这个装饰器trace_agent_execution可以轻松地包装任何函数自动为其创建追踪Span。4. 完整实战案例构建可观测的天气查询Agent现在我们将综合运用以上组件构建一个能够查询天气的AI Agent。这个Agent会先判断用户意图如果需要查询天气则调用一个模拟的天气工具。4.1 创建项目结构与依赖首先创建项目目录并安装依赖。# 创建项目目录 mkdir observable_ai_agent cd observable_ai_agent mkdir -p config core tools logs # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 创建requirements.txt cat requirements.txt EOF langchain0.1.0 langchain-openai0.0.5 openai1.0.0 pydantic2.0.0 pydantic-settings2.0.0 python-dotenv1.0.0 loguru0.7.0 opentelemetry-sdk1.20.0 opentelemetry-exporter-otlp1.20.0 opentelemetry-instrumentation0.40b0 EOF # 安装依赖 pip install -r requirements.txt将前面章节的config/settings.py,core/logging.py,core/tracing.py文件创建好。4.2 实现自定义工具并注入追踪工具是Agent能力的延伸。我们实现一个模拟的天气查询工具并在其中集成日志和追踪。# tools/custom_tools.py from langchain.tools import BaseTool from pydantic import Field from typing import Optional, Type from core.logging import app_logger from core.tracing import tracer import random import time class WeatherQueryTool(BaseTool): 一个模拟的天气查询工具。 name: str get_weather description: str 根据城市名称查询该城市的当前天气情况。输入应为城市名例如北京。 city: str Field(..., description要查询天气的城市名称) # 为了简化我们重写 _run 方法。更标准的做法是使用 args_schema。 def _run(self, city: str) - str: 执行工具的主逻辑。 # 为工具执行创建一个独立的Span with tracer.start_as_current_span(tool.get_weather) as span: span.set_attribute(tool.input.city, city) app_logger.info(f开始查询天气城市: {city}, extra{tool: self.name}) # 模拟网络延迟 time.sleep(random.uniform(0.1, 0.5)) # 模拟根据城市返回天气这里只是随机生成 weather_conditions [晴, 多云, 阴, 小雨, 中雨, 大雪] temperatures range(-10, 35) result_weather random.choice(weather_conditions) result_temp random.choice(temperatures) result f{city}的天气是{result_weather}气温{result_temp}摄氏度。 span.set_attribute(tool.output.weather, result_weather) span.set_attribute(tool.output.temperature, result_temp) app_logger.info(f天气查询完成结果: {result}, extra{tool: self.name}) return result # 为了支持异步调用如果需要 async def _arun(self, city: str) - str: 异步执行。本例中与同步相同。 return self._run(city)关键设计继承BaseTool 这是Langchain工具的标准基类。清晰的name和description 这至关重要因为LLM会根据描述来决定是否以及如何使用该工具。在_run内部创建Span 这样每次工具调用都会在追踪链路中生成一个子Span清晰地展示其在整体请求中的位置和耗时。结构化日志 使用app_logger.info并添加extra参数可以丰富日志的上下文。4.3 构建可观测的Agent接下来我们创建Agent本身并用我们之前编写的装饰器来追踪其核心执行方法。# core/agent.py from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.tools import Tool from tools.custom_tools import WeatherQueryTool from core.config import settings from core.logging import app_logger from core.tracing import trace_agent_execution, tracer class ObservableAgent: 具备可观测性的AI Agent def __init__(self): # 1. 初始化LLM self.llm ChatOpenAI( modelsettings.openai_model, openai_api_keysettings.openai_api_key, temperature0, # 降低随机性使Agent行为更确定 ) app_logger.debug(fLLM初始化完成模型: {settings.openai_model}) # 2. 初始化工具列表 weather_tool_instance WeatherQueryTool() self.tools [ Tool( nameweather_tool_instance.name, funcweather_tool_instance._run, descriptionweather_tool_instance.description, args_schemaweather_tool_instance.args_schema if hasattr(weather_tool_instance, args_schema) else None, ) ] app_logger.info(fAgent工具加载完成共{len(self.tools)}个工具) # 3. 定义Prompt模板ReAct框架典型模板 self.prompt PromptTemplate.from_template( 你是一个乐于助人的助手可以回答问题并使用工具。 如果你需要查询天气请使用工具。 工具列表 {tools} 使用以下格式 问题用户输入的问题 思考你需要思考做什么是否需要使用工具 行动要使用的工具名称必须是[{tool_names}]中的一个 行动输入工具的输入 观察工具返回的结果 ... (这个思考/行动/观察循环可以重复多次) 最终答案根据观察得出的最终答案 开始 问题{input} 思考{agent_scratchpad} ) # 4. 创建Agent和Executor self.agent create_react_agent(llmself.llm, toolsself.tools, promptself.prompt) self.agent_executor AgentExecutor( agentself.agent, toolsself.tools, verboseFalse, # 我们用自己的日志所以关闭Langchain的verbose handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate, # 提前停止策略 ) app_logger.info(ObservableAgent 初始化成功) trace_agent_execution def run(self, user_input: str) - str: 执行Agent的主入口该方法已被追踪装饰器包装 app_logger.info(f收到用户请求: {user_input}) # 在Span中添加用户输入作为属性 current_span trace.get_current_span() if current_span: current_span.set_attribute(user.input, user_input) try: # 调用Langchain Agent执行 result self.agent_executor.invoke({input: user_input}) final_answer result.get(output, 抱歉我没有得到答案。) app_logger.info(f请求处理完成最终答案: {final_answer}) # 记录本次执行的元数据到Span if current_span: current_span.set_attribute(agent.output, final_answer) current_span.set_attribute(agent.iteration_count, result.get(intermediate_steps, [])) return final_answer except Exception as e: app_logger.error(fAgent执行过程中发生异常: {e}, exc_infoTrue) # 异常会被装饰器自动记录到Span中 return f处理您的请求时出现错误{str(e)}核心要点trace_agent_execution装饰器 这是关键它自动为每次run方法调用创建了一个顶层的Span所有后续的工具调用Span都会成为它的子Span。关闭verbose Langchain自带的verbose输出虽然详细但格式杂乱不利于收集。我们使用结构化的日志替代它。错误处理handle_parsing_errorsTrue能防止因为LLM输出格式错误导致整个Agent崩溃。资源限制max_iterations防止Agent陷入思考循环消耗过多Token和时间。4.4 应用入口与可观测性后端启动最后我们创建应用入口并提供一个docker-compose文件来一键启动追踪后端Jaeger。# main.py import asyncio from core.agent import ObservableAgent from core.config import settings from core.logging import app_logger def main(): 主函数 app_logger.info(f启动可观测AI Agent服务服务名: {settings.service_name}) # 初始化Agent agent ObservableAgent() # 模拟处理一些请求 test_queries [ 你好今天天气怎么样, 请问北京和上海的天气分别如何, 你能做什么, 帮我查一下巴黎的天气。, ] for query in test_queries: print(f\n用户: {query}) response agent.run(query) print(fAgent: {response}) # 添加短暂间隔方便观察 asyncio.sleep(0.5) app_logger.info(演示请求处理完毕。) if __name__ __main__: main()为了可视化追踪数据我们需要一个后端。使用Docker Compose可以轻松启动Jaeger。# docker-compose.yml version: 3.8 services: jaeger-all-in-one: image: jaegertracing/all-in-one:latest container_name: jaeger ports: - 16686:16686 # Jaeger UI - 4317:4317 # OTLP gRPC接收端口 - 4318:4318 # OTLP HTTP接收端口 environment: - COLLECTOR_OTLP_ENABLEDtrue networks: - observability-net networks: observability-net: driver: bridge运行步骤启动可观测性后端docker-compose up -d访问Jaeger UIhttp://localhost:16686在项目根目录运行AgentOPENAI_API_KEYyour_key python main.py4.5 结果验证与可视化运行main.py后你将在控制台看到结构化的日志输出。同时所有追踪数据已通过OTLP协议发送到Jaeger。日志分析 查看./logs/agent.log文件里面是JSON格式的日志包含了时间、级别、服务名、函数、行号以及我们注入的trace_id。链路追踪可视化 打开Jaeger UI (http://localhost:16686)在服务下拉框中选择production-ai-agent或你在配置中设置的服务名点击Find Traces。你将看到每次Agent执行的追踪链路图。点击一个Trace可以看到树状结构最顶层是runSpan其下可能有多个tool.get_weatherSpan如果一次查询涉及多个城市。每个Span都显示了耗时、属性如user.input,tool.input.city和状态。这直观地展示了Agent内部的工作流程和性能瓶颈。5. 常见问题与排查思路在企业级部署中你可能会遇到以下问题问题现象可能原因排查思路与解决方案Jaeger UI中看不到Trace1. OTLP导出配置错误地址、端口2. 网络策略阻止3. Agent未成功发送数据1. 检查docker-compose端口映射和settings.py中的OTLP_ENDPOINT。2. 使用curl测试端点连通性curl http://localhost:4318/。3. 启用控制台导出器(ConsoleSpanExporter)确认Span是否生成。日志文件未生成或为空1. 目录权限不足2.loguru配置路径错误3. 日志级别设置过高1. 检查logs/目录是否存在且可写。2. 检查settings.log_file路径是否为绝对路径或相对路径正确。3. 将LOG_LEVEL设置为DEBUG查看控制台是否有输出。Agent执行缓慢1. LLM API调用延迟高2. 工具调用如网络请求超时3. Agent陷入循环max_iterations过大1. 在Jaeger中查看runSpan下各子Span的耗时定位瓶颈。2. 为工具调用设置超时timeout参数。3. 检查日志中agent.iteration_count适当调低max_iterations。LLM不调用工具1. 工具描述(description)不清晰2. Prompt模板未优化3. LLM温度(temperature)过高输出不稳定1. 精炼工具描述确保LLM能理解其用途和输入格式。2. 使用Langchain的AgentType如ZERO_SHOT_REACT_DESCRIPTION或微调Prompt。3. 将temperature设为0或更低值。内存消耗持续增长1. 对话历史Memory未限制长度2. 产生了内存泄漏1. 使用ConversationBufferWindowMemory等有长度限制的Memory。2. 使用如tracemalloc等工具进行内存分析检查工具或自定义组件。6. 最佳实践与工程建议将AI Agent投入生产环境除了可观测性还需考虑以下工程实践配置管理进阶多环境配置 使用不同的.env文件如.env.production或配置中心如Apollo, Consul管理环境变量。密钥安全 永远不要将API密钥提交到代码仓库。使用云厂商的密钥管理服务如AWS KMS, GCP Secret Manager或专业的密钥管理工具如HashiCorp Vault。增强的日志与追踪关联日志与追踪 将Trace ID和Span ID注入到每一条业务日志中如通过loguru的filter或patcher实现通过一个ID查询全链路日志。业务指标埋点 使用OpenTelemetry Metrics API记录业务指标如“每日请求量”、“各工具调用次数”、“平均响应时间”、“Token消耗分布”。LLM特定监控 监控LLM调用的延迟、速率限制、Token使用成本和特定错误如上下文过长。Agent架构优化超时与重试 为所有外部调用LLM、工具API设置合理的超时和重试策略如指数退避。限流与熔断 使用如tenacity进行重试使用circuitbreaker实现熔断防止下游故障拖垮整个Agent。异步化 如果工具是I/O密集型如网络请求考虑使用Langchain的异步接口(ainvoke,_arun)提升并发性能。测试与验证单元测试 为每个工具和关键的Agent逻辑编写单元测试。集成测试 模拟端到端流程验证Agent在给定输入下能否产生预期输出和正确的工具调用序列。评估与基准测试 建立测试数据集定期运行评估Agent回答的准确性和可靠性监控性能指标变化。部署与运维容器化 使用Docker将Agent及其依赖打包确保环境一致性。健康检查 为Agent服务添加/health端点检查LLM连通性、工具可用性等。版本化与回滚 Agent的Prompt、工具集、模型版本都应进行版本控制并具备快速回滚能力。通过本文的实战你已经掌握了为Langchain AI Agent构建企业级可观测性的核心方法。从基础的结构化日志和分布式追踪入手逐步扩展到配置管理、错误处理和性能优化这套方法论是构建可靠、可维护AI应用系统的基石。真正的价值在于当线上出现问题时你能快速定位根因而不是在庞杂的日志中大海捞针。下一步你可以尝试将追踪数据导出到PrometheusGrafana栈进行指标监控或者探索LangGraph等更复杂的Agent编排框架下的可观测性实践。