一次 AI 问答，背后是十几个 Agent 在 “传纸条” — RocketMQ Lite Topic 实战

一、设想这样一个场景

你打开一个 AI 助手，敲下一句话：

"帮我看看明天去上海出差，天气怎么样、机票多少钱、我那天的日程排得开吗？"

你以为它就是一个模型在吐字？不是。

它背后是好几个 Agent 在协作：一个编排 Agent（主 Agent） 把你的问题拆成三件事 -> 查天气、查机票、查日程 -> 分别交给三个服务 Agent（子 Agent）去办，等结果都回来了，再汇总成一段顺畅的回答推给你。

这条看似轻巧的链路里，藏着几个让人头疼的细节：

• 三个工具谁也不依赖谁，本该 同时去查，可一旦用同步调用串起来，就变成"查完天气再查机票再查日程"，用户干等；
• 机票接口偶尔慢个 30 秒，整次回答就此卡住，你看到的是屏幕上 "转圈圈无响应"；
• 回答正流式地往外蹦字，你顺手锁了屏进了地铁，网络一断，已经生成的半段回答眼睁睁丢了，唤醒后只能从头再问一遍；
• 隔壁同事的公司刚上线营销活动，上千人在跑别的 AI 任务，你的请求被卡在共享队列后面排队。

焦急地等待着……

二、问题的本质：AI 原生应用的 "通信难题"

上面这个看起来很简单的场景，背后藏着的是 AI 原生应用（AI Native Application） 带来的全新通信挑战。

与微服务架构相比，AI 原生应用呈现出三个显著特征：

• 异步化通信成为主流：Agent 之间是任务驱动的异步协作，单次任务动辄数秒到数分钟。
• 会话上下文重要性凸显：每个用户会话需要独立的上下文管理，不能被相邻租户挤占。
• 百万级独立通道需求：每个 AI 用户会话、每次长任务、每个知识库都需要独立通道。

传统消息中间件的"Topic 预创建 + 共享消费组"模型，在这三个挑战面前几乎全部失守：

• Topic 数量动辄数千上限，扛不住百万级会话独立通道的需求
• 共享消费组里慢租户拖垮快租户，没法做到会话级隔离
• 同步 RPC 调用一卡全卡，没法支撑长任务链路

腾讯云 TDMQ 团队，针对这些挑战，给 RocketMQ 上新了一个新能力 - 轻量主题（Lite Topic）。它的核心使命就一句话：像创建文件一样简单地创建一个消息通道，用完即弃、自动回收、严格保序、断线不丢。

接下来，我们通过 2 个真实的 AI 场景，看看 Lite Topic 到底解决了什么问题。

三、场景 1：Agent 并行通信 + 断线续传

痛点：同步调用一卡全卡，断线则前功尽弃

回到开头那次出差问答。如果编排 Agent（主 Agent）和服务 Agent（子 Agent） 之间走传统的 HTTP/RPC 同步调用，会面临一连串瓶颈：

• 同步阻塞：三个工具调用本可并行，串行调用却要逐个等待；其中任一个慢 30 秒，编排 Agent 就被阻塞 30 秒，前端只能"转圈圈"
• 扩展性受限：编排 Agent 想新增一个 Agent 服务类型（如"查酒店"），就要改它自己的调用、重试、超时处理逻辑
• 系统脆弱：某个服务 Agent 短暂不可用，整次回答失败，用户被迫从头再问
• 断线即丢：回答以流式 token 往外推，用户网络一抖、网关一升级，已经吐出的半段内容就丢了，无法续上

Lite Topic 解法：请求异步并行 + 回答可断线续推

我们把这条链路拆成两段来看：请求侧（编排 Agent 把活派给各服务 Agent）和 回答侧（结果流式回推给用户）。

💡 先理清两个角色：谁在"思考"、谁在"干活" —— 开头说的"好几个 Agent 协作"，落到工程上其实是两类单元：编排 Agent（主 Agent）负责拆解问题、决定派什么活、最后汇总回答；服务 Agent（子 Agent）负责接到活就去执行（查天气、查机票），是订阅消息、调用服务、回写结果的执行者。

1）前提准备

• 为每个子 Agent 创建一个一级 topic，如 weather，flight 和 calendar，用于接收主 Agent 的任务请求。一级 topic 按 Agent 职能划分，少量且固定。
• 为主 Agent 创建一个一级 Topic，如 reply，用于接收子 Agent 的处理结果。
• 用户的每个任务在上述的各一级 Topic 下，按照 task ID 自动创建不同的二级 Lite Topic，支持百万级别，并保证不同任务之间的隔离。

2）请求链路：Request Lite Topic 承载海量请求 + 通配符订阅自动订阅

同一个子 Agent（如天气 Agent）一天内可能被不同用户的不同任务调用成千上万次，产生海量且临时的请求。

• 主 Agent 拆解完任务后，在每个子 Agent 已创建的一级 Topic（如"weather"）下，根据 task ID 直接指定一个二级 Request Lite Topic（如 weather.{taskId}），发送任务信息。Lite Topic 可以设置 TTL 过期时间（如 12 小时），即如果 12 小时内没有新的消息写入，则这个 Request Lite Topic 用完即弃，自动回收，节约系统资源，及时清理脏数据。
• 子 Agent 通过 Lite Topic 的通配符订阅能力，可以直接 subscribeLite("*") 一次性订阅"weather"这个一级 Topic 下的所有 Lite Topic，无需感知具体的 task ID 和对应的 Lite Topic 的存在。

子 Agent 通过通配符订阅自动感知并消费新请求，三个 Agent 的调用互相独立、无先后依赖，谁先回来都行，任一个慢也不阻塞其他。主 Agent 的派发动作与子 Agent 的处理动作 彻底解耦。新增一个 Agent 类型（如酒店 Agent），只需新建一个"hotel"一级 Topic，让新加入的酒店 Agent 通过通配符订阅所有的 hotel.{taskId} Lite Topic 即可。

3）响应链路：用 Lite Topic 为每次任务创建一个专属通道

• 子 Agent 处理完请求后，在"reply"这个一级 Topic 下，根据 task ID 直接指定一个二级 Response Lite Topic（如 reply.{taskId}），把结果追加写入。
• 主 Agent 通过 subscribeLite(taskId) 精确订阅、收齐三份回答后汇总，再经 SSE 流式推送给前端。通过 TTL 到点自动回收的功能，同样可以做到回答用完即弃。

💡 两侧 "发现" 机制对称而不同：在请求时，子 Agent 通过 通配符订阅（如 subscribeLite("*")）自动感知并消费新请求，子 Agent 无需知道具体 Lite Topic 名称；而响应时，Lite Topic 由派发方（主 Agent）自己创建、自己知道 taskId，直接 subscribeLite(taskId) 即可，无需使用通配符订阅。

（图 1：Agent 并行调用链路）

4）断线续传：把 "流式不丢" 降维成 "维护一个位点"

回答侧用 Lite Topic 还顺手解决了开头那个最恼人的问题 - 断线丢字。

因为 Lite Topic 天生具备这些能力：单队列严格保序、消息持久化、排他消费 "最新连接胜出"。于是一次"地铁里断网"的经历会变成这样：

1. 服务 Agent 把回答 token 持续写入 Response Lite Topic，浏览器每收到一条就记录 lastReceivedOffset。
2. 你锁屏进地铁，浏览器与网关 A 断开——但 写入端完全无感，token 仍在持续落盘。
3. 你出站唤醒手机，浏览器自动重连，携带 lastReceivedOffset=50，命中网关 B。
4. 网关 B 以排他模式订阅该 Lite Topic，触发 "最新连接胜出"自动踢掉网关 A；从断点一次性补齐 token_51 ~ 最新，你感知到的只是"网络恢复后接着看"。
5. 会话结束、TTL 到点，Lite Topic 连同位点、订阅关系自动销毁。

（图 2：断线续传流程）

Lite Topic 在这里并不"自动"实现端到端断线续传，而是用一组协议级能力，把这个原本极其复杂的工程问题，简化为业务侧维护一个客户端位点。请求侧并行不阻塞、回答侧不丢不乱，一次问答就此稳稳落地。

四、场景 2：大模型记忆服务的 "记忆写入"

痛点：同一用户要严格保序，海量用户又要互不干扰

场景 1 是 "单次问答内部" 的协作。当 AI 产品要 "越用越懂你"，就需要一层 长期记忆服务：把用户对话中沉淀出的偏好、事实、画像持续写入存储，下次对话时检索出来注入 prompt。

这里有一个看似矛盾的双重要求。我们以一个典型例子说明：

A 用户 周一说"我喜欢喝咖啡"，周二 又说"改喝茶了"。

• 同一用户必须严格保序：这两条记忆一旦乱序写入，记忆系统做冲突消解时就会得出"用户喜欢咖啡"的错误画像。
• 海量用户必须互不干扰：线上同时有数十万、上百万用户在写记忆，任何一个用户写得慢，绝不能拖累其他用户。

📌 顺带说明一个常见误解：消息中间件只参与 记忆写入侧（异步落库 + 保序 + 隔离）；用户提问时的 记忆检索是在线请求，直接查向量库 / KV，不经过 MQ。

Lite Topic 解法：每个用户独占一条记忆通道

为每个用户直接在 "mem" 这个一级 Topic 下自动创建一条 Lite Topic，可以使用 user ID 进行命名，便于理解，我们暂时把这个 Lite Topic 记为 mem.{userId}。

这样的好处是：

• 天然 FIFO 保序：同一用户的 "咖啡→茶" 严格按序写入，冲突消解不会出错
• Lite Topic 间隔离：A 用户写得再慢，也只影响 A 自己，B、C 完全不受波及
• 消费隔离：每个用户独立通道，单个慢用户不影响其他用户
• 并行度说明：消费并行度由 Worker 实例数和分配策略决定。若有 N 个 Worker 实例，可同时处理 N 个用户的记忆写入；但同用户的消息始终由同一个 Worker 实例消费（保证保序），不同用户可并行消费。
• 百万级轻量通道：Topic 量级从千级跃升到 百万级，自动创建、TTL 自动回收，故障仅影响 1 个用户

生产者与消费者怎么用？

• 生产者 = 记忆写入服务：按 mem.{userId} 进行投递，把同一用户的记忆写入其专属 Lite Topic。
• 消费者 = 记忆处理服务：subscribeLite(mem.{userId}) 订阅某个用户的通道，每个用户独立保序消费，完成 embedding、记忆演化、写入向量库 / KV。

那么下游 Memory Worker 怎么知道"有新用户上线"、该订阅哪条新 Lite Topic？

借助 Lite Topic 的通配符订阅能力，记忆处理 Agent 可以直接 subscribeLite("*") 订阅 mem 一级 Topic 下所有 Lite Topic，一次性订阅所有匹配的用户记忆通道。当有新用户首次写入时，其专属的 mem.{userId} Lite Topic 被自动创建，下游 Memory Worker 通过通配符订阅自动感知并消费该用户的记忆消息，无需任何额外的通知机制。会话结束后，该 Lite Topic 随 TTL 到期自动回收，Memory Worker 的通配符订阅自动停止消费已回收的 Lite Topic。

（图 3：大模型记忆写入链路）

一句话总结：记忆写入要的就是 "每个用户一条独立、保序、可弹性伸缩的写入流水线"，而 Lite Topic 的"每用户独占 Topic + Topic 内保序 + Topic 间隔离 + 百万级轻量通道" 恰好严丝合缝。

五、小结

通过 2 个真实场景，我们看到 Lite Topic 是怎么把 AI 应用的几类经典通信难题，从 "业务侧自己拼凑" 提升到 "协议级保证"：

留给下篇的 3 个谜题

读到这里，相信你已经注意到 Lite Topic 表现出的几个 "不寻常" 的能力。

但问题来了：

🤔 谜题 1：单实例 支持百万级 Lite Topic 自动创建，怎么做到不爆内存，不爆文件句柄的？

🤔 谜题 2：海量的 Lite Topic 必然带来 海量的订阅关系，这些订阅关系的维护如何保证效率和稳定？

🤔 谜题 3："最新连接胜出"的排他消费 是怎么做到的？传统 RocketMQ 不是 Consumer Group 共享消费吗？

这些"魔法"背后是怎么实现的？腾讯云 TDMQ 团队又是怎么在 RocketMQ 既有架构里做这套能力增强的？

🎬 下篇预告 · 同系列 —— 《上篇答疑：RocketMQ 轻量主题是怎么把百万通道做"轻"的》。我们将从 LMQ、CommitLog、Ready Event Set、Exclusive Mode 四个内核组件入手，逐一拆开三个谜题，看 Lite Topic 是怎么在保持 RocketMQ 既有存储体系的前提下，做出这套面向 AI 原生应用的能力增强的。

敬请期待。

关于腾讯云 TDMQ

腾讯云 TDMQ 是腾讯云推出的企业级消息队列服务，深度结合 RocketMQ 开源社区能力，为大模型时代的 AI 原生应用提供消息中间件底座。

Lite Topic 能力当前已在腾讯云内部多个 AI 业务灰度运行，后续将面向外部客户商业化，欢迎加入 👇 产品交流群沟通了解更多信息！

往期推荐

重磅发布！为 AI 原生应用而生 - TDMQ RocketMQ 轻量主题 Lite Topic！

智能路由与业务连续性保障：AI 网关如何让大模型稳定跑在生产链路上

MCP 网关：让存量接口 “零” 代码改造接入 AI 生态

TDMQ RabbitMQ 托管版全新推出 4.2：原生支持 AMQP 1.0，吞吐性能翻倍！

车联网实战：TDMQ MQTT + 云函数打造车辆状态智能响应链路

扫描下方二维码关注本公众号，

了解更多微服务、消息队列的相关信息！