📚 LangGraph 学习笔记 - 从入门到精通

---

🎯 第 1 章：LangGraph 概述（核心概念）

1.1 什么是 LangGraph？【⭐⭐⭐ 重点】

一句话理解：

LangGraph = 用 “流程图” 的方式来写 AI 应用程序

就像你画流程图一样：

• 节点（Node） = 每个要做的事情（比如调用 LLM、搜索网络、查数据库）
• 边（Edge） = 事情之间的先后顺序（做完 A 再做 B，或者 A 做完可以选 B 或 C）
• 状态（State） = 整个流程中共享的 “笔记本”，所有节点都能读写


LangGraph 能做什么？

1. ✅ 持久化执行：程序断了能从断点继续，不用从头再来
2. ✅ 人机协作：可以中途暂停，让人来审核决策，再继续
3. ✅ 记忆管理：支持多轮对话上下文，像 ChatGPT 一样有记忆
4. ✅ 流式输出：LLM 打字效果实时展示
5. ✅ 生产级部署：适合复杂、长时间运行的 AI 应用

---

1.2 LangGraph vs LangChain 区别【⭐⭐⭐ 重点】

对比项	LangGraph	LangChain
定位	低级、精细控制	高级、开箱即用
状态管理	内置自动管理	需要自己写代码管理
执行方式	图结构、可循环、可分支	线性链条、一条路走到黑
持久化	原生支持	需要额外开发
适用场景	复杂智能体、多轮对话	简单的链式调用

大白话总结：

当你的 AI 应用需要 “循环、分支、记忆、暂停继续” 这些复杂逻辑时，LangChain 很难写优雅，必须用 LangGraph！

---

🚀 第 2 章：快速入门（5 步学会写 LangGraph）

0.1 环境安装

# 创建环境
conda create --name=langgraph_venv python=3.12
conda activate langgraph_venv

# 安装核心包
pip install langgraph==1.0.5

---

0.2 写第一个 LangGraph 程序【⭐⭐⭐ 重点】

记住这 5 步，所有 LangGraph 程序都这么写！

步骤1：定义 State（共享笔记本）
步骤2：定义 Node（每个节点做什么）
步骤3：创建图，添加节点和边（画流程图）
步骤4：编译图（把流程图变成可执行程序）
步骤5：调用运行

完整示例（RAG 问答系统）：

# ========== 步骤1：定义 State（共享笔记本） ==========
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class MyState(TypedDict):
    query: str           # 用户问题
    rag_result: str      # 知识库检索结果
    web_search_result: str  # 网络搜索结果
    final_answer: str    # 最终回答

# ========== 步骤2：定义 Node（每个节点做什么） ==========
def rag_search_node(state: MyState):
    """节点1：从知识库检索"""
    query = state["query"]
    return {"rag_result": f"关于{query}的知识库结果"}

def web_search_node(state: MyState):
    """节点2：从网络搜索"""
    query = state["query"]
    return {"web_search_result": f"关于{query}的网络结果"}

def final_answer_node(state: MyState):
    """节点3：LLM总结回答"""
    return {"final_answer": f"基于知识库和网络搜索的回答"}

# ========== 步骤3：画流程图（添加节点和边） ==========
graph = StateGraph(state_schema=MyState)

graph.add_node(rag_search_node)    # 添加3个节点
graph.add_node(web_search_node)
graph.add_node(final_answer_node)

graph.add_edge(START, "rag_search_node")    # 开始 → 知识库检索
graph.add_edge(START, "web_search_node")    # 开始 → 网络搜索（并行！）
graph.add_edge("rag_search_node", "final_answer_node")  # 检索完 → 总结
graph.add_edge("web_search_node", "final_answer_node")  # 搜索完 → 总结
graph.add_edge("final_answer_node", END)    # 总结完 → 结束

# ========== 步骤4：编译 ==========
compiled_graph = graph.compile()

# ========== 步骤5：运行 ==========
result = compiled_graph.invoke({"query": "如何使用LangGraph"})
print(result)

💡 理解要点：

• 两个搜索节点并行执行！这就是图的优势
• 每个节点只返回自己修改的部分，不是整个 state
• 所有节点共享同一个 state 字典

---

🧠 第 3 章：状态（State）—— LangGraph 的灵魂【⭐⭐⭐ 最重要】

状态是 LangGraph 最核心、最容易搞错的概念，必须彻底理解！

3.1 状态的定义

3.1.1 三种定义方式（推荐第一种）

方式 1：TypedDict（推荐）

from typing import TypedDict

class MyState(TypedDict):
    query: str
    answer: str

方式 2：Pydantic BaseModel（带数据校验）

from pydantic import BaseModel

class MyState(BaseModel):
    query: str
    answer: str

方式 3：Dataclass

from dataclasses import dataclass

@dataclass
class MyState():
    query: str
    answer: str

---

3.1.2 输入输出隔离【⭐⭐ 重要】

可以限制：

• 用户只能输入哪些字段（input_schema）
• 最终只返回哪些字段（output_schema）

用户输入 → [input_schema过滤] → 图内部完整state → [output_schema过滤] → 返回给用户

class MyStateFull(TypedDict):
    query: str
    rag_result: str
    web_result: str
    final_answer: str
    internal_key: str  # 内部使用，用户看不到

class InputSchema(TypedDict):
    query: str  # 用户只能传query

class OutputSchema(TypedDict):
    final_answer: str  # 只返回最终答案

graph = StateGraph(
    state_schema=MyStateFull,
    input_schema=InputSchema,
    output_schema=OutputSchema
)

---

3.1.4 Reducer 函数【⭐⭐⭐ 重点难点】

问题： 两个节点同时修改同一个字段，怎么合并？

答案： 用 Reducer 函数定义合并规则

from typing import Annotated, List
from langchain_core.messages import BaseMessage

# 自定义合并函数：把新消息追加到列表末尾
def add_message(old_list: list, new_list: list):
    return old_list + new_list

class MyState(TypedDict):
    # Annotated[类型, 合并函数]
    messages: Annotated[List[BaseMessage], add_message]

三种 Reducer：

1. 默认（覆盖）：不指定时，后执行的覆盖先执行的
2. 内置 add_messages：专门用于消息列表，自动追加
3. 自定义：自己写合并逻辑

---

3.2 状态的存储（Checkpointer）【⭐⭐⭐ 核心功能】

为什么需要存储状态？

场景 1：多轮对话保持上下文

第一次问：“北京天气怎么样？” → 第二次问：“适合出去玩吗？”

如果不存状态，第二次就忘了第一次说了什么

场景 2：断点续传

程序执行到一半报错了，修复 bug 后从断点继续，不用从头跑

---

怎么用？3 步搞定

# 1. 导入Checkpointer
from langgraph.checkpoint.memory import InMemorySaver  # 内存存储
from langgraph.checkpoint.sqlite import SqliteSaver    # 数据库持久化

# 2. 创建实例
checkpointer = InMemorySaver()

# 3. 编译时传入
graph = builder.compile(checkpointer=checkpointer)

# 4. 调用时传 thread_id（会话ID）
# 同一个thread_id = 同一个会话，共享状态
result = graph.invoke(
    {"messages": "你好"},
    config={"configurable": {"thread_id": "user_1_session_1"}}
)

💡 thread_id 理解：

• 就像 “房间号”，同一个房间里的人共享记忆
• 不同用户用不同 thread_id，互不干扰
• 同一个用户新开对话用新 thread_id

---

从故障中恢复

# 第一次执行，node2报错了，但node1已经执行完了
graph.invoke({}, config={"configurable": {"thread_id": "123"}})

# ===== 修复bug后 =====
# 传 None + 相同 thread_id，自动从断点继续
graph.invoke(None, config={"configurable": {"thread_id": "123"}})

---

3.3 底层运行原理（Pregel 算法）【了解即可】

LangGraph 不是传统的有向无环图（DAG），而是基于 Pregel 算法：

核心概念：

• Actors（节点）：干活的人
• Channels（通道）：节点之间传消息的管道

执行流程（超步 SuperStep）：

1. Plan：看哪些通道有新消息，决定执行哪些节点
2. Execute：并行执行所有选中的节点
3. Update：把节点输出写入通道
4. 重复，直到没有节点需要执行

💡 这就是为什么能支持循环！

传统 DAG 只能往前走一次，Pregel 可以反复触发节点执行

---

🔧 第 4 章：节点（Node）

4.1 节点的输入输出

节点能接收 3 个参数：

def my_node(state, config, runtime):
    """
    state:   共享状态字典（必选）
    config:  配置信息（thread_id等，可选）
    runtime: 运行时对象（依赖注入、流式输出等，可选）
    """
    pass

❌ 节点输出的大坑！【⭐⭐⭐ 必看】

错误写法（千万不要）：

def bad_node(state):
    state['result'] = 'xxx'
    return state  # ❌ 不要返回整个state！

正确写法：

def good_node(state):
    return {"result": "xxx"}  # ✅ 只返回你修改的部分

为什么？

LangGraph 会自动把节点输出和全局状态合并。如果你返回整个 state，会导致重复合并、数据混乱！

---

4.2 特殊节点

• START：虚拟起点节点，标记图的入口
• END：虚拟终点节点，标记图的结束

就是两个字符串常量，不用自己定义。

---

4.3 节点缓存

对于耗时的节点（比如调用 API），可以缓存结果：

from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache

builder.add_node(
    "expensive_node", 
    my_function,
    cache_policy=CachePolicy(ttl=60)  # 缓存60秒
)

graph = builder.compile(cache=InMemoryCache())

---

4.4 节点重试

对于不稳定的 API 调用，可以自动重试：

from langgraph.types import RetryPolicy

builder.add_node(
    "unstable_api",
    my_api_call,
    retry_policy=RetryPolicy(max_attempts=5)  # 最多重试5次
)

---

4.5 流式输出（5 种模式）【⭐⭐ 实用】

模式	用途
values	每步输出完整状态
updates	只输出增量更新
custom	节点内自定义输出（比如进度条）
messages	LLM Token 流式输出（打字效果）
debug	所有调试信息

# LLM打字效果
for chunk, metadata in graph.stream(input, stream_mode="messages"):
    print(chunk.content, end="")

---

4.6 人工审核节点（Interrupt）【⭐⭐⭐ 高级功能】

场景： 转账、发布内容等重要操作，需要人工确认

from langgraph.types import interrupt, Command

def review_node(state):
    # 生成待审核数据
    data = {"amount": 100, "recipient": "张三"}
    
    # 在这里暂停，把data传给用户审核
    user_decision = interrupt({
        "title": "转账审核",
        "data": data
    })
    
    # 用户审核后，程序从这里继续执行
    if user_decision["approved"]:
        return {"status": "已转账"}
    else:
        return {"status": "已取消"}

# ===== 使用方式 =====
# 1. 第一次调用，执行到interrupt暂停
result = graph.invoke(initial_state, config=config)
print(result["__interrupt__"][0].value)  # 展示给用户审核

# 2. 用户审核后，传Command继续执行
final = graph.invoke(
    Command(resume={"approved": True}),
    config=config
)

---

🔗 第 5 章：边（Edge）

5.1 两种边

普通边：A 执行完一定执行

graph.add_edge("node_a", "node_b")

条件边：A 执行完，根据状态决定走 B 还是 C

def router(state):
    if state["value"] % 2 == 0:
        return "node_b"
    else:
        return "node_c"

graph.add_conditional_edges("node_a", router)

---

5.2 可控循环（REACT 模式）【⭐⭐⭐ Agent 核心】

用条件边实现循环：

model → 需要工具？→ tools → model → 需要工具？→ tools → ... → 输出
        ↓
      output

⚠️ 防死循环！

设置递归限制，默认 25 步，超过就报错：

graph.invoke(input, config={"recursion_limit": 50})

---

学习总结

核心知识点（必须掌握）

1. 5 步编程法：定义 State → 写 Node → 加边 → 编译 → 调用
2. State 三大要点：TypedDict 定义、Reducer 合并、Checkpointer 持久化
3. thread_id：同一个 ID 共享记忆，不同 ID 隔离
4. 节点输出：只返回增量，不要返回整个 state
5. 条件边：实现分支和循环

下一步建议

1. 把快速入门的代码跑一遍
2. 试着写一个简单的 Agent：LLM 判断是否需要搜索，需要就搜索，不需要就直接回答
3. 学习 LangGraph 官方文档的更多高级用法