1. 从“黑盒”到“白盒”为什么我们需要Middleware来定制Agent Harness如果你最近在折腾AI Agent开发尤其是用过LangChain、LangGraph这类框架大概率会碰到一个瓶颈Agent的执行流程像个黑盒。你定义好工具Tools、设定好提示词Prompt然后调用agent.run()接下来就只能祈祷它按你预想的路径去执行。当它出错、卡住或者行为不符合预期时你很难介入只能一遍遍地调整提示词或者重构工具过程非常被动且低效。这就是Middleware中间件的价值所在。它不是一个新概念在Web开发、后端服务中早已是基础设施。简单来说Middleware就是在核心处理逻辑的前后插入你自己的代码逻辑从而实现对流程的观察、控制和修改。在Agent Harness你可以理解为承载和运行Agent的“马具”或“框架”的语境下Middleware让你能深入到Agent决策和执行的每一个关键环节把黑盒变成白盒。想象一下你的Agent是一个在迷宫里执行任务的机器人。没有Middleware时你只能站在迷宫入口给它一张任务清单然后等它在出口给你结果。你完全不知道它在里面是走错了路、被卡住了还是中途改变了主意。而Middleware就像是在迷宫的每个岔路口、每个房间都安装了摄像头和遥控门。你不仅能实时看到它的每一步日志和监控还能在它即将走入死胡同时远程打开另一扇门修改输入或决策甚至能在它拿到错误物品时让它先做个验证预处理输出。Agent Harness比如LangChain的AgentExecutor、LangGraph的StateGraph它们提供了Agent运行的基础设施管理对话历史、调用工具、迭代执行直到得出最终答案。但它们的默认行为是固定的。Middleware则是你定制这个基础设施的“手术刀”。通过它你可以监控与可观测性记录每一次LLM调用、工具调用的输入输出、耗时、token消耗。流程干预在LLM思考前修改用户问题在工具调用后验证或修正结果甚至在Agent决定结束前插入额外的确认步骤。错误处理与韧性增强捕获工具调用异常并尝试重试或提供降级方案而不是让整个Agent直接崩溃。功能扩展轻松地为所有Agent统一添加缓存、限流、审计等能力。没有Middleware构建一个健壮、可控、可调试的Agent系统会异常困难。你可能会把大量逻辑硬编码到工具函数或提示词里导致代码臃肿且难以维护。Middleware提供了一种清晰、解耦的方式来增强你的Agent Harness这正是从“玩具Demo”走向“生产级应用”的关键一步。2. 解剖Agent Harness理解Middleware可以插入的“钩子”要有效使用Middleware首先得明白你的Agent Harness在运行时究竟经历了哪些步骤。不同的框架LangChain, LangGraph, 自定义框架细节略有不同但核心流程是相似的。我们以一个典型的基于ReActReasoning and Action模式的Agent执行流程为例拆解出Middleware可以介入的关键“钩子”。一个标准的Agent运行周期One Agent Step通常包含以下阶段输入预处理原始用户输入或上一轮的输出进入系统。Agent决策LLM调用Harness将当前的对话历史、可用工具列表等信息组织成提示词发送给LLM请求其给出下一步的决策思考过程thought和行动action。工具分发与执行解析LLM返回的action通常是工具名和输入参数找到对应的工具函数并执行。工具结果处理获取工具的执行结果observation。结果组装与循环判断将工具结果组装进对话历史并判断Agent是否应该继续执行产生最终答案final_answer还是进入下一轮循环。Middleware正是在这些阶段之间提供了注入点。以LangChain的AgentExecutor为例它支持callbacks这本质上就是一种Middleware模式。更现代的框架如LangGraph其StateGraph的checkpointer和interrupts机制也提供了强大的流程干预能力。我们可以将这些注入点归纳为三类核心Middleware围绕LLM调用的Middleware在调用LLM之前和之后插入逻辑。Pre-LLM可以修改发送给LLM的提示词messages。例如自动为问题添加当前日期时间上下文或者过滤掉历史消息中的敏感信息。Post-LLM可以解析、验证或修改LLM返回的原始响应。例如确保返回的JSON格式正确或者检测到模型产生了有害内容时替换成一个安全回复。围绕工具调用的Middleware在调用工具之前和之后插入逻辑。Pre-Tool可以修改工具调用的参数。例如对用户输入的查询参数进行标准化处理如城市名补全或进行权限校验。Post-Tool可以处理、验证或包装工具返回的结果。这是最常用的钩子之一。例如当工具返回一个庞大的JSON时你可以用另一个LLM调用对其进行总结提炼再将精简后的结果交给Agent。或者当工具调用失败如网络超时时自动重试或返回一个友好的错误信息。围绕执行流程的Middleware在Agent每一步开始或结束时插入逻辑。On-Step-Start/End用于全局的日志记录、性能监控、状态快照。你可以在这里记录每一步的完整状态State便于事后调试和复现问题。On-Stream如果Agent支持流式输出这个Middleware可以拦截中间生成的token用于实现实时展示、内容过滤或特殊格式处理。理解这些“钩子”的位置是设计有效Middleware的前提。它让你明确知道当你想实现某个特定功能时代码应该“挂”在流程的哪个环节。3. 实战为LangChain Agent添加自定义Middleware理论说再多不如动手写一行代码。我们以最流行的LangChain框架为例演示如何为其AgentExecutor添加两个实用的自定义Middleware。虽然LangChain官方文档可能更倾向于使用Callbacks但我们可以通过继承和包装的方式实现Middleware模式。假设我们有一个简单的天气查询Agent。它有一个工具get_weather(city: str)能返回该城市的天气信息。3.1 实现一个工具结果验证与摘要Middleware这个Middleware将在工具调用之后生效。它的目标是如果工具返回的原始数据过于冗长比如包含未来7天每小时的详细数据则自动调用一个LLM对其进行摘要只将核心信息如明天是否下雨、最高最低温度传递给Agent的下一步思考。from typing import Any, Dict, List, Optional, Union from langchain.agents import AgentExecutor from langchain.callbacks.manager import CallbackManagerForChainRun from langchain.schema import AgentAction, AgentFinish from langchain.tools import BaseTool from langchain.chat_models import ChatOpenAI import json class ToolSummaryMiddleware: 工具结果摘要中间件 def __init__(self, llm): # 用于做摘要的LLM可以与主Agent使用不同的模型 self.summary_llm llm def post_tool_process(self, tool_name: str, tool_input: str, raw_output: str, run_manager: Optional[CallbackManagerForChainRun] None) - str: 在工具原始输出返回给Agent前进行处理。 # 判断是否需要摘要例如仅对get_weather工具且返回数据较长时处理 if tool_name get_weather and len(raw_output) 500: try: # 构造摘要提示词 summary_prompt f 请对以下天气数据进行简洁摘要提取最关键的信息给一个旅行助理。 关键信息包括明天及后天的整体天气状况晴/雨/阴、最高最低气温、是否需要带伞。 原始数据 {raw_output[:2000]} # 防止过长 messages [{role: user, content: summary_prompt}] summary_response self.summary_llm.invoke(messages) summarized_output f[经摘要] {summary_response.content} # 记录日志 if run_manager: run_manager.on_text(fMiddleware: 已对工具 {tool_name} 的输出进行摘要长度从 {len(raw_output)} 缩减至 {len(summarized_output)}\n) return summarized_output except Exception as e: # 摘要失败返回原始输出并记录错误 if run_manager: run_manager.on_text(fMiddleware: 工具结果摘要失败错误: {e}返回原始输出\n) return raw_output # 其他情况直接返回原始输出 return raw_output # 使用示例 from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool # 1. 定义工具模拟 def get_weather(city: str) - str: # 模拟返回一个冗长的JSON天气数据 return json.dumps({ city: city, forecast: [ {date: 2023-10-27, condition: Sunny, high: 22, low: 15, precipitation_prob: 10}, {date: 2023-10-28, condition: Rainy, high: 18, low: 12, precipitation_prob: 90}, # ... 更多数据 ], alerts: [明天下午有暴雨黄色预警] }, indent2) weather_tool Tool.from_function( funcget_weather, nameget_weather, description查询指定城市的天气信息 ) # 2. 创建Agent简化流程 from langchain_openai import ChatOpenAI from langchain import hub from langchain.agents import create_react_agent llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) prompt hub.pull(hwchase17/react-chat) agent create_react_agent(llm, tools[weather_tool], promptprompt) # 3. 创建Middleware实例 summary_middleware ToolSummaryMiddleware(llmChatOpenAI(modelgpt-3.5-turbo, temperature0)) # 4. 包装原始的AgentExecutor class MiddlewareEnhancedAgentExecutor(AgentExecutor): 包装了中间件的Agent执行器 def __init__(self, agent_executor: AgentExecutor, middlewares: List[Any] None): # 这里简化了实际需要继承并重写相关方法 super().__init__(**agent_executor.dict()) # 复制属性 self.middlewares middlewares or [] def _call(self, inputs: Dict[str, str], run_manager: Optional[CallbackManagerForChainRun] None) - Dict[str, Any]: # 这里是简化示意实际需要重写_take_next_step等方法在调用工具后插入middleware逻辑 # 伪代码在得到tool_output后 # for middleware in self.middlewares: # if hasattr(middleware, post_tool_process): # tool_output middleware.post_tool_process(tool.name, tool_input, tool_output, run_manager) # 然后继续原有流程 return super()._call(inputs, run_managerrun_manager) # 注意上述包装类是一个概念演示。在实际的LangChain中更规范的做法是通过自定义Callback或继承AgentExecutor并重写_iter_next_step等核心方法来实现。提示在LangChain的最新版本中实现Middleware更优雅的方式可能是创建自定义的BaseCallbackHandler并在on_tool_end事件中处理结果。或者直接使用LangGraph其基于状态机的设计对Middleware通过checkpointer和node的pre/post函数的支持更为原生和强大。3.2 实现一个执行流程日志与监控Middleware这个Middleware更关注全局流程它在每一步的开始和结束时记录详细的状态信息便于调试和性能分析。import time from datetime import datetime class MonitoringMiddleware: 执行监控中间件 def __init__(self, log_file: str agent_execution.log): self.log_file log_file def on_step_start(self, step_id: int, inputs: Dict, run_manager: Optional[CallbackManagerForChainRun] None): 记录步骤开始 start_time time.time() log_entry { timestamp: datetime.now().isoformat(), step_id: step_id, event: step_start, inputs: inputs, start_time: start_time } self._write_log(log_entry) if run_manager: run_manager.on_text(fStep {step_id} started at {log_entry[timestamp]}\n) # 将开始时间附着到上下文中供结束时使用 return {step_start_time: start_time} def on_step_end(self, step_id: int, outputs: Dict, context: Dict, run_manager: Optional[CallbackManagerForChainRun] None): 记录步骤结束计算耗时 end_time time.time() start_time context.get(step_start_time, end_time) duration end_time - start_time log_entry { timestamp: datetime.now().isoformat(), step_id: step_id, event: step_end, outputs: outputs, duration_seconds: round(duration, 2) } self._write_log(log_entry) # 可以在这里设置性能阈值告警 if duration 5.0: # 假设5秒为慢查询阈值 warning_msg fWarning: Step {step_id} took {duration:.2f}s, which is too slow. if run_manager: run_manager.on_text(warning_msg \n) # 在实际生产中这里可以触发告警如发送到Slack、邮件 def _write_log(self, entry: Dict): 将日志写入文件生产环境应接入ELK等系统 import json with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(entry, ensure_asciiFalse) \n) # 使用方式在Agent执行循环中手动调用这些钩子。 # 例如在自定义的AgentExecutor循环里 # monitoring_middleware MonitoringMiddleware() # for step in range(max_iterations): # context monitoring_middleware.on_step_start(step, current_inputs, run_manager) # # ... 执行Agent的一步 ... # monitoring_middleware.on_step_end(step, step_output, context, run_manager)这两个例子展示了Middleware的两种典型应用业务逻辑增强摘要和运维可观测性监控。通过这种方式你将业务核心逻辑工具函数、Agent策略与辅助性、跨切面的逻辑日志、缓存、验证清晰分离。4. 超越LangChain在LangGraph与自定义Harness中设计Middleware体系虽然LangChain的Callback机制能实现部分Middleware功能但在设计上LangGraph和自定义的Agent Harness能给你更清晰、更强大的Middleware集成体验。4.1 LangGraph的Middleware哲学状态、节点与检查点LangGraph的核心是状态机StateGraph。你的Agent流程被建模为一张图Graph节点Node是处理单元如调用LLM、执行工具边Edge定义了状态流转的条件。这种设计让Middleware的植入点变得非常直观。节点级Middleware你可以在每个节点的pre和post函数中注入逻辑。pre函数在节点主逻辑前执行可以修改传入该节点的状态post函数在主逻辑后执行可以修改节点的输出状态。这完美对应了“Pre-LLM”和“Post-Tool”等钩子。图级Middleware通过Checkpointer和Interrupts。Checkpointer允许你在每一步之后持久化整个状态这本身就是一种强大的监控和恢复机制。Interrupts允许你在状态满足特定条件时例如检测到用户输入了“暂停”中断标准流程跳转到你定义的异常处理节点这实现了流程级的动态路由。# LangGraph 节点Middleware概念示例 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 消息历史 needs_human_help: bool # 是否需要人工介入 def llm_node(state: AgentState): 调用LLM的节点 # ... 调用LLM的逻辑 ... return {messages: [new_ai_message]} def human_check_pre_middleware(state: AgentState): 在LLM节点前执行检查是否需要人工优先处理 if state.get(needs_human_help): # 直接返回跳过LLM节点或者跳转到人工处理节点 return {messages: [{role: system, content: 已转接人工。}]} # 否则返回原状态继续执行LLM节点 return state # 构建图 graph_builder StateGraph(AgentState) graph_builder.add_node(call_llm, llm_node) # 理想情况下LangGraph应支持为节点添加pre/post函数。当前版本可能需要通过包装节点函数或使用条件边来实现类似效果。4.2 设计一个自定义的、Middleware友好的Agent Harness如果你想完全掌控或者现有框架不能满足需求自己设计一个轻量级的Agent Harness并不复杂。关键在于采用管道Pipeline或责任链Chain of Responsibility模式来组织Middleware。from abc import ABC, abstractmethod from typing import Any, Dict, List, Callable from dataclasses import dataclass dataclass class AgentContext: Agent执行上下文贯穿整个Middleware链 input: str history: List[Dict] current_step: int 0 llm_response: Any None tool_name: str tool_input: Any None tool_output: Any None final_output: Any None metadata: Dict[str, Any] None # 用于Middleware间传递额外信息 class Middleware(ABC): Middleware基类 abstractmethod async def handle(self, context: AgentContext, next_fn: Callable): 处理请求并决定是否/如何调用下一个Middleware pass class LoggingMiddleware(Middleware): 日志中间件 async def handle(self, context: AgentContext, next_fn: Callable): print(f[Step {context.current_step}] 输入: {context.input[:100]}...) start time.time() # 调用下一个Middleware或核心Agent逻辑 await next_fn(context) elapsed time.time() - start print(f[Step {context.current_step}] 完成耗时: {elapsed:.2f}s) class ValidationMiddleware(Middleware): 验证中间件检查工具输入 async def handle(self, context: AgentContext, next_fn: Callable): if context.tool_name book_flight: if not self._validate_city(context.tool_input.get(city)): context.tool_output 错误城市名称无效。 return # 中断链不再调用下一个中间件/核心逻辑 await next_fn(context) def _validate_city(self, city: str) - bool: return city in [北京, 上海, 广州, 深圳] class AgentHarness: 支持Middleware的简易Agent Harness def __init__(self, llm, tools: List): self.llm llm self.tools {t.name: t for t in tools} self.middlewares: List[Middleware] [] def use(self, middleware: Middleware): 注册Middleware self.middlewares.append(middleware) async def run(self, user_input: str, history: List[Dict] None) - str: 执行Agent顺序通过Middleware链 context AgentContext(inputuser_input, historyhistory or []) # 构建Middleware责任链 async def chain_handler(idx: int): if idx len(self.middlewares): # 调用当前Middleware并传入下一个handler await self.middlewares[idx].handle(context, lambda: chain_handler(idx 1)) else: # 所有Middleware处理完毕执行核心Agent逻辑 await self._core_agent_logic(context) await chain_handler(0) return context.final_output async def _core_agent_logic(self, context: AgentContext): 核心Agent逻辑简化版ReAct # 1. 调用LLM决策 prompt self._build_prompt(context.input, context.history) context.llm_response await self.llm.ainvoke(prompt) # 2. 解析决策调用工具... # 3. 更新context.final_output... pass在这个自定义设计中AgentContext对象承载了所有数据并沿着Middleware链传递。每个Middleware可以读取和修改Context并决定是否中断链或继续。这种模式提供了极大的灵活性是许多成熟Web框架如Express.js, Koa中间件系统的核心思想。5. Middleware的典型应用场景与避坑指南理解了如何实现Middleware我们来看看在哪些具体场景下它会大放异彩以及实践中容易踩的坑。5.1 五大高价值应用场景调试与可观测性这是Middleware最直接的用途。记录每一次LLM请求和响应的原始内容、token用量、耗时。当Agent给出一个匪夷所思的回答时你可以回溯完整的思维链和工具调用历史精准定位问题是出在提示词、工具输出还是LLM本身。错误处理与降级网络请求失败、工具异常、LLM输出格式错误……生产环境中异常无处不在。一个健壮的ErrorHandlingMiddleware可以捕获这些异常根据策略进行重试、替换工具、或返回预设的友好提示避免整个会话崩溃。缓存与性能优化对于重复或相似的查询例如“北京天气怎么样”在短时间内被问多次CachingMiddleware可以在LLM调用或工具调用前检查缓存直接返回结果显著降低延迟和成本。注意缓存键的设计要合理避免语义相似但字面不同的查询无法命中。安全与合规输入过滤在用户输入传递给LLM前过滤敏感词、个人身份信息PII。输出审查在LLM或工具输出最终给用户前进行内容安全审查防止生成有害或不当内容。权限控制在调用特定工具如“转账”、“删除数据”前校验用户身份和权限。流程定制与业务规则注入强制确认在Agent执行“下单”工具前插入一个ConfirmationMiddleware主动向用户发起一次确认“您确定要购买XX吗”。结果后处理对所有Agent的最终答案进行统一的格式化比如添加公司标识、免责声明或翻译成特定语言。5.2 实践中的常见陷阱与解决方案Middleware执行顺序问题多个Middleware注册时执行顺序至关重要。例如一个负责记录原始输入的Middleware应该放在最前面而一个负责最终格式化的Middleware应该放在最后面。在自定义Harness中你需要明确定义注册顺序即执行顺序。在LangChain中Callback的执行顺序可能由框架内部决定需要仔细查阅文档。解决方案为Middleware设计优先级priority字段或在注册时显式声明顺序。在架构设计文档中明确各Middleware的职责和预期位置。Middleware导致的性能瓶颈每个Middleware都增加了一次函数调用和可能的IO操作如写日志、网络请求。如果Middleware逻辑复杂或数量众多会显著拖慢Agent的响应速度。解决方案对非关键Middleware如详细调试日志提供开关在生产环境中关闭。将IO操作日志写入、远程调用异步化async/await避免阻塞主线程。对耗时操作进行性能剖析确保其必要性。状态污染与循环依赖Middleware之间通过共享的Context对象通信。如果Middleware A修改了Context中的某个字段可能会意外影响Middleware B或核心逻辑的行为导致难以调试的bug。解决方案严格定义Context的数据结构避免随意添加字段。如果Middleware需要产生只供自己或特定下游Middleware使用的数据应将其放在context.metadata[namespace]这样的隔离区域。编写清晰的Middleware文档说明其读写哪些字段。错误处理中的错误在ErrorHandlingMiddleware中如果你捕获了异常并进行了降级处理务必确保这个降级处理本身不会抛出新的异常否则会导致程序崩溃。同时要小心避免“吞噬”掉本应向上抛出的关键异常。解决方案在Middleware内部使用细致的try...except只处理你预期中可恢复的异常如网络超时、特定API错误。对于未知异常最好记录日志后重新抛出raise或者将其包装成一个特定的“中间件处理失败”异常。为降级逻辑编写单元测试。过度设计Middleware是强大的工具但不应滥用。如果某个逻辑只服务于一个特定的Agent或工具将其硬编码在该工具函数内部可能更简单、更清晰。Middleware最适合用于横切关注点cross-cutting concerns——那些影响系统中多个部分且逻辑相似的功能。解决方案在决定引入一个新的Middleware前问自己三个问题1) 这个功能会在两个以上的Agent或工具中使用吗2) 这个功能与核心业务逻辑是正交的吗3) 用Middleware实现会比直接写在业务代码里更易于管理和测试吗如果答案都是“是”那么Middleware是一个好选择。Middleware是将你的Agent从脆弱的脚本升级为健壮系统的关键构件。它通过关注点分离让核心业务逻辑保持简洁同时赋予系统强大的可扩展性、可观测性和可控性。开始在你的下一个Agent项目中尝试引入Middleware吧从最简单的日志中间件开始你会立刻感受到它对开发效率和系统稳定性的提升。