使用 LangGraph 构建 AI Agent：RAG 与长期记忆
python约 4 分钟RAG 查出来三段资料，模型还是答偏了。 我第一反应不是 prompt 写得烂，而是上下文里少了一块东西：用户以前说过的偏好、项目约束、踩过的坑，RAG 根本查不到。
这就是很多 Agent 做着做着变蠢的地方。 知识库只能回答“公司文档里有什么”，长期记忆才知道“这个人上次为什么嫌弃这个方案”。
LangGraph 这里要分清楚两个东西。checkpointer 管当前 thread 的状态，适合多轮对话、恢复执行；store 管跨 thread 的长期记忆，适合存用户偏好、项目事实、历史决策。官方文档也是这么拆的：checkpointer 是线程级短期记忆，store 是跨线程长期记忆，而且 graph 编译时可以两个一起传进去。
我一般不会一上来就堆一堆节点，先把链路压到最短：
用户问题 -> 取长期记忆 -> RAG 检索 -> 生成回答 -> 必要时写回记忆
先看一段能跑思路的骨架，代码别写成培训班那种全家桶。
from dataclasses import dataclass
from typing import TypedDict
from uuid import uuid4from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
@dataclass
classUserCtx:
user_id: str
classAgentState(TypedDict):
question: str
docs: list[str]
memories: list[str]
answer: str
KB = [
"退款接口走 order_refund 表，状态 WAIT_PAY 不允许退款。",
"订单导出超过 5 万行必须走异步任务，结果写入 oss_file。",
"会员等级变更后要刷新 user_rank_cache，否则前端仍显示旧权益。",
]
defpick_docs(question: str) -> list[str]:
hit = []
for row in KB:
score = sum(1for w in question if w in row)
if score > 2:
hit.append(row)
return hit[:3]
defload_memory(state: AgentState, runtime: Runtime[UserCtx]):
ns = (runtime.context.user_id, "agent_memory")
rows = runtime.store.search(ns, query=state["question"], limit=5)
return {
"memories": [
item.value["text"]
for item in rows
if item.value.get("text")
]
}
defretrieve(state: AgentState):
return {"docs": pick_docs(state["question"])}
defanswer(state: AgentState):
prompt = f"""
你是内部系统排障助手。
用户长期记忆：
{chr(10).join(state["memories"]) or"无"}
知识库命中：
{chr(10).join(state["docs"]) or"无"}
问题：
{state["question"]}
要求：不知道就说不知道，不要编接口。
"""
# 这里换成真实 LLM 调用，我写文章不贴 key，也不贴一堆无关初始化
return {"answer": "根据现有资料看：" + prompt[-180:]}defsave_memory(state: AgentState, runtime: Runtime[UserCtx]):
q = state["question"]
if"以后"notin q and"记住"notin q:
return {}
ns = (runtime.context.user_id, "agent_memory")
runtime.store.put(
ns,
str(uuid4()),
{"text": q.replace("记住", "").strip()}
)
return {}
这段里我最在意的是 load_memory 放在 retrieve 前面。 因为很多时候，长期记忆会影响检索词。
比如用户问：
导出又卡住了，按上次那个限制看下。
只靠 RAG，这句话没法搜。 但长期记忆里如果有一条：
用户负责订单后台，导出超过 5 万行必须异步。
那 Agent 至少知道该往“订单导出”“异步任务”“文件落库”方向查，而不是去翻什么 Excel 库参数。
图接起来：
builder = StateGraph(AgentState)builder.add_node("load_memory", load_memory)
builder.add_node("retrieve", retrieve)
builder.add_node("answer", answer)
builder.add_node("save_memory", save_memory)
builder.add_edge(START, "load_memory")
builder.add_edge("load_memory", "retrieve")
builder.add_edge("retrieve", "answer")
builder.add_edge("answer", "save_memory")
builder.add_edge("save_memory", END)
graph = builder.compile(
checkpointer=InMemorySaver(),
store=InMemoryStore(),
)
调用时有个细节别漏，thread_id 和 user_id 不是一回事。
resp = graph.invoke(
{"question": "记住，我这边订单导出超过 5 万行都走异步"},
config={"configurable": {"thread_id": "chat-20260626"}},
context=UserCtx(user_id="u_1024"),
)resp = graph.invoke(
{"question": "导出又卡住了，按上次那个限制看下"},
config={"configurable": {"thread_id": "chat-new"}},
context=UserCtx(user_id="u_1024"),
)
print(resp["answer"])
thread_id 换了，短期对话上下文断了，这没问题。 但 user_id 没变，store 里的长期记忆还在，这才叫“跨会话”。
这里我踩过一个小坑：别什么都往长期记忆里塞。 用户随口一句“今天网络好慢”，没必要记。 真正值得记的，一般是这几类：
项目约束、用户偏好、业务规则、已经确认过的排查结论。
更稳一点，可以把写记忆的节点改得保守些：
defshould_keep(text: str) -> bool:
keys = ["记住", "以后", "默认", "必须", "不要", "上次确认"]
return any(k in text for k in keys)defsave_memory(state: AgentState, runtime: Runtime[UserCtx]):
text = state["question"].strip()
ifnot should_keep(text):
return {}
ns = (runtime.context.user_id, "agent_memory")
runtime.store.put(ns, str(uuid4()), {
"text": text,
"source": "user_confirmed"
})
return {}
线上别用 InMemoryStore。进程一重启，全没了。官方文档也明确说内存版适合示例，生产要换数据库后端；长期记忆在 LangGraph 里是按 namespace 和 key 组织的 JSON 文档，这个结构很适合按用户、组织、应用拆开存。
RAG 解决的是“资料在哪”。 长期记忆解决的是“这个人是谁，他上次怎么定的”。 LangGraph 的价值不在于把 Agent 画成一张漂亮流程图，而是你能把这些状态切开：临时对话放 checkpointer，长期事实放 store，知识库仍然走检索。
这几个边界分不清，Agent 看着能聊，真接业务时迟早开始胡说。