企业级Agent记忆系统架构：从Context到Long-term Memory

企业级 Agent 开发做到后面，最让人头疼的不是模型选型，也不是 Prompt 写不好，而是上下文窗口炸了。API 报错、请求超时、回答丢失前文信息、多轮任务跑一半直接挂掉，这些问题背后几乎都指向同一个模块：记忆系统没做治理。

这次我们来看一套企业级 Agent 记忆系统架构，重点拆解从 Context 到 Long-term Memory 的工程化路径，以及如何使用 LangChain、LangGraph 和 DeepAgent 三个框架把记忆层落地。文章会覆盖框架选型、环境准备、状态图设计、记忆存取、API 治理、批量任务和常见报错排查。

本文适合正在做 Agent 开发的工程师、想把原型工具转成企业服务的架构师，以及准备从 LangChain 迁移到 LangGraph 或 DeepAgent 的技术团队。先给结论：只要你的 Agent 要处理多轮对话、长文档、复杂工具链，记忆系统就不能继续依赖“一把梭塞进上下文”。

1. 核心能力速览

先把几个关键问题回答清楚：这套架构能做什么，需要什么环境，能跑到什么程度。

这里需要说清楚：LangChain、LangGraph、DeepAgent 不是“三选一”的关系，而是一套分层的组合。LangChain 提供模型调用和工具封装，LangGraph 负责把流程变成可控制的图结构，DeepAgent 则偏向企业级 Agent 的整体治理。实际工程中，你可以用 LangGraph 做状态控制，同时用 LangChain 组件作为节点内部实现，再以 DeepAgent 的思路做 Session 管理和权限隔离。

2. 记忆系统解决的问题与企业级使用边界

2.1 上下文窗口为什么总是不够用

很多团队在 Agent 开发初期只有一个朴素做法：把所有对话历史、检索结果、工具返回内容全部塞进messages里。小规模测试没问题，一旦进入生产环境，就会出现类似下面这些报错：

api error: 400 this model's maximum context length is 1048576 tokens. however...

codex ran out of room in the model's context window. start a new thread or c...

context is too large and auto-compaction could not recover this turn.

这些报错的本质是：上下文窗口是硬资源，模型对 Token 数量有明确上限。超过上限，服务端直接拒绝请求，或者自动压缩失败后被迫丢弃信息。企业级 Agent 面临的问题更严重——多用户并发、多次工具调用、长文档注入、多轮任务链条，都会快速耗尽上下文。

2.2 从 Context 到 Long-term Memory 的演进路径

记忆系统需要分三层设计：

• 临时记忆：当前对话轮次内的消息、工具返回、中间推理结果，放在内存或状态对象中。
• 工作记忆：整个会话周期内需要保留的关键信息，比如用户偏好、任务目标、已完成步骤，以结构化状态保存。
• 长期记忆：跨会话的信息，比如用户历史行为、项目背景知识、领域规则，持久化到向量数据库或关系型数据库。

这三层不是互相替代，而是过滤和升华的关系。每次对话结束时，应该有一个“记忆提炼”过程，把工作记忆中的高价值信息写入长期记忆，同时把低价值信息丢弃。

2.3 适用场景与合规边界

这套架构适合：

• 需要多轮交互的客服 Agent。
• 需要跨 Session 记住用户偏好和历史的推荐 Agent。
• 需要处理长文档、长对话的知识库助手。
• 需要并行执行多个子任务的复杂工作流。
• 需要对 Agent 调用过程进行审计和回溯的企业系统。

使用时必须注意：

• 长期记忆中存储的用户数据涉及隐私合规，需要明确告知用户并取得必要授权。
• 涉及人脸、声音、身份信息、企业敏感数据时，必须做好脱敏和权限控制。
• 不要把未经授权的第三方版权内容写入长期记忆库用于商用。
• 记忆系统的读写操作要留有日志，方便审计和追溯。

3. 环境准备与前置条件

在部署这套框架之前，建议先确认以下环境项：

3.1 基础环境检查清单

3.2 虚拟环境搭建示例

# 创建项目目录 mkdir agent-memory-system && cd agent-memory-system # 创建 Python 虚拟环境 python3 -m venv venv # 激活虚拟环境 source venv/bin/activate # Windows 下使用 venv\Scripts\activate

然后安装基础依赖。这里注意，LangChain 和 LangGraph 的版本选择要匹配，不能随意装最新版。

pip install --upgrade pip pip install langchain langchain-openai langgraph pip install chromadb pip install fastapi uvicorn pip install python-dotenv

如果你使用 DeepAgent 框架，需要根据对应文档安装。由于该框架在不同团队的封装程度不同，建议以实际项目提供的安装方式为准。

3.3 环境变量配置

在项目根目录创建.env文件，保存模型 API Key。

# .env 示例 OPENAI_API_KEY=sk-xxxxxxxxxxxx # 如果使用其他兼容 API，可配置自定义 base_url OPENAI_API_BASE=https://your-api-endpoint

加载方式使用 python-dotenv：

from dotenv import load_dotenv load_dotenv()

4. Agent 记忆系统分层设计与 LangGraph 状态图实现

4.1 为什么选择 LangGraph 而不是纯 LangChain

LangChain 早期版本提供了ConversationBufferMemory等记忆组件，但它的流程控制能力相对有限。Agent 任务一旦涉及条件分支、循环、并行执行、子图嵌套，LangChain 的 Chain 结构会显得不够灵活。

LangGraph 的核心改进是：把 Agent 流程抽象成有向图，节点（Node）执行具体操作，边（Edge）控制流转逻辑，状态（State）在不同节点之间传递并更新。这种设计天然适合实现记忆系统：

• State 本身就是工作记忆的载体。
• 条件边可以判断是否需要压缩上下文。
• 子图可以把不同子任务隔离成独立的记忆空间。
• 并行分支可以同时处理多个上下文片段。

简单来说：LangChain 负责“用什么工具”，LangGraph 负责“流程怎么走”，DeepAgent 负责“整个 Agent 怎么治理”。

4.2 定义状态对象

在记忆系统中，State 设计是最重要的一步。下面给出一个具体的状态定义示例：

from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): # 核心消息列表，使用 add_messages 自动追加 messages: Annotated[List[dict], add_messages] # 当前会话 ID session_id: str # 用户 ID user_id: str # 任务目标 task_goal: str # 已完成步骤记录 completed_steps: List[str] # 当前步骤 current_step: str # 是否需要压缩上下文的标志 needs_compaction: bool # 长期记忆检索结果 long_term_memory: str

这里的关键是Annotated[List[dict], add_messages]，它告诉 LangGraph：每次节点返回新的消息时，自动追加到原有消息列表，而不是覆盖。

4.3 构建基础图结构：从输入到长期记忆检索

有了 State 之后，我们来构建一个包含记忆读取的 LangGraph 图。这个图包含四个节点：

1. 读取长期记忆。
2. 构建 Prompt。
3. 调用模型。
4. 提炼并写入长期记忆。

from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI # 初始化模型 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 节点1：读取长期记忆 def retrieve_memory(state: AgentState): # 从向量库中检索与当前任务相关的历史记忆 # 这里假设有一个 memory_store 对象 memory_context = memory_store.search(state["task_goal"]) return {"long_term_memory": memory_context} # 节点2：构建消息 def build_messages(state: AgentState): system_prompt = f"你是企业级Agent助手。\n历史记忆:\n{state['long_term_memory']}" messages = [{"role": "system", "content": system_prompt}] # 保留最近 N 条消息，避免上下文超限 recent_messages = state["messages"][-20:] messages.extend(recent_messages) return {"messages": messages} # 节点3：调用模型 def call_model(state: AgentState): response = llm.invoke(state["messages"]) return {"messages": [{"role": "assistant", "content": response.content}]} # 节点4：提炼长期记忆 def extract_memory(state: AgentState): # 简单实现：将用户最后一条消息和助手回复写入长期记忆 if len(state["messages"]) >= 2: memory_store.save( session_id=state["session_id"], user_id=state["user_id"], content=state["messages"][-1] ) return {} # 构建图 graph = StateGraph(AgentState) graph.add_node("retrieve_memory", retrieve_memory) graph.add_node("build_messages", build_messages) graph.add_node("call_model", call_model) graph.add_node("extract_memory", extract_memory) graph.add_edge(START, "retrieve_memory") graph.add_edge("retrieve_memory", "build_messages") graph.add_edge("build_messages", "call_model") graph.add_edge("call_model", "extract_memory") graph.add_edge("extract_memory", END) app = graph.compile()

从上面的流程可以看到，每次调用 Agent 时，会先读取长期记忆，再构建 Prompt，这就在源头减少了无效信息的注入。

4.4 条件路由：上下文超限自动决策

这是 LangGraph 最有价值的能力之一。我们可以增加一个判断节点：如果当前上下文 Token 数超过阈值，就进入压缩节点；否则正常调用模型。这正好解决真实业务中最常见的context is too large报错。

from langgraph.graph import ConditionalEdge def should_compact(state: AgentState) -> str: # 粗略估算 token 数：按字符数 / 4 total_chars = sum(len(m["content"]) for m in state["messages"]) estimated_tokens = total_chars / 4 if estimated_tokens > 3000: return "compact" return "call_model" graph.add_conditional_edges( "build_messages", should_compact, { "compact": "compact_memory", "call_model": "call_model" } )

条件路由的核心价值在于：不需要每次都在代码里手动判断上下文长度，LangGraph 会在状态流转过程中自动选择路径。

5. 长期记忆存储设计与 DeepAgent 治理思路

5.1 长期记忆的存储选型

长期记忆存储方案需要满足三个要求：写入快、检索准、容易扩展。推荐以“向量数据库 + 关系型数据库”组合方式实现：

• 向量数据库：用于语义检索，存储对话摘要、知识片段。
• 关系型数据库：用于结构化查询，存储用户 ID、Session ID、时间戳、消息原文。

向量库选择上，小规模项目可以用 Chroma，企业级生产环境更建议使用 Milvus 或 Qdrant。如果团队已有 PostgreSQL 基础设施，也可以使用 pgvector 扩展，减少组件数量。

5.2 简单记忆存储类实现

import chromadb class MemoryStore: def __init__(self, collection_name="agent_memory"): self.client = chromadb.Client() self.collection = self.client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"} ) def save(self, session_id: str, user_id: str, content: str): # 实际项目中需要将 content 转为向量 # 这里使用简单 id 和 embedding 占位 self.collection.add( documents=[content], ids=[f"{session_id}-{user_id}-{len(self.collection.get()['ids'])}"], metadatas=[{"session_id": session_id, "user_id": user_id}] ) def search(self, query: str, n_results: int = 5): # 实际项目需要使用 embedding model results = self.collection.query( query_texts=[query], n_results=n_results ) return "\n".join(results["documents"][0])

这里是一个最小实现。生产级系统需要考虑字段裁剪批量写入用upsert、定期清理过期记忆、权限隔离等。

5.3 DeepAgent 框架在企业级治理中的定位

DeepAgent 框架本身并不是 LangChain 的替代品，而是面向企业 Agent 场景的整体治理框架。它的核心关注点包括：

• 多 Agent 协作时的任务分发。
• 记忆系统的统一读写接口。
• 权限模型与数据隔离。
• 调用链路的监控与日志。
• 批量任务的调度和限流。

在工程项目中，建议以“LangGraph 负责流程内状态流转，DeepAgent 负责全局 Agent 治理”的方式来组合使用，避免所有逻辑都堆在同一个图里。

6. 接口 API 与批量任务治理

6.1 FastAPI 封装 Agent 接口

不管本地测试还是企业集成，建议把 Agent 服务封装成 API。这样前端、后端、自动化脚本都可以统一调用。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI(title="Agent Memory Service") class ChatRequest(BaseModel): session_id: str user_id: str message: str task_goal: str = "" class ChatResponse(BaseModel): reply: str session_id: str @app.post("/api/chat", response_model=ChatResponse) async def chat(request: ChatRequest): try: initial_state = { "messages": [{"role": "user", "content": request.message}], "session_id": request.session_id, "user_id": request.user_id, "task_goal": request.task_goal, "completed_steps": [], "current_step": "start", "needs_compaction": False, "long_term_memory": "" } result = app.invoke(initial_state) last_message = result["messages"][-1] return ChatResponse(reply=last_message["content"], session_id=request.session_id) except Exception as e: raise HTTPException(status_code=500, detail=str(e))

启动服务：

uvicorn main:app --host 127.0.0.1 --port 8000

6.2 Python 客户端调用示例

import requests url = "http://127.0.0.1:8000/api/chat" payload = { "session_id": "session_001", "user_id": "user_001", "message": "帮我总结上一轮对话中的关键结论", "task_goal": "多轮对话总结" } response = requests.post(url, json=payload, timeout=120) if response.status_code == 200: print(response.json()["reply"]) else: print("Error:", response.status_code, response.text)

6.3 批量任务与会话隔离

企业场景下，批量任务是一项硬需求。比如批量处理历史对话、批量生成摘要、批量分析用户反馈。这里的核心设计原则是：每个批量任务使用独立的 Session ID，任务级别日志单独记录。

import uuid from concurrent.futures import ThreadPoolExecutor def process_batch(messages_list): results = [] def process_one(item): session_id = str(uuid.uuid4()) response = requests.post( "http://127.0.0.1:8000/api/chat", json={ "session_id": session_id, "user_id": item["user_id"], "message": item["message"], "task_goal": item.get("task_goal", "") }, timeout=120 ) return response.json() with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(process_one, messages_list)) return results

批量任务需要注意：

• 并发数不能过高，否则模型 API 会限流。
• 每个任务要有独立 Session，避免数据串线。
• 失败任务要支持重试，建议使用消息队列（Redis / RabbitMQ）而不是裸线程池。
• 单条任务的 Token 消耗要控制，避免批量任务导致费用飙升。

7. 资源占用与性能观察

7.1 记忆框架本身的资源占用

LangChain、LangGraph、FastAPI 都是纯 Python 框架，框架本身的内存占用通常不高，一般在几百 MB 以内。真正的资源消耗来自两块：

• LLM 推理：如果调用云端 API，本地资源消耗很低；如果本地部署模型，显存占用取决于模型参数量和量化方式。
• 向量检索：如果使用内存型向量库（如 Chroma），向量数据会占用系统内存；如果使用 Milvus 这类独立服务，则需要单独分配资源。

建议在部署环境安装psutil和nvidia-smi监控脚本，观察 Agent 服务在推理过程中的资源变化：

pip install psutil nvidia-smi -l 2

7.2 如何降低 Token 消耗

企业级 Agent 的 Token 成本往往被低估。以下方法能有效减少 Token 消耗：

7.3 观察指标

建议在 Agent 服务中记录以下指标：

{ "session_id": "xxx", "input_tokens": 1200, "output_tokens": 350, "total_latency_ms": 2800, "memory_retrieved": True, "context_compacted": False, "model_name": "gpt-4o-mini" }

这些指标可以接入 Prometheus 或写入日志，用于后续的成本分析和性能优化。

8. LangChain / LangGraph / DeepAgent 常见问题与排查方法

关于 LangChain 和 LangGraph 的区别，这是社区里高频讨论的问题。简单概括：

• LangChain 是组件库和工具链，提供模型封装、Prompt 模板、向量库接口。
• LangGraph 是状态化编排框架，用图结构管理复杂流程。
• 两者不是替代关系，LangGraph 节点内部可以调用 LangChain 的组件。
• 如果只做简单的链式调用，LangChain 足够；如果你的 Agent 需要条件分支、循环、并行、子图，建议直接用 LangGraph。

9. 工程治理最佳实践

9.1 记忆系统的数据生命周期治理

记忆数据不是越多越好。企业级系统必须定义数据生命周期：

对话输入 -> 短期 Context -> 会话结束 -> 提炼关键信息 -> 长期记忆存储 -> 定期清理过期记忆

建议在数据库中为每条长期记忆增加expires_at字段，由定时任务定期清理。这样既能保证记忆的时效性，也能控制存储成本。

9.2 多租户与权限隔离

如果 Agent 服务面向多个团队或多个客户，长期记忆必须做租户隔离。最简单的方案是在记忆数据结构中加入tenant_id字段，所有查询和写入都强制带上租户条件：

class MemoryStore: def save(self, tenant_id: str, session_id: str, content: str): self.collection.add( documents=[content], ids=[f"{tenant_id}-{session_id}-{uuid.uuid4()}"], metadatas=[{"tenant_id": tenant_id, "session_id": session_id}] )

9.3 日志与审计

记忆系统接触的是用户输入和模型输出，在合规审计上有较高要求。建议完整记录：

• 每次记忆写入的内容摘要。
• 长期记忆检索的查询关键字。
• 会话级 Token 消耗。
• 哪个用户、哪个 Session 访问了哪些记忆。

9.4 测试策略

企业级 Agent 记忆系统不能只在 Happy Path 上测试。建议至少准备以下测试用例：

1. 短对话：验证基本问答和记忆写入。
2. 长对话：验证上下文压缩是否生效。
3. 跨会话问答：验证长期记忆是否能被检索到。
4. 上下文超限：刻意构造超长输入，验证系统不崩溃。
5. 批量并发：验证 Session 隔离是否正常。
6. 模型 API 异常：验证系统是否能降级返回。

9.5 模型与框架的适配注意

在使用 LangChain 或 LangGraph 时，不要盲目追求新版本。生产环境升级依赖前，必须做回归测试。尤其是 LangChain 0.1 到 0.2 再到后续版本，很多 API 都发生了变化。一个稳妥的做法是：在requirements.txt中锁定主版本号，并在 CI 中运行记忆模块的集成测试。

# requirements.txt 示例 langchain>=0.2,<0.4 langgraph>=0.1,<0.3 langchain-openai>=0.1,<0.3 chromadb>=0.4,<1.0 fastapi>=0.110,<1.0

10. 总结与下一步

这套架构最值得尝试的点，是把 Agent 记忆从“塞上下文”升级为“分层治理”：LangGraph 负责状态流转与条件路由，LangChain 负责模型调用和组件封装，DeepAgent 负责企业级 Session 与任务治理。三者的组合解决的核心问题，就是上下文超限和多轮任务信息丢失。

最先应该验证的功能是条件路由和上下文压缩。用一个长对话测试，观察当消息数量达到阈值时，是否正确进入压缩节点而不是直接报错。这是整套架构中收益最高、也最容易踩坑的一环。

最容易踩的坑有三个：第一，State 字段设计不完整，导致节点之间数据传递丢失；第二，向量库的 embedding 模型选择不一致，导致检索结果为空；第三，批量任务并发过高，触发了 API 限流。这三个问题在开发阶段就要提前设计好应对方案。

后续可以继续扩展的方向有：用 LangGraph 的 Subgraph 拆分更复杂的多 Agent 协作场景，把长期记忆从向量库升级为“向量库 + 知识图谱”的混合结构（GraphRAG），以及为记忆读写接口接入完整的权限体系和审计平台。掌握这套记忆系统架构之后，再做 Agent 开发时，上下文窗口超限就不会再是你第一个要担心的问题了。