====== 第十二章：高级编排 - LangGraph ======

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

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

===== 12.1 为什么需要图式编排 =====

线性链适合这样的流程：

<code>
A -> B -> C -> 结束
</code>

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

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

===== 12.2 LangGraph 的核心概念 =====

==== 12.2.1 State ====

State 是图运行时共享的数据结构。你可以把它理解为当前工作流的“上下文快照”。它通常包含：
  * messages；
  * 当前任务目标；
  * 检索结果；
  * 工具返回；
  * 是否已完成；
  * 是否需要审批；
  * 当前阶段产物。

==== 12.2.2 Node ====

Node 是图中的处理单元。一个节点可以是：
  * 一次模型调用；
  * 一次工具调用；
  * 一段纯 Python 逻辑；
  * 一个分类器；
  * 一个校验器；
  * 一个人工确认点。

==== 12.2.3 Edge ====

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

例如：
  * `need_tool=True` -> 工具节点；
  * `need_human=True` -> 审批节点；
  * `done=True` -> END。

===== 12.3 定义状态结构 =====

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

==== 12.3.1 一个简单状态定义 ====

<code python>
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]
</code>

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

===== 12.4 构建一个最小 StateGraph =====

==== 12.4.1 基础示例 ====

<code python>
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)
</code>

这个例子虽然简化，但已经展示了 LangGraph 的基本骨架：
  * 状态；
  * 节点；
  * 边；
  * 编译和执行。

===== 12.5 条件边与分支控制 =====

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

==== 12.5.1 一个简单的条件路由 ====

<code python>
def route_after_answer(state: WorkflowState):
    if state.get("need_human"):
        return "human_review"
    return "finalize"
</code>

将其接入图：

<code python>
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",
    }
)
</code>

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

===== 12.6 在节点中调用模型或 Agent =====

LangGraph 的节点不一定只是纯 Python，也可以是：
  * 一次 LLM 调用；
  * 一次 RAG 链调用；
  * 一个 Agent；
  * 一个结构化输出分类器。

==== 12.6.1 在节点中调用模型 ====

<code python>
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}
</code>

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

===== 12.7 检查点与可恢复执行 =====

LangGraph 的一个关键价值是支持检查点（checkpoint）。这意味着：
  * 图可以暂停；
  * 出错后可以恢复；
  * 人工审批后可以继续；
  * 长流程可以跨请求维持状态。

==== 12.7.1 使用 InMemorySaver 作为检查点 ====

<code python>
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)
</code>

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

===== 12.8 Human-in-the-loop：人工中断节点 =====

很多业务并不允许模型自动执行最终动作，例如：
  * 发正式邮件；
  * 提交审批单；
  * 修改主数据；
  * 发起退款。

LangGraph 非常适合在这类环节插入“人工确认节点”：
  * 图运行到某节点暂停；
  * 将当前状态展示给人；
  * 等待确认或修改；
  * 再恢复执行。

==== 12.8.1 一个审批节点的状态设计 ====

<code python>
class ApprovalState(TypedDict):
    draft_action: str
    need_human: bool
    approved: NotRequired[bool]
    approver_comment: NotRequired[str]
</code>

这种设计的价值是：
  * 人工行为成为显式状态；
  * 可以审计谁在何时批准了什么；
  * 方便失败恢复和责任追踪。

===== 12.9 多 Agent 协作模式 =====

LangGraph 不只是“单 Agent 加分支”，它也很适合多 Agent 协作：
  * 规划 Agent：拆解问题；
  * 检索 Agent：搜集资料；
  * 分析 Agent：形成结论；
  * 审校 Agent：检查风险；
  * 执行 Agent：在批准后执行动作。

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

===== 12.10 一个合同审查工作流案例 =====

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

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

对应状态可能包括：
  * `question`
  * `retrieved_clauses`
  * `draft_review`
  * `risk_level`
  * `need_human`
  * `final_review`

===== 12.11 设计 LangGraph 的实践建议 =====

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

===== 12.12 什么时候不必上 LangGraph =====

以下场景不一定要使用 LangGraph：
  * 单次摘要或改写；
  * 只有 1~2 步的简单任务；
  * 原型验证阶段；
  * 团队尚未建立基础日志和状态管理能力。

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

===== 12.13 本章小结 =====

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

===== 练习 =====

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

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

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

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

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

===== 参考资源 =====

  * [[https://docs.langchain.com/oss/python/langgraph/overview|LangGraph 官方文档：Overview]]


===== 12.14 补充案例：带条件边的审查流程 =====

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

<code python>
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": "请评估这份合同中的付款违约风险"}))
</code>

===== 12.15 补充案例：把 Agent 作为图节点 ====

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

这种方式很适合：
  * 节点内部做局部自主决策；
  * 图整体控制风险和边界；
  * 多 Agent 系统中让不同 Agent 负责不同节点职责。