FDE W4D2 学习手册 · LangGraph

W4 Week1 Day2 · A 级（必须掌握，面试核心）· 4h · 学完能讲清 LangGraph 如何用 State/Node/Edge/Conditional Edge 组成图，以及 Checkpointer/Interrupt 在 HITL 中的作用

一、为什么需要 LangGraph（图编排动机）

1.1 朴素 Loop 的局限

Day1 的手写 Agent Loop 在"玩具场景"够用，但生产里会遇到：循环状态怎么持久化（进程崩了怎么办）？中途要"等人审批"怎么暂停又恢复？多分支怎么表达？这些手写很痛苦。

1.2 LangGraph 的解法

LangGraph 把 Agent 抽象成一张有向图（StateGraph）：

• Node（节点）：一个计算步骤（调模型 / 调工具 / 校验）。
• Edge（边）：节点间的流转，普通边是固定走向，条件边按 State 动态选分支。
• State（状态）：贯穿整张图的共享数据结构，每个节点读写它。
• Checkpointer / Interrupt：让图能"存盘、暂停、恢复"，支撑持久化与人工介入。

二、State（状态）

2.1 原理

State 是一份贯穿整张图的共享数据。每个节点执行后返回对 State 的更新，LangGraph 按Reducer（归约函数）合并到全局 State。最常见的是 messages 字段，用"追加（append）"作为 reducer——这就是 Day1 里那个不断增长的 messages 列表。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

class ClaimState(TypedDict):
    messages: Annotated[list, add_messages]   # 追加式 reducer
    policy_no: str
    risk_level: str
    need_human: bool

2.2 Reducer 的关键

• append（如 messages）：返回的新值追加到旧值后面（不直接覆盖）。
• replace（默认）：返回的字段覆盖旧值。

三、Node（节点）

3.1 定义

Node 是一个 Python 函数（或 Runnable），签名 (state) -> 部分更新。它读 State、做计算、返回要合并进 State 的字段。

def call_model(state: ClaimState):
    resp = model.invoke(state["messages"])
    return {"messages": [resp]}          # append 到 messages

def check_completeness(state: ClaimState):
    ok = bool(state.get("policy_no"))
    return {"need_human": not ok}        # replace 覆盖

3.2 节点类型

四、Edge（边）与 Conditional Edge（条件边）

4.1 普通 Edge

固定从一个节点走到另一个节点：graph.add_edge("node_a", "node_b")。对应"必然的下一步"。

4.2 Conditional Edge（条件边）

根据 State 动态选择下一个节点：add_conditional_edges("node_a", 路由函数, {"分支1":"node_b","分支2":"node_c"})。路由函数读 State，返回分支名。

def route_after_model(state):
    last = state["messages"][-1]
    if last.tool_calls:        # 模型要调工具
        return "tools"
    return "risk_check"        # 否则去风险校验

builder.add_conditional_edges("call_model", route_after_model,
    {"tools":"call_tools", "risk_check":"risk_check"})

4.3 为什么重要

条件边是"让模型参与控制流"的机制：模型输出决定走哪条边，这正是 Day1 说的"模型动态决策下一步"在图里的表达。普通边=确定性 Workflow，条件边=把决策点交给模型=Agentic Workflow。

五、StateGraph 如何组成一张图

5.1 拼装步骤

builder = StateGraph(ClaimState)
builder.add_node("call_model", call_model)
builder.add_node("call_tools", ToolNode(tools))
builder.add_node("risk_check", risk_check)
builder.add_node("human_approve", human_node)

builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", route_after_model,
    {"tools":"call_tools", "risk_check":"risk_check"})
builder.add_edge("call_tools", "call_model")        # 工具结果回填后再决策
builder.add_conditional_edges("risk_check", route_after_risk,
    {"human":"human_approve", "done":"END"})
builder.add_edge("human_approve", "risk_check")

graph = builder.compile(checkpointer=memory)        # 持久化

六、ToolNode（工具节点）

6.1 作用

ToolNode 是 LangGraph 内置节点，自动：读取 State 里最后一条 assistant 消息的 tool_calls → 逐个分发到对应工具函数 → 把结果包装成 tool

6.2 工具从哪来

工具通常是 LangChain 的 @tool 装饰函数，或 OpenAI 风格的 function schema。你的 Java 微服务可以包成一个 @tool 函数（内部 HTTP 调你的服务）。

from langchain_core.tools import tool

@tool
def ocr_scan(file_url: str) -> str:
    """识别理赔材料图片，返回结构化字段"""
    return requests.post("http://your-java-svc/ocr", json={"url":file_url}).text

tools = [ocr_scan, query_policy]
tool_node = ToolNode(tools)

七、Checkpointer（检查点 / 持久化）

7.1 原理

Checkpointer 是图的"存档器"：每执行一步，把当前 State 快照按 thread_id 存起来。这样：

• 崩溃恢复：进程挂了，用同一 thread_id 重跑可从断点继续，不重头来。
• 多轮对话：同一 thread_id 的多次 invoke 共享 State（不丢上下文）。
• 中断恢复：配合 Interrupt 实现"暂停等人、再继续"。

7.2 使用

graph = builder.compile(checkpointer=MemorySaver())   # 或 Sqlite/Postgres
cfg = {"configurable": {"thread_id": "claim_12345"}}
graph.invoke({"messages":[("user","审核这张单")]}, cfg)  # 同 thread_id 共享状态

八、Interrupt（中断）与 Command（命令）

8.1 原理

interrupt() 在节点里调用，会让图暂停执行并把控制权交回调用方，当前 State 已被 Checkpointer 存下。等人工/外部输入后，用 Command(resume=...) 带着输入"唤醒"图，从断点继续。

def human_approve(state):
    decision = interrupt({"question":"是否批准该特药理赔？", "risk":state["risk_level"]})
    # 图在此暂停；外部调用 graph.invoke(Command(resume="approved"), cfg) 后继续
    return {"approved": decision == "approved"}

graph.invoke(Command(resume=user_input), cfg)   # 带输入恢复

8.2 与 Checkpointer 的关系

Interrupt 负责"暂停"，Checkpointer 负责"把暂停前的状态存好"，二者配合才能实现 HITL。没有 Checkpointer，Interrupt 一暂停状态就没了，恢复不了。

九、HITL（人在回路）的工程含义

十、实战：构建特药理赔 Agent 的 LangGraph

把 Day1 的 Agent Loop 升级为图，含工具调用、条件分支、人工审批中断。

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(ClaimState)
builder.add_node("call_model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_node("risk_check", risk_check)
builder.add_node("human_approve", human_approve)

builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", tools_condition)  # 内置：有tool_calls→tools
builder.add_edge("tools", "call_model")
builder.add_conditional_edges("risk_check", route_after_risk,
    {"human":"human_approve", "done":END})
builder.add_edge("human_approve", "risk_check")

graph = builder.compile(checkpointer=MemorySaver())

# 运行
cfg = {"configurable": {"thread_id": "claim_8848"}}
graph.invoke({"messages":[("user","审核 /tmp/claim.png，保单 P123")]}, cfg)
# 若中断在 human_approve，前端拿审核员输入后：
# graph.invoke(Command(resume="approved"), cfg)

十一、LangGraph 高频易错点

十二、学习资源（中文 / 非 OpenAI）

十三、面试达标线①：State / Node / Edge / Conditional Edge 如何组成图

十四、面试达标线②：Checkpointer 与 Interrupt 在 HITL 中的作用

十五、W4D2 自测清单

十六、高频面试题速记卡