LangGraph 第3章：核心概念

LangGraph 的核心概念围绕有向图模型展开。理解这四个概念——Graph、State、Node、Edge——就掌握了 LangGraph 的基本思维模型。

四大核心概念

Graph（图）

Graph 是整个工作流的蓝图。它定义了节点（Nodes）和边（Edges）的拓扑结构，描述了数据如何在节点之间流转。

from langgraph.graph import StateGraph

# 创建一个新图
graph = StateGraph(StateType)

Graph 的生命周期：

1. 构建阶段：添加节点和边，定义图的结构
2. 编译阶段：编译为可执行的应用（拓扑排序、验证）
3. 执行阶段：调用应用，输入初始状态，获取最终结果

State（状态）

State 是节点间共享的数据容器。所有节点读取和写入同一个 State 对象。

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import add_messages

# 使用 TypedDict 定义状态
class MyState(TypedDict):
    # 普通字段：后一个值覆盖前一个
    counter: int
    # 带 reducer 的字段：新值合并到列表中
    messages: Annotated[List[str], add_messages]
    # 使用 operator.add 作为 reducer
    steps: Annotated[List[str], operator.add]

State 的关键特性：

Nodes（节点）

Node 是图的基本执行单元。每个节点是一个函数或可调用对象，接收 State 并返回状态更新。

# 普通函数节点
def my_node(state: MyState):
    # 读取状态
    count = state["counter"]
    # 执行逻辑
    new_count = count + 1
    # 返回状态更新
    return {"counter": new_count}

# LLM 调用节点
def llm_node(state: MyState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 异步节点
async def async_node(state: MyState):
    result = await some_async_operation(state)
    return {"result": result}

Edges（边）

Edge 定义了节点之间的流转路径。LangGraph 有两种边：

1. 普通边（Edge）：固定连接，源节点执行完毕后自动进入目标节点。

graph.add_edge("node_a", "node_b")  # A 执行完后自动执行 B

2. 条件边（Conditional Edge）：根据当前状态动态选择目标节点。

# 路由函数：根据状态返回目标节点名称
def router(state: MyState):
    if state["counter"] > 10:
        return "end_node"
    else:
        return "process_node"

graph.add_conditional_edges(
    "decision_node",
    router,  # 路由函数
    {
        "end_node": "end_node",
        "process_node": "process_node"
    }
)

编译与执行流程

编译阶段

编译会将图结构转换为可执行的计算图：

# compile() 执行以下操作：
# 1. 验证图的完整性（入口、出口、所有节点可达）
# 2. 进行拓扑排序（确定执行顺序）
# 3. 优化执行计划
# 4. 创建可执行的应用实例
app = graph.compile()

执行阶段

执行分为两种模式：

# 1. 普通执行：阻塞等待最终结果
result = app.invoke(initial_state)

# 2. 流式执行：逐节点输出状态
for state in app.stream(initial_state):
    print(state)

执行流程：

invoke(input)
  |
  v
[Entry Point] --> [Node 1] --> [Conditional]
                                  |
                          +-------+-------+
                          |               |
                          v               v
                      [Node 2]       [Node 3]
                          |               |
                          +-------+-------+
                                  |
                                  v
                              [END]
                                  |
                                  v
                              output

条件边路由函数（补：多个目标节点的映射）

当使用条件边时，路由函数的返回值必须匹配映射表中的键名：

def grade_router(state):
    """根据评分路由到不同节点"""
    score = state.get("score", 0)
    if score >= 90:
        return "excellent"
    elif score >= 60:
        return "pass"
    else:
        return "fail"

graph.add_conditional_edges(
    "grader",                    # 源节点
    grade_router,                # 路由函数
    {                            # 映射表
        "excellent": "reward_node",
        "pass": "review_node",
        "fail": "retry_node"
    }
)

如果路由函数的返回值直接就是目标节点名称，且映射表与返回值一致，可以省略映射表：

graph.add_conditional_edges(
    "decision",
    router
    # 省略映射表，直接使用 router 的返回值作为节点名
)

可视化展示

LangGraph 通过 get_graph() 提供图结构可视化：

# 打印 ASCII 图结构
print(app.get_graph().draw_ascii())

# 生成 Mermaid 格式（可用于 Markdown）
print(app.get_graph().draw_mermaid())

# 生成 Mermaid PNG（需要安装 pygraphviz）
with open("graph.png", "wb") as f:
    f.write(app.get_graph().draw_mermaid_png())

ASCII 可视化示例：

+--------+     +---------+     +--------+
| start  | --> | process | --> | end    |
+--------+     +---------+     +--------+

特殊节点：START 和 END

LangGraph 有两个内置特殊节点：

• START (__start__)：图的入口点，通过 set_entry_point() 设置
• END (__end__)：图的终止点，执行到此节点后停止

graph.set_entry_point("first_node")    # 执行从 first_node 开始
graph.add_edge("last_node", "__end__") # 执行到 last_node 后结束

理解这四个核心概念后，你已掌握 LangGraph 的基础。下一章我们将开始动手搭建开发环境并编写第一个程序。