1. 项目概述为什么我们需要在Agent执行流中“插一脚”最近和几个做AI应用落地的朋友聊天大家不约而同地提到了同一个痛点好不容易基于LangChain、AutoGPT或者各类大模型API搭建了一个智能体Agent它能跑起来了能回答问题了甚至能调用工具了。但当我们想给它“加点料”的时候比如记录一下每次它思考的完整过程用于后续分析或者在它调用某个关键API前做一层权限校验又或者想在它最终回复给用户前统一给答案加上公司品牌标识——这时候就傻眼了。要么得去魔改框架的底层源码风险极高要么就得在业务代码里到处写重复的“补丁”既丑陋又难以维护。这其实就是“中间件”思想在Agent架构中的缺失。在传统的Web开发里中间件Middleware是个再熟悉不过的概念了它允许你在请求-响应的主流程中无侵入地插入预处理、后处理、日志、鉴权等逻辑。现在当我们的应用核心从处理HTTP请求变成了处理AI Agent的“思考-行动”循环时同样需要这样一套灵活、可插拔的拦截与增强机制。这就是“在Agent执行流中插入自定义逻辑”这个项目的核心价值为AI智能体的执行过程提供一个标准化、非侵入式的“钩子”系统让开发者能够像搭积木一样增强、监控、控制Agent的行为而不必担心破坏其核心逻辑的稳定性。无论是想实现全链路可观测性记录每一步的思考、工具调用和结果还是增强安全与合规对敏感工具调用进行审核对输出内容进行过滤或是提升业务灵活性动态修改提示词、对结果进行后处理这个中间件系统都能提供优雅的解决方案。它本质上是在Agent的“大脑”LLM和“手脚”工具/行动之间以及Agent与外界环境之间架起了一座可编程的桥梁。2. 核心设计思路如何为“思考流”安插观察哨与检查站设计一个Agent中间件系统不能简单地照搬Web中间件的洋葱模型。因为Agent的执行流是动态的、非线性的它可能包含多轮思考Chain of Thought、工具调用Tool Calling、甚至自我反思Self-Reflection等复杂环节。我们的设计必须贴合Agent特有的执行模式。2.1 理解Agent的标准执行流一个典型的、简化的Agent执行流可以抽象为以下几个核心阶段接收输入/问题用户提出问题或指令。规划与思考Agent在LLM驱动下分析问题决定下一步是直接回答还是需要调用工具。执行动作如果决定调用工具则执行对应的工具函数并获取结果。观察与迭代Agent观察工具执行结果结合历史决定是否继续思考、调用新工具还是可以给出最终答案。生成最终输出将最终答案返回给用户。我们的中间件系统就需要能够介入这个流程的关键生命周期节点。2.2 定义中间件的核心生命周期钩子基于上述流程我们可以定义出一套通用的中间件钩子Hooks。每个钩子都在特定时刻被触发并接收当前执行上下文。on_agent_start: 在Agent开始处理一个新输入时触发。这是注入初始上下文、进行权限校验或初始化追踪的绝佳位置。before_llm_call: 在Agent调用大模型进行“思考”之前触发。你可以在这里修改即将发送给LLM的提示词Prompt或者添加一些系统级的指令。after_llm_call: 在收到LLM的回复后触发。这里可以解析和记录LLM的原始输出比如它的思考过程如果启用了CoT和它建议的工具调用。before_tool_execution: 在Agent决定调用某个工具并即将执行该工具函数前触发。这是进行安全检查、参数校验、流量控制或模拟调用Mock的关键节点。after_tool_execution: 在工具执行完成后触发。这里可以处理工具返回的结果例如格式化数据、检查错误、或将结果进行脱敏处理。on_agent_end: 在Agent完成所有工作准备返回最终结果给用户时触发。适合进行最终的结果包装、日志记录、或触发后续业务逻辑如发送通知。2.3 中间件的执行模型责任链模式多个中间件如何组织最经典的模式是责任链Chain of Responsibility。中间件被组织成一个有序的管道Pipeline。当Agent执行到某个生命周期节点时请求上下文数据会依次流过注册在该节点上的所有中间件。每个中间件在管道中都有机会做三件事预处理Pre-process在核心逻辑执行前进行操作如修改输入、进行校验。传递控制权调用next()方法将请求传递给管道中的下一个中间件并最终触达Agent的核心逻辑。后处理Post-process在核心逻辑执行后对结果进行操作如修改输出、记录日志。这种模式的好处是解耦和灵活。你可以写一个只关心日志的中间件一个只关心安全的中间件然后像拼装乐高一样将它们组合起来。注意这里有一个关键决策点——中间件是同步执行还是支持异步考虑到LLM调用和工具调用往往是I/O密集型操作强烈建议从设计之初就支持异步async/await以避免阻塞整个事件循环这对于高并发场景至关重要。3. 技术实现详解从概念到可运行的代码理论说完了我们来点实际的。我将以一个基于Python、面向类似LangChain Agent的中间件系统为例展示其核心实现。我们会自底向上构建。3.1 定义核心数据模型上下文Context首先我们需要一个贯穿整个执行流程的上下文对象它携带了所有必要的信息并能在中间件链中传递和修改。from dataclasses import dataclass, field from typing import Any, Dict, Optional, List import uuid from datetime import datetime dataclass class AgentContext: Agent执行上下文贯穿整个中间件链。 # 会话与请求标识 session_id: str field(default_factorylambda: str(uuid.uuid4())) request_id: str field(default_factorylambda: str(uuid.uuid4())) user_input: str # 原始用户输入 # LLM相关 llm_messages: List[Dict] field(default_factorylist) # 发送给LLM的消息历史 llm_response_raw: Optional[str] None # LLM的原始回复 llm_parsed_action: Optional[Dict] None # 解析出的动作如 {tool_name: search, args: {...}} # 工具执行相关 current_tool_name: Optional[str] None tool_call_args: Optional[Dict] None tool_execution_result: Optional[Any] None tool_execution_error: Optional[Exception] None # Agent状态与结果 agent_thoughts: List[str] field(default_factorylist) # 记录Agent的思考步骤 final_output: Optional[str] None is_finished: bool False # 元数据与自定义存储供中间件使用 metadata: Dict[str, Any] field(default_factorydict) created_at: datetime field(default_factorydatetime.now) def to_dict(self) - Dict: 方便序列化记录 return { session_id: self.session_id, request_id: self.request_id, user_input: self.user_input, final_output: self.final_output, metadata: self.metadata, # ... 其他需要记录的字段 }这个AgentContext对象就像一份“病历”记录了Agent从接诊接收输入到开出药方输出结果的完整过程。每个中间件都可以读取和修改它需谨慎。3.2 构建中间件基类与生命周期管理器接下来定义中间件的抽象基类和负责管理生命周期钩子的核心类。from abc import ABC, abstractmethod from typing import Callable, Awaitable, List import inspect class AgentMiddleware(ABC): 中间件抽象基类。所有自定义中间件需继承此类。 async def on_agent_start(self, context: AgentContext): Agent开始处理时调用。 pass async def before_llm_call(self, context: AgentContext): 调用LLM前调用。 pass async def after_llm_call(self, context: AgentContext): 调用LLM后调用。 pass async def before_tool_execution(self, context: AgentContext): 执行工具前调用。 pass async def after_tool_execution(self, context: AgentContext): 执行工具后调用。 pass async def on_agent_end(self, context: AgentContext): Agent结束时调用。 pass # 一个通用的异步调用方法简化钩子调用 async def _invoke_hook(self, hook_name: str, context: AgentContext): method getattr(self, hook_name, None) if method and inspect.iscoroutinefunction(method): await method(context) class MiddlewareManager: 中间件管理器负责注册中间件并按顺序执行钩子。 def __init__(self): self._middlewares: List[AgentMiddleware] [] # 定义所有可用的钩子名称 self._hook_names [ on_agent_start, before_llm_call, after_llm_call, before_tool_execution, after_tool_execution, on_agent_end, ] def register(self, middleware: AgentMiddleware): 注册一个中间件实例。 self._middlewares.append(middleware) return self # 支持链式调用 async def invoke_hook(self, hook_name: str, context: AgentContext): 依次调用所有中间件的指定钩子。 if hook_name not in self._hook_names: raise ValueError(f未知的钩子名称: {hook_name}) for middleware in self._middlewares: await middleware._invoke_hook(hook_name, context)3.3 实现一个具体的Agent执行引擎现在我们创建一个简单的Agent执行引擎它会集成中间件管理器并在关键节点触发钩子。class SimpleAgentEngine: 一个集成了中间件系统的简易Agent执行引擎。 假设它使用一个LLM客户端和一个工具集。 def __init__(self, llm_client, tools: Dict[str, Callable], middleware_manager: MiddlewareManager): self.llm llm_client self.tools tools self.middleware middleware_manager async def run(self, user_input: str) - str: 主执行方法。 # 1. 初始化上下文 context AgentContext(user_inputuser_input) # 2. 触发 agent_start 钩子 await self.middleware.invoke_hook(on_agent_start, context) max_iterations 5 for i in range(max_iterations): # 3. 准备LLM调用消息这里简化了实际可能包含历史对话 context.llm_messages [{role: user, content: user_input}] # 4. 触发 before_llm_call 钩子中间件可以修改 messages await self.middleware.invoke_hook(before_llm_call, context) # 5. 调用LLM llm_response await self.llm.acomplete(context.llm_messages) context.llm_response_raw llm_response # 6. 触发 after_llm_call 钩子 await self.middleware.invoke_hook(after_llm_call, context) # 7. 解析LLM响应判断是直接回答还是调用工具这里是非常简化的解析 parsed_action self._parse_llm_response(llm_response) context.llm_parsed_action parsed_action if parsed_action.get(action) final_answer: context.final_output parsed_action.get(answer) break elif parsed_action.get(action) use_tool: tool_name parsed_action[tool_name] tool_args parsed_action[tool_args] if tool_name not in self.tools: context.final_output f错误未知工具 {tool_name} break context.current_tool_name tool_name context.tool_call_args tool_args # 8. 触发 before_tool_execution 钩子中间件可以阻止或修改参数 await self.middleware.invoke_hook(before_tool_execution, context) # 注意中间件可以通过设置 context 中的某个标志来跳过工具执行 # 9. 执行工具 try: tool_func self.tools[tool_name] # 注意这里实际执行时应该使用中间件处理后的参数 result await tool_func(**context.tool_call_args) if inspect.iscoroutinefunction(tool_func) else tool_func(**context.tool_call_args) context.tool_execution_result result except Exception as e: context.tool_execution_error e result f工具执行出错: {e} # 10. 触发 after_tool_execution 钩子中间件可以处理或修改结果 await self.middleware.invoke_hook(after_tool_execution, context) # 将工具结果作为下一轮LLM对话的上下文简化处理 user_input f上次你使用了工具{tool_name}结果是{result}。请继续分析。 else: context.final_output 无法解析LLM的指令。 break # 11. 标记结束并触发最终钩子 context.is_finished True await self.middleware.invoke_hook(on_agent_end, context) return context.final_output or Agent未产生有效输出。 def _parse_llm_response(self, response: str) - Dict: 一个极其简化的响应解析器。实际项目中应使用更可靠的方法如JSON模式调用。 # 这里只是一个示例实际应解析LLM返回的结构化数据 if 调用搜索工具 in response: return {action: use_tool, tool_name: search_web, tool_args: {query: 示例查询}} elif 最终答案是 in response: return {action: final_answer, answer: response.split(最终答案是)[-1].strip()} else: return {action: final_answer, answer: response}这个引擎清晰地展示了中间件钩子是如何被嵌入到Agent的核心执行循环中的。每个钩子都位于一个关键决策点或状态变更点。3.4 编写实战中间件示例有了引擎我们来编写几个有实际用途的中间件。示例1日志记录中间件这个中间件负责将Agent执行的每一步都记录下来便于调试和审计。import json import logging class LoggingMiddleware(AgentMiddleware): 全链路日志记录中间件。 def __init__(self, logger_nameagent_middleware): self.logger logging.getLogger(logger_name) async def on_agent_start(self, context: AgentContext): self.logger.info(f[{context.request_id}] Agent开始处理。输入: {context.user_input}) context.metadata[start_time] datetime.now().isoformat() async def before_llm_call(self, context: AgentContext): self.logger.debug(f[{context.request_id}] 准备调用LLM。消息数: {len(context.llm_messages)}) async def after_llm_call(self, context: AgentContext): self.logger.info(f[{context.request_id}] LLM调用完成。原始响应: {context.llm_response_raw[:200]}...) # 截断长文本 async def before_tool_execution(self, context: AgentContext): self.logger.warning(f[{context.request_id}] 即将执行工具: {context.current_tool_name}, 参数: {json.dumps(context.tool_call_args, ensure_asciiFalse)}) async def after_tool_execution(self, context: AgentContext): status 成功 if context.tool_execution_error is None else f失败({context.tool_execution_error}) result_preview str(context.tool_execution_result)[:100] if context.tool_execution_result else None self.logger.info(f[{context.request_id}] 工具执行{status}。结果预览: {result_preview}...) async def on_agent_end(self, context: AgentContext): duration (datetime.now() - context.created_at).total_seconds() self.logger.info(f[{context.request_id}] Agent执行结束。耗时: {duration:.2f}s, 输出: {context.final_output}) # 可以将完整的context.to_dict()发送到日志系统或数据库示例2安全检查中间件这个中间件在工具执行前进行拦截检查是否有敏感操作或未经授权的调用。class SecurityCheckMiddleware(AgentMiddleware): 工具调用安全检查中间件。 def __init__(self, forbidden_toolsNone, allowed_tool_patternsNone): self.forbidden_tools set(forbidden_tools or []) self.allowed_patterns allowed_tool_patterns or [] # 可以用正则表达式定义允许的模式 async def before_tool_execution(self, context: AgentContext): tool_name context.current_tool_name # 检查1是否在禁用名单 if tool_name in self.forbidden_tools: self._block_execution(context, f工具 {tool_name} 被策略禁止执行。) return # 检查2参数中是否包含敏感关键词示例禁止执行删除操作 args_str json.dumps(context.tool_call_args or {}) sensitive_keywords [delete, drop, rm -rf, format] for kw in sensitive_keywords: if kw in args_str.lower(): self._block_execution(context, f工具调用参数包含敏感关键词 {kw}执行被阻止。) return # 检查3可以在这里集成更复杂的权限系统比如检查API Key、用户角色等 # if not self._check_user_permission(context.session_id, tool_name): # self._block_execution(context, 权限不足。) # return # 如果所有检查通过可以继续。也可以在这里对参数进行清洗或转换。 # context.tool_call_args[query] self._sanitize_input(context.tool_call_args.get(query, )) def _block_execution(self, context: AgentContext, reason: str): 阻止工具执行并设置错误信息和最终输出。 context.tool_execution_error Exception(reason) # 直接设置最终输出跳过实际工具调用让Agent进入结束流程 context.final_output f安全策略拦截{reason} context.is_finished True # 这是一个技巧告诉引擎提前结束 # 注意更优雅的方式是在context中设置一个 should_skip_tool 标志由引擎判断。示例3性能监控与缓存中间件这个中间件监控LLM调用耗时并尝试对重复的LLM查询进行缓存。import asyncio from functools import lru_cache class MonitoringAndCacheMiddleware(AgentMiddleware): 性能监控与简单缓存中间件。 def __init__(self): self.llm_call_durations [] # 使用一个简单的内存缓存生产环境应用Redis等 self._cache {} async def before_llm_call(self, context: AgentContext): # 生成一个缓存键例如对消息内容进行哈希 cache_key self._generate_cache_key(context.llm_messages) context.metadata[llm_cache_key] cache_key # 检查缓存 if cache_key in self._cache: context.llm_response_raw self._cache[cache_key] context.metadata[cache_hit] True # 我们需要一个方式来告诉引擎“跳过真正的LLM调用” # 这里我们在metadata里设置一个标志引擎需要检查这个标志 context.metadata[skip_llm_call] True else: context.metadata[cache_hit] False context.metadata[skip_llm_call] False context.metadata[llm_call_start] asyncio.get_event_loop().time() async def after_llm_call(self, context: AgentContext): if context.metadata.get(cache_hit): self.logger.debug(f缓存命中节省了一次LLM调用。) return if llm_call_start in context.metadata: duration asyncio.get_event_loop().time() - context.metadata[llm_call_start] self.llm_call_durations.append(duration) self.logger.info(fLLM调用耗时: {duration:.3f}秒) # 缓存结果注意只缓存成功的、非流式的响应 cache_key context.metadata.get(llm_cache_key) if cache_key and context.llm_response_raw: self._cache[cache_key] context.llm_response_raw # 可以设置TTL或最大缓存大小 def _generate_cache_key(self, messages): 生成缓存键的简单示例。实际应用需要更健壮的序列化和哈希。 import hashlib content json.dumps(messages, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest()实操心得在实现缓存中间件时最大的坑在于如何让引擎知道“跳过本次LLM调用”。上面的示例通过在context.metadata里设置标志位来实现这是一种简单有效的方式。但更优雅的设计是让中间件拥有“短路”整个管道的能力即某个中间件可以直接设置context.final_output并终止后续中间件和核心逻辑。这需要在MiddlewareManager.invoke_hook方法中增加对上下文状态的检查例如检查context.is_finished或context.should_skip如果为真则中断钩子链的传递。这体现了中间件系统的控制力。4. 高级特性与生产级考量一个基础的中间件系统跑起来后我们还需要考虑更多生产环境中会遇到的问题。4.1 中间件的顺序依赖与配置化中间件的执行顺序很重要。例如你肯定希望日志中间件最先执行on_agent_start和最后执行on_agent_end以确保记录下最完整的上下文。而安全中间件必须在工具执行前before_tool_execution生效并且其阻断逻辑的优先级应该很高。因此我们的MiddlewareManager需要支持对中间件进行排序或者允许通过配置文件如YAML来声明中间件栈的顺序。# 示例通过优先级定义顺序 class OrderedMiddlewareManager(MiddlewareManager): def __init__(self): super().__init__() # 使用 (priority, middleware) 列表 priority 数字越小越先执行 self._ordered_middlewares [] def register(self, middleware: AgentMiddleware, priority: int 10): self._ordered_middlewares.append((priority, middleware)) self._ordered_middlewares.sort(keylambda x: x[0]) self._middlewares [mw for _, mw in self._ordered_middlewares] return self # 使用 manager OrderedMiddlewareManager() manager.register(LoggingMiddleware(), priority1) # 最先注册最先执行 manager.register(SecurityCheckMiddleware(forbidden_tools[rm]), priority5) manager.register(MonitoringAndCacheMiddleware(), priority10)4.2 错误处理与中间件容错中间件本身也是代码也可能出错。一个中间件的崩溃不应该导致整个Agent服务不可用。我们需要在MiddlewareManager.invoke_hook中为每个中间件的调用添加try-catch。async def invoke_hook_safe(self, hook_name: str, context: AgentContext): if hook_name not in self._hook_names: raise ValueError(f未知的钩子名称: {hook_name}) for middleware in self._middlewares: try: await middleware._invoke_hook(hook_name, context) except Exception as e: # 记录错误但不要阻断其他中间件 logging.error(f中间件 {middleware.__class__.__name__} 在执行钩子 {hook_name} 时出错: {e}, exc_infoTrue) # 可以选择将错误信息存入上下文供后续中间件或引擎处理 context.metadata.setdefault(middleware_errors, []).append({ middleware: middleware.__class__.__name__, hook: hook_name, error: str(e) })4.3 上下文Context的线程/协程安全与深度拷贝在异步并发环境下同一个Agent的多个执行流例如同时处理多个用户请求可能会共享某些资源或者中间件可能会修改传入的上下文对象。如果AgentContext实例在中间件链中被不适当地共享和修改会导致数据污染。解决方案每个请求独立的上下文确保AgentEngine.run方法每次被调用时都创建一个全新的AgentContext实例。谨慎处理引用传递如果某个中间件需要修改上下文中的复杂对象如llm_messages列表最好进行深拷贝copy.deepcopy尤其是在有缓存功能的中间件里。但深拷贝有性能开销需要权衡。使用不可变数据结构考虑将上下文的核心数据设计为不可变immutable任何修改都返回一个新的上下文实例。这类似于React的状态管理能从根本上避免副作用但实现复杂度较高。4.4 与现有框架如LangChain集成我们上面的实现是原生的。在实际项目中你很可能是在LangChain、LlamaIndex等现有框架上工作。这些框架通常已经提供了某种回调Callback机制。我们的中间件系统可以构建在这些回调机制之上或者将其包装成更统一、强大的形式。以LangChain为例它有一个丰富的BaseCallbackHandler系统。我们可以创建一个MiddlewareCallbackHandler将LangChain的各种回调事件on_llm_start,on_tool_start等映射到我们自定义的中间件钩子上。from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import AgentAction, AgentFinish class LangChainMiddlewareAdapter(BaseCallbackHandler): 将LangChain回调适配到我们的中间件系统。 def __init__(self, middleware_manager: MiddlewareManager, agent_context: AgentContext): self.manager middleware_manager self.context agent_context async def on_chain_start(self, serialized, inputs, **kwargs): # 映射到 on_agent_start await self.manager.invoke_hook(on_agent_start, self.context) async def on_llm_start(self, serialized, prompts, **kwargs): # 映射到 before_llm_call self.context.llm_messages prompts # 可能需要转换格式 await self.manager.invoke_hook(before_llm_call, self.context) async def on_llm_end(self, response, **kwargs): self.context.llm_response_raw response.generations[0][0].text await self.manager.invoke_hook(after_llm_call, self.context) async def on_tool_start(self, serialized, input_str, **kwargs): # 解析出工具名和参数这里简化 self.context.current_tool_name serialized.get(name) self.context.tool_call_args {input: input_str} await self.manager.invoke_hook(before_tool_execution, self.context) # 如果安全中间件阻止了执行这里需要能中断LangChain的流程 # 这可能需要更高级的集成比如修改Agent的执行器。 async def on_tool_end(self, output, **kwargs): self.context.tool_execution_result output await self.manager.invoke_hook(after_tool_execution, self.context) async def on_agent_finish(self, finish: AgentFinish, **kwargs): self.context.final_output finish.return_values.get(output) self.context.is_finished True await self.manager.invoke_hook(on_agent_end, self.context)这样你就可以在LangChain项目中使用同一套中间件系统了。关键在于处理好框架原生回调系统与我们自定义钩子之间的数据转换和流程控制。5. 常见问题排查与实战技巧在实际开发和部署中你肯定会遇到各种问题。下面是一些典型场景和解决思路。5.1 中间件执行顺序混乱或未生效症状日志中间件没记录到工具调用的信息或者安全校验在工具执行后才触发。排查检查注册顺序确认中间件在MiddlewareManager中的注册顺序是否符合预期。使用上面提到的优先级机制。检查钩子绑定确保你的自定义中间件正确重写了目标钩子方法如before_tool_execution。拼写错误或使用了同步方法未加async都会导致钩子不被调用。添加调试中间件写一个最简单的调试中间件在每个钩子里打印一行日志确认引擎是否按预期触发了所有钩子。5.2 中间件修改了上下文但引擎未使用症状在before_llm_call中修改了context.llm_messages但实际发送给LLM的仍是旧消息。排查确认引擎读取点检查你的AgentEngine代码是在调用before_llm_call钩子之前还是之后读取llm_messages并发送给LLM的必须是之后。检查数据流确保上下文对象在钩子链中是按引用传递的同一个对象而不是被复制了。查看引擎逻辑引擎是否在调用钩子后重新从上下文中获取了最新数据例如# 正确做法钩子执行后使用可能被修改过的上下文数据 await self.middleware.invoke_hook(before_llm_call, context) llm_response await self.llm.acomplete(context.llm_messages) # 使用context.llm_messages5.3 异步中间件导致性能问题或死锁症状系统在高并发下变慢或者偶尔卡住无响应。排查避免阻塞操作确保中间件内的所有I/O操作网络请求、数据库查询、文件读写都是异步的。如果一个同步阻塞操作如requests.get在异步中间件中被调用它会阻塞整个事件循环。使用异步库将requests替换为aiohttp或httpx将同步数据库驱动替换为异步驱动如asyncpg用于PostgreSQL。设置超时为中间件中可能长时间运行的操作如外部API调用添加超时控制使用asyncio.wait_for。async def before_tool_execution(self, context): try: # 假设有一个异步的远程权限校验 async with aiohttp.ClientSession() as session: async with session.post(https://auth-service/check, json..., timeout5.0) as resp: # ... except asyncio.TimeoutError: context.tool_execution_error Exception(权限校验超时) context.final_output 系统繁忙请稍后再试。5.4 如何测试中间件中间件的测试策略与普通业务逻辑不同因为它高度依赖于执行上下文和生命周期。单元测试单个中间件Mock一个AgentContext对象直接调用中间件的钩子方法断言其对上下文的修改或产生的副作用如日志记录是否符合预期。pytest.mark.asyncio async def test_security_middleware_blocks_forbidden_tool(): middleware SecurityCheckMiddleware(forbidden_tools[delete_database]) context AgentContext(current_tool_namedelete_database, tool_call_args{}) await middleware.before_tool_execution(context) assert context.final_output is not None assert 禁止 in context.final_output # 或者检查 context.tool_execution_error集成测试中间件链创建一个最小化的MockAgentEngine注册需要测试的中间件模拟完整的执行流程检查最终的上下文状态和输出。端到端测试将中间件系统与真实的LLM和工具集成进行完整的场景测试确保中间件的加入不会破坏Agent原有的核心功能。5.5 中间件系统的扩展性思考当中间件越来越多时管理起来会变得复杂。可以考虑以下方向按功能模块化将中间件分组例如“可观测性模块”包含日志、指标、追踪中间件、“安全模块”、“业务增强模块”。每个模块可以独立配置和启用。动态加载与热重载支持从配置文件或数据库中读取中间件配置在不重启服务的情况下动态添加、移除或更新中间件。这对于SaaS平台或需要频繁调整策略的场景非常有用。中间件市场/仓库如果中间件系统设计得足够通用可以抽象出标准的接口和配置范式鼓励团队内部甚至社区贡献可复用的中间件形成一个生态。6. 总结与个人体会构建一个Agent中间件系统本质上是在为AI应用的“操作系统”增加一套强大的“系统调用”和“驱动接口”。它把那些原本散落在业务代码各处的横切关注点Cross-cutting Concerns——日志、安全、监控、缓存、限流、数据脱敏等——集中管理起来让Agent的核心逻辑保持干净和专注。从我自己的实践来看这套系统的价值在项目规模扩大后体现得尤为明显。当你有十几个不同的Agent每个Agent又调用着数十个工具时如果没有中间件想要统一加上调用链追踪Tracing功能那将是一场灾难。而有了中间件你只需要编写并注册一个TracingMiddleware所有Agent瞬间就具备了全链路追踪能力。最后分享一个关键的取舍经验中间件的强大也带来了复杂性。不是所有逻辑都适合放进中间件。我的原则是只有那些与核心业务逻辑正交的、全局性的、可插拔的功能才适合用中间件实现。如果一个逻辑只针对某一个特定的Agent或工具并且紧密耦合业务那么把它放在Agent或工具本身的实现里会更清晰。过度设计中间件会让系统变得难以理解和调试。从最简单的日志中间件开始当同一个模式出现第三次时再考虑将其抽象成通用中间件这是一个比较稳妥的演进路径。 AI Agent中间件系统设计：为智能体执行流添加可观测与安全控制
1. 项目概述为什么我们需要在Agent执行流中“插一脚”最近和几个做AI应用落地的朋友聊天大家不约而同地提到了同一个痛点好不容易基于LangChain、AutoGPT或者各类大模型API搭建了一个智能体Agent它能跑起来了能回答问题了甚至能调用工具了。但当我们想给它“加点料”的时候比如记录一下每次它思考的完整过程用于后续分析或者在它调用某个关键API前做一层权限校验又或者想在它最终回复给用户前统一给答案加上公司品牌标识——这时候就傻眼了。要么得去魔改框架的底层源码风险极高要么就得在业务代码里到处写重复的“补丁”既丑陋又难以维护。这其实就是“中间件”思想在Agent架构中的缺失。在传统的Web开发里中间件Middleware是个再熟悉不过的概念了它允许你在请求-响应的主流程中无侵入地插入预处理、后处理、日志、鉴权等逻辑。现在当我们的应用核心从处理HTTP请求变成了处理AI Agent的“思考-行动”循环时同样需要这样一套灵活、可插拔的拦截与增强机制。这就是“在Agent执行流中插入自定义逻辑”这个项目的核心价值为AI智能体的执行过程提供一个标准化、非侵入式的“钩子”系统让开发者能够像搭积木一样增强、监控、控制Agent的行为而不必担心破坏其核心逻辑的稳定性。无论是想实现全链路可观测性记录每一步的思考、工具调用和结果还是增强安全与合规对敏感工具调用进行审核对输出内容进行过滤或是提升业务灵活性动态修改提示词、对结果进行后处理这个中间件系统都能提供优雅的解决方案。它本质上是在Agent的“大脑”LLM和“手脚”工具/行动之间以及Agent与外界环境之间架起了一座可编程的桥梁。2. 核心设计思路如何为“思考流”安插观察哨与检查站设计一个Agent中间件系统不能简单地照搬Web中间件的洋葱模型。因为Agent的执行流是动态的、非线性的它可能包含多轮思考Chain of Thought、工具调用Tool Calling、甚至自我反思Self-Reflection等复杂环节。我们的设计必须贴合Agent特有的执行模式。2.1 理解Agent的标准执行流一个典型的、简化的Agent执行流可以抽象为以下几个核心阶段接收输入/问题用户提出问题或指令。规划与思考Agent在LLM驱动下分析问题决定下一步是直接回答还是需要调用工具。执行动作如果决定调用工具则执行对应的工具函数并获取结果。观察与迭代Agent观察工具执行结果结合历史决定是否继续思考、调用新工具还是可以给出最终答案。生成最终输出将最终答案返回给用户。我们的中间件系统就需要能够介入这个流程的关键生命周期节点。2.2 定义中间件的核心生命周期钩子基于上述流程我们可以定义出一套通用的中间件钩子Hooks。每个钩子都在特定时刻被触发并接收当前执行上下文。on_agent_start: 在Agent开始处理一个新输入时触发。这是注入初始上下文、进行权限校验或初始化追踪的绝佳位置。before_llm_call: 在Agent调用大模型进行“思考”之前触发。你可以在这里修改即将发送给LLM的提示词Prompt或者添加一些系统级的指令。after_llm_call: 在收到LLM的回复后触发。这里可以解析和记录LLM的原始输出比如它的思考过程如果启用了CoT和它建议的工具调用。before_tool_execution: 在Agent决定调用某个工具并即将执行该工具函数前触发。这是进行安全检查、参数校验、流量控制或模拟调用Mock的关键节点。after_tool_execution: 在工具执行完成后触发。这里可以处理工具返回的结果例如格式化数据、检查错误、或将结果进行脱敏处理。on_agent_end: 在Agent完成所有工作准备返回最终结果给用户时触发。适合进行最终的结果包装、日志记录、或触发后续业务逻辑如发送通知。2.3 中间件的执行模型责任链模式多个中间件如何组织最经典的模式是责任链Chain of Responsibility。中间件被组织成一个有序的管道Pipeline。当Agent执行到某个生命周期节点时请求上下文数据会依次流过注册在该节点上的所有中间件。每个中间件在管道中都有机会做三件事预处理Pre-process在核心逻辑执行前进行操作如修改输入、进行校验。传递控制权调用next()方法将请求传递给管道中的下一个中间件并最终触达Agent的核心逻辑。后处理Post-process在核心逻辑执行后对结果进行操作如修改输出、记录日志。这种模式的好处是解耦和灵活。你可以写一个只关心日志的中间件一个只关心安全的中间件然后像拼装乐高一样将它们组合起来。注意这里有一个关键决策点——中间件是同步执行还是支持异步考虑到LLM调用和工具调用往往是I/O密集型操作强烈建议从设计之初就支持异步async/await以避免阻塞整个事件循环这对于高并发场景至关重要。3. 技术实现详解从概念到可运行的代码理论说完了我们来点实际的。我将以一个基于Python、面向类似LangChain Agent的中间件系统为例展示其核心实现。我们会自底向上构建。3.1 定义核心数据模型上下文Context首先我们需要一个贯穿整个执行流程的上下文对象它携带了所有必要的信息并能在中间件链中传递和修改。from dataclasses import dataclass, field from typing import Any, Dict, Optional, List import uuid from datetime import datetime dataclass class AgentContext: Agent执行上下文贯穿整个中间件链。 # 会话与请求标识 session_id: str field(default_factorylambda: str(uuid.uuid4())) request_id: str field(default_factorylambda: str(uuid.uuid4())) user_input: str # 原始用户输入 # LLM相关 llm_messages: List[Dict] field(default_factorylist) # 发送给LLM的消息历史 llm_response_raw: Optional[str] None # LLM的原始回复 llm_parsed_action: Optional[Dict] None # 解析出的动作如 {tool_name: search, args: {...}} # 工具执行相关 current_tool_name: Optional[str] None tool_call_args: Optional[Dict] None tool_execution_result: Optional[Any] None tool_execution_error: Optional[Exception] None # Agent状态与结果 agent_thoughts: List[str] field(default_factorylist) # 记录Agent的思考步骤 final_output: Optional[str] None is_finished: bool False # 元数据与自定义存储供中间件使用 metadata: Dict[str, Any] field(default_factorydict) created_at: datetime field(default_factorydatetime.now) def to_dict(self) - Dict: 方便序列化记录 return { session_id: self.session_id, request_id: self.request_id, user_input: self.user_input, final_output: self.final_output, metadata: self.metadata, # ... 其他需要记录的字段 }这个AgentContext对象就像一份“病历”记录了Agent从接诊接收输入到开出药方输出结果的完整过程。每个中间件都可以读取和修改它需谨慎。3.2 构建中间件基类与生命周期管理器接下来定义中间件的抽象基类和负责管理生命周期钩子的核心类。from abc import ABC, abstractmethod from typing import Callable, Awaitable, List import inspect class AgentMiddleware(ABC): 中间件抽象基类。所有自定义中间件需继承此类。 async def on_agent_start(self, context: AgentContext): Agent开始处理时调用。 pass async def before_llm_call(self, context: AgentContext): 调用LLM前调用。 pass async def after_llm_call(self, context: AgentContext): 调用LLM后调用。 pass async def before_tool_execution(self, context: AgentContext): 执行工具前调用。 pass async def after_tool_execution(self, context: AgentContext): 执行工具后调用。 pass async def on_agent_end(self, context: AgentContext): Agent结束时调用。 pass # 一个通用的异步调用方法简化钩子调用 async def _invoke_hook(self, hook_name: str, context: AgentContext): method getattr(self, hook_name, None) if method and inspect.iscoroutinefunction(method): await method(context) class MiddlewareManager: 中间件管理器负责注册中间件并按顺序执行钩子。 def __init__(self): self._middlewares: List[AgentMiddleware] [] # 定义所有可用的钩子名称 self._hook_names [ on_agent_start, before_llm_call, after_llm_call, before_tool_execution, after_tool_execution, on_agent_end, ] def register(self, middleware: AgentMiddleware): 注册一个中间件实例。 self._middlewares.append(middleware) return self # 支持链式调用 async def invoke_hook(self, hook_name: str, context: AgentContext): 依次调用所有中间件的指定钩子。 if hook_name not in self._hook_names: raise ValueError(f未知的钩子名称: {hook_name}) for middleware in self._middlewares: await middleware._invoke_hook(hook_name, context)3.3 实现一个具体的Agent执行引擎现在我们创建一个简单的Agent执行引擎它会集成中间件管理器并在关键节点触发钩子。class SimpleAgentEngine: 一个集成了中间件系统的简易Agent执行引擎。 假设它使用一个LLM客户端和一个工具集。 def __init__(self, llm_client, tools: Dict[str, Callable], middleware_manager: MiddlewareManager): self.llm llm_client self.tools tools self.middleware middleware_manager async def run(self, user_input: str) - str: 主执行方法。 # 1. 初始化上下文 context AgentContext(user_inputuser_input) # 2. 触发 agent_start 钩子 await self.middleware.invoke_hook(on_agent_start, context) max_iterations 5 for i in range(max_iterations): # 3. 准备LLM调用消息这里简化了实际可能包含历史对话 context.llm_messages [{role: user, content: user_input}] # 4. 触发 before_llm_call 钩子中间件可以修改 messages await self.middleware.invoke_hook(before_llm_call, context) # 5. 调用LLM llm_response await self.llm.acomplete(context.llm_messages) context.llm_response_raw llm_response # 6. 触发 after_llm_call 钩子 await self.middleware.invoke_hook(after_llm_call, context) # 7. 解析LLM响应判断是直接回答还是调用工具这里是非常简化的解析 parsed_action self._parse_llm_response(llm_response) context.llm_parsed_action parsed_action if parsed_action.get(action) final_answer: context.final_output parsed_action.get(answer) break elif parsed_action.get(action) use_tool: tool_name parsed_action[tool_name] tool_args parsed_action[tool_args] if tool_name not in self.tools: context.final_output f错误未知工具 {tool_name} break context.current_tool_name tool_name context.tool_call_args tool_args # 8. 触发 before_tool_execution 钩子中间件可以阻止或修改参数 await self.middleware.invoke_hook(before_tool_execution, context) # 注意中间件可以通过设置 context 中的某个标志来跳过工具执行 # 9. 执行工具 try: tool_func self.tools[tool_name] # 注意这里实际执行时应该使用中间件处理后的参数 result await tool_func(**context.tool_call_args) if inspect.iscoroutinefunction(tool_func) else tool_func(**context.tool_call_args) context.tool_execution_result result except Exception as e: context.tool_execution_error e result f工具执行出错: {e} # 10. 触发 after_tool_execution 钩子中间件可以处理或修改结果 await self.middleware.invoke_hook(after_tool_execution, context) # 将工具结果作为下一轮LLM对话的上下文简化处理 user_input f上次你使用了工具{tool_name}结果是{result}。请继续分析。 else: context.final_output 无法解析LLM的指令。 break # 11. 标记结束并触发最终钩子 context.is_finished True await self.middleware.invoke_hook(on_agent_end, context) return context.final_output or Agent未产生有效输出。 def _parse_llm_response(self, response: str) - Dict: 一个极其简化的响应解析器。实际项目中应使用更可靠的方法如JSON模式调用。 # 这里只是一个示例实际应解析LLM返回的结构化数据 if 调用搜索工具 in response: return {action: use_tool, tool_name: search_web, tool_args: {query: 示例查询}} elif 最终答案是 in response: return {action: final_answer, answer: response.split(最终答案是)[-1].strip()} else: return {action: final_answer, answer: response}这个引擎清晰地展示了中间件钩子是如何被嵌入到Agent的核心执行循环中的。每个钩子都位于一个关键决策点或状态变更点。3.4 编写实战中间件示例有了引擎我们来编写几个有实际用途的中间件。示例1日志记录中间件这个中间件负责将Agent执行的每一步都记录下来便于调试和审计。import json import logging class LoggingMiddleware(AgentMiddleware): 全链路日志记录中间件。 def __init__(self, logger_nameagent_middleware): self.logger logging.getLogger(logger_name) async def on_agent_start(self, context: AgentContext): self.logger.info(f[{context.request_id}] Agent开始处理。输入: {context.user_input}) context.metadata[start_time] datetime.now().isoformat() async def before_llm_call(self, context: AgentContext): self.logger.debug(f[{context.request_id}] 准备调用LLM。消息数: {len(context.llm_messages)}) async def after_llm_call(self, context: AgentContext): self.logger.info(f[{context.request_id}] LLM调用完成。原始响应: {context.llm_response_raw[:200]}...) # 截断长文本 async def before_tool_execution(self, context: AgentContext): self.logger.warning(f[{context.request_id}] 即将执行工具: {context.current_tool_name}, 参数: {json.dumps(context.tool_call_args, ensure_asciiFalse)}) async def after_tool_execution(self, context: AgentContext): status 成功 if context.tool_execution_error is None else f失败({context.tool_execution_error}) result_preview str(context.tool_execution_result)[:100] if context.tool_execution_result else None self.logger.info(f[{context.request_id}] 工具执行{status}。结果预览: {result_preview}...) async def on_agent_end(self, context: AgentContext): duration (datetime.now() - context.created_at).total_seconds() self.logger.info(f[{context.request_id}] Agent执行结束。耗时: {duration:.2f}s, 输出: {context.final_output}) # 可以将完整的context.to_dict()发送到日志系统或数据库示例2安全检查中间件这个中间件在工具执行前进行拦截检查是否有敏感操作或未经授权的调用。class SecurityCheckMiddleware(AgentMiddleware): 工具调用安全检查中间件。 def __init__(self, forbidden_toolsNone, allowed_tool_patternsNone): self.forbidden_tools set(forbidden_tools or []) self.allowed_patterns allowed_tool_patterns or [] # 可以用正则表达式定义允许的模式 async def before_tool_execution(self, context: AgentContext): tool_name context.current_tool_name # 检查1是否在禁用名单 if tool_name in self.forbidden_tools: self._block_execution(context, f工具 {tool_name} 被策略禁止执行。) return # 检查2参数中是否包含敏感关键词示例禁止执行删除操作 args_str json.dumps(context.tool_call_args or {}) sensitive_keywords [delete, drop, rm -rf, format] for kw in sensitive_keywords: if kw in args_str.lower(): self._block_execution(context, f工具调用参数包含敏感关键词 {kw}执行被阻止。) return # 检查3可以在这里集成更复杂的权限系统比如检查API Key、用户角色等 # if not self._check_user_permission(context.session_id, tool_name): # self._block_execution(context, 权限不足。) # return # 如果所有检查通过可以继续。也可以在这里对参数进行清洗或转换。 # context.tool_call_args[query] self._sanitize_input(context.tool_call_args.get(query, )) def _block_execution(self, context: AgentContext, reason: str): 阻止工具执行并设置错误信息和最终输出。 context.tool_execution_error Exception(reason) # 直接设置最终输出跳过实际工具调用让Agent进入结束流程 context.final_output f安全策略拦截{reason} context.is_finished True # 这是一个技巧告诉引擎提前结束 # 注意更优雅的方式是在context中设置一个 should_skip_tool 标志由引擎判断。示例3性能监控与缓存中间件这个中间件监控LLM调用耗时并尝试对重复的LLM查询进行缓存。import asyncio from functools import lru_cache class MonitoringAndCacheMiddleware(AgentMiddleware): 性能监控与简单缓存中间件。 def __init__(self): self.llm_call_durations [] # 使用一个简单的内存缓存生产环境应用Redis等 self._cache {} async def before_llm_call(self, context: AgentContext): # 生成一个缓存键例如对消息内容进行哈希 cache_key self._generate_cache_key(context.llm_messages) context.metadata[llm_cache_key] cache_key # 检查缓存 if cache_key in self._cache: context.llm_response_raw self._cache[cache_key] context.metadata[cache_hit] True # 我们需要一个方式来告诉引擎“跳过真正的LLM调用” # 这里我们在metadata里设置一个标志引擎需要检查这个标志 context.metadata[skip_llm_call] True else: context.metadata[cache_hit] False context.metadata[skip_llm_call] False context.metadata[llm_call_start] asyncio.get_event_loop().time() async def after_llm_call(self, context: AgentContext): if context.metadata.get(cache_hit): self.logger.debug(f缓存命中节省了一次LLM调用。) return if llm_call_start in context.metadata: duration asyncio.get_event_loop().time() - context.metadata[llm_call_start] self.llm_call_durations.append(duration) self.logger.info(fLLM调用耗时: {duration:.3f}秒) # 缓存结果注意只缓存成功的、非流式的响应 cache_key context.metadata.get(llm_cache_key) if cache_key and context.llm_response_raw: self._cache[cache_key] context.llm_response_raw # 可以设置TTL或最大缓存大小 def _generate_cache_key(self, messages): 生成缓存键的简单示例。实际应用需要更健壮的序列化和哈希。 import hashlib content json.dumps(messages, sort_keysTrue) return hashlib.md5(content.encode()).hexdigest()实操心得在实现缓存中间件时最大的坑在于如何让引擎知道“跳过本次LLM调用”。上面的示例通过在context.metadata里设置标志位来实现这是一种简单有效的方式。但更优雅的设计是让中间件拥有“短路”整个管道的能力即某个中间件可以直接设置context.final_output并终止后续中间件和核心逻辑。这需要在MiddlewareManager.invoke_hook方法中增加对上下文状态的检查例如检查context.is_finished或context.should_skip如果为真则中断钩子链的传递。这体现了中间件系统的控制力。4. 高级特性与生产级考量一个基础的中间件系统跑起来后我们还需要考虑更多生产环境中会遇到的问题。4.1 中间件的顺序依赖与配置化中间件的执行顺序很重要。例如你肯定希望日志中间件最先执行on_agent_start和最后执行on_agent_end以确保记录下最完整的上下文。而安全中间件必须在工具执行前before_tool_execution生效并且其阻断逻辑的优先级应该很高。因此我们的MiddlewareManager需要支持对中间件进行排序或者允许通过配置文件如YAML来声明中间件栈的顺序。# 示例通过优先级定义顺序 class OrderedMiddlewareManager(MiddlewareManager): def __init__(self): super().__init__() # 使用 (priority, middleware) 列表 priority 数字越小越先执行 self._ordered_middlewares [] def register(self, middleware: AgentMiddleware, priority: int 10): self._ordered_middlewares.append((priority, middleware)) self._ordered_middlewares.sort(keylambda x: x[0]) self._middlewares [mw for _, mw in self._ordered_middlewares] return self # 使用 manager OrderedMiddlewareManager() manager.register(LoggingMiddleware(), priority1) # 最先注册最先执行 manager.register(SecurityCheckMiddleware(forbidden_tools[rm]), priority5) manager.register(MonitoringAndCacheMiddleware(), priority10)4.2 错误处理与中间件容错中间件本身也是代码也可能出错。一个中间件的崩溃不应该导致整个Agent服务不可用。我们需要在MiddlewareManager.invoke_hook中为每个中间件的调用添加try-catch。async def invoke_hook_safe(self, hook_name: str, context: AgentContext): if hook_name not in self._hook_names: raise ValueError(f未知的钩子名称: {hook_name}) for middleware in self._middlewares: try: await middleware._invoke_hook(hook_name, context) except Exception as e: # 记录错误但不要阻断其他中间件 logging.error(f中间件 {middleware.__class__.__name__} 在执行钩子 {hook_name} 时出错: {e}, exc_infoTrue) # 可以选择将错误信息存入上下文供后续中间件或引擎处理 context.metadata.setdefault(middleware_errors, []).append({ middleware: middleware.__class__.__name__, hook: hook_name, error: str(e) })4.3 上下文Context的线程/协程安全与深度拷贝在异步并发环境下同一个Agent的多个执行流例如同时处理多个用户请求可能会共享某些资源或者中间件可能会修改传入的上下文对象。如果AgentContext实例在中间件链中被不适当地共享和修改会导致数据污染。解决方案每个请求独立的上下文确保AgentEngine.run方法每次被调用时都创建一个全新的AgentContext实例。谨慎处理引用传递如果某个中间件需要修改上下文中的复杂对象如llm_messages列表最好进行深拷贝copy.deepcopy尤其是在有缓存功能的中间件里。但深拷贝有性能开销需要权衡。使用不可变数据结构考虑将上下文的核心数据设计为不可变immutable任何修改都返回一个新的上下文实例。这类似于React的状态管理能从根本上避免副作用但实现复杂度较高。4.4 与现有框架如LangChain集成我们上面的实现是原生的。在实际项目中你很可能是在LangChain、LlamaIndex等现有框架上工作。这些框架通常已经提供了某种回调Callback机制。我们的中间件系统可以构建在这些回调机制之上或者将其包装成更统一、强大的形式。以LangChain为例它有一个丰富的BaseCallbackHandler系统。我们可以创建一个MiddlewareCallbackHandler将LangChain的各种回调事件on_llm_start,on_tool_start等映射到我们自定义的中间件钩子上。from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import AgentAction, AgentFinish class LangChainMiddlewareAdapter(BaseCallbackHandler): 将LangChain回调适配到我们的中间件系统。 def __init__(self, middleware_manager: MiddlewareManager, agent_context: AgentContext): self.manager middleware_manager self.context agent_context async def on_chain_start(self, serialized, inputs, **kwargs): # 映射到 on_agent_start await self.manager.invoke_hook(on_agent_start, self.context) async def on_llm_start(self, serialized, prompts, **kwargs): # 映射到 before_llm_call self.context.llm_messages prompts # 可能需要转换格式 await self.manager.invoke_hook(before_llm_call, self.context) async def on_llm_end(self, response, **kwargs): self.context.llm_response_raw response.generations[0][0].text await self.manager.invoke_hook(after_llm_call, self.context) async def on_tool_start(self, serialized, input_str, **kwargs): # 解析出工具名和参数这里简化 self.context.current_tool_name serialized.get(name) self.context.tool_call_args {input: input_str} await self.manager.invoke_hook(before_tool_execution, self.context) # 如果安全中间件阻止了执行这里需要能中断LangChain的流程 # 这可能需要更高级的集成比如修改Agent的执行器。 async def on_tool_end(self, output, **kwargs): self.context.tool_execution_result output await self.manager.invoke_hook(after_tool_execution, self.context) async def on_agent_finish(self, finish: AgentFinish, **kwargs): self.context.final_output finish.return_values.get(output) self.context.is_finished True await self.manager.invoke_hook(on_agent_end, self.context)这样你就可以在LangChain项目中使用同一套中间件系统了。关键在于处理好框架原生回调系统与我们自定义钩子之间的数据转换和流程控制。5. 常见问题排查与实战技巧在实际开发和部署中你肯定会遇到各种问题。下面是一些典型场景和解决思路。5.1 中间件执行顺序混乱或未生效症状日志中间件没记录到工具调用的信息或者安全校验在工具执行后才触发。排查检查注册顺序确认中间件在MiddlewareManager中的注册顺序是否符合预期。使用上面提到的优先级机制。检查钩子绑定确保你的自定义中间件正确重写了目标钩子方法如before_tool_execution。拼写错误或使用了同步方法未加async都会导致钩子不被调用。添加调试中间件写一个最简单的调试中间件在每个钩子里打印一行日志确认引擎是否按预期触发了所有钩子。5.2 中间件修改了上下文但引擎未使用症状在before_llm_call中修改了context.llm_messages但实际发送给LLM的仍是旧消息。排查确认引擎读取点检查你的AgentEngine代码是在调用before_llm_call钩子之前还是之后读取llm_messages并发送给LLM的必须是之后。检查数据流确保上下文对象在钩子链中是按引用传递的同一个对象而不是被复制了。查看引擎逻辑引擎是否在调用钩子后重新从上下文中获取了最新数据例如# 正确做法钩子执行后使用可能被修改过的上下文数据 await self.middleware.invoke_hook(before_llm_call, context) llm_response await self.llm.acomplete(context.llm_messages) # 使用context.llm_messages5.3 异步中间件导致性能问题或死锁症状系统在高并发下变慢或者偶尔卡住无响应。排查避免阻塞操作确保中间件内的所有I/O操作网络请求、数据库查询、文件读写都是异步的。如果一个同步阻塞操作如requests.get在异步中间件中被调用它会阻塞整个事件循环。使用异步库将requests替换为aiohttp或httpx将同步数据库驱动替换为异步驱动如asyncpg用于PostgreSQL。设置超时为中间件中可能长时间运行的操作如外部API调用添加超时控制使用asyncio.wait_for。async def before_tool_execution(self, context): try: # 假设有一个异步的远程权限校验 async with aiohttp.ClientSession() as session: async with session.post(https://auth-service/check, json..., timeout5.0) as resp: # ... except asyncio.TimeoutError: context.tool_execution_error Exception(权限校验超时) context.final_output 系统繁忙请稍后再试。5.4 如何测试中间件中间件的测试策略与普通业务逻辑不同因为它高度依赖于执行上下文和生命周期。单元测试单个中间件Mock一个AgentContext对象直接调用中间件的钩子方法断言其对上下文的修改或产生的副作用如日志记录是否符合预期。pytest.mark.asyncio async def test_security_middleware_blocks_forbidden_tool(): middleware SecurityCheckMiddleware(forbidden_tools[delete_database]) context AgentContext(current_tool_namedelete_database, tool_call_args{}) await middleware.before_tool_execution(context) assert context.final_output is not None assert 禁止 in context.final_output # 或者检查 context.tool_execution_error集成测试中间件链创建一个最小化的MockAgentEngine注册需要测试的中间件模拟完整的执行流程检查最终的上下文状态和输出。端到端测试将中间件系统与真实的LLM和工具集成进行完整的场景测试确保中间件的加入不会破坏Agent原有的核心功能。5.5 中间件系统的扩展性思考当中间件越来越多时管理起来会变得复杂。可以考虑以下方向按功能模块化将中间件分组例如“可观测性模块”包含日志、指标、追踪中间件、“安全模块”、“业务增强模块”。每个模块可以独立配置和启用。动态加载与热重载支持从配置文件或数据库中读取中间件配置在不重启服务的情况下动态添加、移除或更新中间件。这对于SaaS平台或需要频繁调整策略的场景非常有用。中间件市场/仓库如果中间件系统设计得足够通用可以抽象出标准的接口和配置范式鼓励团队内部甚至社区贡献可复用的中间件形成一个生态。6. 总结与个人体会构建一个Agent中间件系统本质上是在为AI应用的“操作系统”增加一套强大的“系统调用”和“驱动接口”。它把那些原本散落在业务代码各处的横切关注点Cross-cutting Concerns——日志、安全、监控、缓存、限流、数据脱敏等——集中管理起来让Agent的核心逻辑保持干净和专注。从我自己的实践来看这套系统的价值在项目规模扩大后体现得尤为明显。当你有十几个不同的Agent每个Agent又调用着数十个工具时如果没有中间件想要统一加上调用链追踪Tracing功能那将是一场灾难。而有了中间件你只需要编写并注册一个TracingMiddleware所有Agent瞬间就具备了全链路追踪能力。最后分享一个关键的取舍经验中间件的强大也带来了复杂性。不是所有逻辑都适合放进中间件。我的原则是只有那些与核心业务逻辑正交的、全局性的、可插拔的功能才适合用中间件实现。如果一个逻辑只针对某一个特定的Agent或工具并且紧密耦合业务那么把它放在Agent或工具本身的实现里会更清晰。过度设计中间件会让系统变得难以理解和调试。从最简单的日志中间件开始当同一个模式出现第三次时再考虑将其抽象成通用中间件这是一个比较稳妥的演进路径。