做 AI Agent 应用时很多人会把大多数精力放在单 Agent 的 Prompt、工具调用和上下文构造上。等 Agent 数量变多、业务链路拉长后才发现真正卡住项目的往往不是单个 Agent 的能力而是多个 Agent、多个会话之间怎么通信、怎么协作、怎么把状态安全地传递过去。本文将围绕“跨 Agent 通信、跨 Session 通信”这一主题从概念拆解、设计模式、代码实战到排错建议逐层展开适合正在规划多智能体架构的开发者也适合想把现有单 Agent 项目改造成协作式系统的同学参考。读完你会理解为什么 Session 不能被简单当作“聊天记录容器”也会拿到一套可以直接运行验证的轻量通信示例。1. 先理解 Agent、Session以及“通信”到底在解决什么问题1.1 Agent 不止是“会对话的机器人”如果你只把 Agent 理解为“一个接入了大模型的聊天机器人”那么在讨论 Agent 通信时很容易走偏。Agent 的完整定义更像是一个具备感知、决策、行动能力的自治单元它接收用户的自然语言目标把它拆解成可执行步骤选择合适的工具或调用外部服务最后将结果整理后返回给用户。在现代智能体框架里一个 Agent 通常包含以下组成部分模型入口大语言模型LLM或本地部署模型负责推理和指令生成Prompt 或指令模板定义 Agent 角色、行为边界、输出格式工具集合包括代码执行器、搜索引擎、内部 API、数据库操作等记忆组件短期记忆来自当前对话上下文长期记忆往往来自向量数据库或结构化存储会话状态包括当前任务目标、已完成步骤、待处理事项和中间产物。正因为 Agent 是一个“能做事”的单元它不可能永远单打独斗。一个复杂任务如果拆给多个专业化 Agent 去做就必然涉及任务分发、结果汇聚和异常上报这就是跨 Agent 通信出现的根本原因。1.2 Session 在 AI Agent 项目中到底指什么Session 这个词来自 Web 开发语境常见的是 Cookie/Session 用户会话Node.js 里的 express-session以及今天各类 AI 应用里的对话 Session。笼统地说Session 是“一段时间内的交互上下文容器”。在 Agent 项目中Session 要比“聊天记录”重得多。它一般包含用户标识这是哪个用户、哪个项目发起的会话消息历史用户历史输入和 Agent 历史输出上下文变量比如当前代码仓库路径、语言、需求文档中间状态Agent 是否正在等待工具执行结果是否处于人工审批状态资源句柄已经打开的数据库连接已经申请的临时凭证等。简单理解Session 是 Agent 上下文的最小隔离单元。每个 Session 代表一条相对独立的执行线索不同 Session 之间的上下文默认互不可见。1.3 既然 Session 是隔离的为什么还要“跨”通信隔离是安全与稳定的基石但现实业务里很多协作场景天然就是“跨 Session”的。举一个很常见的例子用户 A 在 Session 1 里让“需求分析 Agent”整理一份功能清单用户 B 在 Session 2 里让“代码开发 Agent”实现同一个需求。如果两个 Session 完全隔离用户 A 得到的 PRD 无法自动传给用户 B用户 B 的 Agent 也无法读取用户 A 的分析结论。为了让两个 Session 协作你需要一个“Agent 之外”的共享通信层。再看一个例子一个 Agent 任务执行到一半用户关闭浏览器或者网络中断Session 被回收。等用户重新打开页面理想情况是无论从哪个入口进入都能恢复之前的任务进度。这意味着 Session 背后的状态不能只存在内存里也不能只绑定当初那一个会话通道。所以可以提炼出两个核心论点“跨 Agent 通信”解决的是不同智能体之间的协同问题“跨 Session 通信”解决的是不同对话上下文的数据共享、状态迁移与任务接力问题。2. 单体 Agent 的能力边界为什么通信会成为项目瓶颈2.1 上下文窗口不是万能的自从大语言模型流行以来大家习惯用“把更多内容塞进 Prompt”来解决复杂问题。但上下文窗口总有上限即使窗口很大塞入过多无关历史信息也会导致注意力分散、回答质量下降、Token 成本飙升。如果把多个 Agent 的全部记忆都塞进同一个上下文项目很快就会不可维护。跨 Agent 通信的价值在于不把全部信息集中在一个上下文里而是让不同 Agent 只维护自己需要的那部分上下文通过消息去同步关键结果。这样做既减少了 Token 消耗也让每个 Agent 的职责更清晰。2.2 Session 隔离带来了稳定也带来了孤岛随便打开一个聊天类 AI 产品你会发现每次新建对话都是从空白开始。这种设计对普通问答很友好因为不会串上下文。但对 Agent 来说Session 隔离可能带来几个问题任务中断后难以恢复新开 Session 无法自动拿到上一个 Session 的任务状态多角色协作困难用户需要手动复制粘贴 Agent A 的输出再发给 Agent B无法沉淀公共知识每个 Agent 都从零开始理解项目背景每次都要重新解释规则。Session 的隔离性保证了“一个会话内部的上下文不打架”这是好的设计。但当我们希望 Agent 具备持续工作能力或者希望在多个会话之间共享最终产物时就必须引入比 Session 更上层的通信通道。2.3 从单体到多体通信逐渐成为一等公民单 Agent 架构下通信是“用户和 Agent 之间”的对话。多 Agent 架构下除了用户与 Agent 的对话还有 Agent 与 Agent 之间的消息传递、结果同步、任务回执。下面用一组简单的文本图示意用户A Session-1 - 需求Agent - 共享消息总线 用户B Session-2 - 开发Agent - 共享消息总线 共享消息总线 - 测试Agent - 生成报告在这种结构里消息总线、共享存储、任务队列等通信组件不再只是可选优化项而是系统的主动脉。如果没有显式设计Agent 之间往往只能靠人肉复制、共享数据库或隐蔽的全局变量来传递数据越往后越难以调试和维护。3. 跨 Agent 通信与跨 Session 通信的实现模式3.1 跨 Agent 通信的几种典型模式跨 Agent 通信没有统一标准不同框架会有不同实现策略。常见的模式可以整理成长对比通信模式核心思路适用场景需要注意的问题消息队列模式发送方把消息写入队列接收方轮询或订阅异步任务、分布式 Agent需要处理消息丢失与重复消费共享黑板模式多个 Agent 读写同一个共享存储空间协同解题、多阶段流水线需要约定数据结构和写入权限直接调用模式Agent A 直接调用 Agent B 的方法或 API单体应用内协作耦合度高容易互相阻塞事件广播模式Agent A 发出事件关心该事件的 Agent 自行处理订阅式通知、状态同步需要事件路由和去重机制编排协调模式中央 Orchestrator 负责任务分发和结果回收复杂工作流编排节点容易成为单点瓶颈实际项目里几种模式常常混用。比如任务分发走编排模式Agent 之间的阶段性结果走消息队列长期共享知识走共享黑板异常告警走事件广播。3.2 跨 Session 通信的本质Session 是工作区不是牢房很多人设计 Session 时默认把 Session 当成“一个完全封闭的沙箱”。但从工程角度看Session 更应该被理解为一个带有临时状态的工作区。工作区内部的数据当然应该互相隔离但工作区可以对外提供明确的导入导出协议。想要实现跨 Session 通信通常有两条路径通过共享外部存储每个 Session 把需要共享的成果写入同一个数据库或对象存储其他 Session 通过主键或标签读取通过消息总线转发Session A 不直接访问 Session B 的上下文而是把消息投递到 Session B 订阅的频道由 Session B 自己决定是否消费。第二种路径更符合“最小权限”原则。Session 之间不需要彼此知道对方的内部细节只需要共同遵守一个消息协议。3.3 消息协议设计是第一步无论选择哪种通信模式先要定义消息协议。一个松耦合的 Agent 消息通常包含以下字段字段示例作用message_idmsg_20250212_001全局唯一标识幂等与排重依赖它sender_agentrequirement_agent标识发送方 Agentreceiver_agentcoding_agent标识接收方 Agent支持广播时可填 allsender_sessionsession_a_001来源会话 IDreceiver_sessionsession_b_002目标会话 ID可空表示不限定msg_typetask / result / error / heartbeat消息类型接收方据此选择处理逻辑content结构化 JSON实际业务数据或任务指令timestamp2025-02-12T10:00:00Z时间戳用于排序和延迟分析协议设计得越清晰后续做多 Agent 调度、Session 恢复、问题追踪就越容易。一旦协议混乱任何通信框架都会变成难以维护的垃圾通道。下面定义一个最小可运行的消息结构message { id: msg_20250212_0001, sender_agent: requirement_agent, receiver_agent: coding_agent, sender_session: session_user_a, receiver_session: session_user_b, msg_type: task, content: { task: 实现用户登录接口, requirement_doc_id: DOC-20250212 }, created_at: 2025-02-12T10:00:00Z }4. 实战搭建一个支持跨 Agent、跨 Session 的轻量协作框架为了把上面的概念落到代码里本文用 Python 标准库实现一个简化版“消息总线 Session 工作区”。考虑到很多读者本地不一定会安装 Redis 或专门的消息队列组件这里先采用 SQLite 做持久化存储。SQLite 足够轻量代码复制后可以直接运行。4.1 项目设计与文件结构示例的核心设计思路如下MessageBus负责保存和转发 Agent 之间的消息SessionWorkspace负责跨 Session 共享任务状态BaseAgent所有 Agent 的父类封装发送与轮询消息的方法ResearchAgent 与 ReportAgent两个简单的演示 Agent分别运行在不同 Session 中main.py模拟用户 Session-1 把分析任务发送给 Session-2 的过程。文件结构如下agent_comm_demo/ ├── message_bus.py ├── session_workspace.py ├── base_agent.py ├── agents.py ├── main.py └── agent_comm.db # 运行时自动生成4.2 消息总线实现MessageBus第 1 步新建message_bus.py实现最基础的消息持久化与轮询# agent_comm_demo/message_bus.py import json import sqlite3 import uuid from datetime import datetime, timezone class MessageBus: def __init__(self, db_pathagent_comm.db): self.db_path db_path self._init_db() def _connect(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self): conn self._connect() conn.execute( CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, sender_agent TEXT NOT NULL, receiver_agent TEXT NOT NULL, sender_session TEXT NOT NULL, receiver_session TEXT, msg_type TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, processed INTEGER DEFAULT 0 ) ) conn.commit() conn.close() def send( self, sender_agent: str, receiver_agent: str, sender_session: str, content: dict, msg_type: str message, receiver_session: str None, ) - str: message { id: str(uuid.uuid4()), sender_agent: sender_agent, receiver_agent: receiver_agent, sender_session: sender_session, receiver_session: receiver_session, msg_type: msg_type, content: content, created_at: datetime.now(timezone.utc).isoformat(), } conn self._connect() conn.execute( INSERT INTO messages (id, sender_agent, receiver_agent, sender_session, receiver_session, msg_type, content, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) , ( message[id], message[sender_agent], message[receiver_agent], message[sender_session], message[receiver_session], message[msg_type], json.dumps(content, ensure_asciiFalse), message[created_at], ), ) conn.commit() conn.close() return message[id] def poll(self, agent_name: str, session_id: str None, limit: int 10): conn self._connect() sql SELECT * FROM messages WHERE receiver_agent ? AND processed 0 params [agent_name] if session_id: sql AND (receiver_session IS NULL OR receiver_session ?) params.append(session_id) sql ORDER BY created_at ASC LIMIT ? params.append(limit) rows conn.execute(sql, params).fetchall() messages [dict(row) for row in rows] for msg in messages: msg[content] json.loads(msg[content]) conn.close() return messages def mark_done(self, message_id: str): conn self._connect() conn.execute( UPDATE messages SET processed 1 WHERE id ?, (message_id,) ) conn.commit() conn.close()这里的核心设计要点是消息落库而不是只放在内存这样 Session 中断后消息不会立刻丢失receiver_session允许为空表示“接收 Agent 所在的所有会话都可以看见”processed字段用于防止 Agent 重复处理同一条消息发送时使用uuid4生成唯一 ID方便追踪和排重。4.3 跨 Session 工作区实现SessionWorkspace跨 Session 通信经常会遇到“Session 1 写入了一个实体Session 2 需要读取这个实体状态”的场景。这里为多个会话提供共享事件表# agent_comm_demo/session_workspace.py import json import sqlite3 from datetime import datetime, timezone class SessionWorkspace: 轻量级跨 Session 共享存储以 entity_id 作为关联主键 def __init__(self, db_pathagent_workspace.db): self.db_path db_path self._init_db() def _connect(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row return conn def _init_db(self): conn self._connect() conn.execute( CREATE TABLE IF NOT EXISTS workspace_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity_id TEXT NOT NULL, session_id TEXT NOT NULL, event_type TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT NOT NULL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_entity ON workspace_events(entity_id) ) conn.commit() conn.close() def append_event(self, entity_id: str, session_id: str, event_type: str, payload: dict): conn self._connect() conn.execute( INSERT INTO workspace_events (entity_id, session_id, event_type, payload, created_at) VALUES (?, ?, ?, ?, ?) , ( entity_id, session_id, event_type, json.dumps(payload, ensure_asciiFalse), datetime.now(timezone.utc).isoformat(), ), ) conn.commit() conn.close() def get_entity_events(self, entity_id: str, limit: int 50): conn self._connect() rows conn.execute( SELECT * FROM workspace_events WHERE entity_id ? ORDER BY created_at DESC LIMIT ? , (entity_id, limit), ).fetchall() events [dict(row) for row in rows] for event in events: event[payload] json.loads(event[payload]) conn.close() return eventsSessionWorkspace 里没有区分“哪个 Session 能读、哪个 Session 不能读”实际项目中应该在读取前增加权限校验。这里为了演示跨 Session 数据共享的可行性先不做复杂鉴权。4.4 Agent 基类与两个演示 Agent有了消息总线和工作区后再来设计 Agent 基类# agent_comm_demo/base_agent.py import time class BaseAgent: def __init__(self, agent_name: str, session_id: str, bus: MessageBus, workspace: SessionWorkspace): self.agent_name agent_name self.session_id session_id self.bus bus self.workspace workspace def send(self, receiver_agent: str, content: dict, receiver_sessionNone, msg_typemessage): return self.bus.send( sender_agentself.agent_name, receiver_agentreceiver_agent, sender_sessionself.session_id, contentcontent, msg_typemsg_type, receiver_sessionreceiver_session, ) def receive_once(self): messages self.bus.poll(self.agent_name, self.session_id) for msg in messages: self.handle(msg) self.bus.mark_done(msg[id]) return messages def run_listen(self, max_loops10, interval0.5): for _ in range(max_loops): received self.receive_once() if not received: time.sleep(interval) def handle(self, msg): raise NotImplementedError然后定义两个实际 Agent# agent_comm_demo/agents.py from base_agent import BaseAgent class ResearchAgent(BaseAgent): 负责做资料分析的 Agent位于 Session-1 def handle(self, msg): print(f[ResearchAgent] 收到消息来自 {msg[sender_agent]}: {msg[content]}) if msg[msg_type] task: # 模拟调研分析并产生结果 conclusion { summary: 登录模块推荐引入短时令牌 无状态刷新令牌机制, risk: 需要防止令牌重放建议增加设备指纹, } # 把分析结果写入共享工作区entity_id 可作为多个 Session 的关联键 self.workspace.append_event( entity_idTASK-001, session_idself.session_id, event_typeresearch_done, payloadconclusion, ) print(f[ResearchAgent] 完成分析并写入 TASK-001 状态) class ReportAgent(BaseAgent): 负责生成报告的 Agent位于 Session-2 def handle(self, msg): print(f[ReportAgent] 收到消息来自 {msg[sender_agent]}: {msg[content]}) if msg[msg_type] task: entity_id msg[content].get(entity_id, TASK-001) events self.workspace.get_entity_events(entity_id) print(f[ReportAgent] 从 SessionWorkspace 读到 {len(events)} 条共享事件) for event in events: print(f - {event[event_type]}: {event[payload]}) print([ReportAgent] 报告生成完成跨 Session 协作验证成功)4.5 主流程跨 Agent 与跨 Session 联动接下来写启动脚本# agent_comm_demo/main.py from message_bus import MessageBus from session_workspace import SessionWorkspace from agents import ResearchAgent, ReportAgent def main(): bus MessageBus(agent_comm.db) workspace SessionWorkspace(agent_workspace.db) # Session-1 中运行 ResearchAgentSession-2 中运行 ReportAgent research_agent ResearchAgent( agent_nameresearch_agent, session_idsession-user-a, busbus, workspaceworkspace, ) report_agent ReportAgent( agent_namereport_agent, session_idsession-user-b, busbus, workspaceworkspace, ) # 模拟Session-1 的用户要求分析登录方案同时把任务接力给 Session-2 的报告 Agent task_message { task: 分析登录模块安全建议, entity_id: TASK-001, source_session: session-user-a, } message_id research_agent.send( receiver_agentreport_agent, contenttask_message, msg_typetask, receiver_sessionsession-user-b, ) print(f消息已投递message_id{message_id}\n) # Session-1 的 ResearchAgent 执行自己的调研任务并把结果写入共享工作区 research_agent.receive_once() # Session-2 的 ReportAgent 开始轮询并消费消息读取跨 Session 状态 report_agent.run_listen(max_loops3, interval0.5) if __name__ __main__: main()运行命令如下cd agent_comm_demo python main.py预期输出大致如下消息已投递message_id60f57d0d-xxxx-xxxx-xxxx-xxxxxxxxxxxx [ResearchAgent] 收到消息来自 report_agent: ... [ResearchAgent] 完成分析并写入 TASK-001 状态 [ReportAgent] 收到消息来自 research_agent: ... [ReportAgent] 从 SessionWorkspace 读到 1 条共享事件 - research_done: {summary: 登录模块推荐引入短时令牌 无状态刷新令牌机制, risk: 需要防止令牌重放建议增加设备指纹} [ReportAgent] 报告生成完成跨 Session 协作验证成功这里需要说明实际代码中 ResearchAgent 在自己 Session 内还可能有自己收到的用户指令。示例把消息发送和业务执行简化成同步顺序是为了更直观地展示消息流。真实系统里两个 Agent 通常各自运行在线程、进程或独立服务中。4.6 这个示例说明了什么示例看起来简单但已经覆盖了跨 Agent 与跨 Session 通信的几个关键机制发送方 Agent 和目标 Agent 不直接依赖对方的具体实例只依赖 MessageBus目标 Agent 所在 Session 的信息通过receiver_session修饰避免消息被无关 Session 的同一类型 Agent 抢走跨 Session 的共享数据没有塞进聊天上下文而是通过 SessionWorkspace 持久化新 Session 可以通过entity_id把之前 Session 的阶段成果恢复或继续使用。这就是“通信不是上下文拼接”的落地体现。5. 进阶用 Redis 或消息队列承载更复杂通信SQLite 方案适合本地验证和学习一旦 Agent 分散到多个进程或多台服务器建议引入集中式消息组件。常见选择包括 Redis Stream、RabbitMQ、Kafka 以及各类云厂商的托管消息服务。以 Redis 为例你可以把某个 Agent 的收件箱设计成一个 List 结构import json import redis # 需要安装 redis 库生产环境不要把密码写在代码里 r redis.Redis(hostlocalhost, port6379, db0) # 跨 Agent 发送 message { id: msg_20250212_1001, sender_agent: research_agent, receiver_agent: report_agent, sender_session: session-user-a, receiver_session: session-user-b, msg_type: task, content: {task: 生成上线报告}, } r.lpush(agent:report_agent:inbox, json.dumps(message))接收方只需要轮询或阻塞读取# 消费者侧 while True: raw r.brpop(agent:report_agent:inbox, timeout5) if raw is None: continue msg json.loads(raw[1]) print(msg) # TODO: 根据 msg 调用 Agent 处理方法处理成功后确认Redis 方案的优点是部署简单、响应快缺点是需要自己处理消费者故障、消息重复和重试策略。如果追求可靠投递和消息轨迹可考虑 RabbitMQ 或 Kafka。不要迷信某一个组件先分析业务是否需要“概率丢失可接受低延迟优先”还是“必须不丢消息”。6. 常见问题与排查思路6.1 消息发出去了但是接收 Agent 一直没收到问题现象常见原因解决思路接收 Agent 轮询为空receiver_agent拼写不一致对比发送方和接收方注册名同一 Agent 类型有多个会话实例没有传receiver_session消息被错误隔离检查轮询时的 Session 过滤条件消息一直为未处理状态Agent 消费逻辑异常中断查看日志确认是否有异常抛出排查时先把数据库打开看messages表里的receiver_agent、receiver_session和processed字段基本就能定位问题。6.2 Session 恢复后上下文不完整Session 恢复不等于“把历史消息重新灌给模型”。如果只保存聊天记录Agent 执行到一半的任务状态仍然会丢失。更稳妥的做法是把会话事件按entity_id持久化每次关键状态变更都记录事件类型和快照恢复时先加载最近快照再回放未完成事件。6.3 多个 Session 同时写一份共享数据导致覆盖这种问题在引入跨 Session 共享存储后非常常见。解决思路有使用数据库事务和行级锁写入时带上版本号或更新时间不直接修改共享实体而是以追加事件的方式记录变更后续读取时归并出最新状态。上面的 SessionWorkspace 采用“事件追加”而不是“覆盖更新”正是为了规避并发覆盖问题。6.4 消息被重复消费常见原因包括消费者处理成功但没有及时标记processed网络抖动导致确认消息丢失消息重试机制设计不合理。建议给每条消息设计全局唯一 ID并在消费者侧做幂等处理。比如处理消息前先查processed_record表发现已处理就跳过。6.5 Session 与 Agent 概念混淆导致权限过度开放还有一种典型问题开发者以为“跨 Session 通信 让所有 Session 数据互相可见”于是把整个上下文对象传给了共享总线。这样虽然通信方便但会把用户 A 的敏感信息暴露给用户 B。正确的做法是只发送需要协作的最小字段并在共享层增加权限判断。7. 工程最佳实践与安全建议7.1 设计消息协议时显式区分 Agent 与 Session很多通信出问题的根源是把 Agent 标识和 Session 标识混在一起。一个 Agent 类型可以同时运行在多个 Session 里比如每个用户都有一个“个人数据分析 Agent”。因此消息里必须同时包含目标 Agent 类型决定哪个处理逻辑响应目标 Session决定该类型 Agent 的哪个实例接收。如果缺少 Session 维度消息可能被任意实例消费造成跨用户数据污染。7.2 用最小字段原则传递上下文跨 Session 共享信息时不要直接共享原始 Prompt、内部工具调用历史或完整记忆库。推荐的做法是定义“对外 DTO”例如任务状态、结论摘要、产物 ID。其他 Agent 需要细节时再按 ID 去读取可信数据源。# 推荐做法发送给其他 Agent 的 content 尽量精简 content { entity_id: TASK-001, stage: analysis_done, result_summary: 建议采用短时令牌方案, next_action: generate_report }这种设计既降低消息体大小也减少敏感上下文扩散。7.3 跨 Session 恢复时先鉴权再加载Session 恢复功能很容易被误用成“任意 Session 可读取他人数据”。实际项目中至少要增加以下校验用户身份是否与该 Session 的归属者一致目标实体是否允许跨 Session 访问是否有登录态、临时令牌或回调签名验证。尤其是当 Session 被用于浏览器端长连接、移动端 App 或第三方 API 时不要信任客户端传入的 session_id。必须把 session_id 与当前用户凭证绑定后重新校验。7.4 对 Agent 通信链路做可观测性处理Agent 的推理过程本来就不容易调试通信链路若再不可见排查成本会非常高。建议为 Agent 消息增加贯穿全局的 trace_id 或 correlation_id在每个 Agent 的入口、出口打日志[2025-02-12 10:00:01] [trace_idabc123] research_agent send - report_agent [2025-02-12 10:00:02] [trace_idabc123] report_agent receive from research_agent [2025-02-12 10:00:03] [trace_idabc123] report_agent call workspace.get_entity_events这样当任务在某一步失败时能快速定位是消息丢失、会话权限错误还是 Agent 处理异常。7.5 不要把通信层设计成“万能数据池”通信层解决的是消息流转如果所有业务状态都往里塞最终会变成一个大泥潭。建议遵循这样的分层思想Agent 自己的内部记忆 Agent 私有协作结果 按 entity_id 存入共享存储消息总线 只负责轻量通知和指令外部系统状态 由对应 Agent 通过工具访问不复制到通信层。这样做的好处是即使某一天更换消息队列组件业务代码也能保持相对稳定。7.6 数据清理与权限回收跨 Session 与跨 Agent 通信意味着数据会被多份拷贝这扩大了数据暴露面。定期清理已完成的临时消息、归档过期 Session、撤销 Agent 的失效凭证是生产系统必须做的事。建议至少做到消息设置 TTL 或归档周期Session 有效期与用户登录态有效期绑定Agent 访问外部系统时使用临时令牌而不是长期密钥。8. 写在最后通信设计中最容易忽略的三件事第一不要把“跨 Session 通信”等同于“取消 Session 隔离”。Session 仍然是权限、上下文和状态的边界跨通信应该是一扇有门禁的门而不是直接把墙拆掉。第二消息协议是团队协作的契约。即使项目初期只有两个 Agent也建议把收发双方标识、会话标识、消息 ID、类型字段规范起来否则后期接入第三个、第四个 Agent 时会越来越痛苦。第三先确认业务真的需要跨 Agent 通信。如果单个 Agent 把任务拆成步骤、按顺序执行就可以完成没必要为了“多智能体”这个词强行引入通信层。跨 Agent 通信能解决的是职责边界明显、需要并行处理、不同会话间需要接力或分工协作的场景。如果你正在搭建自己的 Agent 应用建议先按照本文的轻量示例做一次技术验证把消息能跨 Agent 传递、状态能跨 Session 恢复这两个核心节点跑通再考虑引入 Redis、RabbitMQ 等生产级组件。通信层是慢变量越早设计清楚后面的迭代越省力。
