LangGraph思维模型与首个图：实战构建跨平台爆款图文Agent第1篇

来源：互联网 2026-06-26 06:25:01

LangGraph是一个通过有向图结构替代线性链条和if/else条件判断来编排Agent工作流的框架。其核心要素包括状态、节点和边，利用StateGraph组织，经compile编译验证后调用invoke执行，并支持Mermaid图形可视化，从而高效实现循环、分支等复杂流程。

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

本章目标

读完这一章，你会发现——原来Agent的工作流，并不需要靠一堆if/else硬撑。你会收获这些能力：

LangGraph思维模型与首个图：实战构建跨平台爆款图文Agent第1篇

长期稳定更新的攒劲资源： >>>点此立即查看<<<

随口就能跟人讲清楚"为什么Agent非要用图，而不是线性链条"
能自己定义一个StateGraph，写几个node函数，用edge把它们串起来
能编译（compile）这个图，再调用（invoke）它，拿到结果
还能用Mermaid图把画的图结构可视化出来，一目了然

知识讲解

从一个生活例子开始

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

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

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

每个环节就是一个分拣站，站与站之间由传送带连接。包裹在每个站被处理完，身上的标签就会更新——比如从"未分类"变成"华东区"——然后顺着传送带滑向下一个站。

关键来了：有些包裹的下一站并不是固定的。如果标签上写着"易碎品"，传送带就会把它分流到"易碎品专用通道"；如果是"加急件"，则直接跳到"优先装车站"。这个决策是在运行时根据标签内容做的，并不是事先定死的。

LangGraph把这种"图"搬到了代码里：

快递中心	LangGraph	说明
包裹上的标签	State（状态）	记录当前所有信息，在节点间流转
分拣站	Node（节点）	一个Python函数，读标签、做处理、更新标签
传送带	Edge（边）	连接两个节点，决定数据往哪走
整个分拣系统	StateGraph	把所有节点和边组织起来的容器
启动前的线路检查	compile()	验证图结构没有死循环、没有断头路
把包裹放进系统	invoke()	传入初始数据，启动执行

思考一下：如果包裹的下一站完全由当前标签内容决定（比如"易碎品走专用通道"），这对应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的核心模型极其简单，只有三个要素：

State（状态）——一个数据结构，所有节点共享。每个节点可以读取它、返回部分更新来修改它。它就像一个背包，在工作流的每个环节之间传递。
Node（节点）——一个Python函数，签名是(state: State) -> dict。它接收当前完整状态，返回一个部分更新字典——只包含你想修改的字段。LangGraph会自动把返回值合并到状态中。
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内部发生了这些事：

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

每一步都是确定性的——同一个输入、同一个图，会产生同一个输出。这种确定性对于调试和测试非常关键。

代码实战

环境准备

在开始之前，打开终端安装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传递数据，每个节点只关心自己的输入和输出，不需要知道上游和下游的细节。

本章小结

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

关键术语

术语	释义
StateGraph	LangGraph的核心类，用于定义、构建、编译一个有向图工作流
State	图中所有节点共享的数据结构（TypedDict / Pydantic），在节点间流转
Node（节点）	一个`(state) -> dict`签名的Python函数，读取State返回部分更新
Edge（边）	定义节点之间的流向，固定边（`add_edge`）总是从A到B
START	内置常量，标记图的入口。必须有一条边从START出发
END	内置常量，标记图的出口。到达END的边意味着执行结束
compile()	验证图结构（边连通性、节点存在性）并返回可执行Runable
invoke()	传入初始State同步执行图，返回最终State