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( nametoken_budget_guard, before_modellambda state: check_budget(state), before_toollambda state, tool_call: validate_tool(state, tool_call), after_toollambda state, result: record_usage(state, result), after_modellambda 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本次会话已执行的工具调用列表每条包含工具名、参数、结果、耗时。做去重和限流全靠它。usagetoken 消耗统计分 input / output 两块。做预算熔断的核心依据。scratchpadAgent 的草稿纸模型可以在里面记中间结论。这个字段容易被忽略但在复杂推理任务里非常有用。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(fmiddleware 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(reasonftoken 预算超限已用 {used}上限 {TOKEN_BUDGET}) return Continue() budget_guard createMiddleware( namebudget_guard, before_modelcheck_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(reasonf工具 {tool_call.name} 重复调用超过 2 次疑似死循环) history.append(key) state.metadata[tool_call_history] history return Continue() dedup createMiddleware( nametool_dedup, before_tooldedup_tool, )注意我用的是hash(str(args))而不是直接比 args 对象因为字典的哈希需要转成可哈希类型。这个写法有个小坑如果参数里包含顺序不同的键值对哈希会不一样。更严谨的做法是先把字典按 key 排序再序列化。阈值设 2 还是 3我建议设 2。因为正常的 Agent 流程里同一个工具用相同参数调两次已经很少见了第三次基本可以确定是循环。设太宽松起不到保护作用。4.4 第三个中间件Redis 缓存命中查询类工具是缓存的重灾区。查订单、查物流、查商品详情这些数据在短时间内不会变完全可以缓存。import redis import json r redis.Redis(hostlocalhost, port6379, db0) def cache_lookup(state, tool_call): if tool_call.name not in [query_order, query_logistics]: return Continue() cache_key ftool:{tool_call.name}:{json.dumps(tool_call.args, sort_keysTrue)} cached r.get(cache_key) if cached: return Modify(state, inject_tool_resultjson.loads(cached)) return Continue() def cache_store(state, result): if result.tool_name in [query_order, query_logistics]: cache_key ftool:{result.tool_name}:{json.dumps(result.args, sort_keysTrue)} r.setex(cache_key, 300, json.dumps(result.output)) # 缓存 5 分钟 return Continue() cache_mw createMiddleware( nameredis_cache, before_toolcache_lookup, after_toolcache_store, )这里Modify的inject_tool_result参数是我按常见实践补的具体 API 名以官方为准。核心思路是命中缓存时跳过真实工具执行直接把缓存结果注入 state。TTL 设 5 分钟是个经验值。订单状态变化不会太频繁5 分钟足够覆盖大部分重复查询又不会让数据太陈旧。物流信息变化快一点可以单独设 60 秒。4.5 把中间件组装起来from deepagents import createAgent agent createAgent( modelgpt-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 三个分位数。假设结果是 P508000、P9025000、P9960000。那么阈值应该设在P99 略高一点的位置比如 70000。这样能拦住那 1% 的异常请求又不会误杀正常的长尾请求。如果你设成 P9025000那 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 常见问题速查表问题现象可能原因排查方向解决方案Agent 陷入死循环工具调用无去重看 tool_calls 列表加 dedup 中间件token 消耗异常高上下文未裁剪看 messages 长度加 trim 中间件工具参数格式错误模型输出不稳定看工具报错日志加参数清洗中间件中间件报错导致 Agent 崩溃未做异常捕获看中间件堆栈加 try-except 降级缓存不命中key 生成不一致打印 cache_key统一序列化方式权限校验失效中间件顺序错误看执行日志权限中间件前置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_promptEXPERIMENTAL_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, contenttext)放在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 强太多了。