企业官网建设流程全解析

Agent 项目一旦从单轮问答走向真实业务，最先暴露的问题往往不是提示词写得不够好，而是多个 Agent 之间不知道怎么传结果、多轮会话之间怎么续上上一次的进度。很多团队在演示环境里用单个 Agent 加一个聊天窗口跑得很顺，进入开发环境后却发现：编排层把任务拆给了执行 Agent，结果执行 Agent 跑完没人拿结果；用户刷新页面、换设备或者隔十分钟回来，Session 里的上下文全部丢失，只能重头再来。跨 Agent 通信和跨 Session 通信这两个能力，就是用来解决这两类问题的。本文从概念讲起，给出一个不依赖第三方框架的最小可运行示例，先实现进程内消息总线，再把会话历史落到 SQLite，最后讨论生产环境下的选型取舍。学完后，你可以把这套思路用于自研 Agent 编排模块，也可以作为理解常见编排框架消息机制的基础。

1. 先摸清两种通信分别卡在哪里

1.1 跨 Agent 通信是“多个角色之间传任务”

如果整个业务流程只由一个 Agent 在单次请求内完成，其实不需要通信。需要通信的场景是任务被拆分：规划 Agent 负责拆解目标，执行 Agent 负责调用工具或生成内容，审核 Agent 负责检查结果，入口 Agent 负责汇总。这些角色之间必然要交换“任务描述”“执行结果”“错误信息”“状态变更”等数据，这就是跨 Agent 通信。

跨 Agent 通信和普通 RPC 调用有一个本质区别：Agent 内部依赖了模型调用，而模型调用存在推理慢、超时、返回格式不稳定、中间步骤失败等不确定性。因此消息不能只传一个“文本”，还需要携带完整的任务标识、发送方、接收方、消息类型、优先级和可追溯信息。跨 Agent 通信的设计重点不是“能不能把变量传过去”，而是“任务失败后由谁重试、结果返回后由谁消费、多个 Agent 并发执行时会不会冲突”。

1.2 跨 Session 通信是“换一个会话仍然延续业务”

Session 这个词在不同场景下的含义不同。Web 开发里的 Session 通常指服务端保存的用户会话状态，用于维持登录态；在 Agent 应用里，Session 更多指向一次对话过程，例如用户在聊天窗口里连续发送多条消息，或者一条任务在浏览器刷新后仍需继续执行。跨 Session 通信要解决的核心问题是：同一业务场景下，新的技术 Session 能不能拿到历史 Session 产生的上下文。

最直观的例子是用户第一次访问时提交了一个“整理一份面试复习计划”的任务，Agent 生成了前半部分；用户第二天打开页面，浏览器新建了一个 Session，如果系统不去查询历史记录，用户只能重新发起请求。真正可用的 Agent 产品必须把“业务任务”和个人技术 Session 解耦，用业务 ID 关联历史，用存储介质保存进度，让新的会话可以恢复旧的任务。

1.3 用一张表分清两种通信

这两类通信并不孤立。一个完整的 Agent 系统中，跨 Agent 通信产生的消息记录同时也是跨 Session 恢复的数据来源；Session 持久化保存的内容里，核心部分往往就是 Agent 之间传递过的事件和结果。

2. 设计通信前先约定会话、消息和上下文

2.1 用统一 ID 把零散请求串成链路

跨 Session 恢复之所以难，并不是因为没有存储方案，而是因为没有把“业务链路”标识出来。一次完整的用户目标可能经历多个技术 Session、多个 Agent、多次工具调用，如果每段过程只有各自的随机 ID，恢复时就没有办法按业务维度聚合。常见做法是同时存在两类 ID：

• task_id：表示一次业务目标的完整生命周期，例如“导出本月销售报表”这一个目标。
• session_id：表示一次前端会话，例如用户某一天打开的聊天窗口。

当用户换了一个session_id继续同一条任务时，系统应该通过task_id或者用户 ID 找到历史会话记录，再把历史记录追加为后续处理的上下文。否则服务端只会看到两个互不关联的 Session，任务自然无法延续。

2.2 消息不要用裸字符串，用信封结构

很多自研 Agent 项目在初期会把通信写成“把这段文本发给另一个 Agent”。这种方式在只有两个角色时勉强能跑，角色一多就会出现问题：执行 Agent 不知道这条消息是该立即执行还是仅通知，也不清楚消息来自规划器还是用户。消息需要一个稳定的信封结构。

{ "msg_id": "msg_001", "task_id": "task_8f21a0c1", "sender": "planner", "receiver": "executor", "msg_type": "task.execute", "payload": { "task": "导出本月销售报表" }, "create_time": "2025-01-01T12:00:00Z" }

msg_type比随机文本更可靠，因为后续代码可以根据类型做分发、过滤、监控和重试。sender和receiver让通信可追踪，payload只放业务数据。消息结构一旦确定，跨 Agent 通信才有可能沉淀出可测试的协议，而不是靠字符串拼接。

2.3 上下文要区分“原始记录”和“可恢复快照”

跨 Session 恢复时，一个常见误区是把全部历史消息原样带入下一次模型调用。这么做在早期有效，但任务执行几十步后，历史消息可能超过模型窗口，还会导致响应变慢、费用升高。更合理的做法是分层管理：

• 原始消息记录：每条 Agent 消息都写入日志表，用于故障排查和审计。
• 压缩快照：每一轮或每几轮执行后，生成一份“当前任务做到哪一步、已确认信息、下一步计划”的摘要。
• 近期明细：只保留最近几条原始消息，让模型能看到用户刚说过什么。

恢复新 Session 时，把“压缩快照 + 近期明细”拼接到提示词里，而不是把整张历史表都塞进去。这条原则在后面实现示例时会用到。

3. 最小可运行示例：用进程内消息总线打通两个 Agent

3.1 环境准备与项目结构

这个示例使用 Python 3.10 以上版本，只依赖标准库。这样做的目的是先把通信机制讲清楚，避免读者在理解业务前还要先配置 Redis 或消息中间件。实际项目不要照搬内存方案，但可以把这里的消息模型和角色划分保留下来。

先创建目录结构：

agent_comms_demo/ ├── messages.py ├── bus.py ├── agents.py └── main.py

3.2 定义消息模型

消息模型对应上一节提到的信封结构。使用dataclass定义字段，并给msg_id和create_time设置默认值，这样调用方可以少传两个字段。

# messages.py from dataclasses import dataclass, field import time import uuid @dataclass class Message: sender: str receiver: str msg_type: str payload: dict msg_id: str = field(default_factory=lambda: uuid.uuid4().hex) create_time: float = field(default_factory=time.time)

这里的关键点是msg_id必须由发送方生成，而不是由接收方生成。因为接收方如果需要做幂等处理，就要用发送方的消息 ID 去重。如果 ID 在接收时重新生成，重复消息就无法识别。

3.3 实现一个线程安全的内存消息总线

内存消息总线的思路是：每个 Agent 对应一个队列，send方法按receiver把消息投递到对应队列，receive方法从该 Agent 自己的队列取消息。加锁是为了避免多线程同时读写同一个队列时出现数据错乱。

# bus.py from collections import defaultdict, deque import threading from messages import Message class MessageBus: def __init__(self): self._queues = defaultdict(deque) self._lock = threading.Lock() def send(self, message: Message): with self._lock: self._queues[message.receiver].append(message) def receive(self, agent_name: str) -> Message | None: with self._lock: queue = self._queues[agent_name] if queue: return queue.popleft() return None def size(self, agent_name: str) -> int: with self._lock: return len(self._queues[agent_name])

receive在队列为空时返回None，而不是阻塞等待。这种轮询方式在单进程示例里足够直观，也容易理解和调试。生产环境建议换成阻塞队列或消息中间件的消费者组，否则空轮询会浪费 CPU。

3.4 编写两个轻量 Agent

示例里规划 Agent 负责把用户请求拆成三步，执行 Agent 负责消费并回传最终结果。实际产品中，handle方法内部会调用模型或工具，这里用字符串模拟，让读者只看通信骨架。

# agents.py from messages import Message class PlannerAgent: name = "planner" def handle(self, msg: Message): task = msg.payload.get("task", "") plan = [ f"校验任务：{task}", "选择执行工具并生成命令", "验证产出文件并生成交付说明", ] print(f"[planner] 拆解任务 -> executor") return Message(self.name, "executor", "task.execute", {"plan": plan}) class ExecutorAgent: name = "executor" def handle(self, msg: Message): plan = msg.payload.get("plan", []) summary = " -> ".join(plan) print(f"[executor] 执行完成 -> entry") return Message(self.name, "entry", "task.done", {"summary": summary})

注意发送给其他 Agent 的消息类型和普通业务消息不一样。task.execute表示执行任务，task.done表示任务完成。接口设计上要避免让接收方通过猜测文本语义来决定行为，类型字段才是分发依据。

3.5 入口逻辑与运行结果

入口函数先让“用户”把初始请求投递到 planner 的队列，然后依次消费 planner 和 executor 的队列，最后从 entry 队列取结果。

# main.py from bus import MessageBus from messages import Message from agents import PlannerAgent, ExecutorAgent def drain(bus: MessageBus, agent_name: str, agent, max_iter: int = 10): for _ in range(max_iter): msg = bus.receive(agent_name) if msg is None: break bus.send(agent.handle(msg)) def main(): bus = MessageBus() planner = PlannerAgent() executor = ExecutorAgent() # 用户请求以消息形式进入 bus.send(Message("user", "planner", "user.request", {"task": "导出本月销售报表"})) drain(bus, "planner", planner) drain(bus, "executor", executor) for _ in range(10): final_msg = bus.receive("entry") if final_msg is None: break print("最终输出:", final_msg.payload["summary"]) if __name__ == "__main__": main()

在项目目录下执行：

python main.py

预期输出类似：

[planner] 拆解任务 -> executor [executor] 执行完成 -> entry 最终输出: 校验任务：导出本月销售报表 -> 选择执行工具并生成命令 -> 验证产出文件并生成交付说明

这个示例说明了一个关键设计：planner 和 executor 没有互相持有实例，它们只依赖消息总线。后续替换通信实现时，两个 Agent 的外部接口可以保持不变。

4. 跨 Session：把消息和上下文落到 SQLite

4.1 为什么不能只留在内存

进程内消息总线解决了“同一个进程内多 Agent 通信”的问题，但没有解决“进程重启后数据丢失”和“多个会话之间状态不共享”的问题。用户刷新页面后，如果状态还停留在 Python 进程的内存里，进程一重启或者前端换了一个实例，历史上下文就找不到了。

跨 Session 通信的基础是持久化。只要 Agent 交互的历史被写入稳定存储，新的 Session 就可以通过task_id或user_id找回旧数据。本节用 SQLite 做演示，因为它是零配置的文件数据库，适合学习阶段；生产环境可以使用 MySQL、PostgreSQL 或 Redis 等组件，设计思路不变。

4.2 设计保存模型

需要两张表：一张存业务会话概要，一张存 Agent 消息明细。实际场景还会增加工具调用记录、产物地址、错误日志等，示例里只保留最小字段。

CREATE TABLE IF NOT EXISTS session ( session_id TEXT PRIMARY KEY, user_id TEXT, task_id TEXT, context TEXT, updated_at TEXT ); CREATE TABLE IF NOT EXISTS message_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, sender TEXT NOT NULL, receiver TEXT NOT NULL, msg_type TEXT NOT NULL, payload TEXT NOT NULL, create_time TEXT NOT NULL );

session表里的context保存的是上一节所说的压缩快照，message_log表保存的是原始消息。恢复会话时，两种数据配合使用。

4.3 用 SessionStore 封装读写

SessionStore的核心方法有三个：保存一条消息、读取某个会话的全部消息、保存业务上下文快照。注意 SQLite 连接在示例里设置了check_same_thread=False，这只说明该连接允许多线程调用，实际写操作仍然需要靠事务保证一致性。

# session_store.py import json import sqlite3 from datetime import datetime, timezone class SessionStore: def __init__(self, db_path="session.db"): self.db_path = db_path self.conn = sqlite3.connect(self.db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self._init_tables() def _init_tables(self): self.conn.executescript(""" CREATE TABLE IF NOT EXISTS session ( session_id TEXT PRIMARY KEY, user_id TEXT, task_id TEXT, context TEXT, updated_at TEXT ); CREATE TABLE IF NOT EXISTS message_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, sender TEXT NOT NULL, receiver TEXT NOT NULL, msg_type TEXT NOT NULL, payload TEXT NOT NULL, create_time TEXT NOT NULL ); """) self.conn.commit() def save_message(self, session_id, sender, receiver, msg_type, payload): now = datetime.now(timezone.utc).isoformat() self.conn.execute( "INSERT INTO message_log(session_id, sender, receiver, msg_type, payload, create_time) " "VALUES (?, ?, ?, ?, ?, ?)", (session_id, sender, receiver, msg_type, json.dumps(payload, ensure_ascii=False), now), ) self.conn.commit() def load_messages(self, session_id): rows = self.conn.execute( "SELECT sender, receiver, msg_type, payload, create_time " "FROM message_log WHERE session_id = ? ORDER BY id", (session_id,), ).fetchall() return rows def save_context(self, session_id, user_id, task_id, context): now = datetime.now(timezone.utc).isoformat() self.conn.execute( """ INSERT INTO session(session_id, user_id, task_id, context, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET user_id = excluded.user_id, task_id = excluded.task_id, context = excluded.context, updated_at = excluded.updated_at """, (session_id, user_id, task_id, json.dumps(context, ensure_ascii=False), now), ) self.conn.commit()

每次修改都执行commit在示例里没有太大问题，但在高频调用场景会带来磁盘 IO 压力。生产项目建议按批提交，或者直接使用为高并发设计的内存型存储。

4.4 演示新的 Session 恢复旧的历史

下面用一段代码模拟跨 Session 场景。第一个session_id是用户昨天的会话，第二个session_id是用户今天重新打开页面产生的新会话。恢复的核心不是把两个 Session 强行合并，而是通过业务用户 ID 找到旧会话，再把旧历史拼接进新会话的处理上下文。

# session_demo.py from session_store import SessionStore def run_turn(store, session_id, user_text): store.save_message(session_id, "user", "planner", "user.request", {"content": user_text}) # 真实场景这里会调用模型和 Agent，本示例只记录结果 store.save_message(session_id, "planner", "entry", "task.done", {"summary": "已经生成：" + user_text}) def main(): store = SessionStore("session.db") # 第一次会话：用户提出需求，Agent 完成两轮处理 run_turn(store, "session-001", "整理一份技术面试复习清单") run_turn(store, "session-001", "重点补充计算机网络部分") # 第二次会话：模拟用户第二天重新打开页面 old_history = store.load_messages("session-001") print("恢复出的旧会话消息条数:", len(old_history)) for row in old_history: print(row["sender"], "->", row["receiver"], row["msg_type"]) # 新会话写入新消息，同时继续引用旧会话 ID run_turn(store, "session-002", "在上次复习清单基础上补充算法题") print("新会话也已保存") if __name__ == "__main__": main()

执行：

python session_demo.py

预期输出：

恢复出的旧会话消息条数: 4 user -> planner user.request planner -> entry task.done user -> planner user.request planner -> entry task.done 新会话也已保存

运行后用 SQLite 命令行可以直接看到表里的数据：

sqlite3 session.db "SELECT session_id, sender, msg_type FROM message_log ORDER BY id;"

这里要特别说明：示例里的session-002并不是跨 Session，真正起作用的是“通过业务用户 ID 找到session-001”这一步。如果代码里只保存了session-002自己的消息，没有关联旧的task_id或user_id，那无论存多少数据都无法实现续聊。

5. 生产环境的演进路径：从内存到 Redis 再到消息中间件

5.1 不同存储方案的适用边界

内存消息总线和 SQLite 只适合教学和原型验证。真实项目需要根据部署形态选择通信存储，这里整理几种常见方案。

选型时不要只看某个组件流行，要先确认你要解决的是“多个 Agent 同步调用”还是“异步任务可靠分发”。示例里的同步轮询方式在 Agent 数量少时够用，但一旦出现一个 Agent 同时处理多个任务，就要给消费过程增加确认机制，否则可能会把一个任务分给多个执行者。

5.2 协议字段和幂等策略要先定

无论底层用哪个组件，消息字段和消息消费语义都应该先定好。消息里最好包含schema_version，方便以后协议升级时做兼容处理。比如第一版字段是payload，第二版改成data，没有版本号就只能靠猜。

消费语义也要明确：是“最多一次”还是“至少一次”。如果允许 Agent 执行失败后重试，消息接收方必须幂等，也就是同一个msg_id执行两次不能产生两份结果。最简单的幂等方案是维护一张processed_message表，处理前先查msg_id是否已经存在，不存在才继续执行。

5.3 学习环境与生产环境的差异

跨 Session 通信在生产环境还要额外处理一个问题：会话数据不能无限增长。用户长时间不使用后，旧会话可以转存为压缩摘要，删除明细；真正需要审计的原始记录可以移到冷存储。

6. 常见问题排查：为什么消息收不到、会话找不回

6.1 消息收不到或没人消费

如果消息总线投递正常但接收方取不到消息，优先按以下顺序检查。

1. 先确认receiver名称是否一致。很多问题出在发送方写的是executor，接收方注册的名字是executor_agent。
2. 再确认send是否真的执行了。检查总线某个队列的大小，如果大小为 0，说明问题在发送方。
3. 接着确认取消息的循环是否在发送动作之前就已经退出。示例里drain使用空队列退出，如果时序写错，接收方可能先轮询一遍发现为空就退出。
4. 最后看角色名是否和业务角色对应。结果投给了entry，消费者却在监听done，同样会拿不到。

6.2 Agent 挂了之后消息丢失

现象是：执行 Agent 处理到一半进程退出，重新启动后任务没有继续。

可能原因是消息已经出队但尚未处理完，进程退出后队列里不再有这条消息。解决方式是把“消息已接收”和“消息已处理完成”分开：接收后先记录日志，处理完成后再发送确认或删除消息。使用 Redis Stream 时可以利用消费者组和 Pending Entries 重投机制；使用数据库时可以在消息表里增加status字段。

6.3 会话记录存在但恢复出来为空

这类问题最常见的根因不是存储坏了，而是查询时用错了session_id。用户在新页面里产生的是session-002，代码却只取了session-002的消息，忘记了通过用户 ID 关联session-001。检查方式是在应用里打印业务用户关联的旧 Session ID，再单独查询数据库确认数据是否存在。

另一种可能是在多个环境切换时把数据库文件指向了不同路径。SQLite 是单文件数据库，测试环境和本地环境用同一个session.db路径会互相覆盖思维混乱。解决方式是让数据库路径和部署环境绑定，并在日志里打印实际路径。

6.4 上下文越来越长导致模型调用失败

现象是刚开发时运行正常，执行几轮后请求超时或者提示超出最大 token 限制。

原因是每次新 Session 都把所有历史明细拼进提示词。正确做法是先把旧历史做摘要，再只保留最近几条原始消息。摘要可以放在session表的context字段里，每次新任务开始前先更新摘要，恢复时读取摘要而不是全量历史。

6.5 多线程读写同一 SQLite 连接报 database is locked

示例里为了简单复用了同一个连接，但多线程高并发写同一个 SQLite 文件时容易出现锁冲突。生产环境不要让多个 Agent 线程共享同一个写连接，推荐做法是每个写操作使用新的短连接，或者把 SQLite 换成真正的数据库服务。