多智能体协作系统设计：基于共享记忆的AI智能体架构与实战-平芜编程栈

1. 从单兵作战到团队协作：为什么需要共享记忆

在AI智能体开发领域，我们常常会遇到一个瓶颈：单个智能体（比如一个OpenClaw实例）的能力是有限的。它可能擅长处理特定类型的任务，比如分析数据、撰写报告或者执行某个API调用，但面对一个需要多步骤、多维度协作的复杂任务时，单个智能体就显得力不从心了。想象一下，你有一个项目需要先进行市场调研，然后根据调研结果设计产品原型，最后撰写一份详细的技术方案。如果让一个智能体从头干到尾，它可能会在切换不同思维模式时“遗忘”之前的上下文，或者因为缺乏特定领域的深度知识而卡壳。

这就是“共享记忆”概念的价值所在。它不再是让一个智能体单打独斗，而是组建一个“特种作战小队”。在这个小队里，每个智能体（OpenClaw实例）都扮演着不同的角色，比如“研究员”、“架构师”或“写手”。它们各自拥有独立的思考和执行能力，但最关键的是，它们共享一个中央“任务白板”和“经验档案库”。这个共享空间记录了任务目标、已完成的步骤、产生的中间结果、遇到的坑以及达成的共识。当一个智能体完成了它的部分工作后，它会将成果和关键上下文写入共享记忆；下一个接手的智能体在开始工作前，会先读取这些记忆，从而无缝地接续任务，仿佛整个流程是由一个拥有“全知视角”的超级智能体完成的。

这种模式的核心优势在于任务解耦与能力复用。复杂的任务被拆解成清晰的子任务，分配给最擅长的智能体去执行。同时，智能体之间通过共享记忆传递的不仅仅是数据，更是“意图”和“上下文”，这极大地减少了沟通损耗和重复劳动。对于开发者而言，这意味着你可以像搭积木一样，组合不同的智能体能力来应对千变万化的需求，而无需每次都从头训练一个“全能”模型。接下来，我们就深入探讨如何构建这样一个协作系统。

2. 架构核心：理解共享记忆的载体与通信机制

要实现多个OpenClaw实例的协作，首要问题是：记忆存在哪里？智能体之间如何安全、高效地交换信息？我们不能依赖智能体自身那不稳定且容量有限的临时记忆，必须建立一个外部的、持久化的记忆中枢。

2.1 记忆载体的选型：数据库 vs 消息队列 vs 向量数据库

根据任务的特性和对记忆的查询需求，我们可以选择不同的技术方案作为共享记忆的载体。

方案一：键值数据库（如Redis）这是实现共享记忆最简单、最快速的方案。我们可以将整个任务流程的上下文抽象为一个大的JSON对象，存储在以任务ID为键的Redis中。每个智能体在执行时，通过任务ID获取整个上下文，更新自己负责的部分后再写回。

优点：速度快，数据结构简单直观，非常适合状态共享和会话管理。
缺点：当记忆内容非常庞大（例如包含长文档、多轮对话历史）时，每次全量读写效率低。更重要的是，它缺乏“语义检索”能力。你无法直接问“之前关于用户画像的结论是什么？”，只能通过预设的结构化字段去获取。
适用场景：任务步骤固定、上下文结构明确、且对延迟极其敏感的轻量级协作流程。

方案二：消息队列（如RabbitMQ, Kafka）这种方案将共享记忆转化为“事件流”。每个智能体完成工作后，会向一个特定的主题（Topic）或队列（Queue）发布一个“事件消息”，消息体内包含了它的工作成果。后续的智能体通过订阅这些主题来获取它们需要的信息。

优点：实现了智能体间的完全解耦，支持异步处理和广播通信，系统扩展性极强。
缺点：消息通常是“一次性”的，消费后若不持久化则历史记忆会丢失。要构建完整的任务上下文，后续智能体需要自己维护或从别处查询历史消息，增加了复杂性。
适用场景：流水线式任务，且各环节相对独立，对事件顺序和流处理有要求的场景。

方案三：向量数据库（如Chroma, Weaviate, Pinecone）这是实现“高级记忆”的推荐方案。我们不仅存储记忆文本本身，还利用嵌入模型（Embedding Model）将文本转换为向量（一组数字），并存储起来。当智能体需要查询记忆时，它可以提出一个自然语言问题，系统将问题也转换为向量，并在向量数据库中搜索与之最相似的记忆片段。

优点：支持基于语义的灵活检索，能从海量记忆中精准定位相关信息。记忆可以分块存储，避免了大文本的传输压力。
缺点：架构比前两者复杂，需要引入嵌入模型，有额外的计算开销。
适用场景：任务上下文复杂、记忆内容非结构化（如长文本、会议纪要）、且需要智能体进行“回忆”和“联想”的复杂协作场景。

对于大多数需要深度协作的OpenClaw任务，我推荐采用“向量数据库为主，键值数据库为辅”的混合架构。向量数据库负责存储所有历史对话、文档片段、决策依据等“经验记忆”，支持语义检索；键值数据库则用来存储当前任务的状态、锁、以及一些简单的配置信息，保证系统的高效运转。

2.2 智能体间的通信协议设计

确定了记忆仓库，我们还需要定义智能体之间“读写”记忆的协议。一个健壮的协议需要包含以下几个关键字段：

{ "task_id": "unique_task_identifier_123", "agent_id": "researcher_01", "action": "read|write|query", "memory_type": "context|decision|artifact", "content": { // 写入时：提供完整的记忆内容 "text": "经过分析，目标用户群体主要为25-35岁的科技从业者...", "metadata": { "step": 2, "keywords": ["用户画像", "年龄分布", "职业"], "confidence": 0.95 } }, "query": "检索时：提供自然语言查询语句", "timestamp": "2023-10-27T08:00:00Z" }

task_id：所有协作智能体的唯一聚合点，确保记忆在正确的任务上下文中被存取。
agent_id：标识记忆的贡献者，便于追溯和权责划分。
action：定义操作类型。write是写入新记忆；read是按ID或类型读取；query是向向量数据库发起语义查询。
memory_type：对记忆进行分类。例如context（任务背景）、decision（关键决策点）、artifact（生成的文档/代码等产出物）。这有助于结构化管理和检索。
content/query：根据action不同而使用。content用于写入，应包含核心文本和丰富的元数据（metadata），元数据是后续高效检索的关键。query用于语义搜索。
timestamp：确保记忆的时序性，对于理解事件流至关重要。

注意：在设计写入（write）操作时，务必考虑幂等性。即同一智能体因网络重试等原因多次发送相同记忆时，系统应能识别并避免产生重复数据。通常可以为每条记忆生成一个唯一ID（如结合task_id、agent_id、step和内容哈希），在写入前先检查是否存在。

3. 实战搭建：构建一个多智能体协作系统

理论说完了，我们动手搭建一个简易但功能完整的多OpenClaw协作系统。我们将模拟一个“技术博客生成”任务，由三个智能体协作完成：主题研究员（Researcher）、大纲架构师（Architect）和内容写手（Writer）。

3.1 环境准备与智能体角色定义

首先，确保你有可用的OpenClaw（或类似的大语言模型API，如OpenAI GPT、Claude等）调用环境。我们将使用Python作为粘合剂。

# 安装核心依赖 pip install openai chromadb pydantic

我们使用pydantic来严格定义数据模型，这是保证系统鲁棒性的好习惯。

from pydantic import BaseModel, Field from enum import Enum from typing import Optional, List, Dict, Any import uuid from datetime import datetime class MemoryType(str, Enum): CONTEXT = "context" # 背景信息 DECISION = "decision" # 关键决策 ARTIFACT = "artifact" # 产出物 CONSTRAINT = "constraint" # 约束条件 class MemoryFragment(BaseModel): """记忆片段的基本单元""" id: str = Field(default_factory=lambda: str(uuid.uuid4())) task_id: str agent_id: str type: MemoryType content_text: str metadata: Dict[str, Any] = Field(default_factory=dict) created_at: datetime = Field(default_factory=datetime.now) class Config: use_enum_values = True class AgentRole(BaseModel): """智能体角色定义""" id: str # 如 "researcher", "architect" name: str system_prompt: str # 定义该角色职责和行为的系统提示词 input_memory_types: List[MemoryType] # 需要读取哪些类型的记忆 output_memory_type: MemoryType # 产出何种类型的记忆

接下来，定义我们的三个智能体角色：

# 定义三个协作智能体 RESEARCHER = AgentRole( id="researcher", name="主题研究员", system_prompt="你是一个资深技术趋势分析师。你的职责是分析给定的技术主题，提炼核心概念、技术原理、应用场景和最新进展。你的输出必须结构清晰、事实准确。", input_memory_types=[MemoryType.CONTEXT], # 读取任务背景 output_memory_type=MemoryType.DECISION # 产出分析结论（决策类记忆） ) ARCHITECT = AgentRole( id="architect", name="大纲架构师", system_prompt="你是一个优秀的文章架构师。基于研究员提供的技术分析，你需要规划出一篇博客的详细大纲，包括标题、引言、核心章节（至少3个）、子章节、以及每个部分的要点和阐述角度。确保逻辑流畅，层层递进。", input_memory_types=[MemoryType.DECISION], # 读取研究员的分析结论 output_memory_type=MemoryType.DECISION # 产出文章大纲（也属于决策） ) WRITER = AgentRole( id="writer", name="内容写手", system_prompt="你是一个文笔流畅的技术博主。根据架构师提供的大纲，将其扩展成一篇生动、易懂、有深度的技术博客文章。注意使用恰当的案例和通俗的类比，避免干巴巴的陈述。", input_memory_types=[MemoryType.DECISION, MemoryType.CONTEXT], # 读取大纲和背景 output_memory_type=MemoryType.ARTIFACT # 产出最终文章（产出物） )

3.2 实现共享记忆管理器（MemoryManager）

我们将使用Chroma这款轻量级向量数据库来实现记忆的语义存储与检索。

import chromadb from chromadb.config import Settings import hashlib class MemoryManager: def __init__(self, persist_directory="./chroma_db"): # 持久化存储，重启后记忆不丢失 self.client = chromadb.PersistentClient(path=persist_directory) # 为每个任务创建一个独立的集合（Collection），以task_id命名 self.collections = {} def _get_or_create_collection(self, task_id: str): """获取或创建指定任务的记忆集合""" if task_id not in self.collections: # 集合名用task_id的哈希值，避免特殊字符问题 collection_name = hashlib.md5(task_id.encode()).hexdigest()[:16] self.collections[task_id] = self.client.get_or_create_collection( name=collection_name, metadata={"task_id": task_id} ) return self.collections[task_id] def write_memory(self, memory: MemoryFragment): """将记忆片段写入向量数据库""" collection = self._get_or_create_collection(memory.task_id) # 元数据用于精确过滤 metadata = { "agent_id": memory.agent_id, "type": memory.type, "task_id": memory.task_id, **memory.metadata } # 存入向量数据库 collection.add( documents=[memory.content_text], metadatas=[metadata], ids=[memory.id] ) print(f"[MemoryManager] 记忆已写入。ID: {memory.id}, 类型: {memory.type}, 贡献者: {memory.agent_id}") def query_memories(self, task_id: str, query_text: str, memory_type: Optional[MemoryType] = None, n_results: int = 5): """在指定任务记忆中，进行语义查询""" collection = self._get_or_create_collection(task_id) # 构建过滤条件 where_filter = {"task_id": task_id} if memory_type: where_filter["type"] = memory_type.value results = collection.query( query_texts=[query_text], n_results=n_results, where=where_filter # 过滤条件确保只查本任务、指定类型的记忆 ) # 将查询结果封装成MemoryFragment列表返回 memories = [] if results['documents']: for i in range(len(results['documents'][0])): mem = MemoryFragment( id=results['ids'][0][i], task_id=task_id, agent_id=results['metadatas'][0][i]['agent_id'], type=results['metadatas'][0][i]['type'], content_text=results['documents'][0][i], metadata={k:v for k,v in results['metadatas'][0][i].items() if k not in ['agent_id', 'type', 'task_id']} ) memories.append(mem) return memories def get_memories_by_type(self, task_id: str, memory_type: MemoryType): """获取指定任务下特定类型的所有记忆（非语义，按类型过滤）""" # 这是一种简单实现：通过查询一个空字符串或通用词来触发检索，然后靠where过滤。 # 更高效的做法是直接使用collection.get(where=...)，但Chroma的get接口可能不支持复杂的语义过滤。 # 这里我们查询一个通用词“的”，并靠where严格过滤类型。 return self.query_memories(task_id, "的", memory_type, n_results=100)

3.3 实现智能体执行器（AgentExecutor）

这个模块负责封装与OpenClaw（或LLM API）的交互，并整合记忆的读写。

import openai # 示例使用OpenAI API，请替换为你的OpenClaw客户端 class AgentExecutor: def __init__(self, role: AgentRole, memory_manager: MemoryManager, llm_client): self.role = role self.memory_manager = memory_manager self.llm_client = llm_client def execute(self, task_id: str, initial_input: str = None) -> MemoryFragment: """智能体执行其任务""" print(f"\n=== 智能体 [{self.role.name}] 开始执行 ===") # 1. 从共享记忆中读取所需上下文 relevant_memories = [] for mem_type in self.role.input_memory_types: mems = self.memory_manager.get_memories_by_type(task_id, mem_type) relevant_memories.extend(mems) print(f" 读取到 {len(mems)} 条类型为 '{mem_type}' 的记忆。") # 构建给LLM的提示词 prompt = self._build_prompt(initial_input, relevant_memories) # 2. 调用LLM执行任务 print(f" 正在调用LLM生成内容...") llm_response = self._call_llm(prompt) # 3. 将执行结果封装为新的记忆片段 new_memory = MemoryFragment( task_id=task_id, agent_id=self.role.id, type=self.role.output_memory_type, content_text=llm_response, metadata={ "step": len(relevant_memories) + 1, # 一个简单的步骤计数器 "triggered_by": [m.id for m in relevant_memories] # 关联上游记忆ID } ) # 4. 将新记忆写入共享记忆库 self.memory_manager.write_memory(new_memory) print(f" 执行完成，已生成并写入新记忆。ID: {new_memory.id}") return new_memory def _build_prompt(self, initial_input: str, memories: List[MemoryFragment]) -> str: """构建给LLM的完整提示词""" prompt_parts = [] prompt_parts.append(f"# 系统角色\n{self.role.system_prompt}\n") if memories: prompt_parts.append("# 共享记忆（上下文）") for mem in memories: prompt_parts.append(f"## 来自 [{mem.agent_id}] 的记忆 (类型:{mem.type}):\n{mem.content_text}\n") if initial_input: prompt_parts.append(f"# 初始任务输入\n{initial_input}\n") prompt_parts.append("# 你的任务\n请基于以上所有信息，完成你作为【{self.role.name}】的职责。请直接输出工作成果，不要添加额外的解释。") return "\n".join(prompt_parts) def _call_llm(self, prompt: str) -> str: """调用大语言模型API（此处以OpenAI为例）""" # 请替换为你的实际OpenClaw调用方式 try: response = self.llm_client.chat.completions.create( model="gpt-4", # 或你的模型 messages=[{"role": "user", "content": prompt}], temperature=0.7, max_tokens=2000 ) return response.choices[0].message.content.strip() except Exception as e: return f"LLM调用失败: {str(e)}"

3.4 编排工作流：让智能体接力跑起来

最后，我们需要一个“指挥员”（Orchestrator）来按顺序启动各个智能体，并传递必要的初始信息。

class TaskOrchestrator: def __init__(self, task_id: str, memory_manager: MemoryManager, llm_client): self.task_id = task_id self.memory_manager = memory_manager self.llm_client = llm_client # 初始化任务背景记忆 self._init_task_context() def _init_task_context(self): """向共享记忆中写入初始任务背景""" context_memory = MemoryFragment( task_id=self.task_id, agent_id="orchestrator", type=MemoryType.CONTEXT, content_text="任务：撰写一篇关于'大语言模型智能体（LLM Agent）协作系统设计'的技术博客。要求：面向中级开发者，内容涵盖架构设计、通信协议、实战示例和避坑指南。", metadata={"category": "technical_blog", "target_audience": "mid_level_developer"} ) self.memory_manager.write_memory(context_memory) def run(self): """按预定流程执行协作任务""" print(f"\n***** 开始执行协作任务: {self.task_id} *****") # 第一棒：研究员 researcher_executor = AgentExecutor(RESEARCHER, self.memory_manager, self.llm_client) research_result = researcher_executor.execute(self.task_id) print(f"\n研究员完成工作。产出摘要: {research_result.content_text[:100]}...") # 第二棒：架构师 architect_executor = AgentExecutor(ARCHITECT, self.memory_manager, self.llm_client) outline_result = architect_executor.execute(self.task_id) print(f"\n架构师完成工作。产出摘要: {outline_result.content_text[:100]}...") # 第三棒：写手 writer_executor = AgentExecutor(WRITER, self.memory_manager, self.llm_client) final_article = writer_executor.execute(self.task_id) print(f"\n写手完成工作。产出长度: {len(final_article.content_text)} 字符。") # 任务完成，检索并展示最终成果 print(f"\n***** 任务完成！最终成果 *****") artifacts = self.memory_manager.get_memories_by_type(self.task_id, MemoryType.ARTIFACT) for art in artifacts: print(f"\n--- 产出物 (由 {art.agent_id} 生成) ---\n") print(art.content_text) print("-"*50) # 主程序入口 if __name__ == "__main__": TASK_ID = "blog_llm_agent_collab_001" llm_client = openai.OpenAI(api_key="your-api-key-here") # 替换为你的客户端 memory_manager = MemoryManager() orchestrator = TaskOrchestrator(TASK_ID, memory_manager, llm_client) orchestrator.run()

运行这段代码，你将看到三个智能体依次被激活。研究员会先读取任务背景，生成一份技术分析报告并存入记忆；架构师随后读取这份报告，规划出博客大纲；最后写手综合背景和大纲，生成完整的博客文章。所有中间产物和最终成果都通过MemoryManager持久化在Chroma向量数据库中。

4. 进阶优化与生产环境考量

上面的示例是一个最小可行系统。要将其用于生产环境或更复杂的场景，还需要考虑以下几个关键点。

4.1 记忆的版本管理与冲突解决

当多个智能体可能并发写入，或任务需要回溯到某个历史状态时，版本管理就变得至关重要。

为记忆添加版本号：每次更新记忆时，不覆盖旧记录，而是创建一条新版本，并通过parent_id字段与旧版本关联。这形成了一个记忆的版本树。
引入乐观锁：在写入记忆时，检查该记忆当前的最新版本号是否与读取时一致。如果不一致，说明在计算过程中记忆已被其他智能体更新，此时需要让当前智能体基于最新记忆重新执行。这可以通过在MemoryFragment中添加一个version字段（每次更新自增）来实现。
定义冲突解决策略：对于关键决策类记忆，可以设计投票机制。当多个智能体对同一问题产生不同结论时，可以引入一个“仲裁者”智能体来评估各方论据，或让所有协作智能体进行投票，最终将共识结果写入记忆。

4.2 动态工作流与条件路由

我们的示例是简单的线性流水线（研究员→架构师→写手）。现实任务往往更复杂，需要根据中间结果动态决定下一步由哪个智能体执行。

工作流引擎集成：可以考虑使用像Prefect或Airflow这样的工作流编排工具。每个智能体作为一个任务节点，节点的触发条件不仅依赖于上游完成，还可以依赖于共享记忆中特定内容的出现或满足某个条件（例如，只有当研究员的分析“置信度”高于阈值时，才触发架构师）。
基于记忆内容的路由：在Orchestrator中，增加一个“决策层”。在每个智能体执行完毕后，决策层会查询最新的共享记忆，运行一些规则或甚至调用另一个“调度员”LLM，来分析当前任务状态并决定下一个最佳行动者是谁。这使得系统能够处理分支、循环等复杂流程。

4.3 记忆的压缩、摘要与遗忘机制

长期运行的任务会产生海量记忆，导致检索效率下降和成本增加。

定期摘要：可以设置一个“摘要员”智能体，定期（例如每完成10个步骤）对近期产生的所有记忆进行阅读、总结，生成一段高度凝练的“摘要记忆”。后续的智能体可以先读取摘要记忆来了解全局，必要时再深入查询细节记忆。这类似于人类的工作记忆与长期记忆。
重要性评分与遗忘：为每条记忆附加一个由系统或智能体打出的“重要性分数”。分数可能基于记忆被检索的频率、关联的智能体权威性、或明确的人工标注。系统可以定期清理分数低于某个阈值的记忆，或者将其转移到更廉价的冷存储中，实现“记忆遗忘”，保持核心记忆库的简洁有效。

4.4 监控、调试与可观测性

当多个智能体通过黑盒般的共享记忆协作时，问题排查会变得困难。

全链路追踪：为每个任务生成唯一的trace_id，并贯穿所有智能体的调用和记忆读写。像OpenTelemetry这样的标准可以帮我们记录每个步骤的耗时、输入输出和状态。
记忆可视化看板：开发一个简单的Web界面，能够以时间线或图谱的形式可视化一个task_id下的所有记忆片段，展示它们是如何被创建、关联和消费的。这对于理解智能体的协作逻辑和调试异常行为至关重要。
设置检查点与回滚：定期将整个任务的关键状态（包括所有智能体的内部状态快照和共享记忆的完整备份）保存为检查点。如果后续流程出现不可恢复的错误，可以快速回滚到上一个稳定状态，而不是从头开始。

通过引入共享记忆机制，我们将多个OpenClaw智能体从独立的执行单元，转变为一个有机的、具备集体智慧的协作系统。这个系统的能力上限，不再受限于单个模型的能力，而取决于我们如何巧妙地设计角色、编排流程和管理集体记忆。从简单的线性流水线到复杂的动态决策网络，这套范式为我们构建下一代AI应用提供了坚实的基础框架。在实际项目中，你可以从本文的简易示例出发，根据具体业务需求，逐步引入上述进阶考量，打造出真正强大、可靠的多智能体协作引擎。