当你的系统开始出现循环、分支、暂停等待、人工审批、失败恢复、多 Agent 协作时，传统的“线性链”就会越来越吃力。LangGraph 的意义就在于：把 LLM 应用从一次次调用，提升为可状态化、可恢复、可审计的工作流系统。

本章将系统讲解 LangGraph 的核心概念、状态图建模、节点与边、检查点、人工中断、多 Agent 协作模式，以及如何把前面章节里的 Agent、工具、Memory、Streaming 与 LangGraph 串起来。

线性链适合这样的流程：

A -> B -> C -> 结束

但真实项目往往长这样：

• 如果资料不足，则回到澄清节点；
• 如果风险过高，则进入人工审批；
• 如果工具失败，则重试或走备用路径；
• 如果任务未完成，则继续循环；
• 如果输出需要验证，则先进入校验节点。

这种结构天然更适合“图”，而不是单链。

State 是图运行时共享的数据结构。你可以把它理解为当前工作流的“上下文快照”。它通常包含：

• messages；
• 当前任务目标；
• 检索结果；
• 工具返回；
• 是否已完成；
• 是否需要审批；
• 当前阶段产物。

Node 是图中的处理单元。一个节点可以是：

• 一次模型调用；
• 一次工具调用；
• 一段纯 Python 逻辑；
• 一个分类器；
• 一个校验器；
• 一个人工确认点。

Edge 定义节点之间如何流转。它既可以是固定边，也可以是条件边。

例如：

• `need_tool=True` → 工具节点；
• `need_human=True` → 审批节点；
• `done=True` → END。

LangGraph 最重要的设计习惯之一，是 先定义状态，再设计节点。

from typing import TypedDict, NotRequired
 
class WorkflowState(TypedDict):
    question: str
    retrieved_docs: NotRequired[list]
    draft_answer: NotRequired[str]
    reviewed: NotRequired[bool]
    need_human: NotRequired[bool]
    final_answer: NotRequired[str]

这个状态已经能支撑一个简单的“检索 → 生成 → 审核 → 输出”流程。

from langgraph.graph import StateGraph, START, END
 
 
def retrieve_node(state: WorkflowState):
    return {"retrieved_docs": ["制度条款 A", "制度条款 B"]}
 
 
def answer_node(state: WorkflowState):
    context = "
".join(state.get("retrieved_docs", []))
    return {"draft_answer": f"基于资料：{context}，这是初步回答。"}
 
 
def finalize_node(state: WorkflowState):
    return {"final_answer": state.get("draft_answer", "暂无回答")}
 
builder = StateGraph(WorkflowState)
builder.add_node("retrieve", retrieve_node)
builder.add_node("answer", answer_node)
builder.add_node("finalize", finalize_node)
 
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "answer")
builder.add_edge("answer", "finalize")
builder.add_edge("finalize", END)
 
graph = builder.compile()
 
result = graph.invoke({"question": "差旅报销多久到账？"})
print(result)

这个例子虽然简化，但已经展示了 LangGraph 的基本骨架：

• 状态；
• 节点；
• 边；
• 编译和执行。

实际业务中，最常见的需求不是固定流转，而是条件分支。

def route_after_answer(state: WorkflowState):
    if state.get("need_human"):
        return "human_review"
    return "finalize"

将其接入图：

def human_review_node(state: WorkflowState):
    return {"reviewed": False}
 
builder.add_node("human_review", human_review_node)
builder.add_conditional_edges(
    "answer",
    route_after_answer,
    {
        "human_review": "human_review",
        "finalize": "finalize",
    }
)

这样图就能根据状态决定流向。

LangGraph 的节点不一定只是纯 Python，也可以是：

• 一次 LLM 调用；
• 一次 RAG 链调用；
• 一个 Agent；
• 一个结构化输出分类器。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_template("""
请根据以下资料回答问题：
问题：{question}
资料：{context}
""")
 
answer_chain = prompt | model
 
 
def llm_answer_node(state: WorkflowState):
    context = "
".join(state.get("retrieved_docs", []))
    response = answer_chain.invoke({
        "question": state["question"],
        "context": context
    })
    return {"draft_answer": response.content}

这样你就可以把 LangChain 组件当作 LangGraph 节点内部实现。

LangGraph 的一个关键价值是支持检查点（checkpoint）。这意味着：

• 图可以暂停；
• 出错后可以恢复；
• 人工审批后可以继续；
• 长流程可以跨请求维持状态。

from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
 
config = {"configurable": {"thread_id": "workflow-001"}}
result = graph.invoke({"question": "请审核这份采购申请"}, config=config)

在真实项目中，你还可以换成持久化 checkpointer，把状态落到数据库或存储系统中。

很多业务并不允许模型自动执行最终动作，例如：

• 发正式邮件；
• 提交审批单；
• 修改主数据；
• 发起退款。

LangGraph 非常适合在这类环节插入“人工确认节点”：

• 图运行到某节点暂停；
• 将当前状态展示给人；
• 等待确认或修改；
• 再恢复执行。

class ApprovalState(TypedDict):
    draft_action: str
    need_human: bool
    approved: NotRequired[bool]
    approver_comment: NotRequired[str]

这种设计的价值是：

• 人工行为成为显式状态；
• 可以审计谁在何时批准了什么；
• 方便失败恢复和责任追踪。

LangGraph 不只是“单 Agent 加分支”，它也很适合多 Agent 协作：

• 规划 Agent：拆解问题；
• 检索 Agent：搜集资料；
• 分析 Agent：形成结论；
• 审校 Agent：检查风险；
• 执行 Agent：在批准后执行动作。

在图里，这些 Agent 可以对应不同节点，而不是把所有能力塞给一个超级 Agent。

下面给出一个更接近业务的状态图思路：

START
  -> 解析用户需求
  -> 检索合同条款与制度
  -> 生成初步审查意见
  -> 风险判断
      -> 若高风险 -> 人工审批
      -> 若低风险 -> 输出结果
END

对应状态可能包括：

• `question`
• `retrieved_clauses`
• `draft_review`
• `risk_level`
• `need_human`
• `final_review`

• 先定义状态，再定义节点；
• 节点职责单一，不要把所有逻辑堆进一个节点；
• 条件边只做路由判断，不做大段业务逻辑；
• 高风险动作单独做审批节点；
• 关键节点前后记录日志；
• 尽量让状态字段可追踪、可审计、可恢复。

以下场景不一定要使用 LangGraph：

• 单次摘要或改写；
• 只有 1~2 步的简单任务；
• 原型验证阶段；
• 团队尚未建立基础日志和状态管理能力。

一句经验是：需要循环、分支、恢复、审批时，再考虑图式编排。

• LangGraph 适合构建有状态、可恢复、可分支的复杂工作流；
• State、Node、Edge 是理解它的三大核心；
• 检查点和 thread_id 让图具备更强的生产能力；
• 它特别适合与 Agent、工具调用、人工审批、多阶段验证结合；
• 新一代复杂智能体系统越来越依赖图，而不是简单线性链。

1. 为“企业制度问答 + 人工审批”场景设计一个 LangGraph 状态图。

2. 用 `StateGraph` 写一个最小工作流：检索 → 生成 → 输出。

3. 给某个业务流程加上条件边：低风险直接通过，高风险进入审批。

4. 设计一个多 Agent 协作图，说明每个 Agent 的节点职责。

5. 说明你的项目在哪些节点需要 checkpoint，为什么。

• LangGraph 官方文档：Overview

下面给出一个更完整的示意图代码，它展示了“生成草稿 → 判断风险 → 高风险走审批、低风险直接结束”的结构：

from typing import TypedDict, NotRequired
from langgraph.graph import StateGraph, START, END
 
class ReviewState(TypedDict):
    question: str
    draft: NotRequired[str]
    risk_level: NotRequired[str]
    final_answer: NotRequired[str]
 
 
def draft_node(state: ReviewState):
    return {"draft": f"针对问题《{state['question']}》的初步意见。"}
 
 
def risk_node(state: ReviewState):
    risk_level = "high" if "合同" in state["question"] else "low"
    return {"risk_level": risk_level}
 
 
def route_risk(state: ReviewState):
    return "human_review" if state.get("risk_level") == "high" else "finalize"
 
 
def human_review_node(state: ReviewState):
    return {"final_answer": f"需人工复核：{state['draft']}"}
 
 
def finalize_node(state: ReviewState):
    return {"final_answer": state.get("draft", "暂无结果")}
 
builder = StateGraph(ReviewState)
builder.add_node("draft", draft_node)
builder.add_node("risk", risk_node)
builder.add_node("human_review", human_review_node)
builder.add_node("finalize", finalize_node)
 
builder.add_edge(START, "draft")
builder.add_edge("draft", "risk")
builder.add_conditional_edges("risk", route_risk, {
    "human_review": "human_review",
    "finalize": "finalize",
})
builder.add_edge("human_review", END)
builder.add_edge("finalize", END)
 
graph = builder.compile()
print(graph.invoke({"question": "请评估这份合同中的付款违约风险"}))

def agent_node(state):
    result = agent.invoke({
        "messages": [{"role": "user", "content": state["question"]}]
    })
    return {"agent_result": result}

这种方式很适合：

• 节点内部做局部自主决策；
• 图整体控制风险和边界；
• 多 Agent 系统中让不同 Agent 负责不同节点职责。