尧图网页设计 
AI Agent 和“失忆”这两个词放在一起几乎成了日常你刚跟智能体说完自己的项目背景下一轮对话它又忘了你让它记住用户的偏好它却只记得最后一句话企业内部的知识库、客户资料、历史决策记录稍微一多Agent 就开始“断片”。这不是模型不够聪明而是它根本没有“记忆体”——大模型本身是一次性推理引擎每次调用都从零开始。如果要在企业场景里让 Agent 真正可用就必须给它搭一套长期记忆系统能写入、能检索、能更新、能过期能跟对话上下文、知识库、任务状态打通。这篇文章会从问题拆解开始讲清楚记忆系统该分几层、需要哪些存储组件、怎么设计存取接口然后给出一个可以直接落地的项目骨架、API 示例和批量任务方案。最后补上性能观察、常见排查和合规边界。不管你是在自建 Agent 框架还是准备给现有智能体产品加记忆能力这篇文章都可以当作一份基础工程手册。1. 核心能力速览能力项说明记忆类型短期记忆当前上下文、工作记忆任务状态、长期记忆跨会话持久化存储方案向量数据库做语义检索 关系型数据库做结构化元数据管理检索方式语义检索、关键词检索、混合检索按场景选择核心功能记忆写入、记忆读取、记忆更新、记忆删除、记忆过期、批量导入上下文管理从长期记忆中召回相关片段注入到 System Prompt 或消息历史API 能力提供 REST API 或内部调用接口支持写入、检索、清理、批量任务批量任务历史对话批量导入、记忆抽取、定时压缩、遗忘策略推荐环境Linux / macOS / Windows WSLPython 3.10需联网访问 LLM API 或本地模型服务扩展性可对接企业知识库、多 Agent 共享记忆、权限隔离这套能力设计的核心思路是记忆不应该是一张越堆越长的聊天记录而应该是一种“可管理的数据资产”。写入的时候做清洗和抽取读取的时候做相关性和时效性筛选长期运行后还要有归档和遗忘机制。2. 智能体为什么“失忆”问题拆解先看一个真实现象用户对 Agent 说“我之前提交过一个关于华东区客户流失的分析报告”Agent 完全不知道。原因不是模型没听懂而是这次请求里根本没带“之前”的信息。大模型调用是无状态的每一次 API 请求只含当前输入的 prompt上下文窗口关掉之后一切归零。再深一层即使你把之前的对话拼进这次 prompt也会遇到三个问题上下文窗口有上限。模型输入长度有限制历史对话达到几十万 token 之后装不下也没必要全装。相关性问题。一万条历史记忆里只有几条跟当前任务相关全塞进去反而会干扰模型判断。时效性问题。昨天的信息可能已经过期比如客户公司联系人已经更换旧记录还留在记忆库里Agent 仍然会用旧数据回复。企业级场景更复杂。智能体需要服务的不是一个人聊天而是客户关系管理、项目复盘、工单处理、知识库问答这些场景要求 Agent 能记住“这个客户的项目阶段”“这个工单的处理进度”“这个团队的技术偏好”。所以“长期记忆系统”本质上要解决的不是“变聪明”而是“有据可查”。可以这样理解记忆拆解短期记忆当前会话窗口内的用户消息、助手回复、工具调用结果。工作记忆当前任务执行到哪一步依赖哪些前置条件需要哪些中间结果。长期记忆跨会话保存的用户画像、业务事实、历史决策、知识点、项目背景。全局记忆多个 Agent 之间共享的团队知识、企业知识库、公共规则。真正的企业级 Agent 记忆系统至少要把这三层分开存储、分开管理。否则就会出现“表面记得、实际不可用”的问题。3. 企业级 Agent 记忆系统架构设计一个可落地的记忆系统可以拆成几个模块。3.1 整体架构Agent 运行时负责调用大模型、执行工具、组织提示词。记忆模块只被 Agent 运行时依赖不直接暴露给用户。记忆服务层提供写入、检索、更新、删除、过期接口是记忆系统的核心。向量存储存储记忆条目的向量表示用于语义检索。结构化存储存储记忆的元数据、实体关系、时间戳、来源会话、权限标签。抽取与归档任务从对话历史中提取记忆定时对旧记忆做压缩、删除、转存。LLM 调用模块负责生成 embedding、抽取关键事实、判断记忆是否需要更新。文字流程是这样的用户对话进入 Agent 运行时Agent 先把当前请求发送给记忆服务做检索召回相关历史记忆把这些记忆拼进 prompt然后调用大模型生成回复。生成完成后如果用户消息或助手回复里有值得记录的新事实再调用记忆服务写入一条新的记忆或更新旧记录。这里最容易踩的坑是“所有记忆都写进去”。正确做法是设置抽取规则和置信度阈值比如用户明确说“我偏好简洁回复”、用户提交了具体的项目需求、助手给出了结论性建议这些才是值得沉淀的记忆。3.2 记忆数据模型把一条记忆看成一条可检索、可追溯的数据记录大致包括字段说明memory_id记忆唯一标识user_id所属用户或会话主体agent_id关联的 Agent 实例session_id来源会话content记忆文本内容summary对记忆的摘要或标签entities实体列表如人名、公司名、项目名memory_type用户偏好、业务事实、任务状态、知识片段importance重要程度评分created_at创建时间updated_at更新时间expires_at过期时间可选access_scope访问范围控制权限如私有、团队、全局内容文本用来生成向量实体可以用于构建知识图谱元数据用于过滤和权限控制。如果用关系型数据库存元数据向量库只存 embedding 和向量对应的 memory_id效果会比较清晰。4. 环境准备与前置条件长期记忆系统的技术栈不复杂但依赖较多先列一套通用环境清单按实际项目做裁剪。4.1 操作系统与运行环境操作系统Linux 优先macOS 也可以Windows 建议使用 WSL2。Python 版本3.10推荐 3.11。虚拟环境工具venv 或 conda。容器环境Docker用于启动向量数据库和关系型数据库。4.2 外部依赖清单依赖项用途建议LLM API 或本地模型对话生成、事实抽取、记忆摘要OpenAI / 通义 / 本地 vLLM 等按企业环境选Embedding 模型将记忆文本转为向量text-embedding 系列或本地 embedding 模型向量数据库语义检索Milvus、Chroma、Weaviate、Qdrant 等关系型数据库存储元数据PostgreSQL、SQLiteRedis缓存和短期记忆可选Python 依赖库开发记忆服务fastapi、pydantic、requests、langchain 等如果没有部署向量数据库可以先用轻量的 SQLite 存储测试再用向量库替换检索模块。生产环境建议用独立数据库服务。4.3 Python 环境检查先确认基础环境。python --version pip --version创建虚拟环境。python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate安装基础依赖。pip install fastapi uvicorn pydantic requests openai如果计划使用向量数据库客户端再按所选库安装。比如使用 Chroma。pip install chromadb langchain需要说明依赖安装失败时优先检查 Python 版本和 pip 源必要时使用国内镜像源。5. 从 0 开始搭建项目骨架与代码实现下面给一个可以实际运行的项目骨架包含记忆服务、存储抽象和 API 层。代码设计上不绑定某个特定向量数据库方便替换为企业自建组件。5.1 项目目录结构agent-memory/ ├── app │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── models.py # 数据模型 │ ├── memory_store.py # 记忆存储抽象与实现 │ ├── memory_service.py # 记忆业务逻辑 │ └── vector_store.py # 向量库适配器 ├── data # 本地测试数据目录 ├── requirements.txt └── README.md5.2 数据模型models.py里定义记忆条目的数据结构。from pydantic import BaseModel, Field from typing import Optional from datetime import datetime class MemoryItem(BaseModel): memory_id: str user_id: str agent_id: str session_id: str content: str summary: str entities: list[str] [] memory_type: str general importance: float Field(default0.5, ge0.0, le1.0) created_at: datetime Field(default_factorydatetime.now) updated_at: datetime Field(default_factorydatetime.now) expires_at: Optional[datetime] None access_scope: str private5.3 记忆存储抽象from abc import ABC, abstractmethod from .models import MemoryItem class MemoryStore(ABC): abstractmethod def add(self, item: MemoryItem) - None: pass abstractmethod def get(self, memory_id: str) - MemoryItem | None: pass abstractmethod def update(self, item: MemoryItem) - None: pass abstractmethod def delete(self, memory_id: str) - None: pass abstractmethod def query_by_metadata(self, user_id: str, **filters) - list[MemoryItem]: pass5.4 基于 SQLite 的内存存储实现先用 SQLite 实现一套保证项目可以跑通。向量检索部分后续再接。import sqlite3 import json from datetime import datetime from .models import MemoryItem from .memory_store import MemoryStore class SQLiteMemoryStore(MemoryStore): def __init__(self, db_path: str ./data/memory.db): self.db_path db_path self._init_db() def _init_db(self) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS memories ( memory_id TEXT PRIMARY KEY, user_id TEXT, agent_id TEXT, session_id TEXT, content TEXT, summary TEXT, entities TEXT, memory_type TEXT, importance REAL, created_at TEXT, updated_at TEXT, expires_at TEXT, access_scope TEXT ) ) staticmethod def _serialize_item(item: MemoryItem) - tuple: return ( item.memory_id, item.user_id, item.agent_id, item.session_id, item.content, item.summary, json.dumps(item.entities, ensure_asciiFalse), item.memory_type, item.importance, item.created_at.isoformat(), item.updated_at.isoformat(), item.expires_at.isoformat() if item.expires_at else None, item.access_scope, ) staticmethod def _deserialize_row(row: tuple) - MemoryItem: return MemoryItem( memory_idrow[0], user_idrow[1], agent_idrow[2], session_idrow[3], contentrow[4], summaryrow[5], entitiesjson.loads(row[6]), memory_typerow[7], importancerow[8], created_atdatetime.fromisoformat(row[9]), updated_atdatetime.fromisoformat(row[10]), expires_atdatetime.fromisoformat(row[11]) if row[11] else None, access_scoperow[12], ) def add(self, item: MemoryItem) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO memories ( memory_id, user_id, agent_id, session_id, content, summary, entities, memory_type, importance, created_at, updated_at, expires_at, access_scope ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , self._serialize_item(item), ) def get(self, memory_id: str) - MemoryItem | None: with sqlite3.connect(self.db_path) as conn: cursor conn.execute( SELECT * FROM memories WHERE memory_id ?, (memory_id,), ) row cursor.fetchone() return self._deserialize_row(row) if row else None def update(self, item: MemoryItem) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( UPDATE memories SET content ?, summary ?, entities ?, memory_type ?, importance ?, updated_at ?, expires_at ?, access_scope ? WHERE memory_id ? , ( item.content, item.summary, json.dumps(item.entities, ensure_asciiFalse), item.memory_type, item.importance, item.updated_at.isoformat(), item.expires_at.isoformat() if item.expires_at else None, item.access_scope, item.memory_id, ), ) def delete(self, memory_id: str) - None: with sqlite3.connect(self.db_path) as conn: conn.execute(DELETE FROM memories WHERE memory_id ?, (memory_id,)) def query_by_metadata(self, user_id: str, **filters) - list[MemoryItem]: conditions [user_id ?] params: list [user_id] if memory_type in filters: conditions.append(memory_type ?) params.append(filters[memory_type]) if agent_id in filters: conditions.append(agent_id ?) params.append(filters[agent_id]) sql fSELECT * FROM memories WHERE { AND .join(conditions)} ORDER BY updated_at DESC with sqlite3.connect(self.db_path) as conn: cursor conn.execute(sql, params) rows cursor.fetchall() return [self._deserialize_row(row) for row in rows]5.5 记忆业务逻辑记忆服务里加入写入、更新、检索、判断记忆是否过期的业务逻辑。from typing import Optional from .models import MemoryItem from .memory_store import MemoryStore class MemoryService: def __init__(self, store: MemoryStore): self.store store def create_memory( self, user_id: str, agent_id: str, session_id: str, content: str, summary: str , entities: list[str] | None None, memory_type: str general, importance: float 0.5, access_scope: str private, expires_at: Optional[str] None, ) - MemoryItem: item MemoryItem( memory_idf{user_id}_{session_id}_{int(__import__(time).time())}, user_iduser_id, agent_idagent_id, session_idsession_id, contentcontent, summarysummary, entitiesentities or [], memory_typememory_type, importanceimportance, access_scopeaccess_scope, expires_atexpires_at, ) self.store.add(item) return item def update_memory(self, item: MemoryItem) - None: item.updated_at __import__(datetime).datetime.now() self.store.update(item) def list_memories(self, user_id: str, memory_type: str | None None) - list[MemoryItem]: items self.store.query_by_metadata(user_id, memory_typememory_type) return [item for item in items if not self._is_expired(item)] staticmethod def _is_expired(item: MemoryItem) - bool: if item.expires_at is None: return False now __import__(datetime).datetime.now(item.expires_at.tzinfo) return now item.expires_at这段代码的核心是把记忆当作结构化数据管理不只是“聊天记录”。当你把记忆抽象成可查询、可更新、可过期的实体后续接向量检索、批量任务都会顺手很多。6. 感知记忆与对话恢复核心功能实现存储层搭好后真正需要解决的是“Agent 怎么用上这些记忆”。6.1 对话历史到记忆的抽取对话并不能直接全部写入记忆库否则越存越多、越存越乱。一般做法是先让 LLM 从对话里抽取关键事实和用户意图再按规则写入。一次简化抽取的输出格式可以是这样的 JSON{ memory_candidates: [ { content: 客户华东区项目目前处于需求确认阶段预算约 80 万决策人是张总。, entities: [张总, 华东区项目], memory_type: project, importance: 0.9 }, { content: 用户偏好每周五下午三点接收项目周报。, entities: [周报], memory_type: preference, importance: 0.7 } ] }拿到这个结果后程序先做过滤再去重然后调用记忆服务写入。抽取这一步可以用大模型的 API 完成也可以先写基于规则的版本。6.2 记忆召回与 Prompt 注入每次新对话开始时先从记忆服务召回跟当前用户、当前主题相关的记忆。伪代码流程def build_prompt_with_memory(user_query, user_id): # 1. 检索相关记忆 memories memory_service.retrieve(user_id, user_query) # 2. 拼装记忆上下文 memory_block format_memories(memories) # 3. 拼入 system prompt system_prompt f 你是企业智能助手。请基于以下已知信息回答用户问题 {memory_block} 如果信息不足请明确说明不要编造。 return system_prompt这里记忆召回有两种方式元数据过滤按用户、Agent、记忆类型直接查。语义检索用 embedding 模型把用户查询向量化在向量库召回相近记忆再取 Top-K 条。企业级系统里建议混合使用先做权限过滤再做语义召回最后按重要度、时间排序。这样可以避免“查到了但不能看”和“看到了但无关”这两种情况。6.3 记忆更新记忆不是“只写一次”。当用户说“我改主意了项目总共 100 万不是 80 万”时系统需要找到旧的那条记忆判断冲突然后更新。这一步没有固定逻辑推荐方式是LLM 对比新信息和旧记忆生成新的记忆版本同时保留历史版本到审计表。对关键业务记忆建议记录变更历史便于回溯。7. 接口 API 与批量任务设计记忆系统最终要暴露给上层 Agent 调用一般用 REST API。7.1 API 设计方法路径功能POST/v1/memories写入记忆GET/v1/memories/{memory_id}获取指定记忆GET/v1/memories按用户、类型查询记忆PATCH/v1/memories/{memory_id}更新记忆DELETE/v1/memories/{memory_id}删除记忆POST/v1/memories/retrieve输入查询文本返回相关记忆 Top-KPOST/v1/memories/batch批量写入记忆POST/v1/memories/archive触发批量归档任务API 服务用 FastAPI 实现。下面给一个可运行的最小示例。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from .models import MemoryItem from .memory_service import MemoryService from .memory_store import SQLiteMemoryStore app FastAPI(titleAgent Memory Service) store SQLiteMemoryStore() service MemoryService(store) class CreateMemoryRequest(BaseModel): user_id: str agent_id: str session_id: str content: str summary: str entities: list[str] [] memory_type: str general importance: float 0.5 access_scope: str private expires_at: Optional[str] None app.post(/v1/memories, response_modelMemoryItem) def create_memory(req: CreateMemoryRequest): return service.create_memory( user_idreq.user_id, agent_idreq.agent_id, session_idreq.session_id, contentreq.content, summaryreq.summary, entitiesreq.entities, memory_typereq.memory_type, importancereq.importance, access_scopereq.access_scope, expires_atreq.expires_at, ) app.get(/v1/memories) def list_memories(user_id: str, memory_type: Optional[str] None): return service.list_memories(user_id, memory_typememory_type) app.post(/v1/memories/retrieve) def retrieve_memories(user_id: str, query: str, top_k: int 5): # 这里先做元数据查询向量检索接入后替换为混合检索 items service.list_memories(user_id) return sorted(items, keylambda x: x.importance, reverseTrue)[:top_k]启动服务uvicorn app.main:app --host 0.0.0.0 --port 8100启动后可以访问http://127.0.0.1:8100/docs查看 Swagger 文档。7.2 curl 调用示例写入一条记忆curl -X POST http://127.0.0.1:8100/v1/memories \ -H Content-Type: application/json \ -d { user_id: user_001, agent_id: agent_sales, session_id: session_001, content: 客户华东区项目预算为 100 万决策人是张总。, memory_type: project, importance: 0.9 }查询记忆curl http://127.0.0.1:8100/v1/memories?user_iduser_0017.3 Python 调用示例import requests base_url http://127.0.0.1:8100/v1/memories payload { user_id: user_001, agent_id: agent_sales, session_id: session_002, content: 用户偏好每周五下午三点接收项目周报。, memory_type: preference, importance: 0.7, } response requests.post(base_url, jsonpayload, timeout10) print(response.status_code) print(response.json())7.4 批量任务设计批量任务主要处理三类情况。批量导入历史对话把之前积累的客服记录、工单记录、聊天记录一次性导入记忆库。导入流程是读取源数据、清洗、拆分成片段、调用 LLM 抽取记忆、写入存储。由于数据量大建议异步执行分批写入。批量归档与压缩长期运行后记忆会膨胀。定时任务可以把三个月前的旧记忆做摘要压缩只保留摘要和高重要度条目删除低价值明细。定时遗忘策略按照expires_at字段和遗忘规则每天清理过期记忆或者把权限变更、离职员工等特殊数据做删除或脱敏。批量任务建议使用消息队列或定时调度框架。最简单的方案是 FastAPI 的 BackgroundTasks复杂场景用 Celery。from fastapi import BackgroundTasks def run_batch_import(file_path: str): # 从文件读取历史记录调用抽取逻辑批量写入记忆库 pass app.post(/v1/memories/batch) def batch_import_memories(file_path: str, background_tasks: BackgroundTasks): background_tasks.add_task(run_batch_import, file_path) return {status: started}批量任务必须考虑失败重试。推荐做法是每条记忆写入都记录状态写入失败的放入重试队列整体任务结束后生成失败报告。8. 资源占用与性能观察记忆系统的资源占用主要来自三部分。第一是存储空间。关系型数据库存元数据向量数据库存向量数据量取决于记忆条目数和 embedding 维度。记忆条目越多检索耗时和存储占用都会增长。建议观察索引大小和单条记忆的平均耗时提前规划分库分表或分区策略。第二是 embedding 生成耗时。每条记忆写入前需要转成向量查询时也需要把用户 query 转成向量。如果用的是远程 embedding API网络延迟会直接影响接口响应。本地 embedding 模型则占用 GPU 或 CPU 资源。性能观察重点看单次 embedding 耗时的 P95、P99。第三是 LLM抽取耗时。从对话里抽取关键事实时如果每个用户消息都同步调用大模型响应会非常慢。更稳妥的做法是对话结束后异步抽取。也就是说用户消息先进入短期记忆等用户离开本轮对话或会话结束再异步把值得沉淀的内容写入长期记忆。资源占用观察清单观察项观察方式关注指标API 响应耗时日志中间件P95、P99embedding 耗时端到端链路埋点单次耗时向量库存储空间向量库监控面板索引大小、文档数量数据库连接数连接池监控活跃连接数内存占用进程内存采样常驻内存批量任务耗时任务日志处理速率、失败率性能优化的通用路径用量小的场景先保持简单量上升后再引入缓存。短期记忆放 Redis长期记忆按用户维度做冷热分层热数据放内存或高性能存储冷数据走低频检索路径。检索链路可以做召回-重排两阶段向量召回 Top 100再用精排模型或规则重排到 Top 5效果和性能都能兼顾。9. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 仍然不记得用户信息长期记忆未在 prompt 中注入查看 prompt 是否包含记忆块检查检索接口是否返回记忆确认注入逻辑记忆检索结果不相关embedding 模型不匹配观察检索日志中 topK 分数更换 embedding 模型或改用混合检索同一事实有多条矛盾记忆写入时未做冲突检测查询记忆库中的重复条目增加实体级去重和更新逻辑记忆膨胀接口变慢缺少过期和压缩任务检查记忆条目的时间分布增加 TTL、归档和删除策略批量导入任务卡住单批数据量过大或 LLM 调用超时查看任务日志和重试次数分批处理、增加超时时间、失败重试权限相关数据泄漏检索时未过滤 access_scope检查查询条件增加范围过滤测试跨用户检索场景API 返回 500数据库表结构或字段错误查看服务日志堆栈检查数据库迁移和字段类型向量库和元数据库数据不一致双写时缺少事务机制比对 memory_id 数量增加补偿任务定期对账这里有一条通用排查原则记忆系统出问题时先验证数据是否写入成功再验证检索是否返回最后验证 prompt 是否正确注入。多数问题都出在其中一个环节。10. 最佳实践与使用建议第一先做最小闭环。不需要一开始就上完整向量数据库和分布式存储。先用 SQLite 存元数据加上一个简单的关键词检索跑通“写入-查询-注入 prompt”这条链路。发现确实需要语义检索再加向量库。第二把记忆分层不把所有内容写进同一张表。用户偏好、业务事实、任务状态、知识片段分门别类不同类型用不同检索策略。比如用户偏好高频使用、任务状态要实时更新、业务事实要可审计。第三给记忆加权限边界。企业场景里隐私和数据安全是底线。同一用户的记忆不应该被其他用户检索到团队知识要有团队级权限客户数据要遵守企业数据管理规范。建议所有检索入口都带上 access_scope 过滤。第四记忆内容要可追溯。每条记忆都记录来源会话和创建时间关键记忆要支持查看历史版本。这样当 Agent 回复出错时可以追溯到是哪条记忆影响了决策。第五定期评估记忆质量。抽一批测试问题观察 Agent 引用记忆中信息的准确率。如果频繁出错大概率是记忆内噪声太多或者更新策略不够严格需要调低“全量写入”的冲动。第六涉及人脸、声音、客户隐私、企业内部文档等数据时必须确认合法授权和数据合规边界。记忆系统不能成为非法收集用户信息的工具批量导入历史数据前要核实数据来源是否合法。11. 总结与下一步这篇文章的核心观点是AI Agent 的“记忆”不是模型参数里长出来的而是一个需要刻意设计、分层管理、持续维护的工程系统。先解决“无状态”问题再解决“检索不准”和“记忆膨胀”问题最后解决“多 Agent 共享记忆”和企业权限问题。最先应该验证的功能是给 Agent 加一个记忆写入接口和检索接口让它能在下一轮对话中看到上一轮沉淀的信息。只要这条链路通了后续的向量检索、批量导入、记忆压缩都只是不断加码。最容易踩的坑是把所有对话记录都当作长期记忆写入不做抽取、不去重、不过期。这样系统的检索质量会迅速劣化Agent 会从“失忆”变成“记忆混乱”。下一步可以继续扩展的方向包括多 Agent 共享记忆团队级知识库接入基于实体的知识图谱记忆以及让 Agent 能自主学习更新记忆的强化反馈机制。建议先搭一套最小记忆服务写入、读取、接口、定时清理跑一周真实对话数据再决定要不要引入向量检索和分布式存储。 AI Agent长期记忆系统架构设计与实现：从失忆到有据可查
AI Agent 和“失忆”这两个词放在一起几乎成了日常你刚跟智能体说完自己的项目背景下一轮对话它又忘了你让它记住用户的偏好它却只记得最后一句话企业内部的知识库、客户资料、历史决策记录稍微一多Agent 就开始“断片”。这不是模型不够聪明而是它根本没有“记忆体”——大模型本身是一次性推理引擎每次调用都从零开始。如果要在企业场景里让 Agent 真正可用就必须给它搭一套长期记忆系统能写入、能检索、能更新、能过期能跟对话上下文、知识库、任务状态打通。这篇文章会从问题拆解开始讲清楚记忆系统该分几层、需要哪些存储组件、怎么设计存取接口然后给出一个可以直接落地的项目骨架、API 示例和批量任务方案。最后补上性能观察、常见排查和合规边界。不管你是在自建 Agent 框架还是准备给现有智能体产品加记忆能力这篇文章都可以当作一份基础工程手册。1. 核心能力速览能力项说明记忆类型短期记忆当前上下文、工作记忆任务状态、长期记忆跨会话持久化存储方案向量数据库做语义检索 关系型数据库做结构化元数据管理检索方式语义检索、关键词检索、混合检索按场景选择核心功能记忆写入、记忆读取、记忆更新、记忆删除、记忆过期、批量导入上下文管理从长期记忆中召回相关片段注入到 System Prompt 或消息历史API 能力提供 REST API 或内部调用接口支持写入、检索、清理、批量任务批量任务历史对话批量导入、记忆抽取、定时压缩、遗忘策略推荐环境Linux / macOS / Windows WSLPython 3.10需联网访问 LLM API 或本地模型服务扩展性可对接企业知识库、多 Agent 共享记忆、权限隔离这套能力设计的核心思路是记忆不应该是一张越堆越长的聊天记录而应该是一种“可管理的数据资产”。写入的时候做清洗和抽取读取的时候做相关性和时效性筛选长期运行后还要有归档和遗忘机制。2. 智能体为什么“失忆”问题拆解先看一个真实现象用户对 Agent 说“我之前提交过一个关于华东区客户流失的分析报告”Agent 完全不知道。原因不是模型没听懂而是这次请求里根本没带“之前”的信息。大模型调用是无状态的每一次 API 请求只含当前输入的 prompt上下文窗口关掉之后一切归零。再深一层即使你把之前的对话拼进这次 prompt也会遇到三个问题上下文窗口有上限。模型输入长度有限制历史对话达到几十万 token 之后装不下也没必要全装。相关性问题。一万条历史记忆里只有几条跟当前任务相关全塞进去反而会干扰模型判断。时效性问题。昨天的信息可能已经过期比如客户公司联系人已经更换旧记录还留在记忆库里Agent 仍然会用旧数据回复。企业级场景更复杂。智能体需要服务的不是一个人聊天而是客户关系管理、项目复盘、工单处理、知识库问答这些场景要求 Agent 能记住“这个客户的项目阶段”“这个工单的处理进度”“这个团队的技术偏好”。所以“长期记忆系统”本质上要解决的不是“变聪明”而是“有据可查”。可以这样理解记忆拆解短期记忆当前会话窗口内的用户消息、助手回复、工具调用结果。工作记忆当前任务执行到哪一步依赖哪些前置条件需要哪些中间结果。长期记忆跨会话保存的用户画像、业务事实、历史决策、知识点、项目背景。全局记忆多个 Agent 之间共享的团队知识、企业知识库、公共规则。真正的企业级 Agent 记忆系统至少要把这三层分开存储、分开管理。否则就会出现“表面记得、实际不可用”的问题。3. 企业级 Agent 记忆系统架构设计一个可落地的记忆系统可以拆成几个模块。3.1 整体架构Agent 运行时负责调用大模型、执行工具、组织提示词。记忆模块只被 Agent 运行时依赖不直接暴露给用户。记忆服务层提供写入、检索、更新、删除、过期接口是记忆系统的核心。向量存储存储记忆条目的向量表示用于语义检索。结构化存储存储记忆的元数据、实体关系、时间戳、来源会话、权限标签。抽取与归档任务从对话历史中提取记忆定时对旧记忆做压缩、删除、转存。LLM 调用模块负责生成 embedding、抽取关键事实、判断记忆是否需要更新。文字流程是这样的用户对话进入 Agent 运行时Agent 先把当前请求发送给记忆服务做检索召回相关历史记忆把这些记忆拼进 prompt然后调用大模型生成回复。生成完成后如果用户消息或助手回复里有值得记录的新事实再调用记忆服务写入一条新的记忆或更新旧记录。这里最容易踩的坑是“所有记忆都写进去”。正确做法是设置抽取规则和置信度阈值比如用户明确说“我偏好简洁回复”、用户提交了具体的项目需求、助手给出了结论性建议这些才是值得沉淀的记忆。3.2 记忆数据模型把一条记忆看成一条可检索、可追溯的数据记录大致包括字段说明memory_id记忆唯一标识user_id所属用户或会话主体agent_id关联的 Agent 实例session_id来源会话content记忆文本内容summary对记忆的摘要或标签entities实体列表如人名、公司名、项目名memory_type用户偏好、业务事实、任务状态、知识片段importance重要程度评分created_at创建时间updated_at更新时间expires_at过期时间可选access_scope访问范围控制权限如私有、团队、全局内容文本用来生成向量实体可以用于构建知识图谱元数据用于过滤和权限控制。如果用关系型数据库存元数据向量库只存 embedding 和向量对应的 memory_id效果会比较清晰。4. 环境准备与前置条件长期记忆系统的技术栈不复杂但依赖较多先列一套通用环境清单按实际项目做裁剪。4.1 操作系统与运行环境操作系统Linux 优先macOS 也可以Windows 建议使用 WSL2。Python 版本3.10推荐 3.11。虚拟环境工具venv 或 conda。容器环境Docker用于启动向量数据库和关系型数据库。4.2 外部依赖清单依赖项用途建议LLM API 或本地模型对话生成、事实抽取、记忆摘要OpenAI / 通义 / 本地 vLLM 等按企业环境选Embedding 模型将记忆文本转为向量text-embedding 系列或本地 embedding 模型向量数据库语义检索Milvus、Chroma、Weaviate、Qdrant 等关系型数据库存储元数据PostgreSQL、SQLiteRedis缓存和短期记忆可选Python 依赖库开发记忆服务fastapi、pydantic、requests、langchain 等如果没有部署向量数据库可以先用轻量的 SQLite 存储测试再用向量库替换检索模块。生产环境建议用独立数据库服务。4.3 Python 环境检查先确认基础环境。python --version pip --version创建虚拟环境。python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate安装基础依赖。pip install fastapi uvicorn pydantic requests openai如果计划使用向量数据库客户端再按所选库安装。比如使用 Chroma。pip install chromadb langchain需要说明依赖安装失败时优先检查 Python 版本和 pip 源必要时使用国内镜像源。5. 从 0 开始搭建项目骨架与代码实现下面给一个可以实际运行的项目骨架包含记忆服务、存储抽象和 API 层。代码设计上不绑定某个特定向量数据库方便替换为企业自建组件。5.1 项目目录结构agent-memory/ ├── app │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── models.py # 数据模型 │ ├── memory_store.py # 记忆存储抽象与实现 │ ├── memory_service.py # 记忆业务逻辑 │ └── vector_store.py # 向量库适配器 ├── data # 本地测试数据目录 ├── requirements.txt └── README.md5.2 数据模型models.py里定义记忆条目的数据结构。from pydantic import BaseModel, Field from typing import Optional from datetime import datetime class MemoryItem(BaseModel): memory_id: str user_id: str agent_id: str session_id: str content: str summary: str entities: list[str] [] memory_type: str general importance: float Field(default0.5, ge0.0, le1.0) created_at: datetime Field(default_factorydatetime.now) updated_at: datetime Field(default_factorydatetime.now) expires_at: Optional[datetime] None access_scope: str private5.3 记忆存储抽象from abc import ABC, abstractmethod from .models import MemoryItem class MemoryStore(ABC): abstractmethod def add(self, item: MemoryItem) - None: pass abstractmethod def get(self, memory_id: str) - MemoryItem | None: pass abstractmethod def update(self, item: MemoryItem) - None: pass abstractmethod def delete(self, memory_id: str) - None: pass abstractmethod def query_by_metadata(self, user_id: str, **filters) - list[MemoryItem]: pass5.4 基于 SQLite 的内存存储实现先用 SQLite 实现一套保证项目可以跑通。向量检索部分后续再接。import sqlite3 import json from datetime import datetime from .models import MemoryItem from .memory_store import MemoryStore class SQLiteMemoryStore(MemoryStore): def __init__(self, db_path: str ./data/memory.db): self.db_path db_path self._init_db() def _init_db(self) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( CREATE TABLE IF NOT EXISTS memories ( memory_id TEXT PRIMARY KEY, user_id TEXT, agent_id TEXT, session_id TEXT, content TEXT, summary TEXT, entities TEXT, memory_type TEXT, importance REAL, created_at TEXT, updated_at TEXT, expires_at TEXT, access_scope TEXT ) ) staticmethod def _serialize_item(item: MemoryItem) - tuple: return ( item.memory_id, item.user_id, item.agent_id, item.session_id, item.content, item.summary, json.dumps(item.entities, ensure_asciiFalse), item.memory_type, item.importance, item.created_at.isoformat(), item.updated_at.isoformat(), item.expires_at.isoformat() if item.expires_at else None, item.access_scope, ) staticmethod def _deserialize_row(row: tuple) - MemoryItem: return MemoryItem( memory_idrow[0], user_idrow[1], agent_idrow[2], session_idrow[3], contentrow[4], summaryrow[5], entitiesjson.loads(row[6]), memory_typerow[7], importancerow[8], created_atdatetime.fromisoformat(row[9]), updated_atdatetime.fromisoformat(row[10]), expires_atdatetime.fromisoformat(row[11]) if row[11] else None, access_scoperow[12], ) def add(self, item: MemoryItem) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( INSERT INTO memories ( memory_id, user_id, agent_id, session_id, content, summary, entities, memory_type, importance, created_at, updated_at, expires_at, access_scope ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , self._serialize_item(item), ) def get(self, memory_id: str) - MemoryItem | None: with sqlite3.connect(self.db_path) as conn: cursor conn.execute( SELECT * FROM memories WHERE memory_id ?, (memory_id,), ) row cursor.fetchone() return self._deserialize_row(row) if row else None def update(self, item: MemoryItem) - None: with sqlite3.connect(self.db_path) as conn: conn.execute( UPDATE memories SET content ?, summary ?, entities ?, memory_type ?, importance ?, updated_at ?, expires_at ?, access_scope ? WHERE memory_id ? , ( item.content, item.summary, json.dumps(item.entities, ensure_asciiFalse), item.memory_type, item.importance, item.updated_at.isoformat(), item.expires_at.isoformat() if item.expires_at else None, item.access_scope, item.memory_id, ), ) def delete(self, memory_id: str) - None: with sqlite3.connect(self.db_path) as conn: conn.execute(DELETE FROM memories WHERE memory_id ?, (memory_id,)) def query_by_metadata(self, user_id: str, **filters) - list[MemoryItem]: conditions [user_id ?] params: list [user_id] if memory_type in filters: conditions.append(memory_type ?) params.append(filters[memory_type]) if agent_id in filters: conditions.append(agent_id ?) params.append(filters[agent_id]) sql fSELECT * FROM memories WHERE { AND .join(conditions)} ORDER BY updated_at DESC with sqlite3.connect(self.db_path) as conn: cursor conn.execute(sql, params) rows cursor.fetchall() return [self._deserialize_row(row) for row in rows]5.5 记忆业务逻辑记忆服务里加入写入、更新、检索、判断记忆是否过期的业务逻辑。from typing import Optional from .models import MemoryItem from .memory_store import MemoryStore class MemoryService: def __init__(self, store: MemoryStore): self.store store def create_memory( self, user_id: str, agent_id: str, session_id: str, content: str, summary: str , entities: list[str] | None None, memory_type: str general, importance: float 0.5, access_scope: str private, expires_at: Optional[str] None, ) - MemoryItem: item MemoryItem( memory_idf{user_id}_{session_id}_{int(__import__(time).time())}, user_iduser_id, agent_idagent_id, session_idsession_id, contentcontent, summarysummary, entitiesentities or [], memory_typememory_type, importanceimportance, access_scopeaccess_scope, expires_atexpires_at, ) self.store.add(item) return item def update_memory(self, item: MemoryItem) - None: item.updated_at __import__(datetime).datetime.now() self.store.update(item) def list_memories(self, user_id: str, memory_type: str | None None) - list[MemoryItem]: items self.store.query_by_metadata(user_id, memory_typememory_type) return [item for item in items if not self._is_expired(item)] staticmethod def _is_expired(item: MemoryItem) - bool: if item.expires_at is None: return False now __import__(datetime).datetime.now(item.expires_at.tzinfo) return now item.expires_at这段代码的核心是把记忆当作结构化数据管理不只是“聊天记录”。当你把记忆抽象成可查询、可更新、可过期的实体后续接向量检索、批量任务都会顺手很多。6. 感知记忆与对话恢复核心功能实现存储层搭好后真正需要解决的是“Agent 怎么用上这些记忆”。6.1 对话历史到记忆的抽取对话并不能直接全部写入记忆库否则越存越多、越存越乱。一般做法是先让 LLM 从对话里抽取关键事实和用户意图再按规则写入。一次简化抽取的输出格式可以是这样的 JSON{ memory_candidates: [ { content: 客户华东区项目目前处于需求确认阶段预算约 80 万决策人是张总。, entities: [张总, 华东区项目], memory_type: project, importance: 0.9 }, { content: 用户偏好每周五下午三点接收项目周报。, entities: [周报], memory_type: preference, importance: 0.7 } ] }拿到这个结果后程序先做过滤再去重然后调用记忆服务写入。抽取这一步可以用大模型的 API 完成也可以先写基于规则的版本。6.2 记忆召回与 Prompt 注入每次新对话开始时先从记忆服务召回跟当前用户、当前主题相关的记忆。伪代码流程def build_prompt_with_memory(user_query, user_id): # 1. 检索相关记忆 memories memory_service.retrieve(user_id, user_query) # 2. 拼装记忆上下文 memory_block format_memories(memories) # 3. 拼入 system prompt system_prompt f 你是企业智能助手。请基于以下已知信息回答用户问题 {memory_block} 如果信息不足请明确说明不要编造。 return system_prompt这里记忆召回有两种方式元数据过滤按用户、Agent、记忆类型直接查。语义检索用 embedding 模型把用户查询向量化在向量库召回相近记忆再取 Top-K 条。企业级系统里建议混合使用先做权限过滤再做语义召回最后按重要度、时间排序。这样可以避免“查到了但不能看”和“看到了但无关”这两种情况。6.3 记忆更新记忆不是“只写一次”。当用户说“我改主意了项目总共 100 万不是 80 万”时系统需要找到旧的那条记忆判断冲突然后更新。这一步没有固定逻辑推荐方式是LLM 对比新信息和旧记忆生成新的记忆版本同时保留历史版本到审计表。对关键业务记忆建议记录变更历史便于回溯。7. 接口 API 与批量任务设计记忆系统最终要暴露给上层 Agent 调用一般用 REST API。7.1 API 设计方法路径功能POST/v1/memories写入记忆GET/v1/memories/{memory_id}获取指定记忆GET/v1/memories按用户、类型查询记忆PATCH/v1/memories/{memory_id}更新记忆DELETE/v1/memories/{memory_id}删除记忆POST/v1/memories/retrieve输入查询文本返回相关记忆 Top-KPOST/v1/memories/batch批量写入记忆POST/v1/memories/archive触发批量归档任务API 服务用 FastAPI 实现。下面给一个可运行的最小示例。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from .models import MemoryItem from .memory_service import MemoryService from .memory_store import SQLiteMemoryStore app FastAPI(titleAgent Memory Service) store SQLiteMemoryStore() service MemoryService(store) class CreateMemoryRequest(BaseModel): user_id: str agent_id: str session_id: str content: str summary: str entities: list[str] [] memory_type: str general importance: float 0.5 access_scope: str private expires_at: Optional[str] None app.post(/v1/memories, response_modelMemoryItem) def create_memory(req: CreateMemoryRequest): return service.create_memory( user_idreq.user_id, agent_idreq.agent_id, session_idreq.session_id, contentreq.content, summaryreq.summary, entitiesreq.entities, memory_typereq.memory_type, importancereq.importance, access_scopereq.access_scope, expires_atreq.expires_at, ) app.get(/v1/memories) def list_memories(user_id: str, memory_type: Optional[str] None): return service.list_memories(user_id, memory_typememory_type) app.post(/v1/memories/retrieve) def retrieve_memories(user_id: str, query: str, top_k: int 5): # 这里先做元数据查询向量检索接入后替换为混合检索 items service.list_memories(user_id) return sorted(items, keylambda x: x.importance, reverseTrue)[:top_k]启动服务uvicorn app.main:app --host 0.0.0.0 --port 8100启动后可以访问http://127.0.0.1:8100/docs查看 Swagger 文档。7.2 curl 调用示例写入一条记忆curl -X POST http://127.0.0.1:8100/v1/memories \ -H Content-Type: application/json \ -d { user_id: user_001, agent_id: agent_sales, session_id: session_001, content: 客户华东区项目预算为 100 万决策人是张总。, memory_type: project, importance: 0.9 }查询记忆curl http://127.0.0.1:8100/v1/memories?user_iduser_0017.3 Python 调用示例import requests base_url http://127.0.0.1:8100/v1/memories payload { user_id: user_001, agent_id: agent_sales, session_id: session_002, content: 用户偏好每周五下午三点接收项目周报。, memory_type: preference, importance: 0.7, } response requests.post(base_url, jsonpayload, timeout10) print(response.status_code) print(response.json())7.4 批量任务设计批量任务主要处理三类情况。批量导入历史对话把之前积累的客服记录、工单记录、聊天记录一次性导入记忆库。导入流程是读取源数据、清洗、拆分成片段、调用 LLM 抽取记忆、写入存储。由于数据量大建议异步执行分批写入。批量归档与压缩长期运行后记忆会膨胀。定时任务可以把三个月前的旧记忆做摘要压缩只保留摘要和高重要度条目删除低价值明细。定时遗忘策略按照expires_at字段和遗忘规则每天清理过期记忆或者把权限变更、离职员工等特殊数据做删除或脱敏。批量任务建议使用消息队列或定时调度框架。最简单的方案是 FastAPI 的 BackgroundTasks复杂场景用 Celery。from fastapi import BackgroundTasks def run_batch_import(file_path: str): # 从文件读取历史记录调用抽取逻辑批量写入记忆库 pass app.post(/v1/memories/batch) def batch_import_memories(file_path: str, background_tasks: BackgroundTasks): background_tasks.add_task(run_batch_import, file_path) return {status: started}批量任务必须考虑失败重试。推荐做法是每条记忆写入都记录状态写入失败的放入重试队列整体任务结束后生成失败报告。8. 资源占用与性能观察记忆系统的资源占用主要来自三部分。第一是存储空间。关系型数据库存元数据向量数据库存向量数据量取决于记忆条目数和 embedding 维度。记忆条目越多检索耗时和存储占用都会增长。建议观察索引大小和单条记忆的平均耗时提前规划分库分表或分区策略。第二是 embedding 生成耗时。每条记忆写入前需要转成向量查询时也需要把用户 query 转成向量。如果用的是远程 embedding API网络延迟会直接影响接口响应。本地 embedding 模型则占用 GPU 或 CPU 资源。性能观察重点看单次 embedding 耗时的 P95、P99。第三是 LLM抽取耗时。从对话里抽取关键事实时如果每个用户消息都同步调用大模型响应会非常慢。更稳妥的做法是对话结束后异步抽取。也就是说用户消息先进入短期记忆等用户离开本轮对话或会话结束再异步把值得沉淀的内容写入长期记忆。资源占用观察清单观察项观察方式关注指标API 响应耗时日志中间件P95、P99embedding 耗时端到端链路埋点单次耗时向量库存储空间向量库监控面板索引大小、文档数量数据库连接数连接池监控活跃连接数内存占用进程内存采样常驻内存批量任务耗时任务日志处理速率、失败率性能优化的通用路径用量小的场景先保持简单量上升后再引入缓存。短期记忆放 Redis长期记忆按用户维度做冷热分层热数据放内存或高性能存储冷数据走低频检索路径。检索链路可以做召回-重排两阶段向量召回 Top 100再用精排模型或规则重排到 Top 5效果和性能都能兼顾。9. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 仍然不记得用户信息长期记忆未在 prompt 中注入查看 prompt 是否包含记忆块检查检索接口是否返回记忆确认注入逻辑记忆检索结果不相关embedding 模型不匹配观察检索日志中 topK 分数更换 embedding 模型或改用混合检索同一事实有多条矛盾记忆写入时未做冲突检测查询记忆库中的重复条目增加实体级去重和更新逻辑记忆膨胀接口变慢缺少过期和压缩任务检查记忆条目的时间分布增加 TTL、归档和删除策略批量导入任务卡住单批数据量过大或 LLM 调用超时查看任务日志和重试次数分批处理、增加超时时间、失败重试权限相关数据泄漏检索时未过滤 access_scope检查查询条件增加范围过滤测试跨用户检索场景API 返回 500数据库表结构或字段错误查看服务日志堆栈检查数据库迁移和字段类型向量库和元数据库数据不一致双写时缺少事务机制比对 memory_id 数量增加补偿任务定期对账这里有一条通用排查原则记忆系统出问题时先验证数据是否写入成功再验证检索是否返回最后验证 prompt 是否正确注入。多数问题都出在其中一个环节。10. 最佳实践与使用建议第一先做最小闭环。不需要一开始就上完整向量数据库和分布式存储。先用 SQLite 存元数据加上一个简单的关键词检索跑通“写入-查询-注入 prompt”这条链路。发现确实需要语义检索再加向量库。第二把记忆分层不把所有内容写进同一张表。用户偏好、业务事实、任务状态、知识片段分门别类不同类型用不同检索策略。比如用户偏好高频使用、任务状态要实时更新、业务事实要可审计。第三给记忆加权限边界。企业场景里隐私和数据安全是底线。同一用户的记忆不应该被其他用户检索到团队知识要有团队级权限客户数据要遵守企业数据管理规范。建议所有检索入口都带上 access_scope 过滤。第四记忆内容要可追溯。每条记忆都记录来源会话和创建时间关键记忆要支持查看历史版本。这样当 Agent 回复出错时可以追溯到是哪条记忆影响了决策。第五定期评估记忆质量。抽一批测试问题观察 Agent 引用记忆中信息的准确率。如果频繁出错大概率是记忆内噪声太多或者更新策略不够严格需要调低“全量写入”的冲动。第六涉及人脸、声音、客户隐私、企业内部文档等数据时必须确认合法授权和数据合规边界。记忆系统不能成为非法收集用户信息的工具批量导入历史数据前要核实数据来源是否合法。11. 总结与下一步这篇文章的核心观点是AI Agent 的“记忆”不是模型参数里长出来的而是一个需要刻意设计、分层管理、持续维护的工程系统。先解决“无状态”问题再解决“检索不准”和“记忆膨胀”问题最后解决“多 Agent 共享记忆”和企业权限问题。最先应该验证的功能是给 Agent 加一个记忆写入接口和检索接口让它能在下一轮对话中看到上一轮沉淀的信息。只要这条链路通了后续的向量检索、批量导入、记忆压缩都只是不断加码。最容易踩的坑是把所有对话记录都当作长期记忆写入不做抽取、不去重、不过期。这样系统的检索质量会迅速劣化Agent 会从“失忆”变成“记忆混乱”。下一步可以继续扩展的方向包括多 Agent 共享记忆团队级知识库接入基于实体的知识图谱记忆以及让 Agent 能自主学习更新记忆的强化反馈机制。建议先搭一套最小记忆服务写入、读取、接口、定时清理跑一周真实对话数据再决定要不要引入向量检索和分布式存储。