06-LangGraph入门

第06章：LangGraph 入门

版本：LangChain v1.3.7 | 讲师：汤姆小白

1. LangGraph 是什么

LangGraph是 LangChain 生态的底层编排运行时，专门用于构建有状态、长时间运行的 Agent。

简单理解：LangChain 做"零件"，LangGraph 做"组装线"。

三者关系

create_agent底层就是基于 LangGraph 实现的。学 LangGraph 就是学 Agent 的底层原理。

安装

pipinstalllanggraph

2. StateGraph：定义工作流

2.1 最简单的图

fromlanggraph.graphimportStateGraph,MessagesState,START,ENDfromlangchain.chat_modelsimportinit_chat_model model=init_chat_model("openai:gpt-4o-mini")defcall_model(state:MessagesState):"""节点函数：调用模型"""response=model.invoke(state["messages"])return{"messages":[response]}# 创建图graph=StateGraph(MessagesState)# 添加节点graph.add_node("model",call_model)# 添加边graph.add_edge(START,"model")graph.add_edge("model",END)# 编译并执行app=graph.compile()result=app.invoke({"messages":[{"role":"user","content":"你好！"}]})print(result["messages"][-1].content)

2.2 StateGraph 核心概念

2.3 自定义 State

fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,ENDclassMyState(TypedDict):messages:listcounter:intfinal_answer:strdefstep1(state:MyState):return{"counter":state["counter"]+1}defstep2(state:MyState):return{"final_answer":f"执行了{state['counter']}步"}graph=StateGraph(MyState)graph.add_node("step1",step1)graph.add_node("step2",step2)graph.add_edge(START,"step1")graph.add_edge("step1","step2")graph.add_edge("step2",END)app=graph.compile()result=app.invoke({"messages":[],"counter":0,"final_answer":""})print(result["final_answer"])# 执行了 1 步

3. 条件边：根据状态选择路径

条件边让工作流能根据运行时状态动态决定下一步。

importrandomfromlanggraph.graphimportStateGraph,MessagesState,START,END model=init_chat_model("openai:gpt-4o-mini")defcall_model(state:MessagesState):response=model.invoke(state["messages"])return{"messages":[response]}defshould_continue(state:MessagesState)->str:"""条件判断函数：返回下一个节点名"""last_message=state["messages"][-1]# 检查是否需要工具调用ifhasattr(last_message,"tool_calls")andlast_message.tool_calls:return"tools"returnENDdefcall_tools(state:MessagesState):"""执行工具"""last_message=state["messages"][-1]# 这里简化处理return{"messages":[{"role":"tool","content":"工具结果"}]}graph=StateGraph(MessagesState)graph.add_node("model",call_model)graph.add_node("tools",call_tools)graph.add_edge(START,"model")graph.add_conditional_edges("model",should_continue,{"tools":"tools",END:END,})graph.add_edge("tools","model")# 工具结果返回模型app=graph.compile()

条件边的三个参数：(源节点, 判断函数, 路由映射)。判断函数返回的字符串必须在路由映射中存在。

4. 持久化：断点恢复

Agent 执行中可能因网络、API 限流等原因中断。持久化让 Agent 能从断点继续，不丢失进度。

fromlanggraph.checkpoint.memoryimportMemorySaver# 创建内存检查点（生产环境推荐用 SqliteSaver 或 PostgresSaver）checkpointer=MemorySaver()graph=StateGraph(MessagesState)# ... 添加节点和边 ...app=graph.compile(checkpointer=checkpointer)# 执行时指定 thread_idconfig={"configurable":{"thread_id":"user_123"}}# 第一次调用result1=app.invoke({"messages":[{"role":"user","content":"列出3个Python框架"}]},config=config)# 即使程序崩溃重启，用同一个 thread_id 继续result2=app.invoke({"messages":[{"role":"user","content":"详细说明第一个"}]},config=config# 同一个 thread_id，自动加载历史)

5. 人机协同：关键节点人工审批

在敏感操作（如删除文件、发送邮件）前暂停，等待人工确认。

fromlanggraph.typesimportinterruptdefsensitive_operation(state:MessagesState):"""需要人工审批的节点"""# 暂停并询问用户approval=interrupt("即将删除重要数据，是否继续？")ifapproval=="yes":# 执行操作return{"messages":[{"role":"ai","content":"操作已执行"}]}else:return{"messages":[{"role":"ai","content":"操作已取消"}]}# 在图中使用graph.add_node("sensitive_op",sensitive_operation)app=graph.compile(checkpointer=checkpointer)# 第一次调用：会在 interrupt() 处暂停result=app.invoke(input_data,config)# 此时 result 包含中断信息# 用户确认后，恢复执行app.invoke(None,# 无新输入config,# 传入用户确认的 Command)

6. Stream：实时监控执行过程

# 流式输出每个节点的结果foreventinapp.stream({"messages":[{"role":"user","content":"你好"}]},config={"configurable":{"thread_id":"1"}},stream_mode="values",):# event 是当前完整状态last_msg=event["messages"][-1]print(f"[{last_msg.get('role','node')}]{last_msg.get('content','')}")

7. LangGraph 与 create_agent 的关系

create_agent本质上就是预先配置好的 LangGraph：

# create_agent 内部做的事 ≈fromlanggraph.prebuiltimportcreate_react_agent# create_agent 自动创建了：# - model 节点# - tools 节点# - 条件边（有工具调用→继续，无→结束）# - 中间件机制# - 流式支持

理解 LangGraph 后，你就能：

• 在create_agent不够灵活时，用 LangGraph 自定义 Agent 行为
• 添加自定义逻辑到 Agent 循环中
• 实现复杂的多 Agent 协作

本章小结

LangGraph 是 Agent 的"骨架"，create_agent 是 LangGraph 的"快捷方式"。先学 create_agent 快速上手，再用 LangGraph 深入定制。