第1章：图的力量——LangGraph 思维模型与首个图实战

本章学习目标

读完本章，你将发现——构建AI Agent工作流完全无需依赖繁复的if/else堆砌。你将收获以下核心能力：

• 能清晰阐述“AI Agent为何必须采用图结构而非线性链条”
• 能够独立定义一个StateGraph，编写多个节点函数，使用边将它们串联起来
• 能够编译（compile）该图，并通过invoke调用它获取最终结果
• 还能借助Mermaid图将图结构可视化，一目了然地展示设计

知识讲解

从一个生活实例切入

想象你走进一家快递分拣中心。

包裹从卡车上卸下，它的旅程是这样的：

包裹到达 → 扫码分类 → 按地区分拣 → 装车 → 出发

每个环节相当于一个分拣站，站点之间由传送带连接。包裹在每个站点被处理完毕，身上的标签信息就会更新——例如从“未分类”变为“华东区”——随后顺着传送带滑向下一个站点。

关键之处在于：有些包裹的下一站并非固定不变。若标签显示“易碎品”，传送带便将其分流至“易碎品专用通道”；若是“加急件”，则直接跳转至“优先装车站”。这个决策是在运行过程中根据标签内容动态做出的，而非事先硬性规定。

LangGraph将这种“图”模型映射到代码中：

想一想：如果包裹的下一站完全由当前标签内容决定（比如“易碎品走专用通道”），这对应于LangGraph中的哪个概念？答案稍后揭晓。

工作原理

为什么线性链条不够用？

假设你正在用代码编写一个简单的“内容分析助手”。没有LangGraph时，你很可能写出这样的流程：

def analyze_content(url: str) -> str:
    raw = fetch_content(url)        # 1. 抓取内容
    cleaned = clean_html(raw)       # 2. 清洗HTML
    summary = summarize(cleaned)    # 3. 生成摘要
    return summary

这是一个纯粹的线性流程——水管式，必须执行完第1步才能到第2步，再到第3步。对于简单的ETL任务，这种方式确实够用。

但若增加一个需求：当摘要质量不足时需要重新生成，线性代码立刻变得别扭：

def analyze_content(url: str) -> str:
    raw = fetch_content(url)
    cleaned = clean_html(raw)
    summary = summarize(cleaned)
    if quality_score(summary) < 0.7:
        # 不满意？
        summary = summarize(cleaned)   # 再来一次
    if quality_score(summary) < 0.7:
        # 还不满意？
        summary = summarize(cleaned)   # 第三次...
    return summary

问题显而易见：你不得不使用if嵌套来模拟循环，代码可读性急剧下降。而实际场景中，“不满意就重试”可能需要重复3次、5次，甚至需要人工介入——用线性代码硬写这种逻辑，很快就会变得杂乱无章。

更糟糕的是：如果需求变为“抓取完成后，根据内容类型走不同分支——视频内容走视频分析流程，图文内容走图文分析流程”，嵌套的if会迅速失控。

这正是LangGraph要解决的痛点：让你用“图”的方式描述复杂工作流——包含循环、分支、并行——而不是依靠if/else硬编码。

StateGraph是如何工作的？

LangGraph的核心模型极为简洁，仅包含三个要素：

1. State（状态）——一个数据结构，所有节点共享。每个节点可以读取它，并通过返回部分更新来修改它。它就像一个背包，在工作流的各个阶段之间传递。
2. Node（节点）——一个Python函数，签名是(state: State) -> dict。它接收当前完整状态，返回一个部分更新字典——只包含你想修改的字段。LangGraph自动将返回值合并到状态中。
3. Edge（边）——决定“从A节点之后去哪里”。固定边（add_edge）意味着总是从A走到B；条件边（add_conditional_edges）意味着根据状态内容动态选择下一站。

一个最小的LangGraph程序如下所示：

START → node_a → node_b → END

用“启动→处理→汇总→结束”来理解：

• 传入初始数据给START
• node_a进行第一轮处理
• node_b进行第二轮处理
• END结束

思考一下：上面的图是否存在循环？没有——它只是一个顺序流程。那么什么时候会出现循环？当某个条件边的返回值指向了前面的节点时，就形成了循环。例如“质量检查不通过→回到生成节点”——这正是第2章要深入探讨的内容。

执行流程：invoke到底做了什么？

当你调用graph.invoke(initial_state)时，LangGraph内部发生以下步骤：

1. 从START出发，找到第一条边连接的目标节点
2. 执行该节点的函数，传入当前状态
3. 将该函数的返回值（一个dict）合并到状态中
4. 根据边的定义，找到下一个节点
5. 重复步骤2-4，直到到达END
6. 返回最终状态

每一步都是确定性的——相同的输入、相同的图，必然产生相同的输出。这种确定性对于调试和测试至关重要。

代码实战

环境准备

开始之前，请在终端中安装LangGraph：

pip install -U langgraph

基础版：最简两节点图

我们从整个教程中最简单的图开始——两个节点，一条边，没有任何分支。

打开编辑器，新建文件chapter01_hello_graph.py：

"""第1章 基础演示：最简两节点LangGraph
LocalTrend项目起点——一个能跑通的空骨架
"""
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# ============================================================
# Step 1: 定义State（图的状态数据结构）
# ============================================================
class LocalTrendState(TypedDict):
    """LocalTrend项目的全局状态。
    所有节点共享这个结构，每个节点只返回自己需要更新的字段。
    """
    raw_content: str   # 原始内容（模拟抓取结果）
    summary: str       # 分析摘要

# ============================================================
# Step 2: 定义Node函数（图的处理单元）
# ============================================================
def fetch_content(state: LocalTrendState) -> dict:
    """模拟从平台抓取内容。
    接收当前状态，返回一个部分更新dict。
    """
    print("[fetch_content] 正在抓取内容...")
    # 实际项目中这里会调用requests.get()或网页抓取工具
    # 本章先用模拟数据建立图的结构直觉
    return {"raw_content": "LangGraph 是构建 AI Agent 的强大框架，它用图来编排复杂工作流。"}

def generate_summary(state: LocalTrendState) -> dict:
    """对抓取到的内容生成摘要。"""
    print(f"[generate_summary] 正在分析内容: {state['raw_content'][:30]}...")
    content = state["raw_content"]
    # 实际项目中这里会调用LLM，第2章会讲
    summary = f"分析结果：这篇内容的核心主题是'{content[:20]}'"
    return {"summary": summary}

# ============================================================
# Step 3: 构建StateGraph，添加节点和边
# ============================================================
def build_graph() -> StateGraph:
    """构建LocalTrend的最简图骨架。"""
    # 创建图容器，指定状态类型
    builder = StateGraph(LocalTrendState)

    # 添加节点：给每个函数起一个名字
    builder.add_node("fetch_content", fetch_content)
    builder.add_node("generate_summary", generate_summary)

    # 添加边：定义执行顺序
    # START → fetch_content → generate_summary → END
    builder.add_edge(START, "fetch_content")
    builder.add_edge("fetch_content", "generate_summary")
    builder.add_edge("generate_summary", END)

    # 编译：验证图结构并返回可执行对象
    return builder.compile()

# ============================================================
# Step 4: 运行
# ============================================================
if __name__ == "__main__":
    graph = build_graph()

    # invoke() 接收初始状态，返回最终状态
    # 注意：raw_content 和 summary 都传空字符串作为初始值
    result = graph.invoke({"raw_content": "", "summary": ""})

    print("\n=== 最终状态 ===")
    print(f"raw_content: {result['raw_content']}")
    print(f"summary: {result['summary']}")

在你的终端中执行以下命令：

python chapter01_hello_graph.py

你应该会看到类似这样的输出：

[fetch_content] 正在抓取内容...
[generate_summary] 正在分析内容: LangGraph 是构建 AI Agent 的强大框架...
=== 最终状态 ===
raw_content: LangGraph 是构建 AI Agent 的强大框架，它用图来编排复杂工作流。
summary: 分析结果：这篇内容的核心主题是'LangGraph 是构建 AI Age'

逐行解析

class LocalTrendState(TypedDict)

这就是我们的“包裹标签”。TypedDict是Python标准库提供的一种类型提示方式——它定义了字典应该包含哪些键、每个键是什么类型。LangGraph利用它来确保节点之间传递的数据结构保持一致。

注意，我们只定义了两个字段：raw_content和summary。State应尽量精简——字段过多意味着每个节点都需要理解更多上下文，增加认知负担。随着教程的深入，我们会逐步为State添加新字段。

def fetch_content(state) -> dict

这是第一个节点。其签名至关重要：接收完整的State，返回一个dict。返回的dict仅包含你想更新的字段——无需返回整个State。LangGraph会自动将你返回的{"raw_content": "..."}合并到全局状态中。

builder = StateGraph(LocalTrendState)

创建图容器。这一行告诉LangGraph：“我要构建一个图，其状态结构是LocalTrendState。”此后所有节点的读写都会参照这个结构。

builder.add_node("fetch_content", fetch_content)

将函数注册为节点，节点名称为"fetch_content"。节点名称很重要——后续添加边、定义条件路由时都要使用该名称来引用。名称可以任意取，但建议与函数名保持一致以降低理解成本。

builder.add_edge(START, "fetch_content")

START是LangGraph内置的特殊常量，代表图的入口。这条边表示“图的执行从fetch_content节点开始”。如果忘记添加此边，编译时会报错——LangGraph不知道从哪里启动执行。

builder.add_edge("generate_summary", END)

END同样是内置常量。连接到END的节点执行完后，图即结束。

builder.compile()

这一行完成多项工作：检查所有边连接的目标节点是否存在、检查是否存在不可达节点、检查循环是否合理。如果图结构存在问题，它会在此处抛出明确错误，而非等到运行时报错。编译通过=图结构正确。

graph.invoke({"raw_content": "", "summary": ""})

invoke()是启动图执行的方法。传入的字典必须是完整的初始State——所有TypedDict定义的字段都要提供。执行完毕后，invoke()返回最终状态的完整字典。

可视化你的图

LangGraph内置了生成Mermaid图的能力。在chapter01_hello_graph.py的if __name__ == "__main__":块中，graph = build_graph()之后加入以下代码：

# 生成Mermaid图并保存
mermaid_code = graph.get_graph().draw_mermaid()
with open("chapter01_graph.md", "w", encoding="utf-8") as f:
    f.write(mermaid_code)
print("图已保存到 chapter01_graph.md")

这段代码会将图的Mermaid描述保存到Markdown文件中。在VS Code中安装“Markdown Preview Mermaid Support”插件后，即可直接预览图形。你应该能看到：

START → fetch_content → generate_summary → END

养成习惯：每次修改图结构后，先将其画出来看一眼——很多问题肉眼就能发现。

扩展版：增加预处理节点

上述基础版仅包含一个“抓取→生成摘要”流程。但实际项目中，抓取到的内容通常需要先进行清洗——去除HTML标签、截断过长文本、处理乱码等。

接下来，我们为LocalTrend添加一个清洗节点，同时演示多节点串行：

"""第1章 扩展演示：三节点串行处理流水线
在基础版上增加一个内容清洗节点，展示多节点图的构建模式
"""
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class LocalTrendState(TypedDict):
    raw_content: str
    cleaned_content: str   # ? 新增：清洗后的内容
    summary: str

def fetch_content(state: LocalTrendState) -> dict:
    """模拟抓取——返回含HTML标签的原始内容"""
    print("[fetch_content] 正在抓取...")
    return {"raw_content": "
LangGraph教程
这是一个很棒的框架。
"}

def clean_content(state: LocalTrendState) -> dict:
    """清洗内容：去除HTML标签。"""
    raw = state["raw_content"]
    print(f"[clean_content] 清洗中... 原始长度: {len(raw)} 字符")
    # 简易HTML去标签——实际项目建议用BeautifulSoup
    import re
    cleaned = re.sub(r"<[^>]+>", "", raw).strip()
    print(f"[clean_content] 清洗后长度: {len(cleaned)} 字符")
    return {"cleaned_content": cleaned}

def generate_summary(state: LocalTrendState) -> dict:
    """基于清洗后的内容生成摘要"""
    # 优先用清洗后的内容
    content = state.get("cleaned_content") or state["raw_content"]
    print(f"[generate_summary] 分析内容: {content[:40]}...")
    return {"summary": f"摘要：{content[:50]}..."}

def build_graph() -> StateGraph:
    builder = StateGraph(LocalTrendState)
    builder.add_node("fetch_content", fetch_content)
    builder.add_node("clean_content", clean_content)
    builder.add_node("generate_summary", generate_summary)

    # 三节点串行流水线
    builder.add_edge(START, "fetch_content")
    builder.add_edge("fetch_content", "clean_content")
    builder.add_edge("clean_content", "generate_summary")
    builder.add_edge("generate_summary", END)
    return builder.compile()

if __name__ == "__main__":
    graph = build_graph()
    result = graph.invoke({
        "raw_content": "",
        "cleaned_content": "",
        "summary": "",
    })
    print(f"\n=== 最终状态 ===")
    print(f"raw: {result['raw_content'][:60]}...")
    print(f"cleaned: {result['cleaned_content']}")
    print(f"summary: {result['summary']}")

运行后你会看到三个节点按顺序执行。关键观察：clean_content读取了fetch_content写入的raw_content，而generate_summary读取了clean_content写入的cleaned_content。这就是图的威力——节点之间通过State传递数据，每个节点只关心自己的输入和输出，无需知晓上下游的细节。

本章小结

1. LangGraph将复杂工作流建模为图——节点执行处理，边控制流向，State在它们之间传递数据。
2. State是以TypedDict定义的数据结构，所有节点共享。节点通过返回dict（而非直接修改state）来更新它。
3. StateGraph是图容器；add_node注册节点函数；add_edge定义固定流向。
4. START和END是内置常量，标记图的入口和出口。忘记START边会导致图无法启动。
5. compile()验证图结构，invoke()同步执行图。先编译后执行——编译通过表示结构正确。
6. 每个节点只返回自己修改的字段——LangGraph自动将部分更新合并到全局State。
7. get_graph().draw_mermaid()可以可视化图结构——养成“画出来看看”的习惯。

关键术语