多Agent应用的核心设计：跨Agent与跨Session通信

在开发多 Agent 应用时很多团队会遇到一个比较典型的分水岭单个 Agent 的演示能跑通但一旦需要多个 Agent 协作或者用户在一个新会话里想继续上一次的进度逻辑就开始变得混乱。比如常见的场景Agent A 处理完任务后需要把中间结果交给 Agent B 继续处理结果 Agent B 的上下文里根本看不到之前的执行记录。又比如用户今天早上在会话 1 里让 Agent 写了一份代码框架下午打开新会话 2 想继续修改却发现两个会话之间是完全隔离的所有上下文都得靠用户重新描述一遍。这类问题的根源往往不是模型能力不够而是没有把“跨 Agent 通信”和“跨 Session 通信”当作 Agent 应用的底层基础设施来设计。这篇文章会围绕这两个概念展开讲清楚它们分别解决什么问题、有哪些常见实现方式并用一套 Python 可运行的轻量示例演示多 Agent 协作、跨会话读取进度、权限控制等核心逻辑。如果你是正在做 Agent 开发、想把单 Agent Demo 升级成多 Agent 工程化项目的开发者这篇文章会比较适合你。1. 为什么需要跨 Agent 和跨 Session 通信1.1 从单 Agent 到多 Agent 是必然趋势最早接触大模型开发时很多项目的结构非常简单用户输入一句话程序把这句话拼进 Prompt然后调用一次模型接口最后输出结果。这是“单次问答”模式。但真实业务不会这么简单。比如一个企业知识库问答系统可能需要先有一个 Agent 判断用户意图再有一个 Agent 去检索 RAG 知识库接着有一个 Agent 负责写答案最后还要有一个 Agent 去做事实校验。如果把这些逻辑全部写在一个大函数里代码会迅速膨胀而且很难维护。于是大家开始把一个复杂任务拆成多个 Agent每个 Agent 只做一类事。这时就产生了新问题多个 Agent 不是一个一个函数它们是独立运行的逻辑单元。A 处理完的结果怎么交给 BB 需要读取用户在某次会话中提供的信息从哪里拿这一层就是跨 Agent 通信。需要注意的是跨 Agent 通信并不一定等于“让两个大模型互相对话”。在很多工程实现里它更接近一套消息路由机制一个 Agent 发出事件或指令消息总线根据目标标识把它转发给正确的接收者。通信的内容可以是任务指令、结构化数据、文件路径也可以是状态变更通知。1.2 Session 到底是什么先理清三个层面很多后端开发者看到“Session”这个词会立刻想到 Web 登录态。在 Web 里Session 通常指服务端保存的用户会话状态Cookie 是浏览器侧的凭证Token 则是无状态的身份令牌。这个认知模型是没错的。但进入 Agent 开发后Session 的含义更重了。它不再只是“用户有没有登录”的标志而是一个承载上下文、任务状态和权限边界的容器。从工程角度看可以把 Agent 应用里的 Session 理解为三层东西交互层它是用户和 Agent 的一次连续对话过程包含用户消息和 Agent 回复。上下文层它保存了用户本次任务的业务状态例如文档内容、中间产物、任务阶段。权限层它定义了哪些用户、哪些 Agent 有权读写这个会话中的数据。在常见的 Agent 框架中有的框架把 Session 称为 Thread有的称为 Conversation有的直接叫 State。不管名字怎么变核心诉求一致一个 Session 应该能独立承载一次业务目标同时能被安全地跨会话访问。1.3 跨 Agent 通信和跨 Session 通信的关系跨 Agent 通信的重点是“消息在 Agent 实例之间流动”解决的是任务怎么拆解、结果怎么接力的问题。跨 Session 通信的重点是“状态在不同会话之间共享”解决的是用户换了新会话后怎么继续上一次任务、怎么共享上下文的问题。这两个概念经常同时出现。一个典型的组合场景是用户先在会话 A 里发起了一个耗时较长的任务系统里负责拆单的 Agent 把任务转给执行 Agent执行 Agent 处理后把结果写回了会话 A 的状态中。过了一段时间用户打开了新会话 B想知道任务是否完成。此时会话 B 需要通过授权机制读到会话 A 的执行状态如果任务还没结束甚至可以在会话 B 中继续推动这个任务。也就是说Agent 之间的协作往往发生在某个 Session 的作用域内而 Session 之间的连接又为 Agent 的跨会话接力提供了通道。2. 跨 Agent 通信的核心实现方式2.1 消息路由方式总线、队列、注册中心最常见的跨 Agent 通信方式是“消息路由”。实现时通常会引入一个中心化的消息总线所有 Agent 都注册到总线上消息统一通过总线转发。消息路由有两种经典模式。第一种是点对点发送也就是一个 Agent 明确指定把消息发给谁。比如规划 Agent 把“生成代码”这个任务发送给代码生成 Agent 时消息里会带上 receiver 字段内容是目标 Agent 的 ID。总线收到消息后从注册表里找到对应的 Agent把消息交给它处理。第二种是发布订阅。某个 Agent 发出一个事件比如“draft.ready”总线会把事件广播给所有订阅了这个事件的 Agent。发布者不关心谁接收订阅者也不关心谁发布双方通过事件类型解耦。这种方式的核心是 Agent 注册表和消息协议。Agent 注册表负责记录当前有哪些 Agent 可用、它们的编号是什么消息协议则定义了消息的字段结构例如消息 ID、发送者、接收者、类型、会话 ID、业务内容。2.2 共享状态方式StateStore跨 Agent 通信并不总是靠“消息”来完成的。另一种常见方式是把共享状态放到一个统一的存储层例如 Redis、数据库或内存 Map。在这种模式下Agent A 不需要显式地把数据发给 Agent B它只需要把结果写到某个共享 Key 下Agent B 在处理自己那部分任务时去读取即可。这种方式的优点是实现简单、调试直观缺点是如果所有 Agent 都直接读写同一个状态对象状态会很快失控出现“谁改了数据都不知道”的问题。比较稳妥的做法是给共享存储封装一层带权限校验的 API所有 Agent 必须通过 API 读写状态而不是直接操作数据结构。2.3 主流 Agent 框架是如何处理通信的如果你在使用 LangGraph、CrewAI、微软 Agent Framework 等 Agent 编排框架会发现它们内部已经封装了一些通信能力。在不同的框架里概念术语会不同但核心思想类似定义状态结构、在 Agent 之间传递状态、通过图结构或编排器控制执行顺序。例如 LangGraph 的节点间会共享一个 state 对象节点执行后返回状态的增量更新CrewAI 则通过 Task 和 Crew 的概念组织多个 Agent 的工作流。学习这些框架时不要只记 API重点理解它们背后的通信模型这样才能在不同框架之间迁移。3. 跨 Session 通信的设计要点3.1 核心前提Session 状态必须可寻址要让一个会话能访问另一个会话的数据首先得保证每个 Session 都有稳定且唯一的 ID并且 Session 的状态存储在 Agent 实例之外。如果你把 Session 状态全部放在单个 Agent 进程的内存变量里那么一旦 Agent 重启、扩容或者消息被路由到另一个实例Session 就丢了。因此工程上通常会引入一个独立的 SessionManager 组件负责管理 Session 的创建、状态读写和授权关系。这个 SessionManager 不一定非要引入 Redis 等外部组件。在早期原型阶段可以用一个内存字典来模拟等架构稳定后再替换成 Redis 或 MySQL。关键是要在 API 层面把“Session 操作”和“Agent 业务逻辑”分离开。定义 Session ID 时也有很多实践细节。比如有些团队直接用 UUID有些团队使用“用户ID 会话序号”的语义化 ID还有团队使用“业务类型 业务单号”的格式。推荐至少保证全局唯一、可追踪、可路由。3.2 显式授权跨 Session 不等于全局可见很多新手在设计跨 Session 通信时容易犯一个错误既然是同一个用户的会话就直接把所有会话数据做成全局可读。这个做法非常危险。考虑这样一个场景用户在会话 A 里上传了一份公司内部敏感文档然后打开会话 B 询问“我上次传的文档你帮我总结一下”。由于两个会话都属于同一个用户如果系统允许会话 B 无条件读取会话 A 的全部数据看起来方便但一旦用户的会话 ID 泄露或者系统里存在越权调用漏洞攻击者就可能通过任意一个用户会话读取该用户其他会话里的敏感数据。更合理的设计是“显式授权”。也就是说跨 Session 访问必须有一个明确的授权动作。例如用户新建会话时点击“继续上次会话”系统自动创建授权关系。会话 A 主动共享某部分内容给会话 B。用户通过命令明确指定“下一步工作基于会话 A 的任务继续”。结合“最小可见原则”即使授权跨会话读取也要考虑只暴露必要字段而不是无脑把整个 state 字典都返回。3.3 和处理长任务结合跨 Session 通信最实用的场景之一是长任务跟踪。用户在一个会话里提交了一个需要几分钟甚至更长时间的任务如果让页面一直等待体验会很差。更好的做法是在会话 A 中创建任务生成一个任务 ID。后台的 Agent 异步执行任务并把进度写入 Session A 的状态。用户随时可以在会话 B 中通过任务 ID 或会话关联关系查询任务状态。任务完成后会话 B 可以直接读取最终结果也可以触发后续 Agent 继续处理。这种做法和传统 Web 中的任务队列设计非常相似。核心差异在于这里的执行单元不是普通 Worker而是具备上下文理解能力的 Agent因此状态记录要比普通任务队列更丰富一些。4. 完整实战用 Python 实现轻量通信层下面我用 Python 写一个简化的多 Agent 协作示例。示例中不依赖第三方库只需要 Python 3.8 以上环境即可运行。为了让读者更容易理解我会把代码拆成几个文件session_comm_demo/ ├── models.py # 消息结构定义 ├── bus.py # 消息总线、Agent 基类 ├── session.py # Session 和 SessionManager ├── agents.py # 各类 Agent └── main.py # 演示入口这个示例会模拟以下流程用户 Alice 在会话 session-A 中提交一个写作任务。规划 Agent 接收任务并把 Writer Agent 和 Reviewer Agent 加入会话。规划 Agent 将任务发送给 Writer Agent这属于跨 Agent 点对点通信。Writer Agent 生成草稿并写入 Session 状态再发送给 Reviewer Agent 审校。Reviewer Agent 审校完成后通过事件广播通知通知 Agent这属于发布订阅通信。Alice 在新会话 session-B 中查询 session-A 的上下文这属于跨 Session 通信。用户 Bob 尝试读取 session-A 的数据因为没有授权被拒绝。4.1 定义消息数据结构# models.py from dataclasses import dataclass from typing import Any, Dict dataclass class Message: msg_id: str sender: str receiver: str msg_type: str session_id: str payload: Dict[str, Any] def __repr__(self): return ( fMessage(id{self.msg_id}, f{self.sender} - {self.receiver}, ftype{self.msg_type}, fsession{self.session_id}) )消息结构中最重要的几个字段是sender发送者标识。可以是用户 ID也可以是 Agent ID。receiver接收者标识。对于点对点消息这是目标 Agent ID对于广播事件一般是通配符。msg_type消息类型。接收方会根据这个字段找到对应的处理方法。session_id问题所属的会话 ID。它保证了消息能关联到具体的业务上下文。payload业务体内容通常是字典结构方便扩展字段。4.2 实现消息总线和 Agent 基类# bus.py import uuid from models import Message class MessageBus: 轻量消息总线: 注册 Agent, 负责点对点路由和事件广播 def __init__(self): self.agents {} self.message_log [] def register(self, agent): 把 Agent 注册到总线上, 后续消息才能路由到它 self.agents[agent.agent_id] agent agent.bind_bus(self) def send(self, msg: Message): 点对点消息路由: 发送给明确指定的 receiver print(f[总线] {msg.sender} - {msg.receiver} | type{msg.msg_type} | session{msg.session_id}) target self.agents.get(msg.receiver) if target is None: print(f[总线] 警告: 未找到目标 Agent: {msg.receiver}, 消息进入死信) return target.receive(msg) self.message_log.append(msg) def publish(self, msg: Message): 发布订阅消息: 广播给所有订阅该事件类型的 Agent print(f[总线] {msg.sender} 发布事件 | type{msg.msg_type} | session{msg.session_id}) for agent in self.agents.values(): if msg.msg_type in agent.subscribe_events: agent.receive(msg) self.message_log.append(msg) class BaseAgent