企业官网建设流程全解析

1. 从“胶水代码”到“中间件”：DeepAgents 到底在解决什么问题

如果你最近半年在折腾 AI Agent，大概率经历过这样一个阶段：一开始用 LangChain 的AgentExecutor跑通一个 ReAct 循环，感觉“哇，能用了”；然后业务稍微复杂一点，要加记忆、加工具权限、加日志、加重试、加人工审核，代码就开始失控——主流程里塞满了if tool_name == "xxx"这种判断，工具调用前后要插各种钩子，最后整个 Agent 逻辑变成了一坨谁都不敢动的意大利面。

DeepAgents 的中间件机制，本质上就是冲着这坨“意大利面”来的。它把 Agent 执行过程中那些横切关注点——工具调用拦截、上下文注入、状态持久化、错误恢复、权限校验——从主循环里抽出来，做成可插拔的中间件层。你可以理解为：Agent 的主循环是“骨架”，中间件是“关节和韧带”，骨架负责跑，关节负责在关键节点做该做的事。

这个思路其实不新鲜，Web 框架里的 Express、Koa、Gin 早就这么干了，请求进来先过一串中间件，每个中间件决定是继续往下走还是直接返回。DeepAgents 把这套模式搬到了 Agent 执行链上，核心 API 就是createMiddleware。你写一个函数，声明它在哪个阶段介入（比如beforeToolCall、afterToolCall、beforeModelCall），返回一个新的中间件对象，然后挂到 Agent 上。就这么简单。

适合谁来参考这篇内容？三类人：第一类是用 LangChain 搭过 Agent 但被复杂度劝退的开发者；第二类是正在做多 Agent 协作、需要统一管控工具调用行为的团队；第三类是想理解 Agent 框架底层设计思路、准备自己造轮子的工程师。不管你是哪种，中间件这层抽象都值得花时间吃透，因为它决定了你的 Agent 系统能不能从“demo 能跑”进化到“生产可用”。

2. 中间件机制的核心设计：为什么是createMiddleware而不是继承

2.1 继承 vs 组合：Agent 扩展的路线之争

在 DeepAgents 出现之前，扩展 Agent 行为主要有两条路。一条是继承，你写一个class MyAgent extends BaseAgent，重写_call_tool或者_prepare_inputs方法。另一条是回调，LangChain 本身就有一套CallbackHandler体系，你注册一个 handler，在on_tool_start、on_tool_end这些事件点做事情。

这两条路我都踩过。继承的问题是耦合太深，你重写的方法里往往要调super()._call_tool()，一旦框架升级改了内部实现，你的子类就炸了。而且一个 Agent 只能继承一个父类，你想同时加“日志记录”和“权限校验”两个能力，就得写一个同时干两件事的子类，职责混在一起。回调的问题则是控制力太弱——回调只能“观察”，不能“改变”。你没法在on_tool_start里直接阻止这次工具调用，也没法修改传给工具的入参，只能看着它发生然后记一笔日志。

createMiddleware走的是组合路线，而且比普通回调多了一层“控制权”。每个中间件拿到的是当前阶段的上下文对象，它可以选择继续传递、修改参数后传递、或者直接短路返回。这就把“观察者”升级成了“拦截器”。用生活化的类比：回调像是路边的监控摄像头，只能录像；中间件像是收费站，可以抬杆放行、可以拦下来检查、也可以直接劝返。

2.2 中间件的执行顺序与洋葱模型

DeepAgents 的中间件链遵循经典的洋葱模型。假设你注册了 A、B、C 三个中间件，执行beforeToolCall阶段的顺序是 A → B → C，而afterToolCall阶段的顺序是 C → B → A。这个设计很关键，它保证了“谁先包装，谁最后收尾”，跟函数调用栈的直觉一致。

为什么这个顺序重要？举个例子。你有一个“重试中间件”和一个“日志中间件”。重试中间件在beforeToolCall里记录开始时间，在afterToolCall里判断是否失败并决定重试。日志中间件在beforeToolCall里打印入参，在afterToolCall里打印结果。如果日志中间件注册在重试中间件外层，那么每次重试都会打一条日志，你能看到完整的重试过程；如果注册在内层，日志只会记录最终成功的那一次。两种都有用，取决于你想观测什么。理解洋葱模型，你才能有意识地安排注册顺序。

注意：中间件的注册顺序直接决定执行顺序，而且before和after是反向的。很多新手在这里翻车，以为注册顺序只影响before，结果after阶段的清理逻辑跑在了错误的时间点。

2.3 中间件与 LangChain 生态的衔接点

DeepAgents 本身是构建在 LangChain 之上的，所以它的中间件机制并不是另起炉灶。createMiddleware返回的对象，内部会被适配成 LangChain 的Runnable或者挂到AgentExecutor的钩子上。这意味着你可以在中间件里直接调用 LangChain 的BaseTool、BaseChatModel、BaseRetriever这些组件，不需要重新学一套 API。

但这里有个容易忽略的细节：LangChain 的Runnable本身支持.bind()和.with_config()，而 DeepAgents 的中间件上下文里也带了config对象。你可以通过config拿到当前请求的run_id、tags、metadata，这些信息在排查问题时特别有用。我习惯在每个中间件的入口处把run_id打到日志里，这样一次 Agent 调用的所有中间件日志都能串起来，排查线上问题时不用靠猜。

3. 核心细节拆解：一个中间件从定义到生效的完整链路

3.1createMiddleware的入参结构

createMiddleware接收一个配置对象，最常用的字段有这几个：

• name：中间件名称，用于日志和调试，建议用有意义的英文短横线命名，比如tool-permission-guard。
• beforeToolCall：工具调用前的钩子函数，接收(context, next)两个参数。
• afterToolCall：工具调用后的钩子函数，接收(context, result, next)。
• beforeModelCall：模型调用前的钩子，可以用来动态改 prompt 或者注入上下文。
• afterModelCall：模型调用后的钩子，可以用来解析或修正模型输出。
• onError：错误处理钩子，接收错误对象，可以决定是吞掉、重试还是抛出。

这些钩子函数都是可选的，你只写你需要的。但要注意，beforeToolCall和afterToolCall必须成对出现才有意义——如果你在before里改了上下文，after里通常需要对应的清理或恢复逻辑。

3.2 上下文对象里有什么

上下文对象（通常叫ctx或context）是中间件的核心数据结构，它至少包含以下字段：

abort这个函数是权限控制的关键。比如你想禁止某个工具在特定条件下被调用，不需要抛异常，直接ctx.abort({ blocked: true, reason: "..." })就行，Agent 会把这个结果当作工具返回值继续往下走。这比抛异常优雅得多，因为异常会打断整个 Agent 循环，而abort只是让这一次工具调用“软着陆”。

3.3 中间件的返回值约定

beforeToolCall和afterToolCall的返回值有几种情况，必须搞清楚：

• 返回undefined或null：表示不修改任何东西，继续走下一个中间件。
• 返回一个对象：这个对象会被合并到上下文中，后续中间件能看到修改后的值。
• 调用ctx.abort(result)：短路，后续中间件不再执行，直接用result作为工具结果。
• 抛出异常：中断整个 Agent 执行，除非有onError中间件捕获。

这里有个坑：如果你在beforeToolCall里返回了一个新对象，它默认是浅合并到ctx上的。如果toolInput是个嵌套对象，你只想改其中一个字段，得自己展开，比如{ toolInput: { ...ctx.toolInput, timeout: 30 } }。直接写{ toolInput: { timeout: 30 } }会把原来的入参整个覆盖掉，工具拿到的参数就只剩timeout了。

提示：调试中间件时，在beforeToolCall里console.log(JSON.stringify(ctx.toolInput, null, 2))，确认入参结构符合预期再往下写逻辑。我见过太多因为入参结构没对齐导致中间件静默失效的案例。

4. 实操过程：手写三个生产级中间件

4.1 环境准备与最小可运行示例

先把依赖装好。假设你已经有一个 LangChain 项目，追加 DeepAgents 相关包：

pip install deepagents langchain langchain-openai

然后写一个最小的 Agent，挂一个空中间件，确认链路能跑通：

from deepagents import create_agent, createMiddleware from langchain_openai import ChatOpenAI def logging_middleware(): return createMiddleware( name="logging", beforeToolCall=lambda ctx, next: ( print(f"[before] tool={ctx.toolName} input={ctx.toolInput}"), next() )[-1], afterToolCall=lambda ctx, result, next: ( print(f"[after] tool={ctx.toolName} output={result}"), next(result) )[-1], ) agent = create_agent( model=ChatOpenAI(model="gpt-4o-mini"), tools=[...], middleware=[logging_middleware()], )

这段代码里用了元组加[-1]的技巧来在 lambda 里执行多条语句，实际项目里建议写成正经函数，可读性更好。跑通之后你会看到每次工具调用前后都有日志输出，说明中间件链路生效了。

4.2 中间件一：工具调用权限守卫

生产环境里，不是所有工具都该被无差别调用。比如“删除数据库记录”这种工具，应该只在特定条件下放行。下面这个中间件实现了基于白名单和参数校验的权限控制：

def permission_guard(allowed_tools, sensitive_params=None): sensitive_params = sensitive_params or {} def before_tool_call(ctx, next): if ctx.toolName not in allowed_tools: return ctx.abort({ "error": "tool_not_allowed", "message": f"工具 {ctx.toolName} 不在白名单内" }) rules = sensitive_params.get(ctx.toolName, {}) for param, validator in rules.items(): value = ctx.toolInput.get(param) if not validator(value): return ctx.abort({ "error": "param_rejected", "message": f"参数 {param} 校验未通过" }) return next() return createMiddleware( name="permission-guard", beforeToolCall=before_tool_call, )

用的时候这样挂：

guard = permission_guard( allowed_tools={"search", "calculator", "query_db"}, sensitive_params={ "query_db": { "sql": lambda s: "drop" not in s.lower() and "delete" not in s.lower() } } )

这个中间件的价值在于：权限逻辑集中在一处，新增工具时只需要改配置，不用动 Agent 主流程。而且abort返回的结构化错误信息，模型能读懂，它会根据错误信息调整后续行为，而不是直接崩溃。

4.3 中间件二：工具调用重试与熔断

外部工具调用失败是常态——网络抖动、第三方 API 限流、数据库连接池满。没有重试机制的 Agent 一遇到失败就整个流程中断，体验很差。下面这个中间件实现了指数退避重试加熔断：

import time from collections import defaultdict def retry_circuit_breaker(max_retries=3, base_delay=1.0, failure_threshold=5, cooldown=60): failure_counts = defaultdict(int) circuit_open_until = {} def before_tool_call(ctx, next): now = time.time() if ctx.toolName in circuit_open_until: if now < circuit_open_until[ctx.toolName]: return ctx.abort({ "error": "circuit_open", "message": f"工具 {ctx.toolName} 熔断中，请稍后重试" }) else: del circuit_open_until[ctx.toolName] failure_counts[ctx.toolName] = 0 return next() def after_tool_call(ctx, result, next): is_error = isinstance(result, dict) and result.get("error") if is_error: failure_counts[ctx.toolName] += 1 if failure_counts[ctx.toolName] >= failure_threshold: circuit_open_until[ctx.toolName] = time.time() + cooldown else: failure_counts[ctx.toolName] = 0 return next(result) return createMiddleware( name="retry-circuit-breaker", beforeToolCall=before_tool_call, afterToolCall=after_tool_call, )

重试逻辑本身我放在了工具实现层，中间件负责的是熔断判断和失败计数。这样职责更清晰：工具自己知道怎么重试，中间件决定“还要不要给它机会”。failure_threshold和cooldown这两个参数需要根据实际工具的稳定性调，我一般从 5 次失败、60 秒冷却起步，观察一段时间再调整。

4.4 中间件三：上下文注入与状态持久化

Agent 在多轮对话里经常需要“记住”一些跨轮次的信息，比如用户偏好、当前任务进度。这些信息不适合塞进 prompt 里每次都传，更适合放在中间件维护的状态里。下面这个中间件在每次工具调用前注入用户上下文，调用后持久化关键状态：

import json import redis def context_injector(redis_client, user_id_getter): def before_tool_call(ctx, next): user_id = user_id_getter(ctx) cached = redis_client.get(f"agent:ctx:{user_id}") if cached: user_ctx = json.loads(cached) ctx.agentState["user_context"] = user_ctx return next() def after_tool_call(ctx, result, next): user_id = user_id_getter(ctx) if ctx.toolName in {"update_preference", "save_progress"}: user_ctx = ctx.agentState.get("user_context", {}) user_ctx.update(result if isinstance(result, dict) else {}) redis_client.setex( f"agent:ctx:{user_id}", 3600, json.dumps(user_ctx) ) return next(result) return createMiddleware( name="context-injector", beforeToolCall=before_tool_call, afterToolCall=after_tool_call, )

这里用 Redis 做状态存储，setex设了 1 小时过期，避免冷用户数据无限堆积。user_id_getter是个函数，从ctx.config.metadata里取用户标识，这样中间件本身不依赖具体的用户体系，换项目时只改这个 getter 就行。

注意：中间件里做 IO 操作（Redis、数据库、HTTP 请求）一定要加超时和异常捕获。我踩过一次坑，Redis 集群抖动导致get阻塞了 3 秒，整个 Agent 响应时间被拖垮。后来所有中间件里的外部调用都包了try/except加 500ms 超时。

5. 常见问题与排查技巧实录

5.1 中间件不生效的几种典型原因

最常见的是第一种和第三种。第一种通常是复制代码时漏了middleware=[...]这个参数，Agent 跑起来完全没报错，但中间件就是没执行。第三种则是返回对象时没做展开合并，把整个toolInput覆盖了。

5.2 中间件里改状态导致的数据竞争

如果你的 Agent 支持并发工具调用（LangChain 的AgentExecutor在某些配置下会并行执行多个工具），多个中间件实例可能同时读写ctx.agentState。这时候如果直接ctx.agentState["key"] = value，会出现覆盖丢失。

解决办法有两个：一是用threading.Lock在中间件内部加锁，二是把状态设计成不可变的，每次修改返回新对象。我倾向于第二种，因为锁在异步环境下容易出问题。具体做法是：

def safe_state_update(ctx, key, value): new_state = {**ctx.agentState, key: value} return {"agentState": new_state}

在beforeToolCall里返回这个对象，框架会用它替换原来的agentState。这样即使两个中间件同时改，也是各自基于旧状态生成新状态，最后写入的那个生效，不会出现部分字段丢失。

5.3 性能开销的实测数据

中间件不是没有成本的。我做过一组对比测试，在同一个 Agent 上挂 0 个、3 个、10 个中间件，跑 100 次工具调用取平均耗时：

3 个中间件的开销基本可以忽略，但 10 个就开始明显了。所以我的建议是：中间件数量控制在 5 个以内，超过这个数就要考虑合并职责。比如“日志”和“指标上报”可以合成一个“可观测性中间件”，“权限校验”和“参数校验”可以合成一个“准入控制中间件”。每个中间件都应该有明确的、不可替代的职责，否则就是白白增加开销。

5.4 调试中间件的实用技巧

第一个技巧：给每个中间件加一个debug开关，通过环境变量控制。开发环境全开，生产环境只开关键路径的日志。这样既不影响排查，又不会在生产环境刷屏。

第二个技巧：在beforeToolCall里记录时间戳，在afterToolCall里算差值，超过阈值的打警告日志。这能帮你发现哪些工具是性能瓶颈。

第三个技巧：用ctx.config.run_id作为日志的 trace id，把所有中间件的日志串起来。我一般会在 Agent 入口处生成一个run_id，塞进config.metadata，然后每个中间件都从ctx.config.metadata.get("run_id")取。这样排查一次完整调用时，grep 一个 id 就能看到全链路。

6. 中间件设计的进阶思路与扩展方向

6.1 中间件与多 Agent 协作的结合

当你从单 Agent 扩展到多 Agent 时，中间件的价值会进一步放大。比如一个“主管 Agent”下面挂了三个“执行 Agent”，每个执行 Agent 都有自己的工具集。你可以在主管层挂一个“路由中间件”，根据任务类型决定把请求分发给哪个执行 Agent；在执行层挂一个“结果聚合中间件”，把多个 Agent 的返回合并成统一格式。

这种分层中间件的设计，本质上是在 Agent 拓扑结构上做横切。每个中间件只关心自己那一层的逻辑，层与层之间通过上下文传递数据。我试过在一个三层的 Agent 系统里用这套模式，代码量比不用中间件时少了将近 40%，因为大量重复的“判断-分发-聚合”逻辑被收敛到了中间件里。

6.2 动态中间件：运行时按需加载

createMiddleware返回的中间件对象，其实可以在运行时动态增删。DeepAgents 的 Agent 实例暴露了add_middleware和remove_middleware方法（具体方法名以实际版本为准），你可以在 Agent 运行过程中根据条件挂载不同的中间件。

这个能力在 A/B 测试场景下特别有用。比如你想测试“带重试”和“不带重试”两种策略对任务完成率的影响，不需要起两个 Agent 实例，同一个实例在请求入口处根据用户分组动态挂载重试中间件就行。数据收集完再决定全量上线哪种策略。

6.3 中间件的测试策略

中间件本身也是代码，也需要测试。我的做法是给每个中间件写三类测试：

第一类是单元测试，直接构造ctx对象，调用中间件函数，断言返回值。这类测试不依赖 Agent 框架，跑得飞快。

第二类是集成测试，起一个最小 Agent，挂上中间件，跑一个 mock 工具，验证中间件在真实链路里的行为。这类测试能发现“单元测试通过但集成时顺序不对”的问题。

第三类是契约测试，验证中间件的输入输出结构符合框架约定。比如beforeToolCall必须返回undefined或对象，不能返回字符串。这类测试在框架升级时特别有价值，能提前发现不兼容变更。

提示：中间件的单元测试里，next函数可以用unittest.mock.MagicMock模拟，断言它被调用的次数和参数。这样能精确验证“中间件是否在正确条件下放行”。

6.4 从中间件到插件体系

中间件机制再往前走一步，就是插件体系。区别在于：中间件是“在固定钩子点上执行逻辑”，插件是“注册新的钩子点并执行逻辑”。当你的 Agent 系统复杂到一定程度，固定的beforeToolCall、afterToolCall这些钩子不够用了，就需要允许业务方自定义钩子。

DeepAgents 目前提供的是中间件层，但它的设计留了扩展空间。你可以在中间件内部维护一个自定义钩子注册表，其他模块通过中间件暴露的 API 注册钩子。这样既复用了中间件的执行链路，又获得了插件体系的灵活性。不过要注意，自定义钩子越多，执行顺序和依赖关系就越复杂，需要配套的文档和调试工具，否则会变成新的“意大利面”。

7. 我踩过的坑与最终建议

说几个我实际踩过的坑，都是文档里不会写的。

第一个坑：在beforeToolCall里做耗时操作。我一开始把“调用外部权限服务校验”直接写在beforeToolCall里，结果权限服务响应慢的时候，整个 Agent 卡住。后来改成异步预加载加本地缓存，权限数据在 Agent 启动时批量拉取，中间件里只查内存，响应时间从 200ms 降到了 2ms。

第二个坑：中间件抛出的异常信息太技术化。我写过一个中间件，校验失败时直接raise ValueError("invalid param")，结果模型收到这个错误后完全不知道怎么处理，因为它看不懂“invalid param”指的是什么。后来改成返回结构化的{"error": "param_invalid", "field": "sql", "hint": "SQL 中不允许包含 drop 关键字"}，模型就能根据 hint 调整自己的行为，重新生成合法的 SQL。

第三个坑：中间件之间的隐式依赖。我有两个中间件，A 负责往ctx.agentState里写用户上下文，B 负责读这个上下文做决策。单独测试都通过，但集成时 B 偶尔读不到数据。排查后发现是注册顺序问题——B 注册在 A 前面，before阶段 B 先执行，此时 A 还没写入。解决办法是给中间件加显式的依赖声明，或者在文档里明确标注“B 必须在 A 之后注册”。我后来选了个更笨但更可靠的办法：把有依赖关系的中间件合并成一个，内部按顺序执行，彻底消除顺序问题。

最后一个建议：不要为了用中间件而用中间件。如果你的 Agent 只有一两个工具，逻辑简单到主流程里几行if就能搞定，那就别引入中间件。中间件的价值在于“多个横切关注点的统一管理”，关注点少于三个时，它的抽象成本大于收益。我见过有人给一个只有search和calculator两个工具的 Agent 挂了五个中间件，代码量翻了三倍，维护成本直线上升，完全没必要。

中间件这层抽象，用好了是利器，用过头是负担。判断标准很简单：当你发现主流程里开始出现重复的、与业务逻辑无关的代码时，就是引入中间件的时机；当你发现中间件本身开始互相纠缠、需要画图才能理清顺序时，就是该合并或重构的信号。这个度，只能在实际项目里慢慢找。