AI Agent的事件驱动架构实践

AI Agent 的事件驱动架构实践Apache RocketMQ × AI阿里云智能云原生解决方案架构师邹星宇Contents目录01 传统消息队列在 AI 场景中的局限性02 Lite-Topic：面向 AI 领域的新范式03 优先级 Topic：面向 AI 场景的分级消费策略04 面向 AI 场景的应用案例1、传统消息队列在 AI 场景中的局限性更长的响应时间• 和传统互联网应用毫秒级的延时比，AI应用的响应时常基本是分钟级以上。• AI应用一次业务的运行时间不可预测性更高。更复杂的交互•AI 应用的的多轮对话持续时间长，历史可达数十轮。• 上下文的传输可能达到几十甚至上百M，上下文管理难度更高。• 多Agent之间的协同更加复杂。更昂贵的计算资源• AI 推理依赖昂贵的 GPU 资源，瞬时高并发流量可能冲击推理服务稳定性，导致算力资源浪费。• 任务失败重试的成本更高。更精细化的事件驱动• 因为计算能力有限，异步事件驱动需要更精准的消费速度控制。• 分级的事件驱动策略，确保高优先级任务优先获得宝贵的计算资源。AI业务场景的新特性轻量化通信模型•轻量级的资源管理，低开销的资源创建支持海量会话场景•更细粒度的的订阅管理•适用于长时会话、AI 工作流和 Agent-to-Agent 交互等场景智能化资源调度•定速消费•优先级消费•优先级修改•确保在高并发和多租户环境下高效利用资源大上下文管理•以连续的消息保存上下文•支持超大消息体•顺序、互斥消费保证上下文的完整AI 时代RocketMQ新特性2、Lite-Topic：面向 AI 领域的新范式轻量化通信模型：•百万级队列支持：RocketMQ 支持在单个集群中高效管理百万级 Lite-Topic，能够为海量并发会话或任务提供独立 Topic，并且保障性能无损。•轻量化资源管理：RocketMQ 队列的创建和销毁极其轻量和自动化，系统可按需自动创建与回收Lite-Topic（如客户端连接断开或 TTL 到期时），避免资源泄漏和手动干预，显著降低使用复杂度和成本。•大消息体传输：RocketMQ 可处理数十 MB 甚至更大的消息体，充分满足 AIGC 场景中常见的庞大数据负载的传输需求，如大量上下文的 Prompt、高清图像或长篇文档等。•顺序消息保障：通过顺序消费，确保推理结果流式输出到客户端的顺序性，保障会话体验连贯流畅。轻量级Topic：•基于百万队列的方案，本质上是一个个 queue•从全局上来看，一个轻量级 Topic 不会存在于每一个broker上，在分配和发送时像顺序Topic 的发送一样要做 queue 的 hash•Queue的消息是某个 broker 专属的，一个轻量级 topic 的发送在只会到一台 broker，而不是轮询发送轻量级订阅：•消费组 group 的概念被弱化•订阅关系粒度更细，以 client_ID 维度维护•新增互斥（ Exclusive）消费模式•没有订阅关系不一致导致的各种消费问题•TTL 到期后自动删除订阅关系消费分发策略：•客户端发起读请求不再指定 topic，而是broker 根据 client_ID 识别订阅关系，并返回多个 topic 的多条消息•引入类似 epoll 机制的 topic ready set，在 pop 请求处理时直接访问就绪的 topic•当订阅上线、新消息发送、消息 ACK 后仍有消息、order lock 释放时往 topic readyset 进行 add 操作3、优先级 Topic：面向 AI 场景的分级消费策略大模型服务在资源调度上，普遍面临两大核心挑战•负载不匹配：前端请求突发性强，而后端算力资源有限且相对稳定，直接对接易导致服务过载崩溃或算力资源浪费。•无差别分配：在实现流量平稳后，如何确保高优先级任务优先获得宝贵的计算资源，成为提升整体服务价值的关键。优先级Topic为实现智能算力调度提供了坚实的基础•削峰填谷保护 AI 算力：RocketMQ 天然具备“流量水库”的作用，能缓存突发请求，使后端 AI 模型服务根据自身处理能力，基于类似滑动窗口模式自适应消费负载均衡，避免系统过载或资源浪费。•定速消费，最大化 AI 算力利用率：RocketMQ 支持定速消费能力，可为消费者组 group 设置消费quota。开发者可灵活定义 AI 算力的每秒调用量，在保障核心 AI 算力不过载的前提下，最大限度提升吞吐量。•抢占式分配：当高价值任务（如 VIP 用户请求、关键系统分析）进入系统时，可将其标记为高优先级消息。RocketMQ 确保这些消息被优先消费，让宝贵的算力资源优先服务于最关键的任务。•按权重分配：在共享算力池场景下，可依据各业务请求的实时执行状态设置请求消息优先级，调整请求执行的先后顺序，既保障整体吞吐效率，又防止个别租户因资源饥饿而无法获得算力。优先级 Topic：•多队列： 对应优先级定义，不同优先级对应不同队列•队列选择：对应调度（存取策略），优先从高P队列取数，依次进行队列选择策略：•永远从高到低依次选择•概率模式，当前普通消息的 pop 其实也在按概率选择队列，只不过每个队列概率相等，对于优先级队列来说，高优先级的被筛选到的概率更高4、面向 AI 场景的应用案例MQ 和 AI 会话网关的续传场景AI 会话网关：•会话消息的收发管理、session 维护，统一不同的接入渠道MQ 作用：•AI 应用通过消息句柄（业务标识）进行回复（发消息），一次回答可以回复多条，网关机器接收结果（收消息）•网关到应用之间，使用通用的 HTTP 协议•AI 应用处理耗时较长，不适合同步等待，通过 MQ 中转，本质上是一种典型的异步推理场景MQ 和 AI 网关的会话续传场景现有 MQ 的解决方案•使用缓存维护 tag 集合，并实现租约机制•预创建若干 Consumer Group，用于分配独立的消费身份•将请求带上身份信息，消费时使用 tag 进行过滤•每次接入新的租户，重复上述过程•在 group 和 tag 上额外做很多业务无关的设计MQ 和 AI 网关的会话续传场景现有 MQ 方案的问题会话网关场景特点：•有状态：浏览器/APP 和网关之间是SSE/WebSocket 等连接, 会话是有状态的•智能应用发送消息时，虽然可以将原始request 标识带上，但 MQ 传统的消费模式不能保证消息被对应机器接收•网关机器在收到 response 时，可能找不到对应 SSE 连接MQ 和 AI 网关的会话续传场景基于轻量级通信模型轻量级通信流程•网关机器在发起请求时带上身份标识，并开始订阅该身份标识对应的消息（无需预创建 group、topic）•智能应用根据请求的标识，发送对应的消息（同样无需预创建）•网关机器各自接收属于自己的response消息续传场景•网关机器在发起请求时带上当前 request 的 session，并开始订阅该 session 对应的消息•智能应用回复对应 session 的消息（同样无需预创建）•网关机器各自接收自己负责处理的 session 的消息•在网关机器下线/宕机时，端上重连，新的网关机器可以动态订阅 session 的消息，自动恢复