如何利用 LangGraph 的 StateGraph 实现带记忆的复杂多轮对话工作流
LangGraph 的核心逻辑在于把对话状态(State)交给一个可持久化的 Checkpointer 托管,而不是每次请求都手动把历史记录塞进 Prompt。很多新手容易把
StateGraph 当成简单的流程图,其实它本质上是一个带状态的有限状态机。实现带记忆的复杂工作流,最关键的配置是定义一个 TypedDict 作为状态容器,并给 Graph 绑定一个内存检查点。
核心配置步骤:
首先定义状态,确保 messages 字段使用 annotated 配合 add_messages,这样新消息会追加到历史中,而不是覆盖掉之前的对话。
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
# add_messages 确保消息是累加的,这是实现记忆的基础
messages: Annotated[list, add_messages]
user_context: str # 用来存储从对话中提取的临时用户偏好接下来是构建图。为了实现“复杂多轮”,我通常会设置一个路由节点(Router),根据 LLM 的判断决定是直接回答,还是调用工具,或者要求用户补充信息。
workflow = StateGraph(State)
# 定义节点
def assistant(state: State):
# 这里调用 LLM,LLM 能看到 state["messages"] 里的全部历史
response = llm.invoke(state["messages"])
return {"messages": [response]}
workflow.add_node("assistant", assistant)
workflow.add_edge(START, "assistant")
workflow.add_edge("assistant", END)
# 关键点:配置 MemorySaver 实现持久化
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)避坑指南与效率技巧:
1. 线程 ID (thread_id) 的误区
很多人发现对话没记忆,是因为调用时没传 config。LangGraph 的记忆是基于 thread_id 隔离的。如果你想让 AI 记得你是谁,必须在 invoke 时传入同一个 ID:
config = {"configurable": {"thread_id": "user_12345"}}
app.invoke({"messages": [HumanMessage(content="我叫老王")]}, config)2. 状态爆炸问题
对话轮数多了以后,messages 列表会极其冗长,导致 Token 成本激增且模型注意力分散。不要依赖默认的 add_messages 盲目堆积。建议写一个 trim_messages 的节点,在进入 LLM 之前,根据 Token 数量或轮数截断历史,只保留最近 10 轮和最初的 System Prompt。
3. 状态更新的颗粒度
不要把所有变量都塞进 messages。比如用户的地理位置、偏好,应该单独开一个 user_context 字段。在节点中通过 return {"user_context": "北京"} 更新,这样 LLM 在处理后续逻辑时,直接读取该字段即可,不需要在历史记录里反复检索,响应速度和准确率会提升很多。
免费 AI 工具箱 · 全部完全免费
各类AI落地变现的详细拆解见AI赚钱方法实操指南,有不少直接可参考的案例。
全部回复 (0)
还没有回复,来发第一条吧!
