ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

AI Agent长期记忆系统设计:从失忆原理到企业级落地实践

AI Agent长期记忆系统设计:从失忆原理到企业级落地实践 在实际 Agent 项目中“AI Agent 总是失忆”不是一句玩笑话。对话刚结束新建一个会话它就不记得用户的偏好、项目背景和刚刚做过的决定任务执行到一半上下文一超限前面的关键信息就被截断。根本原因在于多数 Agent 本质上是无状态服务每次请求只依赖当前上下文窗口里的内容会话一结束、窗口一超限模型对历史就没有感知。要解决这个问题不能只靠“把对话历史全部塞进 Prompt”更合理的做法是搭建一套独立的长期记忆系统把需要长期保留的信息从对话中提取出来、结构化存储、按需检索再在生成回复前把记忆注入上下文。这篇文章会先从原理出发解释 Agent 为什么会失忆然后带你从零搭建一个最小可运行的记忆系统再讨论企业级落地时涉及的会话管理、检索链路、时效与遗忘、权限隔离和常见问题排查。适合正在做 Agent 应用、想解决多轮对话与跨会话记忆问题的开发者。1. 先理解 Agent 为什么“失忆”无状态调用与上下文窗口是根因1.1 把历史对话全部塞进上下文为什么行不通很多人的第一反应是把历史消息全部塞进请求里模型不就记住了吗这种方式在演示阶段有效一旦进入真实场景就会暴露出三个问题。第一是上下文窗口有限。当前模型的上下文窗口从几千 token 到几十万 token 不等但真实业务对话中一次会话可以产生远超窗口上限的文本。窗口超限后只能截断而截断往往丢掉的是最早的关键约定。第二是成本和时延失控。每轮请求携带的 token 越多推理耗时越长费用也越高。把三个月前的寒暄、试错过程、重复问题全部带进每一轮请求对结果没有帮助纯属浪费。第三是注意力被稀释。即使窗口放得下全部历史模型也难以在一大堆冗余内容中准确找到当前任务真正需要的事实。历史越长关键信息反而越容易被噪声覆盖。更关键的是这种方案根本无法处理跨会话记忆。用户关闭页面、重新打开一个新会话旧的请求上下文完全消失模型对上次会话一无所知。所以“塞历史”本质上是缓存不是记忆。1.2 把记忆分成三层短期、工作与长期业界在讨论 Agent 记忆时通常会把记忆拆成三层来设计而不是笼统地说“保存历史”。短期记忆对应当前会话内的消息列表生命周期就是一次会话通常放在 Redis 或内存里用滑动窗口控制长度。工作记忆对应当前任务的状态比如执行计划、中间变量、已经调用过的工具和返回结果任务结束即失效。长期记忆是跨会话保留的信息比如用户偏好、重要事实、历史决策、领域知识必须持久化存储并且按需检索。记忆类型典型内容存储位置生命周期访问方式短期记忆当前会话消息列表Redis、内存会话结束或 TTL 到期全量或窗口读取工作记忆任务计划、中间变量、工具调用结果内存、分布式状态存储任务结束按 key 读取或更新长期记忆用户偏好、关键事实、历史决策、领域知识数据库、向量库长期保存按策略遗忘按需检索召回三者的关系可以这样理解短期记忆负责“当前对话上下文的连续性”工作记忆负责“当前任务的执行连续性”长期记忆负责“跨会话的身份和知识连续性”。Agent 记忆系统的核心工作就是在这三层之间建立写入、读取和更新链路。1.3 企业级记忆系统的四个核心能力一个小玩具可以在代码里硬编码几条消息但企业级 Agent 记忆系统必须具备四个能力。提取能力从非结构化的对话文本中抽取结构化记忆包括用户画像、事件记录、业务结论并给每条记忆打重要度分。存储能力记忆要支持结构化字段、JSON 元数据和高维向量能够按用户、Agent、记忆类型快速过滤。检索能力根据当前问题召回最相关的记忆而不是把全部记忆都塞给模型检索要支持向量召回、关键词召回和重排序。管理能力包括记忆的合并、更新、过期、遗忘、权限隔离和审计日志。下文的技术选型和代码实现都会围绕这四个能力展开。2. 技术选型记忆系统由哪几层组件构成各层怎么选2.1 存储层向量库、关系库、缓存各司其职长期记忆存储最常见的选择是“关系库 向量扩展”典型组合是 PostgreSQL 配合 pgvector。这样做的好处是业务字段、JSON 元数据、向量和 SQL 过滤可以放在同一个数据库里减少组件数量事务也容易保证。中小规模项目用这一套就足够。如果向量规模很大比如千万级以上并且有很高的并发检索要求可以单独引入 Milvus、Qdrant 等专用向量数据库。短期记忆通常放在 Redis 里利用它的 TTL 和列表结构保存最近消息。组件负责内容适用场景PostgreSQL pgvector长期记忆主存储、元数据过滤、向量检索中小规模项目、团队已使用 PostgreSQLMilvus / Qdrant大规模向量检索千万级向量、高并发检索Redis短期记忆、会话消息、缓存会话级上下文、热点记忆缓存MySQL / 其他关系库只存结构化字段不做向量检索已有 MySQL 且数据量不大时这里要注意一点不要把短期记忆和长期记忆放在同一个地方管理。短期记忆是高频写入、短生命周期长期记忆是低频写入、长生命周期两者对存储和清理策略的要求完全不同。2.2 提取层让 LLM 把非结构化对话转换成结构化记忆对话文本是自然语言不能直接作为长期记忆存储。提取层要调用 LLM把原始对话转换成结构化记忆条目。一条典型的记忆条目应该包含记忆类型、内容、重要度、来源会话和时间。常见记忆类型可以设计为user_profile 用户偏好与画像episodic 重要事件和任务决策knowledge 领域知识和业务规则。提取逻辑通常放在对话的异步链路上。对话结束后或者每积累若干轮消息后把最近几轮消息交给 LLM 提取而不是每轮都提取。这样既能控制成本也能避免把中间过程的临时内容写入长期记忆。2.3 检索层按需召回而不是全量加载长期记忆只能按需召回。检索层要做三件事。第一向量召回。把当前用户问题转成向量与记忆表中的 embedding 做相似度检索。第二元数据过滤。检索 SQL 必须强制带上 agent_id、user_id、记忆类型、过期时间等条件保证只召回当前用户在当前 Agent 场景下的记忆。第三重排序。向量相似度不能作为唯一排序依据还要综合考虑记忆的重要度、访问次数和最近访问时间。重排公式可以简化成最终分数 向量相似度 × 权重 重要度 × 权重 时效因子 × 权重。具体权重需要在业务上调试但“相似度 重要度 时效”这三个维度必须同时存在否则会出现“检索到一堆相关但没价值的记忆”的情况。2.4 对话集成层记忆如何安全进入 Prompt检索回来的记忆不能直接拼在用户消息后面。更稳妥的做法是组装出独立的结构化记忆块放在系统提示词里并告诉模型“以下是该用户的长期记忆回答时优先参考但不要虚构记忆中没有的信息。”组装顺序一般是系统提示词 → 长期记忆块 → 短期记忆消息列表 → 当前用户输入。这里有两个关键控制点记忆块必须带来源说明比如“记忆类型user_profile来源2025-06-01 会话”记忆块必须受 token 预算限制超出的部分要截断或减少召回数量。3. 从零搭建最小可运行的 Agent 记忆系统3.1 环境准备依赖与基础设施版本本节的项目结构使用 Python 3.10、FastAPI、PostgreSQL 15 pgvector、Redis 7并使用兼容 OpenAI 接口的 LLM 和 Embedding 服务。生产环境落地前需要先确认项目实际依赖的版本和模型接口是否一致。# docker-compose.yml version: 3.8 services: postgres: image: pgvector/pgvector:pg16 environment: POSTGRES_USER: agent POSTGRES_PASSWORD: agent123 POSTGRES_DB: agent_memory ports: - 5432:5432 volumes: - pg_data:/var/lib/postgresql/data redis: image: redis:7-alpine ports: - 6379:6379 volumes: pg_data:依赖清单如下。# requirements.txt fastapi uvicorn psycopg2-binary pgvector redis openai pydantic-settings启动基础设施后检查两个端口是否可用5432 对应 PostgreSQL6379 对应 Redis。如果本地端口冲突需要修改 docker-compose 中的映射关系。3.2 项目目录与数据模型设计最小系统建议采用如下目录结构。agent-memory/ ├── docker-compose.yml ├── requirements.txt ├── schema.sql └── app/ ├── main.py # FastAPI 入口提供记忆读写接口 ├── memory_store.py # 长期记忆持久化与向量检索 ├── extractor.py # 调用 LLM 提取记忆、调用 Embedding 生成向量 └── memory_service.py # 业务编排召回、重排、写入长期记忆表的设计是这套系统的核心。下面是基础 DDL。-- schema.sql CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS agent_memory ( id BIGSERIAL PRIMARY KEY, agent_id VARCHAR(64) NOT NULL, user_id VARCHAR(64) NOT NULL, session_id VARCHAR(64), memory_type VARCHAR(32) NOT NULL, content TEXT NOT NULL, metadata JSONB NOT NULL DEFAULT {}::jsonb, importance REAL NOT NULL DEFAULT 0.5, embedding VECTOR(1536), access_count INT NOT NULL DEFAULT 0, last_accessed_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), expired_at TIMESTAMPTZ ); CREATE INDEX idx_memory_owner ON agent_memory (agent_id, user_id, memory_type); CREATE INDEX idx_memory_expire ON agent_memory (expired_at);embedding 字段的维度要与 Embedding 模型输出维度保持一致。使用 text-embedding-3-small 时是 1536 维如果换模型表定义要同步改。生产环境中数据量上来之后还需要创建专门的向量索引比如 ivfflat 或 hnsw开发阶段可以先用顺序扫描。3.3 核心代码记忆写入、提取与检索先实现持久化层负责写入和向量检索。# app/memory_store.py import json import os import psycopg2 from pgvector.psycopg2 import register_vector class MemoryStore: def __init__(self): self.conn psycopg2.connect(os.getenv(DATABASE_URL)) register_vector(self.conn) def save(self, memory: dict, embedding: list[float]): cur self.conn.cursor() cur.execute( INSERT INTO agent_memory (agent_id, user_id, session_id, memory_type, content, metadata, importance, embedding, expired_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) , ( memory[agent_id], memory[user_id], memory.get(session_id), memory[memory_type], memory[content], json.dumps(memory.get(metadata, {}), ensure_asciiFalse), memory.get(importance, 0.5), embedding, memory.get(expired_at), ), ) self.conn.commit() def search(self, agent_id, user_id, query_embedding, memory_typeNone, top_k5): cur self.conn.cursor() sql SELECT id, memory_type, content, metadata, 1 - (embedding %s) AS score FROM agent_memory WHERE agent_id %s AND user_id %s AND (expired_at IS NULL OR expired_at now()) params [query_embedding, agent_id, user_id] if memory_type: sql AND memory_type %s params.append(memory_type) sql ORDER BY score DESC LIMIT %s params.append(top_k) cur.execute(sql, params) rows cur.fetchall() return [ { id: r[0], memory_type: r[1], content: r[2], metadata: r[3], score: r[4], } for r in rows ]这里使用计算余弦距离1 - 距离就是相似度分数。SQL 中强制过滤了 agent_id 和 user_id这是多租户隔离的最基本要求。接下来是提取层负责调用 LLM 提取记忆并调用 Embedding 服务生成向量。# app/extractor.py import json import os from openai import OpenAI llm OpenAI( base_urlos.getenv(LLM_BASE_URL), api_keyos.getenv(LLM_API_KEY), ) embedder OpenAI( base_urlos.getenv(EMBEDDING_BASE_URL), api_keyos.getenv(EMBEDDING_API_KEY), ) EXTRACT_PROMPT 你是记忆提取器。从对话中抽取值得长期保留的信息包括 1. user_profile用户的偏好、身份、习惯 2. episodic重要任务事件和决策过程 3. knowledge领域结论和业务规则 只输出 JSON 数组每项包含 memory_type、content、importance。 importance 取值范围 0 到 11 表示非常重要。 如果没有值得保留的信息输出 []。 def extract_memories(conversation: list[dict]) - list[dict]: resp llm.chat.completions.create( modelos.getenv(LLM_MODEL), messages[ {role: system, content: EXTRACT_PROMPT}, {role: user, content: json.dumps(conversation, ensure_asciiFalse)}, ], temperature0, ) text resp.choices[0].message.content.strip() text text.removeprefix(json).removeprefix().removeprefix().removesuffix().strip() return json.loads(text) def embed_text(text: str) - list[float]: resp embedder.embeddings.create( modelos.getenv(EMBEDDING_MODEL), inputtext, ) return resp.data[0].embedding提取 Prompt 是记忆质量的关键。如果模型经常输出格式错误可以在提取层增加重试和格式校验逻辑而不是让它进入主链路报错。最后是业务编排层把提取、存储、召回和重排串起来。# app/memory_service.py from .extractor import embed_text, extract_memories from .memory_store import MemoryStore class MemoryService: def __init__(self): self.store MemoryStore() def recall(self, agent_id, user_id, query, top_k5): query_embedding embed_text(query) candidates self.store.search( agent_id, user_id, query_embedding, top_ktop_k * 3 ) # 重排相似度 重要度 时效 scored [] for c in candidates: meta c[metadata] or {} importance meta.get(importance, 0.5) rerank_score c[score] * 0.6 importance * 0.4 scored.append((rerank_score, c)) scored.sort(keylambda x: x[0], reverseTrue) return [item[1] for item in scored[:top_k]] def remember(self, agent_id, user_id, session_id, conversation): extracted extract_memories(conversation) saved 0 for mem in extracted: mem[agent_id] agent_id mem[user_id] user_id mem[session_id] session_id mem[metadata] mem.get(metadata, {}) mem[metadata][importance] mem.get(importance, 0.5) embedding embed_text(mem[content]) self.store.save(mem, embedding) saved 1 return saved这里需要注意的是重排时如果没有考虑访问时间和访问次数长期不用的重要记忆和刚写入的相关记忆会混在一起。生产环境可以在 SQL 中维护 last_accessed_at 和 access_count再把它纳入重排分数。3.4 与 Agent 对话循环集成一次完整调用链记忆系统只有接入对话循环才有意义。下面是一个简化的集成示例包含短期记忆读写和长期记忆召回。# app/agent.py import json import os from redis import Redis from extractor import embed_text from memory_service import MemoryService redis_client Redis.from_url(os.getenv(REDIS_URL)) memory_service MemoryService() def build_system_prompt(long_memories): if not long_memories: return 你是企业知识助手回答要准确、简洁。 memory_text \n.join( f- [{m[memory_type]}] {m[content]} for m in long_memories ) return ( 你是企业知识助手回答要准确、简洁。\n 以下是该用户的长期记忆回答时优先参考 但不要虚构记忆中没有的信息。\n memory_text ) def run_turn(agent_id, user_id, session_id, user_input): # 1. 读取短期记忆 session_key fsession:{session_id}:msgs short_memory [] for raw in redis_client.lrange(session_key, 0, -1): short_memory.append(json.loads(raw)) # 2. 召回长期记忆 long_memories memory_service.recall(agent_id, user_id, user_input) # 3. 组装 Prompt messages [{role: system, content: build_system_prompt(long_memories)}] messages.extend(short_memory) messages.append({role: user, content: user_input}) # 4. 调用 LLM reply llm.chat.completions.create( modelos.getenv(LLM_MODEL), messagesmessages, ).choices[0].message.content # 5. 把本轮消息写入短期记忆并触发异步记忆提取 redis_client.rpush( session_key, json.dumps({role: user, content: user_input}, ensure_asciiFalse), ) redis_client.rpush( session_key, json.dumps({role: assistant, content: reply}, ensure_asciiFalse), ) redis_client.ltrim(session_key, -20, -1) redis_client.expire(session_key, 86400) # 提取放到后台任务避免阻塞主链路 recent short_memory[-6:] [ {role: user, content: user_input}, {role: assistant, content: reply}, ] memory_service.remember(agent_id, user_id, session_id, recent) return reply这只是一个最小闭环。真实项目中提取操作应该放入消息队列或异步任务框架中执行并记录失败次数。如果提取链路依赖外部 LLM 服务还需要超时控制和失败降级不能让记忆提取失败反过来影响用户请求。4. 企业级落地从“能跑”升级到“能上线”的关键机制4.1 短期记忆会话管理Redis Key 设计与滑动窗口短期记忆的 Redis Key 设计必须带上租户、Agent、用户和会话维度。{tenant_id}:{agent_id}:{user_id}:session:{session_id}:msgs带 tenant_id 是为了多租户隔离。如果只用人维度不同企业租户的用户 ID 可能冲突会发生会话串号。Key 的 TTL 应根据业务会话周期设置比如企业客服场景可以设置 24 小时长周期项目场景可以设置 7 天。滑动窗口的实现方式是 ltrim 只保留最近 N 条N 的取值要根据模型上下文窗口和单条消息平均 token 数计算。更准确的做法是按 token 数裁剪而不是按消息条数裁剪因为一条消息可能包含很长的工具返回结果。4.2 长期记忆的多租户隔离与权限控制长期记忆是最敏感的数据隔离必须同时发生在存储层和应用层。存储层的隔离就是检索 SQL 中强制带 agent_id 和 user_id。应用层的隔离是统一的鉴权入口。不要信任客户端传入的 user_id必须从当前登录态中解析用户身份再由后端注入查询条件。企业内部还要区分普通用户和管理员用户可以查看和删除自己的记忆管理员可以查看统计和审计日志但不能直接修改业务记忆内容。如果 Agent 系统支持多企业租户建议在记忆表增加 tenant_id 字段并把所有索引都带上 tenant_id避免不同租户的数据互相干扰检索结果。4.3 遗忘机制与时效管理长期记忆如果没有遗忘机制数据库会越来越大检索结果会被过期信息污染。遗忘机制可以设计成三个层次。第一是硬删除。对明确失效的记忆比如用户主动要求删除的历史数据直接物理删除。第二是软过期。每条记忆可以设置 expired_at过期后不再参与检索由定时任务统一清理。第三是分数衰减。记忆的最终得分要引入时间衰减因子比如“最后访问时间越久得分越低”这样不常用的旧记忆会自然沉底。最终分数 向量相似度 * 0.5 重要度 * 0.3 时效因子 * 0.2时效因子的具体函数可以根据业务调整比如用最近访问天数进行指数衰减。衰减的目的不是删除记忆而是把用户当下最可能需要的内容排到前面。4.4 一致性、审计与生产保障记忆写入失败不能影响主对话链路这是生产环境的红线。常用做法是写入操作走独立异步任务失败进入重试队列超过重试次数后写入失败日志由值班人员处理。审计方面建议增加一张记忆审计表记录每次写入的来源会话、提取前的原始对话摘要、写入时间和操作人。用户对记忆内容提出异议时可以快速定位是哪轮对话导致了错误记忆。生产环境还需要关注三个点数据库备份与恢复演练、记忆服务的接口限流与超时、以及记忆检索的监控指标包括召回耗时、写入成功率、单次召回 token 数。没有监控的记忆系统出问题时只能靠用户反馈来发现。5. 运行验证怎样确认 Agent 真的“记住了”5.1 设计跨会话验证场景验证记忆系统不能只在同一会话里连续聊天。最有效的验证方式是设计一组跨会话场景。验证场景操作步骤预期结果记住用户偏好会话 1告诉 Agent 你是数据团队输出要带 SQL 方言会话 2提出分析问题回答自动带 SQL且风格匹配记住历史决策会话 1讨论后确定主库用 PostgreSQL会话 2询问上次技术选型结果回答包含 PostgreSQL 及原因提取业务结论会话 1约定每周五发布版本会话 2询问发布节奏回答提到周五记忆不串号用户 A 写一条私密偏好用户 B 询问相同问题不应包含 A 的记忆如果场景 1 能通过说明“写入 提取 检索 Prompt 注入”整条链路是通的。5.2 验证检索召回质量检索层需要单独验证不能只看最终对话效果。可以写一个调试脚本直接调用 recall 方法查询。python -c from memory_service import MemoryService svc MemoryService() for m in svc.recall(agent_demo, user_001, 上次我们决定用什么数据库): print(m[memory_type], round(m[score], 3), m[content][:40]) 预期输出应该类似knowledge 0.862 项目最终选择 PostgreSQL 作为主库原因是团队熟悉 episodic 0.731 2025-06-01 评审会上否决了 MongoDB 方案如果召回结果与问题无关优先检查 Embedding 模型是否一致、记忆内容是否足够语义完整。注意单条记忆内容不能太短比如“好的”“同意”这种碎片不适合作为长期记忆。5.3 验证 Prompt 组装结果检索正确不代表模型会用上。需要把最终发给模型的消息打印出来检查。可以在 run_turn 中临时增加日志输出确认系统提示词里确实包含记忆块且记忆块顺序在用户消息之前。System: 你是企业知识助手。以下是该用户的长期记忆... - [knowledge] 项目最终选择 PostgreSQL 作为主库 - [user_profile] 用户来自数据团队输出需要带 SQL 方言 User: 请帮我写一个查询最近订单量的 SQL如果记忆块出现在正确位置但回答仍然没有参考它问题往往出在提示词措辞上。可以改为更明确的指令“你必须在回答中优先使用上述记忆中的事实记忆冲突时说明不同来源。”5.4 性能与资源验证记忆系统上线前至少要关注四组指标。指标开发环境参考生产环境建议单次记忆召回耗时100ms 以内控制在 200ms 以内记忆写入成功率调试阶段可失败重试99.9% 以上失败走告警单次召回 token 占用不超过上下文 20%设置硬上限超限截断重复记忆比例人工抽查写入链路做相似度去重写入链路要去重否则同一事实反复被提取数据库中会出现大量重复条目。去重的简单做法是写入前先按用户和内容做一次精确匹配更合理的做法是用向量相似度做近似去重。6. 常见问题排查记忆丢失、召回为空、答非所问6.1 先看这条排查链路记忆系统出问题时不要直接改 Prompt。要按照“写入 → 存储 → 检索 → 注入 → 生成”这条链路逐层排查。现象 └─ 1. 日志里是否触发记忆提取 └─ 2. 数据库里是否落库成功 └─ 3. 检索 SQL 是否能召回 └─ 4. 组装后的 Prompt 是否包含记忆 └─ 5. 模型回答是否参考记忆任何一层断了最终表现都是“Agent 没记住”或“记住但没用上”。6.2 记忆没有写入现象是数据库里查不到新的记忆记录。优先检查三处是否到达了触发条件比如对话轮数不足、后台任务没有执行提取 LLM 返回的内容是否是合法 JSON格式错误会导致解析异常写入事务是否因为字段错误回滚比如 embedding 维度不匹配。排查时看应用日志里提取链路的输出。建议在提取服务中记录原始返回文本格式错误时能够直接看到。不要把异常吞掉否则问题非常难追。6.3 写入成功但检索不到数据库有记录但召回结果为空。常见原因有三个。第一检索时使用的问题向量与写入时的内容向量由不同模型生成导致向量空间不一致相似度普遍很低。第二SQL 中 agent_id 或 user_id 过滤条件不匹配比如写入时 user_id 带了前缀查询时没有带。第三expired_at 已经过期或者顺序扫描在大表上超时被截断。检查方法是把查询 SQL 中的过滤条件逐个手工执行一遍。先在库里确认记录存在再确认 embedding 字段有值再执行相似度查询逐步缩小范围。6.4 检索到了但注入 Prompt 无用召回正常但回答没有利用记忆。先看组装后完整 Prompt。如果记忆块被放在很靠后的位置可能已经被上下文截断如果记忆块内容过短模型无法判断它的用途如果系统提示词没有明确说明记忆块的使用方式模型倾向于忽略。解决方式把记忆块放到系统提示词中并增加使用指令在生成阶段对记忆块单独设置 token 上限避免被长任务内容挤掉。6.5 记忆冲突、过期与上下文污染同一事实产生多个矛盾版本是长期记忆系统必然遇到的现象。比如用户在会话 1 说“周报用 Excel”在会话 5 又说“周报改用在线表格”。如果不处理检索会把两个版本同时捞出来模型就会给出矛盾答案。处理策略有两种。一种是覆盖策略对同一用户和同一主题新记忆写入时标记旧记忆为失效。另一种是版本策略保留多个版本并在记忆块中标注时间和来源让模型自行判断哪个更新。生产环境建议先做覆盖策略配合来源时间和会话 ID 的审计字段必要时再升级到版本管理。问题现象常见原因检查方式处理建议新会话完全不记得未触发写入或写入失败查提取日志和 agent_memory 表补全触发条件增加失败重试写入成功但召回为空Embedding 模型不一致对比写入和查询时的模型名统一模型重建失效向量召回结果相关但回答没用记忆块被截断或指令不清打印组装后的完整 Prompt调整 Prompt 结构设置记忆 token 上限回答出现矛盾信息记忆冲突未处理按用户查同主题多条记录实现覆盖策略或版本标注不同用户记忆串号过滤条件缺失检查检索 SQL 的 user_id强制租户条件禁止客户端传身份7. 最佳实践与可复用清单7.1 记忆系统设计原则第一写入异步、读取同步。记忆提取和 Embedding 生成依赖外部 LLM延迟高且可能失败不能放在用户请求的主链路上。用户请求只依赖检索读取写入全部异步化。第二检索必须走“窄召回 精排”。先按租户和类型过滤再向量召回再重排最后限制返回条数。不要在大表上做全量相似度扫描。第三记忆块必须有来源标注。没有来源的记忆在审计和纠错时无法追溯。不要在高频请求里重复生成同一段文本的向量。对 Embedding 结果做本地缓存可以显著降低成本但缓存 Key 必须包含用户维度和模型版本否则模型升级后新旧向量空间不一致召回质量会突然下降。7.2 学习环境与生产环境的差异对照本地跑通最小系统只是起点生产环境还需要补齐很多环节。维度学习环境生产环境数据库单节点 pgvector主从复制、定期备份、慢查询监控向量索引默认无索引按数据量建立 ivfflat 或 hnsw记忆提取同步调用消息队列异步执行、失败重试鉴权Testing 环境直接传 user_id从登录态解析身份、服务间鉴权监控无调用链日志、召回指标、写入成功率告警清理手动清表定时任务做过期清理和分数衰减如果项目早期就并行输出两套方案改造成本会小很多。把表结构调整为带 tenant_id 的通用模型把提取操作放进异步任务这是从开发走向上线前最值得先做的两件事。7.3 上线前检查清单记忆写入检查 [ ] 提取任务是否在用户请求主链路之外 [ ] 提取失败是否会重试并记录日志 [ ] 写入是否做了重复记忆去重 [ ] 每条记忆是否包含来源会话和时间字段 记忆检索检查 [ ] 检索 SQL 是否强制过滤 tenant_id / agent_id / user_id [ ] 写入和检索是否使用同一 Embedding 模型 [ ] 是否设置了召回条数和 token 上限 [ ] 重排是否考虑重要度和时效 数据与安全检查 [ ] 是否支持用户查询和删除自己的记忆 [ ] 是否有审计日志 [ ] 数据库是否有备份和恢复演练 [ ] 是否有过期记忆清理任务 验证检查 [ ] 是否跑通过至少 4 组跨会话验证场景 [ ] 是否打印并检查过组装后的完整 Prompt [ ] 是否验证过多租户隔离不串号7.4 扩展方向知识库、工作流编排与多语言生态长期记忆系统稳定之后可以沿着三个方向继续扩展。第一个方向是接入外部知识库。很多团队会把企业文档、个人笔记导入知识库再通过 RAG 提供给 Agent 查询。这与长期记忆系统天然互补记忆保存的是个性化、动态变化的信息知识库保存的是相对稳定的领域知识。像 Obsidian 这类本地知识库与 Agent 结合时就可以把笔记内容同步到同一个向量库实现“个人知识 对话记忆”的统一检索。第二个方向是与工作流编排平台集成。企业级部署往往不是只有一个 Agent而是多个 Agent 配合工作流执行。n8n 这类支持企业级部署的编排工具可以把“会话消息 → 记忆提取 → 记忆存储 → 下游任务”串成可视化流水线方便非后端团队维护触发条件。第三个方向是多语言生态。当前示例是 Python 技术栈但很多企业在 Java 体系中。Java 技术栈可以参考 Spring AI 的客户端能力把长期记忆检索封装成一个组件供不同的 Agent 客户端复用。无论是哪种语言记忆系统本身的接口设计思想是一致的写入异步、读取同步、租户隔离、检索可控。如果正在准备与 AI Agent 相关的技术面试记忆系统是一个高频考察点。面试官通常会问上下文窗口有限时如何控制成本与质量、跨会话记忆如何设计、向量检索与关键词检索如何结合、记忆与知识库有什么区别。能徒手画出记忆分层架构并能解释写入、检索、遗忘和隔离这四个链路通常就能把这一类问题讲清楚。真正值得优先投入精力的不是无限扩大上下文窗口而是建立一套“知道该记住什么、该忘掉什么、该在什么时候回忆起什么”的管理机制。先把最小闭环跑通再补齐异步化、多租户隔离、遗忘机制和监控告警这套系统就能从“能跑”走向“能上线”。
返回列表