第 2 期：第一张图——状态、节点、边

上册练习 3 的结论是一句话：多轮对话的全部机制，就是一个数组加一个 for 循环。数组装着到目前为止发生过的一切，循环决定下一步做什么。

LangGraph 把这两样拆开，各给了一个名字。数组叫状态，循环里的每一步叫节点，步骤之间的先后叫边。这一期用第 1 期的第二个场景，商品咨询答案草稿，搭出第一张图。三个节点串成一条线，没有工具，没有分支，没有循环。那些留给第 3 期。

敲进去

第 2 期的代码在 code/ep02/，五个文件，各管一件事：

ep02/
  state.py       状态：所有节点共享的那份数据长什么样
  prompts.py     两段提示词
  graph.py       节点函数 + 把节点连成图
  main.py        命令行入口
  data/policies.json   两个商品的改期、退款、使用政策

另外在 common/llm.py 加了一个函数，按第 1 期的三个环境变量造模型客户端，后面每一期都用它。

先定义状态。state.py：

import operator
from typing import Annotated, TypedDict


class DraftState(TypedDict):
    # 输入
    product_id: str
    question: str
    # 中间结果：每个字段由某一个节点写入，后写的覆盖先写的
    category: str
    policy: str
    draft: str
    # 带 reducer 的字段：每个节点往里追加一条，不覆盖
    trace: Annotated[list[str], operator.add]

状态就是一个带类型的字典。六个字段里五个是普通字段，谁写谁覆盖。最后一个 trace 不一样：它用 Annotated 挂了一个 operator.add，意思是“新值追加到旧值后面“。挂在字段上的这个合并规则叫 reducer。练习 3 里你写的 messages = append(messages, ...) 就是手写的 reducer，只是当时没人给它起名字。

三个节点，每个是一个普通函数。graph.py：

import json
from pathlib import Path

from langgraph.graph import END, START, StateGraph

from common.llm import chat_model
from ep02.prompts import CATEGORIES, CLASSIFY, DRAFT
from ep02.state import DraftState

POLICIES = json.loads((Path(__file__).parent / "data" / "policies.json").read_text())
llm = chat_model()


def classify(state: DraftState) -> dict:
    """节点一：模型判断问题类别。只返回自己改动的字段。"""
    reply = llm.invoke(CLASSIFY.format(question=state["question"]))
    word = reply.content.strip().lower()
    category = word if word in CATEGORIES else "usage"
    return {"category": category, "trace": [f"classify -> {category}"]}


def lookup(state: DraftState) -> dict:
    """节点二：纯 Python 查政策，没有模型。"""
    product = POLICIES[state["product_id"]]
    policy = product[state["category"]]
    return {"policy": policy, "trace": [f"lookup -> {product['name']}"]}


def draft(state: DraftState) -> dict:
    """节点三：模型照着政策原文起草回复。"""
    name = POLICIES[state["product_id"]]["name"]
    reply = llm.invoke(
        DRAFT.format(name=name, policy=state["policy"], question=state["question"])
    )
    return {"draft": reply.content.strip(), "trace": ["draft -> done"]}

看节点的签名：进来一份完整状态，出去一个字典，字典里只有这个节点改动的字段。 classify 只写 category，lookup 只写 policy，谁也不碰别人的字段。三个节点都往 trace 里追加了一条，因为它有 reducer，追加不会互相覆盖。

lookup 里没有模型，就是查一个字典。节点是普通函数，里面有没有模型无所谓。

把节点连起来，还在 graph.py：

def build_graph():
    builder = StateGraph(DraftState)
    builder.add_node("classify", classify)
    builder.add_node("lookup", lookup)
    builder.add_node("draft", draft)
    builder.add_edge(START, "classify")
    builder.add_edge("classify", "lookup")
    builder.add_edge("lookup", "draft")
    builder.add_edge("draft", END)
    return builder.compile()


graph = build_graph()

四条边，从 START 到 END 一条线。compile() 之后才是一张能跑的图。

main.py 提供三种跑法，核心就两段：

# 一次跑完拿最终状态。version="v2" 返回的是结构化结果，最终状态在 .value 里
result = graph.invoke(state, version="v2")
print("trace:", " | ".join(result.value["trace"]))
print("草稿:", result.value["draft"])

# stream_mode="updates"：每个节点跑完，吐出它返回的那一小块更新
for update in graph.stream(state, stream_mode="updates"):
    for node, changed in update.items():
        print(f"[{node}] 写入 {', '.join(changed)}")

两段提示词在 prompts.py，分类那段让模型只输出类别单词，起草那段限定只根据政策原文回答。完整文件在仓库里。

跑起来

cd code
uv run python -m ep02.main --show-graph
uv run python -m ep02.main SKU-1001 "我想把日期改到下周六可以吗"
uv run python -m ep02.main --steps SKU-1001 "我想把日期改到下周六可以吗"

你应该看到什么

第一条命令不调模型，把图画出来：

graph TD;
	__start__([<p>__start__</p>]):::first
	classify(classify)
	lookup(lookup)
	draft(draft)
	__end__([<p>__end__</p>]):::last
	__start__ --> classify;
	classify --> lookup;
	lookup --> draft;
	draft --> __end__;

这是 Mermaid 格式，渲染出来是这样：

graph TD;
	__start__([<p>__start__</p>]):::first
	classify(classify)
	lookup(lookup)
	draft(draft)
	__end__([<p>__end__</p>]):::last
	__start__ --> classify;
	classify --> lookup;
	lookup --> draft;
	draft --> __end__;
	classDef default fill:#f2f0ff,line-height:1.2
	classDef first fill-opacity:0
	classDef last fill:#bfb6fc

图是从代码里生成的，代码改了图跟着变，这一点第 10 期图变复杂之后会很有用。

第二条命令跑完整流程：

trace: classify -> reschedule | lookup -> 东京迪士尼一日票 | draft -> done
草稿: 您好，若您的出行日期距今天还有3天以上，可以免费改期到下周六一次；改期后不可再改。若已不足3天，则不支持改期。请确认您的出行日期。

trace 里三条记录，三个节点各追加了一条，顺序就是边的顺序。草稿只说了政策里有的条款，还多问了一句出行日期，因为政策以出行日为界，模型判断得对。

第三条命令看每一步：

[classify] 写入 category, trace
[lookup] 写入 policy, trace
[draft] 写入 draft, trace

每个节点跑完，运行时吐出它返回的那个字典里有哪些键。三个节点各写自己的字段， trace 每次都在。

发生了什么

状态是练习 3 那个数组的一般化。 练习 3 里所有信息都塞在 messages 数组里，这里拆成了六个有名字的字段。节点看到的是整份状态，改的只是其中几个字段。运行时在每个节点跑完之后负责合并：普通字段直接覆盖，带 reducer 的字段按 reducer 合并。你在练习 3 里手写的那行 append，现在是字段声明上的一个标注。

边是 for 循环里写死的顺序。 练习 3 的循环体里，先做什么后做什么是代码定的。这里也一样，四条边把顺序钉死，模型改不了。这就是第 1 期说的第二档：预定义流程。模型出现在两个节点里做分类和起草，路线图是你画的。lookup 没有模型，照样是一个节点。

节点只返回改动。 这个约定让每个节点只需要知道自己的事。classify 不知道后面有没有 lookup，draft 不知道政策是谁查的。第 10 期把图拆成子图的时候，这个约定是能拆开的前提。

出错会告诉你在哪个节点。 传一个不存在的商品编号进去，lookup 会抛 KeyError，报错末尾多一行：

KeyError: 'SKU-9999'
During task with name 'lookup' and id '7c2b86e9-...'

第 1 期说预定义流程的好处是出错能定位到某一行，这里运行时先帮你定位到某个节点。

两种跑法拿的是同一份状态。 invoke 一次跑完，version="v2" 让它返回一个结构化的结果对象，最终状态在 .value 里。stream 按节点吐更新，stream_mode="updates" 每次给你一个节点返回的那个字典。第 12 期把图包进 FastAPI 的时候，流式输出走的就是后一条路。

常见问题

三步的事，为什么要画成图？ 这一期确实啰嗦，三个函数顺序调用就能干同样的事。图的价值从第 4 期开始显现：在某个节点停下、从某个节点续跑、看某个节点之前的状态，这些都以节点为单位。没有节点这个边界，那些能力无处挂。

classify 返回了清单外的词怎么办？ 代码里兜底成 usage。三次真机运行模型都老老实实只输出了类别单词，但你不能指望这一点，兜底那一行是必须的。加分练习让你故意把它弄坏看看。

为什么 trace 要挂 reducer，category 不要？ category 一个流程里只有一个节点写它，覆盖就是想要的行为。trace 三个节点都写，要的是累加。哪个字段挂 reducer，取决于它是“一个人的答案“还是“大家的记录“。第 3 期的 messages 字段属于后一种。

每次 invoke 都从头开始吗？ 是。这一期的图没有记忆，跑完状态就丢了。第 4 期加 checkpointer 之后，同一个 thread_id 再进来会接着上次的状态。

加分练习

加第四个节点 review，纯 Python：检查草稿有没有超过 80 字，超了在 trace 里记一笔。三条边变四条。
把 trace 字段上的 Annotated[..., operator.add] 去掉，改成普通的 list[str]，再跑 --steps，看最终的 trace 里剩几条。
在 classify 里把返回值硬改成 "unknown"，跑一次，确认兜底那一行确实起了作用。
把 --show-graph 的输出贴到 mermaid.live 看看图长什么样，然后加完第 1 题的节点再贴一次。

Keyboard shortcuts

笨办法学 Agent · 用 LangGraph 上线

第 2 期：第一张图——状态、节点、边

敲进去

跑起来

你应该看到什么

发生了什么

常见问题

加分练习