LangGraph

🎯 学习目标

• 理解 StateGraph、节点、普通边与条件边
• 用 ToolNode 执行工具并写回 messages 状态
• 配置 MemorySaver Checkpoint 支持中断恢复

引言

当 AgentExecutor 的黑盒循环不够用时——需要人工审批、多 Agent、精确分支——LangGraph 用显式状态机描述流程。

章节正文

第 1 步：定义 State 与图

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

tool_node = ToolNode(tools)

def call_model(state: AgentState):
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile()

tools_condition：若 AIMessage 含 tool_calls → tools 节点，否则 → END。

第 2 步：Checkpoint 与人工节点

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [("user", "查北京天气")]}, config)

# 后续同 thread_id 可恢复上下文

人工审批：增加 human_review 节点，在敏感工具执行前 interrupt，等待 API 回调后再 resume（生产常见模式）。

第 3 步：与 AgentExecutor 选型

场景	推荐
标准工具循环	AgentExecutor
多步骤工作流、审批、子图	LangGraph
Multi-Agent 协作	LangGraph 多节点 + 共享 State

LangGraph 图结构应可导出为 Mermaid 便于评审，与 5.6 Workflow 文档并存。

动手练习

1. 用 LangGraph 重写 6.5 的天气+计算器 Agent
2. 为 thread_id 模拟两次 invoke，验证 Checkpoint 恢复历史
3. 画一张 Mermaid 图描述 agent → tools 循环

本节小结

LangGraph 用 StateGraph 显式建模 Agent；ToolNode + tools_condition 处理工具边；Checkpoint 持久化 thread；复杂流程优于 AgentExecutor 黑盒。