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.10(libs/langgraph/pyproject.toml:9 requires-python):
python3 --version # 必须 >= 3.10,否则装不上
| 场景 | 怎么装 | 适合谁 |
|---|---|---|
| 只想用 | pip install -U langgraph | 写业务、跟着敲例子 |
| 想读源码 + 改着玩(本课) | 克隆仓库 + uv sync(见 L02) | 我们,要打断点、加 print |
| 要连模型做对话 | 额外装 langchain-openai 等 | D06 之后 |
🤔 痛点:为什么本课要"从源码装",不直接 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 sync读 pyproject.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,但"入口→加工→出口"的完整骨架已经齐了。
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
五步生命周期:从图纸到运行
把刚才的过程抽象成一张控制流图,以后写任何图都照这个骨架:
图注:前三步都在"搭建造器",compile 是分水岭——之后才有能 invoke 的机器。
🎨 设计取舍:为什么非要一步
compile(),不能边建边跑?
朴素实现:add_node 时就地生成可执行逻辑,随加随用。源码实现:先把节点/边攒在建造器里(self.nodes/self.edges 就是普通字典和集合),最后 compile() 时一次性做校验(有没有孤立节点、入口是否存在)、把边翻译成底层通道、绑定 checkpointer。好处:① 校验集中,一次报全所有结构错误;② 编译期能做优化和一致性检查;③ 一张图纸能 compile 出多个配置不同的机器(带/不带存档)。代价:多一次显式调用,新手容易忘(见下讲的坑)。这是"构建期与运行期分离"的经典权衡。L07
新手最常撞的 4 个报错
| 报错/现象 | 原因 | 怎么修 |
|---|---|---|
'StateGraph' object has no attribute 'invoke' | 忘了 compile(),直接对建造器 invoke | 先 g = graph.compile() 再 g.invoke() |
Graph must have an entrypoint | 没连 START | 加 set_entry_point 或 add_edge(START, "A") |
| 结果把 x 覆盖成单值而非列表 | State 里没给 x 加 reducer | 用 Annotated[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 的真实函数签名和内部动作。今天的黑盒,明天变白盒。