我这两年只要一聊 AI Agent 项目被问得最频繁的问题几乎都是同一个——你的 Agent 能扛多少并发为什么一到线上就卡死其实很多团队做 Agent 能力的时候模型选型、Prompt 调优都做得挺到位结果一到多用户场景就直接崩盘问题几乎都出在缺少一层像样的中间件。DeepAgents 中间件就是在这样的背景下被逼出来的它不是某个具体的模型也不是一套 UI 框架而是专门负责 Agent 的运行时调度、状态持久化、工具编排和并发托底的一层基础设施。这篇文章我想把 DeepAgents 的设计思路、核心模块、实战落地步骤以及我踩过的坑全部摊开来讲尤其适合那些已经跑通 Demo、准备往生产环境推的 AI 应用开发者。1. 为什么 Agent 项目必须有中间件很多人一开始会觉得Agent 不就是大模型 工具调用 循环判断吗最多加个提示词能复杂到哪里去。但等你真正把一个 Agent 暴露给几十上百个用户同时使用时你面对的根本不是一个模型调用的问题而是一整套分布式系统问题。1.1 多 Agent 协作带来的状态管理难题一个真实的业务 Agent 往往不是单打独斗而是拆成多个子 Agent 分工协作。比如我做过的一个客服场景一个主 Agent 负责理解用户意图然后分发给售后 Agent、订单 Agent、物流 Agent 去分别处理。每个子 Agent 都有自己的上下文、中间结果和运行状态。如果没有一个统一的中间层去存这些状态每一次子 Agent 调用之间的上下文就只能靠内存里的全局变量硬顶一个进程重启或扩容所有会话直接断掉。这种情况在单体 Demo 里完全看不出来问题因为你所有的任务都串行跑在一个进程里。但一旦并发上来多个用户的会话交错执行如果没有中间件统一管理会话快照和任务状态你会在日志里看到一堆找不到上下文重试后状态丢失的诡异报错。1.2 Agent 天生需要异步化和可恢复性LLM 的响应本身是不稳定的耗时可长可短你不可能让用户请求一直阻塞在一个 HTTP 调用上。更现实的做法是请求进来之后立刻返回一个任务 IDAgent 在后台异步执行执行完通过回调或轮询把结果交给用户。这个思路其实很像 Web 后端里的消息队列但 Agent 比普通任务又特殊很多——它每一步都可能调用外部 API、等待模型返回、触发一个新的子任务执行链路特别长。中间件在这时候要做的就是把运行中等待模型等待工具失败重试已完成这些状态完整接住保证任何一步断了都能从最近的检查点恢复。1.3 工具编排必须集中管理Agent 的能力边界取决于它挂了哪些工具。今天接一个搜索 API明天加一个数据库查询后天接一个内部工单系统如果没有中间件做统一的工具注册和路由所有工具调用逻辑会散落在各个 Agent 的代码里连排查一个某个工具为什么没被触发都要翻半天代码。用中间件做工具编排还有一个额外的好处可以给每个工具定义 QoS 策略比如限流、熔断、超时降级。这在直接调工具函数时是完全做不到的因为你没有统一出口去控制这些行为。2. DeepAgents 的核心设计思路DeepAgents 本质上是一个面向 AI Agent 场景的中间件层它介于应用代码和大模型、外部工具之间。一开始我在选技术路线时参考了不少主流方案包括 Spring AI Agent 的生态、Rust 系的高性能运行时、以及 FastAPI LangChain 的轻量组合最终这套中间件选择了模块化运行时 可插拔状态存储 统一网关的架构。2.1 架构选型的底层逻辑技术选型上Spring AI Agent 的优势是你如果已经踩在 Java/Spring 的坑里整个团队迁移成本低但问题是它的生态相对更偏传统企业级很多轻快的 Agent 编排能力你得自己补。Rust 语言做 Agent 的亮点非常突出内存安全、并发性能极佳我之前用 Rust 写过一版核心调度器单机扛 QPS 确实比 Python 高一个数量级但问题是团队维护成本太高业务功能迭代速度会被严重的类型系统拖住。最终我选择了折中方案核心调度与协议层用 FastAPI 实现业务扩展逻辑走 LangGraph状态存储和并发控制交给 Redis然后把整个调度环境封装成 DeepAgents 中间件。FastAPI 的优势在于异步模型很干净跟 LangGraph 这种基于图编排的库能比较好地接合。LangGraph 则擅长表达Agent 在什么条件下走哪条分支、该调用什么工具这种逻辑比我自己手写状态机省太多力气。2.2 中间件到底接管了哪些事DeepAgents 内部把职责拆成了四大块。第一块是任务调度层负责接收请求、生成任务 ID、按优先级和负载把任务分配给工作线程或子 Agent。第二块是状态管理层用 Redis 保存每个会话的检查点、上下文窗口、已执行步骤的记录。第三块是工具网关层所有 Agent 调用的外部工具都必须经过这层代理统一在这里做鉴权、限流、超时控制和结果格式化。第四块是可观测层负责记录每一个 Agent 的完整轨迹包括每一步的输入输出、Token 消耗、耗时和失败原因。这套拆分带来的最直接效果是业务代码变得特别笨笨到只需要关心我这一步该干什么而不用操心我这一步失败了怎么恢复要不要重试别人怎么知道我卡住了。2.3 和消息队列中间件的本质区别有人会问Redis、Kafka 这些本来就是中间件DeepAgents 和它们什么关系打个比方Kafka 是高速公路负责把数据从一个地方运到另一个地方它不关心运的是什么。DeepAgents 更像是物流调度中心它决定包裹怎么分拣、哪辆车走哪条路、丢了怎么补发。所以在 DeepAgents 内部Redis 被用来做实时状态存储RabbitMQ 或 Kafka 负责解耦耗时任务而 DeepAgents 本身是更上层的那套调度大脑。3. 深入拆解核心模块从并发到状态再到工具链路到了这一节我想把 DeepAgents 中间件里的几个关键模块掰开揉碎来讲。可能你已经看了不少 AI Agent 的文章讲 Prompt、讲模型选型的多把并发、状态、限流这些工程细节讲透的少而恰恰是这些细节决定了你的 Agent 能不能真正上线。3.1 并发模型与全局任务队列AI Agent 的并发问题比普通 Web 接口复杂得多核心原因是它的每个请求耗时可长可短、资源消耗不均匀。你没法像处理普通 API 一样简单地用每秒钟请求数来衡量压力一个需要走 20 步工具调用的复杂任务和一个只需要一次模型调用的简单任务对系统的压力差别可能有几十倍。DeepAgents 的做法是引入两层队列。第一层是接入队列所有用户请求进来后先入队由调度器按会话优先级和工作线程的空闲度进行分发。第二层是内部步骤队列每个 Agent 运行过程中的每一步包括模型调用、工具调用、子 Agent 下发都被拆成一个独立的消息进入 Redis Stream 里由工作进程消费。这套模型最大的好处是精细到步骤级别的并发控制。假设你在做一个期货交易分析 Agent风控类的任务必须优先执行普通的数据汇总任务可以往后排。通过给不同的步骤标记不同的优先级调度器可以在系统繁忙时做到普通任务让路、关键任务插队这个能力在单体代码里几乎没法实现。在实践里我用过一个不错的优化给每个会话设置最大并发步数。单个用户的任务在系统里最多同时占用 2 个步骤槽位防止一个复杂任务把整个线程池堵死。这个值不能设得太低否则复杂 Agent 的完成时间会显著拉长但也不能太高否则一旦某个用户触发了一个失控循环其他人的任务会被拖死。3.2 会话级状态持久化与检查点恢复Agent 系统一个最容易被忽视的问题是上下文到底放在哪里。很多 Demo 把历史对话记录放在一个列表里整个 Agent 跑完就算完事。但生产环境完全不同一次 Agent 任务可能跑几分钟用户中途可能断线服务可能重启模型可能超时。如果没有持久化的会话状态失败恢复就是空谈。DeepAgents 把检查点分为两个维度会话级检查点和步骤级检查点。会话级检查点存的是用户和多轮 Agent 交互的完整上下文包括已经确定的用户意图、已经获取到的关键信息、当前所处的业务阶段这个维度用 Redis Hash 结构来存Key 是 session_idField 是上下文中的各个属性支持随时更新和回滚。步骤级检查点更细粒度它记录当前步骤调用了哪个工具、传了什么参数、拿到了什么结果、模型返回了什么内容。Redis 在这里有个坑就是单个 Key 的 Value 太大会带来严重的内存和网络开销。一个长会话经过几十轮工具调用之后序列化的 JSON 可能上来就几十 KB再叠加并发量Redis 会先撑不住。我的处理方案是设置检查点的最大容量超出后丢弃较早的细节步骤只保留业务关键属性比如用户 ID、业务单号、当前状态、最近三步的结果。这样既不会丢失核心上下文又能保证 Redis 的读写性能。检查点还有一个被很多人忽略的用途审计回溯。用户如果质疑你上次为啥给我推这个结果你可以直接拿当时的检查点重放一遍完整轨迹逐条展示当时的决策依据。这个能力在我们做金融场景时特别重要合规要求每一步的决策都能追溯。3.3 工具执行网关与依赖注入传统 Agent 开发里工具函数直接注册进模型可调用的列表就行调用逻辑散落在各个模块。DeepAgents 把工具调用收敛成了一个网关任何工具要暴露给 Agent必须先注册到这个网关上。每个工具注册的时候需要声明的内容包括工具名称、参数 Schema、超时时间、最大并发数、失败策略重试/跳过/终止、调用是否需要鉴权。网关拿着这些元信息在每次调用前先做一次并发检查如果该工具当前并发已经打满请求直接进入等待队列而不是强行触发一次注定失败的调用。举一个具体例子我之前对接过一个小红书自动发消息的场景。平台接口的限流特别严格每秒钟最多 5 次请求一旦超了就会被封禁。在工具网关里我把这个工具的最大并发数设为 3把失败策略设为重试并退避。实测下来即使主 Agent 疯狂触发调用网关也会把多余的请求挡在外面平台侧一次都没触发限流。工具网关还有一层容易被忽略的作用它是模型输出和实际系统之间的防弹衣。模型的输出偶尔会给你编一个根本不存在的工具名或者传一个格式完全错误的参数。网关可以做一次参数强校验不合法直接返回一条工具调用格式错误请重新组织你的调用的提示给模型让模型自我修正。这一步能避免大量因为模型幻觉导致的执行错误。3.4 Token 消耗的统计、预算控制与成本火警Token 成本是 AI Agent 上线前最容易被低估的一笔账。一个简单的用户请求看起来只是一次模型调用但 Agent 内部为了拆解意图、调用工具、汇总结果可能偷偷消耗了几倍的 Token。如果不加控制月底账单会非常难看。DeepAgents 在中间件层统一做了 Token 计量。每一次模型调用的 prompt 和 completion 的 Token 数都会被记入会话账本同时按用户维度、按 Agent 维度、按工具维度三张大表汇总。基于这套数据我实现了三级成本控制第一级是软提醒某个会话 Token 消耗超过阈值在日志里打一个警告第二级是硬限制超过预算后这个 Agent 实例直接拒绝继续执行返回本次对话预算已用完第三级是熔断某个用户或某个 Agent 的成本异常飙升全局自动切断新任务。这个设置救过我一次。有一次 QA 的测试用例里带了一个死循环倾向的 Prompt让 Agent 反复调用同一个工具如果没有预算熔断那一晚上能把测试账号的额度烧穿。上线这些策略之后第二天一看账单异常会话在触发熔断前就被拦住了损失几乎为零。对于任何一个准备长期运营 Agent 服务的团队来说Token 预算控制系统绝对要优先做。3.5 可观测性用全链路追踪还原 Agent 的每一步决策Python 生态里排查普通 Web 问题可以靠日志和 trace但 Agent 问题特殊在哪特殊在它的每一个步骤都涉及和模型、工具的交互错误出现在任何一环都可能被 LLM 的自动修复逻辑掩盖最终留给你的是一个看起来不太对劲的最终答案。DeepAgents 的可观测层为每个任务生成一个全局唯一的 trace_id从任务进入系统开始到任务完成或失败结束每一步的输入和输出都打点记录。记录的字段包括当前步骤在执行什么操作、调用了哪个模型还是哪个工具、模型返回的原始内容摘要、工具调用的完整参数和响应、步骤耗时、累计 Token 消耗以及每一步的决策依据。有一次业务方反馈某个 Agent 经常给出自相矛盾的回答。我直接打开对应的 trace发现模型在第三步已经获得了用户所在地区为 A但到第七步时由于上下文窗口被新内容挤占模型忘了这个信息重新推断了一个错误地区。基于这个 trace我调整了上下文压缩策略把关键用户属性固定在不可被压缩的头部区域问题立刻解决。如果没有全链路追踪这种偶发问题可能要抓瞎好几天。4. 实操落地用 FastAPI LangGraph 复刻一个 DeepAgents 最小可用版理论说了那么多不落地就是空中楼阁。我把 DeepAgents 的一个简化版本跑通的全过程记录下来技术栈是 FastAPI 加 LangGraph 加 Redis。这个版本麻雀虽小但调度、状态、工具网关、Token 统计都齐了你可以直接照着改。4.1 准备工作与目录规划建议你提前准备一个 Python 3.11 以上的环境安装以下依赖fastapi、uvicorn、langgraph、redis、pydantic。先建目录结构deepagents-mini/ ├── main.py # FastAPI 入口与任务提交接口 ├── scheduler.py # 任务调度接入队列 状态管理 ├── graph.py # LangGraph 图定义Agent 主流程 ├── tools.py # 工具定义与网关注册 ├── redis_client.py # Redis 连接与检查点读写 └── config.py # 全局配置并发数、Token 限额、超时时间目录规划这件事看着不起眼实际对后期排查很有帮助。我见过太多项目把所有逻辑塞进一个 main.pyAgent 的图定义和工具注册混在一起并发一上来根本没法定位问题。从第一天就按职责分文件会省掉后面大量重构成本。4.2 工具网关实现先挡住超限请求先写一个最简单的工具网关。核心思路很简单每个工具进来时先查一下 Redis 里的当前并发计数超过限制就抛异常等待重试。这里用 Redis 的 INCR 命令配合 EXPIRE 实现一个原子的滑动窗口计数import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) class ToolRateLimiter: def __init__(self, tool_name: str, max_concurrency: int, window_seconds: int 10): self.key ftool:qps:{tool_name} self.max_concurrency max_concurrency self.window_seconds window_seconds def acquire(self) - bool: current r.incr(self.key) if current 1: r.expire(self.key, self.window_seconds) if current self.max_concurrency: return True r.decr(self.key) return False注意这里我没有用锁因为 Redis 的 INCR 本来就是原子的在多进程环境下也能保证计数准确。这个限流器可以挡住短时间内的超额调用但如果你需要更平滑的令牌桶算法可以换成 Redis 加 Lua 脚本实现思路类似。我在生产环境用的就是 Lua 版令牌桶效果会更稳。工具注册表用字典维护。每个工具先注册元信息再注册实际执行函数registry {} def register_tool(name: str, timeout: int 10, max_concurrency: int 5): def decorator(func): registry[name] { name: name, func: func, timeout: timeout, limiter: ToolRateLimiter(name, max_concurrency), } return func return decorator register_tool(query_database, timeout15, max_concurrency10) def query_database(sql: str) - dict: # 模拟查询 return {result: fexecuted: {sql}} register_tool(send_message, timeout5, max_concurrency3) def send_message(phone: str, content: str) - dict: # 模拟发送 return {status: sent, to: phone}网关的调用方法统一接收工具名和参数字典先从注册表读取配置做限流和超时控制再执行。这一步很重要的原因是让调用工具这件事变成一个可以统一拦截的出口后续加日志、加鉴权、加熔断都只需要改这一处。4.3 LangGraph 图编排把 Agent 流程变成有向图LangGraph 这个名字听上去很唬人实际上它的核心逻辑就是帮你定义一个状态图每个节点执行一段操作根据条件决定下一步走向哪个节点。DeepAgents 的主流程我用三个节点搞定分析意图节点、执行工具节点、生成回答节点。from typing import TypedDict, Optional from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list user_intent: Optional[str] tool_result: Optional[dict] final_answer: str def analyze_intent(state: AgentState) - AgentState: # 实际项目里这里会调用大模型 intent query_database # 模拟模型判断结果 return {user_intent: intent} def execute_tool(state: AgentState) - AgentState: tool_name state.get(user_intent, ) tool_meta registry.get(tool_name) if not tool_meta: state[tool_result] {error: tool not found} return state if not tool_meta[limiter].acquire(): state[tool_result] {error: tool concurrency limit reached} return state state[tool_result] tool_meta[func](sqlselect 1) return state def generate_answer(state: AgentState) - AgentState: result state.get(tool_result, {}) state[final_answer] f查询结果如下: {result} return state def build_graph(): g StateGraph(AgentState) g.add_node(analyze, analyze_intent) g.add_node(execute, execute_tool) g.add_node(answer, generate_answer) g.add_edge(START, analyze) g.add_conditional_edge( analyze, lambda state: execute if state.get(user_intent) in registry else answer, ) g.add_edge(execute, answer) g.add_edge(answer, END) return g.compile()这里有一个我特别想强调的设计点条件边的判断不能只看意图字符串还要判断工具是否真的存在。否则模型一旦产生幻觉返回一个不在注册表里的工具名图执行会直接报错。上面加了if state.get(user_intent) in registry的判断就是为了兜住这种幻觉情况。4.4 调度器的接入和状态保存调用方通过 FastAPI 接口提交任务调度器把任务压入 Redis Stream然后立即返回 task_id。工作进程从 Stream 里消费任务执行 LangGraph 图并把最终结果存入 Redis。import uuid from fastapi import FastAPI from redis import Redis app FastAPI() r Redis(hostlocalhost, port6379, decode_responsesTrue) app.post(/agent/task) async def submit_task(user_message: str): task_id uuid.uuid4().hex task_data {task_id: task_id, message: user_message} r.xadd(agent:task_queue, task_data) return {task_id: task_id, status: queued} def worker(): while True: entries r.xread({agent:task_queue: 0}, count1, block5000) if not entries: continue _, messages entries[0] for msg_id, data in messages: task_id data[task_id] message data[message] graph build_graph() result graph.invoke({messages: [message]}) r.hset(fagent:result:{task_id}, mappingresult) r.xdel(agent:task_queue, msg_id)这种提交后异步执行的模式要求前端必须支持轮询或 WebSocket。我在实际项目里用的方案是任务完成后把结果写到一个 Redis List前端通过一个轻量 SSE 接口拿结果。你如果不想自己写也可以用 WebSocket 推送原理一样。4.5 状态检查点的写入策略真正生产级的中间件会把检查点直接集成到 LangGraph 的每次节点执行之间。LangGraph 官方支持 checkpointer 参数你可以传入一个自定义的 BaseCheckpointSaver。我这里给一个极简实现核心逻辑就是在 Redis 里以会话为单位存一个 JSONclass RedisCheckpointSaver: def __init__(self, redis_client): self.r redis_client def save(self, session_id: str, state: dict): key fagent:checkpoint:{session_id} self.r.set(key, json.dumps(state, ensure_asciiFalse)) self.r.expire(key, 3600 * 24) def load(self, session_id: str) - dict: key fagent:checkpoint:{session_id} data self.r.get(key) if not data: return {} return json.loads(data)一个比较反直觉的经验是检查点不能保存得太过频繁。如果每执行一个节点就做一次全量持久化当会话状态较大时 Redis 的写入会成为新的瓶颈。我建议只在满足以下条件之一时才落盘状态发生了结构性的变化比如意图被重新识别、关键信息字段被更新、执行进入了一个子 Agent 的边界、或者任务进入等待外部回调的阶段。4.6 Token 预算控制的小型实现Token 预算控制在最小版本里用两个字段实现当前会话已消耗 Token 数和预算上限。每次调用模型前先从 Redis 读取当前消耗如果超过了就直接短路不让模型调用发生。def check_token_budget(session_id: str, limit: int 100000, estimated_tokens: int 1000) - bool: key fagent:token_usage:{session_id} current int(r.get(key) or 0) if current estimated_tokens limit: return False r.incrby(key, estimated_tokens) return True这个方案里估计的 Token 数是一个粗粒度值实际项目里可以调用模型供应商的 Token 统计接口做到精确计量。不过对于预算控制这个目标来说粗粒度已经足够——你要的是别让异常任务烧光账单而不是精确到个位数的统计。真到了需要精确成本分摊的时候再用专门的计量模块接进去。5. 生产环境最容易踩的 5 个坑这部分全部是我自己实操中踩过的坑每一个都有对应的血泪教训。如果你准备把 DeepAgents 或者类似的中间件投到生产环境请一定仔细看完。5.1 死循环不是模型的问题是你的图没兜底Agent 最常见的线上事故就是死循环。模型的推理链路一旦陷入调用工具 - 结果不够满意 - 再调用工具的循环如果图里没有跳出机制整个工作线程会被无限占用最后拖垮所有任务。解决思路是在 LangGraph 图外面包一层循环计数器并在每次节点执行后递增。超过最大步数我一般设 20 步之后立即强制结束当前任务返回一条任务过于复杂请拆解后重试的提示给用户。这一条在低成本防事故清单里排名第一比任何限流都重要。5.2 步骤级重试会放大外部 API 的压力很多人做失败重试时习惯对整个 Agent 任务重跑一遍这在高并发下非常危险。假设你有个工具在高峰期超时了如果对整个任务做重试相当于把这个任务的 10 步工具调用全部重放一遍外部系统承受的压力会被放大好几倍。DeepAgents 的做法是只对失败的步骤做重试而且重试前要判断失败类型网络超时可以重试参数错误不要重试因为错误是确定的重试多少次都一样。我在工具网关里把异常分成了 RetryableException 和 NonRetryableException只有前者才走重试分支。5.3 Redis 连接池配置不当导致整个 Agent 系统变慢Agent 中间件对 Redis 的访问是高频的每个工具调用、每次状态检查、每个任务分发都要访问 Redis。如果连接池开太小Redis 的等待本身就会吃掉大量本该用于模型调用的时间。在一台 4 核 8G 的机器上我把 Redis 连接池的 max_connections 设在了 50 左右配合连接复用实测 QPS 可以从 200 多提 到 400 多。这里有个误区是连接池越大越好实际上连接过多会增加 Redis 服务端的上下文切换开销适中才是关键。建议压测时多试几组参数观察 P99 耗时。5.4 消息重复消费会造成业务重复执行Redis Stream 的消费语义是至少一次不是恰好一次。如果工作进程在处理任务时崩溃重启同一个消息可能被另一个进程重新消费Agent 的某些工具调用比如发消息、创建订单就会重复执行。我的处理方案是在每个任务的执行开始时先往 Redis 里写入一条任务开始执行当前会话 UUID 为 X的记录。工具执行时携带这个会话 UUID如果在工具网关发现同一个 UUID 之前已经处理过该消息就直接跳过。业务层面如果要求更强的幂等可以让工具接收方基于业务幂等键做去重。5.5 上下文窗口被打满导致 Agent失忆长会话场景下历史步骤的中间结果会不断堆积直到撑爆模型上下文窗口。此时如果不做压缩模型会丢掉最前面的关键信息表现出失忆症状回答质量明显下降。DeepAgents 的中间件层内置了一个简单的上下文压缩器当当前会话 Token 数超过窗口的 60% 时自动把最早的历史消息做摘要用一个更短的历史摘要替换掉原始内容。关键点是要把「用户核心诉求」和「业务关键属性」标记为不可压缩把它们一直保留在上下文中。这个策略在长会话场景下对维持模型一致性很有帮助。6. 从 Demo 到生产一套检查清单每次我帮团队评估一个 Agent 项目能不能上线都会拿一套自己总结的检查清单过一遍。这套清单不是理论而是被生产事故反复验证出来的硬性条件。第一并发控制是否做到步骤级别。如果你的 Agent 还是一次性把整个任务塞给线程池不知道当前系统里有多少个 Agent 在跑、每个 Agent 跑到哪一步那离生产还有距离。第二是否所有外部工具调用都经过统一的网关。散落的工具调用意味着你无法统一限流、链路追踪和异常处理。第三是否有会话级的状态持久化和恢复能力。没有这个能力服务重启等于所有用户会话归零。第四是否有 Token 预算硬限制。线上环境没人能保证模型不会失控预算熔断是最后一道防线。第五Trac e 链路是否完整记录了模型输入输出。出了问题如果不能在十分钟内定位到具体是哪个步骤排查效率会非常低。这五条不能打折。我见过太多精力旺盛的团队模型效果调得漂漂亮亮但工程底座四处漏风一上线就被流量打爆。反过来说只要这五条做到位哪怕模型能力稍微弱一点你也有足够的时间和空间去迭代优化。7. 后续扩展方向DeepAgents 这套中间件目前在我这里已经稳定跑了大半年接下来我打算在两个方向上继续深挖。一个方向是把调度器从单机版改成多机版用一致的哈希算法把不同会话固定分配到不同工作节点这样既支持水平扩容又能让同一个会话尽量落在同一台机器上减少跨节点的状态同步开销。另一个方向是给工具网关加一层模型驱动的参数纠错能力当模型传入的参数和工具 Schema 不匹配时不只做简单报错而是让一个小模型根据描述自动补全缺失参数降低整个人机协作链路的失败率。如果你也在做 AI Agent 相关的应用我强烈建议你尽早把中间件意识带入架构设计里。模型能力你可以慢慢调、慢慢换但工程底座一旦草率搭建后面每一次迭代都会为当初的图省事买单。踩过几次坑之后你会发现一个稳定、可观测、可恢复的中间件层才是 Agent 真正能下地干活的关键所在。