最近在给一个企业级 Agent 项目做记忆系统改造时踩了不少坑也把“记忆”这件事从最初理解的“把历史消息塞进 Prompt”慢慢梳理成了分层、持久化、有治理策略的完整架构。这篇文章就围绕“从 Context 到 Long-term Me”这条主线结合 LangChain、LangGraph 和 DeepAgent 框架完整拆解企业级 Agent 记忆系统的设计思路、落地代码和工程治理方案。不管你是准备入门 Agent 开发还是已经在做多轮对话、智能客服、Copilot 类应用这篇文章都值得收藏。我会尽量用工程化的语言把概念、代码、坑点一次讲透。1. Agent 记忆系统背景与核心概念1.1 一次真实的“失忆”事故先看一个很常见的场景。用户在第一轮对话里说我是上海的后端开发主要用 Java 和 Kubernetes 部署。Agent 也正常回复了。但用户第二天再开一个会话问帮我看看这段 Deployment 配置有什么问题Agent 完全不记得用户的技术背景给了通用答案。用户只能重新解释一遍。这只是“失忆”的初级版本。更严重的是会话历史越来越长每次都把全部消息塞进 PromptToken 成本线性增长。多用户共用一套 Agent记忆没有隔离A 用户的偏好被 B 用户的会话检索到。用户明确表达过“不要用 xx 方案”Agent 却在后续对话里反复推荐。这些问题本质上都是同一个大模型本身是无状态的但业务要求 Agent 在长生命周期内保持身份和上下文的一致性。1.2 什么是 Agent 记忆系统Agent 记忆系统可以理解为给 Agent 装载一套跨请求、跨会话的信息基础设施。它负责存储、更新、检索和淘汰与用户、任务、环境相关的信息让 Agent 从“一次性问答工具”进化成“越来越懂用户的长期助手”。从工程角度我们可以把记忆分成几个层次记忆类型生命周期典型载体核心问题Context单次请求Prompt 窗口容量有限、成本高、随用随丢Working Memory单次会话内State Checkpointer需要持久化、需要控制长度Long-term Memory跨会话向量库 / 数据库写入、检索、去重、更新User Profile持续演进画像服务闭环更新、权限控制这里要特别区分两个概念Context是“窗口式的记忆”它只存在于一次请求中模型看完就丢。Long-term Me是“身份式的记忆”它跨越会话存在是 Agent 在时间维度上对用户持续建模的结果。如果一个 Agent 只能靠 Context 工作那它本质上只是“一个没有记忆的函数”。而“Long-term Me”的下一层意思是Agent 不仅要记住用户说了什么还要能整理出用户的身份、偏好、决策和里程碑形成可复用的长期记忆。1.3 从 Context 到 Long-term Me 的演进路线我习惯把记忆系统分成四个演进阶段阶段一纯 Context所有信息都塞进系统提示词或用户消息简单直接但受限于 Token 上限也无法跨会话。阶段二会话级 Message HistoryLangChain 中的ChatMessageHistory、InMemoryChatMessageHistory可以保存消息记录解决单次会话内的连续性。阶段三持久化工作记忆用 LangGraph 的 State 和 Checkpointer 把会话状态持久化到 SQLite、Postgres会话中断后可以恢复。阶段四长期记忆 用户画像从对话中抽取关键事实写入 Long-term Memory 存储并在下一次会话中检索注入 Prompt。此时 Agent 才真正开始形成“Long-term Me”。这篇文章的核心实战案例就是带你从阶段三走向阶段四。2. 环境准备与依赖选型2.1 运行环境本文示例代码使用 Python 3.10建议使用 3.11 或 3.12。操作系统不限Windows、macOS、Linux 都可以。你需要准备一个可以访问的大模型 API本文以 OpenAI 兼容接口为例。如果没有 OpenAI Key也可以使用国内支持 OpenAI 协议的大模型服务只需要修改base_url和model即可。一个虚拟环境推荐使用venv或conda创建隔离环境。2.2 安装依赖pip install langchain langchain-openai langgraph langgraph-checkpoint-sqlite openai简单说明每个包的作用langchain提供模型封装、消息抽象、Prompt 模板等基础组件。langchain-openaiLangChain 对 OpenAI 兼容接口的适配器。langgraph本文的核心编排框架负责状态管理和流程控制。langgraph-checkpoint-sqliteLangGraph 的 SQLite 持久化后端用于保存会话状态。openaiOpenAI 官方 Python SDK作为底层调用依赖。版本方面建议安装当前 PyPI 的最新稳定版本。LangChain 和 LangGraph 迭代较快不同小版本接口可能会有微调但本文使用的核心 API 在 0.2 和 0.3 版本下都可以运行。2.3 项目结构我们用一个最小但完整的工程来演示agent_memory/ ├── main.py # LangGraph 图定义与运行入口 ├── memory_store.py # 长期记忆存储层 ├── memory_extract.py # 记忆抽取与检索 └── requirements.txt这个结构划分得很干净图编排、存储、抽取逻辑各自独立方便后续扩展。2.4 LangChain 和 LangGraph 到底有什么区别这是很多新手常问的问题。简单来说LangChain 更像是一个“零件库”提供模型封装、消息抽象、文档加载、Prompt 模板、输出解析器。LangGraph 更像是一个“流水线引擎”它基于状态图模型允许你定义节点、边、条件路由、循环、持久化状态。如果你的需求是“调用一次 LLM 并处理结果”用 LangChain 就够了。但如果你需要多轮对话、条件分支、循环、人工审批、中断恢复、跨会话状态LangGraph 是更合适的编排层。有人担心 LangChain 过时了其实并没有。LangGraph 仍然依赖 LangChain 提供的基础组件两者是互补关系。在记忆系统场景里LangChain 提供模型调用和消息抽象LangGraph 负责状态流转和持久化配合很自然。3. LangChain 与 LangGraph 的记忆机制拆解3.1 LangChain 中的消息历史LangChain 提供了多种消息历史实现例如InMemoryChatMessageHistory、FileChatMessageHistory、RedisChatMessageHistory等。先看一个最简单的内存版本from langchain_core.chat_history import InMemoryChatMessageHistory history InMemoryChatMessageHistory() history.add_user_message(你好我是上海的后端开发) history.add_ai_message(你好我记住了) print(history.messages)输出[HumanMessage(content你好我是上海的后端开发), AIMessage(content你好我记住了)]这种方式适合聊天机器人简单的会话保存但有几个问题它只是保存在内存中重启进程就丢失。它是“平铺”的消息列表没有做消息摘要、裁剪和重要性评估。它无法区分哪些信息值得长期保存哪些只是寒暄。所以在企业级 Agent 记忆系统里Message History 只能作为最底层的“会话日志”而不是完整的记忆方案。3.2 LangGraph 的 State 与 ReducerLangGraph 的核心是StateGraph。它维护一个全局状态对象每个节点都可以读取状态并返回状态更新。记忆的本质其实就藏在 State 的设计里。看一个典型的状态定义from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str retrieved_memories: list这里有两个关键点messages是一个列表通过add_messages这个 reducer 做更新。默认情况下节点返回一个字典时LangGraph 会用新值覆盖旧字段。但messages字段使用add_messagesreducer 后新消息会被追加到旧消息列表后面而不是直接覆盖。也就是说LangGraph 的 State 天然就是一个“短期工作记忆”。每一轮对话产生的消息都会累积在 State 中这就是 Agent 在单次会话中的上下文来源。3.3 Checkpointer把工作记忆持久化State 默认保存在内存中进程重启后就会丢失。要让工作记忆持久化我们需要使用 Checkpointer。SQLite 版本的使用方式如下from langgraph.checkpoint.sqlite import SqliteSaver checkpointer SqliteSaver.from_conn_string(langgraph_checkpoint.db) graph builder.compile(checkpointercheckpointer)在调用图时通过thread_id区分不同的会话config {configurable: {thread_id: session_xxx}} result graph.invoke(input_data, configconfig)同一个thread_id的多次invoke会共享同一个 State也就是说会话历史不会丢。这就是 LangGraph 对工作记忆的持久化方案。3.4 为什么纯 Context 拼接不可持续很多初学 Agent 的人会把“记忆”等同于“把历史消息全部塞进 Prompt”。小规模 Demo 可以跑通但一旦进入生产环境问题会逐渐暴露对话轮数多了以后Context 长度受限。所有历史消息都消耗 Token成本随对话轮数线性增长。无关的历史信息会干扰模型判断降低回答质量。所以记忆系统必须做分层尽量用“结构化的长期记忆”替代“无差别的历史消息”让模型每次只读取最相关的少量信息而不是把所有历史都搬进上下文。4. 完整实战用 LangGraph 构建带长期记忆的 Agent这一节是全文核心。我们将实现一个带长期记忆的 Agent它的流程如下用户输入retrieve节点按user_id和当前问题检索长期记忆。agent节点把长期记忆注入 System Prompt和历史消息一起交给模型。extract节点从最新对话中抽取值得长期记忆的事实写入记忆库。这样的设计可以保证同一会话内Converation History 由 LangGraph 的 State 管理跨会话后Agent 靠长期记忆库重新“想起”用户。4.1 长期记忆存储层 memory_store.py我们使用 SQLite 作为长期记忆的落库方式。每一个记忆条目都包含user_id用户隔离维度。content记忆内容。kind类型比如fact、preference、decision、milestone。importance重要性分数影响检索排序。created_at/updated_at时间戳用于生命周期管理。# 文件路径agent_memory/memory_store.py import os import sqlite3 from datetime import datetime DB_PATH os.path.join(os.path.dirname(__file__), agent_memory.db) def get_connection(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute( CREATE TABLE IF NOT EXISTS long_term_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, kind TEXT DEFAULT fact, importance REAL DEFAULT 0.5, created_at TEXT DEFAULT (datetime(now)), updated_at TEXT DEFAULT (datetime(now)) ) ) conn.commit() return conn def add_memory(user_id: str, content: str, kind: str fact, importance: float 0.5): conn get_connection() conn.execute( INSERT INTO long_term_memory(user_id, content, kind, importance) VALUES (?, ?, ?, ?) , (user_id, content, kind, importance), ) conn.commit() conn.close() def get_user_memories(user_id: str): conn get_connection() rows conn.execute( SELECT * FROM long_term_memory WHERE user_id ? ORDER BY updated_at DESC LIMIT 200 , (user_id,), ).fetchall() conn.close() return [dict(row) for row in rows] def search_memories(user_id: str, query: str, limit: int 5): 先用 user_id 过滤再做简单的关键词评分。 生产环境建议替换为向量检索 Rerank。 memories get_user_memories(user_id) if not memories: return [] if not query: return memories[:limit] scored [] query_tokens query.lower().split() for m in memories: content_lower m[content].lower() score 0.0 for token in query_tokens: if token in content_lower: score 1.0 final_score score * m[importance] scored.append((final_score, m)) scored.sort(keylambda x: x[0], reverseTrue) matched [m for s, m in scored if s 0] if matched: return matched[:limit] # 兜底没有匹配时返回最新记忆避免长期记忆被“遗忘” return memories[:limit]这段代码有几个设计点值得注意使用user_id过滤是记忆隔离的基础。search_memories先用关键词评分再用重要性加权。这是最简单的检索策略便于理解。生产环境可以替换为 Embedding 向量检索。查询结果限制为 5 条控制注入 Prompt 的 Token 开销。4.2 记忆抽取与检索 memory_extract.py为了让 Agent 能“记住”用户我们需要从对话中抽取结构化记忆。这里使用 LLM 作为抽取引擎让模型把一段对话转成 JSON 记忆条目。# 文件路径agent_memory/memory_extract.py import json import os from langchain_core.messages import AIMessage, HumanMessage from langchain_openai import ChatOpenAI from memory_store import add_memory EXTRACT_PROMPT 你是长期记忆抽取器。请从下面的对话中提取值得长期记住的信息例如 - 用户的身份背景、职业、所在城市 - 用户的技术栈、项目偏好、工作习惯 - 用户明确表达过的偏好、禁忌、目标 - 用户提到的关键日期、约定、决策 只输出 JSON 数组不要输出任何解释。每个元素包含以下字段 - content: 记忆内容 - kind: fact / preference / decision / milestone - importance: 0.0 到 1.0 的浮点数 对话内容 {dialogues} def _format_recent_messages(messages, n: int 6) - str: lines [] for m in messages[-n:]: role User if isinstance(m, HumanMessage) else Assistant text getattr(m, content, ) or lines.append(f{role}: {text}) return \n.join(lines) def extract_memories_from_state(user_id: str, messages, llmNone, min_importance: float 0.6): if not messages: return llm llm or ChatOpenAI( modelos.getenv(OPENAI_MODEL, gpt-4o-mini), temperature0, ) dialogues _format_recent_messages(messages, 6) try: resp llm.invoke(EXTRACT_PROMPT.format(dialoguesdialogues)) items json.loads(resp.content.strip()) if not isinstance(items, list): return except Exception: return for item in items: content (item.get(content) or ).strip() if not content: continue importance float(item.get(importance, 0.5)) if importance min_importance: continue add_memory( user_iduser_id, contentcontent, kinditem.get(kind, fact), importanceimportance, )这里的抽取逻辑并不复杂关键是两点只抽取最近 6 条消息避免每轮都重复处理全部历史。设置min_importance阈值防止把无关紧要的对话内容写入长期记忆。4.3 图编排与核心代码 main.py下面是最核心的 LangGraph 定义。我们把记忆检索、Agent 回复、记忆抽取三个逻辑封装成三个节点。# 文件路径agent_memory/main.py import os import uuid from typing import Annotated, TypedDict from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import END, START, StateGraph from langgraph.graph.message import add_messages import memory_extract import memory_store OPENAI_MODEL os.getenv(OPENAI_MODEL, gpt-4o-mini) class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str retrieved_memories: list def build_system_prompt(state: AgentState) - str: memories state.get(retrieved_memories) or [] if memories: memory_text \n.join( f- [{m[kind]}] {m[content]} for m in memories ) else: memory_text 暂无相关长期记忆。 return f你是一个企业级智能助理。以下是当前用户的长期记忆回答时自然引用不要机械复述。 长期记忆 {memory_text} def retrieve_memories_node(state: AgentState) - dict: user_id state.get(user_id, default) query if state.get(messages): last_msg state[messages][-1] query getattr(last_msg, content, ) or memories memory_store.search_memories(user_id, query, limit5) return {retrieved_memories: memories} def agent_node(state: AgentState) - dict: llm ChatOpenAI(modelOPENAI_MODEL, temperature0.3) system_msg SystemMessage(contentbuild_system_prompt(state)) response llm.invoke([system_msg] state[messages]) return {messages: [response]} def extract_memory_node(state: AgentState) - dict: user_id state.get(user_id, default) memory_extract.extract_memories_from_state(user_id, state.get(messages) or []) return {} def build_agent_graph(): gb StateGraph(AgentState) gb.add_node(retrieve, retrieve_memories_node) gb.add_node(agent, agent_node) gb.add_node(extract, extract_memory_node) gb.add_edge(START, retrieve) gb.add_edge(retrieve, agent) gb.add_edge(agent, extract) gb.add_edge(extract, END) checkpointer SqliteSaver.from_conn_string(langgraph_checkpoint.db)
尧图网络