第 2 章 最小可运行图

本章目标：跑通第一张 LangGraph 图，并吃透它的三个生命周期阶段——定义、编译、运行，以及最容易误解的一件事：节点返回的是"补丁"，不是"全量状态"。

2.1 一张图的三个阶段

LangGraph 中的图和"源代码 → 可执行文件 → 进程"的关系惊人地相似：

• 定义阶段：声明状态结构、注册节点、连接边。此时它只是一份"图纸"，不可执行。
• 编译阶段：compile() 对图纸做校验（悬空节点、不可达路径），生成一个结构固定的运行时对象。编译之后不能再增删节点——这份"固定性"是运行时可以做并发调度和持久化的前提。
• 运行阶段：对编译产物调用 invoke，引擎按拓扑关系执行节点，最后返回最终状态。

先建立这个三段式的心智模型，后面所有 API 都可以按"它属于哪个阶段"来归类记忆。

2.2 第一张图：不依赖模型的纯机械流水线

我们延续简历筛选场景。第一步先做一个最朴素的需求：从简历文本里抽取关键信息，再生成一段格式化摘要。为了让注意力集中在图的机制上，这一版不调用模型，两个节点都是普通 Python 函数：

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

# ---- 1. 定义状态结构 ----
class ScreenState(TypedDict):
    resume_text: str      # 原始简历文本
    candidate: dict       # 抽取出的关键信息
    summary: str          # 格式化摘要


# ---- 2. 定义节点：普通函数，入参是完整状态，返回值是状态更新 ----
def extract(state: ScreenState) -> dict:
    """从简历文本中抽取关键信息（朴素实现）"""
    lines = [ln.strip() for ln in state["resume_text"].splitlines() if ln.strip()]
    name = lines[0].split("|")[0].strip() if lines else "未知"
    skills = [ln for ln in lines if ln.startswith("- 技能")]
    candidate = {
        "name": name,
        "skills": skills[0].replace("- 技能：", "").split("、") if skills else [],
    }
    return {"candidate": candidate}


def summarize(state: ScreenState) -> dict:
    """根据抽取结果生成格式化摘要"""
    c = state["candidate"]
    summary = f"候选人 {c['name']}，掌握技能：{'、'.join(c['skills']) or '未识别'}"
    return {"summary": summary}


# ---- 3. 组装图 ----
builder = StateGraph(ScreenState)
builder.add_node("extract", extract)
builder.add_node("summarize", summarize)

builder.add_edge(START, "extract")        # 入口：先执行 extract
builder.add_edge("extract", "summarize")  # extract 完成后执行 summarize
builder.add_edge("summarize", END)        # summarize 完成后结束

# ---- 4. 编译并运行 ----
graph = builder.compile()

resume = """李思远 | 求职意向：Python 后端工程师
- 本科，计算机科学与技术，6 年开发经验
- 技能：Python、FastAPI、PostgreSQL、Redis、Docker
- 近期项目：电商订单中台重构，QPS 从 800 提升至 3000
"""

result = graph.invoke({"resume_text": resume})
print(result["summary"])
# 候选人 李思远，掌握技能：Python、FastAPI、PostgreSQL、Redis、Docker

保存运行，控制台输出摘要。这个例子只有两个节点，却已经覆盖了 LangGraph 最核心的五个 API。下面逐一拆开。

2.3 逐个拆解五个核心构件

状态：TypedDict 声明"图级共享数据结构"

ScreenState 描述了整张图共享的数据有哪些字段。它不是某个节点的私有数据——每个节点执行时，都能读到完整状态；执行完，通过返回值把更新写回去。

用 TypedDict 只是习惯写法，Pydantic BaseModel 和 dataclass 也可以（第 3 章会对比三种方式）。状态字段建议在初始 invoke 输入中提供，或在节点中首次写入。

节点：签名约定

节点就是一个函数：

• 入参：当前完整状态（LangGraph 会把 dict 传进来）
• 返回值：一个 dict，表示"要更新哪些字段"

注意返回值不需要是完整状态。extract 只返回 {"candidate": ...}，resume_text 保持不变。这就是"增量更新"语义——下一节详细展开。

add_node：把函数注册为图中节点

第一个参数是节点名（字符串，是边引用它的唯一标识），第二个参数是可调用对象。除了普通函数，任何 Runnable（比如一个 prompt + llm 链）都可以直接注册为节点——这对已熟悉 LangChain 的读者是个好消息：你写的任何 Runnable 链，一行代码就能变成图里的一个节点。

add_edge / START / END：连接顺序

add_edge("A", "B") 声明"A 完成后执行 B"。START 和 END 是两个保留节点：

• START 是虚拟入口，add_edge(START, "extract") 声明了图的第一个真实节点
• END 是虚拟出口，连到它意味着"流程可以在此结束"

一张合法的图，至少要有一条从 START 出发、最终能到达 END 的路径。

compile：从图纸到可执行对象

builder.compile() 返回编译产物。此后节点和边的拓扑固定不可变——想改结构只能重新定义、重新编译。编译产物是可复用的：invoke 多次、给不同输入、在不同线程里并发调用同一个编译产物，完全没问题。

2.4 关键语义：节点返回的是"补丁"

这是新手最容易误解的一点，值得单独一节。

有人会把节点写成这样：

def summarize(state: ScreenState) -> dict:
    return {"resume_text": state["resume_text"],
            "candidate": state["candidate"],
            "summary": "..."}   # 把整个状态原样带回去

功能上没错，但语义上多余。节点的返回值是"对状态的补丁"，引擎只合并补丁里出现的字段，其余字段原样保留：

这个设计有两个直接好处：

1. 节点只声明自己关心的变更，读取的数据不需要"带出来"，代码更短
2. 为并行做准备——两个并行节点各自返回补丁，引擎负责合并。如果大家都返回全量状态，并行合并时就会互相覆盖（第 3 章的 reducer 会把这件事讲透）

2.5 给图换上真正的 LLM 节点

现在把 summarize 换成真正调用模型的版本，感受"普通函数节点"和"Runnable 链节点"的等价性：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

load_dotenv()
llm = ChatOpenAI(model=os.getenv("LLM_MODEL", "gpt-4o-mini"), temperature=0)

comment_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是资深技术招聘官，用一句话点评候选人，突出最亮眼的点。"),
    ("human", "候选人信息：{candidate}"),
])

def summarize(state: ScreenState) -> dict:
    chain = comment_prompt | llm          # prompt -> llm 的 Runnable 链
    response = chain.invoke({"candidate": state["candidate"]})
    return {"summary": response.content}

# 组装与编译部分和上一版完全相同
builder = StateGraph(ScreenState)
builder.add_node("extract", extract)
builder.add_node("summarize", summarize)
builder.add_edge(START, "extract")
builder.add_edge("extract", "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()

result = graph.invoke({"resume_text": resume})
print(result["summary"])
# 例：李思远六年后端经验，主导订单中台将 QPS 提升至 3000，工程能力扎实突出。

节点内部怎么组织逻辑（裸函数、Runnable 链、还要不要再拆）是节点自己的事，图只关心"给它状态、它返回补丁"这个契约。

2.6 看见你的图：编译产物的自描述

编译产物能输出自己的结构。draw_mermaid() 返回一段 Mermaid 文本，粘到任何支持 Mermaid 的编辑器即可渲染：

print(graph.get_graph().draw_mermaid())

输出类似：

flowchart TD
    __start__([__start__]):::startEnd
    extract(extract)
    summarize(summarize)
    ...

这个能力在调试复杂图（十几二十个节点）时非常实用——代码里的图和脑子里的图是否一致，一画便知。

2.7 编译产物是一个 Runnable

对已会 LangChain 的读者，这里有个让人踏实的事实：编译产物实现了 Runnable 协议。这意味着：

from langchain_core.runnables import Runnable

isinstance(graph, Runnable)   # True

invoke / ainvoke / stream / astream / batch / abatch 全部可用。你过往对 Runnable 的一切认知——批处理、异步、嵌套进另一个链——都可以直接套用在编译产物上。例如把这张图当成一个节点，嵌进另一张更大的图里（第 9 章子图的基础）：

results = graph.batch(
    [{"resume_text": r} for r in [resume, resume2, resume3]]
)

2.8 运行时发生了什么：一次 invoke 的生命周期

把 invoke 展开看，引擎做的是一个非常朴素的循环：

LangGraph 底层的执行引擎叫 Pregel，源自 Google 2010 年的同名图计算论文：把执行切成一个个"超步"（superstep），每个超步里找出所有当前可执行的节点并行跑，跑完统一合并状态，再进入下一个超步。本章的图是线性的，看起来只是顺序执行；到第 5 章引入并行分支后，你会看到超步模型真正的威力。

常见坑与排错

1. 节点返回 None 之外的非法类型。节点必须返回 dict（或者第 6 章的 Command 对象）。返回字符串、列表都会在运行时报 InvalidUpdateError。最常见的翻车：节点函数里忘了 return，默认返回 None——这是合法的（表示"不更新任何字段"），但如果你本意是更新状态，就会表现为"字段 mysteriously 没变化"。

2. 忘记连 END。add_node 注册了节点却没有任何出边连向 END，编译期或运行期会直接报错。看到"节点不可达"或"图无出口"类错误，先检查边是否画全。

3. 状态字段拼写不一致。TypedDict 的字段名和节点返回的 key 不一致（比如状态定义用 summary，节点返回 summery），运行时会静默新增一个字段而不报错——表现为"下游节点读不到数据"。用 Pydantic 定义状态可以更早暴露这类问题。

4. 把编译产物当一次性对象。有人每次 invoke 前都重新 compile()。没有必要：编译开销不小，且编译产物本身是线程安全、可复用的。正确姿势是模块级编译一次，到处 invoke。

5. 异步节点与同步调用混用。节点定义成 async def 却用同步 invoke 调用可以工作（引擎会兜底），但性能路径不佳；统一使用 ainvoke 或统一用同步节点。

动手练习

1. 在 2.5 节的图里新增一个 translate 节点：把 summary 翻译成英文存入 summary_en 字段，接在 summarize 之后。只改状态定义和图组装部分，summarize 节点代码一行不动。

2. 把 extract 和 summarize 合并成一个节点 process，观察：功能上仍可运行。思考——什么样的逻辑值得拆成两个节点？（提示：单独观测、单独重试、单独复用）

3. 用 graph.batch([...]) 一次传入三份虚构简历，确认批量执行可用。再试试把批量的并发行为和手写 for 循环对比，思考 batch 底层做了什么。

本章速查