1. 项目概述为什么“记忆”是AI Agent的命门最近和几个做AI应用的朋友聊天发现一个挺有意思的现象大家一窝蜂地给Agent加各种工具、搞复杂的工作流但聊到“记忆”这个最基础的能力时往往就含糊其辞了。要么是简单粗暴地把所有对话历史都塞进上下文窗口要么就是用一个简陋的向量数据库存了事结果Agent用着用着就“失忆”了——不记得用户十分钟前刚说过自己的偏好或者把上周定好的项目规则忘得一干二净。这恰恰点中了当前AI Agent开发的一个核心痛点。我们搭建的Agent本质上是一个需要与环境用户进行多轮、长期交互的智能体。它不像传统的单次问答模型问完即走。一个真正有用的Agent必须能记住关键信息并在后续的交互中灵活、准确地调用这些信息形成连贯的“人格”和“服务体验”。这就是“Memory Skills”记忆技能和“上下文工程”要解决的根本问题如何让AI拥有像人一样的工作记忆和长期记忆并高效地管理这些记忆以支持复杂的任务执行。从零开始搭建一个具备优秀记忆能力的AI Agent远不止是调用一个API那么简单。它涉及到记忆的类型划分、存储架构的设计、读取与更新的策略以及最关键的——如何将这些记忆有效地编织到每一次与模型的对话即上下文中。这个过程我称之为“上下文工程”它决定了Agent的“智商”上限。一个只会机械记录、不会智能提取和应用的Agent就像一个记了满脑子笔记却考试时找不到重点的学生空有知识无法发挥。所以这个项目我们将深入Agent的“大脑”内部抛开那些花哨的外围工具聚焦于构建一套扎实、高效、可扩展的记忆与上下文管理系统。无论你是想做一个贴心的个人助理还是一个专业的行业顾问这套核心技能都是让你的Agent从“玩具”蜕变为“工具”的关键。2. 记忆系统核心架构设计分而治之的存储哲学设计记忆系统的第一步是摒弃“一个向量库装所有”的懒惰想法。人的记忆尚且分为短期记忆和长期记忆AI Agent的记忆更应如此。我们需要根据信息的特性、访问频率和重要性设计分层、分类的存储架构。2.1 记忆的三大核心类型与用途在我的实践中通常将Agent的记忆划分为三种核心类型它们各司其职共同构成完整的记忆体系。短期记忆/对话记忆这是最直接、最活跃的记忆层。它完整记录当前会话轮次中的所有消息历史用户输入、Agent思考、工具调用结果等。它的核心作用是维持对话的连贯性让模型理解“刚才我们说到哪了”。实现上它通常就是一个在内存中维护的列表或队列。这里的关键设计点是长度管理。无限制地增长会导致上下文窗口爆炸因此需要设定一个最大轮次限制例如最近20轮对话并实现一个“先进先出”的滚动窗口机制。长期记忆/实体记忆这部分用于存储需要跨会话持久化、并且需要被精确回忆的关键事实信息。例如用户的姓名、职业偏好、项目截止日期、API密钥等。它的特点是高精度、结构化、需快速检索。因此传统的关系型数据库如SQLite、PostgreSQL或文档数据库如MongoDB是更好的选择我们可以通过用户ID、实体类型等字段进行精确查询。例如一张user_preferences表字段包括user_id,key,value,updated_at。语义记忆/知识记忆这是最体现“智能”的一层。它存储的是那些非结构化的、需要通过含义来联想和回忆的信息。比如用户曾说过“我喜欢简洁现代的设计风格”或者“上次讨论的营销方案核心是聚焦年轻群体”。这类信息无法用简单的keyvalue匹配但需要在相关话题出现时被主动唤起。这正是向量数据库如Chroma, Pinecone, Weaviate的用武之地。我们将这些文本片段转换成向量嵌入存储起来。当新问题到来时计算其与存储向量的相似度召回最相关的几条记忆。2.2 记忆的读写策略何时记如何取定义了存储位置接下来是更关键的策略记忆的读写机制。记忆的写入记什么何时记我们不能让Agent事无巨细地记下每一句话那会产生大量噪音。记忆的写入应该是一个有选择的、主动的过程。通常有两种触发方式显式指令用户明确要求“记住这一点”例如“请记住我的邮箱是xxxxx.com”。隐式提取通过一个独立的“记忆提炼”环节来实现。在每轮对话结束后或者检测到对话涉及关键信息变更时让一个大语言模型可以是主模型也可以是一个轻量级模型对对话历史进行分析提取出需要长期或语义存储的事实、观点或承诺并将其格式化后存入相应的记忆库。例如提炼出“用户偏好设计风格 - 简洁现代”存入语义记忆。记忆的读取用什么取多少这是上下文工程的核心。在Agent准备回复用户之前我们需要根据当前查询从庞大的记忆库中筛选出最相关的片段注入到本次对话的上下文提示词中。策略是混合检索精确检索针对长期记忆直接用当前用户ID和可能的关键词如“邮箱”、“截止日期”去数据库查询。语义检索针对语义记忆将用户的当前问题或经过重写的问题进行向量化从向量数据库中召回相似度最高的前K条例如3-5条记忆。对话上下文直接从短期记忆的滚动窗口中取出最近N轮对话。一个常见的误区是“召回即注入”。直接把所有召回的记忆片段堆砌在提示词里会严重干扰模型的判断。我们必须对召回的记忆进行相关性过滤和重要性排序。可以设计一个简单的评分规则近期对话中的信息权重更高用户明确要求记住的信息权重最高通过语义检索到的信息其相似度分数可以作为权重参考。最终只选择权重最高的、最相关的几条记忆放入上下文。3. 上下文工程的实战构建提示词流水线有了记忆的原材料如何将它们烹饪成一份模型能消化并给出优质回答的“提示词大餐”就是上下文工程的任务。这绝不仅仅是字符串拼接而是一个精密的流水线。3.1 系统提示词的角色定义与记忆锚点系统提示词是Agent的“宪法”它定义了Agent的身份、核心能力和行为准则。在涉及记忆的系统中系统提示词必须明确包含关于记忆的指令。例如你是一个专业的个人助理拥有记忆能力。 你可以记住关于我的个人信息、偏好和我们对话中达成共识的事项。 在回答时请主动、恰当地运用你所知道的关于我的信息使服务更具个性化。 如果你对某些信息不确定或记忆模糊可以向我确认。更重要的是我们需要在系统提示词中为“记忆读写”预留接口。一种高效的模式是采用“事实清单”或“已知信息”模块。在拼接最终提示词时我们会将本轮检索到的相关记忆以清晰、结构化的格式如 bullet points插入到这个模块中。这样模型在生成回答时就能明确地“看到”并利用这些背景信息。3.2 动态上下文组装与长度优化每次请求模型时我们组装的上下文通常由以下几部分动态构成系统指令固定的角色定义和行为准则。相关记忆动态检索并格式化后的长期/语义记忆。对话历史从短期记忆中提取的最近若干轮对话。工具定义与历史如果Agent具备工具调用能力相关工具的描述和最近的工具调用结果。当前用户问题。这里的核心挑战是上下文长度限制。主流模型都有token上限如128K、200K我们必须确保组装后的上下文不超过这个限制。因此需要一个上下文窗口管理器。它的职责是实时计算已组装内容的token数。采用优先级压缩策略当前用户问题、系统指令和最关键的相关记忆必须保留。对话历史则从最旧的开始逐轮剔除直到总长度低于安全阈值通常预留一部分token给模型生成回答。对于超长的单个记忆片段可以尝试用另一个LLM进行摘要概括后再注入。3.3 记忆的更新、合并与遗忘机制记忆不是一成不变的。一个成熟的记忆系统必须能处理信息的更新、冲突和清理。更新与合并当从对话中提炼出关于同一事实的新信息时例如用户先说“我喜欢蓝色”后来说“其实我更喜欢深蓝色”系统需要能判断这是“补充”还是“修正”。简单的规则是对于长期记忆中的结构化数据直接根据唯一键如[user_id, key]进行更新。对于语义记忆一种策略是删除旧的相似向量存入新的更复杂的策略是尝试将新旧信息合并成一条更完整的表述后再存入。遗忘机制记忆系统也需要“垃圾回收”。我们可以为每条记忆附加元数据如created_at创建时间、last_accessed_at最后访问时间、access_count访问次数。可以定期如每天运行一个清理任务将那些长期未被访问、且重要性标签低的记忆进行归档或删除。例如删除超过90天未被访问的语义记忆。这能有效控制存储成本并提升检索效率。4. 从零到一的代码实现拆解理论说再多不如一行代码。下面我们以Python为例勾勒一个简化但完整的内存系统核心实现。我们将使用SQLite作为长期记忆库Chroma作为向量数据库并利用LangChain的框架来组织流程。4.1 基础架构与记忆类定义首先定义核心的记忆数据结构。from datetime import datetime from typing import Optional, Dict, Any, List from pydantic import BaseModel, Field import json class MemoryItem(BaseModel): 单条记忆项的基础模型 id: Optional[str] None user_id: str # 记忆所属的用户/会话标识 content: str # 记忆的文本内容 memory_type: str # 类型short_term, long_term_fact, semantic metadata: Dict[str, Any] Field(default_factorydict) # 元数据如来源、重要性、标签 created_at: datetime Field(default_factorydatetime.now) last_accessed_at: datetime Field(default_factorydatetime.now) access_count: int 0 def to_text(self) - str: 将记忆项格式化为可注入上下文的文本 # 根据类型不同格式化方式可调整 return f- {self.content} (记忆时间: {self.created_at.date()})4.2 长期记忆存储SQLite实现我们实现一个基于SQLite的长期记忆管理器用于存储key-value式的确切事实。import sqlite3 from contextlib import contextmanager class LongTermMemoryStore: def __init__(self, db_path: str agent_memory.db): self.db_path db_path self._init_db() def _init_db(self): with self._get_connection() as conn: conn.execute( CREATE TABLE IF NOT EXISTS long_term_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(user_id, key) ) ) conn.commit() contextmanager def _get_connection(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row # 返回字典样式的行 try: yield conn finally: conn.close() def store_fact(self, user_id: str, key: str, value: str): 存储或更新一个事实 with self._get_connection() as conn: conn.execute( INSERT INTO long_term_memory (user_id, key, value, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(user_id, key) DO UPDATE SET valueexcluded.value, updated_atCURRENT_TIMESTAMP , (user_id, key, value)) conn.commit() def retrieve_fact(self, user_id: str, key: str) - Optional[str]: 检索一个事实 with self._get_connection() as conn: cur conn.execute( SELECT value FROM long_term_memory WHERE user_id? AND key?, (user_id, key) ) row cur.fetchone() return row[value] if row else None def search_facts_by_prefix(self, user_id: str, key_prefix: str) - List[Dict]: 通过键的前缀检索相关事实例如检索所有‘preference.’开头的设置 with self._get_connection() as conn: cur conn.execute( SELECT key, value FROM long_term_memory WHERE user_id? AND key LIKE ? , (user_id, f{key_prefix}%)) return [dict(row) for row in cur.fetchall()]4.3 语义记忆存储Chroma实现接下来实现基于向量的语义记忆存储。这里我们使用Chroma的本地持久化模式。import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import numpy as np class SemanticMemoryStore: def __init__(self, persist_dir: str ./chroma_db, embedding_model_name: str all-MiniLM-L6-v2): self.client chromadb.PersistentClient(pathpersist_dir, settingsSettings(anonymized_telemetryFalse)) # 使用SentenceTransformer本地模型生成嵌入避免API调用延迟和成本 self.embedder SentenceTransformer(embedding_model_name) self.collection self.client.get_or_create_collection(namesemantic_memories) def _generate_embedding(self, text: str) - List[float]: 为文本生成向量嵌入 return self.embedder.encode(text).tolist() def store_memory(self, user_id: str, content: str, metadata: Optional[Dict] None): 存储一条语义记忆 embedding self._generate_embedding(content) # Chroma要求ID为字符串我们组合用户ID和时间戳生成唯一ID memory_id f{user_id}_{datetime.now().timestamp()} doc_metadata {user_id: user_id, content: content, **(metadata or {})} self.collection.add( embeddings[embedding], documents[content], # 同时存储原文便于召回后直接使用 metadatas[doc_metadata], ids[memory_id] ) def retrieve_similar_memories(self, user_id: str, query: str, n_results: int 3) - List[Dict]: 根据查询文本检索对应用户最相似的N条语义记忆 query_embedding self._generate_embedding(query) # 在检索时通过metadata过滤特定用户的记忆 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, where{user_id: user_id} # 关键按用户ID过滤 ) memories [] if results[documents]: for doc, meta, dist in zip(results[documents][0], results[metadatas][0], results[distances][0]): memories.append({ content: doc, metadata: meta, similarity_score: 1 - dist # 假设使用余弦相似度距离越小越相似 }) return memories4.4 记忆管理器与上下文组装器现在我们将各个部分组合起来创建一个顶层的记忆管理器并实现上下文组装逻辑。class MemoryManager: def __init__(self, user_id: str): self.user_id user_id self.short_term_memory: List[Dict] [] # 存储原始消息字典 self.long_term_store LongTermMemoryStore() self.semantic_store SemanticMemoryStore() self.max_short_term_turns 10 # 短期记忆保留最近10轮对话 def add_to_short_term(self, role: str, content: str): 添加一轮对话到短期记忆 self.short_term_memory.append({role: role, content: content}) # 滚动窗口保持不超过最大轮次 if len(self.short_term_memory) self.max_short_term_turns * 2: # 每轮包含user和assistant两条 self.short_term_memory self.short_term_memory[-(self.max_short_term_turns * 2):] def extract_and_store_memories(self, conversation_turn: List[Dict]): 记忆提炼分析最近一轮或几轮对话提取需要长期/语义存储的信息。 这是一个简化示例实际应用中可能需要调用LLM进行智能提取。 # 示例简单的规则提取实际应用应更复杂 last_user_message None for msg in reversed(conversation_turn): if msg[role] user: last_user_message msg[content] break if last_user_message: # 规则1如果用户说“记住我的X是Y”则存入长期记忆 if 记住我的 in last_user_message and 是 in last_user_message: # 非常简单的解析实际应用需用更稳健的方法如正则或LLM try: parts last_user_message.split(记住我的)[1].split(是) key parts[0].strip() value parts[1].strip( 。.) self.long_term_store.store_fact(self.user_id, key, value) print(f[记忆] 已存储长期事实: {key} - {value}) except: pass # 规则2将用户陈述的明显偏好或事实存入语义记忆 # 这里可以定义更复杂的触发词或使用一个小的分类模型 preference_indicators [我喜欢, 我讨厌, 我觉得, 我认为, 对我来说] if any(indicator in last_user_message for indicator in preference_indicators): self.semantic_store.store_memory(self.user_id, last_user_message, metadata{type: preference}) print(f[记忆] 已存储语义记忆: {last_user_message[:50]}...) def retrieve_relevant_context(self, current_query: str) - str: 检索所有相关记忆并格式化为上下文字符串 context_parts [] # 1. 检索长期事实例如查询中可能包含的关键词 # 这里可以做一个简单的关键词匹配实际应用中可以用NER提取实体 potential_keys [邮箱, 名字, 偏好, 截止日期] # 示例关键词列表 for key in potential_keys: if key in current_query: value self.long_term_store.retrieve_fact(self.user_id, key) if value: context_parts.append(f已知事实我的{key}是{value}。) # 2. 检索语义记忆 semantic_memories self.semantic_store.retrieve_similar_memories(self.user_id, current_query, n_results2) if semantic_memories: context_parts.append(相关背景信息) for mem in semantic_memories: if mem[similarity_score] 0.7: # 设置一个相似度阈值 context_parts.append(f- {mem[content]}) # 3. 获取短期对话历史 short_term_context for msg in self.short_term_memory[-6:]: # 取最近3轮对话6条消息 short_term_context f{msg[role]}: {msg[content]}\n # 组装最终上下文 memory_context \n.join(context_parts) if context_parts else 暂无相关长期记忆 full_context f## 系统指令 你是一个有帮助的、能记住用户信息的助手。 ## 已知关于用户的信息 {memory_context} ## 最近的对话历史 {short_term_context} ## 当前用户问题 用户{current_query} 助手 return full_context4.5 与主循环集成最后我们将记忆管理器集成到Agent的主循环中。# 假设我们使用OpenAI API from openai import OpenAI client OpenAI(api_keyyour-api-key) def run_agent_with_memory(): user_id user_001 memory_manager MemoryManager(user_id) print(Agent已启动。输入‘退出’来结束对话。) while True: user_input input(\n你) if user_input.lower() in [退出, exit, quit]: break # 1. 将用户输入加入短期记忆 memory_manager.add_to_short_term(user, user_input) # 2. 记忆提炼分析上一轮交互提取需要持久化的记忆 # 这里我们简单地将上一轮完整对话传入实际可能更精细 if len(memory_manager.short_term_memory) 2: last_turn memory_manager.short_term_memory[-2:] # 上一轮的用户和助手消息 memory_manager.extract_and_store_memories(last_turn) # 3. 为当前问题检索相关上下文 context_for_this_turn memory_manager.retrieve_relevant_context(user_input) # 4. 调用LLM生成回复 try: response client.chat.completions.create( modelgpt-4, messages[ {role: system, content: 你是一个有帮助的、能记住用户信息的助手。}, {role: user, content: context_for_this_turn} ], temperature0.7, max_tokens500 ) assistant_reply response.choices[0].message.content except Exception as e: assistant_reply f抱歉我遇到了一些问题{e} print(f助手{assistant_reply}) # 5. 将助手回复加入短期记忆 memory_manager.add_to_short_term(assistant, assistant_reply) if __name__ __main__: run_agent_with_memory()这个实现是一个高度简化的原型但它清晰地展示了记忆系统的核心工作流程记录 - 提炼 - 存储 - 检索 - 注入上下文 - 生成。你可以在此基础上扩展更智能的记忆提炼模型、更复杂的检索排序算法以及更健壮的错误处理。5. 避坑指南与进阶优化策略在实际开发和部署中你会遇到许多在Demo中不会出现的问题。以下是我从多个项目中总结出的关键避坑点和优化方向。5.1 常见陷阱与解决方案陷阱一记忆污染与幻觉现象Agent错误地记住了不存在的信息或者将不同用户、不同场景的信息混淆在一起。根因记忆检索时没有做好严格的隔离如用户ID过滤或者记忆提炼环节错误地将模型的推断当成了事实存储。解决方案严格的数据隔离在所有记忆存储和检索的接口中强制传入并校验user_id、session_id等隔离标识。事实性校验在记忆提炼环节对于关键事实如日期、数字、具体承诺可以设计一个“确认”步骤。例如让LLM以“用户是否明确说了以下内容”的格式输出待存储的记忆只有置信度极高的才存入长期记忆库。对于存疑信息可先存入一个“待确认”缓冲区。记忆来源追溯为每条记忆附加source元数据记录它来自哪一轮对话。当记忆之间发生冲突时可以依据来源的新旧、可信度进行裁决。陷阱二上下文窗口爆炸与信息过载现象随着对话进行检索到的记忆越来越多加上长对话历史导致提示词过长模型响应变慢、成本激增甚至因超限而报错。根因无节制地向上下文添加信息缺乏压缩和优先级管理。解决方案摘要式记忆对于较长的对话回合或文档内容不要存储原始文本而是存储由模型生成的摘要。例如每10轮对话后让模型生成一段“本轮对话核心要点”存入语义记忆。动态上下文窗口实现一个智能的上下文组装器。它不仅做简单的截断而是根据当前查询的意图主动选择最相关的记忆和对话历史。可以训练一个小型分类器或使用LLM本身来判断哪些历史片段与当前问题最相关。分层注入并非所有记忆都需要在每一轮都注入。可以将记忆分为“核心档案”如用户名和“情境知识”。核心档案每次对话都带情境知识则仅在检测到相关话题时才动态检索注入。陷阱三检索效率低下与准确性不足现象语义检索总是召回不相关的记忆或者漏掉关键记忆导致Agent表现“健忘”或“答非所问”。根因嵌入模型不匹配、检索策略单一、或记忆存储的“粒度”不合适。解决方案嵌入模型选型通用模型如text-embedding-ada-002在大多数情况下表现良好但对于特定领域如法律、医疗使用在该领域语料上微调过的嵌入模型能大幅提升效果。混合检索不要只依赖向量检索。结合关键词检索如BM25和元数据过滤如时间范围、记忆类型。例如先通过“用户ID标签”过滤出一个子集再在这个子集里做向量相似度计算。优化记忆“块”的大小存储的文本片段chunk大小直接影响检索质量。太大会包含无关信息稀释核心语义太小则缺乏上下文。一个实用的策略是使用“递归分割”法先按较大段落分割如果某段落的向量与查询相似度很高再将其进一步细分成句子进行更精细的检索。5.2 性能与成本优化实战1. 缓存策略频繁访问且不常变动的记忆如用户基本信息可以在服务内存中设置一个带TTL生存时间的缓存避免每次请求都查询数据库。from functools import lru_cache import time class CachedMemoryManager(MemoryManager): def __init__(self, user_id: str, ttl_seconds300): super().__init__(user_id) self._fact_cache {} self._cache_ttl ttl_seconds lru_cache(maxsize128) def get_fact_with_cache(self, key: str) - Optional[str]: 带缓存的长期事实获取 cache_key f{self.user_id}:{key} if cache_key in self._fact_cache: value, timestamp self._fact_cache[cache_key] if time.time() - timestamp self._cache_ttl: return value # 缓存未命中或已过期查询数据库 value self.long_term_store.retrieve_fact(self.user_id, key) if value: self._fact_cache[cache_key] (value, time.time()) return value2. 异步操作记忆的存储尤其是向向量数据库写入和复杂的记忆提炼过程可以设计为异步操作不要阻塞主对话流程。用户发出消息后系统立即开始检索和生成回复同时在后端异步执行本轮对话的记忆分析存储任务。3. 成本控制记忆提炼和摘要生成都需要调用LLM是主要成本来源。可以通过以下方式控制降低提炼频率并非每轮对话都进行提炼可以设定一个时间或轮次间隔如每5轮提炼一次。使用小模型对于记忆摘要、信息提取这类任务使用gpt-3.5-turbo或更小的开源模型如Llama 3.1 8B通常足以胜任成本远低于GPT-4。设置预算上限为每个用户/会话设置每日或每月的记忆处理token预算超出后则暂停自动记忆提炼功能。5.3 评估记忆系统的有效性如何判断你的记忆系统是否真的在“加分”需要设计一些评估方法人工评测设计一系列多轮对话的测试用例其中后一轮的问题需要依赖前一轮的信息才能正确回答。让人工评估员判断Agent的回答是否准确、自然地运用了记忆。自动化指标记忆召回率在测试集中统计Agent应该使用记忆的场景中它成功检索并使用了正确记忆的比例。记忆准确率统计Agent所使用的记忆中信息准确无误的比例防止幻觉。上下文效率统计平均每次请求注入的token数在保证效果的前提下这个数字越低说明你的检索和压缩策略越高效。搭建一个强大的AI Agent记忆系统是一个在准确性、效率、成本和用户体验之间不断寻找平衡点的工程。它没有银弹但通过理解核心原理、采用分层架构、实施精细策略并持续迭代优化你完全能够打造出一个真正“记得住事”、“懂你所需”的智能伙伴。这不仅仅是技术的堆砌更是对交互本质的深入思考。 AI Agent记忆系统架构与上下文工程实战：从短期记忆到长期语义存储
1. 项目概述为什么“记忆”是AI Agent的命门最近和几个做AI应用的朋友聊天发现一个挺有意思的现象大家一窝蜂地给Agent加各种工具、搞复杂的工作流但聊到“记忆”这个最基础的能力时往往就含糊其辞了。要么是简单粗暴地把所有对话历史都塞进上下文窗口要么就是用一个简陋的向量数据库存了事结果Agent用着用着就“失忆”了——不记得用户十分钟前刚说过自己的偏好或者把上周定好的项目规则忘得一干二净。这恰恰点中了当前AI Agent开发的一个核心痛点。我们搭建的Agent本质上是一个需要与环境用户进行多轮、长期交互的智能体。它不像传统的单次问答模型问完即走。一个真正有用的Agent必须能记住关键信息并在后续的交互中灵活、准确地调用这些信息形成连贯的“人格”和“服务体验”。这就是“Memory Skills”记忆技能和“上下文工程”要解决的根本问题如何让AI拥有像人一样的工作记忆和长期记忆并高效地管理这些记忆以支持复杂的任务执行。从零开始搭建一个具备优秀记忆能力的AI Agent远不止是调用一个API那么简单。它涉及到记忆的类型划分、存储架构的设计、读取与更新的策略以及最关键的——如何将这些记忆有效地编织到每一次与模型的对话即上下文中。这个过程我称之为“上下文工程”它决定了Agent的“智商”上限。一个只会机械记录、不会智能提取和应用的Agent就像一个记了满脑子笔记却考试时找不到重点的学生空有知识无法发挥。所以这个项目我们将深入Agent的“大脑”内部抛开那些花哨的外围工具聚焦于构建一套扎实、高效、可扩展的记忆与上下文管理系统。无论你是想做一个贴心的个人助理还是一个专业的行业顾问这套核心技能都是让你的Agent从“玩具”蜕变为“工具”的关键。2. 记忆系统核心架构设计分而治之的存储哲学设计记忆系统的第一步是摒弃“一个向量库装所有”的懒惰想法。人的记忆尚且分为短期记忆和长期记忆AI Agent的记忆更应如此。我们需要根据信息的特性、访问频率和重要性设计分层、分类的存储架构。2.1 记忆的三大核心类型与用途在我的实践中通常将Agent的记忆划分为三种核心类型它们各司其职共同构成完整的记忆体系。短期记忆/对话记忆这是最直接、最活跃的记忆层。它完整记录当前会话轮次中的所有消息历史用户输入、Agent思考、工具调用结果等。它的核心作用是维持对话的连贯性让模型理解“刚才我们说到哪了”。实现上它通常就是一个在内存中维护的列表或队列。这里的关键设计点是长度管理。无限制地增长会导致上下文窗口爆炸因此需要设定一个最大轮次限制例如最近20轮对话并实现一个“先进先出”的滚动窗口机制。长期记忆/实体记忆这部分用于存储需要跨会话持久化、并且需要被精确回忆的关键事实信息。例如用户的姓名、职业偏好、项目截止日期、API密钥等。它的特点是高精度、结构化、需快速检索。因此传统的关系型数据库如SQLite、PostgreSQL或文档数据库如MongoDB是更好的选择我们可以通过用户ID、实体类型等字段进行精确查询。例如一张user_preferences表字段包括user_id,key,value,updated_at。语义记忆/知识记忆这是最体现“智能”的一层。它存储的是那些非结构化的、需要通过含义来联想和回忆的信息。比如用户曾说过“我喜欢简洁现代的设计风格”或者“上次讨论的营销方案核心是聚焦年轻群体”。这类信息无法用简单的keyvalue匹配但需要在相关话题出现时被主动唤起。这正是向量数据库如Chroma, Pinecone, Weaviate的用武之地。我们将这些文本片段转换成向量嵌入存储起来。当新问题到来时计算其与存储向量的相似度召回最相关的几条记忆。2.2 记忆的读写策略何时记如何取定义了存储位置接下来是更关键的策略记忆的读写机制。记忆的写入记什么何时记我们不能让Agent事无巨细地记下每一句话那会产生大量噪音。记忆的写入应该是一个有选择的、主动的过程。通常有两种触发方式显式指令用户明确要求“记住这一点”例如“请记住我的邮箱是xxxxx.com”。隐式提取通过一个独立的“记忆提炼”环节来实现。在每轮对话结束后或者检测到对话涉及关键信息变更时让一个大语言模型可以是主模型也可以是一个轻量级模型对对话历史进行分析提取出需要长期或语义存储的事实、观点或承诺并将其格式化后存入相应的记忆库。例如提炼出“用户偏好设计风格 - 简洁现代”存入语义记忆。记忆的读取用什么取多少这是上下文工程的核心。在Agent准备回复用户之前我们需要根据当前查询从庞大的记忆库中筛选出最相关的片段注入到本次对话的上下文提示词中。策略是混合检索精确检索针对长期记忆直接用当前用户ID和可能的关键词如“邮箱”、“截止日期”去数据库查询。语义检索针对语义记忆将用户的当前问题或经过重写的问题进行向量化从向量数据库中召回相似度最高的前K条例如3-5条记忆。对话上下文直接从短期记忆的滚动窗口中取出最近N轮对话。一个常见的误区是“召回即注入”。直接把所有召回的记忆片段堆砌在提示词里会严重干扰模型的判断。我们必须对召回的记忆进行相关性过滤和重要性排序。可以设计一个简单的评分规则近期对话中的信息权重更高用户明确要求记住的信息权重最高通过语义检索到的信息其相似度分数可以作为权重参考。最终只选择权重最高的、最相关的几条记忆放入上下文。3. 上下文工程的实战构建提示词流水线有了记忆的原材料如何将它们烹饪成一份模型能消化并给出优质回答的“提示词大餐”就是上下文工程的任务。这绝不仅仅是字符串拼接而是一个精密的流水线。3.1 系统提示词的角色定义与记忆锚点系统提示词是Agent的“宪法”它定义了Agent的身份、核心能力和行为准则。在涉及记忆的系统中系统提示词必须明确包含关于记忆的指令。例如你是一个专业的个人助理拥有记忆能力。 你可以记住关于我的个人信息、偏好和我们对话中达成共识的事项。 在回答时请主动、恰当地运用你所知道的关于我的信息使服务更具个性化。 如果你对某些信息不确定或记忆模糊可以向我确认。更重要的是我们需要在系统提示词中为“记忆读写”预留接口。一种高效的模式是采用“事实清单”或“已知信息”模块。在拼接最终提示词时我们会将本轮检索到的相关记忆以清晰、结构化的格式如 bullet points插入到这个模块中。这样模型在生成回答时就能明确地“看到”并利用这些背景信息。3.2 动态上下文组装与长度优化每次请求模型时我们组装的上下文通常由以下几部分动态构成系统指令固定的角色定义和行为准则。相关记忆动态检索并格式化后的长期/语义记忆。对话历史从短期记忆中提取的最近若干轮对话。工具定义与历史如果Agent具备工具调用能力相关工具的描述和最近的工具调用结果。当前用户问题。这里的核心挑战是上下文长度限制。主流模型都有token上限如128K、200K我们必须确保组装后的上下文不超过这个限制。因此需要一个上下文窗口管理器。它的职责是实时计算已组装内容的token数。采用优先级压缩策略当前用户问题、系统指令和最关键的相关记忆必须保留。对话历史则从最旧的开始逐轮剔除直到总长度低于安全阈值通常预留一部分token给模型生成回答。对于超长的单个记忆片段可以尝试用另一个LLM进行摘要概括后再注入。3.3 记忆的更新、合并与遗忘机制记忆不是一成不变的。一个成熟的记忆系统必须能处理信息的更新、冲突和清理。更新与合并当从对话中提炼出关于同一事实的新信息时例如用户先说“我喜欢蓝色”后来说“其实我更喜欢深蓝色”系统需要能判断这是“补充”还是“修正”。简单的规则是对于长期记忆中的结构化数据直接根据唯一键如[user_id, key]进行更新。对于语义记忆一种策略是删除旧的相似向量存入新的更复杂的策略是尝试将新旧信息合并成一条更完整的表述后再存入。遗忘机制记忆系统也需要“垃圾回收”。我们可以为每条记忆附加元数据如created_at创建时间、last_accessed_at最后访问时间、access_count访问次数。可以定期如每天运行一个清理任务将那些长期未被访问、且重要性标签低的记忆进行归档或删除。例如删除超过90天未被访问的语义记忆。这能有效控制存储成本并提升检索效率。4. 从零到一的代码实现拆解理论说再多不如一行代码。下面我们以Python为例勾勒一个简化但完整的内存系统核心实现。我们将使用SQLite作为长期记忆库Chroma作为向量数据库并利用LangChain的框架来组织流程。4.1 基础架构与记忆类定义首先定义核心的记忆数据结构。from datetime import datetime from typing import Optional, Dict, Any, List from pydantic import BaseModel, Field import json class MemoryItem(BaseModel): 单条记忆项的基础模型 id: Optional[str] None user_id: str # 记忆所属的用户/会话标识 content: str # 记忆的文本内容 memory_type: str # 类型short_term, long_term_fact, semantic metadata: Dict[str, Any] Field(default_factorydict) # 元数据如来源、重要性、标签 created_at: datetime Field(default_factorydatetime.now) last_accessed_at: datetime Field(default_factorydatetime.now) access_count: int 0 def to_text(self) - str: 将记忆项格式化为可注入上下文的文本 # 根据类型不同格式化方式可调整 return f- {self.content} (记忆时间: {self.created_at.date()})4.2 长期记忆存储SQLite实现我们实现一个基于SQLite的长期记忆管理器用于存储key-value式的确切事实。import sqlite3 from contextlib import contextmanager class LongTermMemoryStore: def __init__(self, db_path: str agent_memory.db): self.db_path db_path self._init_db() def _init_db(self): with self._get_connection() as conn: conn.execute( CREATE TABLE IF NOT EXISTS long_term_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(user_id, key) ) ) conn.commit() contextmanager def _get_connection(self): conn sqlite3.connect(self.db_path) conn.row_factory sqlite3.Row # 返回字典样式的行 try: yield conn finally: conn.close() def store_fact(self, user_id: str, key: str, value: str): 存储或更新一个事实 with self._get_connection() as conn: conn.execute( INSERT INTO long_term_memory (user_id, key, value, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(user_id, key) DO UPDATE SET valueexcluded.value, updated_atCURRENT_TIMESTAMP , (user_id, key, value)) conn.commit() def retrieve_fact(self, user_id: str, key: str) - Optional[str]: 检索一个事实 with self._get_connection() as conn: cur conn.execute( SELECT value FROM long_term_memory WHERE user_id? AND key?, (user_id, key) ) row cur.fetchone() return row[value] if row else None def search_facts_by_prefix(self, user_id: str, key_prefix: str) - List[Dict]: 通过键的前缀检索相关事实例如检索所有‘preference.’开头的设置 with self._get_connection() as conn: cur conn.execute( SELECT key, value FROM long_term_memory WHERE user_id? AND key LIKE ? , (user_id, f{key_prefix}%)) return [dict(row) for row in cur.fetchall()]4.3 语义记忆存储Chroma实现接下来实现基于向量的语义记忆存储。这里我们使用Chroma的本地持久化模式。import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import numpy as np class SemanticMemoryStore: def __init__(self, persist_dir: str ./chroma_db, embedding_model_name: str all-MiniLM-L6-v2): self.client chromadb.PersistentClient(pathpersist_dir, settingsSettings(anonymized_telemetryFalse)) # 使用SentenceTransformer本地模型生成嵌入避免API调用延迟和成本 self.embedder SentenceTransformer(embedding_model_name) self.collection self.client.get_or_create_collection(namesemantic_memories) def _generate_embedding(self, text: str) - List[float]: 为文本生成向量嵌入 return self.embedder.encode(text).tolist() def store_memory(self, user_id: str, content: str, metadata: Optional[Dict] None): 存储一条语义记忆 embedding self._generate_embedding(content) # Chroma要求ID为字符串我们组合用户ID和时间戳生成唯一ID memory_id f{user_id}_{datetime.now().timestamp()} doc_metadata {user_id: user_id, content: content, **(metadata or {})} self.collection.add( embeddings[embedding], documents[content], # 同时存储原文便于召回后直接使用 metadatas[doc_metadata], ids[memory_id] ) def retrieve_similar_memories(self, user_id: str, query: str, n_results: int 3) - List[Dict]: 根据查询文本检索对应用户最相似的N条语义记忆 query_embedding self._generate_embedding(query) # 在检索时通过metadata过滤特定用户的记忆 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, where{user_id: user_id} # 关键按用户ID过滤 ) memories [] if results[documents]: for doc, meta, dist in zip(results[documents][0], results[metadatas][0], results[distances][0]): memories.append({ content: doc, metadata: meta, similarity_score: 1 - dist # 假设使用余弦相似度距离越小越相似 }) return memories4.4 记忆管理器与上下文组装器现在我们将各个部分组合起来创建一个顶层的记忆管理器并实现上下文组装逻辑。class MemoryManager: def __init__(self, user_id: str): self.user_id user_id self.short_term_memory: List[Dict] [] # 存储原始消息字典 self.long_term_store LongTermMemoryStore() self.semantic_store SemanticMemoryStore() self.max_short_term_turns 10 # 短期记忆保留最近10轮对话 def add_to_short_term(self, role: str, content: str): 添加一轮对话到短期记忆 self.short_term_memory.append({role: role, content: content}) # 滚动窗口保持不超过最大轮次 if len(self.short_term_memory) self.max_short_term_turns * 2: # 每轮包含user和assistant两条 self.short_term_memory self.short_term_memory[-(self.max_short_term_turns * 2):] def extract_and_store_memories(self, conversation_turn: List[Dict]): 记忆提炼分析最近一轮或几轮对话提取需要长期/语义存储的信息。 这是一个简化示例实际应用中可能需要调用LLM进行智能提取。 # 示例简单的规则提取实际应用应更复杂 last_user_message None for msg in reversed(conversation_turn): if msg[role] user: last_user_message msg[content] break if last_user_message: # 规则1如果用户说“记住我的X是Y”则存入长期记忆 if 记住我的 in last_user_message and 是 in last_user_message: # 非常简单的解析实际应用需用更稳健的方法如正则或LLM try: parts last_user_message.split(记住我的)[1].split(是) key parts[0].strip() value parts[1].strip( 。.) self.long_term_store.store_fact(self.user_id, key, value) print(f[记忆] 已存储长期事实: {key} - {value}) except: pass # 规则2将用户陈述的明显偏好或事实存入语义记忆 # 这里可以定义更复杂的触发词或使用一个小的分类模型 preference_indicators [我喜欢, 我讨厌, 我觉得, 我认为, 对我来说] if any(indicator in last_user_message for indicator in preference_indicators): self.semantic_store.store_memory(self.user_id, last_user_message, metadata{type: preference}) print(f[记忆] 已存储语义记忆: {last_user_message[:50]}...) def retrieve_relevant_context(self, current_query: str) - str: 检索所有相关记忆并格式化为上下文字符串 context_parts [] # 1. 检索长期事实例如查询中可能包含的关键词 # 这里可以做一个简单的关键词匹配实际应用中可以用NER提取实体 potential_keys [邮箱, 名字, 偏好, 截止日期] # 示例关键词列表 for key in potential_keys: if key in current_query: value self.long_term_store.retrieve_fact(self.user_id, key) if value: context_parts.append(f已知事实我的{key}是{value}。) # 2. 检索语义记忆 semantic_memories self.semantic_store.retrieve_similar_memories(self.user_id, current_query, n_results2) if semantic_memories: context_parts.append(相关背景信息) for mem in semantic_memories: if mem[similarity_score] 0.7: # 设置一个相似度阈值 context_parts.append(f- {mem[content]}) # 3. 获取短期对话历史 short_term_context for msg in self.short_term_memory[-6:]: # 取最近3轮对话6条消息 short_term_context f{msg[role]}: {msg[content]}\n # 组装最终上下文 memory_context \n.join(context_parts) if context_parts else 暂无相关长期记忆 full_context f## 系统指令 你是一个有帮助的、能记住用户信息的助手。 ## 已知关于用户的信息 {memory_context} ## 最近的对话历史 {short_term_context} ## 当前用户问题 用户{current_query} 助手 return full_context4.5 与主循环集成最后我们将记忆管理器集成到Agent的主循环中。# 假设我们使用OpenAI API from openai import OpenAI client OpenAI(api_keyyour-api-key) def run_agent_with_memory(): user_id user_001 memory_manager MemoryManager(user_id) print(Agent已启动。输入‘退出’来结束对话。) while True: user_input input(\n你) if user_input.lower() in [退出, exit, quit]: break # 1. 将用户输入加入短期记忆 memory_manager.add_to_short_term(user, user_input) # 2. 记忆提炼分析上一轮交互提取需要持久化的记忆 # 这里我们简单地将上一轮完整对话传入实际可能更精细 if len(memory_manager.short_term_memory) 2: last_turn memory_manager.short_term_memory[-2:] # 上一轮的用户和助手消息 memory_manager.extract_and_store_memories(last_turn) # 3. 为当前问题检索相关上下文 context_for_this_turn memory_manager.retrieve_relevant_context(user_input) # 4. 调用LLM生成回复 try: response client.chat.completions.create( modelgpt-4, messages[ {role: system, content: 你是一个有帮助的、能记住用户信息的助手。}, {role: user, content: context_for_this_turn} ], temperature0.7, max_tokens500 ) assistant_reply response.choices[0].message.content except Exception as e: assistant_reply f抱歉我遇到了一些问题{e} print(f助手{assistant_reply}) # 5. 将助手回复加入短期记忆 memory_manager.add_to_short_term(assistant, assistant_reply) if __name__ __main__: run_agent_with_memory()这个实现是一个高度简化的原型但它清晰地展示了记忆系统的核心工作流程记录 - 提炼 - 存储 - 检索 - 注入上下文 - 生成。你可以在此基础上扩展更智能的记忆提炼模型、更复杂的检索排序算法以及更健壮的错误处理。5. 避坑指南与进阶优化策略在实际开发和部署中你会遇到许多在Demo中不会出现的问题。以下是我从多个项目中总结出的关键避坑点和优化方向。5.1 常见陷阱与解决方案陷阱一记忆污染与幻觉现象Agent错误地记住了不存在的信息或者将不同用户、不同场景的信息混淆在一起。根因记忆检索时没有做好严格的隔离如用户ID过滤或者记忆提炼环节错误地将模型的推断当成了事实存储。解决方案严格的数据隔离在所有记忆存储和检索的接口中强制传入并校验user_id、session_id等隔离标识。事实性校验在记忆提炼环节对于关键事实如日期、数字、具体承诺可以设计一个“确认”步骤。例如让LLM以“用户是否明确说了以下内容”的格式输出待存储的记忆只有置信度极高的才存入长期记忆库。对于存疑信息可先存入一个“待确认”缓冲区。记忆来源追溯为每条记忆附加source元数据记录它来自哪一轮对话。当记忆之间发生冲突时可以依据来源的新旧、可信度进行裁决。陷阱二上下文窗口爆炸与信息过载现象随着对话进行检索到的记忆越来越多加上长对话历史导致提示词过长模型响应变慢、成本激增甚至因超限而报错。根因无节制地向上下文添加信息缺乏压缩和优先级管理。解决方案摘要式记忆对于较长的对话回合或文档内容不要存储原始文本而是存储由模型生成的摘要。例如每10轮对话后让模型生成一段“本轮对话核心要点”存入语义记忆。动态上下文窗口实现一个智能的上下文组装器。它不仅做简单的截断而是根据当前查询的意图主动选择最相关的记忆和对话历史。可以训练一个小型分类器或使用LLM本身来判断哪些历史片段与当前问题最相关。分层注入并非所有记忆都需要在每一轮都注入。可以将记忆分为“核心档案”如用户名和“情境知识”。核心档案每次对话都带情境知识则仅在检测到相关话题时才动态检索注入。陷阱三检索效率低下与准确性不足现象语义检索总是召回不相关的记忆或者漏掉关键记忆导致Agent表现“健忘”或“答非所问”。根因嵌入模型不匹配、检索策略单一、或记忆存储的“粒度”不合适。解决方案嵌入模型选型通用模型如text-embedding-ada-002在大多数情况下表现良好但对于特定领域如法律、医疗使用在该领域语料上微调过的嵌入模型能大幅提升效果。混合检索不要只依赖向量检索。结合关键词检索如BM25和元数据过滤如时间范围、记忆类型。例如先通过“用户ID标签”过滤出一个子集再在这个子集里做向量相似度计算。优化记忆“块”的大小存储的文本片段chunk大小直接影响检索质量。太大会包含无关信息稀释核心语义太小则缺乏上下文。一个实用的策略是使用“递归分割”法先按较大段落分割如果某段落的向量与查询相似度很高再将其进一步细分成句子进行更精细的检索。5.2 性能与成本优化实战1. 缓存策略频繁访问且不常变动的记忆如用户基本信息可以在服务内存中设置一个带TTL生存时间的缓存避免每次请求都查询数据库。from functools import lru_cache import time class CachedMemoryManager(MemoryManager): def __init__(self, user_id: str, ttl_seconds300): super().__init__(user_id) self._fact_cache {} self._cache_ttl ttl_seconds lru_cache(maxsize128) def get_fact_with_cache(self, key: str) - Optional[str]: 带缓存的长期事实获取 cache_key f{self.user_id}:{key} if cache_key in self._fact_cache: value, timestamp self._fact_cache[cache_key] if time.time() - timestamp self._cache_ttl: return value # 缓存未命中或已过期查询数据库 value self.long_term_store.retrieve_fact(self.user_id, key) if value: self._fact_cache[cache_key] (value, time.time()) return value2. 异步操作记忆的存储尤其是向向量数据库写入和复杂的记忆提炼过程可以设计为异步操作不要阻塞主对话流程。用户发出消息后系统立即开始检索和生成回复同时在后端异步执行本轮对话的记忆分析存储任务。3. 成本控制记忆提炼和摘要生成都需要调用LLM是主要成本来源。可以通过以下方式控制降低提炼频率并非每轮对话都进行提炼可以设定一个时间或轮次间隔如每5轮提炼一次。使用小模型对于记忆摘要、信息提取这类任务使用gpt-3.5-turbo或更小的开源模型如Llama 3.1 8B通常足以胜任成本远低于GPT-4。设置预算上限为每个用户/会话设置每日或每月的记忆处理token预算超出后则暂停自动记忆提炼功能。5.3 评估记忆系统的有效性如何判断你的记忆系统是否真的在“加分”需要设计一些评估方法人工评测设计一系列多轮对话的测试用例其中后一轮的问题需要依赖前一轮的信息才能正确回答。让人工评估员判断Agent的回答是否准确、自然地运用了记忆。自动化指标记忆召回率在测试集中统计Agent应该使用记忆的场景中它成功检索并使用了正确记忆的比例。记忆准确率统计Agent所使用的记忆中信息准确无误的比例防止幻觉。上下文效率统计平均每次请求注入的token数在保证效果的前提下这个数字越低说明你的检索和压缩策略越高效。搭建一个强大的AI Agent记忆系统是一个在准确性、效率、成本和用户体验之间不断寻找平衡点的工程。它没有银弹但通过理解核心原理、采用分层架构、实施精细策略并持续迭代优化你完全能够打造出一个真正“记得住事”、“懂你所需”的智能伙伴。这不仅仅是技术的堆砌更是对交互本质的深入思考。