LangGraph学习笔记2-快速入门

AI代码蜂巢X

探索编程的无限可能

编辑：嘉禾

📌 目标

• • 理解 LangGraph 是什么以及为什么需要它
• • 掌握 LangGraph 的核心三要素：State、Node、Edge
• • 完成第一个 LangGraph 应用：简单计数器

🤔 什么是 LangGraph？

问题背景

在构建 AI 应用时，我们经常遇到这样的需求：

用户输入 → 处理A → 处理B → 处理C → 输出结果

传统的做法是写一个线性流程：

def process(user_input):
    result_a = process_a(user_input)
    result_b = process_b(result_a)
    result_c = process_c(result_b)
    return result_c

但是，当需求变复杂时：

• • 需要根据条件选择不同的处理路径
• • 需要循环执行某些步骤
• • 需要在多个处理节点间共享状态
• • 需要支持人机交互（中途暂停/恢复）

传统代码会变得难以维护。

LangGraph 的解决方案

LangGraph 将应用抽象为一个有向图：

┌─────────────────────────────────────────────────────┐
│                                                     │
│    START ──→ [节点A] ──→ [节点B] ──→ [节点C] ──→ END │
│                                                     │
│    每个节点可以访问和修改共享的 State                 │
│    边定义了执行的顺序和条件                          │
│                                                     │
└─────────────────────────────────────────────────────┘

核心优势：

🏗️ 核心三要素

1️⃣ State（状态）

State 是在节点间传递的共享数据结构。

from typing_extensions import TypedDict

class CounterState(TypedDict):
    count: int           # 计数器值
    history: list[str]   # 执行历史

关键点：

• • 使用 TypedDict 定义类型
• • 所有节点都可以访问和修改 State
• • 节点返回的更新会合并到 State 中（不是替换）

2️⃣ Node（节点）

Node 是执行具体任务的函数。

def increment_node(state: CounterState):
    """每次调用，计数器+1"""
    new_count = state["count"] + 1
    message = f"节点执行: count从 {state['count']} 增加到 {new_count}"

    return {
        "count": new_count,
        "history": state["history"] + [message]
    }

关键点：

• • 接收当前 State 作为参数
• • 返回需要更新的字段（不是整个 State）
• • 返回值会与当前 State 合并

3️⃣ Edge（边）

Edge 定义节点间的执行顺序。

# 普通边：固定路径
workflow.add_edge(START, "increment")      # START → increment
workflow.add_edge("increment", "double")   # increment → double
workflow.add_edge("double", "report")      # double → report
workflow.add_edge("report", END)           # report → END

关键点：

• • START 和 END 是特殊的内置节点
• • 普通边表示固定的执行顺序
• • 条件边可以根据 State 动态选择路径（后续章节讲解）

💻 完整示例：简单计数器

让我们通过一个完整的例子来理解这些概念。

流程图

START
  ↓
[increment]  ← 计数器+1
  ↓
[double]     ← 计数器×2
  ↓
[report]     ← 报告结果
  ↓
 END

完整代码

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

# 1️⃣ 定义状态结构
class CounterState(TypedDict):
    count: int
    history: list[str]

# 2️⃣ 定义节点函数
def increment_node(state: CounterState):
    """每次调用，计数器+1"""
    new_count = state["count"] + 1
    message = f"节点执行: count从 {state['count']} 增加到 {new_count}"
    return {
        "count": new_count,
        "history": state["history"] + [message]
    }

def double_node(state: CounterState):
    """计数器翻倍"""
    new_count = state["count"] * 2
    message = f"节点执行: count从 {state['count']} 翻倍到 {new_count}"
    return {
        "count": new_count,
        "history": state["history"] + [message]
    }

def report_node(state: CounterState):
    """报告最终结果"""
    message = f"最终结果: count = {state['count']}"
    return {"history": state["history"] + [message]}

# 3️⃣ 创建图
workflow = StateGraph(CounterState)

# 4️⃣ 添加节点
workflow.add_node("increment", increment_node)
workflow.add_node("double", double_node)
workflow.add_node("report", report_node)

# 5️⃣ 连接节点（定义执行流程）
workflow.add_edge(START, "increment")
workflow.add_edge("increment", "double")
workflow.add_edge("double", "report")
workflow.add_edge("report", END)

# 6️⃣ 编译图
app = workflow.compile()

# 7️⃣ 运行图
initial_state = {
    "count": 5,
    "history": ["开始执行"]
}

result = app.invoke(initial_state)

# 8️⃣ 查看结果
print("\n执行历史:")
for step in result["history"]:
    print(f"  - {step}")
print(f"\n最终计数: {result['count']}")

执行过程解析

初始状态: count = 5, history = ["开始执行"]

执行 increment_node:
  - count: 5 → 6
  - history: ["开始执行", "节点执行: count从 5 增加到 6"]

执行 double_node:
  - count: 6 → 12
  - history: [..., "节点执行: count从 6 翻倍到 12"]

执行 report_node:
  - count: 12 (不变)
  - history: [..., "最终结果: count = 12"]

最终结果: count = 12

🔑 关键理解

State 的合并机制

重要：节点返回值是合并到 State，而不是替换整个 State！

# 假设当前 State 是：
state = {"count": 5, "history": ["a", "b"]}

# 节点返回：
return {"count": 10}  # 只更新 count

# 合并后的 State：
state = {"count": 10, "history": ["a", "b"]}  # history 保持不变

节点函数的规范

def my_node(state: MyState):
    # ✅ 正确：只返回需要更新的字段
    return {"field1": new_value}

    # ❌ 错误：不要返回整个 state
    # return state  # 这会导致问题

    # ❌ 错误：不要直接修改 state
    # state["field1"] = new_value  # 这会导致不可预期的问题
    # return {}

📊 图的构建流程

┌────────────────────────────────────────────────────────┐
│                  LangGraph 构建流程                      │
├────────────────────────────────────────────────────────┤
│                                                        │
│  Step 1: 定义 State                                    │
│     └─ class MyState(TypedDict): ...                  │
│                                                        │
│  Step 2: 定义 Node 函数                                │
│     └─ def node_func(state: MyState): ...             │
│                                                        │
│  Step 3: 创建 StateGraph                               │
│     └─ workflow = StateGraph(MyState)                 │
│                                                        │
│  Step 4: 添加节点                                      │
│     └─ workflow.add_node("name", node_func)           │
│                                                        │
│  Step 5: 添加边                                        │
│     └─ workflow.add_edge("from", "to")                │
│                                                        │
│  Step 6: 编译图                                        │
│     └─ app = workflow.compile()                       │
│                                                        │
│  Step 7: 运行图                                        │
│     └─ result = app.invoke(initial_state)             │
│                                                        │
└────────────────────────────────────────────────────────┘

🎯 小结