企业官网建设流程全解析

做智能体开发，真正让人挠头的往往不是工具调用，也不是模型 API 崩了，而是“记忆”。模型上下文窗口再大，也只够装下当前这段对话；用户上周说过的一句话，下周一重启系统就丢了。LangChain、LangGraph 这类框架陆续把 Store 作为一层标准抽象拿出来，解决的就是这个“跨会话、跨线程、甚至跨智能体”的长期记忆问题。这篇文章我会直接拆开讲：Store 是什么、为什么它是长期记忆的底座、LangChain / LangGraph / DeepAgents 三个方向分别怎么接入，以及我在实际项目里踩过的那些坑。适合正在做智能体应用、被“模型记不住事”折磨过的人，哪怕你刚接触这些框架，跟着代码走一遍也能把记忆模块搭起来。

1. 先想明白：Store 到底在解决什么问题

1.1 会话记忆和长期记忆不是一回事

很多刚做 Agent 的人会把“记忆”和“对话历史”混在一起。对话历史是模型的上下文输入，属于短期记忆：用户发来一条消息，你把它塞进 prompt，模型回复，再把新消息追加进去。这个链条的问题在于，一旦会话结束、进程重启、或者切换了 Thread ID，这些内容就没了。就算你用了 Checkpointer 把每轮状态保存下来，它依然绑死在“对话线程”这个维度上。

长期记忆要解决的是另一个问题：用户偏好、事实信息、项目背景、历史决策，这些内容不应该因为换了一个对话而消失。比如用户上个月明确说过“报告里不要放折线图，用表格”，如果新会话里 Agent 又画了一堆图，用户会觉得自己在和金鱼聊天。

我常用一个生活化类比：会话记忆是便利贴，贴在当前任务旁边，任务结束就撕掉；长期记忆是笔记本，按目录和页码存放，任何时候翻出来都能用。Store 就是那本笔记本的底层书架。

1.2 为什么需要一个存储抽象层

你可能会说，长期记忆不就是存数据库吗？我自己建一张 memory 表，user_id、content、created_at，不就完事了？确实能存，但问题在于智能体场景里，记忆的读写路径很特殊：

• 写入频率高，单条数据小，而且往往跟着事件走：用户表达了一个偏好、Agent 完成了一个子任务、系统发现了一个隐含事实，都要写。
• 读取路径不只会按用户 ID 精确查，还要做语义检索：用户说“我不吃辣”，下次 Agent 安排餐厅时应该主动过滤辣菜，这时候得靠向量相似度，而普通数据库不会替你算 embedding。
• 底层存储会切换：开发时用 SQLite，线上可能换 Redis、云数据库，如果记忆模块直接写死某种客户端，后面想换就麻烦。

所以 LangChain / LangGraph 的做法是抽象出一层 Store：接口固定，底层可换。你的业务代码关心的是put、get、search，至于数据落在内存、SQLite 还是 Redis，对上层不可见。这层抽象让长期记忆组件可以独立演进，也让不同智能体框架之间能共用同一套记忆策略。

2. LangChain 的 Store：把记忆落进一个可插拔的层

2.1 从 BaseStore 到 InMemoryStore

LangChain 的存储抽象以BaseStore为核心，它有四个最基本的操作：get、mget、set、mset，以及用来遍历的yield_keys。注意这里的数据是 Key-Value 形态，value 默认要求是字节串，不是 Python 对象。这个设计一开始有点反直觉，但想通了就合理：只有把数据压到字节层，才能兼容 Redis、S3、文件系统这些五花八门的存储后端。

先看最基础的内存实现：

from langchain.storage import InMemoryStore store = InMemoryStore() store.mset( [ ("user_001:preferred_style", b"table_not_bars"), ("user_001:food_restriction", b"no_spicy"), ] ) result = store.mget(["user_001:preferred_style", "user_001:food_restriction"]) print(result) # [b"table_not_bars", b"no_spicy"]

这个例子看起来简单，但它揭示了 langchain 内置 Store 的一条规则：写入的值必须是 bytes。如果你想存字典、列表，请自行json.dumps(...).encode()，读出来再json.loads(...)。很多新手在这里直接传 dict，结果报类型错误，并不是 Store 的问题，而是抽象层选用了通用字节契约。

如果只是写几个 key-value，用原生字典可能更省事。但BaseStore的价值在于接口统一：你可以在本地用InMemoryStore调试，线上替换为RedisStore、SQLiteStore，业务代码不用动。我个人的建议是：即使你的项目现在只需要一种存储，也尽量通过BaseStore类型来依赖它，不要直接 new 一个具体实现到处传。这样后面换存储、加缓存、做埋点，都只需要替换一处实例。

2.2 接入 LangChain 记忆模块的正确姿势

LangChain 早期的记忆模块，比如ConversationBufferMemory，本质上只是把消息列表放进内存，解决不了跨会话长期记忆。真正常用的做法是自己写一个“记忆读写器”，内部依赖 Store，外部提供一个load_memory_variables和save_context接口。

我实际项目里常用的结构是这样的：

import json from typing import Any, Dict, List from langchain.storage import BaseStore class LongTermMemory: def __init__(self, store: BaseStore, user_id: str): self.store = store self.user_id = user_id def _key(self, memory_key: str) -> str: return f"user:{self.user_id}:memory:{memory_key}" def remember(self, memory_key: str, data: Dict[str, Any]) -> None: payload = json.dumps(data, ensure_ascii=False, default=str).encode() self.store.set(self._key(memory_key), payload) def recall(self, memory_key: str) -> Dict[str, Any] | None: raw = self.store.get(self._key(memory_key)) if raw is None: return None return json.loads(raw.decode()) store = InMemoryStore() memory = LongTermMemory(store, user_id="u_123") memory.remember("layout_pref", {"chart_style": "table", "reason": "用户希望快速对比数据"}) print(memory.recall("layout_pref"))

这里有几个关键点：

• key 设计要有层级，user:{id}:memory:{type}比裸的layout_pref安全，因为将来同一个 Store 可能服务多个用户、多个 Agent。
• 序列化要用ensure_ascii=False，避免中文被转成\uXXXX，既难读又占空间。
• default=str是为了兜底，万一带入 datetime、Decimal 这类对象时不会直接报错。

如果你需要做语义检索，只靠精确 key 不够。这时候可以叠加一个向量库，但向量库里存的不是“记忆正文”，而是一个记忆 ID，真正内容仍在 Store 中。也就是：向量负责“找到相关记忆”，Store 负责“拿出完整记忆”。两者职责分开，比把所有记忆都塞进向量库更干净。

3. LangGraph 的 Store：跨线程的长期记忆

3.1 分清 Checkpointer 与 Store

LangGraph 的持久化机制经常让人困惑，因为它同时提供了两个概念：checkpointer和store。

Checkpointer 保存的是图的运行状态，包括当前节点、待处理消息、中间变量。它和 Thread ID 绑定，作用是支持人工中断、断点续跑、时间旅行。但它本质上是“会话状态快照”，不是长期记忆。

Store 是独立的键值存储，namespace 是元组，key 是字符串，value 是可以 JSON 序列化的 dict。它不跟 Thread 绑定，而是跨线程共享。也就是说：Thread A 写入的知识，Thread B 在另一个会话里能读到。

用一个表格就能看清两者的定位：

很多人在 LangGraph 里做长期记忆，第一反应是“我把 checkpointer 存久一点不就行了”。不行。Checkpointer 依赖线程 ID，新对话一旦使用新的 Thread ID，旧状态就读不到。Store 才是那个不依赖线程的“全局记忆”。

3.2 用 BaseStore 读写跨会话记忆

LangGraph 内置了InMemoryStore，并且在图执行时会把 Store 注入到每个节点的 context 中。最常用的 API 是store.put(namespace, key, value)和store.get(namespace, key)，以及用store.search(namespace)做前缀范围内的检索。

看一个实际的例子。假设我们构建一个“用户偏好助手”，无论用户从哪个新会话进来，都能记得他之前说过的偏好：

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langgraph.store.memory import InMemoryStore from typing import TypedDict, Annotated from langgraph.graph.message import add_messages store = InMemoryStore() checkpointer = MemorySaver() class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str def remember_preference(state: AgentState): user_id = state["user_id"] last_message = state["messages"][-1] content = last_message.content if hasattr(last_message, "content") else str(last_message) # 简单假设：如果消息以“我喜欢”开头，就写入长期记忆 if content.startswith("我喜欢"): preference = content.replace("我喜欢", "").strip() store.put( (user_id, "preferences"), "last_preference", {"text": preference, "source": "user"} ) # 无论是否写入，本轮都先读取历史偏好 memories = store.search((user_id, "preferences")) history_text = "；".join(m.item["text"] for m in memories) return {"messages": [{"role": "assistant", "content": f"我记住的偏好：{history_text}"}]} graph = StateGraph(AgentState) graph.add_node("remember", remember_preference) graph.add_edge(START, "remember") graph.add_edge("remember", END) app = graph.compile(checkpointer=checkpointer, store=store)

第一次运行时传thread_id="thread_A"，写入偏好；第二次运行换一个新的thread_id="thread_B"，依然能读到那条记忆：

app.invoke( {"messages": [{"role": "user", "content": "我喜欢深色模式"}], "user_id": "u_1"}, config={"configurable": {"thread_id": "thread_A"}} ) app.invoke( {"messages": [{"role": "user", "content": "你还记得我的偏好吗？"}], "user_id": "u_1"}, config={"configurable": {"thread_id": "thread_B"}} )

这里的关键是：第一次调用写入 Store，第二次调用虽然 Thread 不同，但search结果里依然带出了偏好。用 LangGraph 做长期记忆，我建议遵循一个原则：凡是“用户说了什么、Agent 发现了什么、系统沉淀了什么”这类事实，全部写进 Store；凡是“当前任务执行到哪一步”这类进度，才交给 Checkpointer。这样两者各司其职，排查问题时也清晰。

4. DeepAgents 的场景：把 Store 变成 Agent 的“第二大脑”

4.1 多阶段任务里的记忆读写

LangChain 和 LangGraph 是通用编排层，DeepAgents 则更偏“深度任务型 Agent”。它把一个大目标拆解成多个子任务，再让不同子 Agent 协作完成。这种场景下，记忆有一个特殊之处：单个子 Agent 只看到自己的局部上下文，但整个任务链需要共享前期结论。

比如一个“市场调研 + 报告生成”型智能体，子任务 A 负责收集竞品信息，子任务 B 负责生成报告。如果子任务 B 没有记忆，它就得重新读一遍所有原始资料；如果两个子任务各自维护一份本地缓存，数据又会不一致。DeepAgents 的做法是把 Store 作为任务链的“公共记忆仓库”，每个子 Agent 在启动时从 Store 拉取上游结论，结束时把新结论写回。

这里我说的 DeepAgents，是指我最近在一个内部数据产品里集成的那一类 deep agent 框架，接口抽象和 LangGraph 类似，都具备Store.put/Store.get语义。实际集成时，我给每个 Agent 分配了独立命名空间，但在命名空间之间保留“可继承”关系：

STORE_NAMESPACE = { "researcher": ("task", "research", "findings"), "reporter": ("task", "report", "final"), }

子任务 B 启动时，不是只读自己的 final 命名空间，而是先读上游 researcher 的 findings。这样即使模型上下文被压缩，关键结论也不会丢失。

4.2 记忆更新的三个时机

在实际操作里，我总结出 DeepAgents 更新长期记忆的三个关键时机，不是随便每个节点都写。写太多会让 Store 越来越脏，写太少又起不到记忆作用。

第一个时机是任务规划前。Agent 在拆解目标之前，先查 Store 里有没有同类型任务的历史记录，比如“用户上次用的报告结构是什么”“上次执行时踩过哪类数据坑”。这些信息可以直接影响本轮规划，避免重复犯错。

第二个时机是子任务完成时。每个子 Agent 输出结果之后，把可复用的结论写入 Store。注意不要写原始大段文本，要提炼成结构化的字段。例如：

store.put( ("task", "2025-06-01", "extracted"), "customer_trend", { "category": "demand", "level": "high", "summary": "客户对实时数据看板的需求增长明显", "created_at": "2025-06-01T10:00:00Z" } )

第三个时机是用户反馈之后。用户说“这个报告不要用饼图”或者“下次先做政策分析”，这属于偏好型记忆，价值最高，但也很容易被忽略。我通常会在用户反馈时单独触发一个on_user_feedback回调，把反馈写入 Store，并在之后的任何阶段读取。

这三个时机对应三种记忆类型：任务级历史、事实结论、用户偏好。如果你在实现自己的智能体，可以直接照这个分类去设计命名空间和字段结构。

5. 踩坑记录与排查清单

5.1 namespace 设计错误导致“记忆串味”

我最早做 Store 时，namespace 只有一个元组，比如("memories", user_id)，后来发现同一个 Store 里不同 Agent 写入的内容都混在一起。用户在一个业务里表达的偏好，另一个不相干业务也读到了，非常尴尬。

后来我统一了 namespace 的分层规范，基本是这个模式：

(user_id, agent_type, memory_type)

• 第一层必须是用户维度，防止用户之间互相串。
• 第二层是 Agent 类型，比如shopping_agent、report_agent，不同业务隔离。
• 第三层是记忆类型，比如facts、preferences、task_history。

search的时候也按层级逐步放宽：先精确到(user_id, agent_type, memory_type)，没有结果再放宽到(user_id, agent_type)。这种设计能避免一个 Agent 变笨：它不是没有记忆，而是没有被允许访问其他 Agent 的记忆。

5.2 序列化与 JSON 兼容性问题

很多框架的 Store 要求 value 是 JSON 兼容 dict，但你在代码里可能随手塞了 numpy 类型、datetime、自定义对象。写入的时候没问题，可能某些实现内部帮你转了一层；一旦换到严格的 Store，比如不经过 LangChain 而直接落到普通数据库表时，就会报“Object of type datetime is not JSON serializable”。

我的固定做法是写一个统一的序列化入口，而不是到处json.dumps(...)。比如上面 LangChain 例子里的default=str，只能兜底，不能区分字段。更稳妥的是在写入前就把对象转换成基础类型：

def _to_jsonable(value: Any) -> Any: if isinstance(value, dict): return {k: _to_jsonable(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [_to_jsonable(v) for v in value] if hasattr(value, "isoformat"): return value.isoformat() return value

这样时间字段在 Store 里统一是2025-06-01T10:00:00Z这样的字符串，读取后想转 datetime 再转。避免在数据库里出现多种日期格式，后续做时间过滤会非常痛苦。

5.3 并发写入和过期策略

多个子 Agent 同时写同一个用户的 Store 时，可能发生后写覆盖先写。如果两个写入都针对同一个 key，那么后写入的一定会覆盖前面的。很多 Store 实现没有提供 CAS（Compare And Swap）能力，所以我在设计上会尽量减少对同一个 key 的并发写。比如“用户偏好”在同一时刻只允许一个入口写入，或者把每次写入变成一个带时间戳的新 key，而不是覆盖旧 key。

过期策略也是长期记忆绕不开的问题。用户半年前说“我常用的城市是上海”，半年后可能已经搬走了。我建议给每条记忆打时间戳，并在读取时做两个动作：一是过滤掉超过有效期的条目，二是对同类记忆做“最新值优先”。具体有效期多长，要看业务，但一定要有这个机制，否则 Store 会积累大量过期信息，反而误导 Agent 的决策。

5.4 常见问题速查表

6. 如果你也在做长期记忆，可以参考这几个配置

最后说几个我在实际项目里沉淀下来的配置习惯。首先，Store 的底层选型不要一上来就上 Redis 这种重组件。开发和测试阶段用InMemoryStore，虽然进程一重启就没了，但调试快、无依赖。等到功能稳定，再切换到 SQLite 或 Redis，切换成本会被抽象层挡在业务代码之外。

其次，长期记忆的价值在于“精确读取”，而不是“全部读取”。很多 Agent 性能变慢，就是因为每次对话都把所有记忆塞进 prompt。正确做法是先从 Store 里 search 出候选记忆，做一轮截断，只保留高相关的 5 到 10 条。我也见过有人把所有历史对话直接存 Store，这种“记忆”没有结构，读出来也只是噪声，不如不存。

然后是一个小技巧：在 Store 的 value 里额外存一个version字段。每次业务逻辑升级，旧记忆的字段结构可能不兼容，version 字段能帮你做迁移或忽略旧数据。否则线上跑了一周后你想给记忆模块加字段，会发现历史记录全是旧结构，代码每次读取都要兼容两层。

我个人体会是，做智能体长期记忆，真正难的从来不是怎么调用 Store API，而是怎么定义记忆的边界和生命周期。哪些内容值得跨会话保存？哪些内容只是临时中间变量？每个团队都要有自己的答案。如果你刚开始做，建议先只保存“用户明确表达的偏好”和“任务产出的关键结论”两类，跑通以后再逐步扩大。这样你的 Store 从一开始就是干净的，后续加语义检索、加过期策略都容易得多。