基于FastAPI与Chroma实现AI长期记忆系统：从概念到代码实践

EverMind-AI/EverOS 从命名上可以读出三层信号Ever 强调时间维度上的持久Mind 指向认知和记忆OS 则暗示 AI 应用需要一套类似操作系统的基础设施而不是只有一次性的对话请求。这类以“AI 原生知识工作空间”为目标的项目核心命题可以概括为如何让 AI 在多次会话、多天工作时长、多条项目任务之间稳定地保留信息并能在需要时把信息重新取回。这里不计论某个仓库的具体功能清单而是把 EverMind-AI/EverOS 当作一个工程方向来看。真正值得拆解的是它提出的一个问题当前大模型虽然有长上下文能力但每次对话结束后模型并不会天然记住用户说过什么、决定过什么、推进到什么状态。如果 AI 要成为长期可用的工作伴侣就必须在模型之外拥有一套“记忆系统”。这篇内容会给出一个可以本地运行的参考实现。它基于 Python FastAPI 和向量数据库 Chroma实现一个最小记忆服务接收一条新记忆、在语义空间里索引、在下一次提问时召回最相关的若干条记忆并把召回内容交给大模型组织回答。整体代码量不大适合作为理解 EverOS 类项目记忆子系统的起点。动手跑通后这套结构可以迁移到个人知识库、会议纪要助手、用户画像记忆、客服多轮对话甚至多智能体调度等场景。1. 先看懂 EverOS 这类项目真正要解决的技术命题1.1 持久记忆和上下文窗口是两个不同的问题很长一段时间里大模型工程讨论主要围绕“上下文窗口能塞多少 token”。窗口确实在变大但它不等于记忆。窗口里的内容来自当前请求或者开发者临时拼接的上下文会话结束后这些内容不会自动沉淀为可复用的长期信息。EverOS 这类项目的切入点就是沿着“长期记忆”方向做基础设施。它们的目标不是挑战模型的能力而是让模型拥有一个“体外大脑”模型在推理时读取记忆在工作后写入新记忆。模型本身不需要把所有历史都背下来只需要把可以压缩、检索和更新的记忆层放在模型之外。从工程结构上看这一层很像操作系统里的文件系统或数据库。AI 应用要运行得久、运行得稳前提是数据能被写入、索引、查询和删除。只把一句提示词越拼越长并不会形成真正的项目记忆。1.2 记忆层承担的五个动作回忆一个完整的记忆闭环至少要覆盖以下链路动作说明常见实现接收接受一条新的原始记忆文本输入、对话记录、任务结果结构化给记忆打上类型、时间、项目等元数据分类、字段抽取索引让记忆可以被语义检索生成向量并写入向量库召回根据当前问题筛选最相关内容向量相似度检索反馈把召回内容注入模型回答流程拼接提示词或上下文只完成接收和索引不完成召回等于把笔记本写满却永远不打开。只完成召回不完成反馈记忆也不会影响 AI 的答案。判断一个自建记忆系统是否成熟最简单的方式就是看这五步是否都能闭环。1.3 参考实现的范围与边界后面的演示不会假设 EverOS 当前代码长什么样也不去复刻它的真实源码。而是构造一个命名为 evermind-ref 的小服务用它演示“AI 记忆系统”最基本的工程形态。服务分为两个边界记忆层负责写入、索引、检索。这是本文核心。回答层负责把召回的上下文交给大模型并生成最终答案。在记忆层中输入可以是任意文本片段输出是一组相似记忆及其距离分数。在回答层中输入是用户问题输出是结合记忆后的回答文本。学习阶段重点看记忆层。生产阶段还要额外考虑权限、审计、数据删除、多租户隔离等事情相关内容会在最后一章展开。2. 搭建环境和项目骨架选择一个本地可跑的最小技术栈2.1 为什么使用 FastAPI、Chroma 和 SentenceTransformer一个便于演示的记忆服务必须满足三个条件本地能跑、依赖不复杂、能够体现语义检索真实过程。FastAPI 负责提供 HTTP API它和 Pydantic 配合方便适合快速写出可测试服务。Chroma 负责存储文本和向量并把向量索引放在本地目录中无需额外启动一个数据库服务。SentenceTransformer 负责把文本转成向量它在本地完成计算不需要配置外部模型服务也能规避“没有 API Key 就无法运行”的问题。这套组合不是生产环境唯一答案但它很适合进入学习链路。生产上如果数据量很大把 Chroma 替换为支持分布式部署的向量数据库并不困难因为上层面向的是统一的 collection 查询语义。2.2 环境要求与依赖版本建议使用 Python 3.10 到 3.12。太新的 Python 版本有时会和 Chroma 底层依赖兼容性不一致如果遇到 sqlite 或 pydantic 相关报错先确认 Python 版本再继续排查。创建的 requirements.txt 内容如下fastapi0.110.0 uvicorn[standard]0.29.0 chromadb0.4.24 sentence-transformers3.0.0 openai1.24.0 python-dotenv1.0.0说明chromadb 负责本地持久化集合。sentence-transformers 负责本地向量化。openai 用来请求兼容 OpenAI Chat Completions 接口的大模型服务。python-dotenv 用来读取 .env 配置文件。安装命令pip install -r requirements.txt在不使用大模型的情况下只依赖 fastapi、uvicorn、chromadb、sentence-transformers、python-dotenv 就能跑通记忆写入和检索。openai 包只有在调用最终问答接口时才需要。2.3 目录结构设计项目结构如下evermind-ref/ ├── app │ ├── __init__.py │ ├── config.py │ ├── schemas.py │ ├── memory_store.py │ ├── llm.py │ └── main.py ├── .env.example ├── requirements.txt └── data/data 目录是运行时动态生成的存放 Chroma 的持久化文件。如果数据需要备份只需要备份这个目录并通过环境变量重新指向项目外的独立路径。2.4 配置入口与数据约定config.py 的职责是从环境变量和 .env 文件读取配置import os from pathlib import Path from dotenv import load_dotenv load_dotenv() class Settings: data_dir: Path Path(os.getenv(EVEROS_DATA_DIR, ./data)) collection_name: str os.getenv(EVEROS_COLLECTION_NAME, everos_memory) embedding_model: str os.getenv(EMBEDDING_MODEL, all-MiniLM-L6-v2) llm_api_key: str os.getenv(LLM_API_KEY, ) llm_base_url: str os.getenv(LLM_BASE_URL, ) llm_model: str os.getenv(LLM_MODEL, gpt-4o-mini) settings Settings()配置项含义如下环境变量默认值说明EVEROS_DATA_DIR./data向量数据库持久化目录EVEROS_COLLECTION_NAMEeveros_memoryChroma collection 名称EMBEDDING_MODELall-MiniLM-L6-v2文本向量化模型LLM_API_KEY空调用大模型服务的密钥LLM_BASE_URL空兼容接口服务地址空时使用 SDK 默认地址LLM_MODELgpt-4o-mini实际模型名需要注意load_dotenv 默认只从当前工作目录读取 .env。如果 uvicorn 不是从项目根目录启动环境变量可能加载不到。稳妥做法是显式指定路径或者在启动前手动 export。3. 核心实现把“记住”和“回忆”变成 HTTP API3.1 先用 Pydantic 定义记忆的数据模型开始写代码前先定义输入输出的数据结构后续所有模块都围绕这套结构展开。schemas.py 内容如下from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field MemoryKind Literal[fact, episode, task] class MemoryCreate(BaseModel): content: str Field(..., min_length1, max_length2000, description记忆正文) kind: MemoryKind Field(defaultfact, description记忆类型) memory_id: Optional[str] Field(defaultNone, description业务侧记忆ID为空则自动生成) metadata: Dict[str, Any] Field(default_factorydict, description附加元数据) class MemoryRecord(BaseModel): memory_id: str kind: str content: str created_at: str metadata: Dict[str, Any] class RecallRequest(BaseModel): query: str Field(..., min_length1, description用户问题或检索语句) top_k: int Field(default3, ge1, le20, description返回记忆条数) kind: Optional[str] Field(defaultNone, description按记忆类型过滤) class MemoryHit(BaseModel): memory: MemoryRecord distance: float class RecallResponse(BaseModel): query: str hits: List[MemoryHit] class AskRequest(BaseModel): question: str Field(..., min_length1) top_k: int Field(default3, ge1, le20) class AskResponse(BaseModel): answer: str recalled: List[MemoryHit]这里把记忆类型限制为 fact、episode、task 三种分别表示事实、交互片段和任务状态。类型不是严格的数据库约束但它在实际项目中很有用。例如回答“用户公司主营什么”时只需要查 fact 类型而复盘“上次执行到哪一步”时更适合查 task 类型。3.2 MemoryStore处理元数据归一化Chroma 的 metadata 只能保存 str、int、float、bool 这类标量值。如果直接写入 dict 或 list会抛类型异常。因此写入前需要做一次拍平处理把非标量转换成 JSON 字符串。memory_store.py 的完整实现如下import json import uuid from datetime import datetime, timezone from typing import Any, Optional import chromadb from chromadb.utils import embedding_functions from .schemas import MemoryCreate, MemoryRecord def _normalise_value(value: Any) - Any: if isinstance(value, (str, int, float, bool)): return value return json.dumps(value, ensure_asciiFalse) def _normalise_metadata(metadata: dict[str, Any]) - dict[str, Any]: return {str(key): _normalise_value(value) for key, value in metadata.items()} class MemoryStore: def __init__(self, settings): self.settings settings settings.data_dir.mkdir(parentsTrue, exist_okTrue) embed_fn embedding_functions.SentenceTransformerEmbeddingFunction( model_namesettings.embedding_model ) self.client chromadb.PersistentClient(pathstr(settings.data_dir)) self.collection self.client.get_or_create_collection( namesettings.collection_name, embedding_functionembed_fn, metadata{hnsw:space: cosine}, ) def add_memory(self, payload: MemoryCreate) - MemoryRecord: content payload.content.strip() if not content: raise ValueError(memory content cannot be empty) memory_id payload.memory_id or str(uuid.uuid4()) created_at datetime.now(timezone.utc).isoformat() metadata { kind: payload.kind, created_at: created_at, } metadata.update(_normalise_metadata(payload.metadata)) self.collection.add( ids[memory_id], documents[content], metadatas[metadata], ) return MemoryRecord( memory_idmemory_id, kindpayload.kind, contentcontent, created_atcreated_at, metadatapayload.metadata, ) def recall( self, query: str, top_k: int, kind: Optional[str] None, ) - tuple[list[MemoryRecord], list[float]]: count self.collection.count() if count 0: return [], [] safe_k max(1, min(top_k, count)) kwargs: dict[str, Any] { query_texts: [query], n_results: safe_k, include: [documents, metadatas, distances], } if kind: kwargs[where] {kind: kind} result self.collection.query(**kwargs) ids result[ids][0] documents result[documents][0] metadatas result[metadatas][0] distances result[distances][0] memories [] for index in range(len(ids)): meta metadatas[index] or {} extra_metadata { key: value for key, value in meta.items() if key not in (kind, created_at) } memories.append( MemoryRecord( memory_idids[index], kindmeta.get(kind, fact), contentdocuments[index], created_atmeta.get(created_at, ), metadataextra_metadata, ) ) return memories, distances几个关键点需要解释存储时把 kind 和 created_at 放进 Chroma 的 metadata是为了支持按类型过滤和返回时间。返回给调用方时又把 kind 和 created_at 从普通 metadata 中拆出来让 API 响应更清晰。safe_k max(1, min(top_k, count))是为了避免分页边界错误。Chroma 在请求条数超过集合内文档数时可能直接报错。配置{hnsw:space: cosine}表示使用余弦距离计算相似度。distance 越小表示语义越接近。3.3 召回入口与过滤逻辑“回忆”接口其实不需要单独写另一个类。真正要做的是把用户的查询文本转成向量再在集合中做向量相似度搜索。向量搜索和关键词搜索不一样。用户如果问“上周末我设计了什么功能”关键词搜索只能匹配“上周末”这类字面词很难把“周六下午完成了记忆模块设计”这条记录捞回来。向量搜索把两句话映射到同一个语义空间里即使字面上不重叠只要含义接近距离也会较近。因此 recall 方法保留了 kind 过滤能力。在多条记忆混合存放时可以继续使用 metadata。比如检索任务状态时限定kindtask避免把用户闲聊内容也带进答案。3.4 LLM 模块把召回记忆组装成提示词记忆检索完成后需要把召回结果送给大模型。这里的关键问题是提示词组装不能让模型把所有召回都当成真实结论必须让模型区分“相关记忆”和“无关记忆”。llm.py 内容如下from typing import Any from openai import OpenAI from .config import settings def build_prompt(question: str, recalled_items: list[dict[str, Any]]) - list[dict[str, str]]: lines [] for item in recalled_items: lines.append(f- 类型({item[kind]}): {item[content]}) memory_text \n.join(lines) if lines else - 暂无相关记忆 system_content ( 你是一个带长期记忆的 AI 助手。 回答问题时请优先依据“召回记忆”中与当前问题相关的信息。 如果召回记忆与问题无关请明确说明没有找到相关记忆