为AI Agent构建记忆操作系统:从向量检索到自我进化的工程实践

📅 2026/8/5 3:50:24
为AI Agent构建记忆操作系统:从向量检索到自我进化的工程实践
1. 项目概述当AI助手拥有“记忆”与“进化”能力最近在捣鼓一个叫WorkBuddy的AI助手框架它本质上是一个帮你处理各种任务的智能体AI Agent。但玩久了总觉得差点意思每次对话都像是“重启”它不记得之前聊过什么更别提从历史交互中学习并自我改进了。这就像和一个只有短期记忆的朋友合作效率总卡在瓶颈上。于是我萌生了一个想法能不能给WorkBuddy装上一个“记忆操作系统”这个想法源自对现有AI Agent局限性的观察。大多数Agent包括WorkBuddy的默认配置其“记忆”是短暂且孤立的。一次会话中的上下文窗口消耗完之前的对话、决策逻辑、乃至成功或失败的经验就都消失了。这不仅导致重复劳动更阻碍了Agent的长期学习和能力沉淀。所谓的“记忆操作系统”MemOS并非指一个具体的、现成的软件而是一种架构理念和功能集合。它的核心目标是赋予AI Agent持续的记忆存储、检索、反思和基于记忆进行自主创造如编写新Skill的能力。简单说就是让WorkBuddy从一个“健忘的执行者”变成一个“有经验的、能自我进化的伙伴”。这个项目的价值在于它直接回应了当前AI Agent领域的一个核心痛点如何实现持续学习和能力累积。无论是个人用来管理日常任务、学习新技能还是团队用于构建复杂的自动化工作流一个拥有记忆和进化能力的Agent其效率和智能水平将是指数级提升。它开始能记住你的偏好、总结工作模式、甚至在遇到类似问题时主动调用或组合已有的技能Skill来生成更优的解决方案最终实现“自己写Skill自己进化”。这听起来有点科幻但基于现有的技术栈是完全可实现的工程实践。接下来我就把自己从构思到实现这套“记忆操作系统”的全过程包括核心设计、技术选型、踩过的坑和最终效果毫无保留地分享出来。2. 记忆操作系统MemOS的核心架构设计给WorkBuddy添加记忆能力绝不是简单加个数据库存聊天记录那么简单。它需要一套完整的、分层的架构来处理信息的生命周期从感知、存储、索引、检索到应用与生成。我设计的MemOS架构主要包含四个核心层次每一层都有其明确的职责和技术考量。2.1 记忆的层次化存储模型记忆不能是“一锅粥”必须结构化。我借鉴了认知科学和现有AI系统的设计将记忆分为三个层级瞬时记忆Working Memory对应LLM的上下文窗口。它处理当前对话回合的输入、思考链Chain-of-Thought和即时输出。这部分是“思考的草稿纸”速度快但容量有限。我的设计是在每次Agent运行结束时自动将本轮对话中的关键决策点、执行结果和用户反馈从“瞬时记忆”中提炼出来准备存入长期记忆。短期记忆Short-term Memory存储近期如过去几天的高频、高相关性记忆片段。我使用向量数据库如Chroma或Qdrant来实现。每一段记忆例如“用户要求将‘月度报告.pdf’转换为Markdown格式我成功调用了‘pdf_to_md’这个Skill完成”会被转换成一个文本片段并通过嵌入模型Embedding Model转化为向量。向量数据库支持基于语义相似度的快速检索。当新任务到来时系统会从短期记忆中检索出最相关的几条记忆作为上下文提供给LLM帮助它做出更相关的决策。长期记忆Long-term Memory存储重要的经验、提炼出的知识、以及Agent自己创造的资产如Skill代码。这部分使用传统的关系型数据库如SQLite或PostgreSQL和文件系统。关系库用于存储记忆的元数据ID、类型、创建时间、关联标签等而具体的知识内容或Skill代码则以文本或文件形式存储。长期记忆的关键在于“摘要”和“索引”。系统会定期或触发式对短期记忆中的相关片段进行总结形成更高阶的“经验知识”例如“用户通常在周一上午需要处理文档格式转换”并存入长期记忆。同时所有记忆都通过关键词和语义标签进行索引方便多维度检索。注意记忆的存储不是单向的。一个高效的MemOS需要设计记忆的“激活”机制。当处理任务时系统首先从长期记忆中通过关键词召回一批相关记忆再用这些记忆的摘要去向量库做更精细的语义检索最后将最相关的几条记忆与瞬时记忆结合。这模仿了人类的记忆唤醒过程。2.2 记忆的生成、索引与检索流程记忆不是被动存储的而是主动生成的。我设计了一个“记忆生成器”模块它作为WorkBuddy Action的一个环节在每次任务循环结束后被调用。其工作流程如下事件捕获监听Agent的核心事件如“Skill执行完成”、“用户给出明确反馈好评/差评”、“任务成功/失败”、“出现了需要总结的复杂情况”。内容提炼对于捕获的事件不是原始日志全存。我编写了提炼提示词让一个小型LLM如GPT-3.5-Turbo或本地轻量模型对事件进行总结生成结构化的记忆片段。例如输入原始日志“调用天气查询Skill参数{city: ‘北京’}返回成功气温22度。” 提炼后记忆“成功执行‘天气查询’技能用户询问了北京天气结果正常。关联标签#天气 #查询 #成功”。向量化与存储将提炼后的记忆文本通过嵌入模型我选用text-embedding-3-small平衡效果与成本转化为向量并存入向量数据库短期记忆。同时将记忆文本、元数据类型、时间戳、关联的Skill ID、任务ID和向量ID的映射关系存入关系数据库长期记忆索引。检索策略当新任务到来时“记忆检索器”模块开始工作。首先它解析当前任务描述和上下文生成一组关键词。用这些关键词在关系数据库中进行初步筛选。然后将当前任务描述也向量化用这个向量去向量数据库中执行相似度搜索通常用余弦相似度找出最相关的N条短期记忆。最后将两类结果去重、排序合并成一个增强的上下文注入给负责决策的LLM。这个流程确保了记忆是高质量的、可检索的并且检索过程是高效、精准的。2.3 基于记忆的反思与Skill进化机制记忆的终极价值在于驱动进化。MemOS最核心的功能就是让WorkBuddy能够“反思”记忆并据此创造新Skill或优化旧Skill。我设计了两种进化触发机制定期反思Scheduled Reflection像一个定期的复盘会议。系统设置一个后台任务例如每天凌晨分析过去24小时内存储的所有记忆。它会关注重复模式用户是否频繁提出某一类请求现有的Skill组合起来是否能解决如果不能是否需要新Skill失败案例哪些任务失败了失败原因是什么是Skill能力不足、参数错误还是需要外部信息成功模式哪些任务完成得特别出色其执行路径是否可以抽象成一个更高效的新流程或Skill 反思过程由一个具有较强分析能力的LLM如Claude 3或GPT-4驱动它读取一批相关记忆并输出反思报告报告可能包含“建议创建新Skill”的提案。事件触发反思Event-triggered Reflection当发生特定高价值事件时立即启动。例如用户对某个Skill的输出给出了“大拇指 down”的负面反馈。一个复杂任务经过多轮工具调用才勉强完成。Agent发现自己无法处理一个请求因为缺少对应的Skill。 这时系统会立即围绕该事件检索相关记忆启动一个高优先级的反思会话分析问题根源并生成解决方案如优化现有Skill的提示词或起草一个新Skill的需求描述。从反思到Skill创建当反思报告建议创建新Skill时MemOS不会直接写代码。它会生成一个详细的“Skill需求说明书”PRD包括Skill的功能描述、输入/输出格式、可能用到的API或工具、以及测试用例。然后这个PRD会被送入“Skill编码器”模块。这个模块可以是另一个专精代码生成的LLM如Claude 3.5 Sonnet或DeepSeek-Coder它根据PRD和类似的现有Skill作为参考生成初始的Skill代码可能是Python函数、一段脚本或一个插件配置。生成的代码会经过简单的语法检查并保存到Skill仓库中等待用户或管理员的审核与激活。至此一个基于记忆和反思的“自我进化”循环就完成了。3. 技术栈选型与核心组件集成实现MemOS需要谨慎选择每一个技术组件并让它们在WorkBuddy的生态中无缝协作。我的选型原则是优先考虑成熟度、社区活跃度、与Python生态的兼容性以及轻量级毕竟这是给一个AI Agent增加的“外挂”系统。3.1 记忆存储组件向量数据库与关系数据库向量数据库我选择了ChromaDB。原因很简单它轻量、易嵌入、API简单并且专门为AI应用设计。它可以直接在Python进程中运行无需单独部署服务器这对于集成到WorkBuddy中非常友好。将记忆片段向量化后用chromadb客户端直接存入集合Collection即可。检索时一句collection.query就能搞定。它的持久化模式也足够可靠可以将数据保存在本地目录。备选方案如果对性能和大规模生产环境有更高要求Qdrant或Weaviate是更强大的选择但它们需要单独部署服务增加了系统复杂度。对于个人或小团队使用的WorkBuddyChroma是起步的最佳选择。关系数据库我选择了SQLite。WorkBuddy本身可能就使用SQLite来管理配置和基础数据延续使用可以避免引入新的依赖。它零配置、单文件、性能对于记忆元数据的管理绰绰有余。我设计了几张核心表memory_metadata: 存储记忆ID、类型、摘要、时间戳、关联的skill_id、task_id等。skill_definitions: 存储Skill的元数据名称、描述、作者、版本、代码路径。reflection_logs: 存储每次反思会话的输入、输出和结论。 SQLite的简洁性让整个MemOS的部署变得极其简单。3.2 嵌入模型与LLM的选型考量嵌入模型Embedding Model这是记忆检索准确性的基石。我主要使用OpenAI的text-embedding-3-small。它在效果和成本之间取得了完美平衡对于记忆片段这种短文本其语义捕捉能力足够强且API调用速度快、价格低廉。如果追求完全本地化可以选用BAAI/bge-small-zh-v1.5或thenlper/gte-small这类开源模型通过sentence-transformers库集成。但需要权衡本地推理的计算资源消耗。实操心得对于记忆检索嵌入模型的速度和稳定性比极致的精度更重要。text-embedding-3-small的维度是1536在Chroma中检索效率很高。如果使用开源模型务必测试其编码速度避免拖慢Agent的整体响应。核心LLM用于决策与反思WorkBuddy的主脑LLM选择很多。对于日常任务决策可以使用性价比较高的模型如GPT-3.5-Turbo或Claude 3 Haiku。但对于“反思”和“Skill生成”这类需要深度推理和创造性的任务我强烈建议使用能力更强的模型如GPT-4、Claude 3.5 Sonnet或DeepSeek-V2。这部分投入是值得的因为高质量的反思和代码生成能极大提升进化的有效性减少后续人工修正的工作量。注意事项需要为不同的任务类型配置不同的LLM。在代码中这意味着要有清晰的“路由”逻辑任务决策走经济型模型API反思生成走高性能模型API。这能有效控制成本。3.3 与WorkBuddy框架的融合方式WorkBuddy通常有一个主循环处理用户输入、调用工具Skill、返回结果。集成MemOS需要在几个关键节点插入钩子HooksPost-Action Hook动作后钩子在每个Skill执行完成后无论成功与否自动触发“记忆生成器”将本次执行的关键信息提炼存储。Pre-Decision Hook决策前钩子在主LLM进行任务规划或Skill选择之前先调用“记忆检索器”获取与当前任务相关的历史记忆并将其作为系统提示词的一部分附加进去让LLM做出更明智的决策。Scheduler调度器引入一个轻量级后台调度如使用apscheduler库定时触发“定期反思”任务。Event Bus事件总线建立一个小型的事件系统。当发生用户反馈、任务失败等事件时发布相应事件。“事件触发反思”模块监听这些事件并做出响应。这种“非侵入式”的集成保证了MemOS可以作为WorkBuddy的一个插件或扩展模块来开发而不需要大量修改WorkBuddy的核心代码提高了可维护性。4. 实操搭建从零到一的MemOS实现步骤理论说再多不如动手做一遍。下面是我在本地搭建并集成MemOS到WorkBuddy的具体步骤你可以跟着一步步实现。4.1 基础环境与依赖安装假设你已经有一个基础的WorkBuddy运行环境基于Python。首先为MemOS创建独立的依赖环境。# 在WorkBuddy项目根目录下创建并激活虚拟环境可选但推荐 python -m venv venv_memos source venv_memos/bin/activate # Linux/Mac # venv_memos\Scripts\activate # Windows # 安装核心依赖 pip install chromadb # 向量数据库 pip install openai # 用于嵌入模型和LLM API调用如果使用 pip install sentence-transformers # 如果使用开源嵌入模型 pip install apscheduler # 用于定时任务 pip install sqlalchemy # ORM方便操作SQLite pip install pydantic # 用于数据验证和设置管理接下来在WorkBuddy的项目结构中创建一个新的模块目录例如workbuddy_memos/。4.2 记忆存储模块的代码实现在workbuddy_memos/目录下创建以下几个核心文件1.models.py- 定义数据模型from pydantic import BaseModel from datetime import datetime from typing import Optional, List import sqlalchemy as sa from sqlalchemy.orm import declarative_base Base declarative_base() # Pydantic模型用于API和逻辑处理 class MemoryFragment(BaseModel): id: Optional[str] None content: str # 提炼后的记忆文本 embedding: Optional[List[float]] None memory_type: str # 如 skill_execution, user_feedback, task_result timestamp: datetime skill_id: Optional[str] None task_id: Optional[str] None tags: List[str] [] # SQLAlchemy模型用于数据库持久化 class MemoryMetadata(Base): __tablename__ memory_metadata id sa.Column(sa.String, primary_keyTrue) content sa.Column(sa.Text, nullableFalse) memory_type sa.Column(sa.String, nullableFalse) timestamp sa.Column(sa.DateTime, defaultdatetime.utcnow) skill_id sa.Column(sa.String) task_id sa.Column(sa.String) vector_id sa.Column(sa.String) # 关联Chroma中的ID tags sa.Column(sa.JSON) # 存储标签列表 class SkillDefinition(Base): __tablename__ skill_definitions id sa.Column(sa.String, primary_keyTrue) name sa.Column(sa.String, uniqueTrue, nullableFalse) description sa.Column(sa.Text) code_path sa.Column(sa.String) # Skill代码文件路径 author sa.Column(sa.String, defaultMemOS) created_at sa.Column(sa.DateTime, defaultdatetime.utcnow) is_active sa.Column(sa.Boolean, defaultTrue)2.storage.py- 实现存储与检索类import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import sqlalchemy as sa from sqlalchemy.orm import sessionmaker from .models import Base, MemoryMetadata from typing import List, Dict, Any import uuid class MemoryStorage: def __init__(self, persist_dir: str ./chroma_db, embedding_model_name: str all-MiniLM-L6-v2): # 初始化Chroma客户端持久化模式 self.chroma_client chromadb.PersistentClient(pathpersist_dir) self.collection self.chroma_client.get_or_create_collection(nameworkbuddy_memories) # 初始化嵌入模型这里以开源模型为例 self.embedding_model SentenceTransformer(embedding_model_name) # 初始化SQLite数据库 self.engine sa.create_engine(sqlite:///./workbuddy_memos.db) Base.metadata.create_all(self.engine) self.SessionLocal sessionmaker(bindself.engine) def store_memory(self, memory: MemoryFragment): 存储一段记忆到向量库和关系库 # 生成唯一ID memory_id str(uuid.uuid4()) # 生成向量 embedding self.embedding_model.encode(memory.content).tolist() # 存入Chroma self.collection.add( documents[memory.content], metadatas[{type: memory.memory_type, skill_id: memory.skill_id, task_id: memory.task_id}], embeddings[embedding], ids[memory_id] ) # 存入SQLite with self.SessionLocal() as session: db_memory MemoryMetadata( idmemory_id, contentmemory.content, memory_typememory.memory_type, skill_idmemory.skill_id, task_idmemory.task_id, vector_idmemory_id, tagsmemory.tags ) session.add(db_memory) session.commit() def retrieve_related_memories(self, query: str, n_results: int 5) - List[Dict[str, Any]]: 根据查询文本检索相关记忆 # 将查询文本向量化 query_embedding self.embedding_model.encode(query).tolist() # 从Chroma中查询 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # 组织返回结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memory { content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] } memories.append(memory) return memories4.3 记忆生成与反思引擎的实现3.memory_generator.py- 记忆提炼import openai from .models import MemoryFragment from datetime import datetime class MemoryGenerator: def __init__(self, llm_client): self.llm llm_client # 传入一个配置好的LLM客户端如OpenAI或ChatOLLM def generate_from_event(self, event_type: str, raw_data: dict) - MemoryFragment: 根据原始事件数据生成结构化的记忆片段 prompt f 你是一个记忆提炼助手。请将以下AI Agent执行事件提炼成一段简洁、结构化、未来易于检索的记忆。 事件类型{event_type} 原始数据{raw_data} 请按以下格式输出记忆内容 [动作描述]。结果[成功/失败/部分成功]。关键信息[提取关键参数或结果]。关联标签[列出2-4个相关标签以#开头]。 例如成功执行‘文件转换’技能将‘report.pdf’转换为Markdown格式。结果成功。关键信息输入文件report.pdf输出格式md。关联标签#文件转换 #pdf #markdown #成功。 try: response self.llm.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.2 # 低温度保证输出稳定 ) memory_text response.choices[0].message.content.strip() # 简单解析出标签这里简化处理实际可以更复杂 tags [] if 关联标签 in memory_text: tag_part memory_text.split(关联标签)[-1] tags [tag.strip() for tag in tag_part.split(#) if tag.strip()] return MemoryFragment( contentmemory_text, memory_typeevent_type, timestampdatetime.now(), skill_idraw_data.get(skill_id), task_idraw_data.get(task_id), tagstags ) except Exception as e: print(f记忆生成失败: {e}) # 降级方案生成一个简单的记忆 return MemoryFragment( contentf{event_type}: {str(raw_data)[:200]}, memory_typeevent_type, timestampdatetime.now(), skill_idraw_data.get(skill_id), task_idraw_data.get(task_id), tags[f#{event_type}] )4.reflection_engine.py- 定期反思from apscheduler.schedulers.background import BackgroundScheduler from .storage import MemoryStorage import openai from datetime import datetime, timedelta class ReflectionEngine: def __init__(self, memory_storage: MemoryStorage, powerful_llm_client): self.storage memory_storage self.llm powerful_llm_client # 使用更强的LLM如GPT-4 self.scheduler BackgroundScheduler() def start_periodic_reflection(self, interval_hours24): 启动定时反思任务 self.scheduler.add_job( self._run_reflection, interval, hoursinterval_hours, next_run_timedatetime.now() # 立即运行一次 ) self.scheduler.start() print(定期反思引擎已启动。) def _run_reflection(self): 执行反思分析 print(f{datetime.now()}: 开始执行定期反思...) # 1. 从存储中获取近期记忆例如过去24小时 # 这里需要扩展MemoryStorage添加按时间查询的方法略 # recent_memories self.storage.get_memories_since(datetime.now() - timedelta(hours24)) # 2. 构建反思提示词 reflection_prompt f 你是一个AI Agent的自我进化分析引擎。请分析以下近期的工作记忆并给出进化建议。 近期记忆摘要最近24小时 [这里应拼接近期记忆的摘要内容] 请从以下角度分析 1. **重复模式识别**用户请求中是否存在频繁出现的、未被现有Skill很好满足的需求模式 2. **失败根因分析**哪些任务失败了根本原因是什么例如缺少特定数据、Skill逻辑错误、外部API变化 3. **成功经验固化**哪些任务完成得特别好其执行流程是否可以抽象成新的、可复用的Skill或工作流 4. **Skill缺口发现**是否存在用户明确或隐含请求但当前没有任何Skill能处理的情况 请输出一份结构化的反思报告并至少提出一个具体的“新Skill创建提案”或“现有Skill优化建议”。 提案格式 - 提案类型[新建/优化] - 目标Skill名称 - 功能描述 - 触发场景 - 预期输入/输出 - 参考代码或实现思路如可能 # 3. 调用LLM进行反思 try: response self.llm.chat.completions.create( modelgpt-4, messages[{role: user, content: reflection_prompt}], temperature0.7 ) reflection_report response.choices[0].message.content print(反思报告生成完毕。) # 4. 解析报告提取提案并触发Skill创建流程下一节实现 self._process_reflection_report(reflection_report) except Exception as e: print(f反思过程出错: {e})4.4 Skill自动生成器的实现5.skill_creator.py- 从提案到代码import re import os from pathlib import Path class SkillCreator: def __init__(self, code_llm_client, skill_directory: str ./skills/generated): self.llm code_llm_client # 专精代码生成的LLM如Claude 3.5 Sonnet self.skill_dir Path(skill_directory) self.skill_dir.mkdir(parentsTrue, exist_okTrue) def create_skill_from_proposal(self, proposal: dict): 根据反思报告中的提案生成Skill代码 # proposal 是一个字典包含反思报告解析出的字段 skill_name proposal.get(target_skill_name, new_skill).replace( , _).lower() description proposal.get(function_description, ) trigger proposal.get(trigger_scenario, ) # 构建代码生成提示词 coding_prompt f 你是一个资深的AI Skill开发助手。请根据以下需求编写一个WorkBuddy可用的Skill。 Skill名称{skill_name} 功能描述{description} 触发场景{trigger} 输入/输出要求{proposal.get(input_output, 待定)} WorkBuddy Skill通常是一个Python函数它接收一个字典参数params并返回一个字符串或字典结果。 请遵循以下规范 1. 函数名应为 execute_{skill_name}。 2. 包含详细的文档字符串docstring。 3. 包含必要的错误处理。 4. 如果需要调用外部API请使用requests库并处理好网络异常。 5. 代码应简洁、高效、可读性强。 只输出最终的Python代码无需任何解释。 try: response self.llm.chat.completions.create( modelclaude-3-5-sonnet-20241022, # 示例实际需替换为正确模型名 messages[{role: user, content: coding_prompt}], temperature0.3 ) code response.choices[0].message.content # 清理代码块标记如果LLM返回了python ... code re.sub(rpython\n, , code) code re.sub(r\n, , code) # 保存代码文件 file_path self.skill_dir / f{skill_name}.py with open(file_path, w, encodingutf-8) as f: f.write(code) print(fSkill代码已生成并保存至: {file_path}) # 更新Skill元数据到数据库需要调用storage模块 # self._register_skill_to_db(skill_name, description, str(file_path)) return file_path except Exception as e: print(fSkill代码生成失败: {e}) return None4.5 集成到WorkBuddy主循环最后需要在WorkBuddy的主应用文件中初始化MemOS组件并挂载钩子。6. 在主应用中的集成示例 (app.py或main.py)# 假设这是你的WorkBuddy主文件 from workbuddy_memos.storage import MemoryStorage from workbuddy_memos.memory_generator import MemoryGenerator from workbuddy_memos.reflection_engine import ReflectionEngine from workbuddy_memos.skill_creator import SkillCreator import openai import asyncio # 初始化组件 openai_client openai.OpenAI(api_keyyour-api-key) memory_storage MemoryStorage() memory_gen MemoryGenerator(llm_clientopenai_client) reflection_engine ReflectionEngine(memory_storage, openai_client) skill_creator SkillCreator(code_llm_clientopenai_client) # 启动定期反思例如每12小时一次 reflection_engine.start_periodic_reflection(interval_hours12) # 假设WorkBuddy有一个处理用户请求的核心异步函数 async def handle_user_request(user_input: str, context: dict): WorkBuddy原有的任务处理函数 # 1. 【决策前钩子】检索相关记忆 related_memories memory_storage.retrieve_related_memories(user_input, n_results3) enhanced_prompt build_prompt_with_memories(user_input, related_memories) # 2. WorkBuddy原有逻辑LLM规划选择并执行Skill... # 假设这里调用某个Skill得到结果 skill_result await execute_skill(skill_name, params) # 3. 【动作后钩子】生成并存储记忆 event_data { user_input: user_input, skill_name: skill_name, skill_id: get_skill_id(skill_name), task_id: context.get(task_id), result: skill_result, success: is_success(skill_result) } memory_fragment memory_gen.generate_from_event(skill_execution, event_data) memory_storage.store_memory(memory_fragment) # 4. 【事件触发反思】如果结果失败或用户反馈差立即触发反思 if not event_data[success] or context.get(negative_feedback): reflection_engine.trigger_immediate_reflection(event_data) return skill_result def build_prompt_with_memories(user_input: str, memories: list) - str: 将检索到的记忆构建成提示词的一部分 memory_context \n.join([f- {m[content]} for m in memories[:3]]) # 取前3条 base_system_prompt 你是一个有帮助的AI助手... enhanced_system_prompt f{base_system_prompt} 以下是你过去的相关经验记忆供本次决策参考 {memory_context} 请基于以上经验和当前请求做出最佳回应。 return enhanced_system_prompt至此一个具备记忆存储、检索、反思和Skill生成能力的MemOS就初步集成到WorkBuddy中了。启动你的WorkBuddy它现在不仅能记住过去还能从中学习并创造未来。5. 核心问题排查与优化经验在实际搭建和运行这套系统的过程中我遇到了不少坑也总结出一些优化经验。这里分享几个最关键的问题和解决方案。5.1 记忆检索不准或速度慢问题现象Agent决策时检索到的记忆与当前任务无关或者检索过程明显拖慢了响应速度。排查与解决检查嵌入模型语义搜索不准首先怀疑嵌入模型。对于中文场景text-embedding-3-small对英文优化更好可以尝试专门的中文嵌入模型如BAAI/bge-large-zh-v1.5。用一组已知相关的query-doc对测试不同模型的检索命中率。优化记忆片段质量垃圾进垃圾出。如果存储的是原始冗长的日志检索效果必然差。务必强化“记忆生成器”的提炼能力。我后来改进了提示词要求LLM必须提取“动作主体、对象、结果状态和关键参数”形成标准化更高的记忆文本检索准确率提升了约40%。调整检索策略单纯靠向量相似度可能不够。我引入了混合检索先用关键词从记忆标签和任务描述中提取在关系数据库里做一次筛选缩小范围再用向量相似度在这个子集里精搜。这既提高了相关性也加快了速度。限制检索数量不要一次性检索太多条记忆。对于大多数任务3-5条最相关的记忆已经足够。在retrieve_related_memories函数中n_results参数从默认的10调到了4响应速度有可感知的提升。向量索引优化如果记忆量非常大10万条需要考虑使用支持更高效索引的向量数据库如Qdrant的HNSW索引并确保数据持久化到SSD硬盘。5.2 反思报告空洞或无法生成可行提案问题现象定期反思运行了但生成的报告都是“未发现明显模式”、“一切正常”之类的空话或者提出的Skill提案天马行空无法实现。排查与解决提供更丰富的上下文反思LLM不能只给记忆文本。我修改了_run_reflection方法在提示词中额外提供了现有Skill的清单和功能描述。近期用户请求的原始文本样本而不仅是记忆摘要。历史上成功创建的Skill案例作为参考。 这给了LLM一个更全面的“视野”来做分析。设计更结构化的反思流程将单一的反思提示拆分成多步链式思考Chain-of-Thought。例如第一步总结归纳。要求LLM先将近期记忆分类如文件操作、信息查询、数据整理等。第二步模式识别。针对每一类找出重复的请求、共同的痛点或成功的模式。第三步提案生成。基于第二步的发现结合现有Skill库提出具体、可行的新建或优化提案。 通过分步引导LLM的思考更聚焦输出质量显著提高。设置提案可行性过滤器在skill_creator接收到提案后增加一个“可行性评估”步骤。用一个简单的规则引擎或另一个LLM调用来判断这个提案需要的API或数据是否可获取实现复杂度是否过高如果评估不通过则打回反思引擎重新生成或标记为“需人工审核”。人工反馈循环初期系统生成的Skill代码不可能完美。我建立了一个简单的审核界面所有自动生成的Skill都处于“未激活”状态。我会快速浏览代码进行简单修正或直接批准。系统会记录我的审核动作批准/驳回/修改这些反馈本身又作为新的记忆存入系统用于训练反思引擎的未来判断形成一个持续改进的闭环。5.3 系统资源占用与成本控制问题现象MemOS运行后WorkBuddy内存占用变大响应变慢且API调用费用尤其是GPT-4增长较快。排查与解决记忆存储优化设置记忆过期与降级不是所有记忆都需要永久保存。我实现了策略短期记忆向量库只保留最近30天的30天以上的记忆仅将其摘要文本存入关系数据库长期存档并从向量库中删除以节省空间和检索开销。记忆去重在存储前计算新记忆与已有记忆的向量相似度。如果相似度超过一个阈值如0.95则视为重复记忆只更新原有记忆的时间戳和关联信息而非新增一条。LLM调用优化分级模型使用严格贯彻“轻活用小模型重活用大模型”的原则。记忆提炼、简单的文本处理用GPT-3.5-Turbo核心任务规划用Claude 3 Haiku或同等模型只有深度反思和Skill生成才调用GPT-4或Claude 3.5 Sonnet。反思任务合并将一些低优先级的反思分析合并执行而不是一有事件就触发。例如收集一段时间内的用户负面反馈集中进行一次分析。缓存嵌入结果对于常见的查询关键词或固定的系统提示词部分其嵌入向量是固定的。可以将这些向量的计算结果缓存起来避免重复调用嵌入模型API。异步与非阻塞设计记忆存储、反思分析、Skill生成这些后台任务绝对不能阻塞主Agent的同步响应。务必使用异步asyncio或消息队列如Celery将这些耗时操作放到后台线程或进程中执行。确保用户请求的响应路径是快速的。5.4 生成的Skill代码质量不稳定问题现象SkillCreator生成的代码有时语法错误有时逻辑混乱无法直接运行。排查与解决提供更详细的Skill模板和规范在给代码生成LLM的提示词中提供1-2个写得非常好的现有Skill作为“范例”Few-shot Learning。明确列出代码风格、错误处理、日志记录等要求。引入静态代码分析生成代码后自动调用py_compile或ast模块进行简单的语法检查。如果语法错误则尝试让LLM重新生成或直接标记为失败等待人工处理。分步生成对于复杂Skill不要指望一步到位。我改进了生成流程第一步生成设计稿。先让LLM输出函数签名、输入输出格式、以及主要步骤的伪代码。第二步人工/自动审核设计稿。确认设计合理。第三步根据审核后的设计稿生成完整代码。 虽然多了一步但成功率大幅提高。建立Skill测试套件为生成的Skill自动创建一个简单的单元测试脚本测试其基本功能。如果测试不通过则代码不予激活。这能过滤掉大部分有严重逻辑缺陷的Skill。6. 效果评估与未来演进方向经过几周的运行和迭代这个为WorkBuddy添加的“记忆操作系统”已经展现出明显的价值。最直观的效果是任务处理的上下文连贯性大大增强。当我第二次说“把那个文件像上次一样整理好”时WorkBuddy能准确地回忆起上周我是如何定义“整理好”的特定格式、保存路径等并执行相同的操作。决策质量也提升了尤其是在复杂任务规划时它能借鉴过去的成功或失败经验选择更可靠的Skill组合。进化的萌芽已经出现。系统在运行一周后通过反思我频繁进行的“从网页抓取数据并制表”的操作自动生成了一个名为web_scrape_to_table的新Skill提案。我审核后稍作修改便激活了它现在这个任务从过去需要手动组合3个步骤变成了一个指令直接完成。当然这套系统还处于早期阶段。我规划了几个未来的演进方向记忆的关联与图谱化目前的记忆还是孤立的片段。下一步是建立记忆之间的关联形成知识图谱。例如记忆A学习Python装饰器和记忆B使用装饰器优化Skill应该被关联起来。这样当遇到“如何优化Skill性能”的问题时系统能沿着图谱找到更根本的知识点。个性化与用户建模MemOS目前主要记忆“事”未来可以更主动地记忆“人”。通过分析历史交互逐渐构建用户画像偏好什么沟通风格通常在什么时间处理什么类型的任务对哪些错误容忍度低基于这些画像Agent可以提供更个性化的服务。多模态记忆扩展当前的记忆主要是文本。如果WorkBuddy接入了图像识别、语音处理等能力那么MemOS也需要支持存储和检索图像特征向量、音频片段等实现真正的多模态记忆和联想。安全的进化与伦理边界自我进化是一把双刃剑。必须设置安全围栏。例如任何涉及外部网络访问、文件删除、敏感信息处理的Skill生成都必须经过严格的人工审核。需要建立一个“Skill安全策略”模型自动评估新Skill的潜在风险。给AI Agent装上记忆和进化的能力不再是遥不可及的研究课题而是可以通过工程化落地的实践。这套MemOS的实现就像为WorkBuddy点亮了一盏通往更智能、更自主未来的灯。虽然过程中调试提示词、处理边界情况颇费周折但看到它开始“记住”并“学习”的那一刻感觉所有努力都值了。如果你也在构建或使用AI Agent不妨从为一个核心动作添加记忆开始亲手感受一下智能体“活”起来的过程。