（十）用状态图组织步骤

1. 节点返回字段更新

StateGraph 从 langgraph.graph 导入，用状态结构描述一次执行中共享的数据，节点接收状态并返回字段更新；边说明下一步执行哪里。START 和 END 分别表示图的入口和结束位置。

TypedDict 是带字段类型说明的字典注解，运行中仍是普通字典，不会像 Pydantic 那样自动检查外部输入。Annotated[T, 标记] 给类型补充元数据；这里 LangGraph 读取第二项 reducer，决定怎样合并新旧字段值。

"""同一次节点更新，普通字段替换，带 reducer 的字段累加。"""
from operator import add
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    status: str
    events: Annotated[list[str], add]


def retrieve(state: State):
    # 只返回变化的字段；events 的 add 会把本次列表接在旧列表后。
    return {"status": "retrieved", "events": ["找到 numpy.md:0"]}


def check(state: State):
    return {"status": "checked", "events": ["引用核对通过"]}


builder = StateGraph(State)
builder.add_node("retrieve", retrieve)
builder.add_node("check", check)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "check")
builder.add_edge("check", END)
graph = builder.compile()
initial = {"status": "new", "events": ["接收问题"]}
# updates 只显示本节点的更新；values 显示合并后的完整状态。
for update in graph.stream(initial, stream_mode="updates"):
    print(update)
final = graph.invoke(initial)
assert final == {"status": "checked", "events": ["接收问题", "找到 numpy.md:0", "引用核对通过"]}
print(final)

初始 events 有“接收问题”。retrieve 返回一个新列表，reducer 使用 operator.add 把它接到旧列表后面；check 再加一项，于是最终共有三条事件。status 没有 reducer，后一次节点返回的新字符串会替换旧值。

消息历史常用 add_messages reducer，它按消息标识合并或更新消息，适合聊天状态。普通列表使用 add 就是前后连接；选择 reducer 时先决定希望保存“最新值”“追加记录”还是“按标识更新”，再写类型说明。状态与 reducer

2. 列出业务状态与运行依赖

再把短例里的 status、events 换成完整阅读任务需要保存的字段。首先是 RunState：

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
class RunState(TypedDict, total=False):
    question: str
    write_plan: bool
    minutes: int
    operation_id: str  # 同一保存动作在重试时继续使用这个编号。
    hits: list[dict]  # 检索后写入文本、来源、块编号和得分。
    answer: dict  # 生成后保存 Pydantic 对象导出的字段。
    approved: bool
    plan_id: int  # 保存完成后才出现。
    status: str

total=False 表示字段可以随步骤出现。开始时只有问题、保存请求、时长和操作编号；检索后填 hits，回答后填 answer，保存后才有 plan_id。本项目的这些业务字段保存最新值，沿用默认替换规则；前面用来展示执行顺序的 events 才需要累加。

第五章已经把 SQLite 封装进存储器。本章的 PlanStore 也保留两个业务入口：save(owner, operation_id, question, minutes) 保存计划并返回 id，list(owner) 返回当前用户的记录；具体事务在第十四节展开。节点先按照这些已经明确的输入、输出调用它。

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
@dataclass
class RunContext:
    """运行依赖单独传入，不把连接和当前身份交给模型生成。"""
    owner: str
    index: NoteIndex
    plans: PlanStore
    model: Any = None
    crash_after_commit: bool = False

RunContext 使用第四章的数据类写法，owner、index、plans 分别保存当前身份、检索对象和计划存储器。model 可以放回放模型或本机模型，Any 表示这里接收不同适配器；它的注解不会自动检查模型行为。crash_after_commit 是演示失败恢复的测试开关，正常调用为 False。

3. 构建与执行分开

add_node 注册节点，add_edge 连接顺序，compile 得到可执行图。invoke 执行一次，stream 可逐步观察更新。节点通常只返回它修改的字段，其余字段由图继续保留；并发节点要写同一字段时，则需要明确能合并这些写入的规则。