（十一）从业务判断推导条件边

1. 写出各节点读取和返回的字段

先把已经会调用的功能接成三个普通函数。Runtime[RunContext] 从 langgraph.runtime 导入，用来接收前面定义的运行依赖。state 是任务数据，runtime.context 是执行这次操作时可使用的对象。

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
def retrieve(state: RunState, runtime: Runtime[RunContext]) -> dict:
    # 身份从上下文取得；节点只返回自己更新的两个字段。
    return {"hits": runtime.context.index.search(state["question"], runtime.context.owner), "status": "retrieved"}

retrieve 读取问题和当前身份，返回新的 hits 与 status。它没有返回 question，图仍保留旧问题；也没有改写原字典，返回的局部结果会由图按字段合入。

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
def draft(state: RunState, runtime: Runtime[RunContext]) -> dict:
    # 使用当前状态中的同一份证据，解析后的对象再转回可保存的字典。
    answer = make_answer(state["question"], state["hits"], runtime.context.model)
    return {"answer": answer.model_dump(), "status": "drafted"}

draft 读取刚才的 hits，返回结构化回答。model_dump() 把 Answer 转回可保存的字典。check 再把字典还原成 Answer，复用字段与引用检查：

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
def check(state: RunState) -> dict:
    # 先核对引用，后面的确认和保存只接收通过检查的结果。
    verify_answer(Answer.model_validate(state["answer"]), state["hits"])
    return {"status": "answered" if state["hits"] else "no_evidence"}

节点运行后	新增或替换的字段	继续保留的字段
retrieve	hits 包含两个块；status 为 retrieved	问题、35 分钟、operation_id
draft	answer 包含回答与引用；status 为 drafted	原问题和同一份 hits
check	status 为 answered；材料为空时为 no_evidence	已核对的 answer 与其材料

2. 从保存请求推导去向

任务要求回答后“按需保存计划”。先列情况：没有材料就结束；有材料但只问问题，也结束；有材料且要求保存，进入确认。确认拒绝则结束，批准才进入保存。

项目把这些判断写成条件边，模型生成的自然语言不会改变这张路线：

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
def route_after_check(state: RunState) -> str:
    return "approve" if state["hits"] and state["write_plan"] else END

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
def build_graph(checkpointer):
    builder = StateGraph(RunState, context_schema=RunContext)
    for name, node in (("retrieve", retrieve), ("draft", draft), ("check", check), ("approve", approve), ("save", save)):
        builder.add_node(name, node)
    builder.add_edge(START, "retrieve")
    builder.add_edge("retrieve", "draft")
    builder.add_edge("draft", "check")
    builder.add_conditional_edges("check", route_after_check, {"approve": "approve", END: END})
    builder.add_conditional_edges("approve", lambda state: "save" if state["approved"] else END,
                                  {"save": "save", END: END})
    builder.add_edge("save", END)
    return builder.compile(checkpointer=checkpointer)

add_conditional_edges 先调用判断函数，再按返回的路径名选择目标。第二处判断读取 approved，只有 True 才到 save。END 用作结束标记，并通过路径映射列出可能去向。

沿retrieve、draft和check推进，条件边决定结束或进入确认，批准后才能到save

逐条手算时，可以只看三个量：hits、write_plan、approved。例如 hits 非空、write_plan=False，经过 check 后结束，approve 和 save 都不会执行。把每条路线各写一个测试，比只跑通“批准并保存”更能说明图是否符合需求。