1. Agent模型调用的拦截需求与Middleware解决方案在开发AI Agent时我们经常遇到这样的场景Agent运行过程中需要插入额外的处理逻辑比如记录日志、安全检查、性能监控等。传统做法是直接修改Agent核心代码但这会导致代码臃肿且难以维护。Middleware模式提供了一种优雅的解决方案它就像给Agent安装了一个拦截器可以在不修改核心逻辑的情况下灵活地插入各种处理逻辑。Middleware的核心思想是面向切面编程(AOP)它允许我们在模型调用的关键节点调用前和调用后插入自定义逻辑。这种设计模式在Web开发中很常见如Express.js的中间件现在也被广泛应用于AI Agent开发领域。2. Middleware的核心机制与实现原理2.1 Middleware的基本结构一个标准的Middleware通常需要实现两个关键方法from langchain.agents.middleware import AgentMiddleware from langchain.agents import AgentState from langgraph.runtime import Runtime class CustomMiddleware(AgentMiddleware): def before_model(self, state: AgentState, runtime: Runtime) - dict | None: 模型调用前执行的逻辑 # 可以访问和修改state中的消息、工具等 # 返回None表示继续流程返回dict可以改变流程走向 pass def after_model(self, state: AgentState, runtime: Runtime) - None: 模型调用后执行的逻辑 # 可以访问模型响应结果但不能改变流程 pass2.2 执行流程控制Middleware最强大的特性在于它可以控制执行流程被动观察返回None表示不干预流程仅执行附加逻辑如日志记录主动干预返回dict可以改变流程比如跳过模型调用直接返回结果修改输入消息内容终止当前会话{ jump_to: end, # 跳过模型调用直接结束 messages: [AIMessage(content自定义响应)] # 添加的消息 }3. 实战构建实用的Middleware组件3.1 日志记录Middleware一个完整的日志记录Middleware应该包含以下功能import logging from datetime import datetime class EnhancedLoggingMiddleware(AgentMiddleware): def __init__(self, log_levellogging.INFO): self.logger logging.getLogger(AgentLogger) self.logger.setLevel(log_level) def before_model(self, state: AgentState, runtime: Runtime) - None: context { timestamp: datetime.now().isoformat(), message_count: len(state[messages]), last_user_input: next( (msg.content for msg in reversed(state[messages]) if msg.type human), None) } self.logger.info(fPre-model call: {context}) def after_model(self, state: AgentState, runtime: Runtime) - None: response state[messages][-1].content metrics { response_length: len(response), response_time: runtime.get(model_time, 0) } self.logger.info(fPost-model call: {metrics})提示在生产环境中建议将日志输出到文件或日志系统而非直接打印到控制台。3.2 安全拦截Middleware增强版的安全Middleware可以支持关键词黑名单正则表达式模式匹配敏感操作检测import re from typing import List class AdvancedSafetyMiddleware(AgentMiddleware): def __init__(self, blacklist: List[str] None, dangerous_patterns: List[str] None): self.blacklist blacklist or [删除, 危险, 密码, root] self.patterns [re.compile(p) for p in dangerous_patterns or []] def before_model(self, state: AgentState, runtime: Runtime) - dict | None: last_msg state[messages][-1].content # 检查黑名单关键词 if any(keyword in last_msg for keyword in self.blacklist): return self._block_action(检测到禁用关键词) # 检查危险模式 if any(pattern.search(last_msg) for pattern in self.patterns): return self._block_action(检测到危险操作模式) return None def _block_action(self, reason: str) - dict: return { jump_to: end, messages: [AIMessage( contentf{reason}操作已终止。如需帮助请联系管理员。 )] }4. Middleware的高级应用场景4.1 上下文增强Middleware可以在模型调用前注入相关上下文信息class ContextEnhancementMiddleware(AgentMiddleware): def __init__(self, knowledge_base): self.knowledge knowledge_base def before_model(self, state: AgentState, runtime: Runtime) - None: user_query state[messages][-1].content related_info self.knowledge.search(user_query) if related_info: state[context] related_info # 注入上下文4.2 限流Middleware控制模型调用频率防止滥用from collections import deque import time class RateLimitMiddleware(AgentMiddleware): def __init__(self, max_calls5, period60): self.max_calls max_calls self.period period self.call_times deque() def before_model(self, state: AgentState, runtime: Runtime) - dict | None: now time.time() # 移除过期的调用记录 while self.call_times and now - self.call_times[0] self.period: self.call_times.popleft() if len(self.call_times) self.max_calls: return { jump_to: end, messages: [AIMessage( content请求过于频繁请稍后再试 )] } self.call_times.append(now) return None4.3 缓存Middleware对重复请求返回缓存结果import hashlib class CacheMiddleware(AgentMiddleware): def __init__(self, cache_size100): self.cache {} self.cache_size cache_size def before_model(self, state: AgentState, runtime: Runtime) - dict | None: query state[messages][-1].content query_hash hashlib.md5(query.encode()).hexdigest() if query_hash in self.cache: return { jump_to: end, messages: [AIMessage( contentself.cache[query_hash] )] } return None def after_model(self, state: AgentState, runtime: Runtime) - None: if len(self.cache) self.cache_size: self.cache.popitem() # 简单LRU策略 query state[messages][-2].content # 用户的上一条消息 response state[messages][-1].content query_hash hashlib.md5(query.encode()).hexdigest() self.cache[query_hash] response5. Middleware的链式调用与执行顺序当多个Middleware组合使用时它们的执行顺序很重要before_model调用顺序按照Middleware列表顺序依次执行after_model调用顺序与before_model相反栈式结构流程中断任一Middleware返回非None值都会中断后续Middleware执行# 推荐的Middleware顺序 middlewares [ RateLimitMiddleware(), # 最先执行限流检查 SafetyMiddleware(), # 然后安全检查 LoggingMiddleware(), # 记录原始请求 ContextEnhancementMiddleware(kb), # 上下文增强 CacheMiddleware() # 最后检查缓存 ]6. 性能优化与调试技巧6.1 Middleware性能监控可以添加专门的性能监控Middlewareclass PerformanceMonitorMiddleware(AgentMiddleware): def before_model(self, state: AgentState, runtime: Runtime) - None: runtime[start_time] time.time() def after_model(self, state: AgentState, runtime: Runtime) - None: duration time.time() - runtime[start_time] print(f模型调用耗时: {duration:.3f}秒) if duration 1.0: # 慢请求警告 print(f慢请求警告: {state[messages][-2].content[:50]}...)6.2 调试技巧隔离测试逐个启用Middleware确认各自功能正常状态检查在before_model中打印state完整内容错误处理Middleware内部应该捕获自己的异常避免影响主流程class SafeMiddleware(AgentMiddleware): def before_model(self, state: AgentState, runtime: Runtime) - dict | None: try: # 业务逻辑 return None except Exception as e: print(fMiddleware错误: {str(e)}) return None # 即使出错也不中断流程7. 生产环境最佳实践配置化通过配置文件管理Middleware开关和参数依赖注入避免Middleware直接实例化外部依赖单元测试为每个Middleware编写独立测试用例性能考量IO密集型操作如网络请求应该异步化# 配置示例 MIDDLEWARE_CONFIG { safety: { enable: True, blacklist: [删除, 格式化, 关机], patterns: [rrm -rf, rDROP TABLE] }, logging: { enable: True, level: INFO } } # 根据配置动态创建Middleware链 def setup_middlewares(config): middlewares [] if config[safety][enable]: middlewares.append(SafetyMiddleware( blacklistconfig[safety][blacklist], patternsconfig[safety][patterns] )) if config[logging][enable]: middlewares.append(LoggingMiddleware( log_levelconfig[logging][level] )) return middlewares8. 常见问题与解决方案8.1 Middleware执行顺序问题问题多个Middleware相互影响顺序不当导致功能异常解决方案按照安全→日志→业务→缓存的通用顺序排列为Middleware添加优先级属性动态排序8.2 状态污染问题问题Middleware意外修改了state导致后续流程异常解决方案在修改state前创建深拷贝使用不可变数据结构import copy class SafeStateMiddleware(AgentMiddleware): def before_model(self, state: AgentState, runtime: Runtime) - dict | None: original_state copy.deepcopy(state) # 保存原始状态 try: # 修改state return None except Exception: return {restore_state: original_state} # 出错时恢复状态8.3 性能瓶颈问题问题Middleware引入过多计算或IO导致延迟增加解决方案异步化处理采样记录而非全量记录使用轻量级检查如布隆过滤器import asyncio class AsyncLoggingMiddleware(AgentMiddleware): async def _async_log(self, message): # 异步写入日志系统 pass def after_model(self, state: AgentState, runtime: Runtime) - None: loop asyncio.get_event_loop() message state[messages][-1].content loop.create_task(self._async_log(message)) # 异步记录9. 扩展思考Middleware设计模式的应用Middleware模式不仅适用于模型调用拦截还可以应用于工具调用拦截在工具执行前后插入逻辑消息处理管道对输入/输出消息进行统一处理Agent生命周期钩子在Agent启动/停止时执行操作class ToolMiddleware: def before_tool(self, tool_name: str, input: dict) - dict | None: 工具调用前执行 pass def after_tool(self, tool_name: str, output: str) - str | None: 工具调用后执行 pass class LifecycleMiddleware: def on_agent_start(self): Agent启动时执行 pass def on_agent_stop(self): Agent停止时执行 passMiddleware模式为Agent开发提供了极大的灵活性和可扩展性。通过合理设计和组合Middleware可以实现各种横切关注点而无需修改核心业务逻辑。这种设计模式遵循了开闭原则对扩展开放对修改关闭是构建可维护、可扩展AI系统的重要实践。
