Day 02 / 共 60 天 · 阶段 1 入门与心智

装环境 + 跑通第一个图

昨天有了地图,今天上手。我们把环境装好,然后跑通一个官方源码里就存在的最小 StateGraph 例子——它就写在 state.py 的类文档里,是真代码。跑通并逐行拆解它,"节点 / 边 / 编译 / invoke"这四个词会从抽象名词变成你手里转起来的东西。

📍 阶段 1「入门与心智」(D01-06) · 你在第 2 天
D01 全景分包 D02 跑通第一个图 D03 StateGraph D04 节点与边 D05 State/TypedDict D06 对话图+小结
💡 一句话锚点 跑通一个 LangGraph 程序永远是同一套五步:定义 State(工件长啥样)→ 写节点函数(工位干啥)→ add_node/add_edge(连线)→ compile(把图纸变成机器)→ invoke(塞进工件开跑)。今天这个最小例子把五步全占齐,以后再复杂的图都是这五步的放大版。
L01

装环境:三条路,按需选

先确认 Python 版本达标——昨天在 pyproject 里看到硬要求 ≥3.10libs/langgraph/pyproject.toml:9 requires-python):

python3 --version   # 必须 >= 3.10,否则装不上
场景怎么装适合谁
只想用pip install -U langgraph写业务、跟着敲例子
想读源码 + 改着玩(本课)克隆仓库 + uv sync(见 L02)我们,要打断点、加 print
要连模型做对话额外装 langchain-openaiD06 之后
🤔 痛点:为什么本课要"从源码装",不直接 pip?因为我们要做的是逐行读源码 + 打断点观察。pip 装的是打包好的 wheel,藏在 site-packages 深处,改一行还怕污染环境。克隆仓库用 uv 装成"可编辑模式",源码就在你眼前的 libs/langgraph/,随便加 print 看它内部怎么跑——这才是学源码的姿势。
🍼 uv 是啥uv 是个超快的 Python 包/环境管理器(Rust 写的),本仓库用它管理多包 monorepo。你可以把它理解成"更快的 pip + venv 二合一"。没有它用 pip 也行,只是慢一点、命令啰嗦一点。
L02

从源码把 langgraph 装成可编辑

本课的仓库已经在 /Users/bitmart/work/codes/github/langgraph。核心包在 libs/langgraph/,装它:

# 进核心包目录
cd libs/langgraph

# 方式 A:用 uv(仓库推荐)——建虚拟环境并按 pyproject 装齐依赖
uv sync

# 方式 B:用 pip 装成可编辑模式(-e = editable,改源码立即生效)
python3 -m venv .venv && source .venv/bin/activate
pip install -e .

# 验证:能 import 且版本正确
python3 -c "import langgraph; from langgraph.graph import StateGraph; print('ok')"
uv syncpyproject.toml 的 dependencies(昨天那 6 行),把 langchain-core、checkpoint、prebuilt、pydantic 等一次装齐到一个隔离环境。
pip install -e .关键是 -e(editable):不是把包复制走,而是做个"软链接"指回源码目录。你在 state.py 里加一行 print,下次运行立刻生效——这正是读源码要的。
import 验证from langgraph.graph import StateGraph 就说明昨天 L06 那张"门牌"通了。报 ModuleNotFoundError 多半是没激活虚拟环境。
⚠️ 坑:monorepo 里装错目录仓库根 libs/ 下有 8 个包。如果你在仓库根直接 pip install -e . 会找不到 pyproject(根目录没有可安装包)。必须 cd libs/langgraph 再装。想连 SQLite 存档还要再进 libs/checkpoint-sqlite 装那个包——这正是昨天讲的"分包"带来的副作用:装哪个进哪个目录。
L03

最小图全貌:官方文档里的真代码

不用自己编例子——StateGraph 的类文档里就带了一个完整可跑的最小图。这是仓库源码,不是我杜撰的:

libs/langgraph/langgraph/graph/state.py:158-198(StateGraph 类 docstring 的 Example)
from typing_extensions import Annotated, TypedDict
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime


def reducer(a: list, b: int | None) -> list:
    if b is not None:
        return a + [b]
    return a


class State(TypedDict):
    x: Annotated[list, reducer]


class Context(TypedDict):
    r: float


graph = StateGraph(state_schema=State, context_schema=Context)


def node(state: State, runtime: Runtime[Context]) -> dict:
    r = runtime.context.get("r", 1.0)
    x = state["x"][-1]
    next_value = x * r * (1 - x)
    return {"x": next_value}


graph.add_node("A", node)
graph.set_entry_point("A")
graph.set_finish_point("A")
compiled = graph.compile()

step1 = compiled.invoke({"x": 0.5}, context={"r": 3.0})
# {'x': [0.5, 0.75]}
🍼 它在算什么?别被数学吓到。这个节点算的是"逻辑斯蒂映射"x·r·(1-x)——一个经典的迭代公式。你完全不用管公式含义,只需盯住:输入 x=0.5,节点算出 0.75,返回 {"x": 0.75},最后状态里 x 变成了 [0.5, 0.75]。为什么是列表、为什么两个值都在?答案在 reducer——下一讲拆。
这张最小图长这样(就一个工位 A) START 节点 Anode() END set_entry_point set_finish_point 工件 {"x":[0.5]} 从 START 进 → A 加工 → 从 END 出 {"x":[0.5,0.75]}
图注:最小图只有一个节点 A,但"入口→加工→出口"的完整骨架已经齐了。
L04

逐行走读(上):定义状态与节点

① 状态 State:告诉引擎"工件长啥样"

def reducer(a: list, b: int | None) -> list:      # 合并函数:老列表 a + 新值 b
    if b is not None:
        return a + [b]
    return a

class State(TypedDict):
    x: Annotated[list, reducer]                    # 字段 x,用 reducer 合并
class State(TypedDict)TypedDict(带类型的字典)描述状态。这里状态就一个字段 x。为什么用 TypedDict 而不是普通类?D05 深挖。
Annotated[list, reducer]关键!Annotated[类型, 元数据] 给字段贴了一张"合并规则"标签:x 是个 list,新值来了用 reducer 合并,而不是直接覆盖。
reducer(a, b)合并逻辑:把新值 b 追加到老列表 a 后面(a + [b])。所以每次节点返回一个 x,历史都被保留成列表——这就是为什么最后是 [0.5, 0.75]

这个 reducer 机制正是昨天 L03 说的"铁律③",也是官方在 StateGraph 文档里强调的(state.py:135-137):

    Each state key can optionally be annotated with a reducer function ...
    The signature of a reducer function is `(Value, Value) -> Value`.

② 节点函数:一个工位干的活

def node(state: State, runtime: Runtime[Context]) -> dict:
    r = runtime.context.get("r", 1.0)   # 从"运行时上下文"取参数 r
    x = state["x"][-1]                  # 取当前状态里 x 列表的最后一个值
    next_value = x * r * (1 - x)        # 算一步
    return {"x": next_value}            # 只返回改动的字段(Partial
def node(state, runtime)节点就是普通函数。第一个参数拿到当前状态;第二个 runtime 拿到运行时上下文(不可变配置,如这里的 r)。D59 专讲 Runtime。
state["x"][-1]读状态:x 是个 list,取最后一个元素当"当前值"。第一次进来 x=[0.5],所以取到 0.5。
return {"x": next_value}兑现昨天的铁律②:只返回改动的字段,不返回整个 state。返回 {"x":0.75},引擎再用 reducer 把 0.75 合并进历史列表。
💡 本质:节点是"纯函数",副作用交给引擎注意节点没有去改全局变量、没有 state["x"].append() 这种原地修改,它只是"读输入、返回更新"。真正把更新写回状态的动作由引擎统一做(用 reducer)。这种"节点只算、引擎负责落状态"的分工,是后面能做并行、能做持久化、能做时间旅行的根基。
L05

逐行走读(下):连线、编译、运行

③ 建图:把工位连起来

graph = StateGraph(state_schema=State, context_schema=Context)
graph.add_node("A", node)        # 加一个叫 "A" 的工位,干的活是 node
graph.set_entry_point("A")       # 图从 A 进(= add_edge(START, "A"))
graph.set_finish_point("A")      # A 干完就到 END(= add_edge("A", END))
StateGraph(state_schema=State, ...)造一个建造器。此刻它是空图,只知道"状态长这样"。这一步内部就在解析 State 的字段、建 channel(D03/D05 深入)。
add_node("A", node)登记一个节点:名字 "A",行为 node。名字后面连边、看日志、画图都靠它。
set_entry_point("A")它内部就是一句 add_edge(START, key)——不信看源码:
libs/langgraph/langgraph/graph/state.py:1066-1077 & 1103-1114
def set_entry_point(self, key: str) -> Self:
    """... Equivalent to calling `add_edge(START, key)`."""
    return self.add_edge(START, key)          # ← 入口就是从 START 连一条边

def set_finish_point(self, key: str) -> Self:
    """... Equivalent to calling `add_edge(key, END)`."""
    return self.add_edge(key, END)            # ← 出口就是连一条边到 END
结论set_entry_point("A")add_edge(START, "A") 完全等价,只是更好读。START/END 就是昨天说的两个特殊字符串,"入口/出口"本质就是"和这两个特殊点连边"。

④⑤ 编译 + 运行

compiled = graph.compile()                          # 图纸 → 可执行机器
step1 = compiled.invoke({"x": 0.5}, context={"r": 3.0})
# {'x': [0.5, 0.75]}
graph.compile()把"建造器"翻译成真正能跑的 CompiledStateGraph(内部其实是 Pregel 引擎,D19 起精讲)。没 compile 之前图不能跑——见下方坑。
invoke({"x": 0.5}, ...)塞进初始工件开跑。注意传的是 0.5(会被当成 x 的初值,reducer 把它变列表 [0.5])。context={"r":3.0} 就是节点里 runtime.context.get("r") 拿到的那个 3.0。
结果 [0.5, 0.75]验算:x=0.5,0.5·3·(1-0.5)=0.75,reducer 把 0.75 追加进历史 → [0.5, 0.75]。跟注释完全对上,说明你跑通了!
📝 亲手改一个值验证理解context={"r": 3.0} 改成 {"r": 2.0}:结果会变成 [0.5, 0.5](0.5·2·0.5=0.5)。改 r 就改了节点算出的值——这就是 context 运行时参数的作用:同一张图、不同参数、不同结果。
L06

五步生命周期:从图纸到运行

把刚才的过程抽象成一张控制流图,以后写任何图都照这个骨架:

控制流:一个 LangGraph 程序的五步生命周期 ① 定义 State工件长啥样 ② 写节点函数工位干啥 ③ add_node/edge连线成图 ④ compile()变成机器 ⑤ invoke()塞工件跑 ①②③ 都在操作"建造器 StateGraph"——此时图还不能跑 ④之后才是 CompiledStateGraph(能跑)
图注:前三步都在"搭建造器",compile 是分水岭——之后才有能 invoke 的机器。
🎨 设计取舍:为什么非要一步 compile(),不能边建边跑? 朴素实现:add_node 时就地生成可执行逻辑,随加随用。源码实现:先把节点/边攒在建造器里(self.nodes/self.edges 就是普通字典和集合),最后 compile()一次性做校验(有没有孤立节点、入口是否存在)、把边翻译成底层通道、绑定 checkpointer。好处:① 校验集中,一次报全所有结构错误;② 编译期能做优化和一致性检查;③ 一张图纸能 compile 出多个配置不同的机器(带/不带存档)。代价:多一次显式调用,新手容易忘(见下讲的坑)。这是"构建期与运行期分离"的经典权衡。
L07

新手最常撞的 4 个报错

报错/现象原因怎么修
'StateGraph' object has no attribute 'invoke'忘了 compile(),直接对建造器 invokeg = graph.compile()g.invoke()
Graph must have an entrypoint没连 STARTset_entry_pointadd_edge(START, "A")
结果把 x 覆盖成单值而非列表State 里没给 x 加 reducerAnnotated[list, reducer],否则默认覆盖写
装不上 / import 失败Python < 3.10 或没激活 venv升级 Python;source .venv/bin/activate

"没入口"这个报错不是随口报的,是 compile()validate() 真的在查(libs/langgraph/langgraph/graph/state.py:1116 validate):

    if START not in all_sources:
        raise ValueError(
            "Graph must have an entrypoint: add at least one edge "
            "from START to another node"
        )
⚠️ 最隐蔽的坑:漏了 reducer,静默"覆盖"而不报错 如果 State 写成 x: list(没 Annotated[..., reducer]),程序不会报错,但每次节点返回的 x 会直接覆盖老值——你以为在累积历史,实际只剩最后一个。这类"不报错但结果不对"的 bug 最难查。原理:没标注 reducer 的字段默认走 LastValue 通道(后来者覆盖),这是 D03/D05 会看到的 _get_channel 的兜底行为。记住:想累积就必须显式给 reducer。
L08

今日小结 + 动手 + 明日预告

🧠 今天你应该能回答

  • 本课为什么"从源码可编辑安装"而不是 pip?(能改源码、打断点观察内部)
  • 一个 LangGraph 程序的五步是哪五步?(定义 State → 写节点 → add_node/edge → compile → invoke)
  • 节点函数的型号?((state, runtime) -> 部分状态字典,只返回改动字段)
  • 为什么结果是 [0.5, 0.75] 而不是 0.75?(x 带了追加型 reducer,历史被保留成列表)
  • set_entry_point("A") 的等价写法?(add_edge(START, "A")
  • 为什么必须 compile?(构建期/运行期分离:集中校验、翻译成底层通道、可编译多份配置)
  • 漏写 reducer 会怎样?(不报错,但字段被覆盖,历史丢失——最隐蔽的坑)

✋ 10 分钟动手

# 1. 把 L03 的最小例子存成 first_graph.py,跑起来看是否得到 [0.5, 0.75]
python3 first_graph.py

# 2. 把 context 的 r 从 3.0 改成 2.0,观察结果变成 [0.5, 0.5]

# 3. 故意注释掉 set_entry_point,观察报的正是 "must have an entrypoint"

# 4. 把 State 的 x 改成不带 reducer 的 `x: list`,观察结果只剩最后一个值

# 5. 亲眼核对 set_entry_point 就是 add_edge(START, key)
sed -n '1066,1077p' libs/langgraph/langgraph/graph/state.py
💡 明日预告 · Day 03今天我们"会用"了 StateGraph,明天钻进它的源码:StateGraph.__init__ 到底初始化了哪些内部字典(self.nodes/edges/channels)、add_node/add_edge/compile 的真实函数签名和内部动作。今天的黑盒,明天变白盒。
← Day 01 · 项目全景与分包 Day 03 · StateGraph 基础 →