LangGraph 入门:状态、节点、边、条件路由

今天要解决的问题

Day 9 你用 100 行代码写了一个 Agent 循环,Day 13 又写了五种 workflow 模式。这些代码有个共同点:控制流全靠 if/while 嵌在函数里。只有三五步的时候这很清爽;等到毕业项目那种「分类 → 检索 → 调工具 → 起草 → 自检 → 可能要人审批 → 可能转人工」的流程,一个函数里套四层 if,再加上中途暂停、断点恢复、流式输出,代码会很快变得没人敢动。

LangGraph 的做法是把控制流画成一张图:每一步是一个节点,步骤之间的跳转是边,所有步骤共享一份状态。今天学完,你应该能:

1. 说清 LangGraph 里 State、节点、边、条件边、reducer 这五个概念各自是什么,并且每个都亲手用一个小脚本跑过(包括并行、循环、Command 和 Send)。
2. 用 LangGraph 把 Day 9 的 Agent 循环重写一遍,并且能指着图说出它和 while 循环一一对应在哪。
3. 写一个没有循环的工作流图:工单分类后按类别走不同分支,再汇合。
4. 用结构化输出驱动路由,知道兼容 OpenAI 协议的国产模型该用哪种结构化输出方式。
5. 回答「什么时候用 LangGraph、什么时候手写循环就够」。

预计用时 6–7 小时,其中「先慢下来」那一节的十一个小脚本约 2 小时,觉得概念吃力就先做它。前置:Day 6(结构化输出)、Day 9(Agent 循环)、Day 13(workflow 模式)。

先说清楚 LangChain 和 LangGraph 的关系

很多人第一次接触时会被这两个名字搞糊涂,面试也常问,先理一下。

LangChain 是 2022 年底出来的大模型应用框架,早期以「链」(chain)为核心,封装了大量模型接口、提示词模板、检索器。它在 2023 年火得一塌糊涂,也因为抽象层太多、版本变动太频繁被骂得很惨,「LangChain 太重」至今是很多工程师的第一印象。

LangGraph 是同一家公司 2024 年推出的底层编排库,专门解决「多步、有状态、可能循环、需要持久化」的流程。它不依赖 LangChain 的链,只借用了 langchain-core 里的消息类型和模型接口。2025 年 10 月两者同时发布 1.0,分工变得很清楚:LangChain 1.x 提供高层的 create_agent,底下跑的就是 LangGraph;想要细粒度控制,直接用 LangGraph 画图。

我的建议是:学 LangGraph,别从 LangChain 的高层封装学起。原因是 LangGraph 的概念很少(状态、节点、边),一旦理解,任何复杂流程你都能自己搭;而高层封装把这些都藏起来了,出了问题你不知道里面发生了什么。等 LangGraph 熟了,再去用 create_agent 省代码,心里是有底的。

还有一个版本提醒。网上 2024–2025 年的大量教程用 from langgraph.prebuilt import create_react_agent,这个函数在 LangGraph 1.0 被标记为弃用,官方推荐换成 langchain.agents.create_agent。今天我们手写节点,不用它,这样换版本也不受影响。

四个概念:State、节点、边、reducer

State(状态) 是整张图共享的一份数据,通常用 TypedDict 定义。比如一个 Agent 的状态就是消息列表加一个调用计数:

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    llm_calls: int

节点(node) 是一个普通的 Python 函数,输入是当前 State,输出是一个字典,表示「我要更新状态里的哪些字段」。注意是更新,不是替换整个状态。节点只需要返回它关心的字段:

def call_llm(state: State) -> dict:
    reply = llm_with_tools.invoke(state["messages"])
    return {"messages": [reply], "llm_calls": state["llm_calls"] + 1}

边(edge) 决定执行完一个节点后去哪。普通边是固定的:add_edge("tools", "llm"),工具执行完一定回到模型。条件边根据状态决定去向:add_conditional_edges("llm", route, ["tools", END]),其中 route 是一个函数,看最后一条消息有没有工具调用,返回下一个节点的名字。START 和 END 是两个特殊节点,表示图的入口和出口。

reducer 是最容易被忽略、也最容易踩坑的概念。节点返回 {"messages": [reply]} 时,LangGraph 要决定:是用这个新列表替换旧的 messages,还是追加到后面?默认是替换。Annotated[list, add_messages] 就是告诉它:这个字段用 add_messages 函数来合并,也就是追加(遇到相同 id 的消息则更新)。llm_calls 没有 reducer,所以直接覆盖。

这里是新手最常踩的坑:忘了加 add_messages,每个节点返回的消息就会把整个列表覆盖掉,模型只看得到最后一条,完全不记得前面问了什么,而程序不会报任何错。动手练习第 2 题会让你亲手制造一次这个问题。所以:凡是要累积的字段,一定要配 reducer。除了 add_messages,常见的还有 operator.add(列表拼接、数字相加)。

用后端的经验类比:State 像一个请求上下文对象,节点像中间件或处理器,边像路由表,reducer 像合并策略。区别是这个「请求」可能要跑几十步、中途停下来等人、第二天再接着跑,所以状态需要能完整保存下来,明天我们讲 checkpoint 时就会看到这一点。

先慢下来:十一个小脚本,一次只学一个零件

如果你看了下面「用图重写 Day 9」那段 260 行的代码觉得一下子接不住,先做这一节。尚硅谷那套 LangGraph 课光是 state、边、循环就讲了四十集,我们不必那么细,但每个零件至少要亲手跑一次、看一次输出,后面的大代码才读得进去。

这一节的脚本全都不调模型,节点就是普通 Python 函数,跑起来不花钱,一秒出结果。每个脚本只多一个新东西。建议的做法是:先看我写的那一两段话,再自己敲一遍(别复制),跑出来和我贴的输出对一下,对不上就停下来想为什么。

准备工作只有一步。还没装 LangGraph 的话先装,然后建个目录放这些脚本:

cd agent-60-lab
uv add langgraph langchain-openai python-dotenv
mkdir -p w4/lg_steps

下面所有命令都在 agent-60-lab 根目录跑,形如 uv run python w4/lg_steps/s01_one_node.py。我用的是 langgraph 1.2.14,你装到更新的 1.x 版本,输出应该一样。

第 1 步:只有一个节点的图

先把骨架记住:定义 State → 建 StateGraph → 加节点 → 连边 → compile() → invoke()。不管多复杂的图都是这六步。

# agent-60-lab/w4/lg_steps/s01_one_node.py
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    ticket: str
    category: str


def classify(state: State) -> dict:
    print("classify 收到:", state)
    text = state["ticket"]
    return {"category": "billing" if "发票" in text else "other"}


builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_edge(START, "classify")
builder.add_edge("classify", END)
graph = builder.compile()

result = graph.invoke({"ticket": "上个月的发票开错了"})
print("最终状态:", result)

输出(实跑,langgraph 1.2.14):

classify 收到: {'ticket': '上个月的发票开错了'}
最终状态: {'ticket': '上个月的发票开错了', 'category': 'billing'}

对应尚硅谷第 9–12 集(三要素、图运行过程、完成基础的 graph 流程)。

第 2 步:两个节点串起来,用 stream 看每一步

真实流程不会只有一步。这里加一个 draft 节点,并且换用 stream 代替 invoke:invoke 只给你最后的结果,stream(..., stream_mode="updates") 每跑完一个节点就吐出这个节点改了什么。调试的时候它比 print 好用得多。

# agent-60-lab/w4/lg_steps/s02_sequence.py
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    ticket: str
    category: str
    reply: str


def classify(state: State) -> dict:
    return {"category": "billing" if "发票" in state["ticket"] else "other"}


def draft(state: State) -> dict:
    return {"reply": f"[{state['category']}] 已收到:{state['ticket']}"}


builder = StateGraph(State)
builder.add_sequence([classify, draft])  # 等于 add_node×2 + add_edge("classify","draft")
builder.add_edge(START, "classify")
builder.add_edge("draft", END)
graph = builder.compile()

for step in graph.stream({"ticket": "发票抬头写错了"}, stream_mode="updates"):
    print(step)

输出(实跑,langgraph 1.2.14):

{'classify': {'category': 'billing'}}
{'draft': {'reply': '[billing] 已收到:发票抬头写错了'}}

两行输出就是两个节点各自的「改动记录」。后面调复杂的图,第一件事就是用这个模式看它到底走了哪些节点。对应尚硅谷第 28–29 集(边的简单使用、add_sequence)。

第 3 步:State 的三种写法

State 除了 TypedDict,还能用 dataclass 和 pydantic。区别主要在两点:节点里拿到的是 dict 还是对象;要不要在运行时校验类型。

# agent-60-lab/w4/lg_steps/s03_state_styles.py
from dataclasses import dataclass
from typing import TypedDict

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


class TDState(TypedDict):
    amount: float


@dataclass
class DCState:
    amount: float
    currency: str = "CNY"


class PDState(BaseModel):
    amount: float


def double(state) -> dict:
    amount = state["amount"] if isinstance(state, dict) else state.amount
    print("  节点里拿到的 state 类型:", type(state).__name__)
    return {"amount": amount * 2}


def build(schema):
    b = StateGraph(schema)
    b.add_node("double", double)
    b.add_edge(START, "double")
    b.add_edge("double", END)
    return b.compile()


print("TypedDict:", build(TDState).invoke({"amount": 10}))
print("dataclass:", build(DCState).invoke({"amount": 10}))
print("pydantic :", build(PDState).invoke({"amount": "10"}))
try:
    build(PDState).invoke({"amount": "十块"})
except Exception as e:
    print("pydantic 拦下了:", type(e).__name__)

输出(实跑,langgraph 1.2.14):

  节点里拿到的 state 类型: dict
TypedDict: {'amount': 20}
  节点里拿到的 state 类型: DCState
dataclass: {'amount': 20}
  节点里拿到的 state 类型: PDState
pydantic : {'amount': 20.0}
pydantic 拦下了: ValidationError

本课后面统一用 TypedDict,因为官方文档和大多数开源项目都这么写,读别人代码时不用切换。外部输入不可信、想在入口就拦住脏数据时,再考虑 pydantic。对应尚硅谷第 14–15 集。

第 4 步:不加 reducer,列表会被覆盖

这一步专门演示一个坑。两个节点都往 notes 里写一条记录,你可能以为最后会有三条。

# agent-60-lab/w4/lg_steps/s04_overwrite.py
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    notes: list[str]
    steps: int


def a(state: State) -> dict:
    return {"notes": ["a 查了客户"], "steps": state["steps"] + 1}


def b(state: State) -> dict:
    return {"notes": ["b 查了发票"]}  # 没返回 steps,steps 保持原值


builder = StateGraph(State)
builder.add_sequence([a, b])
builder.add_edge(START, "a")
builder.add_edge("b", END)
graph = builder.compile()

print(graph.invoke({"notes": ["用户来信"], "steps": 0}))

输出:

{'notes': ['b 查了发票'], 'steps': 1}

规则就一句:没有 reducer 的字段,谁最后写谁说了算。这正是开头说的「忘了 add_messages,模型只记得最后一条」那个坑,而且程序不报错。

第 5 步:用 reducer 让字段累积

给字段配上合并函数,写法是 Annotated[类型, 合并函数]。operator.add 对列表是拼接,对数字是相加;add_messages 是专门给消息列表用的,多一个本事:id 相同的消息会被替换而不是重复追加。

# agent-60-lab/w4/lg_steps/s05_reducer.py
import operator
from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


class State(TypedDict):
    notes: Annotated[list[str], operator.add]
    cost: Annotated[float, operator.add]
    messages: Annotated[list[AnyMessage], add_messages]


def a(state: State) -> dict:
    return {"notes": ["a 查了客户"], "cost": 0.01,
            "messages": [AIMessage("草稿 v1", id="draft")]}


def b(state: State) -> dict:
    return {"notes": ["b 查了发票"], "cost": 0.02,
            "messages": [AIMessage("草稿 v2", id="draft")]}  # 同 id:替换而不是追加


builder = StateGraph(State)
builder.add_sequence([a, b])
builder.add_edge(START, "a")
builder.add_edge("b", END)
graph = builder.compile()

out = graph.invoke({"notes": ["用户来信"], "cost": 0.0,
                    "messages": [HumanMessage("发票开错了", id="q1")]})
print("notes:", out["notes"])
print("cost :", round(out["cost"], 2))
print("messages:", [(m.id, m.content) for m in out["messages"]])

输出(实跑,langgraph 1.2.14):

notes: ['用户来信', 'a 查了客户', 'b 查了发票']
cost : 0.03
messages: [('q1', '发票开错了'), ('draft', '草稿 v2')]

add_messages 按 id 替换这个特性,后面做「改写上一条回复」「人工修改草稿后放回去」时会用到。对应尚硅谷第 16–18 集(reducer)和第 26–27 集(预定义状态、MessagesState)。

第 6 步:两个节点同时跑,这时 reducer 是必需的

客服 Agent 常见的需求:查知识库和查订单库互不依赖,应该同时做。LangGraph 里让 START 连出两条边,两个节点就会在同一步并行执行;再用 add_edge([a, b], c) 表示「a 和 b 都跑完才进 c」。

# agent-60-lab/w4/lg_steps/s06_parallel.py
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph


class Bad(TypedDict):
    found: list[str]


class Good(TypedDict):
    found: Annotated[list[str], operator.add]


def search_kb(state) -> dict:
    return {"found": ["kb-003 退款政策"]}


def query_db(state) -> dict:
    return {"found": ["INV-0042 已付款"]}


def merge(state) -> dict:
    print("  merge 看到:", state["found"])
    return {}


def build(schema):
    b = StateGraph(schema)
    b.add_node("search_kb", search_kb)
    b.add_node("query_db", query_db)
    b.add_node("merge", merge)
    b.add_edge(START, "search_kb")
    b.add_edge(START, "query_db")          # START 连两条边 = 两个节点同时跑
    b.add_edge(["search_kb", "query_db"], "merge")  # 两个都跑完才进 merge
    b.add_edge("merge", END)
    return b.compile()


print("有 reducer:")
build(Good).invoke({"found": []})
print("没有 reducer:")
try:
    build(Bad).invoke({"found": []})
except Exception as e:
    print(" ", type(e).__name__, str(e).splitlines()[0])

输出(实跑,langgraph 1.2.14):

有 reducer:
  merge 看到: ['INV-0042 已付款', 'kb-003 退款政策']
没有 reducer:
  InvalidUpdateError At key 'found': Can receive only one value per step. Use an Annotated key to handle multiple values.

对比第 4 步:串行时没 reducer 是静悄悄地覆盖,并行时没 reducer 是直接报错。报错反而是好事。错误信息里那句 Use an Annotated key 就是在提示你加 reducer。对应尚硅谷第 21 集(节点的并行执行)、第 30–31 集和第 38 集(静态扇入)。

第 7 步:条件边,按状态走不同分支

普通边是写死的,条件边是「跑完这个节点后,调一个函数决定下一站」。这就是工单分流的核心。

# agent-60-lab/w4/lg_steps/s07_conditional.py
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    ticket: str
    category: str
    reply: str


def classify(state: State) -> dict:
    t = state["ticket"]
    cat = "billing" if "发票" in t else "refund" if "退款" in t else "other"
    return {"category": cat}


def billing(state: State) -> dict:
    return {"reply": "转账单组"}


def refund(state: State) -> dict:
    return {"reply": "走退款审批"}


def human(state: State) -> dict:
    return {"reply": "转人工"}


def route(state: State) -> Literal["billing", "refund", "other"]:
    return state["category"]


builder = StateGraph(State)
for fn in (classify, billing, refund, human):
    builder.add_node(fn.__name__, fn)
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify", route,
    {"billing": "billing", "refund": "refund", "other": "human"},  # path_map:返回值 → 节点名
)
for n in ("billing", "refund", "human"):
    builder.add_edge(n, END)
graph = builder.compile()

for t in ["发票开错了", "我要退款", "怎么改密码"]:
    print(t, "→", graph.invoke({"ticket": t})["reply"])

输出(实跑,langgraph 1.2.14):

发票开错了 → 转账单组
我要退款 → 走退款审批
怎么改密码 → 转人工

对应尚硅谷第 32–34 集(条件分支、path_map、映射多节点)。

第 8 步:循环,以及怎么防死循环

Agent 循环本质上就是一条指回自己(或指回上一个节点)的边。下面的 write 节点每次多加一个感叹号,check 看够不够三个,不够就回去重写。第二段故意写一个永远不满意的路由,看 LangGraph 怎么兜底。

# agent-60-lab/w4/lg_steps/s08_loop.py
from typing import TypedDict

from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    draft: str
    tries: int


def write(state: State) -> dict:
    tries = state["tries"] + 1
    return {"draft": "您好" + "!" * tries, "tries": tries}


def check(state: State) -> str:
    return "done" if state["tries"] >= 3 else "again"


def broken_check(state: State) -> str:
    return "again"  # 永远不满意:故意写出一个死循环


def build(route):
    b = StateGraph(State)
    b.add_node("write", write)
    b.add_edge(START, "write")
    b.add_conditional_edges("write", route, {"again": "write", "done": END})
    return b.compile()


print(build(check).invoke({"draft": "", "tries": 0}))

try:
    build(broken_check).invoke({"draft": "", "tries": 0}, config={"recursion_limit": 5})
except GraphRecursionError as e:
    print("死循环被拦下:", str(e).splitlines()[0])

输出(实跑,langgraph 1.2.14):

{'draft': '您好!!!', 'tries': 3}
死循环被拦下: Recursion limit of 5 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.

对应尚硅谷第 40–47 集(循环结构、限制循环步数、主动退出和被动退出)。

第 9 步:Command,在节点里同时改状态和指定下一站

条件边要写两个东西:节点改状态,路由函数选去向。有时候判断逻辑和改状态是一回事,拆成两处反而别扭。Command 让节点一次返回两样:update 是要改的字段,goto 是下一个节点。

# agent-60-lab/w4/lg_steps/s09_command.py
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command


class State(TypedDict):
    amount: float
    reply: str


def check(state: State) -> Command[Literal["auto_refund", "need_approval"]]:
    if state["amount"] <= 100:
        return Command(update={"reply": "小额,自动退"}, goto="auto_refund")
    return Command(update={"reply": "大额,要审批"}, goto="need_approval")


def auto_refund(state: State) -> dict:
    return {"reply": state["reply"] + " → 已退款"}


def need_approval(state: State) -> dict:
    return {"reply": state["reply"] + " → 已提交审批"}


builder = StateGraph(State)
builder.add_node("check", check)
builder.add_node("auto_refund", auto_refund)
builder.add_node("need_approval", need_approval)
builder.add_edge(START, "check")
builder.add_edge("auto_refund", END)
builder.add_edge("need_approval", END)
graph = builder.compile()

print(graph.invoke({"amount": 59, "reply": ""})["reply"])
print(graph.invoke({"amount": 6578, "reply": ""})["reply"])

输出(实跑,langgraph 1.2.14):

小额,自动退 → 已退款
大额,要审批 → 已提交审批

什么时候用条件边、什么时候用 Command?我的习惯是:路由只看状态、和节点逻辑无关时用条件边,图结构一眼能看清;路由判断本身就是这个节点的主要工作时用 Command。两种别在同一个节点上混用。对应尚硅谷第 37 集。

第 10 步:Send,数量不固定的并行

第 6 步的并行是写死的两条边。如果要并行检查的发票有 3 张还是 30 张,运行时才知道,就用 Send:路由函数返回一组 Send(节点名, 给这个节点的输入),每个 Send 起一个独立的任务。这就是 map-reduce。

# agent-60-lab/w4/lg_steps/s10_send.py
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send


class State(TypedDict):
    invoices: list[str]
    checked: Annotated[list[str], operator.add]


class OneInvoice(TypedDict):
    invoice: str


def fan_out(state: State) -> list[Send]:
    return [Send("check_one", {"invoice": inv}) for inv in state["invoices"]]


def check_one(state: OneInvoice) -> dict:
    return {"checked": [f"{state['invoice']} ok"]}


builder = StateGraph(State)
builder.add_node("check_one", check_one)
builder.add_conditional_edges(START, fan_out, ["check_one"])
builder.add_edge("check_one", END)
graph = builder.compile()

print(graph.invoke({"invoices": ["INV-1", "INV-2", "INV-3"], "checked": []}))

输出(实跑,langgraph 1.2.14):

{'invoices': ['INV-1', 'INV-2', 'INV-3'], 'checked': ['INV-1 ok', 'INV-2 ok', 'INV-3 ok']}

这一步属于选学,今天用不到,Day 92 做 Deep Research 时要并行查一批子问题,会再见到它。对应尚硅谷第 36 集(动态分支)和第 39 集(动态扇入)。

第 11 步:把图画出来

最后一个小工具:graph.get_graph().draw_mermaid() 输出 Mermaid 格式的图,贴到 mermaid.live 就能看到。图一复杂,先画出来对一眼结构,比读代码快。

# agent-60-lab/w4/lg_steps/s11_draw.py
from s07_conditional import builder

graph = builder.compile()
print(graph.get_graph().draw_mermaid())

运行时要在 w4/lg_steps 目录里跑(cd w4/lg_steps && uv run python s11_draw.py),因为它要 import 同目录的 s07。输出的后半段(实跑,省略了开头三行和样式行):

graph TD;
	__start__([<p>__start__</p>]):::first
	classify(classify)
	billing(billing)
	refund(refund)
	human(human)
	__end__([<p>__end__</p>]):::last
	__start__ --> classify;
	classify -.-> billing;
	classify -. &nbsp;other&nbsp; .-> human;
	classify -.-> refund;
	billing --> __end__;
	human --> __end__;
	refund --> __end__;

在 Jupyter 里可以直接 graph.get_graph().draw_mermaid_png() 出图片,不过那个方法默认要联网调 mermaid.ink 的接口渲染。对应尚硅谷第 13 集。

十一步走完,回头对一下

带着这张表往下读,260 行的代码就只是把这些零件拼在一起,再把「节点函数」从玩具换成真的调模型和调工具。

用图重写 Day 9 的 Agent 循环

回忆一下 Day 9 的循环:

while steps < MAX_STEPS:
    reply = call_model(messages, tools)       # 1. 调模型
    messages.append(reply)
    if not reply.tool_calls:                  # 2. 没有工具调用就结束
        break
    for call in reply.tool_calls:             # 3. 执行工具,结果追加
        messages.append(run_tool(call))

翻译成图:「调模型」是一个节点 llm,「执行工具」是一个节点 tools,第 2 步的判断是 llm 后面的条件边,循环本身就是 tools → llm 那条普通边。

先建一个共用的模型对象。LangGraph 本身不关心你用哪家模型,我们用 langchain-openai 里的 ChatOpenAI,它走 OpenAI 兼容协议,所以 DeepSeek、通义、Kimi 都能接,只改 .env:

cd agent-60-lab
uv add langgraph langchain-openai python-dotenv
mkdir -p w4/lgdemo

目录名我特意叫 lgdemo(LangGraph demo)。千万别叫 langgraph,那会和库本身重名,import langgraph 时 Python 会先找到你的目录,报一堆莫名其妙的 ImportError;也别叫 graph,Day 28 的小项目里会有一个 w4/graph.py,同名的目录和文件放在一起很容易导入错对象。

# agent-60-lab/w4/lgdemo/llm.py
"""LangGraph 这几天共用的模型对象。走 OpenAI 兼容协议,换厂商只改 .env。"""
import json
import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()


def get_llm(temperature: float = 0) -> ChatOpenAI:
    return ChatOpenAI(
        model=os.environ["LLM_MODEL"],
        base_url=os.environ["LLM_BASE_URL"],
        api_key=os.environ["LLM_API_KEY"],
        temperature=temperature,
        # Day 3 的 LLM_EXTRA_BODY:DeepSeek 要靠它关掉思考模式,否则下面的结构化输出会报 400
        extra_body=json.loads(os.environ.get("LLM_EXTRA_BODY", "{}")),
    )

extra_body 那一行是今天最容易漏的:用 DeepSeek 时它负责关掉思考模式,漏了的话,后面用 with_structured_output 的分类节点会直接报 400。

然后是图本身:

# agent-60-lab/w4/lgdemo/agent_graph.py
"""用 LangGraph 重写 Day 9 的 Agent 循环:两个节点(llm、tools)+ 一条条件边。"""
from typing import Annotated, Literal

from langchain_core.messages import AnyMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict

from llm import get_llm

MAX_LLM_CALLS = 6


@tool
def get_weather(city: str) -> str:
    """查询城市当前天气。city 用中文城市名,如「上海」。"""
    fake = {"上海": "小雨 19°C", "北京": "晴 15°C", "广州": "多云 27°C"}
    return fake.get(city, f"没有 {city} 的天气数据")


@tool
def calculate(expression: str) -> str:
    """计算一个只含数字和 + - * / ( ) 的算式,返回结果。"""
    if not set(expression) <= set("0123456789+-*/(). "):
        return "表达式里有不允许的字符"
    return str(eval(expression))  # 字符集已限制,这里只为演示


TOOLS = [get_weather, calculate]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]  # reducer:新消息追加而不是覆盖
    llm_calls: int


llm_with_tools = get_llm().bind_tools(TOOLS)


def call_llm(state: State) -> dict:
    system = SystemMessage("你是一个助手。需要事实或计算时调用工具,不要心算。")
    reply = llm_with_tools.invoke([system] + state["messages"])
    return {"messages": [reply], "llm_calls": state.get("llm_calls", 0) + 1}


def run_tools(state: State) -> dict:
    results = []
    for call in state["messages"][-1].tool_calls:
        output = TOOLS_BY_NAME[call["name"]].invoke(call["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=call["id"]))
    return {"messages": results}


def route(state: State) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None) and state["llm_calls"] < MAX_LLM_CALLS:
        return "tools"
    return END


builder = StateGraph(State)
builder.add_node("llm", call_llm)
builder.add_node("tools", run_tools)
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", route, ["tools", END])
builder.add_edge("tools", "llm")
graph = builder.compile()

if __name__ == "__main__":
    print(graph.get_graph().draw_mermaid())
    out = graph.invoke({"messages": [("user", "上海和北京现在温差几度?")], "llm_calls": 0})
    for m in out["messages"]:
        m.pretty_print()
    print("LLM 调用次数:", out["llm_calls"])

cd w4/lgdemo
uv run python agent_graph.py

开头会先打印一段 Mermaid 格式的图定义,把它贴到任何支持 Mermaid 的地方(GitHub 的 markdown、Typora、mermaid.live)就能看到图。然后是运行过程,输出大致长这样:

================================ Human Message =================================
上海和北京现在温差几度?
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_1)  Args: city: 上海
  get_weather (call_2)  Args: city: 北京
================================= Tool Message =================================
小雨 19°C
================================= Tool Message =================================
晴 15°C
================================== Ai Message ==================================
Tool Calls:
  calculate (call_3)  Args: expression: 19-15
================================= Tool Message =================================
4
================================== Ai Message ==================================
上海现在小雨 19°C,北京晴 15°C,温差 4 度。
LLM 调用次数: 3

第一次模型调用里出现了两个并行的 get_weather,run_tools 节点一次执行完两个,再回到 llm。不同模型的行为会有差别:有的模型会一次调完两个城市,有的会分两轮调;有的会老老实实调 calculate,有的会自己心算。这正是 Day 12 说的「Agent 行为不确定」,W5 的评测要解决的就是怎么在这种不确定性下衡量质量。

几个值得注意的细节:

• @tool 装饰器来自 langchain-core,和 MCP 的 @mcp.tool() 思路一样,从函数签名和 docstring 生成 schema。bind_tools(TOOLS) 把工具定义附在每次请求上。
• route 函数里我加了 llm_calls < MAX_LLM_CALLS 的判断,这就是 Day 12 讲的步数上限。LangGraph 自己还有一个 recursion_limit,超过会抛 GraphRecursionError。早期版本默认是 25 步,我在 langgraph 1.2.14 的源码里查到默认值已经变成 10007,基本挡不住一个失控的 Agent。所以一定要在业务层自己设一个小的上限,出了问题能优雅结束,而不是烧几千次调用之后抛异常。也可以在调用时传 config={"recursion_limit": 20} 做第二道保险。
• 返回值类型标注 Literal["tools", "__end__"] 不是必须的,但加上之后画图时 LangGraph 知道条件边可能去哪,图会画得更准。add_conditional_edges 的第三个参数传了目标列表,作用相同。
• 官方的 langgraph.prebuilt 里有现成的 ToolNode 和 tools_condition,可以替代我手写的 run_tools 和 route。它们还会帮你处理工具抛异常、并行执行等细节。今天手写是为了看清楚原理,项目里用现成的没问题。

第二张图:没有循环的工作流

Agent 循环只是 LangGraph 能画的图之一。实际项目里更多的是「workflow」:步骤基本固定,只在某些点做分支。Day 13 讲过,能用 workflow 解决的问题就别上 Agent,更可控、更便宜、更好测。

CloudLedger 客服的第一步就是一个典型的 workflow:来了一张工单,先分类,账务类交给账务组(后面要查发票和订阅),技术类交给技术支持(先查知识库),其他的交给一线客服;紧急的再加一个标记。

# agent-60-lab/w4/lgdemo/triage_graph.py
"""一个不带循环的工作流图:工单分类 → 按类别走不同分支 → 汇合。演示 State、结构化输出和条件路由。"""
from typing import Literal

from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from typing_extensions import TypedDict

from llm import get_llm


class Triage(BaseModel):
    category: Literal["billing", "technical", "account", "other"] = Field(description="工单类别")
    urgency: Literal["low", "normal", "high"] = Field(description="紧急程度,影响付款或无法登录算 high")
    summary: str = Field(description="一句话概括客户诉求,不超过 30 字")


class TicketState(TypedDict, total=False):
    ticket: str
    triage: Triage
    owner: str
    note: str


# DeepSeek 等兼容厂商不一定支持 json_schema 模式,用 function_calling 方式最稳
classifier = get_llm().with_structured_output(Triage, method="function_calling")


def classify(state: TicketState) -> dict:
    result = classifier.invoke(
        "把下面的客服工单分类。\n"
        "billing=扣款/退款/发票/价格;technical=报错/功能异常;account=登录/权限/注销;其余 other。\n\n"
        f"工单:{state['ticket']}"
    )
    return {"triage": result}


def to_billing(state: TicketState) -> dict:
    return {"owner": "账务组", "note": "需要查发票和订阅,W4 后面会接 MCP 工具"}


def to_technical(state: TicketState) -> dict:
    return {"owner": "技术支持", "note": "先查知识库里的报错码文章"}


def to_human(state: TicketState) -> dict:
    return {"owner": "一线客服", "note": "账号类或无法归类,一线客服先接"}


def pick_branch(state: TicketState) -> Literal["billing", "technical", "human"]:
    c = state["triage"].category
    if c == "billing":
        return "billing"
    if c == "technical":
        return "technical"
    return "human"


def escalate_if_urgent(state: TicketState) -> dict:
    if state["triage"].urgency == "high":
        return {"note": state["note"] + ";紧急,30 分钟内响应"}
    return {}


builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("billing", to_billing)
builder.add_node("technical", to_technical)
builder.add_node("human", to_human)
builder.add_node("escalate", escalate_if_urgent)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", pick_branch, ["billing", "technical", "human"])
for branch in ["billing", "technical", "human"]:
    builder.add_edge(branch, "escalate")
builder.add_edge("escalate", END)
graph = builder.compile()

if __name__ == "__main__":
    for text in [
        "10 月 8 号被扣了两次款,麻烦退一笔",
        "导出年度报表一直提示 ERR_EXPORT_TIMEOUT",
        "验证码一直过期,登录不了,急!",
    ]:
        out = graph.invoke({"ticket": text})
        t = out["triage"]
        print(f"{text[:16]:<18} → {t.category:<9} {t.urgency:<6} {out['owner']} | {out['note']}")

几个要点:

用结构化输出驱动路由。 分类节点不让模型自由发挥,而是用 Day 6 学的结构化输出,要求它返回一个 Triage 对象,category 是四选一的 Literal。这样条件边函数 pick_branch 拿到的一定是合法值,不会出现模型回一句「这应该算是账务问题吧」导致路由失败的情况。

国产模型用 method="function_calling"。 with_structured_output 有几种实现方式:json_schema 依赖厂商支持 OpenAI 的 Structured Outputs(schema 严格约束),function_calling 是把 schema 伪装成一个工具让模型「调用」,json_mode 只保证输出是 JSON。DeepSeek 等兼容厂商对 json_schema 的支持情况不一,function_calling 是最稳的选择,因为只要支持工具调用就能用。这是一个面试时能体现实战经验的小细节。

State 用 total=False。 工作流里的字段是一步步填进去的,刚开始只有 ticket,triage 要等分类后才有。total=False 表示所有字段都是可选的,避免类型检查报错。

分支汇合。 三个分支都有一条普通边指向 escalate,LangGraph 会在任何一个分支完成后执行它。如果你让多个分支同时执行(比如从 START 同时连到两个节点),它们会并行跑,汇合节点会等所有分支都完成,这叫 fan-out / fan-in,Day 26 讲多 Agent 时会用到。

运行结果大致长这样:

10 月 8 号被扣了两次款,麻   → billing   normal 账务组 | 需要查发票和订阅,W4 后面会接 MCP 工具
导出年度报表一直提示 ERR_E   → technical normal 技术支持 | 先查知识库里的报错码文章
验证码一直过期,登录不了,急!    → account   high   一线客服 | 账号类或无法归类,一线客服先接;紧急,30 分钟内响应

你可以多跑几次、多换几张工单(从 Day 23 的 tickets 表里取),看分类稳不稳定。如果发现「取消订阅」时而被分到 billing 时而被分到 account,说明分类标准本身有歧义,该改的是提示词里的类别定义,不是去调模型参数。

图编译之后是什么

builder.compile() 返回的是一个 CompiledStateGraph,它实现了 LangChain 的 Runnable 接口,所以有一组统一的方法:

• invoke(input):同步跑完,返回最终状态;
• stream(input, stream_mode=...):边跑边吐中间结果;
• ainvoke / astream:异步版本,节点用 async def 时用它们;
• get_graph():拿到图结构,可以画图;
• get_state(config) / get_state_history(config):查看持久化的状态,要配合 checkpointer,明天讲。

stream 很常用,它让你在长流程里实时看到每一步。最常用的模式是 "updates",每执行完一个节点吐一次该节点的更新:

for chunk in graph.stream({"messages": [("user", "上海天气?")], "llm_calls": 0}, stream_mode="updates"):
    for node, update in chunk.items():
        print(f"[{node}]", update["messages"][-1].content or update["messages"][-1].tool_calls)

输出类似:

[llm] [{'name': 'get_weather', 'args': {'city': '上海'}, 'id': '...', 'type': 'tool_call'}]
[tools] 小雨 19°C
[llm] 上海小雨 19°C。

这对调试非常有用,也是 W6 做前端流式展示「Agent 正在调用什么工具」的基础。其他模式("values" 每步吐完整状态、"messages" 逐 token 吐模型输出)明天再讲。

什么时候值得用 LangGraph

说实话,今天这两张图手写也就几十行,用 LangGraph 并没有省代码。它的价值不在今天,在后面这几件事:

• 持久化和断点恢复:状态自动存到 SQLite/Postgres,进程挂了能从上一步接着跑,明天讲。
• 人工介入:流程中间停下来等人审批,可能等几个小时,明天讲。
• 流式和可观测:每一步的输入输出都能实时拿到,和 LangSmith、Langfuse 这类工具天然对接,W5 讲。
• 复杂流程的可读性:十几个节点、多处分支和循环的流程,画成图比读嵌套 if 容易得多,新人接手也快。

反过来,这些情况我不建议上 LangGraph:一问一答的简单调用;固定三步、不需要持久化的流水线;团队里没人熟悉它、项目又很赶。框架是有学习和维护成本的。面试被问「为什么用 LangGraph」时,答出上面这几条具体收益,比说「它是业界主流」有说服力得多。Day 27 会把几个主流框架放在一起对比。

动手练习

1. 跑通 agent_graph.py,把打印的 Mermaid 贴到 mermaid.live 看图。

cd agent-60-lab/w4/lgdemo
uv run python agent_graph.py

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

    messages: list[AnyMessage]  # 练习 2:去掉 reducer

# agent-60-lab/w4/lgdemo/ex24_2_probe.py
"""练习 2 的观察脚本:每跑完一个节点,打印一次状态里有哪些消息。"""
from agent_graph import graph

inputs = {"messages": [("user", "上海和北京现在温差几度?")], "llm_calls": 0}
try:
    for state in graph.stream(inputs, stream_mode="values"):
        kinds = [type(m).__name__ if not isinstance(m, tuple) else "tuple" for m in state["messages"]]
        print(f"llm_calls={state['llm_calls']}  messages={kinds}")
except Exception as e:
    print("出错:", type(e).__name__, str(e)[:120])

llm_calls=0  messages=['tuple']
llm_calls=1  messages=['AIMessage']
llm_calls=1  messages=['ToolMessage', 'ToolMessage']
出错: OpenAIInvalidRequestError Error code: 400 - {'error': {'message': "Messages with role 'tool' must be a response to a preceding message with 'tool_

llm_calls=0  messages=['HumanMessage']
llm_calls=1  messages=['HumanMessage', 'AIMessage']
llm_calls=1  messages=['HumanMessage', 'AIMessage', 'ToolMessage', 'ToolMessage']
llm_calls=2  messages=['HumanMessage', 'AIMessage', 'ToolMessage', 'ToolMessage', 'AIMessage']
llm_calls=2  messages=['HumanMessage', 'AIMessage', 'ToolMessage', 'ToolMessage', 'AIMessage', 'ToolMessage']
llm_calls=3  messages=['HumanMessage', 'AIMessage', 'ToolMessage', 'ToolMessage', 'AIMessage', 'ToolMessage', 'AIMessage']

# agent-60-lab/w4/lgdemo/agent_graph_prebuilt.py
"""练习 3:用 LangGraph 自带的 ToolNode 和 tools_condition 换掉手写的 run_tools 和 route。"""
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

from agent_graph import TOOLS, State, call_llm

builder = StateGraph(State)
builder.add_node("llm", call_llm)
builder.add_node("tools", ToolNode(TOOLS))  # 名字必须叫 "tools",tools_condition 只认这个
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", tools_condition)  # 有工具调用 → "tools",否则 → END
builder.add_edge("tools", "llm")
graph = builder.compile()

if __name__ == "__main__":
    out = graph.invoke({"messages": [("user", "上海和北京现在温差几度?")], "llm_calls": 0})
    for m in out["messages"]:
        calls = [(c["name"], c["args"]) for c in getattr(m, "tool_calls", [])]
        print(f"{type(m).__name__:<12} {calls or m.content}")
    print("LLM 调用次数:", out["llm_calls"])

HumanMessage 上海和北京现在温差几度?
AIMessage    [('get_weather', {'city': '上海'}), ('get_weather', {'city': '北京'})]
ToolMessage  小雨 19°C
ToolMessage  晴 15°C
AIMessage    [('calculate', {'expression': '19 - 15'})]
ToolMessage  4
AIMessage    现在上海 19°C（小雨），北京 15°C（晴），两地温差 **4°C**，上海更暖一些。
LLM 调用次数: 3

# agent-60-lab/w4/lgdemo/triage_all.py
"""练习 4:把 cloudledger.db 里的 41 张工单全部过一遍 triage_graph,按类别和主题统计。"""
import sqlite3
import sys
from collections import Counter, defaultdict
from pathlib import Path

import importlib

MODULE = next((a for a in sys.argv[1:] if not a.startswith("-")), "triage_graph")
graph = importlib.import_module(MODULE).graph

DB = Path(__file__).resolve().parent.parent / "cloudledger" / "cloudledger.db"

with sqlite3.connect(DB) as conn:
    rows = conn.execute("SELECT id, subject, body FROM tickets ORDER BY id").fetchall()

inputs = [{"ticket": f"{subject}\n{body}"} for _, subject, body in rows]
outs = graph.batch(inputs, config={"max_concurrency": 8})  # 8 张一起跑,41 张十几秒

by_cat = Counter(o["triage"].category for o in outs)
print("类别统计:", dict(by_cat))

by_subject: dict[str, Counter] = defaultdict(Counter)
for (tid, subject, _), o in zip(rows, outs):
    by_subject[subject][o["triage"].category] += 1
print(f"\n{'主题':<14}{'张数':>4}  分到的类别")
for subject, cats in sorted(by_subject.items(), key=lambda kv: -sum(kv[1].values())):
    print(f"{subject:<14}{sum(cats.values()):>4}  {dict(cats)}")

if "-v" in sys.argv:  # 加 -v 把每张单子的结果都打出来,方便挑错
    for (tid, subject, body), o in zip(rows, outs):
        print(tid, subject, "→", o["triage"].category, o["triage"].urgency, "|", body[:30])

类别统计: {'technical': 6, 'billing': 26, 'account': 9}

主题              张数  分到的类别
登录不了             9  {'account': 9}
重复扣款             8  {'billing': 8}
导出报表报错           6  {'technical': 6}
申请退款             5  {'billing': 5}
发票抬头修改           5  {'billing': 5}
升级套餐             4  {'billing': 4}
取消订阅             3  {'billing': 3}
席位数量不对           1  {'billing': 1}

# agent-60-lab/w4/lgdemo/triage_edge.py
"""练习 4 的第二步:库里的 41 张都太好分,自己写 8 张「两边都沾」的难工单,每张跑 3 次看稳不稳。"""
import importlib
import sys
from collections import Counter

MODULE = sys.argv[1] if len(sys.argv) > 1 else "triage_graph"
graph = importlib.import_module(MODULE).graph

# (工单, 我认为正确的类别)。规则:只要客户要的是钱上的处理,就算 billing
EDGE = [
    ("员工离职了,他的账号怎么删?删了以后席位费会少收吗", "billing"),
    ("想把公司账户注销掉,剩下的年费能退吗", "billing"),
    ("导出发票 PDF 的时候一直转圈,导不出来", "technical"),
    ("升级企业版的钱扣了,但功能还是基础版的", "billing"),
    ("开发票的邮件一直收不到,查了垃圾箱也没有", "billing"),
    ("你们的 API 文档在哪里看", "other"),
    ("管理员权限被别人改掉了,现在我进不了设置页", "account"),
    ("登录页一直报 500 错误,同事也一样", "technical"),
]
RUNS = 3

inputs = [{"ticket": t} for t, _ in EDGE for _ in range(RUNS)]
outs = graph.batch(inputs, config={"max_concurrency": 8})

right = 0
for i, (ticket, want) in enumerate(EDGE):
    got = Counter(o["triage"].category for o in outs[i * RUNS:(i + 1) * RUNS])
    right += got[want]
    mark = "✓" if got[want] == RUNS else "✗"
    print(f"{mark} {ticket[:22]:<24} 期望 {want:<9} 实际 {dict(got)}")
print(f"{MODULE}: {right}/{len(EDGE) * RUNS} 次分对")

✗ 员工离职了,他的账号怎么删?删了以后席位费会   期望 billing   实际 {'account': 3}
✗ 想把公司账户注销掉,剩下的年费能退吗       期望 billing   实际 {'account': 3}
✓ 导出发票 PDF 的时候一直转圈,导不出来    期望 technical 实际 {'technical': 3}
✓ 升级企业版的钱扣了,但功能还是基础版的      期望 billing   实际 {'billing': 3}
✓ 开发票的邮件一直收不到,查了垃圾箱也没有     期望 billing   实际 {'billing': 3}
✓ 你们的 API 文档在哪里看           期望 other     实际 {'other': 3}
✓ 管理员权限被别人改掉了,现在我进不了设置页    期望 account   实际 {'account': 3}
✓ 登录页一直报 500 错误,同事也一样      期望 technical 实际 {'technical': 3}
triage_graph: 18/24 次分对

        "billing=扣款/退款/发票/价格;technical=报错/功能异常;account=登录/权限/注销;其余 other。\n"
        "两类都沾时按客户最终要的结果分:要退钱、少收钱、核对费用的一律 billing,哪怕同时提到注销或删账号;"
        "登录页、导出等功能报错码算 technical,哪怕发生在登录或发票页面。\n\n"

类别统计: {'technical': 15, 'billing': 26}
登录不了             9  {'technical': 9}

        "两类都沾时按客户最终要的结果分:要退钱、少收钱、核对费用的一律 billing,哪怕同时提到注销或删账号;"
        "出现具体报错码(如 500、ERR_ 开头)或页面卡住转圈的算 technical;"
        "验证码收不到或过期、密码、权限这类登录问题仍是 account。\n\n"

cd agent-60-lab/w4/lgdemo
uv run python triage_all.py                  # 原版,41 张
uv run python triage_edge.py                 # 原版,难工单
uv run python triage_edge.py triage_graph_v3
uv run python triage_all.py triage_graph_v3

# agent-60-lab/w4/lgdemo/triage_refund.py
"""练习 5:billing 分支下加一个子步骤,工单里提到「退」就在 note 里标出需要审批。"""
from langgraph.graph import END, START, StateGraph

from triage_graph import (TicketState, classify, escalate_if_urgent, pick_branch,
                          to_billing, to_human, to_technical)


def refund_check(state: TicketState) -> dict:
    if "退" in state.get("ticket", ""):
        return {"note": state.get("note", "") + ";可能涉及退款,需审批"}
    return {}


builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("billing", to_billing)
builder.add_node("refund_check", refund_check)
builder.add_node("technical", to_technical)
builder.add_node("human", to_human)
builder.add_node("escalate", escalate_if_urgent)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", pick_branch, ["billing", "technical", "human"])
builder.add_edge("billing", "refund_check")  # billing 先定负责人,再做退款检查
builder.add_edge("refund_check", "escalate")
builder.add_edge("technical", "escalate")
builder.add_edge("human", "escalate")
builder.add_edge("escalate", END)
graph = builder.compile()

if __name__ == "__main__":
    for text in ["10 月 8 号被扣了两次款,麻烦退一笔", "上个月的发票抬头写错了,能重开吗"]:
        out = graph.invoke({"ticket": text})
        print(f"{text[:16]:<18} → {out['triage'].category:<8} {out['owner']} | {out['note']}")

10 月 8 号被扣了两次款,麻   → billing  账务组 | 需要查发票和订阅,W4 后面会接 MCP 工具;可能涉及退款,需审批;紧急,30 分钟内响应
上个月的发票抬头写错了,能重开吗   → billing  账务组 | 需要查发票和订阅,W4 后面会接 MCP 工具

验收标准:两张图都能跑通;第 2 题能用一句话说清 reducer 的作用;第 4 题有一张统计表和至少一处你修改分类定义后的前后对比;能不看代码画出两张图的结构。

名词解释

常见误区

消息字段忘了加 reducer。 默认是覆盖,不加 add_messages 每个节点返回的消息会把之前的全部替换掉,模型就「失忆」了。

节点里直接修改 state。 state["messages"].append(x) 这种写法会绕过 reducer,在有 checkpoint 时还会导致状态和记录不一致。节点应该只返回要更新的字段,让 LangGraph 去合并。

照抄老教程的 create_react_agent。 1.0 起它被弃用,替代品是 langchain.agents.create_agent。新代码要么手写节点,要么用新 API。

以为用了 LangGraph 就是 Agent。 图里没有「模型决定下一步」的环节,它就是一个 workflow。今天的 triage_graph 就是 workflow,agent_graph 才是 Agent。两者都是正当的设计,别为了显得高级把 workflow 硬改成 Agent。

用自由文本做路由。 让模型输出「我觉得是账务类」再用字符串匹配,迟早会遇到它说「财务相关」导致匹配失败。路由用结构化输出加枚举。

国产模型上用默认的 with_structured_output。 默认方式依赖厂商对 json_schema 的支持,不支持时会报错或静默降级。兼容厂商显式写 method="function_calling"。

学习资料

文档与文章

• LangGraph 官方 Overview 和 Quickstart:今天的起点。Quickstart 用计算器 Agent 演示了和本文几乎一样的结构,30 分钟。
• Graph API 概念:State、reducer、节点、边、Command 的权威说明,reducer 那一节一定要读,40 分钟。
• Workflows and agents:把 Anthropic《Building effective agents》里的五种模式全部用 LangGraph 实现了一遍,和你 Day 13 手写的版本对照着看,1 小时。
• LangGraph v1 迁移说明:列出了 1.0 弃用的 API,看老教程之前先扫一眼。
• LangChain Academy · Introduction to LangGraph:官方免费课,模块 1 对应今天的内容,有 notebook 可以跟着跑。

B 站视频

这套课我按目录过了一遍(2026-10-22 查,B 站播放 83 万、收藏 1.4 万),值得看,但别从头刷到尾。它的长处是把 LangGraph 的 API 一个个拆开讲透:state 和 reducer 讲了十几集,Send 动态扇出、Command、重试/超时/节点缓存、Postgres checkpoint 的表结构、fork 回放、并行中断时的检查点、子图的持久化策略,这些很多教程一笔带过的地方它都单独录了一集。短处也明显:标题写「项目实战」,其实没有一个完整项目,部署只有 3 集、设计模式 5 集几分钟带过;多 Agent(supervisor、handoff)、MCP 接入、评测都没讲,这些还是看本课 Day 26、Day 23 和 W5。环境用的是 miniconda + Jupyter,和本课的 uv 不一样,第 1–7 集直接跳过。

其他经典视频

• freeCodeCamp《LangGraph Complete Course for Beginners – Complex AI Agents with Python》(YouTube,3 小时):想系统补的话看,按标题搜索。

书籍

• Mayo Oshin、Nuno Campos《Learning LangChain》(O’Reilly, 2025):第 5 章「Cognitive Architectures with LangGraph」讲 LLM 调用、链、路由三种架构,第 6 章「Agent Architecture」讲 Agent 循环。作者之一是 LangChain 的创始工程师,讲法比官方文档更有条理。书写于 1.0 之前,API 细节以文档为准。
• 黄佳《大模型应用开发 动手做AI Agent》(人民邮电出版社·异步图书, 2024):讲 ReAct 和 Agent 架构的章节可以作为中文对照,帮助理解「图」到底在表达什么。

开源项目与代码

• langchain-ai/langgraph:源码和 examples/,看 libs/langgraph/langgraph/graph/message.py 里 add_messages 函数的实现,核心逻辑就几十行,读完就彻底明白 reducer 了。
• langchain-ai/langchain-academy:Academy 课程的全部 notebook。
• datawhalechina/hello-agents:中文 Agent 教程,里面有用框架和不用框架的对照章节。

自测

1. 下面这个 State,节点返回 {"messages": [ai_msg], "llm_calls": 3} 之后,两个字段分别怎么变?

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    llm_calls: int

2. 把 Day 9 的 while 循环翻译成 LangGraph,循环本身体现在图的哪里?「步数上限」可以放在哪几个地方?

3. 工单分类节点为什么要用 with_structured_output(Triage, method="function_calling"),而不是让模型直接回答类别名?

4. triage_graph 是 workflow 还是 Agent?判断依据是什么?

5. 同事说:「LangGraph 太重了,我们的客服流程直接手写就行。」你怎么判断他说得对不对?

6. 面试题:「LangChain 和 LangGraph 有什么区别?你为什么选 LangGraph?」

明天预告

Day 25 · LangGraph 进阶:checkpoint、人工审批、流式与中断恢复。今天的图跑完状态就丢了。明天给它接上 SQLite checkpointer,让状态落盘;用 interrupt 实现 CloudLedger 的退款审批,Agent 起草方案后停下来等主管,主管可以隔几个小时再批;还会演示进程退出后用同一个 thread_id 接着跑,以及「时间旅行」回到审批前重来一次。