LangGraph入门指南

Python入萌

83人浏览 · 2026-07-13 09:11:56

Python入萌 · 2026-07-13 09:11:56 发布

文章目录

一、LangGraph是什么？为什么一定要学它？

1.1 核心定位

LangGraph是一个低阶智能体编排框架+运行时环境，专门用来构建、管理和部署有状态、可长期运行的AI智能体（Agent）。

大白话理解：你可以把它当成一个「AI流水线搭建工具」，所有AI应用的流程都可以拆解成一张图：

每个步骤是一个工位（节点）
步骤之间的流向是传送带（边）
全程共享的工作记录是台账（状态）

不管是并行搜索、循环调用工具、人工审核断点，还是多智能体协作，都可以用LangGraph灵活搭建。

1.2 和LangChain的核心区别

很多同学会把两者搞混，直接一张表讲清楚差异：

特性	LangGraph	LangChain
抽象级别	低阶，细粒度控制，流程完全自定义	高阶，开箱即用的链式/Agent封装
状态管理	内置状态机+检查点，原生支持记忆	需要自己手动管理状态
执行模型	图结构，支持并行、条件分支、循环	线性链式执行，很难做复杂分支
持久化	原生支持断点续跑、故障恢复	需要额外自己实现
适用场景	复杂有状态的智能体、生产级应用	简单链式调用、快速原型Demo

一句话总结：简单线性流程用LangChain足够，只要你的应用需要「循环、分支、跨轮记忆、人工介入」，LangGraph就是必选方案。

二、三大核心概念

LangGraph的所有应用，都是由状态、节点、边三个核心元素组成的，我们用「工厂流水线」的类比一次性讲透。

2.1 状态（State）：贯穿全程的共享工作台账

状态就是一本跟着流水线从头走到尾的共享笔记本，所有节点都可以读取里面的信息，也可以把自己的处理结果写进去。它就是LangGraph的“大脑记忆”，用户输入、中间结果、最终输出全部存在这里。

① 三种状态定义方式

Python里有三种常见的状态定义方式，新手推荐直接用第一种：

TypedDict（官方首推）
本质是带类型约束的字典，相当于给笔记本画好固定格子，每个格子标好“这里存用户问题”“这里存搜索结果”。写代码时IDE会自动提示键名，写错会直接标红，避免拼写错误导致的运行bug，是LangGraph最常用的写法。
```
from typing import TypedDict

class MyState(TypedDict):
    query: str          # 用户输入的问题
    rag_result: str     # 知识库检索结果
    final_answer: str   # 最终答案
```
Pydantic BaseModel
不仅定义类型，还会在运行时强制校验数据格式，填错直接报错，适合要求严格的生产环境。
dataclass
自动生成初始化、打印等样板代码，写法简洁，适合简单数据结构。

② 状态更新规则（Reducer）

每个字段都可以单独设定“写入规则”，一共三类：

默认覆盖：新值直接替换旧值，适合最终结果、单次更新的字段
追加更新：新内容接在旧内容后面，不会覆盖历史数据。比如聊天记录、搜索结果列表，用内置的add_messages、operator.add就能实现
自定义规则：自己写合并逻辑，比如给消息加前缀、给结果加序号等

③ 状态隔离：不该看的不看

LangGraph支持精细化的权限控制：

图输入/输出隔离：只允许指定字段进入/流出工作流，防止恶意数据污染内部状态，也避免中间数据泄露
节点级隔离：每个节点只能访问自己需要的字段，比如搜索节点看不到用户隐私信息，各司其职，降低出错概率

2.2 节点（Node）：各司其职的工作工位

节点就是流水线里的一个个工位，每个节点只干一件事：可以是调用大模型、执行工具、检索知识库，或者任何自定义逻辑。

【重点】
节点必须返回增量更新，也就是只返回自己修改的那几个字段，绝对不要返回整个状态对象。
如果两个并行节点都返回完整state，系统会不知道该以哪个版本为准，直接触发更新冲突报错。就像两个厨师同时端一整桌菜上来，服务员根本不知道该上哪份。

✅ 正确写法：

# 只返回当前节点要更新的字段
def web_search_node(state: MyState):
    query = state["query"]
    result = f"关于{query}的网络搜索结果"
    return {"web_result": result}

❌ 错误写法：

# 直接返回整个状态，并行时会报错
def web_search_node(state: MyState):
    state["web_result"] = f"关于{state['query']}的搜索结果"
    return state

2.3 边（Edge）：控制流程的传送路线

边用来定义节点之间的执行顺序，一共两种：

普通边：固定路线，干完A节点直接去B节点，没有其他选择
条件边：动态分支路线，根据当前状态的内容决定下一步去哪，本质就是编程里的if-else。比如用户问技术问题就去查知识库，问售后问题就转人工。

三、核心进阶能力：搞定生产级需求

3.1 状态持久化（Checkpointer）：自动存档+断点续传

默认情况下，每次运行LangGraph都是全新的空白状态，之前的对话历史、执行进度全部清零。Checkpointer就是自动存档机制：

每执行完一个节点，自动把当前状态快照存起来，支持内存、SQLite、PostgreSQL、MongoDB等多种存储介质
thread_id就是存档编号：同一个编号对应同一份存档，多次调用用同一个thread_id，就能自动延续之前的上下文，实现多轮对话记忆
断点续传：流程跑到一半报错崩溃，修复bug后，用同一个thread_id重新调用，就能从断掉的节点继续执行，不用从头再来

3.2 节点四大实用特性

① 节点缓存

相同输入的节点，在缓存有效期内直接返回之前的结果，不用重新执行。适合调用大模型、查数据库这种耗时耗成本的操作，大幅提升效率、降低成本。

② 节点重试

针对网络超时、API限流这种临时性故障，可以配置自动重试次数，不用一次失败就整个流程崩溃。默认只重试临时故障，代码bug这种永久错误不会重复执行。

③ 流式输出

不用等整个流程跑完再返回结果，支持边执行边输出：

大模型生成答案时逐字输出（打字机效果）
实时返回当前执行进度（正在检索→正在生成→完成）
支持values/updates/custom/messages等多种流式模式，按需选择

④ 人工审核中断（interrupt）

涉及转账、删数据、发布内容等关键操作时，可以让流程自动暂停，把待审核数据抛给人工，等人工确认/修改后，再从断点继续执行。

3.3 可控循环：避免死循环

通过条件边可以搭建循环流程，最典型的就是「大模型判断→调用工具→大模型再判断」的Agent循环。
为了防止逻辑错误导致无限循环，LangGraph提供了recursion_limit递归限制，设定最大执行步数，超过就直接抛出异常终止流程，避免程序卡死。

四、实战入门：从零搭建并行RAG问答工作流

我们做一个入门Demo：用户提问后，同时执行知识库检索和网络搜索，两个任务并行跑完后，大模型汇总生成最终答案。

4.1 环境准备

先安装依赖（推荐用uv管理Python环境）：

uv add langgraph
# 可选：打印流程图需要
uv add grandalf

4.2 完整代码+逐行注释

import time
from typing import TypedDict
from langgraph.constants import START, END
from langgraph.graph import StateGraph

# ========== 1. 定义图状态（共享台账） ==========
class MyState(TypedDict):
    query: str                  # 用户输入的问题
    rag_result: str             # 知识库检索结果
    web_search_result: str      # 网络搜索结果
    final_answer: str           # 最终汇总答案

# ========== 2. 定义节点函数（各个工位） ==========
# 知识库检索节点
def rag_search_node(state: MyState):
    print("📚【技术知识库】开始检索学术定义...")
    query = state.get("query")
    # 模拟调用RAG知识库
    time.sleep(2)
    rag_result = f"📚 【学术定义】{query}：指大语言模型生成的内容看似流畅合理，但实际与事实不符、甚至完全虚构的现象。"
    print("📚【技术知识库】检索完成！")
    # 只返回增量更新的字段
    return {"rag_result": rag_result}

# 网络搜索节点
def web_search_node(state: MyState):
    print("🌐【实时网络搜索】全网查询最新案例...")
    query = state["query"]
    # 模拟调用网络搜索
    time.sleep(2)
    web_search_result = f"🌐 【通俗解释】{query}：常被戏称为AI“一本正经胡说八道”，比如捏造不存在的论文、虚假的历史事件。"
    print("🌐【实时网络搜索】搜索完毕！")
    return {"web_search_result": web_search_result}

# 最终答案汇总节点
def final_answer_node(state: MyState):
    print("🤖【AI助手】正在综合多方信息生成回答...")
    rag_result = state["rag_result"]
    web_search_result = state["web_search_result"]
    # 模拟大模型汇总
    final_answer = f"""
【AI助手总结】：
{rag_result}
{web_search_result}
需要我教你如何降低模型的幻觉率吗？
"""
    time.sleep(2)
    print("🤖【AI助手】回答生成结束！")
    return {"final_answer": final_answer}

# ========== 3. 搭建图结构 ==========
# 3.1 基于状态创建图实例
graph = StateGraph(MyState)

# 3.2 添加所有节点
graph.add_node("rag_search_node", rag_search_node)
graph.add_node("web_search_node", web_search_node)
graph.add_node("final_answer_node", final_answer_node)

# 3.3 添加边（定义流程）
# 起点同时流向两个搜索节点（并行执行）
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)

# 3.4 编译图（生成可运行的工作流）
compiled_graph = graph.compile()

# ========== 4. 运行工作流 ==========
if __name__ == "__main__":
    # 传入初始状态，启动工作流
    result = compiled_graph.invoke({"query": "大模型中的“幻觉”是什么意思?"})
    # 打印最终答案
    print("\n" + "="*50)
    print(result["final_answer"])
    
    # 可选：打印ASCII流程图
    print("\n工作流结构图：")
    compiled_graph.get_graph().print_ascii()