企业官网建设流程全解析

1. 为什么Agent应用越做越需要一层"中间件"

1.1 三天能出Demo，三个月上不了线

我见过太多团队栽在同一个地方：用 LangChain 写一个能对话的 Agent 原型只需要三天，但真要把它放到生产环境里扛真实流量，三个月都搞不定。

问题不在于 Agent 本身的推理逻辑，而在于它周围那一圈"脏活累活"——多模型切换、上下文管理、Token 计费、限流降级、会话状态同步、并发控制。这些事单个拎出来都不难，但堆在一起，就会把业务代码搅成一锅粥。你本来只想写"用户问一句，Agent 答一句"，结果代码里全是请求重试、Token 截断、Redis 锁、队列消费的碎片逻辑。

DeepAgents 中间件就是冲着这个问题来的。它把 Agent 应用和大模型 API 之间所有共性横切问题收敛成一层独立服务，让上层业务只关心"这个 Agent 要干什么"，让下层模型只负责"根据上下文生成回答"。我在实际项目中维护这套中间件跑了将近一年，今天把架构设计、选型理由、踩坑记录全部摊开讲。

1.2 "中间件"卡的其实是这三件事

中间件听起来玄乎，落到真实场景就是三件事：请求怎么路由、状态怎么存、流量怎么控。

第一，请求路由。一个像样的 Agent 系统不会只接一家模型。GPT、Claude、国产开源模型，各有各的优势场景。中间件要做的就是把这些模型的 API 差异抹平，统一成一套内部接口，同时支持按策略切换和故障降级。

第二，会话状态。Agent 是有记忆的，但大模型接口本身是无状态的。用户的上下文、Agent 的推理历史、工具调用的中间结果，都得有地方存。存哪里、怎么隔离、怎么恢复，是中间件最核心的设计决策。

第三，流量控制。模型 API 有速率限制，有成本配额，有超时风险。用户不会管这些，他们只关心"为什么我的请求变慢了""为什么突然报错了"。中间件必须在前面挡一道，做排队、限流、熔断，把底层 API 的不稳定因素和上层用户体验隔离开。

1.3 DeepAgents 在整条链路里的坐标

为了说清楚 DeepAgents 的定位，我通常画这样一条链路：

业务应用（小程序/Web/后台管理系统） ↓ Agent 编排框架（LangGraph / Semantic Kernel / 自研状态机） ↓ DeepAgents 中间件（路由 + 会话 + 限流 + Token 管理） ↓ 大模型 API（GPT / Claude / 国产模型 / 私有化模型）

注意我的用词：DeepAgents不是要替代 LangGraph 这类编排框架，它和编排框架是两码事。

编排框架管的是"Agent 内部怎么思考、怎么调工具、怎么决定下一步"，它是单次请求内部的逻辑。DeepAgents 管的是"请求怎么进来、会话怎么维持、流量怎么放行"，它是跨请求、跨实例的系统级问题。打个比方：编排框架是司机，负责开车；中间件是交通系统，负责红绿灯、车道划分和限速。没有交通系统的车只能在院子里转，没有司机的交通系统也跑不起来。

如果你用的是 Coze 这类平台，中间件概念会被平台的托管服务掩盖掉，但代价是自定义空间有限。要真刀真枪把 Agent 嵌进自己的业务系统，中间件这一层躲不掉。

1.4 谁适合读这篇文章

这篇东西适合三类人：一是正在把 Agent 从 Demo 往生产环境搬的后端开发，二是给团队设计 Agent 平台的架构师，三是想搞清楚"ai agent 怎么扛并发""ai agent token 是什么意思"这类问题的初学者。我会尽量把原理讲透，但也会给可直接抄作业的配置和代码片段，大家各取所需。

2. DeepAgents 的模块边界：路由、会话与限流各管一摊

2.1 请求路由层：多模型接入与自动降级

DeepAgents 的路由层做了一件很朴素但极有用的设计：内部定义一套统一的AgentRequest和AgentResponse数据模型，所有上游业务只跟这套模型打交道，底层接的是 GPT 还是 Claude 还是别的什么，业务方完全无感。

每个模型 Provider 被封装成独立适配器，包含三样东西：调用函数、健康检查逻辑、成本/时延元数据。注册一个模型只需要写一个适配器类，其余全是配置。

# 路由层示例：多模型注册与策略选择 @dataclass class ModelSpec: name: str provider: str max_tokens: int cost_per_1k: float healthy: bool = True class DeepAgentsRouter: def __init__(self, specs: list[ModelSpec]): self.specs = {s.name: s for s in specs} def pick(self, strategy: str = "cost_first"): candidates = [s for s in self.specs.values() if s.healthy] if strategy == "cost_first": return min(candidates, key=lambda s: s.cost_per_1k) if strategy == "latency_first": return min(candidates, key=lambda s: s.latency) # fallback: 轮询 return candidates[time.time() % len(candidates)]

路由策略我实际用过三种：cost_first省钱、latency_first保体验、manual指定模型跑测试。生产环境我推荐双策略组合——默认用便宜模型兜底，检测到用户连续追问复杂问题时临时升级到强模型。这个升级逻辑放在中间件里做，上游业务完全不用关心。

2.2 会话管理层：把记忆从"模型上下文"里剥离出来

这是 DeepAgents 最有价值的部分。

OpenAI 也好、Claude 也罢，API 本身都是无状态的，所谓"多轮对话记忆"不过是每次请求把完整历史都塞进上下文。这个方案在小规模下没问题，但一旦有几百上千个并发会话，每个会话都拖着几十轮历史，Token 消耗会爆炸，响应时延也会线性恶化。

DeepAgents 的做法是把会话分成两层。原始历史层放在 Redis 里，用会话 ID 做 key，存完整的对话记录，结构大概是下面这个样子：

{ "session_id": "sess_8f3a...", "user_id": "u_1024", "agent_id": "agent_stock", "messages": [ {"role": "user", "content": "帮我分析一下这个数据"}, {"role": "assistant", "content": "从三个方面来看..."} ], "updated_at": "2025-06-15T10:23:00Z" }

上下文输入层则是每次请求前实时拼装的——只取最近 N 轮对话，加上从长期记忆里检索出来的相关片段，再压缩掉冗余内容，最后才喂给模型。这层由中间件的上下文管理器统一处理。

这个剥离带来的好处非常直接：业务进程可以随意水平扩容，因为会话状态不在进程内存里；Agent 崩溃重启后能从 Redis 恢复会话，用户无感知；排查问题时可以直接 dump 某个会话的完整原始历史，而不是靠日志拼凑。

2.3 流量控制层：令牌桶、排队与优先级

DeepAgents 的限流不是一个简单的计数器，而是分了三档。

第一档是全局配额，针对整个中间件实例集群的每秒请求数，保护下游模型 API 不被打死。第二档是用户配额，每个用户每分钟最多多少个请求、每天最多多少 Token，防止个别用户耗尽你的预算。第三档是会话配额，单个会话内连续请求的最小间隔，防止用户疯狂点按钮导致上下文状态错乱。

用户配额的实现我用的是 Redis 加 Lua 脚本的令牌桶方案。选令牌桶而不是固定窗口计数器，是因为 Agent 请求的突发性很强——用户经常会一次性提出包含多个子任务的问题，系统内部会产生一批子请求。固定窗口会在窗口边界瞬间打死所有突发流量，令牌桶允许一定程度的突发，体验好很多。

-- Redis 令牌桶限流（简化版） local key = KEYS[1] local capacity = tonumber(ARGV[1]) local refill_rate = tonumber(ARGV[2]) local now = tonumber(ARGV[3]) local bucket = redis.call("HMGET", key, "tokens", "ts") local tokens = tonumber(bucket[1]) local ts = tonumber(bucket[2]) if tokens == nil then tokens = capacity ts = now else local elapsed = now - ts tokens = math.min(capacity, tokens + elapsed * refill_rate) end if tokens >= 1 then tokens = tokens - 1 redis.call("HSET", key, "tokens", tokens, "ts", now) return 1 else return 0 end

排队机制也很重要。当流量超过配额时，DeepAgents 不会直接返回 429，而是把请求丢进一个带优先级的队列。会员用户优先处理，普通用户排队等待，超过等待时间上限才拒绝。这样即使用户遇到限流，体感也只是"稍微等了一下"，而不是"服务不可用"。

2.4 模块之间的关系

这三个模块不是孤立的，调用顺序通常是：路由层先接住请求，确认走哪个模型；然后流量控制层检查配额，决定直接放行、排队还是拒绝；放行之后由会话管理层加载上下文，拼装请求；模型返回后会话管理层再回写历史，路由层把结果返回上游。

这套流程里最关键的一点是：所有模块都不保存业务方自定义数据，只处理 Agent 运行所需的元数据。业务数据留在业务库里，中间件保持纯粹，这样才不会演变成一个新的数据孤岛。

3. 技术选型复盘：Spring AI、Rust、Python 我都比过

3.1 为什么最终选了 Python + FastAPI + LangGraph

热搜词里有"基于rust语言ai agent""spring ai agent"这些，说明大家确实在纠结技术栈。我自己做选型时也认真比较过。

Python 的优势不在于性能，而在于 AI 生态的完整度。LangChain、LangGraph、LlamaIndex 这些编排框架的迭代速度，是其他语言生态追不上的。写 Agent 逻辑时，你需要的是大量现成的 Tool 调用、记忆模块、Prompt 模板，这些东西 Python 社区最全。DeepAgents 中间件虽然本身不依赖某个具体编排框架，但它需要能和主流框架顺畅对接，Python 是交集最大的选择。

FastAPI 作为中间件的 API 网关层也很合适。异步原生支持让它在单进程里能扛住大量 IO 密集的模型调用，async/await的模型适配器写法很自然，加上 Pydantic 做请求校验，省掉很多重复劳动。

LangGraph 则解决了我对 Agent 内部状态的管理需求。它把 Agent 的推理过程建模成图结构，每个节点是一个处理步骤，边是条件转移。这让复杂 Agent 的逻辑变得可观测、可打断、可恢复，和 DeepAgents 的会话层配合起来很顺手。

3.2 技术选型对比：Python 之外的选项

说实话，如果你的团队全是 Java 背景，且 Agent 只是现有系统的一个辅助功能，用 Spring AI 是合理选择，它在 Spring 生态里的集成做得越来越成熟。Rust 则更适合做超高性能的模型网关，比如 QPS 上万、延迟要求极端的场景。但 Rust 的 Agent 生态目前还在早期，我调研时可用的 LangGraph 类框架几乎没有，意味着从编排到工具调用全要自己实现，周期太长。

DeepAgents 中间件在设计上刻意不做语言绑定——核心模块可以用任何语言实现。我最初用 Python 快速验证架构，等流量起来后，把路由和限流部分用 Rust 重构成独立网关，Python 部分专注 Agent 编排逻辑。这条渐进式路线是最稳的。

3.3 一个容易忽略的选型点：可观测性

选技术栈时不光看性能，还要看你能否在问题发生时快速定位。Python 生态里有完善的 OpenTelemetry 集成，DeepAgents 中间件会给每个请求生成一个trace_id，贯穿 API 网关、Redis、模型调用、Agent 内部节点全过程。日志里有了这个 ID，排查多实例问题就简单多了——按 trace_id 一搜，整条调用链全部出来。这也是我强烈建议任何 Agent 中间件方案都要预留可观测性接口的原因。

4. 扛并发实战：Redis 撑起 DeepAgents 的分布式骨架

4.1 从单机版跑通到分布式：这一步早晚要走

如果你只是做个内部工具，DeepAgents 完全可以单机部署，会话存本地内存，限流用进程内令牌桶，能跑得很欢。但用户量一上来，"ai agent 怎么扛并发"就是绕不过去的坎了——Gunicorn 多 worker 下内存状态互相不可见，用户的会话一会儿在这个进程一会儿在那个进程，表现就是"为什么我的对话历史偶尔丢失"。

我建议的路径是：先单机跑通核心逻辑，再引入 Redis 做分布式化。不要一开始就上 K8s 集群，那会把问题复杂化，让你分不清是业务 bug 还是基础设施问题。

4.2 Redis 在 DeepAgents 里的三处关键改造

会话存储从内存搬到 Redis 是最优先的一步。用session_id做 Hash key，字段放 messages、agent_id、user_id、过期时间。这里有个细节：一定要设置 TTL，而且不能太长。我遇到过有人设了一周 TTL，Redis 内存爆掉，整个中间件跟着雪崩。Agent 会话的黄金有效期通常不超过半天，TTL 设 24 小时足够，再长就是内存浪费。

分布式限流是第二步。前面那段 Lua 脚本就是标准实现，可以保证多实例共享同一个限流计数。要注意的是 Redis 的持久化配置——如果 Redis 重启丢数据，限流计数器归零，下游模型 API 会瞬间被打满。我把 Redis 配了 AOF 持久化，并且限流丢失后有降级逻辑（本地限流兜底），双保险。

Agent 工作节点的分布式锁是第三步。多实例同时消费任务队列时，同一个会话的请求可能被两个实例同时处理，造成状态写入互相覆盖。DeepAgents 在处理会话写回时用 Redlock 风格的分布式锁，锁粒度是session_id，持有时间 2 秒，确保同一时间只有一个实例在写某个会话的状态。

4.3 实测压测数据

在一次模拟压测中，我们配置 3 个中间件实例 + 1 个 Redis 节点，后端模型 API 模拟 1 秒响应时延，测试结果如下：

数据说明两件事：一是 Redis 化对单并发场景几乎没有负优化，二是开启限流排队后，整体错误率反而降下来了——大量用户涌入时，与其让所有请求冲向模型 API 然后超时，不如排个队慢慢处理，牺牲一点 P95 时延换稳定的成功率。这个 trade-off 我强烈建议做。

4.4 扩容后冒出来的新问题

水平扩容不是加了实例就完事，新问题会一个接一个冒出来。

问题一：WebSocket 连接管理。Agent 应用经常要用流式输出，前端通过 WebSocket 收消息。多实例下，客户端连接到实例 A，但任务被调度到实例 B 处理，结果 B 要往 A 上的连接推消息。我的做法是在 Redis 里维护一份conn_id -> 实例号的映射，实例启动时注册、关闭时注销，推送消息前查一下映射表，跨实例转发给目标实例。

问题二：消息乱序。Redis 的消息队列不保证严格顺序，两个连续请求可能被不同 worker 消费，结果后一个请求先处理完，先写回了状态，导致会话历史顺序颠倒。我改成了按session_id做哈希分片，让同一个会话的消息永远进同一个队列分区，从源头上消灭乱序。

这些坑都不深，但都需要在架构设计阶段留出改造空间，事后加会比较疼。

5. Token 成本与上下文瘦身：中间件省下的都是真金白银

5.1 Token 是什么，为什么代理层必须懂

Token 是模型计费和输入输出的基本单位，大体可以理解成"模型眼里的词的碎片"。一个英文单词大概等于 1 到 2 个 Token，一段中文大概一个字对应 1 到 1.5 个 Token。每次调用模型，请求里所有内容——系统 Prompt、对话历史、工具定义、用户输入——都会按 Token 计费。

很多人不重视 Token，觉得单个请求花不了多少钱。但 Agent 场景完全不同：一次 Agent 任务可能要调用模型好几次，每次都带着全部对话历史，上下文里还有大量工具描述和中间结果。一个实际上只消耗 500 Token 的简单问题，可能因为拖着 20 轮历史，实际消耗 5000 Token。成本翻了十倍，用户毫无感知。

DeepAgents 中间件每处理一个请求都会记录input_tokens、output_tokens、total_tokens三项数据，按会话、按用户、按 Agent 三个维度汇总。账单出来时你可以直接看到：哪个 Agent 最烧钱，哪个用户用量异常，哪个时段成本最高。没有这套数据，成本优化就是凭感觉拍脑袋。

5.2 上下文膨胀是变慢变贵的隐形杀手

上下文越长，模型处理得越慢，费用越高，还有一个隐蔽问题：当上下文超过模型窗口一定比例后，模型会"忘记"早期内容。我实测过，32K 窗口的模型塞到 24K 以上时，对最早指令的遵循能力明显下降。这不是玄学，是注意力机制在长序列下的自然退化。

5.3 三种瘦身策略：滑动窗口、摘要压缩、相似度检索

DeepAgents 默认实现三种策略，可以叠加使用。

滑动窗口最简单：只保留最近 10 轮对话，更早的丢弃。适合销售客服这种"本轮问题基本只依赖近期上下文"的场景。缺点是如果你在 15 轮前提过关键约束，滑动窗口会沉默地丢掉它。

摘要压缩是滑动窗口的升级版：超过 N 轮的历史先让模型生成一段摘要，放进上下文里。相当于给模型配了一本"会议纪要"，而不是全部录音。缺点是摘要本身也要花 Token，而且概括过程会丢失细节。

相似度检索借鉴 RAG 的思路：历史消息向量化存起来，请求进来时只检索与当前问题语义相关的历史片段，拼进上下文。这是三种方案里成本最高、效果也最好的，适合需要长期记忆的 Agent。

我在实际配置里通常组合使用：近 5 轮全文保留 + 更早历史做摘要 + 每日任务记录做相似度检索。这套组合让 Token 消耗下降了约 60%，而模型回答质量没有明显下降。

5.4 一个反直觉的经验：缓存系统 Prompt 的收益最大

在做 Token 优化时我发现一个反直觉的点：很多 Agent 的系统 Prompt 写得极长，动不动几千字，里面有产品规则、人格设定、工具说明、安全限制。这段 Prompt 每次请求都要完整发送，是最大的固定成本。

DeepAgents 里做了一个"Prompt 前缀缓存"：模型侧凡是支持前缀缓存能力的，把系统 Prompt 固定为前缀；中间件保证每次请求的 Prompt 前缀完全一致，命中缓存后，这部分 Token 的成本和时延都会大幅下降。这个优化几乎不需要改业务代码，只是要求团队写 Prompt 时不要频繁改系统内容，属于性价比最高的成本控制手段。

6. 四个生产事故的完整排查链路

6.1 事故一：Redis 连接风暴

现象：上线第二天，中间件整体响应变慢，部分请求超时，Redis 监控显示连接数指数级上涨。

排查过程：第一反应看日志，发现一个异常模式——每次模型调用返回前，代码在finally块里释放连接，但有一个分支在抛出异常时没有走到finally，导致连接泄漏。正常的连接池最多 50 个连接，这堆泄漏连接把 Redis 的连接数撑到了上限，后续所有操作排队等待，连最简单的读写都变慢。

修复：统一改用with语句管理 Redis 客户端上下文，确保异常路径也能释放连接，同时给连接池设置最大值和空闲回收时间。这个事故之后我立了个规矩：所有涉及外部资源的代码，必须用上下文管理器或defer类语法，不允许手动 open/close。

6.2 事故二：多实例下的消息乱序

现象：用户反馈对话偶尔出现"答非所问"，明明先问 A 再问 B，模型却先回答了 B。

排查过程：查看 trace 后发现，同一会话的两次请求被分发到了不同 worker，两个 worker 同时拉取会话历史、同时调用模型、几乎同时写回。后写回的覆盖了先写回的，或者两个结果互相交叉，导致上下文错位。这就是我前面说的哈希分片要解决的问题——没有它，多实例并发处理同一会话必然出乱子。

修复：在队列消费端按session_id哈希分桶，保证同一会话的请求永远被同一 worker 串行处理，同时给会话写入加分布式锁。修复后这个故障再没出现过。

6.3 事故三：限流误伤正常用户

现象：白天高峰期，大量用户报错提示"请求过于频繁"，但看单个用户的请求量并不高。

排查过程：查看限流日志发现，问题出在令牌桶容量设置上。当时容量设的是 5，补充速率每秒 1 个。看起来合理，但没考虑到 Agent 单次用户请求内部会拆成 4 到 6 个子请求，每个子请求都会过限流器。用户一次提问就把令牌耗尽，再点一次就被限流。属于业务模型和限流参数的错配。

修复：限流判断从"子请求粒度"改成"用户请求粒度"——用户请求进来时发一个"总通行证"，内部子请求不再单独限流。同时把令牌桶容量从 5 调到 20，补充速率保持不变。修完后误伤基本消除。

6.4 事故四：LangGraph 检查点串号

现象：两个不同用户的 Agent 回答出现了对方的历史数据，属于严重的数据隔离事故。

排查过程：事故发生当天的发布刚好改了会话检查点逻辑。排查后发现，LangGraph 的检查点机制默认用thread_id做状态隔离，我们的代码从请求里提取thread_id时回退逻辑写得不对——如果请求没带thread_id，就默认用了全局常量而不是生成新 ID。结果所有没传thread_id的请求共享了同一个检查点路径，数据全串了。

修复：一是回退逻辑改为生成 UUID 而不是常量，二是增加校验：请求必须携带显式session_id，缺失直接拒绝而不是默默兜底。这次事故让我定了一个原则：涉及数据隔离的代码，宁可报错也不要默默兜底——兜底往往是数据事故的温床。

7. 把 DeepAgents 接到真实业务里：三个落地场景

7.1 场景一：Django 后台管理系统的 Agent 接入

热搜里看到"用 ai agent 开发 django"，正好说说我实践过的方案。我们有个 Django 写的后台系统，既有历史数据要查询，又有业务流程要操作。直接把 Agent 逻辑塞进 Django 的 views 里会很难维护，现在做法是：Django 只负责处理 Web 请求和用户鉴权，然后把任务投递给 DeepAgents 中间件的 HTTP 接口，由中间件去编排 Agent、调用模型、返回结果。Django 侧只需要写一个 service 类封装 HTTP 调用，十分钟搞定接入，Agent 的并发和会话管理完全交给中间件。

# Django 侧接入 DeepAgents 的示例 import httpx class DeepAgentsClient: def __init__(self, base_url: str, api_key: str): self.client = httpx.Client(base_url=base_url, headers={"X-API-Key": api_key}) def ask(self, session_id: str, message: str): resp = self.client.post("/v1/agent/chat", json={ "session_id": session_id, "message": message, }) resp.raise_for_status() return resp.json()["reply"]

这个模式的核心价值是业务系统和 AI 能力的解耦。Django 团队不需要懂 Agent 怎么编排，中间件团队不需要懂业务表结构，两边只通过接口契约交互。

7.2 场景二：定时任务与自动通知

我自己实践了一个很实用的场景：定时任务会让 Agent 轮询某个数据源，发现变化后生成一条摘要和提醒，通过 Webhook 推送给用户。

这个场景的难点在于任务调度和 Agent 执行的配合，以及如何按用户维度做 token 配额。DeepAgents 提供一个任务队列接口，外部调度系统（我用 Celery Beat）把任务投进来，中间件按用户维度排队，逐个执行并消耗用户配额，防止定时任务把一天的 Token 预算在几分钟内打光。

类似思路可以扩展到很多业务：往企业微信群里发项目进展日报、监控某个页面变化生成简报、等等。做这类自动化通知时，一定记得在中间件层面加"任务频率上限"和"单日 Token 上限"两道闸，防止定时任务出错后无限循环烧完预算。我吃过这个亏，一次误配置让模型在半夜跑了上千次调用，第二天看到账单时人麻了。

7.3 场景三：多 Agent 协作编排

最后说一个进阶用法：DeepAgents 中间件同时接多个 Agent，形成一个协作网络。

典型例子是"内容生产流水线"：一个规划 Agent 负责拆解任务、一个资料 Agent 负责检索信息、一个写作 Agent 负责成稿、一个校对 Agent 负责审校。它们在中间件的调度下协作——每个 Agent 都是无状态的"工人"，状态由中间件的会话层统一保管。这么做的好处是单个 Agent 可以独立升级、独立限流、独立计费，坏处是链路变长，任何一个环节出问题都可能影响整体。

实践下来，我的建议是：多 Agent 协作一定要规划好"超时总预算"。给整个链路设定一个最大完成时间，比如 60 秒，任何环节超时就把控制权交还给用户，返回已完成的阶段性结果而不是一味等下去。这个设计比单环节逐一超时优雅得多。

7.4 这个中间件的边界在哪里

聊到最后我必须泼一盆冷水：DeepAgents 中间件解决的是路由、会话、限流、成本、可观测这些系统级问题，它不解决 Agent"聪明不聪明"的问题。模型选得差、Prompt 写得烂、工具链设计不合理，这些是编排层和业务层的活儿，中间件再完善也救不了。把中间件做扎实，是为了让你的团队可以放心大胆地在上层试错——这其实是它最大的价值：把试错的代价降到最低。

就拿我这边的经验来说，接入 DeepAgents 之前，每接一个新业务场景都要重复处理一遍模型 Key 管理、会话恢复、限流参数这些体力活；接入之后，新场景只需要注册一个 Agent 配置，写清楚路由策略和上下文窗口，剩下的交给中间件。这种复用带来的效率提升，是我在这个项目里最满意的结果。