做企业级 Agent 应用最难的不是把模型接入业务而是让 Agent 在跨会话、跨业务线、长时间运行后还记得上下文。很多团队把几百页文档、几千轮对话全部塞进 PromptContext 越拼越长效果越来越差延迟和成本反而一起涨。这次我们把“企业级 Agent Memory 架构”完整拆开从 Context 窗口聊到真正的 Long-term Memory讲清楚分层记忆模型怎么做短期记忆、长期记忆各用什么存储再落到 RAG 检索增强和 MCP 工具协议两个方向最后给出一套带代码的实现模板。文章不是 PPT 式的概念罗列而是可以直接照着搭的工程方案Redis 做会话级短期记忆向量库做跨会话长期记忆RAG 做记忆召回MCP 把记忆能力暴露给 Agent 工具调用。关注本地部署、接口服务、批量任务、知识库回填和多 Agent 协作的读者这篇文章可以直接收藏。1. 企业级 Agent Memory 架构核心能力速览先把方案的全貌放出来后面所有代码和步骤都是围绕这张表展开的。能力项说明架构方案分层记忆模型短期记忆 工作记忆 长期记忆短期记忆Redis 会话缓存按 session_id 隔离支持 TTL 过期长期记忆向量数据库持久化存储对话摘要、用户偏好、事实性知识记忆召回embedding 向量检索 关键词多路召回 rerank服务接入MCP 协议工具 REST API 双通道批量任务批量会话摘要、记忆归档、向量重建、记忆压缩上下文工程Prompt 动态注入按“当前会话 相关记忆 知识库”三段拼接推理硬件本地 embedding 模型可用 CPU 运行接入云端模型 API 则无需 GPU适合场景智能客服、企业 Copilot、流程助手、多轮数据分析、知识库问答这套架构解决的关键问题只有一个让 Agent 从“无状态调用”变成“有状态服务”。2. 为什么 Agent 需要长效记忆从 Context 到 Memory先看一个真实的工程困境。企业里的 Agent 通常不是一次性问答而是以周、月为周期的持续服务。用户可能三天后回来继续上次的需求或者在钉钉/飞书群里让 Agent 处理跨部门的流程审批。如果每一次对话都从零开始用户每次都要重复需求背景、身份信息、项目上下文这是不可接受的产品体验。Context 窗口是有限资源。以主流大模型为例上下文窗口从 4K 到 200K 甚至更大看起来很大但企业真实场景里一份流程文档可能 20K token一套业务规则 30K token再加上历史对话、用户画像、工具返回结果很快超出窗口边界。更麻烦的是上下文过长会导致模型“迷失在中间”召回准确率下降响应延迟和成本同步上升。所以工程上必须把记忆从 Prompt 里解放出来放到外部存储中。这就是 Agent Memory 架构的出发点短期记忆当前会话内的对话轮次、临时状态存放在 Redis 这类高速缓存里TTL 过期后自动清理。工作记忆当前任务正在处理的中间结果、工具调用记录、待确认字段属于“正在思考”的数据。长期记忆跨会话持久化的用户偏好、关键事实、历史决策、项目上下文存放在向量库或结构化存储中按需召回。真正的 Long-term Memory 不是把历史对话原样存下来而是对记忆做提取、压缩、索引、召回、遗忘的全生命周期管理。这决定了 Agent 是“听起来聪明”还是“真的了解用户”。从上下文工程的视角看Long-term Memory 的本质是让 Agent 在有限的上下文预算内拿到“当前时刻最该知道的信息”。它和 RAG 有天然的结合点把长期记忆当作一类特殊的文档库用语义检索的方式按需召回。区别在于RAG 的知识来源通常是企业文档、网页、数据库记录而记忆库的来源是用户与 Agent 的交互历史。3. 企业级 Agent Memory 架构设计3.1 分层记忆模型整个架构按记忆生命周期分成四层每一层对应不同的存储介质和访问频率。记忆层存储介质生命周期访问频率示例短期记忆Redis分钟到小时极高当前会话最近 20 轮对话工作记忆Redis / 内存任务进行中高工具调用结果、待确认参数长期记忆向量库 关系库天到月中用户偏好、历史决策、项目上下文业务知识库向量库版本化管理低企业文档、规则、FAQ短期记忆和工作记忆可以共用 Redis 实例但建议用不同的 key 前缀区分避免数据互相污染。3.2 记忆生命周期记忆不是“写进去就不管了”它和业务系统一样需要生命周期管理写入每次对话结束后把关键信息写入短期记忆。提取定期或按事件触发从短期记忆中提取用户偏好、事实、决策记录。摘要对长对话进行摘要压缩避免长期记忆库无限膨胀。归档超过一定时效的记忆降级为冷数据减少检索噪音。召回新对话开始时从长期记忆中检索与当前 Query 最相关的记忆片段。遗忘用户要求删除或数据过期时彻底清除相关记忆。这个流程是长效记忆系统最核心的部分也是长期记忆和“历史记录文件”的区别所在。3.3 关键组件职责Memory Service统一管理记忆读写、提取、摘要、删除对外提供 REST API 和 MCP 工具。Redis Store短期记忆和工作记忆的存储使用 Hash 或 String 结构存储 JSON 序列化的消息。Vector Store长期记忆的索引存储保存 embedding 向量和原始内容。Extractor从对话中抽取记忆条目可以用规则匹配、小模型或大模型函数调用实现。RAG Pipeline负责将查询语句 embedding 化、检索、rerank、拼装上下文。MCP Server把记忆能力封装为标准工具Agent 可通过 MCP 协议直接调用。4. 技术选型与本地部署环境准备4.1 技术栈清单下面这套技术栈是通用选择具体组件可按团队已有基础替换组件选型用途缓存Redis 7.x短期记忆、会话状态向量库Milvus / pgvector / Chroma长期记忆向量索引Embeddingtext-embedding 模型或 OpenAI/通义 Embedding API文本向量化Agent 框架LangGraph / LangChain 或自研编排记忆读写逻辑MCP SDKfastmcp 或官方 MCP Python SDK暴露记忆工具应用框架FastAPIREST API部署方式Docker Compose 或 K8s服务编排对于个人开发者或小团队验证最轻量的组合是Redis SQLite Chroma FastAPI全部本地启动不需要 GPU。只要 embedding 使用云端 APICPU 就能跑完整套服务。4.2 环境检查清单部署前先做一轮环境检查Docker 与 Docker Compose 已安装版本不低于 20.10。Python 版本 3.10 至 3.12不建议直接用 3.13部分依赖还未完全兼容。Redis 端口 6379 未被占用。向量库端口如 Milvus 默认 19530未被占用。如需本地 embedding 模型预留至少 8GB 内存磁盘空间预留 20GB 以上。如果要访问云端大模型 API需要有合法的 API Key 和网络权限。4.3 项目目录结构建议按照下面结构组织代码记忆相关的模块单独放置方便后续拆成独立服务。agent-memory-demo/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── memory/ │ │ ├── store.py # Redis 短期记忆 │ │ ├── vector.py # 向量库操作 │ │ ├── extractor.py # 记忆提取与摘要 │ │ └── service.py # 记忆服务统一入口 │ ├── rag/ │ │ ├── pipeline.py # RAG 召回流程 │ │ └── rerank.py # 重排序 │ ├── mcp_server.py # MCP Server │ └── schemas.py # Pydantic 模型 ├── docker-compose.yml ├── requirements.txt └── config.yaml5. 安装部署与启动方式5.1 启动基础设施使用 Docker Compose 启动 Redis 和向量库这里以 Redis Chroma 为例资源占用最小适合本地验证。version: 3.9 services: redis: image: redis:7-alpine container_name: agent-redis ports: - 6379:6379 command: redis-server --appendonly yes volumes: - redis-data:/data chroma: image: chromadb/chroma:latest container_name: agent-chroma ports: - 8001:8000 volumes: - chroma-data:/data volumes: redis-data: chroma-data:启动命令docker compose up -d启动后确认两个服务状态docker ps docker exec -it agent-redis redis-cli ping正常时 Redis 返回 PONG。5.2 Python 依赖安装创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install fastapi uvicorn redis chromadb openai pydantic-settings如果需要用 MCP额外安装 MCP SDKpip install mcp fastmcp注意Chromadb 版本更新较快如果安装后启动报错先检查 Python 版本和依赖版本兼容性优先使用官方文档给出的组合。5.3 启动记忆服务先写一个最简单的 FastAPI 服务验证环境。创建app/main.pyfrom fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgent Memory Service) class HealthResponse(BaseModel): status: str redis: str vector_store: str app.get(/health, response_modelHealthResponse) async def health(): return HealthResponse( statusok, redischecking, vector_storechecking )启动uvicorn app.main:app --host 127.0.0.1 --port 8000访问http://127.0.0.1:8000/health能看到 JSON 响应说明服务正常。后续所有记忆功能都挂载在这个服务上。6. 功能测试与效果验证6.1 Redis 短期记忆读写测试短期记忆的核心功能是按 session_id 存储最近对话支持过期清理。使用 Redis Hash 结构field 是消息 idvalue 是 JSON 序列化的消息对象。实现app/memory/store.pyimport json import time from typing import Optional import redis redis_client redis.Redis(host127.0.0.1, port6379, db0) SESSION_TTL 3600 # 短期记忆保留 1 小时 def append_message(session_id: str, role: str, content: str) - str: 写入一条会话消息 message_id f{session_id}:{int(time.time() * 1000)} message { id: message_id, role: role, content: content, ts: int(time.time()) } redis_client.hset( fsession:{session_id}, message_id, json.dumps(message, ensure_asciiFalse) ) # 刷新过期时间 redis_client.expire(fsession:{session_id}, SESSION_TTL) return message_id def get_recent_messages(session_id: str, limit: int 20) - list: 取最近 N 条消息 raw redis_client.hgetall(fsession:{session_id}) messages [] for msg_id, msg_json in raw.items(): messages.append(json.loads(msg_json)) messages.sort(keylambda x: x[ts]) return messages[-limit:]测试方法from app.memory.store import append_message, get_recent_messages session_id user_1001 append_message(session_id, user, 我想查上个月华东区的销售数据) append_message(session_id, assistant, 请提供具体的时间范围和指标维度) messages get_recent_messages(session_id, limit10) print(len(messages)) # 预期输出 2判断标准消息写入后能按时间顺序读回TTL 过期后自动消失。如果拿不到预期消息优先检查 Redis 连接和 key 前缀。6.2 长期记忆持久化测试长期记忆存在向量库中每条记忆是一个文本片段例如“用户偏好按周查看数据报表”“用户所在部门是销售部”。写入时生成 embedding检索时用 query embedding 做近邻搜索。实现app/memory/vector.pyfrom typing import Optional import chromadb # 假设 py 文件在 app/memory 目录下 PLUGIN None client chromadb.HttpClient(host127.0.0.1, port8001) collection client.get_or_create_collection(nameagent_memories) def embed_text(text: str) - list: 通用 embedding 占位接口。 实际使用时替换为具体 embedding 模型或云端 API。 # 假实现生产环境必须替换 return [0.0] * 768 def save_memory(user_id: str, content: str, metadata: Optional[dict] None): 写入一条长期记忆 memory_id f{user_id}:{abs(hash(content))} embedding embed_text(content) collection.upsert( ids[memory_id], embeddings[embedding], documents[content], metadatas[{ user_id: user_id, **metadata or {} }] ) return memory_id def search_memory(user_id: str, query: str, top_k: int 3) - list: 召回用户相关长期记忆 q_embedding embed_text(query) results collection.query( query_embeddings[q_embedding], n_resultstop_k, where{user_id: user_id} ) return results[documents][0]测试方法先写入三条测试记忆再搜索语义相近的 query确认能召回最相关的一条。from app.memory.vector import save_memory, search_memory save_memory(user_1001, 用户偏好使用柱状图查看销售趋势) save_memory(user_1001, 用户所在的部门是华东销售部) save_memory(user_1001, 用户每天上午查看数据看板) result search_memory(用户喜欢怎么看报表) print(result)预期结果应该包含“用户偏好使用柱状图查看销售趋势”。如果召回结果不相关优先检查 embedding 效果和 top_k 设置。6.3 RAG 多路召回与上下文拼接测试长期记忆召回只是 RAG 的一种特例。在企业级场景中还需要把文档知识、记忆、实时搜索结果统一做成“多路召回”再重排去重后注入 Prompt。这样才能避免向量检索单路召回不准确的问题。实现一个简化版 RAG Pipelinefrom typing import List class RAGPipeline: def __init__(self, retriever_funcs: List[callable]): # retriever_funcs: 每一路召回函数输入 query输出候选片段列表 self.retriever_funcs retriever_funcs def retrieve(self, query: str, top_k: int 5): candidates [] for func in self.retriever_funcs: candidates.extend(func(query)) # 简化版按分数倒序去重 seen set() result [] for item in sorted(candidates, keylambda x: x[score], reverseTrue): # item: {text, score, source} if item[text] not in seen: seen.add(item[text]) result.append(item) return result[:top_k] def build_prompt(self, query: str, memory_items: List[str], documents: List[str]) - str: memory_block \n.join(memory_items) or 暂无相关历史记忆 doc_block \n.join(documents) prompt f你是企业知识助手。请结合历史记忆和知识库回答用户问题。 历史记忆 {memory_block} 知识库内容 {doc_block} 用户问题 {query} return prompt测试的重点是 Prompt 拼接是否正确历史记忆在前、知识库内容居中、用户问题最贴近模型。很多 Agent 效果差不是模型不行而是 Prompt 拼接顺序和信息冗余把模型带偏了。6.4 记忆替换与删除测试长效记忆系统必须支持用户主动删除记忆合规场景下这是硬性要求。def delete_memory(user_id: str, memory_id: str): 按 ID 删除一条长期记忆 collection.delete(ids[memory_id]) def clear_user_memory(user_id: str): 清空某用户的全部长期记忆 all_memories collection.get(where{user_id: user_id}) ids all_memories[ids] if ids: collection.delete(idsids)验证逻辑删除后再搜索结果中不应再出现该条记忆。如果仍然出现多半是向量库缓存或查询条件问题检查 where 条件是否包含 user_id。7. 接口 API 与批量任务7.1 REST API 设计记忆服务对外提供 REST API方便业务系统接入。路径方法功能/v1/memory/shortPOST写入短期记忆/v1/memory/short/{session_id}GET读取短期记忆/v1/memory/longPOST写入长期记忆/v1/memory/long/searchPOST搜索长期记忆/v1/memory/long/{memory_id}DELETE删除单条记忆/v1/rag/promptPOST构建 RAG Prompt在 main.py 中补充路由from fastapi import FastAPI from app.memory import store, vector from app.rag.pipeline import RAGPipeline app FastAPI(titleAgent Memory Service) class ShortMemoryBody(BaseModel): session_id: str role: str content: str class LongMemoryBody(BaseModel): user_id: str content: str class SearchMemoryBody(BaseModel): user_id: str query: str top_k: int 3 app.post(/v1/memory/short) async def write_short_memory(body: ShortMemoryBody): message_id store.append_message(body.session_id, body.role, body.content) return {message_id: message_id} app.post(/v1/memory/long) async def write_long_memory(body: LongMemoryBody): memory_id vector.save_memory(body.user_id, body.content) return {memory_id: memory_id} app.post(/v1/memory/long/search) async def search_long_memory(body: SearchMemoryBody): results vector.search_memory(body.user_id, body.query, body.top_k) return {items: results}7.2 curl 调用示例写入短期记忆curl -X POST http://127.0.0.1:8000/v1/memory/short \ -H Content-Type: application/json \ -d {session_id: user_1001, role: user, content: 帮我查一下华东区销售数据}搜索长期记忆curl -X POST http://127.0.0.1:8000/v1/memory/long/search \ -H Content-Type: application/json \ -d {user_id: user_1001, query: 用户偏好怎么看报表}Python 调用示例import requests base_url http://127.0.0.1:8000 response requests.post( f{base_url}/v1/memory/long/search, json{user_id: user_1001, query: 用户偏好怎么看报表, top_k: 3}, timeout10 ) print(response.json())7.3 批量记忆归档任务生产环境常见需求每天凌晨把当天所有短期记忆做摘要写入长期记忆库。批量任务的实现思路是使用独立队列逐批读取 session调用摘要模型生成记忆条目再写入向量库。import asyncio async def batch_archive_sessions(): 批量归档短期记忆到长期记忆库生产环境由定时任务触发 # 伪代码通过 Redis SCAN 找出当天活跃 session active_sessions [user_1001, user_1002] for session_id in active_sessions: messages store.get_recent_messages(session_id, limit100) summary await summarize_messages(messages) # 调用大模型生成摘要 vector.save_memory( user_idsession_id, contentsummary, metadata{source: archive, type: session_summary} ) return {archived: len(active_sessions)}批量任务必须做好幂等控制同一个 session 不能重复写入同样的记忆。建议在 metadata 中加入 session_id 和 batch_date写入前先检查是否已存在。8. 资源占用与性能观察8.1 Redis 内存观察短期记忆在 Redis 中占用内存与消息量成正比。观察方法redis-cli info memory redis-cli --bigkeys如果内存增长过快先看是不是 TTL 没有生效再检查是否有过大的 value。单条消息不要超过 64KB长响应可以截断或摘要后再存入。8.2 向量库性能观察向量查询的延迟受集合大小、向量维度、索引类型影响。本地 Chroma 在小数据量下查询毫秒级百万级数据则建议切换 Milvus 或专用向量库。性能观察维度写入延迟单条记忆写入耗时。查询延迟一次检索 top_k 的耗时。召回率标注一批测试 query看正确记忆是否排进前 k。如果查询变慢先确认 embedding 维度是否统一不同模型返回的向量维度不一致会导致全表扫描。8.3 大模型调用延迟与上下文长度记忆架构引入后每个请求的成本结构发生变化embedding 调用增加但按条计费成本很低。检索耗时在 RAG 中通常可以控制在 50ms 以内。Prompt 长度从“全量历史”变成“相关记忆 固定知识条目”整体 token 数通常下降。Agent 因为拿到相关记忆重试和澄清次数减少反而节省总成本。建议在日志中记录每个请求的 prompt token 数、检索耗时长、模型响应时间建立基线后再做优化。9. 常见问题与排查方法问题现象可能原因排查方式解决方案短期记忆写入后读不到Redis key 过期或连接被断查看 Redis logs、检查 TTL调整 SESSION_TTL检查连接池配置向量检索结果不相关Embedding 模型质量差或语义不在同一空间对比不同 query 的召回结果换更强的 embedding 模型或多路召回向量维度不一致使用了不同 embedding 模型检查 collection 元信息固定模型版本建立模型映射表MCP 工具调用失败MCP Server 未启动或工具名错误查看 MCP Server 日志检查工具注册名重新加载 Agent 配置Redis 连接超时受保护模式下无法远程连接执行 redis-cli ping修改 redis.conf 绑定地址注意安全组限制Prompt 拼接过长记忆和知识条目去重失效查看日志中的 prompt token 数提高去重阈值限制单路召回条数批量归档重复写入缺少幂等控制检查长期记忆库相同 metadata 数量在写入前做 session_id batch_date 去重服务启动后端口冲突8001 或 8000 被其他进程占用执行 netstat 查看端口修改启动端口或关闭占用进程10. 最佳实践与合规边界10.1 架构实践建议第一版不要做太重。先用 Redis SQLite Chroma 验证记忆链路再迁移到生产级组件。长期记忆写入前一定要做抽取和摘要切忌原样保存所有对话否则向量库会变成垃圾场。记忆召回必须加 user_id 过滤这是多租户隔离的最低要求。多路召回 rerank 比单路向量检索稳定得多特别是业务术语多、近义词多的企业场景。记忆系统要预留删除接口不要只在数据库里硬删最好有软删除和审计日志。10.2 隐私与数据合规记忆系统中存储了大量用户行为和企业业务数据必须遵循最小化原则默认不采集与任务无关的个人敏感信息。涉及用户身份、联系方式、健康信息等内容必须获得明确授权。用户发起删除请求时应在规定时间内完成记忆清理包括向量库中的 embedding。企业内部使用时记忆库的访问权限要与业务系统一致不能比业务系统更高的开放度。不要将真实姓名、手机号等敏感字段作为向量的直接文本内容可先脱敏处理。工程团队上线此类系统前建议交给法务和隐私团队做一次数据流评估。技术能力可以快速搭建但数据合规的坑一旦踩下去修复成本极高。11. 总结与下一步这套 Agent Memory 架构最值得尝试的点是把记忆从 Prompt 移到存储层之后Agent 的多轮对话能力和跨会话连续性会有一个明显的提升而且不会因为上下文无限增长拖垮推理性能。建议最先验证三个功能短期记忆的 Redis 读写是否稳定、长期记忆的语义召回是否准确、MCP 工具调用是否连通。这三个点通了整个记忆链路就通了。最容易踩的坑有两个一是忘记做记忆抽取和摘要直接灌原始对话导致检索效果差二是长期记忆库没有做租户隔离用户 A 的记忆被用户 B 召回这在企业环境里是严重的合规事故。后续可以继续扩展的方向包括GraphRAG 做实体关系记忆、MCP 接入数据库查询工具、记忆压缩策略的自动调优、以及跨多 Agent 的共享记忆池。先把基础记忆链路跑通再逐步往上加深比一开始就堆一套复杂系统要可靠得多。 企业级Agent Memory架构：从Context到长期记忆的工程实践
做企业级 Agent 应用最难的不是把模型接入业务而是让 Agent 在跨会话、跨业务线、长时间运行后还记得上下文。很多团队把几百页文档、几千轮对话全部塞进 PromptContext 越拼越长效果越来越差延迟和成本反而一起涨。这次我们把“企业级 Agent Memory 架构”完整拆开从 Context 窗口聊到真正的 Long-term Memory讲清楚分层记忆模型怎么做短期记忆、长期记忆各用什么存储再落到 RAG 检索增强和 MCP 工具协议两个方向最后给出一套带代码的实现模板。文章不是 PPT 式的概念罗列而是可以直接照着搭的工程方案Redis 做会话级短期记忆向量库做跨会话长期记忆RAG 做记忆召回MCP 把记忆能力暴露给 Agent 工具调用。关注本地部署、接口服务、批量任务、知识库回填和多 Agent 协作的读者这篇文章可以直接收藏。1. 企业级 Agent Memory 架构核心能力速览先把方案的全貌放出来后面所有代码和步骤都是围绕这张表展开的。能力项说明架构方案分层记忆模型短期记忆 工作记忆 长期记忆短期记忆Redis 会话缓存按 session_id 隔离支持 TTL 过期长期记忆向量数据库持久化存储对话摘要、用户偏好、事实性知识记忆召回embedding 向量检索 关键词多路召回 rerank服务接入MCP 协议工具 REST API 双通道批量任务批量会话摘要、记忆归档、向量重建、记忆压缩上下文工程Prompt 动态注入按“当前会话 相关记忆 知识库”三段拼接推理硬件本地 embedding 模型可用 CPU 运行接入云端模型 API 则无需 GPU适合场景智能客服、企业 Copilot、流程助手、多轮数据分析、知识库问答这套架构解决的关键问题只有一个让 Agent 从“无状态调用”变成“有状态服务”。2. 为什么 Agent 需要长效记忆从 Context 到 Memory先看一个真实的工程困境。企业里的 Agent 通常不是一次性问答而是以周、月为周期的持续服务。用户可能三天后回来继续上次的需求或者在钉钉/飞书群里让 Agent 处理跨部门的流程审批。如果每一次对话都从零开始用户每次都要重复需求背景、身份信息、项目上下文这是不可接受的产品体验。Context 窗口是有限资源。以主流大模型为例上下文窗口从 4K 到 200K 甚至更大看起来很大但企业真实场景里一份流程文档可能 20K token一套业务规则 30K token再加上历史对话、用户画像、工具返回结果很快超出窗口边界。更麻烦的是上下文过长会导致模型“迷失在中间”召回准确率下降响应延迟和成本同步上升。所以工程上必须把记忆从 Prompt 里解放出来放到外部存储中。这就是 Agent Memory 架构的出发点短期记忆当前会话内的对话轮次、临时状态存放在 Redis 这类高速缓存里TTL 过期后自动清理。工作记忆当前任务正在处理的中间结果、工具调用记录、待确认字段属于“正在思考”的数据。长期记忆跨会话持久化的用户偏好、关键事实、历史决策、项目上下文存放在向量库或结构化存储中按需召回。真正的 Long-term Memory 不是把历史对话原样存下来而是对记忆做提取、压缩、索引、召回、遗忘的全生命周期管理。这决定了 Agent 是“听起来聪明”还是“真的了解用户”。从上下文工程的视角看Long-term Memory 的本质是让 Agent 在有限的上下文预算内拿到“当前时刻最该知道的信息”。它和 RAG 有天然的结合点把长期记忆当作一类特殊的文档库用语义检索的方式按需召回。区别在于RAG 的知识来源通常是企业文档、网页、数据库记录而记忆库的来源是用户与 Agent 的交互历史。3. 企业级 Agent Memory 架构设计3.1 分层记忆模型整个架构按记忆生命周期分成四层每一层对应不同的存储介质和访问频率。记忆层存储介质生命周期访问频率示例短期记忆Redis分钟到小时极高当前会话最近 20 轮对话工作记忆Redis / 内存任务进行中高工具调用结果、待确认参数长期记忆向量库 关系库天到月中用户偏好、历史决策、项目上下文业务知识库向量库版本化管理低企业文档、规则、FAQ短期记忆和工作记忆可以共用 Redis 实例但建议用不同的 key 前缀区分避免数据互相污染。3.2 记忆生命周期记忆不是“写进去就不管了”它和业务系统一样需要生命周期管理写入每次对话结束后把关键信息写入短期记忆。提取定期或按事件触发从短期记忆中提取用户偏好、事实、决策记录。摘要对长对话进行摘要压缩避免长期记忆库无限膨胀。归档超过一定时效的记忆降级为冷数据减少检索噪音。召回新对话开始时从长期记忆中检索与当前 Query 最相关的记忆片段。遗忘用户要求删除或数据过期时彻底清除相关记忆。这个流程是长效记忆系统最核心的部分也是长期记忆和“历史记录文件”的区别所在。3.3 关键组件职责Memory Service统一管理记忆读写、提取、摘要、删除对外提供 REST API 和 MCP 工具。Redis Store短期记忆和工作记忆的存储使用 Hash 或 String 结构存储 JSON 序列化的消息。Vector Store长期记忆的索引存储保存 embedding 向量和原始内容。Extractor从对话中抽取记忆条目可以用规则匹配、小模型或大模型函数调用实现。RAG Pipeline负责将查询语句 embedding 化、检索、rerank、拼装上下文。MCP Server把记忆能力封装为标准工具Agent 可通过 MCP 协议直接调用。4. 技术选型与本地部署环境准备4.1 技术栈清单下面这套技术栈是通用选择具体组件可按团队已有基础替换组件选型用途缓存Redis 7.x短期记忆、会话状态向量库Milvus / pgvector / Chroma长期记忆向量索引Embeddingtext-embedding 模型或 OpenAI/通义 Embedding API文本向量化Agent 框架LangGraph / LangChain 或自研编排记忆读写逻辑MCP SDKfastmcp 或官方 MCP Python SDK暴露记忆工具应用框架FastAPIREST API部署方式Docker Compose 或 K8s服务编排对于个人开发者或小团队验证最轻量的组合是Redis SQLite Chroma FastAPI全部本地启动不需要 GPU。只要 embedding 使用云端 APICPU 就能跑完整套服务。4.2 环境检查清单部署前先做一轮环境检查Docker 与 Docker Compose 已安装版本不低于 20.10。Python 版本 3.10 至 3.12不建议直接用 3.13部分依赖还未完全兼容。Redis 端口 6379 未被占用。向量库端口如 Milvus 默认 19530未被占用。如需本地 embedding 模型预留至少 8GB 内存磁盘空间预留 20GB 以上。如果要访问云端大模型 API需要有合法的 API Key 和网络权限。4.3 项目目录结构建议按照下面结构组织代码记忆相关的模块单独放置方便后续拆成独立服务。agent-memory-demo/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── memory/ │ │ ├── store.py # Redis 短期记忆 │ │ ├── vector.py # 向量库操作 │ │ ├── extractor.py # 记忆提取与摘要 │ │ └── service.py # 记忆服务统一入口 │ ├── rag/ │ │ ├── pipeline.py # RAG 召回流程 │ │ └── rerank.py # 重排序 │ ├── mcp_server.py # MCP Server │ └── schemas.py # Pydantic 模型 ├── docker-compose.yml ├── requirements.txt └── config.yaml5. 安装部署与启动方式5.1 启动基础设施使用 Docker Compose 启动 Redis 和向量库这里以 Redis Chroma 为例资源占用最小适合本地验证。version: 3.9 services: redis: image: redis:7-alpine container_name: agent-redis ports: - 6379:6379 command: redis-server --appendonly yes volumes: - redis-data:/data chroma: image: chromadb/chroma:latest container_name: agent-chroma ports: - 8001:8000 volumes: - chroma-data:/data volumes: redis-data: chroma-data:启动命令docker compose up -d启动后确认两个服务状态docker ps docker exec -it agent-redis redis-cli ping正常时 Redis 返回 PONG。5.2 Python 依赖安装创建虚拟环境并安装依赖python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate pip install fastapi uvicorn redis chromadb openai pydantic-settings如果需要用 MCP额外安装 MCP SDKpip install mcp fastmcp注意Chromadb 版本更新较快如果安装后启动报错先检查 Python 版本和依赖版本兼容性优先使用官方文档给出的组合。5.3 启动记忆服务先写一个最简单的 FastAPI 服务验证环境。创建app/main.pyfrom fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgent Memory Service) class HealthResponse(BaseModel): status: str redis: str vector_store: str app.get(/health, response_modelHealthResponse) async def health(): return HealthResponse( statusok, redischecking, vector_storechecking )启动uvicorn app.main:app --host 127.0.0.1 --port 8000访问http://127.0.0.1:8000/health能看到 JSON 响应说明服务正常。后续所有记忆功能都挂载在这个服务上。6. 功能测试与效果验证6.1 Redis 短期记忆读写测试短期记忆的核心功能是按 session_id 存储最近对话支持过期清理。使用 Redis Hash 结构field 是消息 idvalue 是 JSON 序列化的消息对象。实现app/memory/store.pyimport json import time from typing import Optional import redis redis_client redis.Redis(host127.0.0.1, port6379, db0) SESSION_TTL 3600 # 短期记忆保留 1 小时 def append_message(session_id: str, role: str, content: str) - str: 写入一条会话消息 message_id f{session_id}:{int(time.time() * 1000)} message { id: message_id, role: role, content: content, ts: int(time.time()) } redis_client.hset( fsession:{session_id}, message_id, json.dumps(message, ensure_asciiFalse) ) # 刷新过期时间 redis_client.expire(fsession:{session_id}, SESSION_TTL) return message_id def get_recent_messages(session_id: str, limit: int 20) - list: 取最近 N 条消息 raw redis_client.hgetall(fsession:{session_id}) messages [] for msg_id, msg_json in raw.items(): messages.append(json.loads(msg_json)) messages.sort(keylambda x: x[ts]) return messages[-limit:]测试方法from app.memory.store import append_message, get_recent_messages session_id user_1001 append_message(session_id, user, 我想查上个月华东区的销售数据) append_message(session_id, assistant, 请提供具体的时间范围和指标维度) messages get_recent_messages(session_id, limit10) print(len(messages)) # 预期输出 2判断标准消息写入后能按时间顺序读回TTL 过期后自动消失。如果拿不到预期消息优先检查 Redis 连接和 key 前缀。6.2 长期记忆持久化测试长期记忆存在向量库中每条记忆是一个文本片段例如“用户偏好按周查看数据报表”“用户所在部门是销售部”。写入时生成 embedding检索时用 query embedding 做近邻搜索。实现app/memory/vector.pyfrom typing import Optional import chromadb # 假设 py 文件在 app/memory 目录下 PLUGIN None client chromadb.HttpClient(host127.0.0.1, port8001) collection client.get_or_create_collection(nameagent_memories) def embed_text(text: str) - list: 通用 embedding 占位接口。 实际使用时替换为具体 embedding 模型或云端 API。 # 假实现生产环境必须替换 return [0.0] * 768 def save_memory(user_id: str, content: str, metadata: Optional[dict] None): 写入一条长期记忆 memory_id f{user_id}:{abs(hash(content))} embedding embed_text(content) collection.upsert( ids[memory_id], embeddings[embedding], documents[content], metadatas[{ user_id: user_id, **metadata or {} }] ) return memory_id def search_memory(user_id: str, query: str, top_k: int 3) - list: 召回用户相关长期记忆 q_embedding embed_text(query) results collection.query( query_embeddings[q_embedding], n_resultstop_k, where{user_id: user_id} ) return results[documents][0]测试方法先写入三条测试记忆再搜索语义相近的 query确认能召回最相关的一条。from app.memory.vector import save_memory, search_memory save_memory(user_1001, 用户偏好使用柱状图查看销售趋势) save_memory(user_1001, 用户所在的部门是华东销售部) save_memory(user_1001, 用户每天上午查看数据看板) result search_memory(用户喜欢怎么看报表) print(result)预期结果应该包含“用户偏好使用柱状图查看销售趋势”。如果召回结果不相关优先检查 embedding 效果和 top_k 设置。6.3 RAG 多路召回与上下文拼接测试长期记忆召回只是 RAG 的一种特例。在企业级场景中还需要把文档知识、记忆、实时搜索结果统一做成“多路召回”再重排去重后注入 Prompt。这样才能避免向量检索单路召回不准确的问题。实现一个简化版 RAG Pipelinefrom typing import List class RAGPipeline: def __init__(self, retriever_funcs: List[callable]): # retriever_funcs: 每一路召回函数输入 query输出候选片段列表 self.retriever_funcs retriever_funcs def retrieve(self, query: str, top_k: int 5): candidates [] for func in self.retriever_funcs: candidates.extend(func(query)) # 简化版按分数倒序去重 seen set() result [] for item in sorted(candidates, keylambda x: x[score], reverseTrue): # item: {text, score, source} if item[text] not in seen: seen.add(item[text]) result.append(item) return result[:top_k] def build_prompt(self, query: str, memory_items: List[str], documents: List[str]) - str: memory_block \n.join(memory_items) or 暂无相关历史记忆 doc_block \n.join(documents) prompt f你是企业知识助手。请结合历史记忆和知识库回答用户问题。 历史记忆 {memory_block} 知识库内容 {doc_block} 用户问题 {query} return prompt测试的重点是 Prompt 拼接是否正确历史记忆在前、知识库内容居中、用户问题最贴近模型。很多 Agent 效果差不是模型不行而是 Prompt 拼接顺序和信息冗余把模型带偏了。6.4 记忆替换与删除测试长效记忆系统必须支持用户主动删除记忆合规场景下这是硬性要求。def delete_memory(user_id: str, memory_id: str): 按 ID 删除一条长期记忆 collection.delete(ids[memory_id]) def clear_user_memory(user_id: str): 清空某用户的全部长期记忆 all_memories collection.get(where{user_id: user_id}) ids all_memories[ids] if ids: collection.delete(idsids)验证逻辑删除后再搜索结果中不应再出现该条记忆。如果仍然出现多半是向量库缓存或查询条件问题检查 where 条件是否包含 user_id。7. 接口 API 与批量任务7.1 REST API 设计记忆服务对外提供 REST API方便业务系统接入。路径方法功能/v1/memory/shortPOST写入短期记忆/v1/memory/short/{session_id}GET读取短期记忆/v1/memory/longPOST写入长期记忆/v1/memory/long/searchPOST搜索长期记忆/v1/memory/long/{memory_id}DELETE删除单条记忆/v1/rag/promptPOST构建 RAG Prompt在 main.py 中补充路由from fastapi import FastAPI from app.memory import store, vector from app.rag.pipeline import RAGPipeline app FastAPI(titleAgent Memory Service) class ShortMemoryBody(BaseModel): session_id: str role: str content: str class LongMemoryBody(BaseModel): user_id: str content: str class SearchMemoryBody(BaseModel): user_id: str query: str top_k: int 3 app.post(/v1/memory/short) async def write_short_memory(body: ShortMemoryBody): message_id store.append_message(body.session_id, body.role, body.content) return {message_id: message_id} app.post(/v1/memory/long) async def write_long_memory(body: LongMemoryBody): memory_id vector.save_memory(body.user_id, body.content) return {memory_id: memory_id} app.post(/v1/memory/long/search) async def search_long_memory(body: SearchMemoryBody): results vector.search_memory(body.user_id, body.query, body.top_k) return {items: results}7.2 curl 调用示例写入短期记忆curl -X POST http://127.0.0.1:8000/v1/memory/short \ -H Content-Type: application/json \ -d {session_id: user_1001, role: user, content: 帮我查一下华东区销售数据}搜索长期记忆curl -X POST http://127.0.0.1:8000/v1/memory/long/search \ -H Content-Type: application/json \ -d {user_id: user_1001, query: 用户偏好怎么看报表}Python 调用示例import requests base_url http://127.0.0.1:8000 response requests.post( f{base_url}/v1/memory/long/search, json{user_id: user_1001, query: 用户偏好怎么看报表, top_k: 3}, timeout10 ) print(response.json())7.3 批量记忆归档任务生产环境常见需求每天凌晨把当天所有短期记忆做摘要写入长期记忆库。批量任务的实现思路是使用独立队列逐批读取 session调用摘要模型生成记忆条目再写入向量库。import asyncio async def batch_archive_sessions(): 批量归档短期记忆到长期记忆库生产环境由定时任务触发 # 伪代码通过 Redis SCAN 找出当天活跃 session active_sessions [user_1001, user_1002] for session_id in active_sessions: messages store.get_recent_messages(session_id, limit100) summary await summarize_messages(messages) # 调用大模型生成摘要 vector.save_memory( user_idsession_id, contentsummary, metadata{source: archive, type: session_summary} ) return {archived: len(active_sessions)}批量任务必须做好幂等控制同一个 session 不能重复写入同样的记忆。建议在 metadata 中加入 session_id 和 batch_date写入前先检查是否已存在。8. 资源占用与性能观察8.1 Redis 内存观察短期记忆在 Redis 中占用内存与消息量成正比。观察方法redis-cli info memory redis-cli --bigkeys如果内存增长过快先看是不是 TTL 没有生效再检查是否有过大的 value。单条消息不要超过 64KB长响应可以截断或摘要后再存入。8.2 向量库性能观察向量查询的延迟受集合大小、向量维度、索引类型影响。本地 Chroma 在小数据量下查询毫秒级百万级数据则建议切换 Milvus 或专用向量库。性能观察维度写入延迟单条记忆写入耗时。查询延迟一次检索 top_k 的耗时。召回率标注一批测试 query看正确记忆是否排进前 k。如果查询变慢先确认 embedding 维度是否统一不同模型返回的向量维度不一致会导致全表扫描。8.3 大模型调用延迟与上下文长度记忆架构引入后每个请求的成本结构发生变化embedding 调用增加但按条计费成本很低。检索耗时在 RAG 中通常可以控制在 50ms 以内。Prompt 长度从“全量历史”变成“相关记忆 固定知识条目”整体 token 数通常下降。Agent 因为拿到相关记忆重试和澄清次数减少反而节省总成本。建议在日志中记录每个请求的 prompt token 数、检索耗时长、模型响应时间建立基线后再做优化。9. 常见问题与排查方法问题现象可能原因排查方式解决方案短期记忆写入后读不到Redis key 过期或连接被断查看 Redis logs、检查 TTL调整 SESSION_TTL检查连接池配置向量检索结果不相关Embedding 模型质量差或语义不在同一空间对比不同 query 的召回结果换更强的 embedding 模型或多路召回向量维度不一致使用了不同 embedding 模型检查 collection 元信息固定模型版本建立模型映射表MCP 工具调用失败MCP Server 未启动或工具名错误查看 MCP Server 日志检查工具注册名重新加载 Agent 配置Redis 连接超时受保护模式下无法远程连接执行 redis-cli ping修改 redis.conf 绑定地址注意安全组限制Prompt 拼接过长记忆和知识条目去重失效查看日志中的 prompt token 数提高去重阈值限制单路召回条数批量归档重复写入缺少幂等控制检查长期记忆库相同 metadata 数量在写入前做 session_id batch_date 去重服务启动后端口冲突8001 或 8000 被其他进程占用执行 netstat 查看端口修改启动端口或关闭占用进程10. 最佳实践与合规边界10.1 架构实践建议第一版不要做太重。先用 Redis SQLite Chroma 验证记忆链路再迁移到生产级组件。长期记忆写入前一定要做抽取和摘要切忌原样保存所有对话否则向量库会变成垃圾场。记忆召回必须加 user_id 过滤这是多租户隔离的最低要求。多路召回 rerank 比单路向量检索稳定得多特别是业务术语多、近义词多的企业场景。记忆系统要预留删除接口不要只在数据库里硬删最好有软删除和审计日志。10.2 隐私与数据合规记忆系统中存储了大量用户行为和企业业务数据必须遵循最小化原则默认不采集与任务无关的个人敏感信息。涉及用户身份、联系方式、健康信息等内容必须获得明确授权。用户发起删除请求时应在规定时间内完成记忆清理包括向量库中的 embedding。企业内部使用时记忆库的访问权限要与业务系统一致不能比业务系统更高的开放度。不要将真实姓名、手机号等敏感字段作为向量的直接文本内容可先脱敏处理。工程团队上线此类系统前建议交给法务和隐私团队做一次数据流评估。技术能力可以快速搭建但数据合规的坑一旦踩下去修复成本极高。11. 总结与下一步这套 Agent Memory 架构最值得尝试的点是把记忆从 Prompt 移到存储层之后Agent 的多轮对话能力和跨会话连续性会有一个明显的提升而且不会因为上下文无限增长拖垮推理性能。建议最先验证三个功能短期记忆的 Redis 读写是否稳定、长期记忆的语义召回是否准确、MCP 工具调用是否连通。这三个点通了整个记忆链路就通了。最容易踩的坑有两个一是忘记做记忆抽取和摘要直接灌原始对话导致检索效果差二是长期记忆库没有做租户隔离用户 A 的记忆被用户 B 召回这在企业环境里是严重的合规事故。后续可以继续扩展的方向包括GraphRAG 做实体关系记忆、MCP 接入数据库查询工具、记忆压缩策略的自动调优、以及跨多 Agent 的共享记忆池。先把基础记忆链路跑通再逐步往上加深比一开始就堆一套复杂系统要可靠得多。