最近做 Agent 落地时遇到一个很典型的问题对话一多智能体就开始“失忆”。上午刚确认过的用户偏好下午再问一次它完全不记得上一轮说好的任务约束下一轮直接丢掉跨会话更是基础信息清零。表面看是上下文窗口不够实际上是没有一套真正的记忆系统。这次我们不聊概念直接从 0 开始搭一套企业级 Agent 长期记忆系统。重点解决三件事对话记忆怎么写进去、长期记忆怎么存、需要时怎么检索回来并注入给大模型。整套方案以自建服务为主不绑定某个具体 Agent 框架后面接 LangChain、Dify、Coze 或者自研 Agent 都能用。读完这篇文章你能拿到一套可落地的记忆服务代码结构、向量库选型建议、与 Agent 框架的集成方式以及批量导入历史对话完成记忆初始化的方法。1. 核心能力速览能力项说明面向场景智能体多轮对话记忆、跨会话长期记忆、企业级个性化 Agent 构建记忆模型工作记忆 情景记忆 语义记忆 程序记忆存储方案关系型数据库 向量数据库组合检索方式向量相似度检索 元数据过滤 时间衰减排序部署形态Docker Compose 编排或 Python 服务自部署是否需要 GPU不强依赖向量化环节可选本地 Embedding 模型或调用 API是否支持 API支持提供记忆写入、检索、删除、批量导入接口是否支持批量任务支持可批量导入历史对话支持异步任务队列适合场景客服 Agent、知识库 Agent、个人助理、企业知识管理这套方案的核心思路是把记忆从“提示词里的临时上下文”升级为“独立的记忆存储层”通过标准的写入和检索接口与任意大模型应用对接。2. 为什么 AI Agent 总是失忆先回答一个基础问题大模型本身有没有记忆能力没有。大模型每次推理都是无状态的它只能看到当前请求里携带的文本。所谓“记忆”本质上是在请求前把历史信息拼进上下文或者请求后把新信息持久化到外部存储。常见的“失忆”有三种会话内失忆。上下文窗口有限超出窗口后最早的消息被截断。解决思路是做上下文压缩和摘要。跨会话失忆。每次新会话都是空上下文Agent 完全不记得用户是谁、以前聊过什么。解决思路是持久化存储 检索召回。信息提取失败。对话里明明说了偏好但 Agent 没有把关键信息抽出来存好。解决思路是在记忆写入链路里增加结构化抽取。企业级场景比个人使用更苛刻比如多用户隔离、部门数据权限、记忆合并与冲突处理、过期记忆清理。个人玩具可以把“聊天记录全文存下来”当记忆但企业级必须把记忆作为独立模块做服务化让所有 Agent 共用一套记忆基础设施。3. 长期记忆系统的架构设计企业级 Agent 记忆系统建议分成五个模块记忆写入模块。从对话流里抽取关键信息结构化成记忆条目。记忆存储模块。向量库存语义索引关系库存结构化属性。记忆检索模块。根据当前对话内容召回相关记忆。记忆管理模块。负责合并相似记忆、处理冲突、设置过期时间。接入层。以 API 形式暴露给 Agent 框架。记忆类型可以分成四层工作记忆当前会话内的上下文短期、易变通常由 Agent 框架自己管理。情景记忆过去发生过的具体事件比如“上周用户反馈过登录失败”。语义记忆从历史中提炼出的稳定事实比如“用户偏好使用 Python”。程序记忆Agent 执行任务的流程和规则比如“处理退款时需要两步审批”。长期记忆系统重点处理的是情景记忆和语义记忆。工作记忆继续由上下文窗口负责程序记忆则逐步沉淀到 Agent 的流程配置里。整体数据流如下用户消息 - Agent 主流程 - 查询记忆服务注入相关记忆 - LLM 生成回复 - 从对话中抽取新记忆 - 写入记忆服务 - 返回回复给用户画成流程就是“先查再答、答完再存”这样才能保证下一次对话能用到当前这次的信息。4. 技术选型与环境准备4.1 向量数据库选型长期记忆的核心是语义检索因此向量数据库是必选项。常见的选型方案特点适用场景Chroma轻量Python 内嵌适合快速原型个人项目、内部测试QdrantRust 编写性能好支持 Docker 部署生产环境中等规模Milvus分布式能力更强支持百亿级向量大规模企业场景pgvector基于 PostgreSQL扩展方便已有 PG 基础设施的团队如果团队已经使用了 PostgreSQL建议优先考虑 pgvector可以减少一套中间件的运维成本。如果从零开始且规模不大Qdrant 的体验更顺畅。本文示例代码以统一的向量库接口为例具体 SDK 需要按实际选型替换。4.2 环境依赖操作系统Linux 或 macOS 推荐Windows 可用 WSL2。Python3.10 或更高版本。Docker 与 Docker Compose用于启动向量数据库和可选的关系型数据库。大模型 API使用 OpenAI、国内大模型 API或本地部署的模型服务。Embedding 服务可以是 API 形式也可以本地加载模型。不需要 GPU 也能跑通整个流程。向量化如果使用本地 Embedding 模型则建议准备 CPU 即可除非要为高并发场景做 GPU 加速。4.3 项目目录结构agent-memory/ ├── docker-compose.yml ├── requirements.txt ├── app/ │ ├── main.py │ ├── config.py │ ├── schemas.py │ ├── memory_store.py │ ├── memory_extract.py │ └── api_routes.py ├── tests/ │ └── test_memory.py └── scripts/ └── import_history.py5. Docker 部署向量数据库这里以 Qdrant 为例给出 docker-compose 配置。如果你是第一次接触建议先按这套配置搭建。version: 3.8 services: qdrant: image: qdrant/qdrant:latest container_name: agent-memory-qdrant ports: - 6333:6333 - 6334:6334 volumes: - ./data/qdrant:/qdrant/storage restart: unless-stopped postgres: image: postgres:15 container_name: agent-memory-postgres environment: POSTGRES_USER: memory_user POSTGRES_PASSWORD: memory_pass POSTGRES_DB: agent_memory ports: - 5432:5432 volumes: - ./data/postgres:/var/lib/postgresql/data restart: unless-stopped启动命令docker compose up -d启动后确认服务状态docker ps看到 qdrant 和 postgres 两个容器都在运行说明环境就绪。如果不方便用 Docker也可以直接使用 Chroma 的本地文件模式零外部依赖。但生产环境不建议文件锁和并发问题会让你很难受。6. 记忆服务核心代码实现6.1 配置管理先写配置文件把向量库连接和模型 API 都放在环境变量里。# app/config.py import os class Settings: def __init__(self): self.qdrant_host os.getenv(QDRANT_HOST, 127.0.0.1) self.qdrant_port int(os.getenv(QDRANT_PORT, 6333)) self.collection_name os.getenv(MEMORY_COLLECTION, agent_memory) self.embedding_model os.getenv(EMBEDDING_MODEL, text-embedding-3-small) self.embedding_api_key os.getenv(EMBEDDING_API_KEY, ) self.llm_api_key os.getenv(LLM_API_KEY, ) self.llm_base_url os.getenv(LLM_BASE_URL, ) self.llm_model os.getenv(LLM_MODEL, gpt-4o-mini) self.top_k int(os.getenv(MEMORY_TOP_K, 5)) self.memory_score_threshold float(os.getenv(MEMORY_SCORE_THRESHOLD, 0.3)) settings Settings()6.2 记忆数据模型记忆条目建议至少包含以下字段# app/schemas.py from pydantic import BaseModel, Field from typing import Optional class MemoryItem(BaseModel): memory_id: str user_id: str session_id: str content: str memory_type: str Field(defaultsemantic) # episodic / semantic created_at: str expires_at: Optional[str] None metadata: dict Field(default_factorydict) class MemoryQuery(BaseModel): user_id: str query: str top_k: int 5 memory_type: Optional[str] None class MemoryWriteRequest(BaseModel): user_id: str session_id: str content: str memory_type: str semantic metadata: dict Field(default_factorydict)这些字段基本能覆盖企业级需求特别是 user_id 隔离多用户系统里必须严格区分。6.3 向量化封装为了兼容不同的 Embedding 服务建议封装一层。这里给出一个通用接口具体调用逻辑按你实际的模型 API 调整。# app/memory_store.py import requests def get_embedding(text: str, settings) - list: 统一封装向量化接口需要按实际模型 API 调整 headers { Authorization: fBearer {settings.embedding_api_key}, Content-Type: application/json } payload { model: settings.embedding_model, input: text } url f{settings.embedding_base_url}/embeddings response requests.post(url, jsonpayload, headersheaders, timeout30) response.raise_for_status() return response.json()[data][0][embedding]实际的 Embedding API 地址、鉴权方式、返回结构因服务而异这里只是通用模板。如果你用的是 Ollama 本地 Embedding 模型接口地址和参数都不一样需要自行替换。6.4 记忆写入记忆写入需要先向量化再存入向量库。这里以 Qdrant 为例但保持代码与具体 SDK 解耦需要按实际包调整。# app/memory_store.py import uuid from datetime import datetime, timezone def write_memory(client, settings, memory: MemoryItem): vector get_embedding(memory.content, settings) point_id str(uuid.uuid4()) payload memory.model_dump() payload[memory_id] memory.memory_id or point_id try: client.upsert( collection_namesettings.collection_name, points[{ id: point_id, vector: vector, payload: payload }] ) except Exception as e: raise RuntimeError(fmemory write failed: {e}) return payload[memory_id]写入时要注意不要存敏感明文不要存系统内部密钥也不要无限制存对话全文。好的记忆条目是提炼后的、结构化的、经过筛选的信息。6.5 记忆检索检索是最关键的环节。检索质量直接决定 Agent 能不能想起该想的事。# app/memory_store.py def query_memory(client, settings, query: MemoryQuery): query_vector get_embedding(query.query, settings) query_filter { must: [ {key: user_id, match: {value: query.user_id}} ] } if query.memory_type: query_filter[must].append( {key: memory_type, match: {value: query.memory_type}} ) try: results client.query_points( collection_namesettings.collection_name, queryquery_vector, query_filterquery_filter, limitquery.top_k, with_payloadTrue ) except Exception as e: raise RuntimeError(fmemory query failed: {e}) memories [] for res in results.points: if res.score settings.memory_score_threshold: continue memories.append({ content: res.payload.get(content, ), memory_type: res.payload.get(memory_type, ), score: res.score, created_at: res.payload.get(created_at, ) }) return memories检索时加 user_id 过滤是强制要求目的是防止用户 A 的对话记忆被用户 B 检索到。按 memory_type 过滤可以控制召回范围比如只召回语义记忆。还可以引入时间衰减。简单做法是在返回结果后用 created_at 计算一个衰减系数将 score 乘以系数后再排序。这样近期记忆优先级更高。7. API 服务与调用示例记忆服务最终要以 API 形式暴露给 Agent。推荐使用 FastAPI 封装启动速度快自带 Swagger 文档方便联调。# app/main.py from fastapi import FastAPI from qdrant_client import QdrantClient from config import settings from memory_store import write_memory, query_memory from schemas import MemoryItem, MemoryQuery, MemoryWriteRequest app FastAPI(titleAgent Memory Service, version0.1.0) client QdrantClient( hostsettings.qdrant_host, portsettings.qdrant_port ) app.get(/health) def health_check(): return {status: ok} app.post(/v1/memory) def create_memory(request: MemoryWriteRequest): item MemoryItem( memory_id, user_idrequest.user_id, session_idrequest.session_id, contentrequest.content, memory_typerequest.memory_type, created_atstr(datetime.now(timezone.utc)), metadatarequest.metadata ) memory_id write_memory(client, settings, item) return {memory_id: memory_id} app.post(/v1/memory/query) def search_memory(query: MemoryQuery): memories query_memory(client, settings, query) return {memories: memories}启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000启动后访问http://127.0.0.1:8000/docs可以看到接口文档。调用写入接口curl -X POST http://127.0.0.1:8000/v1/memory \ -H Content-Type: application/json \ -d { user_id: user_001, session_id: session_20250101, content: 用户偏好使用 Python 和 FastAPI 开发后端服务, memory_type: semantic, metadata: {source: chat, confidence: 0.95} }调用检索接口curl -X POST http://127.0.0.1:8000/v1/memory/query \ -H Content-Type: application/json \ -d { user_id: user_001, query: 这个用户偏好什么编程语言, top_k: 5 }在 Agent 主流程里你应该在发送给大模型之前先调用检索接口把返回内容拼到 system prompt 里。比如memory_result requests.post( http://127.0.0.1:8000/v1/memory/query, json{user_id: user_id, query: user_message} ).json() memory_text \n.join([ f- {item[content]} for item in memory_result[memories] ]) system_prompt f 你是企业内部智能助手。 下面是与用户相关的长期记忆可能不完整仅供参考 {memory_text} 请根据这些记忆结合当前对话给出回复。 这样就完成了“先查再答”的第一步。8. 从对话中抽取记忆并自动入库只提供手动写入接口还不够Agent 需要在每轮对话后自动抽取记忆。这一步可以交给大模型来做。通用做法是提供一个抽取模板让 LLM 判断当前对话中是否有值得长期记住的信息。没有就返回空列表有就返回结构化记忆条目。# app/memory_extract.py import json import requests EXTRACT_PROMPT 你是一个记忆抽取器。请从下面的对话中抽取值得长期记住的信息。 要求 1. 只抽取事实性、长期有效的信息 2. 忽略寒暄、临时性内容 3. 输出 JSON 数组格式为 [{content: ..., memory_type: semantic}] 4. 如果没有值得抽取的记忆输出 [] 对话内容 {dialogue} def extract_memory_from_dialogue(dialogue_text: str, settings) - list: prompt EXTRACT_PROMPT.format(dialoguedialogue_text) headers { Authorization: fBearer {settings.llm_api_key}, Content-Type: application/json } payload { model: settings.llm_model, messages: [ {role: system, content: prompt} ], temperature: 0.1 } response requests.post( f{settings.llm_base_url}/chat/completions, jsonpayload, headersheaders, timeout60 ) response.raise_for_status() content response.json()[choices][0][message][content] try: return json.loads(content) except json.JSONDecodeError: return []在 Agent 主流程的“答完再存”阶段调用dialogue f用户说{user_message}\n助手说{bot_reply} new_memories extract_memory_from_dialogue(dialogue, settings) for mem in new_memories: write_memory(client, settings, MemoryItem( memory_id, user_iduser_id, session_idsession_id, contentmem[content], memory_typemem.get(memory_type, semantic), created_atstr(datetime.now(timezone.utc)), metadata{} ))注意抽取动作可以在回复完成后异步执行不影响用户请求的响应速度。企业级系统建议放到消息队列里比如 Redis Stream 或 RabbitMQ。9. 批量导入历史对话很多企业已经有大量历史聊天记录。如果能让 Agent 直接学到这些历史冷启动效果会好很多。批量导入流程# scripts/import_history.py import csv import json history_file history.csv def read_history(file_path): with open(file_path, r, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: yield row def build_dialogue(row): return f用户{row[user_content]}\n助手{row[bot_content]}对每条历史记录调用记忆抽取再将结果批量写入向量库。建议分批提交每批 100 条左右避免请求体和向量库压力过大。批量写入时还要注意只导入有长期价值的内容不要全套照搬。对敏感数据进行脱敏处理后再入库。导入前设置好 user_id确保归属正确。建议先抽 100 条做效果验证再全量导入。10. 资源占用与性能观察记忆服务本身不是一个重资源应用主要消耗来自三部分向量数据库的内存和磁盘。向量的数量决定索引大小每条 Embedding 只占几 KB但千万级数据也要规划磁盘空间。Embedding 服务。如果调用 API费用与调用量成正比如果本地部署CPU 占用会明显上升。LLM 抽取环节。每轮对话后调用 LLM 抽取记忆会带来额外 token 开销。可以在用户不活跃时异步批量处理或者用较小的模型做抽取。观察方法使用docker stats查看容器 CPU 和内存占用。在 FastAPI 接口里记录检索耗时正常情况下单次向量检索应在几十毫秒以内。监控 token 用量统计记忆写入链路消耗的比例。如果检索变慢优先检查集合索引和向量数量。如果记忆注入后回答效果变差优先减少 top_k 或提高 score 阈值。11. 常见问题与排查方法问题现象可能原因排查方式解决方案检索不到相关记忆Embedding 维度不一致或阈值过高检查向量库集合配置打印返回 score 值调低 score 阈值或重新建立集合记忆串用户检索时未按 user_id 过滤检查查询条件是否包含 user_id在 query_filter 中强制增加 user_id记忆内容太杂抽取环节没有过滤掉临时信息查看记忆抽取 prompt 效果优化抽取 prompt增加过滤词表每次对话写入太多记忆没有合并或去重统计单轮写入条数增加相似度去重判断注入记忆后回答被干扰top_k 太大相关记忆混入噪声检查注入的 memory_text减小 top_k提高 score 阈值向量库连接失败服务未启动或端口不对检查 docker ps 和防火墙启动容器或修改连接配置批量导入速度慢逐条调用 Embedding 接口看日志确认耗时分布改为批量向量化加并发记忆存在但回答不用检索返回后未注入 prompt检查 Agent 调用逻辑确认 system prompt 拼接处12. 企业级落地的几个关键建议第一记忆系统必须是独立服务不要跟 Agent 业务代码耦合。记忆是基础设施多个 Agent 应该共享同一套记忆服务只是通过 user_id 和 metadata 做隔离。第二必须做权限设计。不同部门、不同角色能检索到的记忆范围应该不同。至少要做到 user_id 级隔离更严格一点要支持 team_id、org_id 多级过滤。第三记忆写入要做质量控制。宁可少存不要乱存。抽取的 prompt 要反复调优设置置信度字段低于阈值的不入库。第四一定要有删除和遗忘机制。企业合规要求用户有权删除自己的数据。提供记忆删除接口定期清理过期记忆既符合合规要求也能减少检索噪声。第五记忆要与 Agent 评测结合。上线前准备一组“记忆能力测试题”比如“这个用户上个月提过什么需求”验证检索准确率不要只靠主观感受。13. 总结与下一步这套从 0 搭建的 Agent 长期记忆系统核心就三步先查再答、答完再存、定期清理。技术上用向量库做语义检索用 LLM 做记忆抽取用 API 服务把能力暴露给任意 Agent 框架。第一步建议先把记忆写入和检索两个接口跑通用测试数据验证检索效果。第二步再接 Agent 主流程实现对话后自动抽取记忆。第三步再考虑批量导入历史数据、权限隔离和异步化处理。最容易踩的坑是记忆检索质量。很多人以为存进去就能用实际上存了一大堆噪声反而把 Agent 的回答带偏。先把抽取 prompt 和 score 阈值调好再扩大数据量这是成功率最高的路线。后续可以扩展的方向包括记忆自动合并与冲突消解、记忆可视化回放、多 Agent 共享记忆、基于用户反馈的记忆权重调整以及把记忆系统接入 RAG 知识库形成双层检索。先把基础链路跑通这些问题就有了真实的迭代基础。