lang

1. 本章目标 #

上一章讲清了两件事：create_agent 本身就是一张 LangGraph 图；什么时候该自己动手写图。这一章开始动手，从零搭出你的第一张图。

好消息是，LangGraph 的核心概念只有三个：状态、节点、边。节点就是普通 Python 函数，不必继承基类，也不必加装饰器。真正需要花点力气理解的，只有「状态怎么在节点之间流动」这一件事。

本章目标：

搭出一张最小的两节点图，把状态、节点、边这三个概念的边界摸清楚——包括那些不报错、却会悄悄出问题的地方。

学完这一章，你应该能做到：

用 StateGraph + add_node + add_edge + compile 四步搭出一张能跑的图
说清「节点返回的字典」和「图的完整状态」是什么关系
在 TypedDict、Pydantic、dataclass 三种状态写法里做出选择，知道它们在输入校验上差在哪，也知道哪些问题是三者都管不了的
认识 START 和 END 这两个特殊节点，明白为什么只有前者是必须的
用 draw_mermaid() 把图画出来，并知道什么情况下画出来的图会误导你
认出四种静默失败：不抛异常，但图的行为和你想的不一样

参考文档：

2. 三个概念 #

2.1. 一句话版 #

概念	一句话	对应代码
状态 State	一个字典，图运行期间所有节点共享它	`class MyState(TypedDict): ...`
节点 Node	一个普通函数：读状态，返回要改的字段	`def clean(state) -> dict: ...`
边 Edge	规定「这个节点跑完，下一个跑谁」	`builder.add_edge("a", "b")`

用一句话串起来：

图 = 一堆函数（节点） + 它们的执行顺序（边） + 它们之间传递的数据（状态）。

如果觉得抽象，可以套一个类比：工厂流水线。 节点是工位上的工人，每人只做一道工序；边是传送带，决定零件下一站交给谁；状态是那个在流水线上一路传下去的托盘——每个工人从托盘上取自己要的零件，把加工结果放回托盘，而不是直接把东西递给下一个人。

这个类比还能顺便解释一件事：为什么工人之间不直接交接，非要走托盘？ 因为一旦要在中间插一道新工序，走托盘的方案里其他工人完全不用改——这正是 §4.1 要说的解耦。

2.2. 为什么不直接写 f(g(x)) #

初学最容易冒出的疑问是：三个节点顺序执行，我直接写 assign(parse(raw)) 不就完了，绕这么大一圈干什么？

如果需求真的只是「顺序跑三个函数」，那确实不需要图。图的价值体现在下面这些地方——普通函数调用要么做不到，要么写起来很难看：

能力	普通函数调用	图
顺序执行	直接写就行	也行，但代码更多
条件分支	`if/else` 也行	`add_conditional_edges`，分支关系能画出来（第 23 章）
循环并控制次数	`while` 也行	有内置的步数上限保护（第 24 章）
中途暂停等人工审批，几小时后继续	很难，要自己存状态	内置（第 26 章）
每一步的中间状态自动持久化	要自己写	内置 `checkpointer`（第 11 章）
流式观察每个步骤的进展	要自己埋点	内置 `stream`（第 27 章）
出错后从失败的那一步恢复	要自己做	内置
把整个流程画成图给同事看	靠口头讲	`draw_mermaid()`

所以更准确的说法是：图不是为了「让函数按顺序跑」，而是为了给流程加上持久化、可暂停、可观测、可视化这几样能力。 如果你的流程一样都不需要，那确实不该用图（见第 19 章 §5.2 的反向清单）。

本章先把最基础的顺序执行搭起来；上面那些能力从第 23 章起再逐个加。

2.3. 状态是怎么流动的 #

这是本章唯一需要认真琢磨的机制。先看一张示意图：

invoke({"text": "  你好  "})
        │
        ▼
   ┌─────────────────────────────┐
   │ 状态: {"text": "  你好  "}   │
   └─────────────────────────────┘
        │
        ▼  clean 节点：读到完整状态，只返回 {"text": "你好"}
   ┌─────────────────────────────┐
   │ 状态: {"text": "你好"}       │  ← 图把返回值【合并】进状态
   └─────────────────────────────┘
        │
        ▼  count 节点：读到更新后的状态，只返回 {"n": 2}
   ┌─────────────────────────────┐
   │ 状态: {"text": "你好", "n": 2}│
   └─────────────────────────────┘
        │
        ▼
   invoke 的返回值就是最终状态

三条规则，记住就够用：

节点收到的是完整状态，不是上一个节点的返回值
节点只需返回要修改的字段，没提到的字段保持不变
图负责把返回值合并进状态，你不需要手动拼

第 2 条是最容易搞错的地方。很多人以为节点必须返回完整状态，其实不用——返回一个只有一两个键的小字典就行。

第 1 条和第 3 条合起来，还能推出一个不太直观的结论：节点函数的返回值和图的返回值，含义正好相反。

	拿到 / 返回的是什么
节点函数收到的参数	完整状态
节点函数的返回值	部分更新（只有你要改的那几个键）
`graph.invoke()` 的返回值	完整状态（不是最后一个节点的返回值）

中间那行是「部分」，上下两行是「完整」。刚开始写图时，这三者最容易混在一起；哪里跑得不对劲，回来对一下这张表。

3. 第一张图 #

3.1. 四步走 #

搭任何一张图都是这四步，顺序固定：

① 定义状态   class MyState(TypedDict): ...
② 写节点函数 def my_node(state) -> dict: ...
③ 组装       builder.add_node(...) / builder.add_edge(...)
④ 编译       graph = builder.compile()

为什么顺序是固定的？ 因为每一步都要用上一步的产物：StateGraph(S) 得先有状态类 S，add_node 得先有函数，add_edge 得先有节点名，compile() 得等边都连完。这也意味着四步里漏掉哪一步，症状都不一样——§7.1 和 §10 会把这些症状一一对上。

3.2. 完整代码 #

一个「清洗文本 → 统计字数」的两节点图。：

# TypedDict 用来声明「这个字典有哪些键、每个键是什么类型」
from typing import TypedDict

# START/END 是图的入口出口标记，StateGraph 是图的构建器
from langgraph.graph import END, START, StateGraph

# ① 定义状态：这张图运行期间要传递哪些数据
class S(TypedDict):
    # 待处理的文本，由外部输入
    text: str
    # 文本长度，由 count 节点写入
    n: int

# ② 写节点：普通函数，参数是状态，返回要更新的字段
def clean(state: S) -> dict:
    # 把收到的完整状态打印出来，方便观察状态是怎么流动的
    print("   [clean] 收到 state:", state)
    # 只返回 text，没提到的 n 会保持原样
    return {"text": state["text"].strip()}

# 第二个节点：统计清洗后的字数
def count(state: S) -> dict:
    # 这里打印出来的 text 应该已经是清洗过的，可以验证状态确实被更新了
    print("   [count] 收到 state:", state)
    # len() 算的是清洗后的长度，因为节点收到的是最新状态
    return {"n": len(state["text"])}

# ③ 组装：先声明节点，再连边
builder = StateGraph(S)
# 注册第一个节点，"clean" 是节点名，clean 是函数本身（注意没有括号）
builder.add_node("clean", clean)
# 注册第二个节点
builder.add_node("count", count)
# START 是图的入口，必须有这条边，否则图不知道从哪开始
builder.add_edge(START, "clean")
# clean 跑完接着跑 count，这是一条无条件边
builder.add_edge("clean", "count")
# END 是图的出口，走到这里就结束
builder.add_edge("count", END)

# ④ 编译：得到一个可运行的图
graph = builder.compile()

# invoke 传入初始状态，返回跑完之后的完整状态
print("结果:", graph.invoke({"text": "  你好世界  "}))
# 打印类型，和第 19 章 create_agent 返回的类型做个对照
print("类型:", type(graph).__name__)

运行输出：

   [clean] 收到 state: {'text': '  你好世界  '}
   [count] 收到 state: {'text': '你好世界'}
结果: {'text': '你好世界', 'n': 4}
类型: CompiledStateGraph

跑通了。这段输出里有三个值得注意的地方：

第一，count 节点收到的 text 已经是清洗过的。 这就是 §2.3 说的「节点收到的是完整状态」——clean 的修改已经合并进去了。

第二，clean 收到的 state 里根本没有 n 这个键。 看输出 {'text': ' 你好世界 '}，只有一个键。原因是 invoke 时只传了 text，而 TypedDict 不会自动填默认值。所以在节点里写 state["n"] 会直接 KeyError，这个坑 §4.3 会细讲。

第三，type(graph).__name__ 是 CompiledStateGraph，和上一章 create_agent 返回的类型一模一样。你现在搭出来的东西，和 create_agent 是同一种对象。

3.3. 拆解 #

代码	在做什么
`class S(TypedDict)`	声明状态的字段名和类型，相当于图的数据契约
`StateGraph(S)`	创建一个构建器，告诉它状态长什么样
`add_node("clean", clean)`	注册节点：第一个参数是节点名（字符串），第二个是函数
`add_edge(START, "clean")`	从入口连到 `clean`，这条边不写图就编译不过
`add_edge("clean", "count")`	`clean` 跑完接着跑 `count`
`add_edge("count", END)`	`count` 跑完就结束
`builder.compile()`	检查图是否合法，产出可运行的图
`graph.invoke({...})`	传入初始状态，跑完返回最终状态

注意 builder 和 graph 是两样东西：

StateGraph 是图纸，compile() 之后才是能跑的机器。 图纸阶段可以随便加节点、加边；编译之后结构就固定了。

两个容易踩的细节：

第一，add_node("clean", clean) 的第二个参数不能带括号。 写成 add_node("clean", clean()) 的意思是「立刻调用这个函数」，而我们要传的是函数本身。这种写错之后的报错信息往往离出错的地方很远，值得单独记一下。

第二，add_node 返回的是 builder 自己，所以可以链式调用：

# 状态与节点的定义和 §3.2 相同，这里只换建图的写法
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    return {"text": state["text"].strip()}

def count(state: S) -> dict:
    return {"n": len(state["text"])}

# add_node / add_edge 都返回 builder 本身，因此可以串起来写
graph = (
    # 起点仍然是创建构建器
    StateGraph(S)
    # 注册第一个节点
    .add_node("clean", clean)
    # 注册第二个节点
    .add_node("count", count)
    # 连入口边
    .add_edge(START, "clean")
    # 连中间边
    .add_edge("clean", "count")
    # 连出口边
    .add_edge("count", END)
    # 最后编译，这一步返回的才是可运行的图
    .compile()
)

# 跑一遍，结果和 §3.2 完全一致
print("结果:", graph.invoke({"text": "  你好世界  "}))

运行输出：

结果: {'text': '你好世界', 'n': 4}

4. 状态（State） #

4.1. 状态是图的共享数据 #

状态的作用是让节点之间传数据。节点函数彼此不直接调用，也不互相传参，只跟状态打交道：

节点 A ──写──► 状态 ◄──读── 节点 B

这个设计有一个很实际的好处：加一个节点，不用改其他节点的签名。 换成 assign(parse(raw)) 这种写法，中间插一步就得改函数签名和整条调用链；用状态的话，只要在状态里加个字段就行。

4.2. 只返回要改的字段 #

节点返回的是部分更新（partial update），不是完整状态。下面四种返回值都合法，跑一遍就能看清各自的效果：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

# 四个节点分别演示四种合法返回值
# 一：只改一个字段，其余字段保持不变
def one(state: S) -> dict:
    return {"text": "新值"}

# 二：一次改多个字段，返回多个键即可
def many(state: S) -> dict:
    return {"text": "新值", "n": 10}

# 三：什么都不改，返回空字典
def nothing(state: S) -> dict:
    return {}

# 四：什么都不改，返回 None（不写 return 语句时 Python 就是返回 None）
def none(state: S) -> None:
    print("   [none] 只做点副作用，不碰状态")

# 把四个节点各建一张单节点的图，跑同一份初始状态做对照
for name, node in [("只改一个字段", one), ("改多个字段", many),
                   ("返回空字典", nothing), ("返回 None", none)]:
    builder = StateGraph(S)
    # 节点名随便起，这里统一叫 n
    builder.add_node("n", node)
    builder.add_edge(START, "n")
    builder.add_edge("n", END)
    # 每轮都重新编译一张图，变量名统一用 graph
    graph = builder.compile()
    print(f"{name}: {graph.invoke({'text': 'a', 'n': 1})}")

运行输出：

只改一个字段: {'text': '新值', 'n': 1}
改多个字段: {'text': '新值', 'n': 10}
返回空字典: {'text': 'a', 'n': 1}
   [none] 只做点副作用，不碰状态
返回 None: {'text': 'a', 'n': 1}

后两种都没有报错，状态原样保留。

「返回 None 也合法」这件事，在写「只做副作用的节点」时很有用，比如只负责打日志、发通知的节点：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    owner: str

# 假装是发短信的第三方接口
def send_sms(who: str) -> None:
    print(f"   [短信] 已通知 {who}")

# 返回类型标注成 None，明确表达「这个节点不改状态」
def notify(state: S) -> None:
    """只发通知，不改状态。"""
    # 副作用：发短信，读状态但不写状态
    send_sms(state["owner"])
    # 不需要 return，Python 默认返回 None

builder = StateGraph(S)
builder.add_node("notify", notify)
builder.add_edge(START, "notify")
builder.add_edge("notify", END)
graph = builder.compile()

# 状态进去什么样，出来还是什么样
print("结果:", graph.invoke({"owner": "财务组"}))

运行输出：

   [短信] 已通知 财务组
结果: {'owner': '财务组'}

但返回值只能是字典或 None，返回别的类型会直接报错。 这是唯一一种「返回值不合法」的情况：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str

# 错误示范：返回字符串
def bad(state: S):
    # 图不知道该把这个字符串合并到哪个字段上
    return "我是字符串"

builder = StateGraph(S)
builder.add_node("bad", bad)
builder.add_edge(START, "bad")
builder.add_edge("bad", END)
graph = builder.compile()

# 预期会抛异常，用 try 包住好看清报错原文
try:
    graph.invoke({"text": "a"})
except Exception as e:
    print("报错:", type(e).__name__, str(e))

运行输出：

报错: InvalidUpdateError Expected dict, got 我是字符串
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/INVALID_GRAPH_NODE_RETURN_VALUE

这个报错很好认，而且它属于本章少数「立刻报错」的情况，比后面那些静默失败友好得多。最常见的触发原因是把节点写成了「返回处理结果」而不是「返回状态更新」，比如 return state["text"].strip() 少包了一层字典。

4.3. 三种状态写法 #

状态可以用 TypedDict、Pydantic 的 BaseModel 或 dataclass 三种写法。它们最关键的差别不在语法，而在要不要校验输入。

这个选择比看上去更重要：它决定了非法输入是在门口就被挡住，还是一路混进来、到某个节点里才炸。 后一种情况在图变大之后特别难排查，因为报错的位置和真正的原因往往隔着好几个节点。

4.3.1 TypedDict：不校验 #

用它写一张两节点的图，再用四种输入试探它的底线：

# TypedDict 来自标准库，不需要装任何东西
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

# 继承 TypedDict 就是在声明「状态有哪些键」
class S(TypedDict):
    # 字段名和类型，运行时不生效，只给 IDE 和类型检查器看
    text: str
    # 同上，声明成 int 也不会阻止你传别的类型进来
    n: int

# 节点里用字典语法访问
def clean(state: S) -> dict:
    # 打印收到的状态，方便观察哪些键真的进来了
    print("   [clean] 收到 state:", state)
    # 注意是 state["text"] 这种方括号写法
    return {"text": state["text"].strip()}

# 第二个节点，用来确认状态确实往下传了
def count(state: S) -> dict:
    print("   [count] 收到 state:", state)
    return {"n": len(state["text"])}

builder = StateGraph(S)
builder.add_node("clean", clean)
builder.add_node("count", count)
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# 四种输入依次试，看 TypedDict 能拦住哪些
CASES = [
    # 少传一个字段
    ("只传 text，不传 n", {"text": " hi "}),
    # 连必需的 text 都没有
    ("什么都不传", {}),
    # 多传一个 S 里没声明的键
    ("传了状态里没声明的键", {"text": " hi ", "多余键": 1}),
    # 类型不对：text 声明的是 str，这里传 int
    ("类型不对（text 传 int）", {"text": 123}),
]
for title, payload in CASES:
    print(f"\n{title}：")
    # 有两种输入会抛异常，用 try 包住才能一次跑完四组
    try:
        print("结果:", graph.invoke(payload))
    except Exception as e:
        print("报错:", type(e).__name__, e)

运行输出：

只传 text，不传 n：
   [clean] 收到 state: {'text': ' hi '}
   [count] 收到 state: {'text': 'hi'}
结果: {'text': 'hi', 'n': 2}

什么都不传：
   [clean] 收到 state: {}
报错: KeyError 'text'

传了状态里没声明的键：
   [clean] 收到 state: {'text': ' hi '}
   [count] 收到 state: {'text': 'hi'}
结果: {'text': 'hi', 'n': 2}

类型不对（text 传 int）：
   [clean] 收到 state: {'text': 123}
报错: AttributeError 'int' object has no attribute 'strip'

四个结论，后两个很多人没意识到：

少传字段不报错，只要没有节点去读它就相安无事
真正报错时，错在节点内部（KeyError: 'text'），而不是在入口。你看到的现象是「某个节点炸了」，得自己往回倒推：是不是入口少传了字段
多传的键被静默丢掉了。 注意第三组输出，clean 收到的 state 里完全没有「多余键」，最终结果里也没有。入参的键名拼错，和节点返回值的键名拼错（§5.3）其实是同一种静默失败，只是发生的位置在入口
类型不对也不拦，而且报什么错取决于节点怎么用这个值。这里 clean 调了 .strip()，所以报 AttributeError；如果节点拿它去做算术，报的就是 TypeError。同一个原因，会表现成各种不同的报错

TypedDict 的类型标注只对 IDE 和类型检查器有意义，运行时完全不生效。 它不会填默认值，不会拦住类型错误的输入，也不会提醒你多传了键。

4.3.2 Pydantic：会校验 #

# Pydantic 是 LangChain 已经依赖的库，不用额外安装
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel

# 继承 BaseModel，Pydantic 会在构造时做运行时校验
class PS(BaseModel):
    # 没有默认值就是必填字段
    text: str
    n: int = 0  # 可以有默认值

# 节点里用属性语法访问，注意是 state.text 不是 state["text"]
def pnode(state: PS) -> dict:
    # 打印出来能看到状态实例的类型是 PS，不是 dict
    print("   节点收到:", state, type(state).__name__)
    # 读字段用点号，返回值仍然是普通字典
    return {"n": len(state.text)}

# 建图的写法完全不变，只是状态类换成了 Pydantic 模型
builder = StateGraph(PS)
builder.add_node("pnode", pnode)
builder.add_edge(START, "pnode")
builder.add_edge("pnode", END)
graph = builder.compile()

# 同样试四种输入
CASES = [
    ("正常输入", {"text": "abc"}),
    ("类型错误的输入（n 传字符串）", {"text": "abc", "n": "不是数字"}),
    ("缺必填字段", {}),
    ("传了模型里没声明的键", {"text": "abc", "多余键": 1}),
]
for title, payload in CASES:
    print(f"\n{title}：")
    try:
        print("结果:", graph.invoke(payload))
    except Exception as e:
        print("报错:", type(e).__name__, e)

运行输出：

正常输入：
   节点收到: text='abc' n=0 PS
结果: {'text': 'abc', 'n': 3}

类型错误的输入（n 传字符串）：
报错: ValidationError 1 validation error for PS
n
  Input should be a valid integer, unable to parse string as an integer
  [type=int_parsing, input_value='不是数字', input_type=str]
    For further information visit https://errors.pydantic.dev/2.13/v/int_parsing

缺必填字段：
报错: ValidationError 1 validation error for PS
text
  Field required [type=missing, input_value={}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing

传了模型里没声明的键：
   节点收到: text='abc' n=0 PS
结果: {'text': 'abc', 'n': 3}

差别很明显：

默认值会自动填充（n=0，不传也有）
类型不对直接拦住，而且错误信息会点名是哪个字段出了问题
报错发生在入口，不用等到某个节点内部炸掉。注意错误信息里的 input_value={}——它把你实际传进去的东西也打印出来了，排查线上问题时特别有用

但最后一组输出提醒了一件事：多传的键照样被静默丢掉，Pydantic 也不管。 因为 LangGraph 会先按状态定义把输入过滤一遍，再交给 Pydantic 校验，多余的键在校验之前就已经没了。

更要紧的是：Pydantic 只守得住入口，守不住节点的返回值。 实测一下——状态用 Pydantic，让节点返回一个不存在的字段：

from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel

class PS(BaseModel):
    text: str
    n: int = 0

# 状态用 Pydantic，看它能不能拦住节点返回值里的错别字
def pbad(state: PS) -> dict:
    # "不存在的字段" 不在 PS 的定义里
    return {"n": 1, "不存在的字段": 123}

builder = StateGraph(PS)
builder.add_node("pbad", pbad)
builder.add_edge(START, "pbad")
builder.add_edge("pbad", END)
graph = builder.compile()

# 期待它报错拦下来，实际会怎样？
print("结果:", graph.invoke({"text": "abc"}))

运行输出：

结果: {'text': 'abc', 'n': 1}

一声不响，照样丢弃。 所以别指望换成 Pydantic 就能躲开 §5.3 那个「返回值字段名拼错」的坑——它只在入口有用。这个边界要划清楚，不然容易生出一种虚假的安全感。

4.3.3 dataclass：介于两者之间 #

# dataclass 来自标准库，用装饰器自动生成 __init__ 等方法
from dataclasses import dataclass

from langgraph.graph import END, START, StateGraph

# 加上 @dataclass 装饰器，它会自动生成 __init__；类本身不继承任何基类
@dataclass
class DS:
    # 没有默认值就是必填
    text: str
    # 有默认值，不传也会自动填 0
    n: int = 0

# 和 Pydantic 一样用点号访问
def dnode(state: DS) -> dict:
    # 打印出来是 DS(text='abcd', n=0) 这种 dataclass 的标准形式
    print("   节点收到:", state, type(state).__name__)
    # 返回值依然是普通字典，不是 DS 实例
    return {"n": len(state.text)}

builder = StateGraph(DS)
builder.add_node("dnode", dnode)
builder.add_edge(START, "dnode")
builder.add_edge("dnode", END)
graph = builder.compile()

# 三种输入，重点看类型错误那一组
CASES = [
    ("正常输入", {"text": "abcd"}),
    ("类型错误（n 传字符串）", {"text": "abcd", "n": "不是数字"}),
    ("缺必填字段", {}),
]
for title, payload in CASES:
    print(f"\n{title}：")
    try:
        print("结果:", graph.invoke(payload))
    except Exception as e:
        print("报错:", type(e).__name__, e)

运行输出：

正常输入：
   节点收到: DS(text='abcd', n=0) DS
结果: {'text': 'abcd', 'n': 4}

类型错误（n 传字符串）：
   节点收到: DS(text='abcd', n='不是数字') DS
结果: {'text': 'abcd', 'n': 4}

缺必填字段：
报错: TypeError DS.__init__() missing 1 required positional argument: 'text'

默认值会填，缺必填字段会在入口报错，但类型完全不校验——注意第二组，n='不是数字' 就这么大摇大摆地进去了，dataclass 一个字都没说。

所谓「介于两者之间」，到这里就有了准确的含义：

	缺必填字段	类型不对
`TypedDict`	节点内 `KeyError`（最晚发现）	不管
`dataclass`	入口 `TypeError`（位置对了，信息一般）	不管
`Pydantic`	入口 `ValidationError`（位置对，信息最好）	拦住

4.3.4 怎么选 #

	`TypedDict`	Pydantic	`dataclass`
节点里访问	`state["x"]`	`state.x`	`state.x`
默认值	× 不填充	√	√
缺必填字段	× 节点内才炸	√ 入口拦截	√ 入口拦截
类型校验	×	√	×
多传的键	静默丢弃	静默丢弃	静默丢弃
节点返回值拼错	静默丢弃	静默丢弃（管不了）	静默丢弃
错误信息质量	差（节点内 `KeyError`）	好（指明字段 + 打印实际输入）	一般
额外依赖	无	pydantic（LangChain 已依赖）	无

注意中间那两行三列全是「静默丢弃」——换状态写法解决不了键名拼错，那得靠静态类型检查（§5.3）。

场景	建议
学习、内部脚本、字段少	`TypedDict`（本文默认，官方文档也用它）
图的入口对外暴露（Web API、别的服务调用）	Pydantic，把非法输入挡在门外
已有 dataclass 想直接复用	`dataclass`

注意一个不一致：无论用哪种写法，节点的返回值都是普通字典。 就算状态是 Pydantic 模型，节点也是 return {"n": 3}，而不是 return PS(...)。刚开始可能觉得别扭，习惯就好。

4.4. 同一个字段被写两次会怎样 #

默认行为是后写覆盖先写：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

# 第一个节点写 text
def w1(state: S) -> dict:
    # 先写入一个值
    return {"text": "第一个节点写的"}

# 第二个节点写同一个 text
def w2(state: S) -> dict:
    # 再写入另一个值，没有任何「已经有值了」的提示
    return {"text": "第二个节点写的"}

# 按 w1 → w2 的顺序串起来
builder = StateGraph(S)
builder.add_node("w1", w1)
builder.add_node("w2", w2)
builder.add_edge(START, "w1")
builder.add_edge("w1", "w2")
builder.add_edge("w2", END)
graph = builder.compile()

# 看最后 text 里剩下的是谁写的
print("串行覆盖:", graph.invoke({"text": "", "n": 0}))

运行输出：

串行覆盖: {'text': '第二个节点写的', 'n': 0}

第一个节点写的内容没留下任何痕迹：不是拼接，也不是报错，就是干干净净地被替换掉了。

对「文本清洗」这类场景，这个行为是对的——后一步的结果本来就该覆盖前一步。但换成下面这几种场景就不对了：

场景	你想要的	默认行为给你的
对话消息列表	追加新消息	只剩最后一条
操作日志	每个节点记一条	只剩最后一条
累计计数 / 累计金额	累加	只剩最后一个值
多路检索的结果汇总	合并成一个大列表	只剩一路的结果

这几种场景在真实项目里非常常见，尤其是第一个——Agent 的对话历史，本质上就是一个需要不断追加的列表。

那怎么改成追加？这正是下一章 Reducer 要解决的问题。这里先记住两点：默认行为是覆盖，而且覆盖是悄悄发生的，不会有任何警告。

5. 节点（Node） #

5.1. 节点就是普通函数 #

节点没有任何特殊要求：不用继承基类，不用加装饰器，也不用单独注册。它只需要满足一个签名约定：

def 节点名(state) -> dict | None:
        ↑            ↑
   一个参数接收状态   返回字典或 None

一个参数（状态），返回字典或 None。 就这么简单。由此带来三个好处：

节点可以单独拿出来测试，直接 clean({"text": " a "}) 调用就行
已有的业务函数，稍微改一下签名就能当节点用
节点内部想干什么都行：查数据库、调 API、调模型、发消息

第一条值得强调，它是图相比 create_agent 的一个实际优势：节点就是普通函数，可以脱离图直接做单元测试。

from typing import TypedDict

class S(TypedDict):
    text: str
    n: int

# 一个普通的节点函数，没有任何框架痕迹
def clean(state: S) -> dict:
    return {"text": state["text"].strip()}

# 不需要建图、不需要 compile，直接把状态当普通字典传进去
assert clean({"text": "  abc  "}) == {"text": "abc"}
# 边界情况也能单测：全是空格
assert clean({"text": "   "}) == {"text": ""}
print("两个断言都通过了")

运行输出：

两个断言都通过了

对比第 19 章那个 create_agent 版本：想测「风控有没有被调用」，得真的发一次模型请求，还要在工具里埋点。改成图之后，每个业务节点都能用普通的 assert 测掉，不花钱，也不用看模型心情。

5.2. 三种注册方式 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    return {"text": state["text"].strip()}

# 换个函数名，用来演示方式二
def my_clean(state: S) -> dict:
    return {"text": state["text"].strip()}

# 方式一：显式指定节点名（推荐，最清晰）
b1 = StateGraph(S)
b1.add_node("clean", clean)
b1.add_edge(START, "clean")
graph = b1.compile()
print("方式一 -> 节点列表:", list(graph.get_graph().nodes))
print("  结果:", graph.invoke({"text": "x", "n": 0}))

# 方式二：只传函数，自动用函数名当节点名
b2 = StateGraph(S)
b2.add_node(my_clean)          # 节点名就是 "my_clean"
b2.add_edge(START, "my_clean")
graph = b2.compile()
print("方式二 -> 节点列表:", list(graph.get_graph().nodes))
print("  结果:", graph.invoke({"text": "x", "n": 0}))

# 方式三：lambda（简单逻辑可用，但不便于调试）；顺便验证节点名能用中文
b3 = StateGraph(S)
b3.add_node("清洗", lambda s: {"text": s["text"].strip()})
b3.add_edge(START, "清洗")
b3.add_edge("清洗", END)
graph = b3.compile()
print("方式三 -> 节点列表:", list(graph.get_graph().nodes))
print("  结果:", graph.invoke({"text": "x", "n": 0}))

运行输出：

方式一 -> 节点列表: ['__start__', 'clean', '__end__']
  结果: {'text': 'x', 'n': 0}
方式二 -> 节点列表: ['__start__', 'my_clean', '__end__']
  结果: {'text': 'x', 'n': 0}
方式三 -> 节点列表: ['__start__', '清洗', '__end__']
  结果: {'text': 'x', 'n': 0}

顺便确认了两件事：只传函数时节点名就是函数名，以及节点名可以用中文。

三种方式各自的问题：

方式	隐患
方式二（只传函数）	改函数名等于改节点名，而 `add_edge` 里的字符串不会跟着改，一重构就断链
方式三（lambda）	图上和 `stream` 里显示的名字得靠你自己起；逻辑一长就没法单测
中文节点名	能用，但 Mermaid 渲染和某些日志系统对非 ASCII 的支持不一致，生产环境建议用英文

推荐用方式一，并给一个业务化的名字。 节点名会出现在图上、stream 的输出里、LangSmith 的 Trace 里——起个好名字，排障时能省很多事。node1 / step2 这种名字，图一大就毫无信息量。

节点名重复会在注册时立刻报错，不会等到编译：

from typing import TypedDict

from langgraph.graph import StateGraph

class S(TypedDict):
    text: str

builder = StateGraph(S)
# 第一次注册没问题
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
# 用同一个名字再注册一次，报错就发生在这一行
try:
    builder.add_node("clean", lambda s: {"text": "别的逻辑"})
except Exception as e:
    print("报错:", type(e).__name__, e)

运行输出：

报错: ValueError Node `clean` already present.

「在注册时就报」这一点值得注意，因为本章大部分问题都是等到 invoke 才暴露，甚至永远不暴露。报错越早越好——add_node 这个立刻报错的行为，是 LangGraph 少有的严格之处。

5.3. 返回值里的多余字段会被静默丢弃 #

这是本章第一个静默失败，也是最容易踩的一个：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

# 返回值里混进一个状态定义里没有的键
def bad(state: S) -> dict:
    # "不存在的字段" 不在 S 的定义里
    return {"text": "x", "不存在的字段": 123}

builder = StateGraph(S)
builder.add_node("bad", bad)
builder.add_edge(START, "bad")
builder.add_edge("bad", END)
graph = builder.compile()

# 期待报错或者看到那个字段，结果两个都没有
print("结果:", graph.invoke({"text": "", "n": 0}))

运行输出：

结果: {'text': 'x', 'n': 0}

没有报错，那个字段直接消失了。

这意味着字段名拼错不会有任何提示。你写成 return {"txet": "..."}，图照跑不误，只是那次更新彻底丢失了。等后面某个节点读到旧值，你会一路排查到怀疑人生。

这个坑之所以特别危险，是因为它的症状看起来像别的问题：

你观察到的现象                     你会怀疑的方向          真正的原因
「这个节点好像没执行」        →    边没连对 / 条件分支     →   返回值键名拼错
「值一直是旧的」              →    状态被覆盖了            →   返回值键名拼错
「下游节点读到 KeyError」     →    入口少传了字段          →   上游节点的键名拼错

而且它不分状态写法：§4.3 已经验证过，换成 Pydantic 也照样静默丢弃。

防范办法有三个，按性价比排序：

让编辑器的类型检查开着（Pylance / mypy）。TypedDict 的标注运行时不生效，但静态检查能直接标红拼错的键名——这是唯一能在写代码时就发现问题的办法

状态字段有变动时，先改状态定义再改节点，别反过来

改完用 stream 跑一遍（§7.3），逐节点核对返回的键名，拼错的那个一眼就能看见

6. 边（Edge） #

6.1. add_edge：固定的下一步 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

builder = StateGraph(S)
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
builder.add_node("count", lambda s: {"n": len(s["text"])})
builder.add_edge(START, "clean")
# 两个参数都是节点名（字符串），方向是从第一个到第二个
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# 无论 text 是什么，clean 跑完一定会跑 count
print("结果:", graph.invoke({"text": "  abc  ", "n": 0}))

运行输出：

结果: {'text': 'abc', 'n': 3}

add_edge("clean", "count") 的含义是「clean 跑完，无条件跑 count」。无条件是关键词——这条边不看状态、不做判断，永远走这条路。

第 19 章 §4.1 那个「唯一入边就是结构性保证」的说法，依据就在这里：无条件边在图上是一条实线，意思是「不存在不走这条路的可能」。 这是提示词永远给不了的保证。

需要判断的分支得用 add_conditional_edges，那是第 23 章的内容。本章只用固定边。

6.2. START 和 END #

这两个是 LangGraph 预置的特殊节点，不需要你再 add_node：

	是什么	规则
`START`	图的入口	必须至少有一条 `add_edge(START, ...)`，否则编译报错
`END`	图的出口	走到这里图就结束

它们在 get_graph() 里显示为 __start__ 和 __end__（上一章见过）。注意：START / END 是代码里用的常量名，__start__ / __end__ 是它们在图里的节点名，说的是同一个东西。

漏了 START 边的报错很直白：

报错: ValueError Graph must have an entrypoint: add at least one edge from START to another node

空图编译也是同一个错——因为空图当然没有入口。

两者为什么不对称？可以这么理解：图必须知道「从哪开始」才能跑起来，但「在哪结束」它可以自己推断——跑到一个没有出边的节点，自然就结束了。所以下面两种写法完全等价：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

# 写法一：显式声明结束
b1 = StateGraph(S)
b1.add_node("clean", lambda s: {"text": s["text"].strip()})
b1.add_node("count", lambda s: {"n": len(s["text"])})
b1.add_edge(START, "clean")
b1.add_edge("clean", "count")
b1.add_edge("count", END)          # 显式连到 END
graph = b1.compile()
print("显式写 END:", graph.invoke({"text": "  abc  ", "n": 0}))

# 写法二：干脆不写，count 没有出边，跑完自然结束
b2 = StateGraph(S)
b2.add_node("clean", lambda s: {"text": s["text"].strip()})
b2.add_node("count", lambda s: {"n": len(s["text"])})
b2.add_edge(START, "clean")
b2.add_edge("clean", "count")
graph = b2.compile()               # 少一条边，照样编译通过
print("不写 END:  ", graph.invoke({"text": "  abc  ", "n": 0}))

运行输出：

显式写 END: {'text': 'abc', 'n': 3}
不写 END:   {'text': 'abc', 'n': 3}

代价是：漏写通向 END 的边不会报错。 这是本章第二个静默失败，§10.2 会细讲——它的排查方式和你想的不太一样。

建议始终显式写 add_edge(..., END)。 不是因为不写会出错，而是因为「有意结束」和「忘记连边」这两件事，在代码里长得一模一样。显式写出来，下一个读代码的人（包括三个月后的你）才能确定这是故意的。

6.3. add_sequence：串行的简写 #

一串节点顺序执行时，可以少写几行：

from typing import TypedDict

from langgraph.graph import START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    print("   [clean] 收到 state:", state)
    return {"text": state["text"].strip()}

def count(state: S) -> dict:
    print("   [count] 收到 state:", state)
    return {"n": len(state["text"])}

# 照常创建构建器
builder = StateGraph(S)
# 等价于 add_node(clean) + add_node(count) + add_edge("clean", "count")
builder.add_sequence([clean, count])
# START 那条边不在 add_sequence 的职责范围内，仍然要自己连
builder.add_edge(START, "clean")
# 编译，得到可运行的图
graph = builder.compile()

print("结果:", graph.invoke({"text": "  abc  "}))

# 把节点和边都打印出来核对
print("节点列表:", list(graph.get_graph().nodes))
for e in graph.get_graph().edges:
    print(f"   {e.source} --> {e.target}")

运行输出：

   [clean] 收到 state: {'text': '  abc  '}
   [count] 收到 state: {'text': 'abc'}
结果: {'text': 'abc', 'n': 3}
节点列表: ['__start__', 'clean', 'count', '__end__']
   __start__ --> clean
   clean --> count
   count --> __end__

看到 count --> __end__，别急着以为是 add_sequence 帮你连了 END。 它没连——这条边是 get_graph() 渲染出来的：任何没有出边的节点，画图时都会被补一条指向 __end__ 的线。 这个渲染规则很重要，§10.2 会看到它怎样把一张断链的图「画成」正常图。

add_sequence 的取舍很清楚：

省下的是：N 个节点能省 N-1 行 add_edge，加上 N 行 add_node
付出的是：节点名自动取函数名（改函数名就断链），而且一旦要加分支就得拆回去显式写

所以本文后面仍然全部显式写。 真实项目里，纯顺序结构很少能一直保持到最后，早晚要加分支；到时候拆回去的成本，往往比省下的那几行更贵。

6.4. 一个节点连出多条边 = 并行 #

从同一个节点连出两条固定边，两个下游节点会并行执行：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

# 两个并行节点，各写一个不同的字段
def wa(state: S) -> dict:
    print("   [wa] 执行")
    return {"text": "A写的"}

def wb(state: S) -> dict:
    print("   [wb] 执行")
    return {"n": 99}

builder = StateGraph(S)
builder.add_node("wa", wa)
builder.add_node("wb", wb)
# 从 START 连出第一条边
builder.add_edge(START, "wa")
# 再从同一个 START 连出第二条边，wa 和 wb 就会在同一步里并行执行
builder.add_edge(START, "wb")
builder.add_edge("wa", END)
builder.add_edge("wb", END)
graph = builder.compile()

print("结果:", graph.invoke({"text": "", "n": 0}))

# 换成 stream，看两个并行节点是各出一个 chunk 还是合成一个
print("--- stream ---")
for chunk in graph.stream({"text": "", "n": 0}):
    print("   chunk:", chunk)

运行输出：

   [wa] 执行
   [wb] 执行
结果: {'text': 'A写的', 'n': 99}
--- stream ---
   [wa] 执行
   chunk: {'wa': {'text': 'A写的'}}
   [wb] 执行
   chunk: {'wb': {'n': 99}}

这里有个反直觉的点，值得细想：你没有写任何「并行」关键字，并行是「同一个节点连出多条边」这个结构自动带来的。 换句话说，在 LangGraph 里串行和并行的区别只是边怎么连，没有别的开关。

两个节点写的是不同字段，所以结果自动合并到了一起。stream 的输出还说明：并行的两个节点各出一个 chunk，不会合成一个。

顺便留意 stream 那段输出的顺序——[wa] 的 chunk 出现在 [wb] 执行之前。所谓「并行」指的是「同一步内、彼此看不到对方的写入」，并不保证真的同时在跑。 默认的同步执行下，LangGraph 是逐个跑完再合并的；能不能真正并发，取决于节点是同步函数还是异步函数。这个区别第 24 章还会再碰到。

但如果它们同时写同一个字段，就会直接报错：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str

builder = StateGraph(S)
# 这次两个节点抢着写同一个 text
builder.add_node("wa", lambda s: {"text": "A写的"})
builder.add_node("wb", lambda s: {"text": "B写的"})
builder.add_edge(START, "wa")
builder.add_edge(START, "wb")
builder.add_edge("wa", END)
builder.add_edge("wb", END)
graph = builder.compile()

try:
    graph.invoke({"text": ""})
except Exception as e:
    print("报错:", type(e).__name__, e)

运行输出：

报错: InvalidUpdateError At key 'text': Can receive only one value per step.
Use an Annotated key to handle multiple values.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/INVALID_CONCURRENT_GRAPH_UPDATE

这个报错说得很清楚——并行分支写同一个键时，图不知道该听谁的，需要你用 Annotated 指定一个合并规则。这又指向下一章的 Reducer。

对比 §4.4，两种「写同一个字段」的处理方式完全不同：

	串行写同一字段	并行写同一字段
行为	静默覆盖，后写赢	直接报错
为什么	有明确先后，「后来的更新」是合理默认	同一步内没有先后，无法选择
你该做什么	确认覆盖是你想要的	必须加 reducer（第 21 章）

报错信息里的 per step 是个关键词。LangGraph 的执行是按「步」（superstep）推进的：同一步里能并行跑的节点一起跑完，再统一合并结果，然后进入下一步。串行覆盖之所以不报错，是因为两次写入落在不同的步里。这个「步」的概念，第 24 章讲循环上限时还会用到。

本章只用串行结构。 知道「并行是把两条边从同一个节点连出去」，以及「并行写同一个键会报错」，就够了。

7. 编译与运行 #

7.1. compile() 检查什么 #

编译时 LangGraph 会做一遍结构检查。实测下来，它管的事情比大多数人以为的少：

检查项	会不会报错	报错信息
有没有从 `START` 出发的边	√ 报错	`Graph must have an entrypoint: add at least one edge from START to another node`
边指向了不存在的节点	√ 报错	`Found edge ending at unknown node ...`
节点名有没有重复	√ 报错（在 `add_node` 时就报了）	`Node 'x' already present.`
有没有节点通向 `END`	× 不检查	—
有没有孤立节点	× 不检查	—
中间的边有没有断	× 不检查	—
节点返回的字段在不在状态里	× 不检查	—
输入状态完不完整	× 不检查（Pydantic / dataclass 状态除外）	—

前三项和后五项的分界很清楚：会报错的，都是「图根本跑不起来」的问题；不报错的，都是「图能跑，但跑得不是你想的那样」。

第二项值得单独说，因为它很容易和第六项混淆：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def build(wire: str) -> StateGraph:
    """按不同的接线方式建一张图，用来对比编译时的检查。"""
    b = StateGraph(S)
    b.add_node("clean", lambda s: {"text": s["text"].strip()})
    b.add_node("count", lambda s: {"n": len(s["text"])})
    b.add_edge(START, "clean")
    if wire == "拼错节点名":
        # 节点名写成了 coutn
        b.add_edge("clean", "coutn")
    elif wire == "忘记连边":
        # 这里本该写 b.add_edge("clean", "count")，整条漏掉了
        pass
    b.add_edge("count", END)
    return b

# 一：拼错节点名 → 编译时就被抓住
try:
    build("拼错节点名").compile()
except Exception as e:
    print("拼错节点名 -> 报错:", type(e).__name__, e)

# 二：彻底忘记连边 → 编译一路绿灯
graph = build("忘记连边").compile()
print("忘记连边 -> 编译通过，结果:", graph.invoke({"text": "  abc  "}))

运行输出：

拼错节点名 -> 报错: ValueError Found edge ending at unknown node `coutn`
忘记连边 -> 编译通过，结果: {'text': 'abc'}

同样是「边没连对」，拼错节点名会被抓住，彻底忘记连边则不会。 第二组的结果里连 n 都没有，因为 count 根本没执行。这个不对称是很多人踩坑的地方——它意味着「编译通过了」完全不能说明边连对了。

一句话总结：

compile() 只保证「图能启动」，不保证「图符合你的意图」。

7.2. invoke：跑完拿最终状态 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int
    owner: str

builder = StateGraph(S)
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
builder.add_node("count", lambda s: {"n": len(s["text"])})
# 第三个节点写 owner，下面用它说明「调用方不必关心是谁写的」
builder.add_node("assign", lambda s: {"owner": "客服一线"})
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", "assign")
builder.add_edge("assign", END)
graph = builder.compile()

# 参数是初始状态（一个字典），返回值是跑完之后的完整状态
result = graph.invoke({"text": "  你好世界  "})
# 拿到的是完整状态，包含所有节点写过的字段
print("完整状态:", result)
# 想要哪个字段就直接取，不用管它是哪个节点写进去的
print("owner:", result["owner"])

运行输出：

完整状态: {'text': '你好世界', 'n': 4, 'owner': '客服一线'}
owner: 客服一线

传入初始状态，返回最终状态。返回值是完整状态，不是最后一个节点的返回值——这一点和节点的返回值规则正好相反，别搞混（§2.3 那张表就是为这个准备的）。

最后那行 result["owner"] 体现了状态解耦的好处：调用方只依赖状态的字段名，不依赖图的内部结构。 你把图从三个节点重构成八个节点，只要 owner 还在，调用方一行都不用改。

7.3. stream：看每一步 #

invoke 只能看到最终结果，中间发生了什么全是黑箱。换成 stream，就能一步一步看：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    print("   [clean] 收到 state:", state)
    return {"text": state["text"].strip()}

def count(state: S) -> dict:
    print("   [count] 收到 state:", state)
    return {"n": len(state["text"])}

builder = StateGraph(S)
builder.add_node("clean", clean)
builder.add_node("count", count)
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# stream 返回一个生成器，每执行完一个节点就吐一个 chunk
for chunk in graph.stream({"text": "  hello  ", "n": 0}):
    # 默认模式下 chunk 的形状是 {节点名: 该节点返回的更新}
    print("  chunk:", chunk)

运行输出（[clean] / [count] 是节点内部自己的 print，会和 chunk 交错出现）：

   [clean] 收到 state: {'text': '  hello  ', 'n': 0}
  chunk: {'clean': {'text': 'hello'}}
   [count] 收到 state: {'text': 'hello', 'n': 0}
  chunk: {'count': {'n': 5}}

默认模式是 updates：每个 chunk 都是一个 {节点名: 这个节点返回的更新}。这正好是排障时最想看的信息——哪个节点改了什么。§5.3 那个键名拼错的坑，症状往往就是「某个节点的 chunk 里，冒出了一个你没见过的键」。

换成 values 模式，看到的是每一步之后的完整状态：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    print("   [clean] 收到 state:", state)
    return {"text": state["text"].strip()}

def count(state: S) -> dict:
    print("   [count] 收到 state:", state)
    return {"n": len(state["text"])}

builder = StateGraph(S)
builder.add_node("clean", clean)
builder.add_node("count", count)
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# stream_mode="values" 让每个 chunk 变成「这一步之后的完整状态」
for chunk in graph.stream({"text": "  hello  ", "n": 0}, stream_mode="values"):
    # 这里打印的是全量状态，不再是单个节点的更新
    print("  chunk:", chunk)

运行输出：

  chunk: {'text': '  hello  ', 'n': 0}
   [clean] 收到 state: {'text': '  hello  ', 'n': 0}
  chunk: {'text': 'hello', 'n': 0}
   [count] 收到 state: {'text': 'hello', 'n': 0}
  chunk: {'text': 'hello', 'n': 5}

注意 values 模式会先吐一次初始状态（在任何节点执行之前），所以两个节点的图会输出三条。别把第一条当成某个节点的执行结果——想用 chunk 个数判断跑了几个节点，values 模式要记得减一。

模式	每个 chunk 是什么	chunk 个数	什么时候用
`updates`（默认）	`{节点名: 更新内容}`	节点数	排障：想知道谁改了什么
`values`	完整状态	节点数 + 1	想看状态的演化过程

两者怎么选，可以这样记：想知道「谁干的」用 updates，想知道「现在长什么样」用 values。 排障绝大多数时候用前者，因为它自带节点名这个定位信息。

stream_mode 还有别的取值（messages、debug、custom），留到第 27 章讲。

调试习惯：图跑出来不对时，把 invoke 换成 stream 打一遍。 一眼就能看出是哪个节点没执行，或者改错了字段。

8. 把图画出来 #

mermaid.live

draw_mermaid() 导出标准的 Mermaid 语法，粘进 Markdown 预览、飞书文档、GitHub README 都能直接渲染：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

builder = StateGraph(S)
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
builder.add_node("count", lambda s: {"n": len(s["text"])})
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# get_graph() 拿到结构描述，draw_mermaid() 把它转成 Mermaid 文本
print(graph.get_graph().draw_mermaid())

运行输出：

---
config:
  flowchart:
    curve: linear
---
graph TD;
    __start__([<p>__start__</p>]):::first
    clean(clean)
    count(count)
    __end__([<p>__end__</p>]):::last
    __start__ --> clean;
    clean --> count;
    count --> __end__;
    classDef default fill:#f2f0ff,line-height:1.2
    classDef first fill-opacity:0
    classDef last fill:#bfb6fc

输出里首尾那几行（开头的 ---config--- 块和末尾的 classDef）都要一起粘贴：前者控制连线样式，后者控制配色。留意 :::first 和 :::last 这两个标记，§10.2 会拿它们当诊断信号用。

另一个方法是 draw_ascii()，能在终端里直接画出来，不用复制到别处渲染：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

builder = StateGraph(S)
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
builder.add_node("count", lambda s: {"n": len(s["text"])})
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# draw_ascii() 直接输出可在终端查看的 ASCII 图
print(graph.get_graph().draw_ascii())

运行输出：

+-----------+
| __start__ |
+-----------+
      *
      *
      *
  +-------+
  | clean |
  +-------+
      *
      *
      *
  +-------+
  | count |
  +-------+
      *
      *
      *
 +---------+
 | __end__ |
 +---------+

它需要一个额外依赖，没装的话会报：

报错: ImportError Install grandalf to draw graphs: `pip install grandalf`.

装上就能用：

pip install grandalf

两个画图方法怎么分工：

	`draw_mermaid()`	`draw_ascii()`
额外依赖	无	需要 grandalf
看的地方	要粘到能渲染 Mermaid 的地方	终端里直接看
循环边（回边）	画得出来	画不出来
条件边（虚线）	区分得出来	区分不出来
适合	分支、循环、复杂图	快速确认节点有哪些

本章的图是纯串行的，两个都够用。但到了第 23、24 章的分支和循环，就只能用 draw_mermaid() 了——回边和虚线是那两章的主角，而 ASCII 图恰好这两样都表达不了。

画图不只是为了好看。 节点一多，「我以为连上了，其实没连」是很常见的事；看一眼图，往往比读代码快得多。上一章 §2.4 那个八节点的 middleware 图，就是最好的例子。

9. 实战：工单分派最小图 #

把本章的内容合成一个有实际意义的两节点图：读进一条工单文本，解析出类别和金额，再分派给对应的小组。

挑这个例子有个用意：它是一段纯确定性逻辑，完全不需要模型。 第 19 章 §3.5 算过账，这类 if-else 交给模型要三次往返、两千个 token，而写成图是零成本。「图套 Agent」这个架构里，最外层那一圈就该长成这个样子。

9.1. ticket_graph.py #

"""最小两节点图：工单解析 → 分派"""

# 让类型标注可以延迟求值，写 dict 而不用 typing.Dict
from __future__ import annotations

# 标准库正则，用来从工单文本里抠金额
import re
# 状态用 TypedDict，本章默认写法
from typing import TypedDict

# 图的三件套
from langgraph.graph import END, START, StateGraph

# ① 状态：这张图从头到尾要传的四个字段
class TicketState(TypedDict):
    raw: str        # 工单原文（输入）
    category: str   # 解析出的类别
    amount: float   # 解析出的金额
    owner: str      # 分派到的小组（输出）

# 分派规则：类别 -> 负责小组
RULES = {
    # 退款类走财务
    "退款": "财务组",
    # 故障类走技术支持
    "故障": "技术支持组",
    # 兜底类别，同时也是「都没命中」时的默认值
    "咨询": "客服一线",
}

def parse(state: TicketState) -> dict:
    """第一个节点：从工单原文里抽出类别和金额。"""
    # 从状态里取输入，这个字段由 invoke 传进来
    text = state["raw"]
    # 关键词匹配，命中哪个算哪个，都没命中算「咨询」
    category = "咨询"
    # 遍历 RULES 的键，也就是三个类别关键词
    for keyword in RULES:
        # 只要关键词出现在工单原文里就算命中
        if keyword in text:
            # 把命中的关键词直接当类别用
            category = keyword
            # 命中第一个就停，所以 RULES 的顺序会影响结果（§9.3 会讲这个坑）
            break
    # 用正则抓「数字 + 元」，抓不到算 0
    m = re.search(r"(\d+(?:\.\d+)?)\s*元", text)
    # 匹配到就转成 float，没匹配到用 0.0 兜底，避免下游 KeyError
    amount = float(m.group(1)) if m else 0.0
    # 只返回这个节点负责的两个字段
    return {"category": category, "amount": amount}

def assign(state: TicketState) -> dict:
    """第二个节点：按类别和金额决定负责人。"""
    # 直接读 parse 写进状态的 category，这就是节点间的数据传递
    owner = RULES[state["category"]]
    # 大额工单加一道复核标记
    if state["amount"] >= 1000:
        # 拼字符串只是演示，第 26 章会把这里换成真正的人工审批
        owner = f"{owner}（需主管复核）"
    # 同样只返回自己负责的字段
    return {"owner": owner}

# ③ 组装
builder = StateGraph(TicketState)
# 注册解析节点
builder.add_node("parse", parse)
# 注册分派节点
builder.add_node("assign", assign)
# 入口边：所有工单都必须先解析
builder.add_edge(START, "parse")
# 解析完接着分派
builder.add_edge("parse", "assign")
# 分派完结束（显式写出来，表明这是有意的终点）
builder.add_edge("assign", END)

# ④ 编译
graph = builder.compile()

# 只有直接运行本文件时才执行下面的演示代码，被 import 时不执行
if __name__ == "__main__":
    # 四条测试工单，覆盖三种类别和一个大额场景
    TICKETS = [
        # 有关键词「退款」，有金额
        "订单 A1002 申请退款，金额 68 元",
        # 明显是故障，但文本里没有「故障」二字（故意留的缺陷，见 §9.3）
        "设备开机报 E2，无法启动，请尽快处理",
        # 有关键词「咨询」，没有金额
        "想咨询一下年假怎么算",
        # 大额退款，会触发复核标记
        "批量退款申请，总计 2500 元，请走流程",
    ]
    # 逐条跑一遍
    for raw in TICKETS:
        # 只传 raw，其余三个字段由节点写入
        result = graph.invoke({"raw": raw})
        # 先打印原文，方便对照
        print(f"\n工单：{raw}")
        # 这两个字段是 parse 写的
        print(f"  类别：{result['category']}  金额：{result['amount']}")
        # 这个字段是 assign 写的
        print(f"  分派：{result['owner']}")

    # 分隔标题
    print("\n=== 图结构 ===")
    # 把图导出成 Mermaid，验收清单第 5 条要用
    print(graph.get_graph().draw_mermaid())

    # 分隔标题
    print("=== 逐节点观察 ===")
    # 用 stream 看每个节点各自改了什么
    for chunk in graph.stream({"raw": "订单 A1002 申请退款，金额 68 元"}):
        # 每个 chunk 形如 {'parse': {...}}
        print("  ", chunk)

9.2. 运行结果 #

先看四条工单的分派结果（中间 === 图结构 === 那段 Mermaid 输出和 §8 演示的一样，这里略去）：

工单：订单 A1002 申请退款，金额 68 元
  类别：退款  金额：68.0
  分派：财务组

工单：设备开机报 E2，无法启动，请尽快处理
  类别：咨询  金额：0.0
  分派：客服一线

工单：想咨询一下年假怎么算
  类别：咨询  金额：0.0
  分派：客服一线

工单：批量退款申请，总计 2500 元，请走流程
  类别：退款  金额：2500.0
  分派：财务组（需主管复核）

stream 的输出清楚地显示了两个节点各自改了什么：

   {'parse': {'category': '退款', 'amount': 68.0}}
   {'assign': {'owner': '财务组'}}

9.3. 第二条工单分错了——这不是 bug，是伏笔 #

「设备开机报 E2，无法启动」明明是设备故障，却被分到了「咨询 / 客服一线」。

原因很简单：parse 用的是关键词匹配，而这句话里压根没出现「故障」两个字。这和第 15 章 §5.2 里 BM25 的短板是同一个病根——字面匹配理解不了同义表达。

顺着这个思路往下试，会发现第二个更隐蔽的缺陷。给它几条同时含多个关键词的工单：

# 把 ticket_graph.py 的核心逻辑精简复制过来，方便单独跑这个实验
import re
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class TicketState(TypedDict):
    raw: str
    category: str
    amount: float
    owner: str

# 注意「退款」写在「故障」前面，这是下面那个坑的根源
RULES = {"退款": "财务组", "故障": "技术支持组", "咨询": "客服一线"}

def parse(state: TicketState) -> dict:
    text = state["raw"]
    category = "咨询"
    # 按 RULES 的定义顺序遍历，命中第一个就 break
    for keyword in RULES:
        if keyword in text:
            category = keyword
            break
    m = re.search(r"(\d+(?:\.\d+)?)\s*元", text)
    return {"category": category, "amount": float(m.group(1)) if m else 0.0}

def assign(state: TicketState) -> dict:
    owner = RULES[state["category"]]
    if state["amount"] >= 1000:
        owner = f"{owner}（需主管复核）"
    return {"owner": owner}

builder = StateGraph(TicketState)
builder.add_node("parse", parse)
builder.add_node("assign", assign)
builder.add_edge(START, "parse")
builder.add_edge("parse", "assign")
builder.add_edge("assign", END)
graph = builder.compile()

# 三条含多个关键词的工单，看关键词匹配会选哪个
for raw in ["退款还是故障？我要退款", "故障导致要退款 500 元", "设备故障，无法启动"]:
    # 每条都单独跑一遍图
    r = graph.invoke({"raw": raw})
    # 打印类别和分派结果
    print(f"  {raw!r} -> {r['category']} / {r['owner']}")

运行输出：

  '退款还是故障？我要退款' -> 退款 / 财务组
  '故障导致要退款 500 元' -> 退款 / 财务组
  '设备故障，无法启动' -> 故障 / 技术支持组

第二条错了。 「故障导致要退款」这条工单里，「故障」出现在「退款」之前，业务上显然该先按故障处理，却被分给了财务组。

原因藏在 parse 的循环里：for keyword in RULES 是按 RULES 字典的定义顺序遍历的，而「退款」写在「故障」前面，命中就 break 了。所以：

这个分类器的优先级不由业务规则决定，而由 RULES 字典里键的书写顺序决定。 有人为了「让代码整齐」把字典按拼音排个序，分派逻辑就悄悄变了——而且不会有任何测试失败，除非你恰好有一条同时含两个关键词的用例。

这类「行为依赖于一个看起来无关的书写顺序」的隐患，和 §5.3 那个键名拼错属于同一类问题：代码没错，只是真实行为和你以为的不一样。 修的办法是把优先级显式写出来（练习 9 就是干这个的），而不是依赖字典顺序。

这两个「缺陷」都是故意留的，正好标出了后面几章的位置：

怎么修	在哪一章
类别不同就走不同的处理流程，而不是共用一个 `assign`	第 23 章条件边
把 `parse` 换成一个能理解自然语言的 Agent 节点	第 25 章确定性 + Agent 节点
大额工单暂停等人工审批，而不是只打个标记	第 26 章 interrupt

注意这个结构的价值：要换掉 parse，只需要保证它仍然返回 {"category": ..., "amount": ...}，assign 一行都不用改。 这就是「节点通过状态解耦」带来的好处。

9.4. 验收清单 #

四步齐全：状态定义、节点函数、add_node / add_edge、compile
能跑：四条工单都有输出，没有异常
状态流动正确：assign 能读到 parse 写进去的 category
部分更新：两个节点都只返回自己负责的字段，没有人返回完整状态
画得出来：draw_mermaid() 输出 __start__ --> parse --> assign --> __end__
看得见过程：stream 能打印出每个节点改了什么

10. 四种静默失败 #

这是本章最该记住的一节。 前面说过 compile() 只保证图能启动，下面这四种情况都不会报任何错，但图的行为和你想的不一样。

10.1. 返回值的字段名拼错 → 更新被丢弃 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

# 返回值里有一个状态定义里不存在的键
def bad(state: S) -> dict:
    # text 会正常写入，"不存在的字段" 会被悄悄丢掉
    return {"text": "x", "不存在的字段": 123}

builder = StateGraph(S)
builder.add_node("bad", bad)
builder.add_edge(START, "bad")
builder.add_edge("bad", END)
graph = builder.compile()

print("结果:", graph.invoke({"text": "", "n": 0}))

结果: {'text': 'x', 'n': 0}

多余的键静默消失。字段名拼错（txet、catgory）时，症状是「这个节点好像没生效」。

排查： 用 stream 打印每个节点的更新，对照状态定义看字段名。注意换成 Pydantic 状态也一样会丢（§4.3 实测过），别指望状态类型帮你挡住这个。

10.2. 节点没有出边 → 图提前结束 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    print("   [clean] 执行")
    return {"text": state["text"].strip()}

def count(state: S) -> dict:
    print("   [count] 执行")
    return {"n": len(state["text"])}

builder = StateGraph(S)
# 第一个节点注册了
builder.add_node("clean", clean)
# 第二个节点也注册了
builder.add_node("count", count)
# 入口边也连了
builder.add_edge(START, "clean")
# 忘了 builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

print("compile 通过了，invoke 看看:")
print("结果:", graph.invoke({"text": " a "}))

mermaid_graph = graph.get_graph().draw_mermaid()
print(mermaid_graph)

compile 通过了，invoke 看看:
   [clean] 执行
结果: {'text': 'a'}

编译通过、运行也不报错，clean 跑完图就结束了。返回值里连 n 都没有，因为 count 根本没执行。

这个坑很阴险：漏 START 那条边会报错，边指向拼错的节点名也会报错（§7.1），但彻底忘记连一条边偏偏不报。

这一节的排查方式要特别说明，因为「看图」这招在这里有个陷阱。 把上面这张断链的图画出来：

graph TD;
    __start__(<p>__start__</p>)
    clean(clean)
    count(count)
    __end__(<p>__end__</p>)
    __start__ --> clean;
    clean --> __end__;
    classDef default fill:#f2f0ff,line-height:1.2
    classDef first fill-opacity:0
    classDef last fill:#bfb6fc

注意 clean --> __end__ 这一行：你的代码里从来没写过这条边。 它是画图时补出来的——§6.3 提过这个渲染规则，任何没有出边的节点，都会被画一条指向 __end__ 的线。

于是就有了一个反直觉的结论：

缺出边时，draw_mermaid() 画出来的是一张「看起来完全正常」的图。 __start__ → clean → __end__ 是一条完整链路，怎么看都不像坏的。

那看图还有用吗？有用，但要看对地方。真正的线索有两条：

线索	具体表现
有节点没有任何连线	`count(count)` 被声明了，但没有任何箭头进出它——它孤零零挂在那儿
`:::first` / `:::last` 标记消失了	正常图是 `__start__([<p>__start__</p>]):::first`，断链图变成了 `__start__(<p>__start__</p>)`，圆角和标记都没了

第二条是个意外好用的信号。只要图的结构不完整（有断链或者孤立节点），LangGraph 就不再给首尾节点打 :::first / :::last。 把两种输出对照一下：

正常：  __start__([<p>__start__</p>]):::first     ← 有圆括号包方括号，有 :::first
异常：  __start__(<p>__start__</p>)               ← 只有普通圆括号，没有 :::first

所以看图时先扫一眼有没有 :::first，比逐条数边快得多。

更可靠的排查方式是用 stream：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

builder = StateGraph(S)
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
builder.add_node("count", lambda s: {"n": len(s["text"])})
builder.add_edge(START, "clean")
# 同样漏掉 clean -> count 这条边
builder.add_edge("count", END)
graph = builder.compile()

# 期望两个 chunk，实际只出来一个，缺的那个就是没执行的节点
for chunk in graph.stream({"text": " a "}):
    # chunk 的键就是节点名，一眼能看出谁跑了谁没跑
    print(chunk)

运行输出：

{'clean': {'text': 'a'}}

只有一个 chunk，count 压根没出现。stream 直接告诉你哪些节点执行了、哪些没有，不受画图渲染规则的干扰。这也是为什么 §10.5 把 stream 排在 draw_mermaid() 前面。

10.3. 孤立节点 → 根本不执行 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

builder = StateGraph(S)
# 这两个节点是正常连好的
builder.add_node("clean", lambda s: {"text": s["text"].strip()})
builder.add_node("count", lambda s: {"n": len(s["text"])})
# 节点注册了，但一条边都没给它连
builder.add_node("orphan", lambda s: {"n": -1})   # 加了，但没连任何边
builder.add_edge(START, "clean")
builder.add_edge("clean", "count")
builder.add_edge("count", END)
graph = builder.compile()

# 如果 orphan 执行了，n 会是 -1
print("compile 通过, 结果:", graph.invoke({"text": " a "}))

compile 通过, 结果: {'text': 'a', 'n': 1}

孤立节点被静默忽略，n 还是 count 算出来的 1，不是 -1。

常见触发场景：加了新节点，但忘了把边从旧路径改道过来。

排查： 看图。孤立节点在 Mermaid 里，是一个被声明了、却没有任何连线的方块：

    __start__(<p>__start__</p>)
    clean(clean)
    count(count)
    orphan(orphan)          ← 声明了，但下面的边里找不到它
    __end__(<p>__end__</p>)
    __start__ --> clean;
    clean --> count;
    count --> __end__;

:::first 标记同样消失了，和 §10.2 的信号一致。这说明 §10.2 和 §10.3 在图上的症状几乎一样——都是「有个方块没连线」。区别只在于：断链时那个孤立方块是你本来想连上的节点，孤立节点则是你新加了却忘了连的。两者的修法相同：把缺的边补上。

10.4. 输入少传字段 → 等到节点里才炸 #

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    n: int

def clean(state: S) -> dict:
    print("   [clean] 收到 state:", state)
    # 状态里没有 text，这一行就会炸
    return {"text": state["text"].strip()}

builder = StateGraph(S)
builder.add_node("clean", clean)
builder.add_edge(START, "clean")
builder.add_edge("clean", END)
graph = builder.compile()

try:
    # 什么都不传，连必需的 text 都没有
    graph.invoke({})
except Exception as e:
    print("报错:", type(e).__name__, e)

   [clean] 收到 state: {}
报错: KeyError 'text'

报错发生在节点内部，信息是 KeyError: 'text'，它不会告诉你「入口少传了字段」。图越大，从这个报错倒推回入口就越费劲。

更麻烦的是，这个错误还会「换皮」。 §4.3 验证过：如果传的是类型不对的值（比如 {"text": 123}），报的就不是 KeyError，而是 AttributeError: 'int' object has no attribute 'strip'。同一个根因（入参有问题），会因为节点怎么用这个值，表现成完全不同的报错。 看到 AttributeError 时，很少有人第一反应是去查 invoke 传了什么。

还有一种更安静的版本：入参的键名拼错。传 {"txet": " a "} 时，那个键会被静默丢弃（§4.3），然后节点报 KeyError: 'text'——你明明「传了」，它却说没有。

防范： 图的入口对外暴露时，用 Pydantic 作为状态类型（§4.3），把校验做在门口。注意 Pydantic 只解决「缺字段」和「类型不对」，解决不了「键名拼错」——多传的键在校验之前就被过滤掉了。

10.5. 汇总 #

症状	可能原因	第一步排查
某个节点「好像没生效」	返回值字段名拼错	`stream` 看该节点返回的键名
后面的节点没跑	缺出边，图提前结束	`stream` 数一下哪些节点出现了（看图会被假的 `__end__` 边骗，§10.2）
新加的节点没执行	孤立节点，没连边	`draw_mermaid()` 找没连线的方块，或看 `:::first` 是否消失
节点里 `KeyError`	入口少传字段，或入参键名拼错	看 `invoke` 传了什么
节点里 `AttributeError` / `TypeError`	入参类型不对，`TypedDict` 没拦	同上
值一直是旧的	上游节点返回值键名拼错，或被覆盖	`stream` 逐节点看

两把万能钥匙：stream() 看数据，draw_mermaid() 看结构。

顺序很重要，要先 stream。它直接告诉你「哪些节点执行了、各自改了什么」，这是事实；画图给的是「结构应该长什么样」，而 §10.2 已经证明画图会补出不存在的边，反而可能误导你。这两招能解决本章 90% 的问题。

11. 实用约定与坑 #

约定	说明
节点名用业务词，别用 `node1`	节点名会出现在图、`stream`、Trace 里（§5.2）
节点只返回自己负责的字段	别返回完整状态，容易误覆盖（§4.2）
显式写 `add_edge(..., END)`	不写也能跑，但「有意结束」和「忘记连边」得能区分开（§6.2）
改状态字段先改定义	否则拼错也不报错（§10.1）
开着编辑器的类型检查	这是唯一能在写代码时发现键名拼错的办法（§5.3）
调试用 `stream` 不用 `invoke`	能看到每个节点改了什么（§7.3）
加完节点看一眼图	找没连线的孤块，顺手确认 `:::first` 还在（§10.2）
对外暴露的图用 Pydantic 状态	把非法输入挡在门口（§4.3）
业务优先级别依赖字典顺序	显式写出优先级，别靠书写顺序（§9.3）
本章只用固定边	分支和循环从第 23 章开始

常见的坑按「会报错」和「不报错」分开列，不报错的那组才是真正耗时间的：

一、会立刻报错（好排查）

现象	原因	处理
`ValueError: Graph must have an entrypoint`	漏了 `add_edge(START, ...)`，或者图是空的	补上入口边（§6.2）
`ValueError: Found edge ending at unknown node`	`add_edge` 里的节点名拼错了	核对节点名（§7.1）
`ValueError: Node 'x' already present`	节点名重复，在 `add_node` 时就报	换个名字（§5.2）
`InvalidUpdateError: Expected dict, got ...`	节点返回了字符串/列表等非字典	包成字典 `return {"字段": 值}`（§4.2）
`InvalidUpdateError: Can receive only one value per step`	并行分支写了同一个键	用 `Annotated` 加 reducer（第 21 章）
`ImportError: Install grandalf`	`draw_ascii()` 需要额外依赖	`pip install grandalf` 或改用 `draw_mermaid()`

二、不报错，但行为不对（静默失败）

现象	原因	处理
节点返回值没生效	字段名不在状态定义里，被静默丢弃。换 Pydantic 也拦不住	核对字段名，开类型检查（§5.3、§10.1）
后半段节点没执行	缺出边，图静默提前结束	用 `stream` 数节点，别只看图（§10.2）
新加的节点没执行	孤立节点，没连边	看图找孤块（§10.3）
入参里的键凭空消失	传了状态定义里没有的键，被静默过滤	核对键名（§4.3）
分类/分派的优先级和预期不符	依赖了字典的书写顺序	显式写优先级（§9.3）
前一个节点写的值不见了	串行写同一字段，后写覆盖先写，且不报错	需要追加就用 reducer（§4.4、第 21 章）

三、报错了但指错了方向

现象	真正的原因	处理
节点里 `KeyError: 'text'`	入口少传字段，或入参键名拼错	看 `invoke` 实际传了什么（§10.4）
节点里 `AttributeError` / `TypeError`	入参类型不对，`TypedDict` 不校验	同上，或换 Pydantic（§4.3）
断链的图「看起来是正常的」	画图会给无出边的节点补一条假的 `__end__` 边	看有没有孤块、`:::first` 是否消失（§10.2）

四、概念性误解

误解	纠正
以为节点会收到上一个节点的返回值	节点收到的是完整状态（§2.3）
状态是 Pydantic，却想在节点里 `return PS(...)`	返回值永远是普通字典（§4.3）
以为 `compile()` 通过就说明图连对了	它只保证图能启动（§7.1）
以为换成 Pydantic 就不会有静默失败了	它只守入口，管不了节点返回值（§4.3）
以为 `add_sequence` 帮你连了 `END`	那条边是画图渲染出来的（§6.3）

口诀：

节点读完整状态，只写自己那几个字段；边只管顺序，不管数据。

12. 练习 #

练习分两组：前五题是机制验证，跑一遍就知道有没有真看懂；后四题是能力建设，需要动手改代码。

机制验证

制造静默失败：故意把 parse 的返回值写成 {"catgory": ...}（拼错），运行看看会发生什么，再用 stream 定位。然后把状态换成 Pydantic 模型再试一次，确认它同样拦不住（§4.3）。
制造断链：删掉 add_edge("parse", "assign")，先用 draw_mermaid() 看——注意你会看到一条 parse --> __end__，而你从没写过这条边。找出图上真正的两个异常信号（孤立的 assign 方块、消失的 :::first），再用 stream 确认哪个节点没执行。
换成 Pydantic：把 TicketState 改成 Pydantic 模型，给 category / amount / owner 加默认值，然后依次试 invoke({})、invoke({"raw": 123})、invoke({"raw": "x", "多余键": 1})，对比三种输入的错误信息（或者没有错误）。
并行实验：让两个节点都从 START 出发，一个写 category 一个写 amount，确认能正常合并；然后改成都写 category，复现 InvalidUpdateError。再用 stream 观察并行时 chunk 的形态是两个还是一个。
返回值边界：让一个节点分别返回 None、{}、{"不存在的字段": 1}、"字符串"，四种情况里只有一种会报错——先猜是哪个，再验证。

能力建设

加第三个节点：给 ticket_graph.py 加一个 notify 节点，接在 assign 之后，打印「已通知 XX 组」且返回 None（只做副作用）。确认 stream 里能看到三个节点、状态没被破坏。
补一个入口校验节点：在 parse 之前加一个节点，检查 raw 是否存在且非空，不合法就写一个明确的错误信息到状态里。对比一下：这样得到的报错，比 §10.4 那个 KeyError 好懂多少？
修掉字典顺序的坑：把 parse 改成不依赖 RULES 书写顺序的写法——显式定义一个优先级列表（比如故障 > 退款 > 咨询），让「故障导致要退款 500 元」正确分给技术支持组（§9.3）。
（扩展）覆盖 vs 追加：给状态加一个 log: list[str] 字段，让每个节点都往里追加一条记录。你会发现后一个节点把前一个的记录冲掉了——这正是下一章 Reducer 要解决的问题。

13. 本章小结 #

图的三个概念：状态（共享数据）、节点（普通函数）、边（执行顺序）。核心机制只有一条——节点读完整状态，返回部分更新，图负责合并。记住那张「完整 / 部分 / 完整」的对照表（§2.3）。
四步搭图：定义状态 → 写节点函数 → add_node / add_edge → compile()。StateGraph 是图纸，compile() 之后才是能跑的机器。
节点就是普通 Python 函数，一个参数、返回字典或 None。因此每个节点都能用普通 assert 单测，不用建图、不花钱——这是图相比 create_agent 的一个实际优势。
返回值只能是字典或 None，返回字符串/列表会报 InvalidUpdateError: Expected dict。这是本章少数「立刻报错」的情况。
状态的三种写法差在校验上：TypedDict 什么都不管（官方主流，本文默认）；Pydantic 在入口拦住缺字段和类型错误，报错信息还会打印你实际传的值；dataclass 缺字段会在入口报 TypeError，但不校验类型。
但三种写法都拦不住两件事：入参多传的键被静默过滤、节点返回值里的错别字被静默丢弃。换 Pydantic 也躲不开静默失败，那得靠编辑器的类型检查。
START 是必须的，END 不是。漏 START 边编译就报错；漏出边则静默提前结束。建议仍然显式写 END，因为「有意结束」和「忘记连边」在代码里长得一样。
同一个字段被写两次，串行是静默覆盖，并行是直接报错。 差别在于并行发生在同一「步」内，没有先后可依据。两者都指向下一章的 Reducer。
compile() 只保证「图能启动」，不保证「图符合意图」。它能抓住「边指向拼错的节点名」，但抓不住「彻底忘记连边」——所以「编译通过」完全不能说明边连对了。
四种静默失败要背下来：字段名拼错（更新被丢弃）、缺出边（提前结束）、孤立节点（不执行）、入口少传字段（节点内 KeyError）。第四种还会「换皮」成 AttributeError / TypeError。
排障顺序是先 stream() 再 draw_mermaid()，不能反。因为画图会给没有出边的节点补一条不存在的 __end__ 边，把断链的图画成看起来正常的样子。图上真正的信号是「有孤立方块」和「:::first 标记消失」。
本章产出：工单分派最小图——两个节点、三条边，能解析工单并分派。它有两个故意留下的缺陷：关键词匹配理解不了同义表达（第二条工单分错了），以及优先级依赖 RULES 字典的书写顺序（「故障导致要退款」被分给了财务组）。前者是第 21、23、24 章的伏笔，后者是练习 8。

下一章解决本章留下的那个问题：同一个字段被多次写入时，怎么让它「追加」而不是「覆盖」——这就是状态与 Reducer。

导航菜单

1. 本章目标 #

2. 三个概念 #

2.1. 一句话版 #

2.2. 为什么不直接写 f(g(x)) #

2.3. 状态是怎么流动的 #

3. 第一张图 #

3.1. 四步走 #

3.2. 完整代码 #

3.3. 拆解 #

4. 状态（State） #

4.1. 状态是图的共享数据 #

4.2. 只返回要改的字段 #

4.3. 三种状态写法 #

4.3.1 TypedDict：不校验 #

4.3.2 Pydantic：会校验 #

4.3.3 dataclass：介于两者之间 #

4.3.4 怎么选 #

4.4. 同一个字段被写两次会怎样 #

5. 节点（Node） #

5.1. 节点就是普通函数 #

5.2. 三种注册方式 #

5.3. 返回值里的多余字段会被静默丢弃 #

6. 边（Edge） #

6.1. add_edge：固定的下一步 #

6.2. START 和 END #

6.3. add_sequence：串行的简写 #

6.4. 一个节点连出多条边 = 并行 #

7. 编译与运行 #

7.1. compile() 检查什么 #

7.2. invoke：跑完拿最终状态 #

7.3. stream：看每一步 #

8. 把图画出来 #

9. 实战：工单分派最小图 #

9.1. ticket_graph.py #

9.2. 运行结果 #

9.3. 第二条工单分错了——这不是 bug，是伏笔 #

9.4. 验收清单 #

10. 四种静默失败 #

10.1. 返回值的字段名拼错 → 更新被丢弃 #

10.2. 节点没有出边 → 图提前结束 #

10.3. 孤立节点 → 根本不执行 #

10.4. 输入少传字段 → 等到节点里才炸 #

10.5. 汇总 #

11. 实用约定与坑 #

12. 练习 #

13. 本章小结 #