最近在开发AI智能体项目时常常遇到一个棘手问题智能体与用户进行多轮对话后总是“忘记”之前的聊天内容导致每次回答都像初次见面体验非常割裂。为了解决智能体的“记忆”难题业界一直在探索各种方案。恰逢此时向量数据库领域的明星项目Chroma发布了其全新的Foundation 智能体记忆方案这无疑为构建具备长期、稳定记忆能力的AI应用提供了新的强大工具。本文将深入解析 Chroma Foundation 记忆方案的核心原理并提供一个从零开始的完整实战教程。无论你是正在探索智能体开发的初学者还是寻求为现有项目集成高级记忆功能的中高级开发者都能通过本文掌握如何利用 Chroma 为你的 AI 应用构建一个可靠、高效的“记忆大脑”。我们将从环境搭建、核心概念、代码实战到生产级最佳实践一步步拆解确保你能亲手实现并理解其背后的设计思想。1. 背景与核心概念为什么智能体需要“记忆”在深入技术细节之前我们首先要理解问题的本质。传统的基于大语言模型LLM的对话系统其上下文通常受限于模型的“上下文窗口”如 4K、8K、128K tokens。一旦对话轮次或内容长度超出这个窗口模型就无法“看到”更早的历史信息从而导致记忆丢失。智能体记忆Agent Memory就是为了突破这一限制而设计的。它允许智能体将重要的历史交互信息如用户偏好、任务上下文、决策依据持久化存储并在后续的交互中动态地检索和利用这些信息从而实现连贯、个性化且具备长期规划能力的对话体验。Chroma 发布的Foundation方案正是其面向智能体记忆场景推出的一个开箱即用的解决方案。它并非一个独立的软件而是基于 Chroma 向量数据库核心能力构建的一套最佳实践、设计模式与工具链的集合。1.1 Chroma Foundation 的核心价值标准化记忆流程Foundation 定义了一套清晰的记忆处理流程包括信息的提取、编码、存储、检索和更新让开发者无需从零设计架构。与向量搜索深度集成记忆的本质是关联信息的快速召回。Foundation 充分利用 Chroma 的向量化存储与相似性搜索能力将记忆片段转换为向量实现基于语义的智能检索而不仅仅是关键词匹配。支持复杂的记忆结构记忆不是简单的键值对。Foundation 支持对记忆进行分片、打标签、关联元数据如时间戳、重要性分数、关联实体并能处理记忆的衰减与合并更贴近人类记忆的特点。开箱即用的工具提供了易于集成的客户端库、示例代码和预设配置大幅降低了为智能体添加高级记忆功能的门槛。1.2 核心组件解析一个典型的 Foundation 记忆系统包含以下关键组件记忆编码器Memory Encoder负责将一段文本信息如用户的一句话、智能体的一个回复转换为固定维度的向量Embedding。这通常使用一个嵌入模型如text-embedding-3-small来完成。记忆存储Memory Store即 Chroma 向量数据库集合Collection。每个记忆片段及其对应的向量、元数据都存储在这里。记忆检索器Memory Retriever根据当前对话的上下文从记忆存储中查找最相关的历史记忆片段。这通过计算当前上下文向量与存储向量之间的相似度如余弦相似度来实现。记忆管理器Memory Manager这是系统的大脑负责协调编码、存储、检索的整个生命周期并实施记忆策略例如哪些信息需要被记住记忆保存多久如何防止记忆存储爆炸理解了这些概念我们就可以开始动手搭建了。2. 环境准备与版本说明本教程将使用 Python 作为开发语言这是与 Chroma 和大多数 AI 框架集成最广泛的选择。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文示例在 Ubuntu 22.04 上验证。Python版本 3.8 或更高。推荐使用 3.10 以获得最佳兼容性。包管理工具pip(Python 自带) 或poetry(用于更专业的依赖管理)。代码编辑器VS Code, PyCharm 或任何你熟悉的编辑器。2.2 安装核心依赖首先创建一个新的项目目录并初始化虚拟环境这是管理 Python 项目依赖的最佳实践。# 创建项目目录并进入 mkdir chroma-agent-memory cd chroma-agent-memory # 创建并激活虚拟环境 (Linux/macOS) python3 -m venv venv source venv/bin/activate # 创建并激活虚拟环境 (Windows) # python -m venv venv # venv\Scripts\activate接下来安装 Chroma 客户端以及我们将用到的嵌入模型和 OpenAI 客户端用于示例中的LLM调用。# 升级pip pip install --upgrade pip # 安装核心库 pip install chromadb openai tiktoken # 安装 sentence-transformers 作为本地嵌入模型备选可选避免网络调用 pip install sentence-transformers版本说明chromadb: 本文基于0.4.22版本。Chroma 更新较快建议关注其官方文档。openai: 用于调用 OpenAI 的嵌入模型和聊天模型。确保你已设置有效的OPENAI_API_KEY。sentence-transformers: 提供本地运行的嵌入模型如all-MiniLM-L6-v2适合离线或隐私要求高的场景。2.3 项目结构预览在开始编码前我们先规划一个清晰的项目结构chroma-agent-memory/ ├── .env # 存储环境变量如API密钥 ├── requirements.txt # 项目依赖列表 ├── src/ │ ├── __init__.py │ ├── memory_system.py # 核心记忆系统类 │ └── agent.py # 简单的智能体逻辑 └── examples/ └── basic_conversation.py # 使用示例3. 核心原理与架构拆解Foundation 方案倡导的记忆处理流程可以概括为以下四个核心步骤理解它们对后续编码至关重要。3.1 记忆处理的生命周期观察与提取Observe Extract观察智能体监控与用户的每一次交互用户输入、自身输出、工具调用结果。提取并非所有信息都值得记忆。这里需要策略例如提取用户明确陈述的偏好“我喜欢用Markdown格式”。提取任务的关键事实和约束“帮我订下周五北京到上海的机票”。提取决策的逻辑或原因“因为用户是VIP所以推荐了高级套餐”。这一步的输出是待存储的“记忆片段”文本。编码与存储Encode Store编码使用嵌入模型将“记忆片段”文本转换为向量。存储将向量、原始文本以及相关的元数据存入 Chroma 集合。元数据是记忆的“标签”对于高效检索至关重要例如{“type”: “user_preference”, “entity”: “format”, “timestamp”: “2023-10-27T10:00:00”, “importance”: 0.8}。检索与关联Retrieve Associate检索当新的对话发生时将当前对话上下文或一个问题同样编码为向量然后在 Chroma 集合中执行相似性搜索找出最相关的 K 个历史记忆片段。关联将检索到的记忆片段与当前上下文组合形成“增强的上下文”输入给大语言模型LLM。这相当于给了 LLM 一个“记忆提示”。维护与更新Maintain Update维护定期清理过时、无效或低重要性的记忆防止存储无限膨胀。更新当获得新信息时可以更新已有的记忆例如用户说“我其实更喜欢PDF”来更新之前的格式偏好而不是简单新增。这需要设计记忆的唯一标识和更新逻辑。3.2 Chroma 在此流程中的角色Chroma 的核心价值体现在第2步和第3步高效的向量存储与索引Chroma 内置了用于快速近似最近邻搜索的索引如 HNSW使得在海量记忆片段中快速检索成为可能。灵活的元数据过滤除了语义搜索你还可以结合元数据进行过滤。例如“检索所有type为user_preference且entity为format的记忆”。这实现了精确查找与语义查找的结合。集成的嵌入函数Chroma 客户端简化了编码过程可以直接配置嵌入函数存储时自动完成向量化。4. 完整实战构建一个具备记忆的对话智能体现在我们将把理论付诸实践构建一个简单的对话智能体它能记住用户的姓名、喜好和之前的对话主题。4.1 创建核心记忆系统类首先在src/memory_system.py中创建记忆系统的核心类。# src/memory_system.py import chromadb from chromadb.config import Settings from typing import List, Dict, Any, Optional import uuid import time import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class FoundationMemorySystem: 基于 Chroma 的 Foundation 风格记忆系统。 def __init__(self, collection_name: str “agent_memory”, embedding_model: str “local”): 初始化记忆系统。 Args: collection_name: Chroma 集合的名称。 embedding_model: 嵌入模型类型‘local’ 使用 sentence-transformers‘openai’ 使用 OpenAI API。 # 初始化 Chroma 客户端持久化到磁盘 self.client chromadb.PersistentClient(path“./chroma_db”) # 获取或创建集合 self.collection self.client.get_or_create_collection( namecollection_name, metadata{“hnsw:space”: “cosine”} # 使用余弦相似度进行搜索 ) self.embedding_model embedding_model self._init_embedding_function() def _init_embedding_function(self): 初始化嵌入函数。 if self.embedding_model “openai”: # 注意需要设置 OPENAI_API_KEY 环境变量 from chromadb.utils import embedding_functions self.embedding_func embedding_functions.OpenAIEmbeddingFunction( api_keyos.getenv(“OPENAI_API_KEY”), model_name“text-embedding-3-small” ) # 为集合设置嵌入函数 self.collection self.client.get_or_create_collection( nameself.collection.name, embedding_functionself.embedding_func, metadataself.collection.metadata ) else: # 使用本地模型 from sentence_transformers import SentenceTransformer self.model SentenceTransformer(‘all-MiniLM-L6-v2’) # Chroma 期望的嵌入函数格式 def local_embedding_function(texts: List[str]) - List[List[float]]: return self.model.encode(texts).tolist() self.embedding_func local_embedding_function def add_memory(self, content: str, metadata: Optional[Dict[str, Any]] None, memory_id: Optional[str] None): 添加一条记忆。 Args: content: 记忆的文本内容。 metadata: 关联的元数据例如 {“type”: “fact”, “source”: “user”, “importance”: 0.5}。 memory_id: 可选的唯一ID。如果为None则自动生成。 if metadata is None: metadata {} # 确保有基本的时间戳 if “timestamp” not in metadata: metadata[“timestamp”] time.time() mem_id memory_id if memory_id else str(uuid.uuid4()) # 使用集合的嵌入函数自动处理向量化 # 注意如果使用自定义嵌入函数需要手动生成向量并调用 add 方法。 # 这里我们演示更通用的手动方式以兼容本地模型。 embedding self._generate_embedding(content) self.collection.add( embeddings[embedding], documents[content], metadatas[metadata], ids[mem_id] ) logger.info(f“Memory added with ID: {mem_id}”) return mem_id def _generate_embedding(self, text: str) - List[float]: 生成文本的向量嵌入。 if self.embedding_model “openai”: # 对于OpenAIembedding_func 是 Chroma 的函数对象不能直接调用。 # 我们需要模拟其内部调用。更简单的方式是直接使用OpenAI客户端。 # 这里为了简化假设使用本地模型或已通过collection配置。 # 实际使用OpenAI时建议通过collection配置嵌入函数。 pass # 默认使用本地模型 return self.model.encode([text])[0].tolist() def search_memories(self, query: str, filter_conditions: Optional[Dict[str, Any]] None, n_results: int 5) - List[Dict[str, Any]]: 搜索相关记忆。 Args: query: 搜索查询文本。 filter_conditions: 元数据过滤条件例如 {“type”: “user_preference”}。 n_results: 返回的结果数量。 Returns: 包含记忆内容、元数据和相似度得分的字典列表。 # 生成查询的向量 query_embedding self._generate_embedding(query) # 执行搜索 results self.collection.query( query_embeddings[query_embedding], n_resultsn_results, wherefilter_conditions, # 元数据过滤 include[“documents”, “metadatas”, “distances”] ) memories [] if results[‘documents’]: for i in range(len(results[‘documents’][0])): memory { “content”: results[‘documents’][0][i], “metadata”: results[‘metadatas’][0][i], “similarity_score”: 1 - results[‘distances’][0][i] # 将距离转换为相似度余弦 } memories.append(memory) return memories def get_conversation_context(self, recent_query: str, user_id: Optional[str] None) - str: 为当前对话构建增强的上下文。 检索相关记忆并将其格式化为LLM可理解的提示。 Args: recent_query: 用户最近的一次查询。 user_id: 可选的用户ID用于过滤特定用户的记忆。 Returns: 格式化的上下文字符串。 filter_condition None if user_id: filter_condition {“user_id”: user_id} relevant_memories self.search_memories(recent_query, filter_conditionsfilter_condition, n_results3) if not relevant_memories: return “No relevant past memories found.\n” context_lines [“Relevant memories from past interactions:”] for i, mem in enumerate(relevant_memories, 1): # 可以在这里根据重要性分数或类型进行格式化 context_lines.append(f“{i}. {mem[‘content’]} (Relevance: {mem[‘similarity_score’]:.2f})”) return “\n”.join(context_lines) “\n\n” def clear_memories(self): 清空所有记忆谨慎使用。 self.client.delete_collection(self.collection.name) logger.warning(“All memories have been cleared.”)4.2 实现一个简单的对话智能体接下来在src/agent.py中创建一个利用记忆系统的简单智能体。# src/agent.py import os from openai import OpenAI from .memory_system import FoundationMemorySystem from typing import Optional class ConversationalAgent: def __init__(self, memory_system: FoundationMemorySystem, user_id: str “default_user”): 初始化对话智能体。 Args: memory_system: 记忆系统实例。 user_id: 当前对话用户的标识。 self.memory memory_system self.user_id user_id # 初始化 OpenAI 客户端用于生成回复 self.llm_client OpenAI(api_keyos.getenv(“OPENAI_API_KEY”)) self.llm_model “gpt-3.5-turbo” # 可根据需要更换 def _extract_memory_from_conversation(self, user_input: str, ai_response: str): 一个简单的记忆提取策略。 在实际应用中这里可以集成更复杂的NLP模型来识别值得记忆的信息。 memories_to_add [] # 示例规则1如果用户提到自己的名字记住它 if “my name is” in user_input.lower(): # 这里应该用更稳健的方法提取名字例如使用LLM或正则表达式 # 此处为演示简单处理 name_part user_input.lower().split(“my name is”)[-1].strip().split(”.”)[0].split(”,”)[0] if len(name_part) 50: # 简单防错 memory_content f“The users name is {name_part}.” metadata {“type”: “user_fact”, “entity”: “name”, “user_id”: self.user_id, “importance”: 0.9} memories_to_add.append((memory_content, metadata)) # 示例规则2记住用户明确表达的偏好 preference_keywords [“i like”, “i prefer”, “i hate”, “i dont like”] for keyword in preference_keywords: if keyword in user_input.lower(): memory_content f“User stated: ‘{user_input}’” metadata {“type”: “user_preference”, “user_id”: self.user_id, “importance”: 0.7} memories_to_add.append((memory_content, metadata)) break # 找到第一个即可 # 将提取的记忆存入系统 for content, metadata in memories_to_add: self.memory.add_memory(content, metadata) def chat(self, user_input: str) - str: 处理用户输入并生成回复。 Args: user_input: 用户的输入文本。 Returns: 智能体的回复文本。 # 1. 检索相关记忆构建增强上下文 enhanced_context self.memory.get_conversation_context(user_input, self.user_id) # 2. 构建LLM提示 system_prompt “””You are a helpful assistant with a memory. Use the provided past memories to inform your response, making the conversation coherent and personalized. If memories are provided, acknowledge or use them naturally. If no memories are relevant, just answer normally.“”” full_prompt f“{system_prompt}\n\n{enhanced_context}Current user says: {user_input}\n\nAssistant:” # 3. 调用LLM生成回复 try: response self.llm_client.chat.completions.create( modelself.llm_model, messages[ {“role”: “system”, “content”: system_prompt}, {“role”: “user”, “content”: f“{enhanced_context}Current user says: {user_input}”} ], temperature0.7, max_tokens500 ) ai_response response.choices[0].message.content.strip() except Exception as e: ai_response f“I encountered an error: {e}. Please try again.” # 4. 从本轮对话中提取可能的新记忆 self._extract_memory_from_conversation(user_input, ai_response) # 5. 可选将本轮有意义的对话也存储为一般记忆 # self.memory.add_memory( # f“User: {user_input}\nAssistant: {ai_response}”, # {“type”: “conversation_turn”, “user_id”: self.user_id, “importance”: 0.5} # ) return ai_response4.3 运行与验证示例创建一个示例脚本examples/basic_conversation.py来测试我们的智能体。# examples/basic_conversation.py import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.memory_system import FoundationMemorySystem from src.agent import ConversationalAgent def main(): # 0. 环境检查假设OPENAI_API_KEY已设置 if not os.getenv(“OPENAI_API_KEY”): print(“Warning: OPENAI_API_KEY not set. LLM calls will fail. Using local embedding only.”) # 1. 初始化记忆系统和智能体 print(“Initializing memory system and agent...”) memory FoundationMemorySystem(embedding_model“local”) # 使用本地嵌入模型 agent ConversationalAgent(memory_systemmemory, user_id“alice”) # 2. 模拟多轮对话 conversations [ “Hi there!”, “My name is Alice.”, “I really prefer coffee over tea.”, “Whats my name and what drink do I like?” ] for i, user_input in enumerate(conversations): print(f“\n[Turn {i1}] User: {user_input}”) response agent.chat(user_input) print(f“Agent: {response}”) # 3. 演示搜索记忆 print(“\n--- Demonstrating memory search ---”) test_query “What does the user like to drink?” relevant_memories memory.search_memories(test_query, filter_conditions{“user_id”: “alice”}) print(f“Query: ‘{test_query}’”) for mem in relevant_memories: print(f“ - {mem[‘content’]} (Score: {mem[‘similarity_score’]:.2f})”) if __name__ “__main__”: main()运行示例 在项目根目录下执行export OPENAI_API_KEY‘your_api_key_here’ # Linux/macOS # set OPENAI_API_KEYyour_api_key_here # Windows python examples/basic_conversation.py预期输出Initializing memory system and agent... [Turn 1] User: Hi there! Agent: Hello! How can I help you today? [Turn 2] User: My name is Alice. Agent: Nice to meet you, Alice! What can I do for you? [Turn 3] User: I really prefer coffee over tea. Agent: Got it, Alice! Ill remember that youre a coffee fan. Is there anything specific about coffee youd like to discuss? [Turn 4] User: Whats my name and what drink do I like? Agent: Your name is Alice, and youve mentioned that you prefer coffee over tea. --- Demonstrating memory search --- Query: ‘What does the user like to drink?’ - User stated: ‘I really prefer coffee over tea.’ (Score: 0.82) - The users name is Alice. (Score: 0.31)可以看到智能体在第四轮成功回忆起了用户的姓名和饮品偏好这正是记忆系统在起作用。搜索演示也显示系统能根据语义找到相关的记忆片段。5. 常见问题与排查思路在实际集成和使用 Chroma Foundation 记忆方案时你可能会遇到以下典型问题。问题现象常见原因解决思路ModuleNotFoundError: No module named ‘chromadb’依赖未正确安装或虚拟环境未激活。1. 确认已激活虚拟环境。2. 运行pip install chromadb。3. 检查 Python 解释器路径是否正确。嵌入向量维度不匹配创建集合时使用的嵌入函数与后续添加/查询时使用的函数不一致。1. 确保embedding_function在初始化集合时一次性配置好。2. 如果更换模型需要重建集合或进行向量迁移。检索结果不相关1. 嵌入模型不适合领域。2. 元数据过滤条件太严格。3. 搜索参数n_results或相似度阈值设置不当。1. 尝试不同的嵌入模型如text-embedding-3-large。2. 检查元数据格式确保过滤条件正确。3. 调整n_results或对结果进行后过滤按相似度分数。4. 在存储前优化记忆文本的提取质量。内存/磁盘占用增长过快1. 存储了过多低价值记忆。2. 未设置记忆过期或清理策略。3. Chroma 的索引文件累积。1. 实现记忆重要性评分定期清理低分记忆。2. 为记忆添加timestamp实现基于时间的过期策略。3. 定期检查并清理chroma_db目录下的chroma.sqlite3文件如果使用持久化客户端。查询速度变慢记忆数量增长到数十万以上默认索引可能效率下降。1. 在创建集合时调整 HNSW 索引参数如hnsw:construction_ef,hnsw:M。2. 考虑按用户或时间分片使用多个集合。3. 升级 Chroma 版本关注性能优化更新。无法连接到 OpenAI 嵌入服务网络问题、API密钥错误或额度不足。1. 检查OPENAI_API_KEY环境变量。2. 测试网络连通性。3. 考虑降级到本地嵌入模型作为备选方案。6. 最佳实践与工程建议将记忆系统投入生产环境需要考虑更多工程化细节。6.1 记忆提取策略优化基础的规则提取如我们示例中的_extract_memory_from_conversation非常脆弱。生产环境建议使用 LLM 进行提取用一个小型、高效的 LLM如 GPT-3.5-turbo来分析对话轮次结构化地提取事实、偏好、承诺和待办事项。定义记忆模式Schema提前定义好记忆的类型和结构。例如memory_schema { “user_fact”: [“name”, “location”, “job”], “user_preference”: [“topic”, “value”, “strength”], “conversation_goal”: [“goal”, “status”], “action_item”: [“task”, “assignee”, “deadline”] }设置重要性评分在提取时或后续由 LLM 评估该记忆的长期重要性0.0 到 1.0便于后续的存储管理和检索排序。6.2 检索增强生成RAG的集成我们的示例是将记忆作为上下文直接拼接。更高级的做法是将其作为RAGRetrieval-Augmented Generation流程的一部分多路召回结合语义检索向量搜索和关键词检索基于元数据过滤取并集或重排序。重排序Re-ranking使用一个更精细的交叉编码器模型对召回的记忆进行二次排序提升相关性。上下文压缩与总结如果召回的记忆太多可以用 LLM 先对其进行总结或压缩再放入上下文以节省 Token 并提升信息密度。6.3 生产环境部署考量Chroma 服务化在开发中使用PersistentClient很方便但在生产环境中建议部署Chroma 服务端让多个智能体实例通过客户端连接实现记忆共享和集中管理。记忆版本化与更新重要的记忆如用户地址可能会变更。设计机制来更新记忆而非简单新增例如为同一实体如user_identity_type的记忆维护最新版本。安全性记忆可能包含敏感信息PII。务必对存储的文本和元数据进行加密。实施严格的访问控制确保智能体只能访问其授权用户的记忆。提供用户查询、更正和删除其个人记忆的接口以符合数据隐私法规。监控与评估记录记忆的检索命中率、用户对“记忆力”的反馈显式或隐式。定期评估记忆系统的有效性并迭代改进提取和检索策略。6.4 与现有智能体框架集成Foundation 方案是理念和模式的集合可以轻松集成到 LangChain、LlamaIndex、Dify、Coze 等主流智能体框架中。LangChain可以将FoundationMemorySystem包装成一个BaseMemory类集成到ConversationChain或Agent中。LlamaIndex可以将 Chroma 记忆系统作为一个“知识库”使用VectorStoreIndex进行管理并通过QueryEngine进行检索。Dify/Coze在这些低代码平台中你可以将记忆系统封装成一个可调用的工具Tool或技能Skill在工作流节点中调用它来存储和读取记忆。通过遵循以上最佳实践你可以构建出一个健壮、可扩展且高效的智能体记忆系统真正让你的 AI 应用具备“长期记忆”能力大幅提升用户体验和任务完成效率。