1. 项目概述为什么我们需要Agent中间件如果你正在用LangChain构建AI Agent大概率遇到过这样的场景Agent执行一个查询任务中途调用了搜索工具但返回的结果总是不尽如人意你想知道它到底搜了什么关键词或者为什么它决定调用这个工具而不是另一个。又或者在生产环境中你需要记录每一次Agent的决策过程用于审计、计费或性能分析。这时候直接修改Agent的核心逻辑不仅侵入性强而且容易出错。这就是Agent中间件或者说钩子函数Hooks大显身手的地方。简单来说Agent中间件就像给Agent的执行流程安装了一系列“监控探头”和“干预开关”。它允许你在Agent生命周期的关键节点如调用工具前、收到LLM响应后、执行动作前插入自定义逻辑而不需要改动Agent本身的代码。这带来了几个核心价值可观测性Observability让你能看清Agent内部的“黑盒”决策可控性Control让你能在关键时刻修正或引导Agent的行为以及非侵入式扩展保持核心代码的整洁与稳定。我最初接触这个概念时觉得它有点“高级”似乎只有构建复杂系统时才用得上。但实际用下来发现无论是调试一个简单的对话机器人还是部署一个需要严格合规的生产级Agent中间件都是提升开发效率和系统健壮性的利器。接下来我会结合六种核心的钩子函数从最基础的日志记录到高级的流程拦截与修改手把手带你实现一套完整的中间件系统并附上可直接运行的教学代码。2. 核心钩子函数深度解析与设计思路LangChain的Agent执行器AgentExecutor提供了一套丰富的回调Callbacks系统而我们要构建的中间件本质上是基于这套回调系统的、更结构化、更面向业务逻辑的封装。理解每种钩子触发的时机和能获取到的上下文信息是灵活运用的关键。2.1 六种钩子函数的定位与分工我们可以将这六种钩子分为三大类观测类、拦截/修改类和生命周期类。它们覆盖了Agent从思考到行动的全过程。观测类钩子用于“看”和“记”这类钩子主要用于记录日志、收集指标和监控通常不修改执行流程。on_llm_start / on_llm_end分别在向大语言模型LLM发送请求前和收到响应后触发。这是记录原始Prompt和Completion、计算Token消耗和延迟的黄金位置。on_tool_start / on_tool_end分别在调用一个工具如搜索、计算、数据库查询的前后触发。你可以在这里记录工具的名称、输入参数、执行结果和耗时对于调试工具调用逻辑异常有用。拦截/修改类钩子用于“控”和“改”这类钩子能力最强允许你修改即将发生的行为的参数甚至直接返回一个结果来“短路”整个调用。 3.on_llm_error在LLM调用出错时触发。你可以在这里实现重试逻辑例如遇到速率限制时等待后重试或者将错误信息转换为对用户更友好的提示。 4.on_tool_error在工具调用出错时触发。例如当调用一个外部API失败时你可以在这里尝试备用方案或者给Agent一个预设的兜底答案避免整个流程崩溃。生命周期类钩子用于感知“阶段”这类钩子让你感知Agent执行过程中的关键阶段转换。 5.on_chain_start / on_chain_end在Agent执行一个“链”Chain的前后触发。在Agent的上下文中每一次完整的“思考-行动”循环可以看作一个链。这里适合记录高层次的任务进度和整体状态。 6.on_agent_action / on_agent_finish这是Agent特有的钩子。on_agent_action在Agent决定要采取某个具体行动即调用某个工具时触发on_agent_finish在Agent得出最终答案并结束时触发。它们是理解Agent决策逻辑的核心。注意on_llm_error和on_tool_error虽然归类为拦截类但它们的首要职责是错误处理。修改流程如重试是建立在处理错误这个基础之上的高级应用。2.2 中间件系统的架构设计考量在设计中间件时我们面临几个关键选择这直接影响到代码的灵活性和复杂度。单一中间件 vs. 中间件链单一中间件一个类实现所有需要的钩子。优点是简单、内聚所有逻辑都在一处。缺点是随着功能增加这个类会变得臃肿难以维护。中间件链定义多个专注单一职责的中间件类如LoggingMiddleware、ErrorHandlingMiddleware、ValidationMiddleware然后按顺序组合。这是更符合生产环境的做法它遵循了单一职责原则方便独立测试和复用。例如你可以先经过验证中间件检查输入再经过日志中间件记录最后交给业务逻辑。同步 vs. 异步LangChain原生支持异步执行。如果你的Agent需要处理高并发请求或者内部调用的工具/LLM本身就是异步的那么实现异步中间件使用async/await能极大提升吞吐量。教学代码我们将以同步版本为主因为它更易于理解但我会指出关键位置如何改为异步。状态管理中间件经常需要共享状态。例如一个用于计费的中间件需要累计整个会话的Token使用量。这个状态应该存在哪里有两种常见模式实例属性在中间件类内部定义self.token_count。这适用于该中间件独享的状态。上下文传递通过LangChain回调系统的run_id、parent_run_id等属性将状态存储在外部存储如Redis、内存字典中键为run_id。这适用于需要在多个独立中间件或不同请求间共享的状态。我们的示例将采用更简单的实例属性方式。3. 手把手实现六类钩子中间件理论讲完了我们开始动手。我将实现一个功能相对完整的AgentMiddleware类它集成了观测、错误处理和简单干预的能力。之后我们再探讨如何将其拆分为链式结构。3.1 基础框架与观测类钩子实现首先我们需要创建一个继承自BaseCallbackHandler的类。这是LangChain所有回调的基类。from typing import Any, Dict, List, Optional from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import LLMResult, AgentAction, AgentFinish from langchain.schema.messages import BaseMessage import time import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class ComprehensiveAgentMiddleware(BaseCallbackHandler): 一个综合的Agent中间件示例演示六类钩子。 def __init__(self): super().__init__() # 用于存储当前会话的元数据 self.session_metrics { total_llm_calls: 0, total_tool_calls: 0, total_tokens_used: 0, start_time: time.time() } # 用于临时存储当前LLM或工具调用的开始时间 self._current_start_time None # 1. 观测类钩子LLM调用 def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any ) - None: LLM调用开始时触发。 self._current_start_time time.time() self.session_metrics[total_llm_calls] 1 run_id kwargs.get(run_id, N/A) # 记录Prompt生产环境中可能需要对长文本进行截断或脱敏 prompt_preview prompts[0][:100] ... if len(prompts[0]) 100 else prompts[0] logger.info(f[LLM_START] RunId: {run_id} | Prompt Preview: {prompt_preview}) # 在实际应用中你可以在这里将Prompt发送到监控平台 def on_llm_end(self, response: LLMResult, **kwargs: Any) - None: LLM调用成功结束时触发。 if self._current_start_time is None: return latency time.time() - self._current_start_time run_id kwargs.get(run_id, N/A) # 提取Token使用量如果LLM提供商返回了该信息 token_usage response.llm_output.get(token_usage, {}) if response.llm_output else {} completion_tokens token_usage.get(completion_tokens, 0) prompt_tokens token_usage.get(prompt_tokens, 0) total_tokens token_usage.get(total_tokens, 0) self.session_metrics[total_tokens_used] total_tokens # 记录响应和耗时 # 注意response.generations 是一个列表的列表结构复杂通常我们取第一条结果的第一条文本 try: first_generation response.generations[0][0] response_text first_generation.text[:200] ... if len(first_generation.text) 200 else first_generation.text except (IndexError, AttributeError): response_text Unable to parse response logger.info( f[LLM_END] RunId: {run_id} | Latency: {latency:.2f}s | fTokens: P({prompt_tokens})/C({completion_tokens}) | fResponse Preview: {response_text} ) self._current_start_time None # 2. 观测类钩子工具调用 def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs: Any ) - None: 工具调用开始时触发。 self._current_start_time time.time() self.session_metrics[total_tool_calls] 1 run_id kwargs.get(run_id, N/A) tool_name serialized.get(name, unknown_tool) logger.info(f[TOOL_START] RunId: {run_id} | Tool: {tool_name} | Input: {input_str}) def on_tool_end(self, output: str, **kwargs: Any) - None: 工具调用成功结束时触发。 if self._current_start_time is None: return latency time.time() - self._current_start_time run_id kwargs.get(run_id, N/A) # 对输出进行安全处理避免日志过长或包含敏感信息 safe_output output[:300] ... if len(output) 300 else output logger.info(f[TOOL_END] RunId: {run_id} | Latency: {latency:.2f}s | Output: {safe_output}) self._current_start_time None在上面的代码中我们实现了四个基础的观测钩子。关键点在于run_id这是LangChain为每一次执行链或Agent运行生成的唯一标识符用于串联所有相关的日志事件。信息截断在生产环境中Prompt和Response可能非常长直接打印会拖慢日志系统并可能泄露敏感数据。进行适当截断是必要的。耗时计算我们利用_current_start_time这个临时变量来配对开始和结束事件计算精确的延迟。这里有一个潜在风险如果on_llm_start被调用但on_llm_end因异常未被调用这个变量将不会被重置。更健壮的做法是使用一个以run_id为键的字典来管理多个并发调用的计时器。3.2 拦截修改类与生命周期钩子实现接下来我们实现错误处理和感知Agent生命周期的钩子。# 3. 拦截/修改类钩子错误处理 def on_llm_error(self, error: Exception, **kwargs: Any) - None: LLM调用出错时触发。 run_id kwargs.get(run_id, N/A) logger.error(f[LLM_ERROR] RunId: {run_id} | Error: {repr(error)}) # 重要这里可以尝试重试逻辑但注意不要陷入无限重试。 # 例如如果是速率限制错误可以等待一段时间。 # 但在这个基础示例中我们仅记录日志。 self._current_start_time None # 确保清理计时器 def on_tool_error(self, error: Exception, **kwargs: Any) - None: 工具调用出错时触发。 run_id kwargs.get(run_id, N/A) logger.error(f[TOOL_ERROR] RunId: {run_id} | Error: {repr(error)}) # 示例针对特定错误进行干预 # 假设我们有一个调用外部天气API的工具当它失败时我们提供一个友好的兜底信息。 # 注意直接在这里修改输出是困难的因为错误已经发生。 # 更常见的模式是在构建工具时在工具内部进行错误处理和兜底。 # 这个钩子更适合记录错误、触发告警或更新熔断器状态。 self._current_start_time None # 4. 生命周期类钩子Agent决策 def on_agent_action(self, action: AgentAction, **kwargs: Any) - None: Agent决定采取一个具体行动时触发。 run_id kwargs.get(run_id, N/A) # action.log 包含了Agent的思考过程非常有用 thought_process action.log[:200] ... if len(action.log) 200 else action.log logger.info( f[AGENT_ACTION] RunId: {run_id} | fTool: {action.tool} | Tool Input: {action.tool_input} | fThought: {thought_process} ) # 这里是一个潜在的“拦截点”你可以根据策略检查action.tool和tool_input # 如果不符合规则例如试图调用一个高风险工具可以抛出异常来终止本次行动。 # 但这需要更精细的流程控制可能结合自定义的AgentExecutor来实现。 def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) - None: Agent完成所有思考输出最终答案时触发。 run_id kwargs.get(run_id, N/A) total_time time.time() - self.session_metrics[start_time] logger.info( f[AGENT_FINISH] RunId: {run_id} | fFinal Output: {finish.return_values[output][:500]}... | fSession Total Time: {total_time:.2f}s | fMetrics: {self.session_metrics} ) # 此时可以将会话指标如总token数、调用次数上报到监控系统用于计费和性能分析。 # 注意on_chain_start/end 对于理解复杂链的嵌套很有用但在简单Agent中可能不常使用。 # 为保持示例完整这里也简单实现。 def on_chain_start( self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any ) - None: run_id kwargs.get(run_id, N/A) chain_name serialized.get(name, unknown_chain) # logger.debug(f[CHAIN_START] RunId: {run_id} | Chain: {chain_name}) # 通常用debug级别 def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) - None: run_id kwargs.get(run_id, N/A) # logger.debug(f[CHAIN_END] RunId: {run_id})错误处理钩子on_llm_error,on_tool_error是增强系统鲁棒性的关键。但需要理解它们的局限性它们是在错误发生后被调用的。如果你想防止错误发生或者错误发生后提供替代结果通常需要在更早的阶段介入。例如如果你想在工具调用超时时自动重试更好的地方可能是在工具类的_call方法内部实现重试逻辑或者使用一个专门的“重试工具包装器”中间件。错误钩子更适合做最后的日志记录和告警。on_agent_action和on_agent_finish是理解Agent“思维过程”的窗口。action.log里通常包含了LLM生成的在最终工具调用之前的推理文本这对于调试Agent为什么做出某个决定至关重要。4. 集成中间件到LangChain Agent实战现在我们有了一个功能齐全的中间件类。接下来看看如何将它实际应用到你的LangChain Agent中。我将创建一个简单的“天气查询Agent”作为示例。4.1 创建工具与Agentfrom langchain.agents import Tool, AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate import requests # 1. 定义一个简单的天气查询工具模拟 def get_weather(location: str) - str: 查询指定城市的天气。输入应为城市名。 # 这里是模拟真实情况应调用天气API # 模拟网络错误 # if location error_city: # raise ConnectionError(模拟网络错误无法连接到天气服务) weather_data { 北京: 晴15-25°C微风, 上海: 多云18-28°C东南风3级, 广州: 阵雨22-30°C南风4级, } return weather_data.get(location, f抱歉未找到{city}的天气信息。) # 将函数封装成LangChain Tool weather_tool Tool( nameWeatherLookup, funcget_weather, description用于查询城市天气。输入一个城市名称如‘北京’。 ) # 2. 初始化LLM # 请替换为你的实际API Key或使用其他兼容的LLM llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, openai_api_keyyour-api-key-here # 请务必替换 ) # 3. 使用ReAct框架创建Agent from langchain import hub prompt hub.pull(hwchase17/react-chat) # 一个标准的ReAct提示模板 agent create_react_agent(llm, tools[weather_tool], promptprompt) # 4. 创建Agent执行器并注入我们的中间件 middleware ComprehensiveAgentMiddleware() agent_executor AgentExecutor( agentagent, tools[weather_tool], verboseFalse, # 设为False因为我们用自己的中间件记录日志 handle_parsing_errorsTrue, # 处理Agent输出解析错误 callbacks[middleware] # 关键步骤注入回调/中间件 )4.2 运行并观察中间件效果现在让我们运行这个Agent并观察中间件打印的日志。# 运行Agent print( 第一次查询正常情况 ) try: result agent_executor.invoke({input: 今天北京的天气怎么样, chat_history: []}) print(f最终答案: {result[output]}) except Exception as e: print(f执行出错: {e}) print(\n *50 \n) print( 第二次查询查询不存在的城市 ) try: result agent_executor.invoke({input: 火星的天气呢, chat_history: []}) print(f最终答案: {result[output]}) except Exception as e: print(f执行出错: {e}) # 打印本次会话的汇总指标 print(f\n会话汇总指标: {middleware.session_metrics})运行这段代码记得替换有效的OpenAI API Key你会在控制台看到类似如下的输出 第一次查询正常情况 [LLM_START] RunId: xxxx-xxxx... | Prompt Preview: ...包含ReAct指令和问题 [AGENT_ACTION] RunId: xxxx-xxxx... | Tool: WeatherLookup | Tool Input: {location: 北京} | Thought: 用户想知道北京的天气我需要使用WeatherLookup工具... [TOOL_START] RunId: xxxx-xxxx... | Tool: WeatherLookup | Input: {location: 北京} [TOOL_END] RunId: xxxx-xxxx... | Latency: 0.00s | Output: 晴15-25°C微风 [LLM_START] RunId: xxxx-xxxx... | Prompt Preview: ...包含工具返回结果的后续思考 [LLM_END] RunId: xxxx-xxxx... | Latency: 1.23s | Tokens: P(350)/C(45) | Response Preview: 根据查询北京今天的天气是晴气温在15到25摄氏度之间有微风。 [AGENT_FINISH] RunId: xxxx-xxxx... | Final Output: 根据查询北京今天的天气是晴... | Session Total Time: 1.45s | Metrics: {...} 最终答案: 根据查询北京今天的天气是晴气温在15到25摄氏度之间有微风。通过日志你可以清晰地看到Agent的完整思考-行动循环它先思考AGENT_ACTION然后调用工具TOOL_START/END拿到结果后再进行一轮思考LLM_START/END最后给出答案AGENT_FINISH。所有的耗时、Token用量一目了然。4.3 构建中间件链生产级最佳实践对于生产环境我强烈推荐使用中间件链模式。每个中间件只负责一件事比如日志、错误处理、输入验证、速率限制等。下面我们拆解上面的综合中间件class LoggingMiddleware(BaseCallbackHandler): 只负责日志记录的中间件 def on_llm_start(self, serialized, prompts, **kwargs): ... def on_llm_end(self, response, **kwargs): ... def on_tool_start(self, serialized, input_str, **kwargs): ... # ... 实现具体的日志逻辑可以更轻量、更专注 class ErrorHandlingMiddleware(BaseCallbackHandler): 只负责错误处理和告警的中间件 def on_llm_error(self, error, **kwargs): # 发送告警到Slack/钉钉 send_alert_to_slack(fLLM Error in run {kwargs.get(run_id)}: {error}) def on_tool_error(self, error, **kwargs): ... class MetricsMiddleware(BaseCallbackHandler): 只负责收集和上报性能指标的中间件 def __init__(self): self.metrics {} def on_llm_end(self, response, **kwargs): run_id kwargs.get(run_id) # 累加token到对应run_id的指标中 def on_agent_finish(self, finish, **kwargs): # 将本次运行的指标上报到Prometheus/StatsD report_metrics(self.metrics.pop(kwargs.get(run_id), {}))然后在创建AgentExecutor时传入一个中间件列表agent_executor AgentExecutor( agentagent, toolstools, callbacks[LoggingMiddleware(), ErrorHandlingMiddleware(), MetricsMiddleware()] # 中间件链 )LangChain的回调管理器会按顺序调用每个中间件的对应方法。这种设计的好处是你可以随时增删中间件比如在开发环境开启详细的日志中间件在生产环境关闭它但开启监控和告警中间件。5. 生产环境进阶技巧与避坑指南在实际项目中应用Agent中间件你会遇到一些在教程中不常提及的挑战。这里分享几个关键的进阶技巧和踩过的坑。5.1 异步中间件的实现如果你的应用是异步的例如使用FastAPI你需要实现异步的钩子方法。LangChain的AsyncCallbackHandler提供了对应的异步接口。from langchain.callbacks.base import AsyncCallbackHandler import asyncio class AsyncComprehensiveMiddleware(AsyncCallbackHandler): async def on_llm_start(self, serialized, prompts, **kwargs): # 注意这里是async方法 await self._send_log_to_async_service(prompts) async def on_tool_end(self, output, **kwargs): # 异步处理工具输出 pass async def _send_log_to_async_service(self, data): # 模拟异步发送日志 await asyncio.sleep(0.01) print(fAsync Log: {data[:50]}...)在异步环境中创建执行器时使用callbacks参数传入即可。确保你的整个调用链包括工具函数都是异步兼容的。5.2 中间件中的状态管理与并发安全我们的示例中session_metrics是中间件实例的一个属性。这在单线程、顺序执行时没问题。但在多线程或异步并发环境下例如一个Web服务器同时处理多个用户请求这会引发状态混乱。解决方案使用线程/任务本地存储或者将状态与run_id绑定存储在外部的共享存储中。from threading import local import threading class ThreadSafeMetricsMiddleware(BaseCallbackHandler): _thread_local local() def _get_metrics(self): # 获取当前线程的metrics字典 if not hasattr(self._thread_local, metrics): self._thread_local.metrics {} return self._thread_local.metrics def on_llm_start(self, serialized, prompts, **kwargs): run_id kwargs.get(run_id) if run_id: metrics self._get_metrics() metrics[run_id] metrics.get(run_id, {llm_calls: 0}) metrics[run_id][llm_calls] 1对于异步环境可以使用contextvars模块来管理上下文变量。5.3 性能开销与采样率为每一个LLM调用和工具调用都记录全量日志在高频场景下可能会成为性能瓶颈并产生巨大的日志存储成本。应对策略采样记录只对一部分请求例如1%进行详细日志记录。可以在中间件的on_llm_start中根据run_id哈希后取模来决定是否记录。聚合上报对于指标如延迟、Token数不要在每次调用时都写入数据库或日志而是在内存中聚合如每10秒批量上报到监控系统如Prometheus。分级日志使用logging模块的DEBUG、INFO、WARNING等级别。在开发环境使用DEBUG级别输出详细日志在生产环境使用INFO或WARNING级别只记录关键事件和错误。5.4 敏感信息过滤与脱敏Agent处理的输入和输出可能包含用户隐私、API密钥等敏感信息。在中间件中记录这些数据前必须进行脱敏处理。def sanitize_text(text: str, sensitive_patterns: List[str]) - str: 简单的脱敏函数用于演示。 sanitized text for pattern in sensitive_patterns: # 这里可以用正则表达式进行更复杂的匹配和替换 if pattern in sanitized: sanitized sanitized.replace(pattern, [REDACTED]) # 对于过长文本也进行截断 if len(sanitized) 500: sanitized sanitized[:250] ... [TRUNCATED] ... sanitized[-250:] return sanitized class SanitizingLoggingMiddleware(BaseCallbackHandler): def __init__(self, sensitive_keywordsNone): self.sensitive_keywords sensitive_keywords or [password, api_key, token] def on_llm_start(self, serialized, prompts, **kwargs): safe_prompts [sanitize_text(p, self.sensitive_keywords) for p in prompts] logger.info(fSafe Prompt: {safe_prompts[0]})5.5 常见问题排查速查表在实际使用中你可能会遇到以下问题问题现象可能原因排查步骤与解决方案中间件钩子完全没有被触发1. 中间件类未正确继承BaseCallbackHandler。2. 未将中间件实例传入AgentExecutor的callbacks参数。3. Agent执行时设置了verboseTrueLangChain使用了默认的回调。1. 检查类定义。2. 确认创建AgentExecutor时传入了callbacks[your_middleware]。3. 将verbose设为False或确保你的中间件也被包含在默认回调管理器中更复杂。on_llm_end收不到token_usageLLM提供商如某些本地模型或配置未在响应中返回token使用信息。检查response.llm_output字典的内容。如果为空可以考虑在on_llm_start时估算Prompt Token数例如使用tiktoken库在on_llm_end时估算Completion Token数。并发时日志和指标错乱中间件使用了实例属性存储状态多个请求共享了同一个实例。采用“状态与run_id绑定”的模式使用线程本地存储或外部字典来管理状态。工具调用出错后on_tool_end仍被调用通常不会。如果工具抛出异常会先触发on_tool_error而不会触发on_tool_end。这是正常行为。确保你的错误处理逻辑写在on_tool_error中并且清理可能在on_tool_start中设置的临时状态如计时器。想修改Agent的决策如禁止调用某个工具on_agent_action钩子只提供了“观察”能力无法直接修改或否决行动。需要更底层的控制。可以考虑1. 创建自定义的AgentExecutor子类重写_take_next_step方法。2. 或者在工具层面做限制让工具在收到非法请求时返回特定错误信息引导Agent重新思考。最后记住中间件的核心价值是非侵入式。它让你能够增强、监控和控制Agent的行为而无需修改其核心逻辑。从简单的日志开始逐步根据需要添加错误处理、指标收集、安全检查等中间件是构建健壮、可维护的AI Agent应用的最佳路径。附上的完整代码已经提供了一个坚实的起点你可以根据项目的具体需求对这些钩子进行任意的组合和扩展。 LangChain Agent中间件实战：六类钩子函数实现可观测性与流程控制
1. 项目概述为什么我们需要Agent中间件如果你正在用LangChain构建AI Agent大概率遇到过这样的场景Agent执行一个查询任务中途调用了搜索工具但返回的结果总是不尽如人意你想知道它到底搜了什么关键词或者为什么它决定调用这个工具而不是另一个。又或者在生产环境中你需要记录每一次Agent的决策过程用于审计、计费或性能分析。这时候直接修改Agent的核心逻辑不仅侵入性强而且容易出错。这就是Agent中间件或者说钩子函数Hooks大显身手的地方。简单来说Agent中间件就像给Agent的执行流程安装了一系列“监控探头”和“干预开关”。它允许你在Agent生命周期的关键节点如调用工具前、收到LLM响应后、执行动作前插入自定义逻辑而不需要改动Agent本身的代码。这带来了几个核心价值可观测性Observability让你能看清Agent内部的“黑盒”决策可控性Control让你能在关键时刻修正或引导Agent的行为以及非侵入式扩展保持核心代码的整洁与稳定。我最初接触这个概念时觉得它有点“高级”似乎只有构建复杂系统时才用得上。但实际用下来发现无论是调试一个简单的对话机器人还是部署一个需要严格合规的生产级Agent中间件都是提升开发效率和系统健壮性的利器。接下来我会结合六种核心的钩子函数从最基础的日志记录到高级的流程拦截与修改手把手带你实现一套完整的中间件系统并附上可直接运行的教学代码。2. 核心钩子函数深度解析与设计思路LangChain的Agent执行器AgentExecutor提供了一套丰富的回调Callbacks系统而我们要构建的中间件本质上是基于这套回调系统的、更结构化、更面向业务逻辑的封装。理解每种钩子触发的时机和能获取到的上下文信息是灵活运用的关键。2.1 六种钩子函数的定位与分工我们可以将这六种钩子分为三大类观测类、拦截/修改类和生命周期类。它们覆盖了Agent从思考到行动的全过程。观测类钩子用于“看”和“记”这类钩子主要用于记录日志、收集指标和监控通常不修改执行流程。on_llm_start / on_llm_end分别在向大语言模型LLM发送请求前和收到响应后触发。这是记录原始Prompt和Completion、计算Token消耗和延迟的黄金位置。on_tool_start / on_tool_end分别在调用一个工具如搜索、计算、数据库查询的前后触发。你可以在这里记录工具的名称、输入参数、执行结果和耗时对于调试工具调用逻辑异常有用。拦截/修改类钩子用于“控”和“改”这类钩子能力最强允许你修改即将发生的行为的参数甚至直接返回一个结果来“短路”整个调用。 3.on_llm_error在LLM调用出错时触发。你可以在这里实现重试逻辑例如遇到速率限制时等待后重试或者将错误信息转换为对用户更友好的提示。 4.on_tool_error在工具调用出错时触发。例如当调用一个外部API失败时你可以在这里尝试备用方案或者给Agent一个预设的兜底答案避免整个流程崩溃。生命周期类钩子用于感知“阶段”这类钩子让你感知Agent执行过程中的关键阶段转换。 5.on_chain_start / on_chain_end在Agent执行一个“链”Chain的前后触发。在Agent的上下文中每一次完整的“思考-行动”循环可以看作一个链。这里适合记录高层次的任务进度和整体状态。 6.on_agent_action / on_agent_finish这是Agent特有的钩子。on_agent_action在Agent决定要采取某个具体行动即调用某个工具时触发on_agent_finish在Agent得出最终答案并结束时触发。它们是理解Agent决策逻辑的核心。注意on_llm_error和on_tool_error虽然归类为拦截类但它们的首要职责是错误处理。修改流程如重试是建立在处理错误这个基础之上的高级应用。2.2 中间件系统的架构设计考量在设计中间件时我们面临几个关键选择这直接影响到代码的灵活性和复杂度。单一中间件 vs. 中间件链单一中间件一个类实现所有需要的钩子。优点是简单、内聚所有逻辑都在一处。缺点是随着功能增加这个类会变得臃肿难以维护。中间件链定义多个专注单一职责的中间件类如LoggingMiddleware、ErrorHandlingMiddleware、ValidationMiddleware然后按顺序组合。这是更符合生产环境的做法它遵循了单一职责原则方便独立测试和复用。例如你可以先经过验证中间件检查输入再经过日志中间件记录最后交给业务逻辑。同步 vs. 异步LangChain原生支持异步执行。如果你的Agent需要处理高并发请求或者内部调用的工具/LLM本身就是异步的那么实现异步中间件使用async/await能极大提升吞吐量。教学代码我们将以同步版本为主因为它更易于理解但我会指出关键位置如何改为异步。状态管理中间件经常需要共享状态。例如一个用于计费的中间件需要累计整个会话的Token使用量。这个状态应该存在哪里有两种常见模式实例属性在中间件类内部定义self.token_count。这适用于该中间件独享的状态。上下文传递通过LangChain回调系统的run_id、parent_run_id等属性将状态存储在外部存储如Redis、内存字典中键为run_id。这适用于需要在多个独立中间件或不同请求间共享的状态。我们的示例将采用更简单的实例属性方式。3. 手把手实现六类钩子中间件理论讲完了我们开始动手。我将实现一个功能相对完整的AgentMiddleware类它集成了观测、错误处理和简单干预的能力。之后我们再探讨如何将其拆分为链式结构。3.1 基础框架与观测类钩子实现首先我们需要创建一个继承自BaseCallbackHandler的类。这是LangChain所有回调的基类。from typing import Any, Dict, List, Optional from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import LLMResult, AgentAction, AgentFinish from langchain.schema.messages import BaseMessage import time import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class ComprehensiveAgentMiddleware(BaseCallbackHandler): 一个综合的Agent中间件示例演示六类钩子。 def __init__(self): super().__init__() # 用于存储当前会话的元数据 self.session_metrics { total_llm_calls: 0, total_tool_calls: 0, total_tokens_used: 0, start_time: time.time() } # 用于临时存储当前LLM或工具调用的开始时间 self._current_start_time None # 1. 观测类钩子LLM调用 def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any ) - None: LLM调用开始时触发。 self._current_start_time time.time() self.session_metrics[total_llm_calls] 1 run_id kwargs.get(run_id, N/A) # 记录Prompt生产环境中可能需要对长文本进行截断或脱敏 prompt_preview prompts[0][:100] ... if len(prompts[0]) 100 else prompts[0] logger.info(f[LLM_START] RunId: {run_id} | Prompt Preview: {prompt_preview}) # 在实际应用中你可以在这里将Prompt发送到监控平台 def on_llm_end(self, response: LLMResult, **kwargs: Any) - None: LLM调用成功结束时触发。 if self._current_start_time is None: return latency time.time() - self._current_start_time run_id kwargs.get(run_id, N/A) # 提取Token使用量如果LLM提供商返回了该信息 token_usage response.llm_output.get(token_usage, {}) if response.llm_output else {} completion_tokens token_usage.get(completion_tokens, 0) prompt_tokens token_usage.get(prompt_tokens, 0) total_tokens token_usage.get(total_tokens, 0) self.session_metrics[total_tokens_used] total_tokens # 记录响应和耗时 # 注意response.generations 是一个列表的列表结构复杂通常我们取第一条结果的第一条文本 try: first_generation response.generations[0][0] response_text first_generation.text[:200] ... if len(first_generation.text) 200 else first_generation.text except (IndexError, AttributeError): response_text Unable to parse response logger.info( f[LLM_END] RunId: {run_id} | Latency: {latency:.2f}s | fTokens: P({prompt_tokens})/C({completion_tokens}) | fResponse Preview: {response_text} ) self._current_start_time None # 2. 观测类钩子工具调用 def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs: Any ) - None: 工具调用开始时触发。 self._current_start_time time.time() self.session_metrics[total_tool_calls] 1 run_id kwargs.get(run_id, N/A) tool_name serialized.get(name, unknown_tool) logger.info(f[TOOL_START] RunId: {run_id} | Tool: {tool_name} | Input: {input_str}) def on_tool_end(self, output: str, **kwargs: Any) - None: 工具调用成功结束时触发。 if self._current_start_time is None: return latency time.time() - self._current_start_time run_id kwargs.get(run_id, N/A) # 对输出进行安全处理避免日志过长或包含敏感信息 safe_output output[:300] ... if len(output) 300 else output logger.info(f[TOOL_END] RunId: {run_id} | Latency: {latency:.2f}s | Output: {safe_output}) self._current_start_time None在上面的代码中我们实现了四个基础的观测钩子。关键点在于run_id这是LangChain为每一次执行链或Agent运行生成的唯一标识符用于串联所有相关的日志事件。信息截断在生产环境中Prompt和Response可能非常长直接打印会拖慢日志系统并可能泄露敏感数据。进行适当截断是必要的。耗时计算我们利用_current_start_time这个临时变量来配对开始和结束事件计算精确的延迟。这里有一个潜在风险如果on_llm_start被调用但on_llm_end因异常未被调用这个变量将不会被重置。更健壮的做法是使用一个以run_id为键的字典来管理多个并发调用的计时器。3.2 拦截修改类与生命周期钩子实现接下来我们实现错误处理和感知Agent生命周期的钩子。# 3. 拦截/修改类钩子错误处理 def on_llm_error(self, error: Exception, **kwargs: Any) - None: LLM调用出错时触发。 run_id kwargs.get(run_id, N/A) logger.error(f[LLM_ERROR] RunId: {run_id} | Error: {repr(error)}) # 重要这里可以尝试重试逻辑但注意不要陷入无限重试。 # 例如如果是速率限制错误可以等待一段时间。 # 但在这个基础示例中我们仅记录日志。 self._current_start_time None # 确保清理计时器 def on_tool_error(self, error: Exception, **kwargs: Any) - None: 工具调用出错时触发。 run_id kwargs.get(run_id, N/A) logger.error(f[TOOL_ERROR] RunId: {run_id} | Error: {repr(error)}) # 示例针对特定错误进行干预 # 假设我们有一个调用外部天气API的工具当它失败时我们提供一个友好的兜底信息。 # 注意直接在这里修改输出是困难的因为错误已经发生。 # 更常见的模式是在构建工具时在工具内部进行错误处理和兜底。 # 这个钩子更适合记录错误、触发告警或更新熔断器状态。 self._current_start_time None # 4. 生命周期类钩子Agent决策 def on_agent_action(self, action: AgentAction, **kwargs: Any) - None: Agent决定采取一个具体行动时触发。 run_id kwargs.get(run_id, N/A) # action.log 包含了Agent的思考过程非常有用 thought_process action.log[:200] ... if len(action.log) 200 else action.log logger.info( f[AGENT_ACTION] RunId: {run_id} | fTool: {action.tool} | Tool Input: {action.tool_input} | fThought: {thought_process} ) # 这里是一个潜在的“拦截点”你可以根据策略检查action.tool和tool_input # 如果不符合规则例如试图调用一个高风险工具可以抛出异常来终止本次行动。 # 但这需要更精细的流程控制可能结合自定义的AgentExecutor来实现。 def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) - None: Agent完成所有思考输出最终答案时触发。 run_id kwargs.get(run_id, N/A) total_time time.time() - self.session_metrics[start_time] logger.info( f[AGENT_FINISH] RunId: {run_id} | fFinal Output: {finish.return_values[output][:500]}... | fSession Total Time: {total_time:.2f}s | fMetrics: {self.session_metrics} ) # 此时可以将会话指标如总token数、调用次数上报到监控系统用于计费和性能分析。 # 注意on_chain_start/end 对于理解复杂链的嵌套很有用但在简单Agent中可能不常使用。 # 为保持示例完整这里也简单实现。 def on_chain_start( self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any ) - None: run_id kwargs.get(run_id, N/A) chain_name serialized.get(name, unknown_chain) # logger.debug(f[CHAIN_START] RunId: {run_id} | Chain: {chain_name}) # 通常用debug级别 def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) - None: run_id kwargs.get(run_id, N/A) # logger.debug(f[CHAIN_END] RunId: {run_id})错误处理钩子on_llm_error,on_tool_error是增强系统鲁棒性的关键。但需要理解它们的局限性它们是在错误发生后被调用的。如果你想防止错误发生或者错误发生后提供替代结果通常需要在更早的阶段介入。例如如果你想在工具调用超时时自动重试更好的地方可能是在工具类的_call方法内部实现重试逻辑或者使用一个专门的“重试工具包装器”中间件。错误钩子更适合做最后的日志记录和告警。on_agent_action和on_agent_finish是理解Agent“思维过程”的窗口。action.log里通常包含了LLM生成的在最终工具调用之前的推理文本这对于调试Agent为什么做出某个决定至关重要。4. 集成中间件到LangChain Agent实战现在我们有了一个功能齐全的中间件类。接下来看看如何将它实际应用到你的LangChain Agent中。我将创建一个简单的“天气查询Agent”作为示例。4.1 创建工具与Agentfrom langchain.agents import Tool, AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import PromptTemplate import requests # 1. 定义一个简单的天气查询工具模拟 def get_weather(location: str) - str: 查询指定城市的天气。输入应为城市名。 # 这里是模拟真实情况应调用天气API # 模拟网络错误 # if location error_city: # raise ConnectionError(模拟网络错误无法连接到天气服务) weather_data { 北京: 晴15-25°C微风, 上海: 多云18-28°C东南风3级, 广州: 阵雨22-30°C南风4级, } return weather_data.get(location, f抱歉未找到{city}的天气信息。) # 将函数封装成LangChain Tool weather_tool Tool( nameWeatherLookup, funcget_weather, description用于查询城市天气。输入一个城市名称如‘北京’。 ) # 2. 初始化LLM # 请替换为你的实际API Key或使用其他兼容的LLM llm ChatOpenAI( modelgpt-3.5-turbo, temperature0, openai_api_keyyour-api-key-here # 请务必替换 ) # 3. 使用ReAct框架创建Agent from langchain import hub prompt hub.pull(hwchase17/react-chat) # 一个标准的ReAct提示模板 agent create_react_agent(llm, tools[weather_tool], promptprompt) # 4. 创建Agent执行器并注入我们的中间件 middleware ComprehensiveAgentMiddleware() agent_executor AgentExecutor( agentagent, tools[weather_tool], verboseFalse, # 设为False因为我们用自己的中间件记录日志 handle_parsing_errorsTrue, # 处理Agent输出解析错误 callbacks[middleware] # 关键步骤注入回调/中间件 )4.2 运行并观察中间件效果现在让我们运行这个Agent并观察中间件打印的日志。# 运行Agent print( 第一次查询正常情况 ) try: result agent_executor.invoke({input: 今天北京的天气怎么样, chat_history: []}) print(f最终答案: {result[output]}) except Exception as e: print(f执行出错: {e}) print(\n *50 \n) print( 第二次查询查询不存在的城市 ) try: result agent_executor.invoke({input: 火星的天气呢, chat_history: []}) print(f最终答案: {result[output]}) except Exception as e: print(f执行出错: {e}) # 打印本次会话的汇总指标 print(f\n会话汇总指标: {middleware.session_metrics})运行这段代码记得替换有效的OpenAI API Key你会在控制台看到类似如下的输出 第一次查询正常情况 [LLM_START] RunId: xxxx-xxxx... | Prompt Preview: ...包含ReAct指令和问题 [AGENT_ACTION] RunId: xxxx-xxxx... | Tool: WeatherLookup | Tool Input: {location: 北京} | Thought: 用户想知道北京的天气我需要使用WeatherLookup工具... [TOOL_START] RunId: xxxx-xxxx... | Tool: WeatherLookup | Input: {location: 北京} [TOOL_END] RunId: xxxx-xxxx... | Latency: 0.00s | Output: 晴15-25°C微风 [LLM_START] RunId: xxxx-xxxx... | Prompt Preview: ...包含工具返回结果的后续思考 [LLM_END] RunId: xxxx-xxxx... | Latency: 1.23s | Tokens: P(350)/C(45) | Response Preview: 根据查询北京今天的天气是晴气温在15到25摄氏度之间有微风。 [AGENT_FINISH] RunId: xxxx-xxxx... | Final Output: 根据查询北京今天的天气是晴... | Session Total Time: 1.45s | Metrics: {...} 最终答案: 根据查询北京今天的天气是晴气温在15到25摄氏度之间有微风。通过日志你可以清晰地看到Agent的完整思考-行动循环它先思考AGENT_ACTION然后调用工具TOOL_START/END拿到结果后再进行一轮思考LLM_START/END最后给出答案AGENT_FINISH。所有的耗时、Token用量一目了然。4.3 构建中间件链生产级最佳实践对于生产环境我强烈推荐使用中间件链模式。每个中间件只负责一件事比如日志、错误处理、输入验证、速率限制等。下面我们拆解上面的综合中间件class LoggingMiddleware(BaseCallbackHandler): 只负责日志记录的中间件 def on_llm_start(self, serialized, prompts, **kwargs): ... def on_llm_end(self, response, **kwargs): ... def on_tool_start(self, serialized, input_str, **kwargs): ... # ... 实现具体的日志逻辑可以更轻量、更专注 class ErrorHandlingMiddleware(BaseCallbackHandler): 只负责错误处理和告警的中间件 def on_llm_error(self, error, **kwargs): # 发送告警到Slack/钉钉 send_alert_to_slack(fLLM Error in run {kwargs.get(run_id)}: {error}) def on_tool_error(self, error, **kwargs): ... class MetricsMiddleware(BaseCallbackHandler): 只负责收集和上报性能指标的中间件 def __init__(self): self.metrics {} def on_llm_end(self, response, **kwargs): run_id kwargs.get(run_id) # 累加token到对应run_id的指标中 def on_agent_finish(self, finish, **kwargs): # 将本次运行的指标上报到Prometheus/StatsD report_metrics(self.metrics.pop(kwargs.get(run_id), {}))然后在创建AgentExecutor时传入一个中间件列表agent_executor AgentExecutor( agentagent, toolstools, callbacks[LoggingMiddleware(), ErrorHandlingMiddleware(), MetricsMiddleware()] # 中间件链 )LangChain的回调管理器会按顺序调用每个中间件的对应方法。这种设计的好处是你可以随时增删中间件比如在开发环境开启详细的日志中间件在生产环境关闭它但开启监控和告警中间件。5. 生产环境进阶技巧与避坑指南在实际项目中应用Agent中间件你会遇到一些在教程中不常提及的挑战。这里分享几个关键的进阶技巧和踩过的坑。5.1 异步中间件的实现如果你的应用是异步的例如使用FastAPI你需要实现异步的钩子方法。LangChain的AsyncCallbackHandler提供了对应的异步接口。from langchain.callbacks.base import AsyncCallbackHandler import asyncio class AsyncComprehensiveMiddleware(AsyncCallbackHandler): async def on_llm_start(self, serialized, prompts, **kwargs): # 注意这里是async方法 await self._send_log_to_async_service(prompts) async def on_tool_end(self, output, **kwargs): # 异步处理工具输出 pass async def _send_log_to_async_service(self, data): # 模拟异步发送日志 await asyncio.sleep(0.01) print(fAsync Log: {data[:50]}...)在异步环境中创建执行器时使用callbacks参数传入即可。确保你的整个调用链包括工具函数都是异步兼容的。5.2 中间件中的状态管理与并发安全我们的示例中session_metrics是中间件实例的一个属性。这在单线程、顺序执行时没问题。但在多线程或异步并发环境下例如一个Web服务器同时处理多个用户请求这会引发状态混乱。解决方案使用线程/任务本地存储或者将状态与run_id绑定存储在外部的共享存储中。from threading import local import threading class ThreadSafeMetricsMiddleware(BaseCallbackHandler): _thread_local local() def _get_metrics(self): # 获取当前线程的metrics字典 if not hasattr(self._thread_local, metrics): self._thread_local.metrics {} return self._thread_local.metrics def on_llm_start(self, serialized, prompts, **kwargs): run_id kwargs.get(run_id) if run_id: metrics self._get_metrics() metrics[run_id] metrics.get(run_id, {llm_calls: 0}) metrics[run_id][llm_calls] 1对于异步环境可以使用contextvars模块来管理上下文变量。5.3 性能开销与采样率为每一个LLM调用和工具调用都记录全量日志在高频场景下可能会成为性能瓶颈并产生巨大的日志存储成本。应对策略采样记录只对一部分请求例如1%进行详细日志记录。可以在中间件的on_llm_start中根据run_id哈希后取模来决定是否记录。聚合上报对于指标如延迟、Token数不要在每次调用时都写入数据库或日志而是在内存中聚合如每10秒批量上报到监控系统如Prometheus。分级日志使用logging模块的DEBUG、INFO、WARNING等级别。在开发环境使用DEBUG级别输出详细日志在生产环境使用INFO或WARNING级别只记录关键事件和错误。5.4 敏感信息过滤与脱敏Agent处理的输入和输出可能包含用户隐私、API密钥等敏感信息。在中间件中记录这些数据前必须进行脱敏处理。def sanitize_text(text: str, sensitive_patterns: List[str]) - str: 简单的脱敏函数用于演示。 sanitized text for pattern in sensitive_patterns: # 这里可以用正则表达式进行更复杂的匹配和替换 if pattern in sanitized: sanitized sanitized.replace(pattern, [REDACTED]) # 对于过长文本也进行截断 if len(sanitized) 500: sanitized sanitized[:250] ... [TRUNCATED] ... sanitized[-250:] return sanitized class SanitizingLoggingMiddleware(BaseCallbackHandler): def __init__(self, sensitive_keywordsNone): self.sensitive_keywords sensitive_keywords or [password, api_key, token] def on_llm_start(self, serialized, prompts, **kwargs): safe_prompts [sanitize_text(p, self.sensitive_keywords) for p in prompts] logger.info(fSafe Prompt: {safe_prompts[0]})5.5 常见问题排查速查表在实际使用中你可能会遇到以下问题问题现象可能原因排查步骤与解决方案中间件钩子完全没有被触发1. 中间件类未正确继承BaseCallbackHandler。2. 未将中间件实例传入AgentExecutor的callbacks参数。3. Agent执行时设置了verboseTrueLangChain使用了默认的回调。1. 检查类定义。2. 确认创建AgentExecutor时传入了callbacks[your_middleware]。3. 将verbose设为False或确保你的中间件也被包含在默认回调管理器中更复杂。on_llm_end收不到token_usageLLM提供商如某些本地模型或配置未在响应中返回token使用信息。检查response.llm_output字典的内容。如果为空可以考虑在on_llm_start时估算Prompt Token数例如使用tiktoken库在on_llm_end时估算Completion Token数。并发时日志和指标错乱中间件使用了实例属性存储状态多个请求共享了同一个实例。采用“状态与run_id绑定”的模式使用线程本地存储或外部字典来管理状态。工具调用出错后on_tool_end仍被调用通常不会。如果工具抛出异常会先触发on_tool_error而不会触发on_tool_end。这是正常行为。确保你的错误处理逻辑写在on_tool_error中并且清理可能在on_tool_start中设置的临时状态如计时器。想修改Agent的决策如禁止调用某个工具on_agent_action钩子只提供了“观察”能力无法直接修改或否决行动。需要更底层的控制。可以考虑1. 创建自定义的AgentExecutor子类重写_take_next_step方法。2. 或者在工具层面做限制让工具在收到非法请求时返回特定错误信息引导Agent重新思考。最后记住中间件的核心价值是非侵入式。它让你能够增强、监控和控制Agent的行为而无需修改其核心逻辑。从简单的日志开始逐步根据需要添加错误处理、指标收集、安全检查等中间件是构建健壮、可维护的AI Agent应用的最佳路径。附上的完整代码已经提供了一个坚实的起点你可以根据项目的具体需求对这些钩子进行任意的组合和扩展。