从零理解 LangGraph
State · Node · Edge · StateGraph

1 LangChain vs LangGraph：你已知道什么，还缺什么

你学过 LangChain，已经知道怎么调用 LLM、拼接 Chain、使用 Tool。LangChain 的核心模型是"流水线"：输入 → 处理步骤 A → 处理步骤 B → 输出，线性地往下走。

LangChain 解决了"怎么调用 LLM"的问题，LangGraph 解决了"LLM 调用的流程控制和状态管理"的问题。当你的 AI 应用需要"根据上一步结果决定下一步做什么"，你就需要 LangGraph。

2 LangGraph 的核心思想：用"图"来描述 AI 的执行流程

什么是"图"？在计算机科学中，图（Graph）由节点（Node）和边（Edge）组成。LangGraph 借用了这个概念：

• Node（节点） = 一个处理步骤，可以是调用 LLM、调用工具、做判断……
• Edge（边） = 节点之间的连接，决定"下一步去哪儿"
• State（状态） = 在整张图里流转的数据，相当于节点之间共享的"便签纸"

想象你在规划一段旅程：

这就是 Module 1 那个 notebook 的完整逻辑：node_1 执行后，根据某个条件（随机/天气/用户输入），决定走 node_2 还是 node_3，最终都到达 END。

3 State（状态）——图的"记忆"

State 是什么？

State 是一个 Python 字典（TypedDict），它在整张图的执行过程中一直存在，从 START 到 END，每个 Node 都可以读取它、修改它。你可以把它理解为节点之间传递的"便签纸"或"共享白板"。

from typing_extensions import TypedDict

# 定义 State 的"模板"：图里有哪些共享数据
class State(TypedDict):
    graph_state: str  # 这张图只有一个字段：graph_state

在这个最简单的例子里，State 只有一个字段 graph_state，类型是字符串。但在真实 Agent 里，State 可以包含：

• 用户的问题
• LLM 的历史对话
• 工具调用的结果
• 搜索到的文档
• 当前执行的步骤数

State 如何流转？

每个 Node 接收当前 State，做完处理后返回一个部分更新的字典，LangGraph 会自动把这个更新合并进 State：

在真实 Agent 中，State 长什么样？

from langchain_core.messages import BaseMessage
from typing import Annotated, List
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # Annotated + add_messages 表示"追加"而不是"覆盖"
    messages: Annotated[List[BaseMessage], add_messages]
    # 其他字段可以根据需要添加
    user_query: str
    search_results: list
    iteration_count: int

4 Node（节点）——图的"工作单元"

Node 就是普通 Python 函数

这是 LangGraph 最优雅的设计之一：Node 就是一个普通的 Python 函数，没有任何魔法。它只有一个约定：

• 输入：接受当前 State（一个字典）作为第一个参数
• 输出：返回一个字典，字典里是你想更新的 State 字段

# 这就是一个完整的 Node！
def node_1(state):
    print("---Node 1---")
    # 读取 State 中的值
    current = state['graph_state']
    # 返回要更新的字段
    return {"graph_state": current + " I am"}

def node_2(state):
    print("---Node 2---")
    return {"graph_state": state['graph_state'] + " happy!"}

def node_3(state):
    print("---Node 3---")
    return {"graph_state": state['graph_state'] + " sad!"}

在真实 Agent 中，Node 里会做什么？

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def call_llm(state):
    # 从 State 里取消息历史
    messages = state["messages"]
    # 调用 LLM（这里就用到了 LangChain！）
    response = llm.invoke(messages)
    # 把 LLM 回复追加到 messages
    return {"messages": [response]}

from langchain_community.tools import TavilySearchResults

search_tool = TavilySearchResults()

def search_web(state):
    query = state["user_query"]
    results = search_tool.invoke(query)
    return {"search_results": results}

5 Edge（边）——图的"决策路线"

Edge 连接 Node，告诉图"执行完这个节点之后，下一步去哪儿"。LangGraph 有两种 Edge：

普通 Edge（Normal Edge）

始终从 A 走到 B，没有条件判断。

# 总是从 node_2 走到 END
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# 总是从 START 走到 node_1
builder.add_edge(START, "node_1")

条件 Edge（Conditional Edge）

根据一个路由函数的返回值，决定下一步去哪个 Node。路由函数接受 State，返回下一个 Node 的名字。

import random
from typing import Literal

# 路由函数：根据 State 决定下一步
def decide_mood(state) -> Literal["node_2", "node_3"]:
    # 可以根据 state 里的任何信息来决定
    if random.random() < 0.5:
        return "node_2"   # 走 happy 路线
    return "node_3"       # 走 sad 路线

# 把路由函数挂到 node_1 之后
builder.add_conditional_edges("node_1", decide_mood)

在真实 Agent 中，条件 Edge 决定"是否继续循环"

def should_continue(state) -> Literal["tools", "__end__"]:
    messages = state["messages"]
    last_message = messages[-1]
    # 如果 LLM 要调用工具 → 去 tools 节点
    if last_message.tool_calls:
        return "tools"
    # 否则 LLM 已经给出最终答案 → 结束
    return "__end__"

# 这就构成了 Agent 的"思考→工具→思考→工具→…→结束"循环
builder.add_conditional_edges("agent", should_continue)

6 StateGraph Class——把所有零件组装成图

StateGraph 是 LangGraph 的核心类，它是一个"图的构建器"（Builder Pattern）。你把所有 Node 和 Edge 的定义"告诉"它，最后 compile() 生成一个可以运行的图。

完整的构建流程

• 1

  初始化：告诉 StateGraph 用哪个 State 类

  StateGraph(State) — 这确定了图中所有节点共享的数据结构模板

• 2

  add_node：注册每个处理步骤

  builder.add_node("名字", 函数) — "名字"是在图里引用这个节点的标识符，"函数"是实际执行的逻辑

• 3

  add_edge / add_conditional_edges：定义流转路线

  普通边用 add_edge，需要判断条件的路线用 add_conditional_edges

• 4

  compile()：校验并生成可运行的图

  LangGraph 在这一步做基本的正确性检查（比如有没有孤立节点），返回一个 CompiledGraph 对象

• 5

  invoke()：运行图

  graph.invoke({"graph_state": "输入"}) — 传入初始 State，图从 START 开始执行，返回最终 State

START 和 END 是什么？

START 和 END 是 LangGraph 内置的两个特殊节点：

• START：图的入口，你调用 invoke() 时，从 START 连接的第一个 Node 开始执行。它让你可以用 add_edge(START, "my_first_node") 来明确声明"从哪里开始"
• END：图的出口，Edge 指向 END 意味着"到这里流程终止了"。一张图可以有多个 END 入口（比如不同分支都可以结束）

7 逐行读懂那段构建代码

现在来逐行分析 notebook 里那段你"完全摸不着头脑"的代码：

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# ① 创建一个图的"蓝图"，使用 State 作为共享数据结构
builder = StateGraph(State)

# ② 注册三个节点（"标识符名" → Python函数）
builder.add_node("node_1", node_1)   # node_1 函数处理第一步
builder.add_node("node_2", node_2)   # node_2 = happy 路线
builder.add_node("node_3", node_3)   # node_3 = sad 路线

# ③ 定义固定边：START → node_1（图从 node_1 开始）
builder.add_edge(START, "node_1")

# ④ 定义条件边：node_1 执行完后，调用 decide_mood() 决定去 node_2 还是 node_3
builder.add_conditional_edges("node_1", decide_mood)

# ⑤ 定义固定边：无论走哪条路，最终都到 END
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# ⑥ 编译：校验图的结构，生成可执行的 CompiledGraph
graph = builder.compile()

# ⑦ 可视化（仅用于学习/调试，生产代码里不需要）
display(Image(graph.get_graph().draw_mermaid_png()))

执行这段代码后，图的形状如下：

调用图并观察状态流转

# 传入初始 State，运行图
result = graph.invoke({"graph_state": "Hi, this is Lance."})

# 可能的输出 1（走 node_2）：
# ---Node 1---
# ---Node 2---
# {'graph_state': 'Hi, this is Lance. I am happy!'}

# 可能的输出 2（走 node_3）：
# ---Node 1---
# ---Node 3---
# {'graph_state': 'Hi, this is Lance. I am sad!'}

完整的 State 演变过程：

8 这只是示例吗？以后会经常用吗？

Module 1 用"随机选择 happy/sad"只是为了屏蔽 LLM 的复杂性，让你先理解图的流程控制机制。在后续 Module 里，你会看到：

9 真实 Agent 场景：LangGraph 如何接管 LLM 调用

现在把 Module 1 的简单示例，升级成一个真正调用 LLM 和工具的 Agent，你会发现结构完全相同：

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
from typing import Literal

# ─── Step 1: 定义 State（比 Module 1 更丰富）───
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 追加而非覆盖

# ─── Step 2: 准备 LLM 和工具 ───
tools = [TavilySearchResults(max_results=2)]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)  # 告诉 LLM 有哪些工具可用

# ─── Step 3: 定义 Nodes（和 Module 1 结构完全一样！）───
def call_llm(state: AgentState):
    # Node：调用 LLM，LLM 会决定是直接回答还是调用工具
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# ─── Step 4: 定义路由函数（就是 Module 1 的 decide_mood）───
def should_continue(state: AgentState) -> Literal["tools", END]:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:  # LLM 决定要调用工具
        return "tools"
    return END              # LLM 给出了最终答案

# ─── Step 5: 构建图（和 Module 1 完全一样的模式！）───
builder = StateGraph(AgentState)

builder.add_node("agent", call_llm)
builder.add_node("tools", ToolNode(tools))  # 内置的工具执行节点

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")   # 工具执行完回到 LLM！这就是循环

agent = builder.compile()

# ─── Step 6: 运行 ───
result = agent.invoke({
    "messages": [HumanMessage(content="2024年奥运会在哪里举行？")]
})

注意 add_edge("tools", "agent") 这一行：它把 tools 节点的输出连回了 agent 节点，形成了一个循环。这就是 Agent 反复"思考→工具→思考→工具"直到得出答案的机制。LangChain 的 AgentExecutor 在底层就是这个图，只是以前你看不到它。

10 思维导图：一张图总结所有概念

学习路线建议

Module 1 的学习重点不是代码细节，而是建立这个心智模型：