DeepAgents中间件实战：解决Agent流程失控与Token爆炸

1. 从一次真实的踩坑说起：为什么我要折腾 DeepAgents 中间件

去年年底我接了个私活，帮一家做跨境电商的朋友搭一套自动处理客服工单的 AI Agent 系统。需求听起来不复杂：用户发来的售后消息，Agent 自动判断意图、查订单、决定是退款还是换货、最后生成回复。我一开始用的是最朴素的 LangChain Agent 写法，一个initialize_agent加几个 Tool，跑通 Demo 只花了半天。

结果一上真实流量就崩了。问题不是模型不行，而是流程失控：Agent 有时候查完订单忘了退款，有时候退款了又重复查订单，还有时候在"要不要转人工"这个判断上反复横跳，一个请求烧掉几万 token。我盯着日志看了整整两天，才意识到问题的本质——我一直在优化"模型能力"，但真正卡脖子的是Agent 的执行流程编排。

这就是 DeepAgents 中间件要解决的事。简单说，DeepAgents 是构建在 LangChain 之上的一层 Agent 中间件框架，它把 Agent 从"一个会调工具的大模型"升级成"一个有明确执行阶段、可插拔拦截逻辑的运行时系统"。你可以把它理解成 Web 开发里的 Express/Koa 中间件——请求进来，经过一层层中间件处理，每层都能读状态、改状态、决定要不要继续往下走。Agent 也一样：思考、调工具、观察结果、再思考，每个环节都可以挂中间件。

这篇文章适合三类人看：一是已经用 LangChain 写过 Agent、但被流程失控折磨过的开发者；二是正在选型 Agent 框架、纠结 LangChain / Dify / CrewAI 到底选哪个的技术负责人；三是想搞清楚"中间件"这个概念在 Agent 领域到底怎么落地的人。我会从设计思路讲到实操代码，再把我踩过的坑一个个摊开讲，尽量让你少走我走过的弯路。

2. DeepAgents 中间件的整体设计与思路拆解

2.1 为什么 Agent 需要"中间件"这层抽象

先说个反直觉的观点：大部分 Agent 项目失败，不是因为模型不够聪明，而是因为缺少工程化的执行骨架。

传统的 LangChain Agent 执行循环大概是这样的：模型输出一个 Action，框架解析出工具名和参数，调用工具，把结果塞回 Prompt，再让模型继续。这个循环本身没问题，问题在于它是个黑盒——你没法在"模型决定调工具"和"工具真正执行"之间插入任何逻辑。想加个权限校验？改源码。想加个 token 计数？改源码。想在某类工具调用前先做一次缓存查询？还是改源码。

DeepAgents 的思路是把这条执行链显式地拆成若干阶段，每个阶段暴露钩子（hook），你用createMiddleware注册自己的逻辑。我画不出图（这里也不打算用图表），但你可以脑补一条流水线：

用户输入 → [前置中间件] → 模型推理 → [工具调用前中间件] → 工具执行 → [工具调用后中间件] → 模型再推理 → ... → [后置中间件] → 最终输出

每一层中间件都能拿到当前的state（对话历史、已调用工具、token 消耗等），也能决定是continue（继续）、modify（改状态后继续）还是halt（中断整个流程）。这个设计直接解决了我前面说的三个痛点：流程失控可以用状态机约束、重复调用可以用去重中间件拦截、token 爆炸可以用预算中间件提前熔断。

2.2 和 LangChain、Dify、CrewAI 的定位差异

很多人会问：既然有 LangChain 了，为什么还要 DeepAgents？既然有 Dify 这种可视化平台了，为什么还要写代码？

我的理解是这样的。LangChain 是"零件库"，它给你 LLM、Tool、Memory、Retriever 这些积木，但怎么搭、搭成什么样，全靠你自己。Dify 是"成品家具"，拖拖拽拽就能出一个能用的 Agent，但你想改个螺丝的材质都费劲。CrewAI 是"多 Agent 协作框架"，它擅长的是让几个角色分工合作，但对单个 Agent 内部的执行流程控制比较粗。

DeepAgents 卡在中间：它比 LangChain 高一层，给你现成的执行骨架和中间件机制；又比 Dify 低一层，所有逻辑都在代码里，想怎么改怎么改。如果你的 Agent 需要复杂的条件分支、严格的权限控制、精细的成本管理，DeepAgents 这种"代码优先 + 中间件可插拔"的路子会比可视化平台更合适。

至于"基于 Rust 语言的 AI Agent"这个热搜词，我得说句实话：Rust 写 Agent 在性能和并发上确实有优势，但生态成熟度跟 Python 差得远。DeepAgents 目前是 Python 生态的东西，如果你团队没有强 Rust 背景，别为了追新而追新。

2.3 中间件的核心抽象：createMiddleware 到底做了什么

createMiddleware是 DeepAgents 里最核心的 API，它的签名大概长这样（我按常见实践补全，具体以官方文档为准）：

from deepagents import createMiddleware my_middleware = createMiddleware( name="token_budget_guard", before_model=lambda state: check_budget(state), before_tool=lambda state, tool_call: validate_tool(state, tool_call), after_tool=lambda state, result: record_usage(state, result), after_model=lambda state, output: finalize(state, output), )

四个钩子对应执行链的四个关键节点。before_model在每次调用模型前触发，适合做预算检查、上下文裁剪；before_tool在工具执行前触发，适合做权限校验、参数清洗、缓存命中判断；after_tool在工具返回后触发，适合做结果过滤、用量统计；after_model在模型输出最终答案后触发，适合做格式化、敏感词过滤、日志落盘。

关键在于每个钩子都能返回一个控制指令。返回None或Continue表示放行；返回Modify(new_state)表示改完状态继续；返回Halt(reason)表示直接中断。这个设计让中间件既能"观察"也能"干预"，比单纯的 callback 强太多。

3. 核心细节解析与实操要点

3.1 状态对象 state 里到底有什么

中间件能不能写好，取决于你对state的理解够不够深。根据我的使用经验，state通常包含这几类信息：

• messages：完整的对话历史，包括用户输入、模型输出、工具调用记录。这是最占 token 的部分，也是上下文裁剪中间件的主要操作对象。
• tool_calls：本次会话已执行的工具调用列表，每条包含工具名、参数、结果、耗时。做去重和限流全靠它。
• usage：token 消耗统计，分 input / output 两块。做预算熔断的核心依据。
• scratchpad：Agent 的"草稿纸"，模型可以在里面记中间结论。这个字段容易被忽略，但在复杂推理任务里非常有用。
• metadata：自定义元数据，你可以往里塞任何东西，比如用户 ID、会话 ID、业务标签。

我踩过的一个坑是：早期我直接在中间件里改messages，结果把工具调用的配对关系搞乱了。LangChain 对消息格式有严格要求，工具调用消息和工具结果消息必须成对出现，你删一个留一个，模型直接报错。正确做法是用官方提供的trim_messages工具函数，或者自己写裁剪逻辑时严格保证配对。

3.2 中间件的执行顺序与优先级

多个中间件同时注册时，执行顺序很关键。DeepAgents 一般按注册顺序执行before_*钩子，按逆序执行after_*钩子——这跟 Web 中间件的洋葱模型是一个道理。

假设你注册了 A、B、C 三个中间件，执行顺序是：

A.before_model → B.before_model → C.before_model → 模型推理 → C.after_model → B.after_model → A.after_model

这个顺序意味着：越早注册的中间件，越"外层"，越晚注册的越"内层"。所以权限校验、预算熔断这类"守门员"逻辑应该注册在最前面；日志、埋点这类"记录员"逻辑可以注册在最后面。

注意：如果你的中间件之间有依赖关系（比如 B 依赖 A 修改后的 state），一定要确认执行顺序符合预期。我见过有人把预算检查注册在日志中间件后面，结果日志里记的 token 数是熔断前的，对不上账。

3.3 工具调用的拦截与改写

before_tool钩子是整个中间件体系里最有价值的部分，因为它能在工具真正执行前做文章。我常用的几个套路：

权限校验：根据用户角色决定某个工具能不能调。比如普通用户不能调refund_order，只有客服角色可以。这个逻辑放在 Prompt 里让模型自己判断是不可靠的，模型会"心软"，必须用代码硬拦。

参数清洗：模型生成的工具参数经常有格式问题，比如日期格式不统一、金额带了货币符号。在before_tool里统一清洗，比在每个工具函数里各写一遍强。

缓存命中：如果同样的工具调用在最近 N 分钟内出现过，直接返回缓存结果，不真正执行。这对查询类工具（查订单、查物流）效果特别明显，能省下大量 API 调用和 token。

限流熔断：统计单个会话的工具调用次数，超过阈值就Halt。我设的阈值是 20 次，超过基本可以判定 Agent 陷入了死循环。

3.4 中间件的错误处理与降级

中间件本身也会出错。比如你调外部缓存服务，缓存挂了怎么办？我的原则是：中间件出错不能拖垮整个 Agent。

具体做法是给每个钩子包一层 try-except，出错时记录日志并返回Continue（放行），而不是让异常往上抛。除非是安全相关的中间件（比如权限校验），那种情况下出错应该Halt（拒绝），宁可误杀不可放过。

def safe_before_tool(state, tool_call): try: return validate_tool(state, tool_call) except Exception as e: logger.error(f"middleware error: {e}") return Continue() # 降级放行

这个模式我用了大半年，救过好几次场。有一次缓存服务抽风，如果没有这层降级，整个客服系统会全线不可用。

4. 实操过程与核心环节实现

4.1 环境准备与依赖安装

先把环境搭起来。我用的 Python 3.11，太老的版本有些异步特性支持不好。

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install deepagents langchain langchain-openai

如果你要用 Redis 做中间件的状态存储（热搜词里提到"redis 做中间件"，这个思路是对的），再装一个：

pip install redis

提示：DeepAgents 的版本迭代比较快，建议在 requirements.txt 里锁死版本号，别用>=，不然某天自动升级后 API 变了你会很懵。

4.2 第一个中间件：Token 预算守卫

我们从最实用的开始。这个中间件的目标是：当单次会话的 token 消耗超过预算时，强制中断并返回友好提示。

from deepagents import createMiddleware, Continue, Halt TOKEN_BUDGET = 50000 def check_budget(state): used = state.usage.get("total_tokens", 0) if used > TOKEN_BUDGET: return Halt(reason=f"token 预算超限：已用 {used}，上限 {TOKEN_BUDGET}") return Continue() budget_guard = createMiddleware( name="budget_guard", before_model=check_budget, )

这里有个细节：预算检查放在before_model而不是before_tool。因为 token 主要消耗在模型推理上，工具调用本身不烧 token（除非工具内部又调了模型）。放在模型调用前检查，能在下一次推理发生前就拦住。

预算值怎么定？我的经验是：简单问答类 Agent 给 10000 就够，带多轮工具调用的复杂 Agent 给 50000，涉及长文档处理的给 100000。别一上来就给很大，先跑一周看实际分布，再定阈值。

4.3 第二个中间件：工具调用去重

这个中间件解决"Agent 反复调同一个工具"的问题。

from collections import Counter def dedup_tool(state, tool_call): key = f"{tool_call.name}:{hash(str(tool_call.args))}" history = state.metadata.get("tool_call_history", []) count = history.count(key) if count >= 2: return Halt(reason=f"工具 {tool_call.name} 重复调用超过 2 次，疑似死循环") history.append(key) state.metadata["tool_call_history"] = history return Continue() dedup = createMiddleware( name="tool_dedup", before_tool=dedup_tool, )

注意我用的是hash(str(args))而不是直接比 args 对象，因为字典的哈希需要转成可哈希类型。这个写法有个小坑：如果参数里包含顺序不同的键值对，哈希会不一样。更严谨的做法是先把字典按 key 排序再序列化。

阈值设 2 还是 3？我建议设 2。因为正常的 Agent 流程里，同一个工具用相同参数调两次已经很少见了，第三次基本可以确定是循环。设太宽松起不到保护作用。

4.4 第三个中间件：Redis 缓存命中

查询类工具是缓存的重灾区。查订单、查物流、查商品详情，这些数据在短时间内不会变，完全可以缓存。

import redis import json r = redis.Redis(host="localhost", port=6379, db=0) def cache_lookup(state, tool_call): if tool_call.name not in ["query_order", "query_logistics"]: return Continue() cache_key = f"tool:{tool_call.name}:{json.dumps(tool_call.args, sort_keys=True)}" cached = r.get(cache_key) if cached: return Modify(state, inject_tool_result=json.loads(cached)) return Continue() def cache_store(state, result): if result.tool_name in ["query_order", "query_logistics"]: cache_key = f"tool:{result.tool_name}:{json.dumps(result.args, sort_keys=True)}" r.setex(cache_key, 300, json.dumps(result.output)) # 缓存 5 分钟 return Continue() cache_mw = createMiddleware( name="redis_cache", before_tool=cache_lookup, after_tool=cache_store, )

这里Modify的inject_tool_result参数是我按常见实践补的，具体 API 名以官方为准。核心思路是：命中缓存时，跳过真实工具执行，直接把缓存结果注入 state。

TTL 设 5 分钟是个经验值。订单状态变化不会太频繁，5 分钟足够覆盖大部分重复查询，又不会让数据太陈旧。物流信息变化快一点，可以单独设 60 秒。

4.5 把中间件组装起来

from deepagents import createAgent agent = createAgent( model="gpt-4o", tools=[query_order, query_logistics, refund_order, send_message], middleware=[budget_guard, dedup, cache_mw], ) result = agent.invoke({"messages": [{"role": "user", "content": "帮我查下订单 12345 的物流"}]})

注册顺序有讲究：budget_guard放最前面，因为它是全局守门员；dedup其次，防止循环；cache_mw最后，因为它只关心特定工具。这个顺序下，预算检查最先执行，缓存查询最后执行，逻辑上最合理。

4.6 参数计算：预算阈值到底怎么定

很多人问我预算阈值怎么算。我给个可复用的方法：

先跑 100 次真实请求，记录每次的 token 消耗，算出 P50、P90、P99 三个分位数。假设结果是 P50=8000、P90=25000、P99=60000。那么阈值应该设在P99 略高一点的位置，比如 70000。这样能拦住那 1% 的异常请求，又不会误杀正常的长尾请求。

如果你设成 P90=25000，那 10% 的正常请求会被误杀，用户体验会很差。设成 P99 的 1.2 倍是比较稳妥的做法。

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

5.1 中间件不生效？先查这三个地方

我遇到过好几次"中间件写了但没反应"的情况，排查下来基本是这三类问题：

注册顺序错了：中间件必须传给createAgent的middleware参数，不是传给 Tool 或 Model。我见过有人把中间件塞进tools列表里，那当然不生效。

钩子名拼错了：before_model、before_tool、after_tool、after_model，这四个名字必须完全一致。Python 不会报错，只会静默忽略。

返回值类型不对：钩子必须返回Continue、Modify或Halt对象，返回None或True在某些版本里会被当成"无操作"。这个坑很隐蔽，建议每个钩子都显式返回。

5.2 常见问题速查表

5.3 独家避坑技巧

技巧一：给中间件加"干跑模式"。新写一个中间件时，先让它只记录日志不真正干预，跑一周看日志，确认逻辑符合预期后再开启干预。我有个权限中间件就是这么上线的，干跑阶段发现它误判了 30% 的请求，如果直接上线会拦掉大量正常业务。

技巧二：中间件的日志要带 trace_id。Agent 一次请求会触发多次中间件，没有 trace_id 你根本串不起来。我在每个中间件入口都打一行logger.info(f"[{trace_id}] {middleware_name} triggered")，排查问题时一目了然。

技巧三：Halt 的 reason 要写清楚。Agent 被中断后，用户看到的是 reason 内容。写"预算超限"比写"error 500"友好一万倍。我甚至会在 reason 里带上建议，比如"预算超限，请简化问题后重试"。

技巧四：别在中间件里做重活。中间件是同步执行在 Agent 主流程里的，你在里面调个慢接口，整个 Agent 就卡住了。缓存查询、日志落盘这类操作要么用异步，要么用本地内存缓存兜底。

技巧五：中间件要能单独测试。把每个中间件的钩子函数写成纯函数，输入 state 输出指令，这样你可以脱离 Agent 单独写单元测试。我现在的中间件测试覆盖率都在 80% 以上，改起来心里有底。

6. 中间件的扩展玩法与进阶思路

6.1 用中间件做 A/B 测试

这个玩法我是从 Web 开发那边借鉴过来的。你可以写一个中间件，根据用户 ID 的哈希值决定走哪套 Prompt 或哪套工具集，然后对比两组的成功率、token 消耗、用户满意度。

def ab_test(state): user_id = state.metadata.get("user_id", "") group = "A" if hash(user_id) % 2 == 0 else "B" state.metadata["ab_group"] = group if group == "B": return Modify(state, system_prompt=EXPERIMENTAL_PROMPT) return Continue()

这个中间件注册在最外层，后续所有逻辑都能读到ab_group，日志里也能按组统计。比在业务代码里到处埋 if-else 优雅多了。

6.2 用中间件做敏感信息过滤

Agent 输出给用户之前，过一遍敏感信息过滤中间件。这个在客服、教育类场景里是刚需。

SENSITIVE_PATTERNS = [r"\d{18}", r"\d{11}"] # 身份证、手机号 def filter_output(state, output): text = output.content for pattern in SENSITIVE_PATTERNS: text = re.sub(pattern, "***", text) return Modify(state, content=text)

放在after_model钩子里，模型输出后、返回用户前执行。注意别过滤太狠，把订单号也当成手机号给屏蔽了，那就闹笑话了。

6.3 中间件与可观测性

生产环境的 Agent 必须可观测。我一般会加一个"埋点中间件"，把每次模型调用、工具调用的耗时、token、成功率都打到监控系统里。

import time def track_model(state): state.metadata["model_start"] = time.time() return Continue() def track_model_end(state, output): duration = time.time() - state.metadata["model_start"] metrics.record("model_latency", duration) metrics.record("model_tokens", output.usage.total_tokens) return Continue()

这些指标积累下来，你就能回答"Agent 到底慢在哪""token 都花在哪"这类问题。没有这些数据，优化就是瞎猜。

6.4 关于"AI Agent 学习路线"的一点个人建议

热搜里有人问 AI Agent 学习路线，我借这个地方说两句。我的建议是：别一上来就啃框架源码，先手写一个最朴素的 Agent 循环。就是用 while 循环 + 模型调用 + 工具执行，跑通一个能查天气的 Agent。这个过程能让你真正理解 Agent 的本质是"模型 + 工具 + 循环"。

然后你再去看 LangChain、DeepAgents 这些框架，就会发现它们做的事情无非是把循环标准化、把扩展点抽象出来。这时候你学中间件、学状态管理，就是水到渠成的事。反过来，如果你连基础循环都没写过，直接看框架文档，很容易被各种概念绕晕。

至于"AI Agent 部署"和"用 AI Agent 开发 Django"这类话题，核心还是先把单机跑通，再考虑容器化、水平扩展、状态外置这些工程问题。别本末倒置。

7. 我在实际项目里的一些体会

DeepAgents 中间件这套东西，我用了大概半年，最大的感受是：它把 Agent 开发从"调 Prompt 的玄学"拉回到了"写代码的工程"。以前优化 Agent 靠反复改 Prompt、祈祷模型听话；现在我可以精确地控制每一步执行，出了问题能定位、能复现、能修复。

但我也得说句公道话：中间件不是银弹。如果你的 Agent 逻辑很简单，就一两个工具、没有复杂分支，那用中间件反而是过度设计，直接写 LangChain 更省事。中间件的价值在复杂场景下才体现得出来——多工具、多轮次、有权限、有成本约束、需要可观测。

最后分享一个小技巧：把中间件当成"可复用的业务规则库"来积累。我现在的项目里，预算守卫、去重、缓存、权限、埋点这几个中间件已经成了标配，新项目直接复制过去改改参数就能用。这种积累带来的复利，比每次从零写 Agent 强太多了。