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从思考到行动的全过程。

观测类钩子：用于“看”和“记”这类钩子主要用于记录日志、收集指标和监控，通常不修改执行流程。

1. on_llm_start / on_llm_end：分别在向大语言模型（LLM）发送请求前和收到响应后触发。这是记录原始Prompt和Completion、计算Token消耗和延迟的黄金位置。
2. 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使用量。这个状态应该存在哪里？有两种常见模式：

1. 实例属性：在中间件类内部定义self.token_count。这适用于该中间件独享的状态。
2. 上下文传递：通过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(level=logging.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 | " f"Tokens: P({prompt_tokens})/C({completion_tokens}) | " f"Response 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} | " f"Tool: {action.tool} | Tool Input: {action.tool_input} | " f"Thought: {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} | " f"Final Output: {finish.return_values['output'][:500]}... | " f"Session Total Time: {total_time:.2f}s | " f"Metrics: {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 创建工具与Agent

from 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( name="WeatherLookup", func=get_weather, description="用于查询城市天气。输入一个城市名称，如‘北京’。" ) # 2. 初始化LLM # 请替换为你的实际API Key，或使用其他兼容的LLM llm = ChatOpenAI( model="gpt-3.5-turbo", temperature=0, openai_api_key="your-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], prompt=prompt) # 4. 创建Agent执行器，并注入我们的中间件 middleware = ComprehensiveAgentMiddleware() agent_executor = AgentExecutor( agent=agent, tools=[weather_tool], verbose=False, # 设为False，因为我们用自己的中间件记录日志 handle_parsing_errors=True, # 处理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(f"LLM 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( agent=agent, tools=tools, 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(f"Async 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. 采样记录：只对一部分请求（例如1%）进行详细日志记录。可以在中间件的on_llm_start中根据run_id哈希后取模来决定是否记录。
2. 聚合上报：对于指标（如延迟、Token数），不要在每次调用时都写入数据库或日志，而是在内存中聚合（如每10秒），批量上报到监控系统（如Prometheus）。
3. 分级日志：使用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_keywords=None): 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(f"Safe Prompt: {safe_prompts[0]}")

5.5 常见问题排查速查表

在实际使用中，你可能会遇到以下问题：

最后，记住中间件的核心价值是非侵入式。它让你能够增强、监控和控制Agent的行为，而无需修改其核心逻辑。从简单的日志开始，逐步根据需要添加错误处理、指标收集、安全检查等中间件，是构建健壮、可维护的AI Agent应用的最佳路径。附上的完整代码已经提供了一个坚实的起点，你可以根据项目的具体需求，对这些钩子进行任意的组合和扩展。