你有没有遇到过这样的场景：同一个 Agent 上一轮还在正确记住用户的偏好，下一轮对话就像“失忆”一样，把所有上下文忘得一干二净。要么只是把记忆塞进对话历史，一旦超长就截断；要么用一个全局字典存会话变量，用户一多、业务一复杂就互相串场。问题通常不是“该不该用记忆”，而是“记忆系统本身缺乏设计”。

这篇文章就来拆解一套可落地的企业级 Agent 记忆系统：三层结构负责分层解耦，五层机制负责写、读、隔离、检索和持久化。全文以 Python + FastAPI + Redis + 向量库为例，手写一个支持 State 自定义、Commander 更新、Namespace 读写的最小可运行项目。适合已经跑通 LLM 调用的后端开发者，也适合想从“临时变量”升级到“结构化记忆”的初学者。读完后，你不仅知道怎么避免 Agent 失忆，还能自己动手搭一个可以扩展的记忆服务。

1. 背景与核心概念

1.1 先理解 Agent 为什么会“失忆”

很多人以为 Agent 的“记忆”就是对话历史字符串。实际上，LLM 本身没有真正的记忆能力，它每次请求都只看到你喂给它的上下文。所谓的“记住”，不过是你把历史信息重新放回 Prompt 里。

当业务复杂起来后，这种原始方案会非常脆弱：

• 上下文长度有限。对话一长，早期信息被截断，Agent 自然“忘记”了。
• 记忆粒度混乱。用户偏好、任务状态、业务数据全混在一段文本里，检索时无法定位。
• 没有隔离机制。多用户、多会话、多 Agent 实例共用一个全局存储，互相污染。
• 写路径不可控。任何代码都能直接改记忆状态，越权覆盖、脏写、并发冲突层出不穷。

所以企业级 Agent 记忆系统的目标不是“记住更多”，而是“有结构、有权限、有作用域、可检索、可追溯”地管理记忆。

1.2 三层结构：接入层、记忆管理层、持久化层

我在生产项目中习惯把记忆系统拆成三层：

三层之间单向依赖：接入层调记忆管理层，记忆管理层调持久化层。这样做的好处是，将来替换向量库、改成分布式存储，Agent 主业务代码几乎不需要改动。

1.3 五层机制：State、Commander、Namespace、Retrieval、Store

三层是物理结构，五层是逻辑机制。五层机制全部落在记忆管理层内部：

1. State：记忆的数据模型。通过 Schema 自定义字段、类型、默认值、版本。
2. Commander：记忆的唯一写入口。任何状态变更都必须构造命令对象，统一校验、审计、幂等。
3. Namespace：读写作用域。通过命名空间隔离用户、会话、业务线，避免串数据。
4. Retrieval：读取与召回。支持多路检索、打分排序、上下文压缩。
5. Store：持久化适配。屏蔽底层 Redis、PostgreSQL、向量库差异。

这套设计的内核是“读写分离 + 路径收敛”：读可以走多条路，写只能走一条路。

1.4 适用范围与设计取舍

如果你的 Agent 只是单机脚本、短期对话，用 dict 缓存即可，没必要上整套系统。本文的记忆系统适合这些场景：

• 多用户 SaaS 产品，Agent 需要长期记住每个用户的偏好。
• 企业知识库问答，需要对不同部门、不同权限做隔离。
• 复杂任务 Agent，需要把中间状态持久化，支持中断恢复。
• 需要审计追溯的金融、政务、医疗类应用，写操作必须留痕。

需要提前说明：记忆系统和“长期规划”“工具调用”是两回事。记忆系统的职责是可靠存储和快速召回，不负责 Agent 决策，不要把记忆逻辑和业务逻辑紧密耦合。

2. 环境准备与版本说明

2.1 技术栈规划

本文示例采用以下技术栈：

• Python 3.10+
• FastAPI 用于暴露记忆服务 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_factory=dict) updated_at: datetime = Field(default_factory=datetime.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)

但直接改有三个隐患：

1. 没有校验。任何字段都能被脏数据覆盖。
2. 没有审计。不知道谁改的、什么时候改的。
3. 没有幂等保护。并发请求可能导致旧数据覆盖新数据。

Commander 模式把“修改”变成“命令对象”。所有变更先构造一个命令，经过校验后再执行。命令本身也是 Pydantic 模型，可序列化、可审计、可重放。

# 文件路径: app/schema.py from typing import Literal, Any import uuid class MemoryCommand(BaseModel): command_id: str = Field(default_factory=lambda: str(uuid.uuid4())) namespace: str state_type: str action: Literal["upsert", "patch", "delete"] payload: dict operator: str = "system" timestamp: datetime = Field(default_factory=datetime.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 前缀写入 Redis：

mem:ai-assist:user_123:session_456:preference -> {"age": 18}

这样天然支持按前缀遍历，也方便后期迁移。

3.4 Retrieval 调度：多路召回与排序

读取记忆不能只按 key 精确查找。真实业务里，Agent 需要回答“用户上个月提过什么需求”，这就要语义检索。

Retrieval 层通常做三件事：

1. 多路召回。从 Redis 按 key 取近期记忆，从向量库按语义相似度召回历史记忆，从关系库按标签筛选。
2. 统一打分。把各路召回结果转成标准结构，按时间衰减、相关度、权限范围排序。
3. 上下文压缩。只把最相关的数据放入 Prompt，塞不下时优先保留目标和当前意图。

一个简单检索请求的输入输出如下：

输入: query="用户喜欢什么咖啡", namespace="ai-assist:user_123:*", top_k=5 输出: [{"content": "用户偏好美式咖啡", "score": 0.92, "timestamp": "..."}]

3.5 Store 持久化与失效策略

Store 层不直接暴露 Redis 命令，而是提供get、save、delete、search四个统一接口。这样上层只依赖抽象接口，不关心底层实现。

记忆还要考虑生命周期：

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_responses=True) def save(self, key: str, value: dict, ttl: Optional[int] = None): self.redis.setex(key, ttl or settings.default_ttl, json.dumps(value, ensure_ascii=False)) 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(f"mem:{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_results=top_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 = f"cmd:{command.namespace}:{command.command_id}" if self.store.get(dedup_key): return {"status": "duplicated"} self.store.save(dedup_key, {"done": True}, ttl=3600) # 3. 构造存储 key state_key = f"mem:{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(f"mem:{namespace}:{state_type}") def recall(self, query: str, namespace: str, top_k: int = 5): # 精确查询 direct_result = self.store.get(f"mem:{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( namespace=namespace, state_type="user_profile", action="patch", payload=new_memory, operator="user_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(title="Agent 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 8000

4.8 运行验证

先用 curl 调用聊天接口：

curl -X POST "http://localhost:8000/agent/chat?user_id=user_123&session_id=session_456&message=你好"

预期返回：

{ "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=美式&namespace=ai-assist:user_123:session_456"

预期语义检索结果包含 coffee 相关的偏好。

5. 常见问题与排查思路

5.1 常见问题对照表

5.2 排查清单

如果你线上遇到 Agent “失忆”问题，按以下顺序排查：

1. 确认 namespace 规则是否全局统一。
2. 确认记忆写入是否走 Commander，还是有人绕过写路径。
3. 查看 Redis key，确认数据到底有没有写进去。
4. 查看召回逻辑，确认是数据丢失还是检索不到。
5. 查看 Prompt 组装逻辑，确认上下文是否被截断或排序错误。
6. 查看权限校验，确认当前 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 负责自定义记忆 Schema，Commander 守住所有写路径并支持幂等，Namespace 隔离用户和会话，Retriever 统一召回，Store 屏蔽底层存储差异。对照这份代码，你可以快速给现有 Agent 加上“不串场、可追溯、可召回”的企业级记忆能力。

下一步建议这样进阶：

• 把内存模拟检索替换成真实向量库，理解 embedding 和索引参数。
• 引入消息队列和事件溯源，让命令日志成为系统事实来源。
• 为多余 Agent 业务线设计独立命名空间，并完善权限校验。
• 读一下主流 Agent 框架的长短期记忆设计源码，对比它们的 State 结构与 Command 流程差异。

实操时优先关注两个风险点：并发写覆盖和权限串读。这两类问题在用户量增长后一定会暴露，越早处理越好。建议先把本文示例跑通，再逐步替换真实组件。如果这篇文章对你有帮助，收藏备用，也欢迎在评论区聊聊你遇到的 Agent 记忆问题。