LangGraph 最简图（Simple Graph）详细解读

📚 概述

本文档详细解读 LangGraph 的 最简图（Simple Graph） 教程。这是 LangGraph 入门的第一课，通过构建一个包含 3 个节点和 1 个条件边的简单图，帮助您掌握 LangGraph 的核心概念和基础架构。

学习目标：

理解 LangGraph 的核心组件（State、Node、Edge）
掌握图的构建和执行流程
学习条件边的使用方法
熟悉 Python 类型提示和 TypedDict

🎯 核心概念

什么是 LangGraph？

LangGraph 是一个用于构建有状态的多步骤应用的框架。它将应用程序建模为一个图（Graph），其中：

节点（Node）：执行具体操作的函数
边（Edge）：定义执行流程的路径
状态（State）：在节点间传递的数据

图的基本结构

START → Node 1 → [条件判断] → Node 2 或 Node 3 → END

特点：

流程控制：通过边控制节点的执行顺序
状态管理：每个节点可以读取和更新状态
动态路由：条件边根据状态动态选择下一个节点

🎭 实战案例：情绪生成器

我们将构建一个简单的情绪生成器，演示 LangGraph 的完整工作流程：

需求：

接收用户输入（如 "Hi, this is Lance."）
添加 "I am" 到文本
随机选择添加 "happy!" 或 "sad!"

系统架构图

用户输入: "Hi, this is Lance."
        ↓
    [START]
        ↓
    [Node 1] 添加 "I am"
        ↓
  [条件边: decide_mood]
        ↓
   ┌────────────┐
   ↓            ↓
[Node 2]     [Node 3]
添加 "happy!"  添加 "sad!"
   ↓            ↓
   └─────┬──────┘
        ↓
      [END]
        ↓
输出: "Hi, this is Lance. I am happy!"
     或 "Hi, this is Lance. I am sad!"

🔧 代码实现详解

1. 环境准备

python

%%capture --no-stderr
%pip install --quiet -U langgraph

说明：

%%capture --no-stderr：Jupyter 魔法命令，隐藏安装输出
--quiet：静默安装模式
-U：升级到最新版本

2. 定义状态（State）

状态是 LangGraph 的核心数据结构，贯穿整个图的执行过程。

python

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: str

Python 知识点：TypedDict

TypedDict 是 Python 3.8+ 引入的类型提示工具，用于定义字典的结构：

python

# 传统字典 - 无类型提示
state = {"graph_state": "Hello"}

# TypedDict - 有类型提示
class State(TypedDict):
    graph_state: str  # 明确指定字段类型

state: State = {"graph_state": "Hello"}

优势：

IDE 支持：自动补全和类型检查
文档作用：清晰展示数据结构
错误预防：静态类型检查发现潜在错误

LangGraph 中的作用：

State 定义了图中所有节点共享的数据结构
每个节点都可以访问和修改 State 中的字段
提供了类型安全的状态管理

3. 定义节点（Nodes）

节点是执行具体操作的 Python 函数。

python

def node_1(state):
    print("---Node 1---")
    return {"graph_state": state['graph_state'] + " 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!"}

节点的核心特性

1. 输入参数

python

def node_1(state):
    #         ^^^^^
    #         必须接收 state 参数

第一个参数是当前状态（State 类型）
可以通过 state['graph_state'] 访问状态字段

2. 返回值

python

return {"graph_state": state['graph_state'] + " I am"}
#      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
#      返回一个字典，更新状态中的字段

返回的字典会更新状态
默认行为：覆盖原有值（非合并）

3. 状态更新机制

python

# 执行前：state = {"graph_state": "Hi, this is Lance."}
node_1(state)
# 执行后：state = {"graph_state": "Hi, this is Lance. I am"}

重要：返回的值会完全替换对应字段的原值，除非使用了 Reducer（后续课程讲解）。

4. 定义边（Edges）

边定义了节点之间的连接关系，分为两种类型：

普通边（Normal Edge）

python

builder.add_edge(START, "node_1")

固定路由：总是从起点流向终点
无条件执行：每次都会走这条路径

条件边（Conditional Edge）

python

import random
from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:
    # 从状态中获取数据（这里未使用，仅作示例）
    user_input = state['graph_state']

    # 50/50 随机选择
    if random.random() < 0.5:
        return "node_2"  # 50% 概率返回 Node 2

    return "node_3"      # 50% 概率返回 Node 3

Python 知识点：Literal 类型

Literal 是 Python 3.8+ 的高级类型提示，用于限定返回值的可能取值：

python

from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:
    #                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^
    #                     返回值只能是 "node_2" 或 "node_3"
    return "node_2"  # ✅ 合法
    # return "node_4"  # ❌ 类型检查会报错

优势：

类型安全：IDE 会检查返回值是否合法
自动补全：编辑器会提示可用选项
文档作用：清晰展示所有可能的路径

条件边的工作原理

python

# 条件函数的职责
def decide_mood(state) -> Literal["node_2", "node_3"]:
    # 1. 读取状态
    user_input = state['graph_state']

    # 2. 执行逻辑判断
    if some_condition:
        return "node_2"  # 路由到节点 2
    else:
        return "node_3"  # 路由到节点 3

执行流程：

节点 1 执行完毕
LangGraph 调用 decide_mood(state)
根据返回值（"node_2" 或 "node_3"）选择下一个节点
执行被选中的节点

5. 构建图（Graph Construction）

现在将所有组件组装成完整的图。

python

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

# 1. 初始化图构建器
builder = StateGraph(State)

# 2. 添加节点
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 3. 添加边
builder.add_edge(START, "node_1")                    # 开始 → 节点1
builder.add_conditional_edges("node_1", decide_mood) # 节点1 → 条件判断
builder.add_edge("node_2", END)                      # 节点2 → 结束
builder.add_edge("node_3", END)                      # 节点3 → 结束

# 4. 编译图
graph = builder.compile()

# 5. 可视化
display(Image(graph.get_graph().draw_mermaid_png()))

LangGraph 核心 API 详解

1. StateGraph

python

builder = StateGraph(State)
#                    ^^^^^
#                    传入状态类型

图的构建器，用于添加节点和边
泛型类，接受 State 类型参数
确保所有节点使用相同的状态结构

2. add_node

python

builder.add_node("node_1", node_1)
#                ^^^^^^^^  ^^^^^^
#                节点名称   节点函数

注册节点到图中
节点名称用于边的连接（字符串形式）
节点函数是实际执行的逻辑

3. add_edge

python

builder.add_edge(START, "node_1")
#                ^^^^^  ^^^^^^^^
#                起点    终点

添加固定路由的边
START 和 END 是特殊节点（后续说明）

4. add_conditional_edges

python

builder.add_conditional_edges("node_1", decide_mood)
#                             ^^^^^^^^  ^^^^^^^^^^^
#                             起点节点   条件函数

添加条件路由的边
条件函数返回下一个节点的名称
自动根据返回值路由到对应节点

5. compile

python

graph = builder.compile()

将构建器转换为可执行的图
执行基本验证（如检查边的连接是否有效）
返回 CompiledGraph 对象，可以调用 invoke 等方法

特殊节点：START 和 END

START 节点

图的入口点
接收用户输入
将输入传递给第一个节点

python

builder.add_edge(START, "node_1")
# 用户输入 → node_1

END 节点

图的出口点
标记执行结束
返回最终状态

python

builder.add_edge("node_2", END)
# node_2 → 结束并返回结果

可视化图结构

python

display(Image(graph.get_graph().draw_mermaid_png()))

生成 Mermaid 图，显示图的结构：

节点：矩形框
边：箭头连接
条件边：菱形判断框

6. 执行图（Graph Invocation）

图实现了 LangChain 的 Runnable 协议，提供统一的执行接口。

python

graph.invoke({"graph_state": "Hi, this is Lance."})

输出示例：

---Node 1---
---Node 3---
{'graph_state': 'Hi, this is Lance. I am sad!'}

invoke 方法详解

1. 输入参数

python

graph.invoke({"graph_state": "Hi, this is Lance."})
#            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
#            初始状态（字典形式）

输入必须是字典
字段必须与 State 定义匹配
设置图的初始状态

2. 执行流程

python

# 1. 初始状态
state = {"graph_state": "Hi, this is Lance."}

# 2. START → node_1
state = node_1(state)
# {"graph_state": "Hi, this is Lance. I am"}

# 3. 条件边判断
next_node = decide_mood(state)  # 返回 "node_3"

# 4. node_3 执行
state = node_3(state)
# {"graph_state": "Hi, this is Lance. I am sad!"}

# 5. END 节点
return state

3. 返回值

python

result = graph.invoke({"graph_state": "Hi, this is Lance."})
print(result)
# {'graph_state': 'Hi, this is Lance. I am sad!'}

返回最终状态（字典形式）
包含所有状态字段的最新值

同步执行特性

python

invoke() # 同步方法

阻塞执行：等待每个节点完成后才继续
顺序执行：按边的定义依次执行
最终返回：所有节点执行完毕后返回最终状态

对比异步方法（后续课程）：

invoke()：同步，等待完成
ainvoke()：异步，非阻塞
stream()：流式，逐步返回

🎓 核心知识点总结

LangGraph 特有概念

1. 状态（State）

定义方式：

python

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: str

作用：

定义图中所有节点共享的数据结构
提供类型安全的状态管理
支持状态的读取和更新

关键点：

必须使用 TypedDict 或类似类型
所有节点都使用相同的状态结构
状态字段可以是任意类型（str, list, dict, 自定义类等）

2. 节点（Node）

定义方式：

python

def node_name(state):
    # 读取状态
    value = state['field_name']

    # 执行逻辑
    new_value = process(value)

    # 更新状态
    return {"field_name": new_value}

特性：

必须接收 state 参数
返回字典更新状态
可以读取和修改任意状态字段

3. 边（Edge）

普通边：

python

builder.add_edge("node_a", "node_b")
# node_a 总是流向 node_b

条件边：

python

def condition(state) -> Literal["node_b", "node_c"]:
    if some_logic:
        return "node_b"
    return "node_c"

builder.add_conditional_edges("node_a", condition)
# node_a 根据 condition 返回值选择下一个节点

4. 特殊节点

节点	作用	用法
START	图的入口	`builder.add_edge(START, "first_node")`
END	图的出口	`builder.add_edge("last_node", END)`

5. 图的构建流程

python

# 1. 创建构建器
builder = StateGraph(State)

# 2. 添加节点
builder.add_node("name", function)

# 3. 添加边
builder.add_edge(source, target)
builder.add_conditional_edges(source, condition_function)

# 4. 编译
graph = builder.compile()

# 5. 执行
result = graph.invoke(initial_state)

Python 特有知识点

1. TypedDict

python

from typing_extensions import TypedDict

class State(TypedDict):
    field1: str
    field2: int
    field3: list[str]

特点：

定义字典的键和值类型
静态类型检查（IDE 支持）
运行时不强制验证（与 Pydantic 不同）

2. Literal 类型

python

from typing import Literal

def function() -> Literal["option1", "option2"]:
    return "option1"  # 只能返回这两个值之一

作用：

限定返回值的可能取值
提供更精确的类型提示
用于条件边的返回类型

3. 类型提示（Type Hints）

python

# 函数参数和返回值类型
def node_1(state: State) -> dict[str, str]:
    return {"graph_state": "..."}

# 变量类型
graph_state: str = "Hello"

优势：

代码可读性更好
IDE 自动补全和错误检查
文档作用

💡 最佳实践

1. 何时使用简单图？

✅ 适用场景：

学习 LangGraph 基础概念
简单的流程控制（3-5 个节点）
需要条件路由的简单任务
原型开发和概念验证

❌ 不适用场景：

复杂的多步骤流程（考虑子图）
需要并行执行（考虑 Map-Reduce）
需要循环和重试（考虑循环图）
复杂的状态管理（考虑多状态设计）

2. 状态设计原则

原则 1：保持状态简洁

python

# ✅ 好的设计 - 只包含必需字段
class State(TypedDict):
    user_input: str
    result: str

# ❌ 不好的设计 - 过多字段
class State(TypedDict):
    user_input: str
    intermediate_1: str
    intermediate_2: str
    intermediate_3: str
    temp_value: str
    result: str

原则 2：使用有意义的字段名

python

# ✅ 清晰的命名
class State(TypedDict):
    user_query: str
    search_results: list[str]
    final_answer: str

# ❌ 模糊的命名
class State(TypedDict):
    data: str
    result: list
    output: str

原则 3：适当的字段类型

python

class State(TypedDict):
    # 简单类型
    message: str
    count: int
    is_done: bool

    # 复杂类型
    items: list[str]
    metadata: dict[str, Any]

    # 自定义类型（后续课程）
    # response: ChatMessage

3. 节点设计技巧

技巧 1：单一职责

python

# ✅ 每个节点做一件事
def fetch_data(state):
    data = api.fetch()
    return {"data": data}

def process_data(state):
    result = process(state["data"])
    return {"result": result}

# ❌ 节点做太多事情
def do_everything(state):
    data = api.fetch()
    result = process(data)
    final = format(result)
    return {"final": final}

技巧 2：添加日志

python

def node_1(state):
    print(f"---Node 1 执行---")
    print(f"输入状态: {state}")

    result = process(state)

    print(f"输出状态: {result}")
    return result

技巧 3：错误处理

python

def safe_node(state):
    try:
        result = risky_operation(state)
        return {"result": result, "error": None}
    except Exception as e:
        return {"result": None, "error": str(e)}

4. 条件边的设计

技巧 1：基于状态的路由

python

def route_by_state(state) -> Literal["node_a", "node_b"]:
    # 基于状态字段做决策
    if state["score"] > 0.8:
        return "node_a"
    return "node_b"

技巧 2：多条件判断

python

def complex_routing(state) -> Literal["node_a", "node_b", "node_c"]:
    value = state["value"]

    if value < 10:
        return "node_a"
    elif value < 50:
        return "node_b"
    else:
        return "node_c"

技巧 3：添加默认路由

python

def safe_routing(state) -> Literal["node_a", "node_b", "error_node"]:
    try:
        if state["condition"]:
            return "node_a"
        return "node_b"
    except:
        return "error_node"  # 异常情况的默认路由

🚀 进阶扩展

1. 多字段状态

python

class State(TypedDict):
    user_input: str
    processing_step: int
    results: list[str]
    metadata: dict

def node(state):
    # 可以更新多个字段
    return {
        "processing_step": state["processing_step"] + 1,
        "results": state["results"] + ["new_item"]
    }

2. 嵌套状态结构

python

from typing import Any

class State(TypedDict):
    config: dict[str, Any]
    data: dict[str, list[str]]

def node(state):
    # 访问嵌套结构
    threshold = state["config"]["threshold"]
    items = state["data"]["items"]

    return {"data": {"items": processed_items}}

3. 可选字段（Python 3.11+）

python

from typing import NotRequired

class State(TypedDict):
    required_field: str
    optional_field: NotRequired[str]  # 可选字段

🎯 实际应用案例

案例 1：简单的聊天机器人

python

class ChatState(TypedDict):
    user_message: str
    bot_response: str
    sentiment: str

def analyze_sentiment(state):
    # 分析用户消息的情感
    sentiment = sentiment_analyzer(state["user_message"])
    return {"sentiment": sentiment}

def generate_response(state):
    # 根据情感生成回复
    if state["sentiment"] == "positive":
        response = "很高兴听到这个消息！"
    else:
        response = "我理解你的感受。"
    return {"bot_response": response}

def route_by_sentiment(state) -> Literal["positive_response", "negative_response"]:
    return "positive_response" if state["sentiment"] == "positive" else "negative_response"

案例 2：内容审核流程

python

class ModerationState(TypedDict):
    content: str
    is_safe: bool
    action: str

def check_content(state):
    # 检查内容是否安全
    is_safe = content_filter(state["content"])
    return {"is_safe": is_safe}

def approve_content(state):
    return {"action": "approved"}

def reject_content(state):
    return {"action": "rejected"}

def route_content(state) -> Literal["approve", "reject"]:
    return "approve" if state["is_safe"] else "reject"

案例 3：数据处理管道

python

class DataState(TypedDict):
    raw_data: str
    cleaned_data: str
    processed_data: dict
    quality_score: float

def clean_data(state):
    cleaned = remove_noise(state["raw_data"])
    return {"cleaned_data": cleaned}

def process_data(state):
    processed = transform(state["cleaned_data"])
    return {"processed_data": processed}

def check_quality(state):
    score = calculate_quality(state["processed_data"])
    return {"quality_score": score}

def route_quality(state) -> Literal["save_result", "retry_processing"]:
    return "save_result" if state["quality_score"] > 0.7 else "retry_processing"

🔍 常见问题

Q1: 为什么节点返回字典而不是直接返回值？

因为 LangGraph 需要知道更新哪个状态字段：

python

# ✅ 正确 - 明确指定更新的字段
def node(state):
    return {"graph_state": "new_value"}

# ❌ 错误 - LangGraph 不知道如何更新状态
def node(state):
    return "new_value"

Q2: 节点可以不返回任何值吗？

可以，但状态不会更新：

python

def node(state):
    print("This node only prints")
    # 不返回任何值，状态保持不变

适用于只执行副作用（如日志、通知）的节点。

Q3: 条件边的返回值必须是 Literal 类型吗？

不是必须的，但强烈推荐：

python

# 推荐 ✅ - 类型安全
def decide(state) -> Literal["node_a", "node_b"]:
    return "node_a"

# 可行但不推荐 - 无类型检查
def decide(state):
    return "node_a"  # 可能拼写错误导致运行时错误

Q4: 可以在节点之间共享变量吗？

不建议。应该通过状态传递数据：

python

# ❌ 不好的做法 - 全局变量
shared_data = []

def node_1(state):
    shared_data.append("data")
    return {}

# ✅ 好的做法 - 通过状态
class State(TypedDict):
    shared_data: list[str]

def node_1(state):
    return {"shared_data": state["shared_data"] + ["data"]}

Q5: START 和 END 节点可以省略吗？

不可以。它们是必需的特殊节点：

python

# ❌ 错误 - 缺少 START
builder.add_edge("node_1", "node_2")

# ✅ 正确
builder.add_edge(START, "node_1")
builder.add_edge("node_2", END)

📊 图的执行模式

invoke - 同步执行

python

result = graph.invoke({"graph_state": "input"})
# 阻塞直到图执行完成
print(result)

特点：

同步阻塞
返回最终状态
适合简单场景

stream - 流式执行（预告）

python

for step in graph.stream({"graph_state": "input"}):
    print(step)  # 逐步返回每个节点的输出

特点：

逐步返回结果
可以观察执行过程
适合长时间运行的图

📖 扩展阅读

🎉 总结

通过这个简单图教程，我们学习了：

LangGraph 核心概念

State：定义图的数据结构
Node：执行具体操作的函数
Edge：控制执行流程的连接
START/END：图的入口和出口

Python 技术要点

TypedDict：类型安全的字典定义
Literal：限定返回值的可能取值
类型提示：提升代码质量和可维护性

关键技能

构建简单的状态机
使用条件边实现动态路由
理解图的执行流程

下一步： 学习更复杂的图结构，如链式图、并行图、循环图等，构建更强大的 AI 应用！

LangGraph 最简图（Simple Graph）详细解读 ​

📚 概述 ​

🎯 核心概念 ​

什么是 LangGraph？ ​

图的基本结构 ​

🎭 实战案例：情绪生成器 ​

系统架构图 ​

🔧 代码实现详解 ​

1. 环境准备 ​

2. 定义状态（State） ​

Python 知识点：TypedDict ​

3. 定义节点（Nodes） ​

节点的核心特性 ​

4. 定义边（Edges） ​

普通边（Normal Edge） ​

条件边（Conditional Edge） ​

Python 知识点：Literal 类型 ​

条件边的工作原理 ​

5. 构建图（Graph Construction） ​

LangGraph 核心 API 详解 ​

特殊节点：START 和 END ​

可视化图结构 ​

6. 执行图（Graph Invocation） ​

invoke 方法详解 ​

同步执行特性 ​

🎓 核心知识点总结 ​

LangGraph 特有概念 ​

1. 状态（State） ​

2. 节点（Node） ​

3. 边（Edge） ​

4. 特殊节点 ​

5. 图的构建流程 ​

Python 特有知识点 ​

1. TypedDict ​

2. Literal 类型 ​

3. 类型提示（Type Hints） ​

💡 最佳实践 ​

1. 何时使用简单图？ ​

2. 状态设计原则 ​

原则 1：保持状态简洁 ​

原则 2：使用有意义的字段名 ​

原则 3：适当的字段类型 ​

3. 节点设计技巧 ​

技巧 1：单一职责 ​

技巧 2：添加日志 ​

技巧 3：错误处理 ​

4. 条件边的设计 ​

技巧 1：基于状态的路由 ​

技巧 2：多条件判断 ​

技巧 3：添加默认路由 ​

🚀 进阶扩展 ​

1. 多字段状态 ​

2. 嵌套状态结构 ​

3. 可选字段（Python 3.11+） ​

🎯 实际应用案例 ​

案例 1：简单的聊天机器人 ​

案例 2：内容审核流程 ​

案例 3：数据处理管道 ​

🔍 常见问题 ​

Q1: 为什么节点返回字典而不是直接返回值？ ​

Q2: 节点可以不返回任何值吗？ ​

Q3: 条件边的返回值必须是 Literal 类型吗？ ​

Q4: 可以在节点之间共享变量吗？ ​

Q5: START 和 END 节点可以省略吗？ ​

📊 图的执行模式 ​

invoke - 同步执行 ​

stream - 流式执行（预告） ​

📖 扩展阅读 ​

🎉 总结 ​

LangGraph 核心概念 ​

Python 技术要点 ​

关键技能 ​

LangGraph 最简图（Simple Graph）详细解读

📚 概述

🎯 核心概念

什么是 LangGraph？

图的基本结构

🎭 实战案例：情绪生成器

系统架构图

🔧 代码实现详解

1. 环境准备

2. 定义状态（State）

Python 知识点：TypedDict

3. 定义节点（Nodes）

节点的核心特性

4. 定义边（Edges）

普通边（Normal Edge）

条件边（Conditional Edge）

Python 知识点：Literal 类型

条件边的工作原理

5. 构建图（Graph Construction）

LangGraph 核心 API 详解

特殊节点：START 和 END

可视化图结构

6. 执行图（Graph Invocation）

invoke 方法详解

同步执行特性

🎓 核心知识点总结

LangGraph 特有概念

1. 状态（State）

2. 节点（Node）

3. 边（Edge）

4. 特殊节点

5. 图的构建流程

Python 特有知识点

1. TypedDict

2. Literal 类型

3. 类型提示（Type Hints）

💡 最佳实践

1. 何时使用简单图？

2. 状态设计原则

原则 1：保持状态简洁

原则 2：使用有意义的字段名

原则 3：适当的字段类型

3. 节点设计技巧

技巧 1：单一职责

技巧 2：添加日志

技巧 3：错误处理

4. 条件边的设计

技巧 1：基于状态的路由

技巧 2：多条件判断

技巧 3：添加默认路由

🚀 进阶扩展

1. 多字段状态

2. 嵌套状态结构

3. 可选字段（Python 3.11+）

🎯 实际应用案例

案例 1：简单的聊天机器人

案例 2：内容审核流程

案例 3：数据处理管道

🔍 常见问题

Q1: 为什么节点返回字典而不是直接返回值？

Q2: 节点可以不返回任何值吗？

Q3: 条件边的返回值必须是 Literal 类型吗？

Q4: 可以在节点之间共享变量吗？

Q5: START 和 END 节点可以省略吗？

📊 图的执行模式

invoke - 同步执行

stream - 流式执行（预告）

📖 扩展阅读

🎉 总结

LangGraph 核心概念

Python 技术要点

关键技能