1. 项目概述为什么我们需要Agent中间件如果你正在用LangChain构建Agent大概率遇到过这样的场景Agent执行一个查询任务中途调用了三次工具最后返回了结果。但你想知道它每一步具体想了什么、调用了哪个工具、耗时多久、有没有出错。如果只是看最终输出整个过程就是个黑盒。更麻烦的是当你想在生产环境监控Agent的性能、记录日志、控制成本或者重试失败步骤时会发现原始的Agent框架提供的控制粒度远远不够。这就是Agent中间件或者说钩子函数Hooks要解决的问题。简单来说LangChain的Agent中间件是一系列可以在Agent执行生命周期关键节点插入自定义逻辑的函数。它就像给你的Agent装上了一套“监控探头”和“控制手柄”让你能观察、记录、甚至干预Agent的思考与行动过程。从简单的日志打印到复杂的错误重试、成本审计、敏感信息过滤都离不开中间件。我见过不少团队初期只关注核心逻辑上线后才发现缺乏可观测性出了问题无从排查回头补中间件反而更费劲。因此理解并熟练运用这六种核心钩子是从“玩具Demo”迈向“生产级应用”的关键一步。本文将以一个完整的“天气查询与新闻总结”Agent项目为例手把手带你实现六种最核心的中间件钩子。你将不仅知道每个钩子怎么用更能理解其背后的设计意图、适用场景以及我踩过的一些坑。所有代码均提供完整可运行的示例你可以直接复制到你的项目中。2. 核心钩子函数全景解析与设计思路在深入代码之前我们必须先建立对Agent执行生命周期的整体认知。一个典型的LangChain Agent执行一次任务如“查询上海天气并总结要点”并不是一步到位的它会经历一个多步骤的循环我们称之为“Agent执行循环”。这个循环的核心是思考LLM决定下一步行动 - 行动执行工具 - 观察获取工具结果 - 再思考...直到LLM认为任务完成并输出最终答案。六种钩子函数正是精准地挂载在这个循环的不同阶段。理解它们的位置就理解了其作用。2.1 钩子函数地图与执行顺序下图清晰地展示了Agent执行过程中各个钩子被触发的顺序与时机[任务开始] | v on_chain_start (整个Agent开始) | v 循环开始: on_llm_start (LLM开始思考) | v on_llm_end (LLM生成思考结果如“我需要调用天气工具”) | v on_tool_start (根据LLM指示开始执行工具) | v on_tool_end (工具执行完毕返回结果) | v on_agent_action (Agent处理了本次“思考-行动”单元) | v [判断是否继续循环] --是-- 回到 on_llm_start | 否 v on_chain_end (整个Agent结束输出最终答案)关键点解析on_chain_start/end这是最外层的包装标志着整个Agent任务的开始和结束。它适合做全局性的初始化如启动计时器、分配追踪ID和收尾工作如汇总日志、发送通知。on_llm_start/end聚焦于LLM的调用。每次Agent需要“思考”下一步时都会触发。这里是记录和分析LLM输入Prompt和输出思考内容的最佳位置也是进行Prompt动态修改或成本估算的切入点。on_tool_start/end聚焦于工具的执行。当Agent决定调用一个工具如搜索、计算、API查询时触发。这里是监控工具性能、重试失败调用、过滤或加工工具输入/输出的关键。on_agent_action这是一个综合性钩子在每次LLM思考后、且决定要采取一个具体行动通常是调用工具时触发。它接收一个AgentAction对象其中包含了LLM决定要执行的工具名和输入参数。它非常适合进行行动前的最终校验或路由。我的踩坑心得初学者常混淆on_llm_end和on_agent_action。记住on_llm_end只告诉你LLM输出了什么文本。而on_agent_action是LangChain框架将LLM的文本输出解析成一个结构化行动如toolweather, tool_input{location: 上海}之后才触发的。如果LLM的输出格式不对无法被解析则可能只有on_llm_end没有on_agent_action。2.2 为何是这六种设计哲学与选型考量LangChain提供了更多的回调函数但这六种是构建生产级中间件的基石。为什么覆盖核心生命周期它们完整覆盖了“计划LLM- 执行Tool- 观察”这个核心循环的起点、终点和关键节点没有冗余。提供足够的控制粒度既有关注全局的chain有关注推理的llm有关注执行的tool还有关注决策的agent_action。这种分层设计允许开发者针对不同层级的问题实施不同的策略。平衡灵活性与复杂度更多的钩子意味着更复杂的学习和维护成本。这六种钩子已经能解决95%的中间件需求如日志、监控、安全、重试、缓存等。更特殊的需求可以通过组合这些钩子或自定义工具/LLM来实现。在你的项目中不必一开始就实现所有钩子。我建议按需引入开发调试阶段优先实现on_llm_end和on_tool_end进行日志打印快速定位问题。预生产阶段加入on_chain_start/end用于追踪完整会话加入on_tool_start进行输入验证或重试准备。生产阶段引入on_agent_action进行更复杂的决策审计或安全策略检查。3. 实战构建一个可观测的天气新闻Agent理论讲完了我们动手建一个项目。这个Agent的功能是“获取指定城市的当前天气并根据天气情况查找与该城市或该天气相关的近期新闻最后生成一份简要报告。”我们将为这个Agent逐步添加六种中间件让它从“哑巴”变成“透明、健壮、可管理”的智能体。3.1 项目初始化与基础Agent搭建首先确保你的环境已安装必要库pip install langchain langchain-openai requests。我们使用OpenAI的模型和简单的HTTP请求工具来模拟。# agent_core.py import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.tools import tool import requests from typing import Any, Dict # 1. 定义工具 tool def get_weather(location: str) - str: 获取指定城市的当前天气。这是一个模拟函数。 # 模拟API调用实际项目中请替换为真实天气API weather_map { 上海: 晴25°C东南风2级, 北京: 多云18°C北风3级, 广州: 雷阵雨28°C南风1级, 深圳: 阴27°C微风, } return weather_map.get(location, f未找到{city}的天气信息。模拟返回晴22°C。) tool def search_news(keyword: str) - str: 根据关键词搜索近期新闻摘要。这是一个模拟函数。 # 模拟新闻搜索 news_db { 上海 晴: 【上海】今日晴空万里市民纷纷出游外滩客流增多。, 北京 多云: 【北京】多云天气适宜出行公园赏花活动正当时。, 广州 雷阵雨: 【广州】午后有雷阵雨市政部门提醒注意排水防涝。, 深圳 阴: 【深圳】阴天持续科技创新大会今日开幕不影响室内议程。, 天气: 全球气候变暖议题持续引发关注多国研讨减排方案。, } # 简单模拟优先匹配“城市天气”组合其次匹配关键词 return news_db.get(keyword, f关于“{keyword}”的模拟新闻相关话题近期受到讨论。) # 2. 配置LLM和Prompt llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手可以查询天气和新闻。请根据用户问题按步骤思考并调用工具。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 3. 创建Agent和Executor tools [get_weather, search_news] agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseFalse) # 先关闭原生verbose # 4. 运行测试 if __name__ __main__: result agent_executor.invoke({input: 上海天气怎么样有什么相关新闻吗}) print(最终结果:, result[output])运行这个基础版本Agent应该能正常工作并返回结果。但除了最终输出我们对内部过程一无所知。接下来我们引入中间件。3.2 实现日志中间件on_llm_endon_tool_end日志是最基本的需求。我们将创建一个中间件类它继承自BaseCallbackHandler并重写我们需要的方法。# middleware_logging.py from langchain_core.callbacks import BaseCallbackHandler from datetime import datetime import json class LoggingMiddleware(BaseCallbackHandler): 基础日志中间件记录LLM和Tool的输入输出。 def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) - Any: # 这里可以记录LLM调用的开始例如分配一个本次调用的唯一ID run_id kwargs.get(run_id, unknown) self.llm_start_time datetime.now() print(f\n[LLM_START] RunId: {run_id} | Time: {self.llm_start_time.isoformat()}) def on_llm_end(self, response: Any, **kwargs: Any) - Any: 记录LLM的响应。 run_id kwargs.get(run_id, unknown) end_time datetime.now() duration (end_time - self.llm_start_time).total_seconds() if hasattr(self, llm_start_time) else 0 # 提取LLM生成的内容 # response 结构可能因LLM类型而异这里处理OpenAI的ChatResult if hasattr(response, generations): content response.generations[0][0].text if response.generations else No content elif hasattr(response, content): content response.content else: content str(response)[:200] # 截断防止过长 print(f[LLM_END] RunId: {run_id} | Duration: {duration:.2f}s) print(f Content: {content}\n) # 在实际生产中这里应该写入文件或日志系统而非打印。 def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) - Any: self.tool_start_time datetime.now() tool_name serialized.get(name, serialized.get(id, [unknown])[0]) run_id kwargs.get(run_id, unknown) print(f[TOOL_START] RunId: {run_id} | Tool: {tool_name} | Input: {input_str}) def on_tool_end(self, output: str, **kwargs: Any) - Any: run_id kwargs.get(run_id, unknown) end_time datetime.now() duration (end_time - self.tool_start_time).total_seconds() if hasattr(self, tool_start_time) else 0 # 注意output 可能是字符串也可能是其他类型 output_str output if isinstance(output, str) else str(output) print(f[TOOL_END] RunId: {run_id} | Duration: {duration:.2f}s) print(f Output: {output_str[:150]}...\n) # 截断长输出 # 修改agent_executor的初始化加入回调 from langchain.callbacks import CallbackManager callback_manager CallbackManager([LoggingMiddleware()]) agent_executor AgentExecutor( agentagent, toolstools, callback_managercallback_manager, verboseFalse )现在再次运行Agent你会在控制台看到详细的步骤日志。这已经比原始的verboseTrue输出更结构化、更易于集成到日志管道中。实操心得直接打印日志只适用于调试。在生产中应将日志发送到如ELK、Loki或云日志服务。建议在__init__中初始化日志客户端在各个钩子方法中异步发送日志事件避免阻塞主流程。另外run_id是串联一次执行中所有事件的唯一标识务必在on_chain_start中生成并传递下去。3.3 实现性能监控与耗时分析中间件on_chain_start/end全局耗时和每个步骤的耗时对于性能优化至关重要。我们扩展日志中间件加入链级别的计时。# middleware_performance.py import time from collections import defaultdict class PerformanceMonitoringMiddleware(BaseCallbackHandler): 性能监控中间件记录链、LLM、工具的总耗时和调用次数。 def __init__(self): self.metrics defaultdict(lambda: {total_time: 0.0, count: 0}) self.chain_start_time None self.current_llm_run_id None self.current_tool_run_id None def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any) - Any: self.chain_start_time time.time() run_id kwargs.get(run_id, chain_run) print(f\n CHAIN START: {run_id} ) print(fInput: {inputs}) def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) - Any: chain_duration time.time() - self.chain_start_time if self.chain_start_time else 0 run_id kwargs.get(run_id, chain_run) print(f\n CHAIN END: {run_id} ) print(fOutput: {outputs.get(output, outputs)}) print(fTotal Chain Duration: {chain_duration:.2f}s) print(\n--- Performance Summary ---) for component, data in self.metrics.items(): avg data[total_time] / data[count] if data[count] 0 else 0 print(f {component}: Called {data[count]} times, Total {data[total_time]:.2f}s, Avg {avg:.2f}s) print(\n) def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) - Any: self.current_llm_run_id kwargs.get(run_id) self.metrics[LLM][count] 1 self._start_timer(self.current_llm_run_id) def on_llm_end(self, response: Any, **kwargs: Any) - Any: duration self._end_timer(kwargs.get(run_id)) if duration is not None: self.metrics[LLM][total_time] duration def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) - Any: tool_name serialized.get(name, unknown_tool) self.current_tool_run_id kwargs.get(run_id) self.metrics[fTool_{tool_name}][count] 1 self._start_timer(self.current_tool_run_id) def on_tool_end(self, output: str, **kwargs: Any) - Any: duration self._end_timer(kwargs.get(run_id)) if duration is not None: run_id kwargs.get(run_id) # 我们需要知道是哪个工具的耗时这里简化处理假设顺序执行。 # 更严谨的做法是用一个字典存储run_id到工具名的映射。 # 为了示例清晰我们这里统计到总工具时间或需要更复杂的设计。 # 我们先记录到通用“Tools”类别下。 self.metrics[Tools][total_time] duration self.metrics[Tools][count] 1 def _start_timer(self, run_id: str): if run_id: self.timers[run_id] time.time() def _end_timer(self, run_id: str) - float: if run_id and run_id in self.timers: duration time.time() - self.timers.pop(run_id) return duration return None def __init__(self): self.metrics defaultdict(lambda: {total_time: 0.0, count: 0}) self.timers {} # 用于存储各个run_id的开始时间 self.chain_start_time None # 使用组合的回调管理器 callback_manager CallbackManager([LoggingMiddleware(), PerformanceMonitoringMiddleware()]) agent_executor AgentExecutor( agentagent, toolstools, callback_managercallback_manager, verboseFalse )这个中间件会在任务结束时打印一份漂亮的性能报告帮助你一眼看出是LLM思考慢还是工具调用慢。3.4 实现错误处理与重试中间件on_tool_starton_tool_end网络工具调用失败是家常便饭。一个健壮的Agent应该具备重试能力。我们可以在工具执行层实现。# middleware_retry.py import asyncio from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests.exceptions class RetryableToolMiddleware(BaseCallbackHandler): 工具重试中间件。注意这个中间件主要通过在on_tool_start前介入来实现。 更常见的模式是直接包装工具本身但这里展示用回调思想来组织重试逻辑。 def __init__(self, max_retries: int 2): self.max_retries max_retries # 定义重试装饰器 self.retry_decorator retry( stopstop_after_attempt(max_retries 1), # 1 包含第一次尝试 waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((requests.exceptions.RequestException, TimeoutError)), reraiseTrue # 重试耗尽后抛出原异常 ) def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) - Any: 这里不能直接修改工具执行但我们可以在这里记录并可能影响执行策略。 实际上重试逻辑最好直接实现在工具函数内部或者通过包装工具来实现。 以下代码演示一种思路但并非标准做法。 tool_name serialized.get(name) print(f[RETRY_MIDDLEWARE] 准备执行工具 {tool_name}最多重试{self.max_retries}次。) # 在实际实现中你可能需要在这里将一个“可重试”的版本注入到执行流程中。 # 更简洁的做法是创建一个支持重试的Tool Wrapper类。 # 因此更实用的做法是直接创建支持重试的工具 from langchain_core.tools import StructuredTool from functools import wraps def retryable_tool(func): 装饰器使工具函数具备重试能力。 wraps(func) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max5)) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: print(f工具 {func.__name__} 调用失败正在重试... 错误: {e}) raise # 让tenacity捕获并决定是否重试 return wrapper # 用装饰器包装原始工具函数 tool retryable_tool def get_weather_retryable(location: str) - str: 获取天气支持重试。 # 模拟偶尔失败 import random if random.random() 0.3: # 30%概率模拟失败 raise requests.exceptions.ConnectionError(模拟网络连接失败) return get_weather.func(location) # 调用原始函数逻辑 # 更新工具列表 tools [get_weather_retryable, search_news]注意事项将重试逻辑放在中间件on_tool_start中实现是复杂且侵入性的因为它需要拦截和修改工具的执行过程。更优雅、更符合LangChain哲学的方式是包装工具本身如上所示或使用LangChain内置的Retry工具包装器。中间件更适合做观测和通知例如在工具失败时发送告警而非直接改变执行逻辑。3.5 实现安全与审计中间件on_agent_actionon_agent_action钩子让我们能在Agent决定行动前进行最后审查。例如检查工具输入是否包含敏感词或者记录决策日志用于合规审计。# middleware_security_audit.py class SecurityAuditMiddleware(BaseCallbackHandler): 安全审计中间件检查Agent决策并记录审计日志。 def __init__(self, blocked_keywords: list None): self.blocked_keywords blocked_keywords or [密码, 密钥, delete, drop table] self.audit_log [] def on_agent_action(self, action: Any, **kwargs: Any) - Any: 在Agent决定行动时触发。action是一个AgentAction对象。 run_id kwargs.get(run_id, unknown) tool_name action.tool tool_input action.tool_input # 1. 安全检查检查输入中是否包含敏感关键词 input_str str(tool_input) for keyword in self.blocked_keywords: if keyword in input_str: # 可以选择阻止、替换或仅记录告警 print(f⚠️ [SECURITY_BLOCKED] RunId: {run_id} | 工具 {tool_name} 的输入包含敏感词 {keyword}。) # 这里可以抛出一个异常来终止本次行动或者修改tool_input。 # 为了演示我们仅记录并允许继续。 # raise ValueError(f输入包含敏感词: {keyword}) # 2. 审计日志 audit_entry { run_id: run_id, timestamp: datetime.now().isoformat(), action: tool_invocation, tool: tool_name, input: tool_input, decision_phase: agent_selected } self.audit_log.append(audit_entry) print(f[AUDIT_LOG] 决策记录: Agent选择执行工具 {tool_name}输入: {tool_input}) def get_audit_log(self): 获取本次运行的所有审计日志。 return self.audit_log # 使用这个中间件 security_middleware SecurityAuditMiddleware(blocked_keywords[密码, 密钥]) callback_manager CallbackManager([LoggingMiddleware(), PerformanceMonitoringMiddleware(), security_middleware])这个中间件在生产中非常有用特别是对于处理用户输入或执行数据库操作的Agent可以防止意外或恶意的危险操作。3.6 完整集成与运行示例现在我们将所有中间件集成到一个可执行的脚本中。# main.py import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.callbacks import CallbackManager from middleware_logging import LoggingMiddleware from middleware_performance import PerformanceMonitoringMiddleware from middleware_security_audit import SecurityAuditMiddleware # ... (工具定义 get_weather_retryable, search_news 同上) ... def create_agent_with_middleware(): 创建集成了所有中间件的Agent执行器。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手可以查询天气和新闻。请根据用户问题按步骤思考并调用工具。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) tools [get_weather_retryable, search_news] agent create_openai_tools_agent(llm, tools, prompt) # 1. 初始化所有中间件 logging_mw LoggingMiddleware() performance_mw PerformanceMonitoringMiddleware() security_mw SecurityAuditMiddleware(blocked_keywords[密码]) # 2. 创建回调管理器 callback_manager CallbackManager([logging_mw, performance_mw, security_mw]) # 3. 创建Agent执行器 agent_executor AgentExecutor( agentagent, toolstools, callback_managercallback_manager, verboseFalse, # 我们用自己的中间件输出关闭LangChain原生冗长输出 handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5 # 防止无限循环 ) return agent_executor, security_mw if __name__ __main__: agent_executor, security_middleware create_agent_with_middleware() queries [ 上海天气怎么样有什么相关新闻吗, # 你可以测试一个包含敏感词的查询观察审计中间件的反应 # 帮我删除密码文件。, ] for query in queries: print(f\n{*60}) print(f处理查询: {query}) print(*60) try: result agent_executor.invoke({input: query}) print(f\n最终答案: {result[output]}) # 打印安全审计日志 # print(f\n安全审计日志: {security_middleware.get_audit_log()}) except Exception as e: print(fAgent执行出错: {e})运行这个脚本你将看到层次分明的日志输出、性能总结并且如果查询触发了安全规则也会看到相应的警告。4. 生产环境部署与高级技巧将这套中间件用于生产还需要考虑以下几个关键点4.1 中间件的管理与配置在大型应用中你可能需要根据不同环境开发、测试、生产启用不同的中间件组合。建议使用配置化或依赖注入的方式来管理。# config.py from enum import Enum class Environment(Enum): DEV development STAGING staging PROD production def get_middleware_for_env(env: Environment): middlewares [] if env in [Environment.DEV, Environment.STAGING]: middlewares.append(LoggingMiddleware()) middlewares.append(PerformanceMonitoringMiddleware()) if env Environment.PROD: # 生产环境可能使用更轻量、异步的日志中间件 middlewares.append(AsyncLoggingMiddleware()) middlewares.append(SecurityAuditMiddleware(blocked_keywordsload_blocked_list())) middlewares.append(RateLimitingMiddleware()) # 假设有速率限制中间件 # 性能监控可能集成到APM如OpenTelemetry中而非简单打印 return middlewares4.2 异步支持与性能默认的BaseCallbackHandler方法是同步的。如果中间件逻辑涉及网络IO如远程日志会阻塞Agent执行。LangChain提供了AsyncCallbackHandler。from langchain_core.callbacks import AsyncCallbackHandler import aiohttp import asyncio class AsyncLoggingMiddleware(AsyncCallbackHandler): async def on_llm_end(self, response, **kwargs): run_id kwargs.get(run_id) log_data {event: llm_end, run_id: run_id, content: str(response)[:500]} async with aiohttp.ClientSession() as session: async with session.post(http://your-log-server/log, jsonlog_data): pass # 发送日志不等待响应或根据需要等待在初始化时使用AsyncCallbackManager。from langchain.callbacks import AsyncCallbackManager async_manager AsyncCallbackManager([AsyncLoggingMiddleware()]) agent_executor AgentExecutor(..., callback_managerasync_manager)4.3 与分布式追踪OpenTelemetry集成对于微服务架构你需要将Agent的追踪信息接入现有的分布式追踪系统如Jaeger。你可以创建一个中间件将每个run_id映射为Trace ID并在各个钩子中记录Span。from opentelemetry import trace tracer trace.get_tracer(__name__) class OpenTelemetryMiddleware(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, run_id, **kwargs): ctx trace.set_span_in_context(trace.Span()) # 简化示例实际需管理上下文 self.span tracer.start_span(fagent_chain, contextctx) self.span.set_attribute(run_id, run_id) def on_tool_start(self, serialized, input_str, run_id, **kwargs): tool_span tracer.start_span(ftool_{serialized.get(name)}, parentself.span) tool_span.set_attribute(input, input_str) self.current_tool_span tool_span def on_tool_end(self, output, run_id, **kwargs): if hasattr(self, current_tool_span): self.current_tool_span.set_attribute(output, str(output)[:200]) self.current_tool_span.end()4.4 常见问题排查速查表问题现象可能原因排查步骤中间件钩子没有被触发1. 回调管理器未正确设置到AgentExecutor。2. 使用的LLM或Tool对象本身也需支持回调LangChain内置组件通常支持。3. 钩子方法签名不正确未被识别。1. 检查AgentExecutor的callback_manager参数。2. 确保LLM和Tool也是通过支持回调的框架创建的。3. 继承BaseCallbackHandler并确保方法名完全匹配。on_agent_action不触发Agent的输出未能被成功解析为AgentAction。可能是LLM的输出格式不符合OpenAIToolsAgent的预期。1. 检查on_llm_end中LLM的输出内容看是否是有效的工具调用格式。2. 启用handle_parsing_errorsTrue并观察错误。性能中间件计时不准多个异步操作并行执行时简单的time.time()计时会混乱。1. 确保每个run_id的start和end配对正确。2. 对于异步环境使用asyncio的高精度计时器并妥善管理状态。重试中间件不生效重试逻辑没有应用到实际的工具执行函数上。1. 确认重试装饰器是否包装了真正的工具函数。2. 确认工具异常类型是否被retry_if_exception_type覆盖。审计日志丢失部分操作某些步骤如LLM思考但未触发工具不会产生on_agent_action。1. 结合on_llm_end的日志来补充审计信息。2. 在on_chain_end中汇总所有日志。5. 超越基础中间件设计模式与扩展当你熟悉了基本钩子后可以探索更高级的模式中间件链Middleware Chain像Express.js或Koa框架一样将多个中间件组织成链每个中间件处理请求并传递给下一个。这在LangChain中可以通过回调管理器的顺序来实现但需要注意有些钩子可能希望中断流程如安全校验失败。条件中间件根据运行状态动态启用或禁用某些中间件。例如只在错误率超过阈值时开启详细调试日志。中间件与Agent状态中间件可以读写kwargs中的run_manager或自定义上下文来在不同钩子间传递信息。例如在on_chain_start中生成一个请求ID并在后续所有钩子中使用。最后记住中间件的核心价值是分离关注点。业务逻辑Agent的任务与横切关注点日志、监控、安全、重试应该解耦。通过这六种钩子你就能为你的LangChain Agent构建起一套强大、灵活、非侵入式的“神经系统”让它真正具备生产就绪的能力。
1. 项目概述为什么我们需要Agent中间件如果你正在用LangChain构建Agent大概率遇到过这样的场景Agent执行一个查询任务中途调用了三次工具最后返回了结果。但你想知道它每一步具体想了什么、调用了哪个工具、耗时多久、有没有出错。如果只是看最终输出整个过程就是个黑盒。更麻烦的是当你想在生产环境监控Agent的性能、记录日志、控制成本或者重试失败步骤时会发现原始的Agent框架提供的控制粒度远远不够。这就是Agent中间件或者说钩子函数Hooks要解决的问题。简单来说LangChain的Agent中间件是一系列可以在Agent执行生命周期关键节点插入自定义逻辑的函数。它就像给你的Agent装上了一套“监控探头”和“控制手柄”让你能观察、记录、甚至干预Agent的思考与行动过程。从简单的日志打印到复杂的错误重试、成本审计、敏感信息过滤都离不开中间件。我见过不少团队初期只关注核心逻辑上线后才发现缺乏可观测性出了问题无从排查回头补中间件反而更费劲。因此理解并熟练运用这六种核心钩子是从“玩具Demo”迈向“生产级应用”的关键一步。本文将以一个完整的“天气查询与新闻总结”Agent项目为例手把手带你实现六种最核心的中间件钩子。你将不仅知道每个钩子怎么用更能理解其背后的设计意图、适用场景以及我踩过的一些坑。所有代码均提供完整可运行的示例你可以直接复制到你的项目中。2. 核心钩子函数全景解析与设计思路在深入代码之前我们必须先建立对Agent执行生命周期的整体认知。一个典型的LangChain Agent执行一次任务如“查询上海天气并总结要点”并不是一步到位的它会经历一个多步骤的循环我们称之为“Agent执行循环”。这个循环的核心是思考LLM决定下一步行动 - 行动执行工具 - 观察获取工具结果 - 再思考...直到LLM认为任务完成并输出最终答案。六种钩子函数正是精准地挂载在这个循环的不同阶段。理解它们的位置就理解了其作用。2.1 钩子函数地图与执行顺序下图清晰地展示了Agent执行过程中各个钩子被触发的顺序与时机[任务开始] | v on_chain_start (整个Agent开始) | v 循环开始: on_llm_start (LLM开始思考) | v on_llm_end (LLM生成思考结果如“我需要调用天气工具”) | v on_tool_start (根据LLM指示开始执行工具) | v on_tool_end (工具执行完毕返回结果) | v on_agent_action (Agent处理了本次“思考-行动”单元) | v [判断是否继续循环] --是-- 回到 on_llm_start | 否 v on_chain_end (整个Agent结束输出最终答案)关键点解析on_chain_start/end这是最外层的包装标志着整个Agent任务的开始和结束。它适合做全局性的初始化如启动计时器、分配追踪ID和收尾工作如汇总日志、发送通知。on_llm_start/end聚焦于LLM的调用。每次Agent需要“思考”下一步时都会触发。这里是记录和分析LLM输入Prompt和输出思考内容的最佳位置也是进行Prompt动态修改或成本估算的切入点。on_tool_start/end聚焦于工具的执行。当Agent决定调用一个工具如搜索、计算、API查询时触发。这里是监控工具性能、重试失败调用、过滤或加工工具输入/输出的关键。on_agent_action这是一个综合性钩子在每次LLM思考后、且决定要采取一个具体行动通常是调用工具时触发。它接收一个AgentAction对象其中包含了LLM决定要执行的工具名和输入参数。它非常适合进行行动前的最终校验或路由。我的踩坑心得初学者常混淆on_llm_end和on_agent_action。记住on_llm_end只告诉你LLM输出了什么文本。而on_agent_action是LangChain框架将LLM的文本输出解析成一个结构化行动如toolweather, tool_input{location: 上海}之后才触发的。如果LLM的输出格式不对无法被解析则可能只有on_llm_end没有on_agent_action。2.2 为何是这六种设计哲学与选型考量LangChain提供了更多的回调函数但这六种是构建生产级中间件的基石。为什么覆盖核心生命周期它们完整覆盖了“计划LLM- 执行Tool- 观察”这个核心循环的起点、终点和关键节点没有冗余。提供足够的控制粒度既有关注全局的chain有关注推理的llm有关注执行的tool还有关注决策的agent_action。这种分层设计允许开发者针对不同层级的问题实施不同的策略。平衡灵活性与复杂度更多的钩子意味着更复杂的学习和维护成本。这六种钩子已经能解决95%的中间件需求如日志、监控、安全、重试、缓存等。更特殊的需求可以通过组合这些钩子或自定义工具/LLM来实现。在你的项目中不必一开始就实现所有钩子。我建议按需引入开发调试阶段优先实现on_llm_end和on_tool_end进行日志打印快速定位问题。预生产阶段加入on_chain_start/end用于追踪完整会话加入on_tool_start进行输入验证或重试准备。生产阶段引入on_agent_action进行更复杂的决策审计或安全策略检查。3. 实战构建一个可观测的天气新闻Agent理论讲完了我们动手建一个项目。这个Agent的功能是“获取指定城市的当前天气并根据天气情况查找与该城市或该天气相关的近期新闻最后生成一份简要报告。”我们将为这个Agent逐步添加六种中间件让它从“哑巴”变成“透明、健壮、可管理”的智能体。3.1 项目初始化与基础Agent搭建首先确保你的环境已安装必要库pip install langchain langchain-openai requests。我们使用OpenAI的模型和简单的HTTP请求工具来模拟。# agent_core.py import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.tools import tool import requests from typing import Any, Dict # 1. 定义工具 tool def get_weather(location: str) - str: 获取指定城市的当前天气。这是一个模拟函数。 # 模拟API调用实际项目中请替换为真实天气API weather_map { 上海: 晴25°C东南风2级, 北京: 多云18°C北风3级, 广州: 雷阵雨28°C南风1级, 深圳: 阴27°C微风, } return weather_map.get(location, f未找到{city}的天气信息。模拟返回晴22°C。) tool def search_news(keyword: str) - str: 根据关键词搜索近期新闻摘要。这是一个模拟函数。 # 模拟新闻搜索 news_db { 上海 晴: 【上海】今日晴空万里市民纷纷出游外滩客流增多。, 北京 多云: 【北京】多云天气适宜出行公园赏花活动正当时。, 广州 雷阵雨: 【广州】午后有雷阵雨市政部门提醒注意排水防涝。, 深圳 阴: 【深圳】阴天持续科技创新大会今日开幕不影响室内议程。, 天气: 全球气候变暖议题持续引发关注多国研讨减排方案。, } # 简单模拟优先匹配“城市天气”组合其次匹配关键词 return news_db.get(keyword, f关于“{keyword}”的模拟新闻相关话题近期受到讨论。) # 2. 配置LLM和Prompt llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手可以查询天气和新闻。请根据用户问题按步骤思考并调用工具。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 3. 创建Agent和Executor tools [get_weather, search_news] agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseFalse) # 先关闭原生verbose # 4. 运行测试 if __name__ __main__: result agent_executor.invoke({input: 上海天气怎么样有什么相关新闻吗}) print(最终结果:, result[output])运行这个基础版本Agent应该能正常工作并返回结果。但除了最终输出我们对内部过程一无所知。接下来我们引入中间件。3.2 实现日志中间件on_llm_endon_tool_end日志是最基本的需求。我们将创建一个中间件类它继承自BaseCallbackHandler并重写我们需要的方法。# middleware_logging.py from langchain_core.callbacks import BaseCallbackHandler from datetime import datetime import json class LoggingMiddleware(BaseCallbackHandler): 基础日志中间件记录LLM和Tool的输入输出。 def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) - Any: # 这里可以记录LLM调用的开始例如分配一个本次调用的唯一ID run_id kwargs.get(run_id, unknown) self.llm_start_time datetime.now() print(f\n[LLM_START] RunId: {run_id} | Time: {self.llm_start_time.isoformat()}) def on_llm_end(self, response: Any, **kwargs: Any) - Any: 记录LLM的响应。 run_id kwargs.get(run_id, unknown) end_time datetime.now() duration (end_time - self.llm_start_time).total_seconds() if hasattr(self, llm_start_time) else 0 # 提取LLM生成的内容 # response 结构可能因LLM类型而异这里处理OpenAI的ChatResult if hasattr(response, generations): content response.generations[0][0].text if response.generations else No content elif hasattr(response, content): content response.content else: content str(response)[:200] # 截断防止过长 print(f[LLM_END] RunId: {run_id} | Duration: {duration:.2f}s) print(f Content: {content}\n) # 在实际生产中这里应该写入文件或日志系统而非打印。 def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) - Any: self.tool_start_time datetime.now() tool_name serialized.get(name, serialized.get(id, [unknown])[0]) run_id kwargs.get(run_id, unknown) print(f[TOOL_START] RunId: {run_id} | Tool: {tool_name} | Input: {input_str}) def on_tool_end(self, output: str, **kwargs: Any) - Any: run_id kwargs.get(run_id, unknown) end_time datetime.now() duration (end_time - self.tool_start_time).total_seconds() if hasattr(self, tool_start_time) else 0 # 注意output 可能是字符串也可能是其他类型 output_str output if isinstance(output, str) else str(output) print(f[TOOL_END] RunId: {run_id} | Duration: {duration:.2f}s) print(f Output: {output_str[:150]}...\n) # 截断长输出 # 修改agent_executor的初始化加入回调 from langchain.callbacks import CallbackManager callback_manager CallbackManager([LoggingMiddleware()]) agent_executor AgentExecutor( agentagent, toolstools, callback_managercallback_manager, verboseFalse )现在再次运行Agent你会在控制台看到详细的步骤日志。这已经比原始的verboseTrue输出更结构化、更易于集成到日志管道中。实操心得直接打印日志只适用于调试。在生产中应将日志发送到如ELK、Loki或云日志服务。建议在__init__中初始化日志客户端在各个钩子方法中异步发送日志事件避免阻塞主流程。另外run_id是串联一次执行中所有事件的唯一标识务必在on_chain_start中生成并传递下去。3.3 实现性能监控与耗时分析中间件on_chain_start/end全局耗时和每个步骤的耗时对于性能优化至关重要。我们扩展日志中间件加入链级别的计时。# middleware_performance.py import time from collections import defaultdict class PerformanceMonitoringMiddleware(BaseCallbackHandler): 性能监控中间件记录链、LLM、工具的总耗时和调用次数。 def __init__(self): self.metrics defaultdict(lambda: {total_time: 0.0, count: 0}) self.chain_start_time None self.current_llm_run_id None self.current_tool_run_id None def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any) - Any: self.chain_start_time time.time() run_id kwargs.get(run_id, chain_run) print(f\n CHAIN START: {run_id} ) print(fInput: {inputs}) def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) - Any: chain_duration time.time() - self.chain_start_time if self.chain_start_time else 0 run_id kwargs.get(run_id, chain_run) print(f\n CHAIN END: {run_id} ) print(fOutput: {outputs.get(output, outputs)}) print(fTotal Chain Duration: {chain_duration:.2f}s) print(\n--- Performance Summary ---) for component, data in self.metrics.items(): avg data[total_time] / data[count] if data[count] 0 else 0 print(f {component}: Called {data[count]} times, Total {data[total_time]:.2f}s, Avg {avg:.2f}s) print(\n) def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any) - Any: self.current_llm_run_id kwargs.get(run_id) self.metrics[LLM][count] 1 self._start_timer(self.current_llm_run_id) def on_llm_end(self, response: Any, **kwargs: Any) - Any: duration self._end_timer(kwargs.get(run_id)) if duration is not None: self.metrics[LLM][total_time] duration def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) - Any: tool_name serialized.get(name, unknown_tool) self.current_tool_run_id kwargs.get(run_id) self.metrics[fTool_{tool_name}][count] 1 self._start_timer(self.current_tool_run_id) def on_tool_end(self, output: str, **kwargs: Any) - Any: duration self._end_timer(kwargs.get(run_id)) if duration is not None: run_id kwargs.get(run_id) # 我们需要知道是哪个工具的耗时这里简化处理假设顺序执行。 # 更严谨的做法是用一个字典存储run_id到工具名的映射。 # 为了示例清晰我们这里统计到总工具时间或需要更复杂的设计。 # 我们先记录到通用“Tools”类别下。 self.metrics[Tools][total_time] duration self.metrics[Tools][count] 1 def _start_timer(self, run_id: str): if run_id: self.timers[run_id] time.time() def _end_timer(self, run_id: str) - float: if run_id and run_id in self.timers: duration time.time() - self.timers.pop(run_id) return duration return None def __init__(self): self.metrics defaultdict(lambda: {total_time: 0.0, count: 0}) self.timers {} # 用于存储各个run_id的开始时间 self.chain_start_time None # 使用组合的回调管理器 callback_manager CallbackManager([LoggingMiddleware(), PerformanceMonitoringMiddleware()]) agent_executor AgentExecutor( agentagent, toolstools, callback_managercallback_manager, verboseFalse )这个中间件会在任务结束时打印一份漂亮的性能报告帮助你一眼看出是LLM思考慢还是工具调用慢。3.4 实现错误处理与重试中间件on_tool_starton_tool_end网络工具调用失败是家常便饭。一个健壮的Agent应该具备重试能力。我们可以在工具执行层实现。# middleware_retry.py import asyncio from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests.exceptions class RetryableToolMiddleware(BaseCallbackHandler): 工具重试中间件。注意这个中间件主要通过在on_tool_start前介入来实现。 更常见的模式是直接包装工具本身但这里展示用回调思想来组织重试逻辑。 def __init__(self, max_retries: int 2): self.max_retries max_retries # 定义重试装饰器 self.retry_decorator retry( stopstop_after_attempt(max_retries 1), # 1 包含第一次尝试 waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((requests.exceptions.RequestException, TimeoutError)), reraiseTrue # 重试耗尽后抛出原异常 ) def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) - Any: 这里不能直接修改工具执行但我们可以在这里记录并可能影响执行策略。 实际上重试逻辑最好直接实现在工具函数内部或者通过包装工具来实现。 以下代码演示一种思路但并非标准做法。 tool_name serialized.get(name) print(f[RETRY_MIDDLEWARE] 准备执行工具 {tool_name}最多重试{self.max_retries}次。) # 在实际实现中你可能需要在这里将一个“可重试”的版本注入到执行流程中。 # 更简洁的做法是创建一个支持重试的Tool Wrapper类。 # 因此更实用的做法是直接创建支持重试的工具 from langchain_core.tools import StructuredTool from functools import wraps def retryable_tool(func): 装饰器使工具函数具备重试能力。 wraps(func) retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max5)) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: print(f工具 {func.__name__} 调用失败正在重试... 错误: {e}) raise # 让tenacity捕获并决定是否重试 return wrapper # 用装饰器包装原始工具函数 tool retryable_tool def get_weather_retryable(location: str) - str: 获取天气支持重试。 # 模拟偶尔失败 import random if random.random() 0.3: # 30%概率模拟失败 raise requests.exceptions.ConnectionError(模拟网络连接失败) return get_weather.func(location) # 调用原始函数逻辑 # 更新工具列表 tools [get_weather_retryable, search_news]注意事项将重试逻辑放在中间件on_tool_start中实现是复杂且侵入性的因为它需要拦截和修改工具的执行过程。更优雅、更符合LangChain哲学的方式是包装工具本身如上所示或使用LangChain内置的Retry工具包装器。中间件更适合做观测和通知例如在工具失败时发送告警而非直接改变执行逻辑。3.5 实现安全与审计中间件on_agent_actionon_agent_action钩子让我们能在Agent决定行动前进行最后审查。例如检查工具输入是否包含敏感词或者记录决策日志用于合规审计。# middleware_security_audit.py class SecurityAuditMiddleware(BaseCallbackHandler): 安全审计中间件检查Agent决策并记录审计日志。 def __init__(self, blocked_keywords: list None): self.blocked_keywords blocked_keywords or [密码, 密钥, delete, drop table] self.audit_log [] def on_agent_action(self, action: Any, **kwargs: Any) - Any: 在Agent决定行动时触发。action是一个AgentAction对象。 run_id kwargs.get(run_id, unknown) tool_name action.tool tool_input action.tool_input # 1. 安全检查检查输入中是否包含敏感关键词 input_str str(tool_input) for keyword in self.blocked_keywords: if keyword in input_str: # 可以选择阻止、替换或仅记录告警 print(f⚠️ [SECURITY_BLOCKED] RunId: {run_id} | 工具 {tool_name} 的输入包含敏感词 {keyword}。) # 这里可以抛出一个异常来终止本次行动或者修改tool_input。 # 为了演示我们仅记录并允许继续。 # raise ValueError(f输入包含敏感词: {keyword}) # 2. 审计日志 audit_entry { run_id: run_id, timestamp: datetime.now().isoformat(), action: tool_invocation, tool: tool_name, input: tool_input, decision_phase: agent_selected } self.audit_log.append(audit_entry) print(f[AUDIT_LOG] 决策记录: Agent选择执行工具 {tool_name}输入: {tool_input}) def get_audit_log(self): 获取本次运行的所有审计日志。 return self.audit_log # 使用这个中间件 security_middleware SecurityAuditMiddleware(blocked_keywords[密码, 密钥]) callback_manager CallbackManager([LoggingMiddleware(), PerformanceMonitoringMiddleware(), security_middleware])这个中间件在生产中非常有用特别是对于处理用户输入或执行数据库操作的Agent可以防止意外或恶意的危险操作。3.6 完整集成与运行示例现在我们将所有中间件集成到一个可执行的脚本中。# main.py import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.callbacks import CallbackManager from middleware_logging import LoggingMiddleware from middleware_performance import PerformanceMonitoringMiddleware from middleware_security_audit import SecurityAuditMiddleware # ... (工具定义 get_weather_retryable, search_news 同上) ... def create_agent_with_middleware(): 创建集成了所有中间件的Agent执行器。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手可以查询天气和新闻。请根据用户问题按步骤思考并调用工具。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) tools [get_weather_retryable, search_news] agent create_openai_tools_agent(llm, tools, prompt) # 1. 初始化所有中间件 logging_mw LoggingMiddleware() performance_mw PerformanceMonitoringMiddleware() security_mw SecurityAuditMiddleware(blocked_keywords[密码]) # 2. 创建回调管理器 callback_manager CallbackManager([logging_mw, performance_mw, security_mw]) # 3. 创建Agent执行器 agent_executor AgentExecutor( agentagent, toolstools, callback_managercallback_manager, verboseFalse, # 我们用自己的中间件输出关闭LangChain原生冗长输出 handle_parsing_errorsTrue, # 优雅处理解析错误 max_iterations5 # 防止无限循环 ) return agent_executor, security_mw if __name__ __main__: agent_executor, security_middleware create_agent_with_middleware() queries [ 上海天气怎么样有什么相关新闻吗, # 你可以测试一个包含敏感词的查询观察审计中间件的反应 # 帮我删除密码文件。, ] for query in queries: print(f\n{*60}) print(f处理查询: {query}) print(*60) try: result agent_executor.invoke({input: query}) print(f\n最终答案: {result[output]}) # 打印安全审计日志 # print(f\n安全审计日志: {security_middleware.get_audit_log()}) except Exception as e: print(fAgent执行出错: {e})运行这个脚本你将看到层次分明的日志输出、性能总结并且如果查询触发了安全规则也会看到相应的警告。4. 生产环境部署与高级技巧将这套中间件用于生产还需要考虑以下几个关键点4.1 中间件的管理与配置在大型应用中你可能需要根据不同环境开发、测试、生产启用不同的中间件组合。建议使用配置化或依赖注入的方式来管理。# config.py from enum import Enum class Environment(Enum): DEV development STAGING staging PROD production def get_middleware_for_env(env: Environment): middlewares [] if env in [Environment.DEV, Environment.STAGING]: middlewares.append(LoggingMiddleware()) middlewares.append(PerformanceMonitoringMiddleware()) if env Environment.PROD: # 生产环境可能使用更轻量、异步的日志中间件 middlewares.append(AsyncLoggingMiddleware()) middlewares.append(SecurityAuditMiddleware(blocked_keywordsload_blocked_list())) middlewares.append(RateLimitingMiddleware()) # 假设有速率限制中间件 # 性能监控可能集成到APM如OpenTelemetry中而非简单打印 return middlewares4.2 异步支持与性能默认的BaseCallbackHandler方法是同步的。如果中间件逻辑涉及网络IO如远程日志会阻塞Agent执行。LangChain提供了AsyncCallbackHandler。from langchain_core.callbacks import AsyncCallbackHandler import aiohttp import asyncio class AsyncLoggingMiddleware(AsyncCallbackHandler): async def on_llm_end(self, response, **kwargs): run_id kwargs.get(run_id) log_data {event: llm_end, run_id: run_id, content: str(response)[:500]} async with aiohttp.ClientSession() as session: async with session.post(http://your-log-server/log, jsonlog_data): pass # 发送日志不等待响应或根据需要等待在初始化时使用AsyncCallbackManager。from langchain.callbacks import AsyncCallbackManager async_manager AsyncCallbackManager([AsyncLoggingMiddleware()]) agent_executor AgentExecutor(..., callback_managerasync_manager)4.3 与分布式追踪OpenTelemetry集成对于微服务架构你需要将Agent的追踪信息接入现有的分布式追踪系统如Jaeger。你可以创建一个中间件将每个run_id映射为Trace ID并在各个钩子中记录Span。from opentelemetry import trace tracer trace.get_tracer(__name__) class OpenTelemetryMiddleware(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, run_id, **kwargs): ctx trace.set_span_in_context(trace.Span()) # 简化示例实际需管理上下文 self.span tracer.start_span(fagent_chain, contextctx) self.span.set_attribute(run_id, run_id) def on_tool_start(self, serialized, input_str, run_id, **kwargs): tool_span tracer.start_span(ftool_{serialized.get(name)}, parentself.span) tool_span.set_attribute(input, input_str) self.current_tool_span tool_span def on_tool_end(self, output, run_id, **kwargs): if hasattr(self, current_tool_span): self.current_tool_span.set_attribute(output, str(output)[:200]) self.current_tool_span.end()4.4 常见问题排查速查表问题现象可能原因排查步骤中间件钩子没有被触发1. 回调管理器未正确设置到AgentExecutor。2. 使用的LLM或Tool对象本身也需支持回调LangChain内置组件通常支持。3. 钩子方法签名不正确未被识别。1. 检查AgentExecutor的callback_manager参数。2. 确保LLM和Tool也是通过支持回调的框架创建的。3. 继承BaseCallbackHandler并确保方法名完全匹配。on_agent_action不触发Agent的输出未能被成功解析为AgentAction。可能是LLM的输出格式不符合OpenAIToolsAgent的预期。1. 检查on_llm_end中LLM的输出内容看是否是有效的工具调用格式。2. 启用handle_parsing_errorsTrue并观察错误。性能中间件计时不准多个异步操作并行执行时简单的time.time()计时会混乱。1. 确保每个run_id的start和end配对正确。2. 对于异步环境使用asyncio的高精度计时器并妥善管理状态。重试中间件不生效重试逻辑没有应用到实际的工具执行函数上。1. 确认重试装饰器是否包装了真正的工具函数。2. 确认工具异常类型是否被retry_if_exception_type覆盖。审计日志丢失部分操作某些步骤如LLM思考但未触发工具不会产生on_agent_action。1. 结合on_llm_end的日志来补充审计信息。2. 在on_chain_end中汇总所有日志。5. 超越基础中间件设计模式与扩展当你熟悉了基本钩子后可以探索更高级的模式中间件链Middleware Chain像Express.js或Koa框架一样将多个中间件组织成链每个中间件处理请求并传递给下一个。这在LangChain中可以通过回调管理器的顺序来实现但需要注意有些钩子可能希望中断流程如安全校验失败。条件中间件根据运行状态动态启用或禁用某些中间件。例如只在错误率超过阈值时开启详细调试日志。中间件与Agent状态中间件可以读写kwargs中的run_manager或自定义上下文来在不同钩子间传递信息。例如在on_chain_start中生成一个请求ID并在后续所有钩子中使用。最后记住中间件的核心价值是分离关注点。业务逻辑Agent的任务与横切关注点日志、监控、安全、重试应该解耦。通过这六种钩子你就能为你的LangChain Agent构建起一套强大、灵活、非侵入式的“神经系统”让它真正具备生产就绪的能力。
