LangGraph入门
很简陋的入门状态，第一部分
LangGraph 本质上是一个低层的 Agent 编排与运行框架，重点不是“帮你写 Prompt”，而是管理状态、流程、持久化、流式输出和人工介入
模型 → 工具 → 模型 → 工具 → 模型 → END,实际上是一个循环
1. 五个核心概念
1.1 State: 共享状态
State可以理解为：整个 Agent 共用的一张工作表
例如：
{"question":"什么是 RAG？","intent":"技术问题","documents":["文档1","文档2"],"answer":"RAG 是……","messages":[...]}所有节点都能读取它，
但是节点一般不应该直接修改原state，而是返回：
{"answer":"新的答案"}意思是：我要更新state中的answer字段，
官方对节点的定义就是State → Partial，
也就是接收完整状态 → 返回部分状态更新
1.2 Node：节点
Node 本质上就是一个普通 Python 函数：
defnode_name(state):# 1. 读取 State# 2. 执行任务# 3. 返回需要更新的字段return{"字段":新值}例如：
defclassify_question(state):question=state["question"]if"Python"inquestion:intent="python"else:intent="general"return{"intent":intent}
1.3 Edge：边
Edge决定：一个节点执行完，接下来执行谁
普通边：
""" 顺序就是： 执行节点A ↓ 节点A执行结束 ↓ 沿着Edge前往节点B ↓ 执行节点B """builder.add_edge("节点A","节点B")条件边：
""" 执行分类节点 ↓ 分类节点执行结束 ↓ 调用路由函数 ↓ 根据路由函数返回值 ↓ 选择下一个节点 """ builder.add_conditional_edges( "分类节点", 路由函数 )节点负责“做事情”，Edge 负责“做完以后去哪”。普通 Edge 固定去一个节点，条件 Edge 先通过路由函数判断再决定去哪。
1.4 START / END
START和END是两个虚拟节点START：图从哪里开始
END：图在哪里结束
builder.add_edge(START,"classify")builder.add_edge("answer",END)
1.5 Checkpointer
Checkpointer 可以理解为：Agent 的自动存档系统
每执行到一个关键节点，就保存：
当前 State 是什么
下一步准备执行谁
属于哪个 thread
有哪些 interrupt
它让 Agent 可以：
多轮对话
中断后继续
失败后恢复
查看历史状态
时间旅行
2. 第一个 LangGraph：线性图
先用普通 Python 函数理解图
安装：
pipinstall-Ulanggraph langchain
- 当前官方 Python 文档要求 Python 3.10 或更高版本。
完整代码：
fromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,END# 1. 定义 StateclassAgentState(TypedDict):question:str#输入问题normalized_question:NotRequired[str]#标准化后的问题answer:NotRequired[str]#答案# State是整个图的共享数据结构，所有节点都通过它读写数据# 2. 定义节点，每个节点是一个纯函数，接收 state，返回要更新的字段defnormalize_question(state:AgentState):print("执行 normalize_question")question=state["question"]normalized=question.strip()# 只返回要更新的字段return{"normalized_question":normalized}defgenerate_answer(state:AgentState):print("执行 generate_answer")question=state["normalized_question"]return{"answer":f"已经收到你的问题：{question}"}# 3. 创建图builder=StateGraph(AgentState)# 4. 添加节点builder.add_node("normalize",normalize_question)builder.add_node("answer",generate_answer)# 5. 添加边builder.add_edge(START,"normalize")builder.add_edge("normalize","answer")builder.add_edge("answer",END)# 6. 编译graph=builder.compile()# 7. 调用图result=graph.invoke({"question":" 什么是 RAG？ "})print(result)
3. 为什么节点只返回“要更新的字段
错误的思路：
defnode(state):state["answer"]="结果"returnstate推荐写法：
defnode(state):return{"answer":"结果"}因为LangGraph 要自己管理：
状态合并
并行更新
Checkpoint
恢复执行
状态历史
Reducer
如果到处直接修改原对象，状态变化会越来越难追踪。
4. 条件分支：add_conditional_edges
→ Python回答 →
START → 问题分类 END
→ 普通回答 →
代码：
这段代码看懂就应该看懂了(废话哈哈)
fromtypingimportLiteralfromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,ENDclassAgentState(TypedDict):question:str# 输入问题route:NotRequired[str]# 路由决策（"python" 或 "general"）answer:NotRequired[str]# 答案# 问题分类defclassify_question(state:AgentState):question=state["question"].lower()# 转为小写#根据问题是否包含 "python" 来决定路由if"python"inquestion:route="python"else:route="general"return{"route":route}# 路由函数，根据路由选择对应的路由函数，注意不是节点。defroute_question(state:AgentState)->Literal["python_answer","general_answer"]:ifstate["route"]=="python":return"python_answer"return"general_answer"defpython_answer(state:AgentState):return{"answer":f"这是一个 Python 问题：{state['question']}"}defgeneral_answer(state:AgentState):return{"answer":f"这是一个普通问题：{state['question']}"}builder=StateGraph(AgentState)builder.add_node("classify",classify_question)builder.add_node("python_answer",python_answer)builder.add_node("general_answer",general_answer)builder.add_edge(START,"classify")builder.add_conditional_edges("classify",route_question# 路由函数，用这个函数来选择路由)builder.add_edge("python_answer",END)builder.add_edge("general_answer",END)graph=builder.compile()result=graph.invoke({# "question": "Python 的装饰器是什么？""question":"明天天气怎么样？"})print(result["answer"])
5. Reducer(必须掌握点…(gpt说的))
Reducer 是状态合并策略，决定多个值怎么合并到一起。
假设 State 里有：
{"logs":["节点A执行完成"]}节点 B 返回：
return{"logs":["节点B执行完成"]}默认情况下，新值会覆盖旧值：
{"logs":["节点B执行完成"]}但我们想要的是:
{"logs":["节点A执行完成","节点B执行完成"]}这时就需要Reducer。
importoperatorfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictclassAgentState(TypedDict):logs:Annotated[list[str],operator.add]
operator.add相当于告诉 LangGraph：
- 这个字段收到新值时，不要覆盖。 把旧值和新值相加。
Annotated[list[str], operator.add]= 这个字段是字符串列表，多个节点写入时用 + 拼接，不覆盖。完整代码：
importoperatorfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassAgentState(TypedDict):logs:Annotated[list[str],operator.add]defnode_a(state:AgentState):return{"logs":["A 执行完成"]}defnode_b(state:AgentState):return{"logs":["B 执行完成"]}builder=StateGraph(AgentState)builder.add_node("a",node_a)builder.add_node("b",node_b)builder.add_edge(START,"a")builder.add_edge("a","b")builder.add_edge("b",END)graph=builder.compile()result=graph.invoke({"logs":[]})print(result["logs"])
6. add_messages：消息专用 Reducer
聊天 Agent 一般会有：
{"messages":[用户消息,AI消息,用户消息,工具消息,AI消息]}有时还需要：
根据 message ID 更新已有消息 删除消息 接受{"role":"user"}这种简写 处理 HumanMessage、AIMessage、ToolMessage所以 LangGraph 提供了专用 Reducer：
add_messagesfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlangchain.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messagesclassChatState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]
7. MemorySaver + thread_id：多轮记忆
首先记住：
add_messages 负责合并消息。
Checkpointer 负责保存消息。
thread_id 负责找到属于哪段对话的消息。
缺一不可
fromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromlangchain.messagesimportAnyMessage,AIMessagefromlanggraph.graph.messageimportadd_messagesfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverclassChatState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]defchatbot(state:ChatState):message_count=len(state["messages"])return{"messages":[AIMessage(content=f"我目前能看到{message_count}条历史消息。")]}builder=StateGraph(ChatState)builder.add_node("chatbot",chatbot)builder.add_edge(START,"chatbot")builder.add_edge("chatbot",END)# 创建内存 Checkpointermemory=MemorySaver()# 编译时传入 checkpointergraph=builder.compile(checkpointer=memory)# 同一段对话使用同一个 thread_idconfig={"configurable":{"thread_id":"bells-chat-001"}}# 第一轮result1=graph.invoke({"messages":[{"role":"user","content":"你好，我叫 Bells。"}]},config=config)print(result1["messages"][-1].content)# 第二轮result2=graph.invoke({"messages":[{"role":"user","content":"这是第二轮对话。"}]},config=config)print(result2["messages"][-1].content)
8. 长短期记忆
8.1 短期记忆
短期记忆依赖：Checkpointer + thread_id
保存的是：
- 用户刚才说了什么
- 工具调用了什么
- 图执行到哪个节点
- 当前任务进行到哪里
8.2 长期记忆
长期记忆是跨 thread 的。
例如当前有两段不同对话：
- thread_id = rag-chat-001
- thread_id = python-chat-003
但两段对话都属于同一个用户：user_id = bells
你希望 Agent 在两个不同对话中都知道：用户主要学习 Python
这时候需要：Store + user_id namespace
9. interrupt：让 Agent 停下来等人
interrupt()的作用：图先暂停 保存当前状态 把问题交给外部用户 等待用户输入 之后从暂停位置继续典型场景：
删除数据前确认 发送邮件前确认 调用付费接口前确认 执行 SQL 写操作前确认 发布文章前人工审核完整示例:
fromtypingimportLiteralfromtyping_extensionsimportTypedDict,NotRequiredfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportinterrupt,Commandfromlanggraph.checkpoint.memoryimportMemorySaverclassReviewState(TypedDict):draft:strstatus:NotRequired[str]defreview_node(state:ReviewState)->Command[Literal["publish","cancel"]]:# 图会暂停在这里decision=interrupt({"question":"是否发布这段内容？","draft":state["draft"]})# 恢复后，decision 就是 Command(resume=...) 传进来的值ifdecision["approved"]:returnCommand(update={"status":"审核通过"},goto="publish")returnCommand(update={"status":"审核拒绝"},goto="cancel")defpublish_node(state:ReviewState):print("发布内容：",state["draft"])return{"status":"已经发布"}defcancel_node(state:ReviewState):return{"status":"已经取消"}builder=StateGraph(ReviewState)builder.add_node("review",review_node)builder.add_node("publish",publish_node)builder.add_node("cancel",cancel_node)builder.add_edge(START,"review")builder.add_edge("publish",END)builder.add_edge("cancel",END)graph=builder.compile(checkpointer=MemorySaver())config={"configurable":{"thread_id":"review-001"}}# 第一次执行：运行到 interrupt 后暂停paused=graph.invoke({"draft":"这是待发布的内容。"},config=config)print(paused)# 用户审核通过后，继续执行result=graph.invoke(Command(resume={"approved":True}),config=config)print(result)
