LangGraph 01：图、State 与节点——从工作流原理到代码

2026-09-01

944 字约 4 分钟

...

1、本篇任务：把退款问答拆成显式工作流

LangChain Agent 的循环适合模型自主选择工具，但“先验证输入、再并行查订单和政策、证据不足则拒答、最后生成结果”更适合显式图。本课只学习四个概念：State、Node、Edge、Reducer。

START → validate → load_order ─┐
                   search_policy ├→ decide → answer / abstain → END

2、图的心智模型

State 是当前运行快照；Node 是读取 state 并返回局部更新的函数；Edge 决定下一个节点；Reducer 决定多个更新怎样合并。节点可以调用模型，也可以是普通代码。

图按 super-step 推进。同一 super-step 的节点可以并行，完成后各自更新 state；这也是并发写入字段必须定义 reducer 的原因。

3、先设计 State 的所有权

from operator import add
from typing import Annotated, Literal
from typing_extensions import TypedDict

class RefundState(TypedDict, total=False):
    question: str
    order_id: str
    order: dict
    evidence: Annotated[list[dict], add]
    decision: Literal["answer", "abstain"]
    answer: str
    errors: Annotated[list[str], add]

order 只由订单节点写；decision 只由判断节点写；answer 只由最终节点写。evidence/errors 可能由并行节点追加，因此使用 reducer。

没有 reducer 的字段默认覆盖。节点只返回自己修改的字段，不返回完整 state。把整个 state 原样复制回来会掩盖所有权问题并扩大 checkpoint。

4、Python：完成第一张图

from langgraph.graph import StateGraph, START, END

def validate(state: RefundState):
    if not state.get("order_id"):
        return {"errors": ["MISSING_ORDER_ID"]}
    return {}

def load_order(state: RefundState):
    return {"order": order_service.get(state["order_id"])}

def search_policy(state: RefundState):
    chunks = retriever.search(state["question"])
    return {"evidence": chunks}

def decide(state: RefundState):
    ok = bool(state.get("order")) and bool(state.get("evidence")) and not state.get("errors")
    return {"decision": "answer" if ok else "abstain"}

def answer(state: RefundState):
    return {"answer": answer_model(state["question"], state["order"], state["evidence"])}

def abstain(state: RefundState):
    return {"answer": "证据不足，暂时无法判断，请联系人工客服。"}

builder = StateGraph(RefundState)
builder.add_node("validate", validate)
builder.add_node("load_order", load_order)
builder.add_node("search_policy", search_policy)
builder.add_node("decide", decide)
builder.add_node("answer", answer)
builder.add_node("abstain", abstain)
builder.add_edge(START, "validate")
builder.add_edge("validate", "load_order")
builder.add_edge("validate", "search_policy")
builder.add_edge("load_order", "decide")
builder.add_edge("search_policy", "decide")
builder.add_conditional_edges("decide", lambda s: s["decision"], {"answer": "answer", "abstain": "abstain"})
builder.add_edge("answer", END)
builder.add_edge("abstain", END)
graph = builder.compile()

这里有一个值得修正的设计：validate 失败后仍会进入两个查询节点。下一课用条件路由在入口立即拒绝，展示 Edge 如何表达业务控制。

5、TypeScript 的等价 State

import { Annotation, START, END, StateGraph } from "@langchain/langgraph";

const State = Annotation.Root({
  question: Annotation<string>,
  orderId: Annotation<string>,
  order: Annotation<Record<string, unknown>>,
  evidence: Annotation<Record<string, unknown>[]>({
    reducer: (left, right) => left.concat(right),
    default: () => [],
  }),
  decision: Annotation<"answer" | "abstain">,
  answer: Annotation<string>,
  errors: Annotation<string[]>({ reducer: (l, r) => l.concat(r), default: () => [] }),
});