你有没有遇到过这样的场景同一个 Agent 上一轮还在正确记住用户的偏好下一轮对话就像“失忆”一样把所有上下文忘得一干二净。要么只是把记忆塞进对话历史一旦超长就截断要么用一个全局字典存会话变量用户一多、业务一复杂就互相串场。问题通常不是“该不该用记忆”而是“记忆系统本身缺乏设计”。这篇文章就来拆解一套可落地的企业级 Agent 记忆系统三层结构负责分层解耦五层机制负责写、读、隔离、检索和持久化。全文以 Python FastAPI Redis 向量库为例手写一个支持 State 自定义、Commander 更新、Namespace 读写的最小可运行项目。适合已经跑通 LLM 调用的后端开发者也适合想从“临时变量”升级到“结构化记忆”的初学者。读完后你不仅知道怎么避免 Agent 失忆还能自己动手搭一个可以扩展的记忆服务。1. 背景与核心概念1.1 先理解 Agent 为什么会“失忆”很多人以为 Agent 的“记忆”就是对话历史字符串。实际上LLM 本身没有真正的记忆能力它每次请求都只看到你喂给它的上下文。所谓的“记住”不过是你把历史信息重新放回 Prompt 里。当业务复杂起来后这种原始方案会非常脆弱上下文长度有限。对话一长早期信息被截断Agent 自然“忘记”了。记忆粒度混乱。用户偏好、任务状态、业务数据全混在一段文本里检索时无法定位。没有隔离机制。多用户、多会话、多 Agent 实例共用一个全局存储互相污染。写路径不可控。任何代码都能直接改记忆状态越权覆盖、脏写、并发冲突层出不穷。所以企业级 Agent 记忆系统的目标不是“记住更多”而是“有结构、有权限、有作用域、可检索、可追溯”地管理记忆。1.2 三层结构接入层、记忆管理层、持久化层我在生产项目中习惯把记忆系统拆成三层层级职责对应模块接入层承接 Agent 运行时、会话路由、Prompt 组装FastAPI、Agent 框架、会话中间件记忆管理层负责状态定义、写命令、命名空间隔离、检索调度State、Commander、Namespace、Retriever持久化层真正落盘存储支撑恢复、扩展、审计Redis、PostgreSQL、向量数据库三层之间单向依赖接入层调记忆管理层记忆管理层调持久化层。这样做的好处是将来替换向量库、改成分布式存储Agent 主业务代码几乎不需要改动。1.3 五层机制State、Commander、Namespace、Retrieval、Store三层是物理结构五层是逻辑机制。五层机制全部落在记忆管理层内部State记忆的数据模型。通过 Schema 自定义字段、类型、默认值、版本。Commander记忆的唯一写入口。任何状态变更都必须构造命令对象统一校验、审计、幂等。Namespace读写作用域。通过命名空间隔离用户、会话、业务线避免串数据。Retrieval读取与召回。支持多路检索、打分排序、上下文压缩。Store持久化适配。屏蔽底层 Redis、PostgreSQL、向量库差异。这套设计的内核是“读写分离 路径收敛”读可以走多条路写只能走一条路。1.4 适用范围与设计取舍如果你的 Agent 只是单机脚本、短期对话用 dict 缓存即可没必要上整套系统。本文的记忆系统适合这些场景多用户 SaaS 产品Agent 需要长期记住每个用户的偏好。企业知识库问答需要对不同部门、不同权限做隔离。复杂任务 Agent需要把中间状态持久化支持中断恢复。需要审计追溯的金融、政务、医疗类应用写操作必须留痕。需要提前说明记忆系统和“长期规划”“工具调用”是两回事。记忆系统的职责是可靠存储和快速召回不负责 Agent 决策不要把记忆逻辑和业务逻辑紧密耦合。2. 环境准备与版本说明2.1 技术栈规划本文示例采用以下技术栈Python 3.10FastAPI 用于暴露记忆服务 HTTP 接口Pydantic v2 用于 State Schema 定义与校验Redis 用于短期工作记忆和命令队列向量库以 Chroma 为例用于语义检索PostgreSQL 或 SQLite 用于长期记忆的关系型存储SQLite 方便本地演示不同版本差异可能较大尤其是 Pydantic 和向量库客户端。本文代码以“展示工程思路”为主你需要根据实际版本微调导入路径和配置参数。2.2 项目目录结构建议按以下结构组织代码agent_memory/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置项 │ ├── schema.py # State 与命令 Schema │ ├── namespace.py # Namespace 管理 │ ├── commander.py # Commander 写路径 │ ├── retriever.py # 检索服务 │ ├── store.py # Store 持久化适配 │ └── agent_service.py # Agent 接入示例 ├── requirements.txt └── .env.example这个目录对应三层结构main.py和agent_service.py属于接入层schema.py、namespace.py、commander.py、retriever.py属于记忆管理层store.py属于持久化层适配。2.3 依赖安装创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn pydantic redis chromadb sqlalchemy如果你只需要先跑通逻辑可以暂时不安装 Chroma用内存列表模拟向量检索。代码里我会标注可替换位置。3. 核心机制拆解在写代码前先把三个核心机制讲清楚State 怎么自定义、Commander 怎么更新、Namespace 怎么读写。3.1 State 自定义记忆的数据模型为什么不能用 dict 一把梭很多 Agent 项目会把记忆写成memory {user_name: 张三, age: 18}这在小 Demo 里没问题但一旦业务变多就会失控字段拼写错误没人管、类型变化没人感知、版本升级无法迁移。State 层的核心是用 Schema 管住数据形状。举个例子# 文件路径: app/schema.py from pydantic import BaseModel, Field from typing import Optional from datetime import datetime class UserProfileState(BaseModel): user_id: str user_name: Optional[str] None age: Optional[int] None preferences: dict Field(default_factorydict) updated_at: datetime Field(default_factorydatetime.utcnow) version: int 1为什么这样设计字段类型明确防止把数字写成字符串。默认值固定缺省字段不会报错。version字段标记 Schema 版本将来升级时可以写迁移逻辑。updated_at精确记录修改时间方便审计。State 不只是一个类而是一组“状态族谱”。不同业务可以定义不同 State 子类再通过state_type字段区分。例如UserProfileState、TaskProgressState、ConversationSummaryState。3.2 Commander 更新为什么不能直接改 State 字段写一段业务代码直接改 State 很省事state.age 19 store.save(state)但直接改有三个隐患没有校验。任何字段都能被脏数据覆盖。没有审计。不知道谁改的、什么时候改的。没有幂等保护。并发请求可能导致旧数据覆盖新数据。Commander 模式把“修改”变成“命令对象”。所有变更先构造一个命令经过校验后再执行。命令本身也是 Pydantic 模型可序列化、可审计、可重放。# 文件路径: app/schema.py from typing import Literal, Any import uuid class MemoryCommand(BaseModel): command_id: str Field(default_factorylambda: str(uuid.uuid4())) namespace: str state_type: str action: Literal[upsert, patch, delete] payload: dict operator: str system timestamp: datetime Field(default_factorydatetime.utcnow)action字段限定为upsert、patch、delete避免任意操作。operator记录操作来源是用户还是 Agent 内部任务。command_id保证幂等重放时不会重复执行。Commander 的执行流程如下收到命令 → Schema 校验 → 幂等检查 → 加载旧状态 → 合并新状态 → 写入存储 → 记录审计日志这个流程必须在事务或加锁环境下执行。单机可以用 Redis 分布式锁分布式环境建议把命令写入消息队列由消费者串行执行。3.3 Namespace 读写如何避免多用户互相串数据Namespace 是这套系统的“隔离护栏”。每次读写都要带一个 namespace格式可以设计如下{产品线}:{用户维度}:{会话维度}:{业务子域} 示例: ai-assist:user_123:session_456:preference为什么需要 Namespace产品线隔离。客服 Agent 和营销 Agent 共用一个服务但记忆不能混。用户隔离。A 用户的偏好不能出现在 B 用户会话里。会话隔离。同一个用户的不同会话有独立工作区。安全边界。低权限业务不能读高权限数据。Namespace 不是简单拼接字符串它需要支持通配匹配和层级过滤。例如检索ai-assist:user_123:*表示读取该用户所有会话下的记忆检索ai-assist:user_123:session_456:*表示只读取当前会话的记忆。我习惯让 Store 层把 Namespace 作为 key 前缀写入 Redismem:ai-assist:user_123:session_456:preference - {age: 18}这样天然支持按前缀遍历也方便后期迁移。3.4 Retrieval 调度多路召回与排序读取记忆不能只按 key 精确查找。真实业务里Agent 需要回答“用户上个月提过什么需求”这就要语义检索。Retrieval 层通常做三件事多路召回。从 Redis 按 key 取近期记忆从向量库按语义相似度召回历史记忆从关系库按标签筛选。统一打分。把各路召回结果转成标准结构按时间衰减、相关度、权限范围排序。上下文压缩。只把最相关的数据放入 Prompt塞不下时优先保留目标和当前意图。一个简单检索请求的输入输出如下输入: query用户喜欢什么咖啡, namespaceai-assist:user_123:*, top_k5 输出: [{content: 用户偏好美式咖啡, score: 0.92, timestamp: ...}]3.5 Store 持久化与失效策略Store 层不直接暴露 Redis 命令而是提供get、save、delete、search四个统一接口。这样上层只依赖抽象接口不关心底层实现。记忆还要考虑生命周期记忆类型存储位置生命周期短期工作记忆Redis会话结束或 24 小时后过期长期叙事记忆关系库/向量库按业务策略归档通常保留数月以上语义偏好记忆向量库长期保留但支持用户主动删除Redis 可以设置 TTL 实现过期redis_client.setex(key, ttl_seconds, value)长期记忆则要在业务层设计清理任务或者提供“用户注销后删除记忆”的接口。3.6 核心设计误区我见过不少团队把记忆系统做复杂后反而更难维护常见误区有三个把记忆系统和业务逻辑强耦合。Agent 里到处调用记忆服务导致记忆结构稍变就要改业务代码。正确做法是业务代码只传命令和查询不直接操作 State。读写不分路径。有的代码读记忆时顺手写缓存破坏 audit。建议读路径全部收敛到 Retriever写路径全部收敛到 Commander。把上下文窗口当记忆。给 Prompt 塞越长越多的历史模型表现不升反降。记忆系统的目标是把“最相关”的信息交给模型不是全量堆历史。4. 完整实战案例这一节我们实现一个最小可运行的企业级 Agent 记忆系统。为了便于读者运行我用内存模拟向量检索实际生产可替换成 Chroma、Milvus 或 Qdrant。4.1 配置模块先写配置统一管理 Redis 地址、默认 TTL 等参数。# 文件路径: app/config.py import os from dataclasses import dataclass dataclass class Settings: redis_url: str os.getenv(REDIS_URL, redis://localhost:6379/0) default_ttl: int int(os.getenv(DEFAULT_TTL, 86400)) # 24小时 index_name: str os.getenv(VECTOR_INDEX, agent_memory) max_recall: int int(os.getenv(MAX_RECALL, 10)) settings Settings()4.2 Store 层Store 层统一封装 Redis 和向量检索。这里我写一个简化版思路清晰且方便替换。# 文件路径: app/store.py import json import redis from typing import Optional, Any from app.config import settings class MemoryStore: def __init__(self): self.redis redis.Redis.from_url(settings.redis_url, decode_responsesTrue) def save(self, key: str, value: dict, ttl: Optional[int] None): self.redis.setex(key, ttl or settings.default_ttl, json.dumps(value, ensure_asciiFalse)) def get(self, key: str) - Optional[dict]: raw self.redis.get(key) if not raw: return None return json.loads(raw) def delete(self, key: str): self.redis.delete(key) def list_by_namespace(self, namespace_prefix: str): keys self.redis.keys(fmem:{namespace_prefix}*) result [] for k in keys: result.append({key: k, data: self.get(k)}) return result def semantic_search(self, query: str, namespace_prefix: str, top_k: int): # 实际生产中使用向量数据库查询 # collection.query(query_texts[query], n_resultstop_k) # 这里用关键词匹配模拟方便本地运行。 items self.list_by_namespace(namespace_prefix) scored [ item for item in items if query in str(item[data]) ] return scored[:top_k]semantic_search在本地用关键词匹配模拟如果换成 Chroma只需要把query_texts改成真实调用即可。4.3 Namespace 管理器Namespace 管理器负责拼接、解析、校验命名空间。# 文件路径: app/namespace.py from typing import Optional class NamespaceManager: def build(self, product: str, user_id: str, session_id: str, sub_domain: str *) - str: return f{product}:{user_id}:{session_id}:{sub_domain} def build_prefix(self, product: str, user_id: str, session_id: Optional[str] None) - str: if session_id: return f{product}:{user_id}:{session_id}: return f{product}:{user_id}: def parse(self, namespace: str) - dict: parts namespace.split(:) return { product: parts[0] if len(parts) 0 else , user_id: parts[1] if len(parts) 1 else , session_id: parts[2] if len(parts) 2 else , sub_domain: parts[3] if len(parts) 3 else *, }为什么需要单独建一个类因为在大型项目里命名空间规则会频繁调整。比如从四段变成五段如果各处手写拼接字符串改动成本会非常高。4.4 Commander 更新器Commander 是写路径的守门员。所有记忆写入必须经过它。# 文件路径: app/commander.py from app.schema import MemoryCommand, UserProfileState from app.store import MemoryStore from app.namespace import NamespaceManager from datetime import datetime class Commander: def __init__(self, store: MemoryStore, ns_manager: NamespaceManager): self.store store self.ns ns_manager def _state_class(self, state_type: str): # 实际项目可以用注册表映射 state_type - State 类 mapping { user_profile: UserProfileState, } return mapping[state_type] def execute(self, command: MemoryCommand): # 1. 根据 state_type 获取 State 类校验 payload schema_cls self._state_class(command.state_type) validated schema_cls(**command.payload) # 2. 幂等检查按 command_id 判断是否已执行 dedup_key fcmd:{command.namespace}:{command.command_id} if self.store.get(dedup_key): return {status: duplicated} self.store.save(dedup_key, {done: True}, ttl3600) # 3. 构造存储 key state_key fmem:{command.namespace}:{command.state_type} # 4. 加载旧数据执行不同操作 if command.action upsert: payload_data validated.model_dump() self.store.save(state_key, payload_data) elif command.action patch: old self.store.get(state_key) or {} new validated.model_dump() old.update({k: v for k, v in new.items() if v is not None}) old[updated_at] datetime.utcnow().isoformat() self.store.save(state_key, old) elif command.action delete: self.store.delete(state_key) # 5. 审计日志略 return {status: ok, command_id: command.command_id, state_key: state_key}注意patch动作中None值不覆盖旧数据这一点很重要。否则调用方漏传某个字段会把原值清空。4.5 Retriever 检索服务读取路径全部收敛到 Retriever不直接访问 Store。# 文件路径: app/retriever.py from app.store import MemoryStore from app.namespace import NamespaceManager class MemoryRetriever: def __init__(self, store: MemoryStore, ns_manager: NamespaceManager): self.store store self.ns ns_manager def get_state(self, namespace: str, state_type: str): return self.store.get(fmem:{namespace}:{state_type}) def recall(self, query: str, namespace: str, top_k: int 5): # 精确查询 direct_result self.store.get(fmem:{namespace}:*) # 语义检索 semantic_results self.store.semantic_search(query, namespace, top_k) return { direct: direct_result, semantic: semantic_results, } def build_prompt_context(self, query: str, namespace: str, max_tokens_est: int 2000) - str: result self.recall(query, namespace) lines [] for item in result[semantic]: lines.append(f- {item[data]}) context \n.join(lines) return context[:max_tokens_est]build_prompt_context是给 Agent 用的核心方法。它会组装一段适合放进 Prompt 的记忆说明并且截断长度防止上下文爆炸。4.6 Agent 集成逻辑下面模拟一个 Agent 服务先读记忆再调用 LLM最后通过 Commander 写回新记忆。# 文件路径: app/agent_service.py from app.commander import Commander from app.retriever import MemoryRetriever from app.namespace import NamespaceManager from app.schema import MemoryCommand class AgentService: def __init__(self, commander: Commander, retriever: MemoryRetriever, ns_manager: NamespaceManager): self.cmd commander self.retriever retriever self.ns ns_manager def handle_message(self, user_id: str, session_id: str, message: str): namespace self.ns.build(ai-assist, user_id, session_id) # 1. 召回记忆 context self.retriever.build_prompt_context(message, namespace) # 2. 实际场景这里调用 LLM # response llm.chat(context message) response f[模拟LLM输出] 根据记忆: {context[:80]} 回答: {message} # 3. 解析新记忆写入命令 # 实际项目中应由模型输出结构化记忆比如 {age: 18, preferences: {coffee: 美式}} new_memory {user_name: 张三, age: 18, preferences: {coffee: 美式}} cmd MemoryCommand( namespacenamespace, state_typeuser_profile, actionpatch, payloadnew_memory, operatoruser_123, ) self.cmd.execute(cmd) return response这段代码特意用“模拟 LLM 输出”因为真实项目里每个模型返回的记忆结构不同。更成熟的方案是让模型返回 JSON用 Pydantic 解析后生成命令。4.7 FastAPI 接口与运行最后写一个 FastAPI 入口供外部调用。# 文件路径: app/main.py from fastapi import FastAPI from app.store import MemoryStore from app.namespace import NamespaceManager from app.commander import Commander from app.retriever import MemoryRetriever from app.agent_service import AgentService from app.schema import MemoryCommand app FastAPI(titleAgent Memory System) store MemoryStore() ns_manager NamespaceManager() commander Commander(store, ns_manager) retriever MemoryRetriever(store, ns_manager) agent AgentService(commander, retriever, ns_manager) app.post(/agent/chat) def chat(user_id: str, session_id: str, message: str): return {response: agent.handle_message(user_id, session_id, message)} app.post(/memory/command) def execute_command(command: MemoryCommand): return commander.execute(command) app.get(/memory/recall) def recall(query: str, namespace: str): return retriever.recall(query, namespace)启动服务uvicorn app.main:app --reload --port 80004.8 运行验证先用 curl 调用聊天接口curl -X POST http://localhost:8000/agent/chat?user_iduser_123session_idsession_456message你好预期返回{ response: [模拟LLM输出] 根据记忆: 回答: 你好 }再查看记忆是否写入curl -X POST http://localhost:8000/memory/command \ -H Content-Type: application/json \ -d { namespace: ai-assist:user_123:session_456, state_type: user_profile, action: upsert, payload: {user_name: 李四, age: 20}, operator: tester }最后召回记忆curl http://localhost:8000/memory/recall?query美式namespaceai-assist:user_123:session_456预期语义检索结果包含 coffee 相关的偏好。5. 常见问题与排查思路5.1 常见问题对照表问题现象常见原因解决思路Agent 回复时完全想不起来用户信息Namespace 拼错或传错检查 namespace 是否一致重点核对 user_id 和 session_id多个用户记忆串场Namespace 缺少用户维度统一用产品线:用户:会话:子域四段式命名写入报 Schema 校验失败payload 字段类型错误看 Pydantic 报错信息校准调用方传参记忆被旧数据覆盖并发写同一 State没有幂等Commander 增加 command_id 去重写入改为串行检索结果相关度差只用关键词匹配接入向量库并增加时间衰减因子排序Redis key 越来越多没有清理策略短期记忆设置 TTL长期记忆建立归档任务5.2 排查清单如果你线上遇到 Agent “失忆”问题按以下顺序排查确认 namespace 规则是否全局统一。确认记忆写入是否走 Commander还是有人绕过写路径。查看 Redis key确认数据到底有没有写进去。查看召回逻辑确认是数据丢失还是检索不到。查看 Prompt 组装逻辑确认上下文是否被截断或排序错误。查看权限校验确认当前 Agent 是否有权读取该 namespace。大部分“失忆”都不是模型问题而是写入路径或检索路径断了。5.3 关于向量库的注意点替换关键词检索为向量检索时需要先建索引再写入数据。不要在写入时动态创建 collection否则高并发下会反复锁冲突。正确做法是启动时检查索引是否存在不存在则创建。6. 最佳实践与工程建议6.1 State 设计建议一个业务域一个 State 类不要把所有字段塞进一个大类。字段使用强类型禁止无约束的 dict 泛滥。不要删除字段尽量新增并用deprecated标记保证线上兼容。State 版本升级时显式写迁移函数而不是直接改字段。6.2 Commander 生产化命令进入消息队列由 worker 串行消费避免单机锁依赖。命令表要记录command_id、operator、timestamp、payload_hash用于重放和审计。写操作失败时执行回滚或者用事件溯源的思路以命令日志重建状态。确保patch动作对空字段的处理是“不覆盖”而不是“清空”否则极易丢数据。6.3 Namespace 权限边界网关层做 namespace 白名单校验防止用户传入任意前缀访问他人记忆。管理端接口需要单独权限不能与普通 Agent 共用读写接口。Namespace 解析要预留扩展位比如从四段扩展到五段时老数据的兼容迁移要提前设计。6.4 检索与 Prompt 优化优先级排序建议当前会话意图 用户长期偏好 全局常识。给每条记忆加时间戳排序时按时间衰减避免几个月前的偏好淹没最近信息。Prompt 中记忆部分要显式标注“这是用户历史记忆可能存在过期信息”引导模型谨慎使用。6.5 安全与合规记忆包含用户隐私时需支持“导出我的数据”和“删除我的数据”接口。日志中不得记录完整记忆内容建议记录payload_hash代替。连接 Redis 和数据库使用最小权限账号生产环境不要用 root/admin。6.6 可观测性为每条命令增加 trace_id贯穿接入层、Commander、Store。采集三个核心指标写入成功率、检索召回率、记忆命中率。记录“Agent 当前轮是否使用了记忆”的标签方便判断失忆问题到底发生在写入还是召回。7. 总结与学习路线这篇教程落地了一套最小可用记忆系统State 负责自定义记忆 SchemaCommander 守住所有写路径并支持幂等Namespace 隔离用户和会话Retriever 统一召回Store 屏蔽底层存储差异。对照这份代码你可以快速给现有 Agent 加上“不串场、可追溯、可召回”的企业级记忆能力。下一步建议这样进阶把内存模拟检索替换成真实向量库理解 embedding 和索引参数。引入消息队列和事件溯源让命令日志成为系统事实来源。为多余 Agent 业务线设计独立命名空间并完善权限校验。读一下主流 Agent 框架的长短期记忆设计源码对比它们的 State 结构与 Command 流程差异。实操时优先关注两个风险点并发写覆盖和权限串读。这两类问题在用户量增长后一定会暴露越早处理越好。建议先把本文示例跑通再逐步替换真实组件。如果这篇文章对你有帮助收藏备用也欢迎在评论区聊聊你遇到的 Agent 记忆问题。