从Context到Long-term Memory：企业级Agent记忆架构与MCP落地

企业级对话系统一旦跨过 Demo 阶段，最先暴露的问题通常是记忆。不是模型不会回答，而是它记不住：用户三分钟前提过的需求，换个会话就完全消失；管理员整理好的业务偏好，每次都要重新描述。AI 大模型虽然有越来越大的 Context 窗口，但 Context 是输入拼盘，不是真正的 Long-term Memory。为了解决这个问题，需要在上下文工程、RAG 和 MCP 协议之间找到一条可落地的组合方案。

这篇文章会围绕“从 Context 到 Long-term Memory”这条主线，先讲清概念差异，再给出企业级 Agent Memory 的分层架构，然后带着你实现一个最小可运行模块。它会用到 Redis 做短期会话记忆，用向量检索做长期事实记忆，并用一个简化版 MCP Server 把记忆能力开放给 Agent。最后会补齐检索质量优化、常见问题排查和生产环境改造建议。读完后，你可以把这套思路接进自己的对话系统，而不是只记住几个术语。

1. 先理解 Context 与 Long-term Memory 的本质区别

1.1 Context 窗口的物理边界与会话短路

Context 窗口是模型一次请求能接收的 Token 数量，它决定了“当前输入能装下多少字”。常见模型的窗口从几千到几十万不等，但窗口扩大并不等于记忆能力增强。

以当前常见模型为例，可以大致把窗口分成几个档位：

实际项目里最常见的错误，是把所有历史消息、文档片段、用户画像一股脑塞进 Context。这样做的直接后果是请求变慢、费用变高，而且很多内容离当前问题很远，模型反而更容易被噪声带偏。Context 窗口越大，越需要上下文工程来控制“什么该进窗口、以什么顺序进、保留多少”。

1.2 Long-term Memory 要解决的不只是“记得久”

Long-term Memory 是指 Agent 能跨会话保存和回溯状态、事实和知识的能力。它至少分成三类：

• 情节记忆：历史交互记录，比如“上周二用户问过部署流程”。
• 事实记忆：用户属性和明确偏好，比如“用户负责数据分析团队”“用户不喜欢邮件通知”。
• 程序性记忆：Agent 学习到或约定好的执行流程，比如“对生产环境变更必须二次确认”。

Context 只是一次请求的输入快照，而 Long-term Memory 是持续变化的资产。用户纠正过一次偏好后，下一次对话应该直接使用纠正后的结果。比如用户说“别再发邮件了，只发企业微信”，Agent 需要更新事实记忆，而不仅仅是把这条消息留在历史里。

从工程角度看，Long-term Memory 还涉及写入、更新、过期、权限、检索质量等一系列问题。只靠把 Context 调大，无法解决这些状态管理问题。

1.3 上下文工程：把记忆变成 LLM 能消费的输入

上下文工程的核心，是根据当前需求从记忆池中选出高价值的片段，并组装成对模型最友好的输入。

一个简化版 Prompt 结构可以这样设计：

SYSTEM: 你是企业知识助手。回答前先参考下面提供的用户记忆和背景。 用户画像： {user_profile} 长期记忆： {long_term_memory} 当前会话近期记录： {recent_messages} HISTORY: {truncated_history} USER: {current_query}

这里的重点是：长期记忆不是原始聊天记录，而是经过抽取、去重和重要性打分的记忆片段。近期对话可以保留原始消息，但也要限制条数。这样做，既利用了短期上下文的细节，又让长期事实跨会话复用。

注意：不要把所有历史消息原封不动塞回 Context。记忆是有损压缩，核心是提取“对后续对话有价值”的信息。

2. 企业级 Agent Memory 架构的分层设计

2.1 四层架构：接入层、服务层、存储层、模型层

企业级 Agent Memory 不能只写一个函数，它需要分成几层，每一层有明确职责。

接入层解决“Agent 怎么找到记忆服务”的问题。服务层解决“记忆怎么写入和读出来”。存储层解决“记忆放在哪里”。模型层解决“哪些文本值得存，哪段记忆与当前问题相关”。

2.2 数据模型与记忆实体设计

设计记忆表时，至少需要区分短期消息和长期记忆。短期消息可以按会话维度保存，长期记忆必须按用户或业务主体维度保存。

一个长期记忆表的核心字段如下：

短期消息表相对简单，核心字段可以包括：session_id、role、content、created_at。Redis 中的 Key 设计可以这样考虑：

session:{session_id}:messages session:{session_id}:meta

短期消息采用 List 结构，固定保留最近 N 条，并设置过期时间。这样既能支撑多轮对话，又不会无限增长。

2.3 记忆写入与读取主链路

记忆写入链路和读取链路要分开设计，因为它们对延迟和一致性的要求不一样。

写入链路：

1. 用户和 Agent 完成一轮对话。
2. 消息进入记忆抽取模块，由 LLM 提取潜在事实、偏好或事件。
3. 对提取结果做去重和覆盖判断。
4. 计算重要性分数和向量。
5. 写入长期记忆存储，同时更新 Redis 短期会话记录。
6. 异步记录更新日志，便于审计和排查。

读取链路：

1. 用户发起新请求。
2. 读取 Redis 中的会话近期消息。
3. 从长期记忆库召回与 query 相关的记忆片段。
4. 对候选记忆做重排和过滤。
5. 将用户画像、长期记忆、近期对话按顺序组装进 Prompt。
6. 调用 LLM 生成回答。

写入和读取之间的关键点是：写入尽量异步化，不影响主对话延迟；读取是在线链路，需要控制召回数量和耗时。

3. 用 Redis 和 SQLite 实现一个最小可运行记忆模块

3.1 环境准备与项目结构

为了让方案可复现，这里实现一个简化但完整的内存模块。环境要求如下：

• Python 3.10+
• Redis 6+
• 可以联网下载向量模型
• 如果使用 LLM 抽取记忆，需要 OpenAI 兼容接口的 API Key

先安装依赖：

pip install fastapi uvicorn redis numpy sentence-transformers openai

项目结构如下：

agent-memory/ ├── requirements.txt ├── config.py ├── redis_store.py ├── vector_store.py ├── memory_service.py ├── mcp_server.py └── main.py

config.py 主要放 Redis 地址、模型名称、TopK 和 TTL 等配置。下面是一个示例：

import os REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0") EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "all-MiniLM-L6-v2") MEMORY_TOPK = int(os.getenv("MEMORY_TOPK", "3")) SESSION_TTL = int(os.getenv("SESSION_TTL", "3600")) SESSION_MAX_MESSAGES = int(os.getenv("SESSION_MAX_MESSAGES", "20")) LLM_BASE_URL = os.getenv("LLM_BASE_URL", "") LLM_API_KEY = os.getenv("LLM_API_KEY", "") LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o-mini")

3.2 用 Redis 做短期会话记忆

Redis 可以很自然地承担短期会话记忆。这里实现一个RedisSessionStore，用 List 保存消息，用 Expire 控制会话生命周期。

import redis import json class RedisSessionStore: def __init__(self, redis_url: str, ttl: int = 3600, max_messages: int = 20): self.client = redis.from_url(redis_url, decode_responses=True) self.ttl = ttl self.max_messages = max_messages def add_message(self, session_id: str, role: str, content: str) -> None: key = f"session:{session_id}:messages" message = json.dumps({"role": role, "content": content}, ensure_ascii=False) self.client.rpush(key, message) self.client.ltrim(key, -self.max_messages, -1) self.client.expire(key, self.ttl) def get_recent_messages(self, session_id: str, last_n: int = 10) -> list: key = f"session:{session_id}:messages" raw_messages = self.client.lrange(key, -last_n, -1) messages = [] for raw in raw_messages: messages.append(json.loads(raw)) return messages def clear(self, session_id: str) -> None: key = f"session:{session_id}:messages" self.client.delete(key)

这里的关键点有三个。第一，ltrim保证列表不会无限增长，超出最近 20 条旧消息会被裁剪。第二，expire设置 TTL，让短期记忆自动过期，避免 Redis 内存被长期占用。第三，decode_responses=True让读写结果默认按字符串处理，省去字节解码。

很多人关心 Redis Agent Memory 怎么用，本质就是把短期会话状态和长期记忆分离。Redis 放短期消息，长期事实记忆放到向量存储中。

3.3 用 SQLite 和向量相似度实现长期记忆

这里不引入外部向量数据库，直接用 SQLite 保存文本和 embedding，再用 numpy 计算余弦相似度。对于几百条学习级别记忆足够，生产环境可以替换为 pgvector 或 Milvus。

import sqlite3 import numpy as np import json from sentence_transformers import SentenceTransformer class VectorMemoryStore: def __init__(self, db_path: str, model_name: str = "all-MiniLM-L6-v2"): self.db_path = db_path self.model = SentenceTransformer(model_name) self.init_db() def init_db(self): conn = sqlite3.connect(self.db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS memories ( memory_id TEXT PRIMARY KEY, user_id TEXT, session_id TEXT, content TEXT, memory_type TEXT, importance_score REAL, embedding TEXT, status TEXT DEFAULT 'active', last_access_at TEXT, created_at TEXT ) """) conn.commit() conn.close() def _embed(self, text: str) -> list: return self.model.encode(text).tolist() @staticmethod def _cosine_similarity(vec_a: list, vec_b: list) -> float: a = np.array(vec_a) b = np.array(vec_b) return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))) def add_memory(self, user_id: str, session_id: str, content: str, memory_type: str, importance_score: float) -> str: import uuid memory_id = f"mem_{uuid.uuid4().hex[:12]}" embedding = self._embed(content) conn = sqlite3.connect(self.db_path) conn.execute( """ INSERT INTO memories (memory_id, user_id, session_id, content, memory_type, importance_score, embedding, status, last_access_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'active', datetime('now'), datetime('now')) """, (memory_id, user_id, session_id, content, memory_type, importance_score, json.dumps(embedding)) ) conn.commit() conn.close() return memory_id def search_topk(self, user_id: str, query: str, top_k: int = 3, min_score: float = 0.3) -> list: conn = sqlite3.connect(self.db_path) rows = conn.execute( """ SELECT memory_id, content, memory_type, importance_score, embedding FROM memories WHERE user_id = ? AND status = 'active' """, (user_id,) ).fetchall() conn.close() query_embedding = self._embed(query) candidates = [] for memory_id, content, memory_type, importance_score, embedding_text in rows: memory_embedding = json.loads(embedding_text) score = self._cosine_similarity(query_embedding, memory_embedding) if score >= min_score: candidates.append({ "memory_id": memory_id, "content": content, "memory_type": memory_type, "importance_score": importance_score, "score": score }) candidates.sort(key=lambda x: x["score"], reverse=True) return candidates[:top_k]

这里把用户 ID 作为过滤条件，避免不同用户之间记忆串线。SQLite 和向量计算虽然简单，但已经具备长期记忆的最小闭环。生产环境要注意模型版本统一，否则不同时间写入的 embedding 向量不在同一语义空间，检索会失真。

3.4 记忆抽取与写入：用 LLM 把对话沉淀为事实

不是所有聊天内容都值得写入长期记忆。需要从对话中提取对后续有用的信息。这里设计一个MemoryExtractor，调用 OpenAI 兼容接口来完成抽取。

import json from openai import OpenAI class MemoryExtractor: def __init__(self, base_url: str, api_key: str, model: str): self.client = OpenAI(base_url=base_url, api_key=api_key) self.model = model def extract(self, user_message: str, assistant_message: str) -> list: prompt = f""" 从以下对话中抽取值得长期记忆的信息。 用户消息：{user_message} 助手消息：{assistant_message} 要求： 1. 只输出 JSON 数组，不输出解释。 2. 每条记忆包括 type、content、importance。 3. type 只能是 fact、preference、event、procedure。 4. importance 是 0 到 1 的浮点数。 """ response = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.2, ) content = response.choices[0].message.content.strip() try: return json.loads(content) except json.JSONDecodeError: return []

抽取后，可以把结果写入VectorMemoryStore。需要注意的是，如果 LLM 返回空，说明该轮对话没有值得长期保存的信息。不要强行把每句都写入长期记忆。

3.5 检索与上下文组装：在请求前把记忆灌进 Prompt

最后实现一个MemoryService，把短期消息和长期记忆组合起来，形成 Prompt 上下文。

from redis_store import RedisSessionStore from vector_store import VectorMemoryStore class MemoryService: def __init__(self, session_store: RedisSessionStore, memory_store: VectorMemoryStore, extractor=None): self.session_store = session_store self.memory_store = memory_store self.extractor = extractor def record_dialog(self, session_id: str, user_message: str, assistant_message: str, user_id: str) -> None: self.session_store.add_message(session_id, "user", user_message) self.session_store.add_message(session_id, "assistant", assistant_message) if self.extractor: memories = self.extractor.extract(user_message, assistant_message) for memory in memories: self.memory_store.add_memory( user_id=user_id, session_id=session_id, content=memory["content"], memory_type=memory["type"], importance_score=memory["importance"], ) def build_context(self, user_id: str, session_id: str, query: str) -> dict: recent_messages = self.session_store.get_recent_messages(session_id, last_n=10) long_term_memories = self.memory_store.search_topk(user_id, query, top_k=3) memory_text = "\n".join( [f"- {item['content']}" for item in long_term_memories] ) history_text = "\n".join( [f"{item['role']}: {item['content']}" for item in recent_messages] ) return { "long_term_memory": memory_text, "recent_messages": history_text, }

这个模块可以直接整合进 FastAPI 接口。学习环境中先通过命令行脚本验证，生产环境再接入消息队列和更复杂的服务治理。

4. 用 MCP 把记忆能力开放给 Agent

4.1 MCP 为什么适合记忆服务

MCP，也就是 Model Context Protocol，是 Agent 与外部工具、数据源之间的标准化协议。它解决的问题是：每个 Agent 都要各自实现一套记忆读写接口，导致重复造轮子。通过 MCP Server，记忆能力可以被统一暴露成 tools，任何支持 MCP 的 Agent 都能直接调用。

Agent Memory 和 MCP 的关系是：MCP 是接入标准，记忆服务是业务能力。把记忆封装成 MCP 工具后，Agent 不需要关心底层是 Redis、SQLite 还是向量数据库。

4.2 实现一个最小的 MCP Server

完整 MCP Server 建议使用官方 SDK。这里用一个简化版 FastAPI 接口演示协议思想：通过 JSON-RPC 方式暴露tools/list和tools/call。

from fastapi import FastAPI, Request from memory_service import MemoryService app = FastAPI() memory_service: MemoryService = None TOOLS = [ { "name": "memory_retrieve", "description": "检索用户的长期记忆", "inputSchema": { "type": "object", "properties": { "user_id": {"type": "string"}, "query": {"type": "string"} }, "required": ["user_id", "query"] } }, { "name": "memory_store", "description": "写入一条长期记忆", "inputSchema": { "type": "object", "properties": { "user_id": {"type": "string"}, "session_id": {"type": "string"}, "content": {"type": "string"}, "memory_type": {"type": "string"} }, "required": ["user_id", "content"] } } ] @app.post("/mcp") async def mcp_endpoint(request: Request): payload = await request.json() method = payload.get("method") if method == "tools/list": return {"jsonrpc": "2.0", "id": payload.get("id"), "result": {"tools": TOOLS}} if method == "tools/call": params = payload.get("params", {}) tool_name = params.get("name") arguments = params.get("arguments", {}) if tool_name == "memory_retrieve": result = memory_service.memory_store.search_topk( user_id=arguments["user_id"], query=arguments["query"], top_k=3 ) return {"jsonrpc": "2.0", "id": payload.get("id"), "result": result} if tool_name == "memory_store": memory_id = memory_service.memory_store.add_memory( user_id=arguments["user_id"], session_id=arguments.get("session_id", ""), content=arguments["content"], memory_type=arguments.get("memory_type", "fact"), importance_score=arguments.get("importance_score", 0.8) ) return {"jsonrpc": "2.0", "id": payload.get("id"), "result": {"memory_id": memory_id}} return {"jsonrpc": "2.0", "id": payload.get("id"), "error": {"code": -32601, "message": "Method not found"}}

这段代码用于理解 MCP 的调用形态。生产接入时，建议使用 MCP 官方 SDK 生成的 Server，它会自动处理初始化握手、错误码、stdio 或 HTTP 传输等细节。

4.3 在 Agent 侧调用记忆工具

Agent 侧发起记忆检索的 JSON-RPC 请求如下：

{ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "memory_retrieve", "arguments": { "user_id": "user_1001", "query": "用户对部署环境有什么要求" } } }

正常响应会返回长期记忆候选列表：

{ "jsonrpc": "2.0", "id": 1, "result": [ { "content": "用户要求私有化部署，不接受公网云服务", "memory_type": "preference", "score": 0.82 } ] }

在 Agent 的工具调用循环里，通常先做一次记忆检索，把结果注入到 Prompt，再让 LLM 生成最终回复。这样可以避免模型靠猜回答用户私有化问题。

4.4 MCP 与 RAG 的分工

MCP 适合暴露所有 Agent 能力，RAG 是其中一种检索增强手段。Agent Memory 和知识库 RAG 要区分开。

两者可以共存。RAG 负责外部知识，Agent Memory 负责用户个性化记忆，MCP 把两个服务都暴露给 Agent。

5. 从单路向量检索到 Agentic RAG：让记忆检索更可靠

5.1 单路向量检索的坑

长期记忆只做向量检索，在真实业务中会遇到几个明显问题。

第一，用户 query 往往很短，比如“上次那个事”，向量相似度会召回很多不相关记忆。第二，语义相似不等于事实正确，可能找到“用户不喜欢邮件”但丢失“用户要求私有化部署”这条关键记忆。第三，向量检索没有时间概念，三年前的旧偏好可能比上周的新偏好分数更高。第四，不相关记忆进入上下文后，会污染模型生成结果。

5.2 多路召回与时间衰减

可靠方案是做多路召回。常见思路包括：

融合打分可以这样理解：

score = alpha * vector_score + beta * keyword_score + gamma * time_recency_score + delta * importance_score

参数需要通过测试集调，不要拍脑袋。更重要的是加一个最低阈值，分数不够就不进上下文，宁可什么都不给模型，也不要给出无关记忆。

5.3 从 RAG 到 Agentic RAG

Agentic RAG 强调让 Agent 自己决定是否需要检索、检索什么、是否要二次验证。应用到记忆场景，可以用一个简单决策函数：

def decide_source(query: str) -> str: memory_keywords = ["我上次", "我说过", "偏好", "要求", "记不记得"] knowledge_keywords = ["文档", "流程", "规范", "FAQ", "手册"] has_memory = any(k in query for k in memory_keywords) has_knowledge = any(k in query for k in knowledge_keywords) if has_memory and has_knowledge: return "both" if has_memory: return "memory" if has_knowledge: return "knowledge" return "memory_first"

这个函数只是一个简化示例。生产环境可以做两层路由：先用 LLM 判断用户意图，再调用不同检索器。目标是一样的：不要每次请求都把所有记忆和知识文档灌给模型。

5.4 记忆的合并、更新与遗忘

长期记忆不是只增不改。用户可能忘记之前说过什么，或者直接推翻之前的偏好。这时候需要覆盖和失效机制。

可以给记忆增加 status 字段，初始为 active。当新抽取的记忆与旧记忆内容冲突时，把旧记忆置为 invalidated，新记忆置为 active。比如用户先说“用邮件通知”，后来改成“改用企业微信”。检索时只返回 active 的记忆。

对于长期未访问且重要度低的记忆，可以定期降级为 archived。这样既能控制存储成本，也能减少上下文噪声。生产环境可以加一个定时任务，扫描 last_access_at 超过 90 天且 importance_score 低于 0.5 的记忆。

注意：检索不到记忆，不一定是系统坏了，也可能是记忆已经被正确失效。排查时要先看记忆状态。

6. 常见问题与排查路径

6.1 检索不到用户记忆

现象：用户明明之前说过某个偏好，下次提问 Agent 完全没反应。

可能原因很多，按顺序排查：

1. user_id 或 session_id 传错，导致检索到别的用户。
2. 短期 Redis key 已过期，且长期记忆没有写入成功。
3. 记忆抽取阶段返回空，原始消息没有被转成长期记忆。
4. 向量检索 min_score 设置过高，相关记忆被过滤。
5. embedding 模型版本不一致，导致相似度分数失真。

建议在接入层打印检索日志，包含 memory_id、score、status，方便排查。

6.2 上下文被噪声记忆污染

现象：用户要求简洁回复，Agent 却总是提两年前的无关节。

这种情况通常是 TopK 太大、没有时间加权、没有重要性过滤。处理方案是：

• 把 TopK 从 5 降到 3。
• 设置最低相似度 0.4，低于 0.4 不注入。
• 在最终排序里引入时间衰减和 importance_score。
• 对同一个用户会话做记忆缓存，避免重复检索。

6.3 MCP 调用不通或超时

MCP 服务接入不稳定，优先检查协议通信层。

生产环境建议给 MCP Server 增加超时重试和熔断，不要让记忆服务故障拖垮主对话流程。

6.4 Redis 会话记录中断

如果 Redis 重启后短期消息丢失，不一定是使用错误，也可能是持久化策略没有配置。

短期记忆可以容忍一定程度丢失，但长期记忆必须可靠保存。不要把重要记忆只存在 Redis。

7. 生产环境最佳实践与可复用清单

7.1 数据安全与隐私

Agent Memory 涉及用户偏好和历史行为，属于敏感数据。生产环境至少要做到：

• 不存储密码、令牌、密钥等敏感凭证。
• 对用户 ID 和会话 ID 做统一脱敏或编码。
• 存储层开启加密，权限按租户和用户隔离。
• 提供用户主动删除记忆的接口，支持“忘记我”的合规要求。
• 对抽取出的记忆内容做敏感信息过滤，避免把身份证、手机号等误存。

删除接口不能只做逻辑删除，还要清理向量和文件缓存。

7.2 上线前检查清单

发布到生产前，可以对照这份清单逐项检查：

每项都是可以落地的动作，不要停留在“注意数据安全”这种口号上。

7.3 从最小模块到企业级：下一步扩展建议

本文的代码是教学级的最小闭环，生产环境可以在以下方向继续演进。

第一，把 SQLite 替换为 pgvector、Milvus 或 Elasticsearch，满足大规模向量检索。第二，引入消息队列，把记忆写入改成异步事件，降低对话延迟。第三，加入 Rerank 模型，对召回结果做精排。第四，用官方 MCP SDK 替换简化版 HTTP 端点，提升协议兼容性。第五，建立记忆效果评估集，统计记忆命中率、用户重复提问率和信息遗忘率。

评估不能只看单条对话是否漂亮，还要看跨会话用户是否明显减少重复描述。最直接的做法，是为每个用户建立小样本测试集：固定几个问题，观察 Agent 在第二次、第三次提问时，能否准确复用第一次提供的信息。

Agent Memory 的难点从来不是“选一个大模型”，而是如何把状态变成资产，再把资产安全、精准地放回上下文里。先从本文的最小模块开始跑通，再逐步加上多路召回、Agentic RAG 和企业级存储，这条路径比试图靠超大 Context 堆出记忆要靠谱得多。