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_factorylambda: str(uuid.uuid4())) task_id: str agent_id: str type: MemoryType content_text: str metadata: Dict[str, Any] Field(default_factorydict) created_at: datetime Field(default_factorydatetime.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( idresearcher, name主题研究员, system_prompt你是一个资深技术趋势分析师。你的职责是分析给定的技术主题提炼核心概念、技术原理、应用场景和最新进展。你的输出必须结构清晰、事实准确。, input_memory_types[MemoryType.CONTEXT], # 读取任务背景 output_memory_typeMemoryType.DECISION # 产出分析结论决策类记忆 ) ARCHITECT AgentRole( idarchitect, name大纲架构师, system_prompt你是一个优秀的文章架构师。基于研究员提供的技术分析你需要规划出一篇博客的详细大纲包括标题、引言、核心章节至少3个、子章节、以及每个部分的要点和阐述角度。确保逻辑流畅层层递进。, input_memory_types[MemoryType.DECISION], # 读取研究员的分析结论 output_memory_typeMemoryType.DECISION # 产出文章大纲也属于决策 ) WRITER AgentRole( idwriter, name内容写手, system_prompt你是一个文笔流畅的技术博主。根据架构师提供的大纲将其扩展成一篇生动、易懂、有深度的技术博客文章。注意使用恰当的案例和通俗的类比避免干巴巴的陈述。, input_memory_types[MemoryType.DECISION, MemoryType.CONTEXT], # 读取大纲和背景 output_memory_typeMemoryType.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(pathpersist_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( namecollection_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_resultsn_results, wherewhere_filter # 过滤条件确保只查本任务、指定类型的记忆 ) # 将查询结果封装成MemoryFragment列表返回 memories [] if results[documents]: for i in range(len(results[documents][0])): mem MemoryFragment( idresults[ids][0][i], task_idtask_id, agent_idresults[metadatas][0][i][agent_id], typeresults[metadatas][0][i][type], content_textresults[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_results100)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_idtask_id, agent_idself.role.id, typeself.role.output_memory_type, content_textllm_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( modelgpt-4, # 或你的模型 messages[{role: user, content: prompt}], temperature0.7, max_tokens2000 ) return response.choices[0].message.content.strip() except Exception as e: return fLLM调用失败: {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_idself.task_id, agent_idorchestrator, typeMemoryType.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_keyyour-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应用提供了坚实的基础框架。在实际项目中你可以从本文的简易示例出发根据具体业务需求逐步引入上述进阶考量打造出真正强大、可靠的多智能体协作引擎。