引言：当你发现AI工作流需要"循环"和"人工审核"时，LangChain的链式思维就开始捉襟见肘了。

一、那个让我朋友放弃Chain的项目

今年年初，我朋友接了一个他们公司内部AI审批系统的活儿。需求听起来不复杂：员工提交报销单，AI先初审一遍，判断票据是否合规、金额是否合理。如果AI觉得没问题，直接自动通过；如果AI拿不准，就转给财务人工复核。

我朋友一开始用LangChain的Chain来实现。第一步，把报销单传给模型做初审；第二步，根据模型的输出决定是"通过"还是"转人工"。代码大概长这样：

# 第一步：AI初审
review_result = review_chain.invoke({"expense": expense_data})

# 第二步：判断
if "通过" in review_result:
    auto_approve(expense_data)
else:
    send_to_human(review_result)

上线第一周，财务主管在群里@我朋友：“这个AI初审太严了，十张单子有八张被转人工，我们根本审不过来。能不能让AI先打标签，标出具体问题，我们再决定是打回修改还是直接通过？”

我朋友想了想，这需求不难，加一步就行：AI初审→打标签→人工判断→如果打回修改，员工改完再提交→AI再审→……

等等，这里出现了一个循环。员工改完重新提交，整个流程要再走一遍。而且如果第二次AI还是拿不准，还得再转人工。

他试着用Chain实现这个循环。Chain是什么？是流水线。原料从A工位流到B工位，再到C工位，一路向前，不能回头。要让Chain支持循环，他得在外面包一层while循环，手动维护状态，判断什么时候退出。代码很快变成了一团乱麻：

while True:
    result = chain.invoke(state)
    if result["status"] == "approved":
        break
    elif result["status"] == "rejected":
        state = get_user_fix(state)
    elif result["status"] == "human_review":
        state = wait_for_human(state)
    if iteration > 5:
        break

更头疼的是状态管理。每一轮循环，报销单的内容、AI的初审意见、人工的复核意见、员工的修改记录，这些信息散落在不同的变量里，没有统一的地方存放。一旦流程中断（比如服务重启），所有中间状态全丢了，用户得从头再来。

他折腾了三天，代码越来越复杂，bug越来越多。最后我不得不承认：Chain的线性思维，根本承载不了这种有循环、有分支、需要持久化状态的复杂工作流。

那用Agent呢？Agent不是能自主决策吗？

他试了一下LangChain的ReAct Agent。确实，Agent可以自己决定调用什么工具，看起来挺智能的。但问题是——Agent是个黑盒。你只知道它输入了一个问题，输出了一个答案，中间经历了多少轮思考、调用了哪些工具、为什么走这条路而不是那条路，你完全看不到，也控制不了。

财务主管提了一个很合理的要求："AI初审金额超过5000元的单子，必须人工复核，不能自动通过。“这个规则在Agent里怎么实现？Agent的决策逻辑藏在模型的"内心独白"里，你没法强行插入一条"硬规则”。要么在Prompt里反复强调，但模型可能听也可能不听；要么在Agent外层再包一层判断，又回到Chain的老路。

那一刻他明白了：Chain太死板，Agent太黑盒。我们需要一种既能精确控制流程、又能灵活处理分支和循环、还能持久化状态的编排方式。

这就是LangGraph诞生的原因。

二、从流水线到交通网：LangGraph到底是什么

如果把LangChain的Chain比作工厂里的流水线，那LangGraph就是城市里的交通网。

流水线的特点是：起点固定、终点固定、中间每个工位的顺序固定。原料从入口进去，按A→B→C→D的顺序依次加工，最后从出口出来。它适合简单、线性、步骤明确的任务，比如"翻译→润色→输出"。

但交通网不一样。交通网有十字路口、有立交桥、有单行道、有环岛。车辆从A点出发，可以根据路况选择走B路还是C路；如果B路堵了，可以掉头回A重新选；甚至可以在D点停下来等人，等完再继续走。

LangGraph的核心思想，就是把AI工作流建模成一个"有向图"：

• 节点（Node）：是路口，是执行具体任务的单元。比如"AI初审"是一个节点，"人工复核"是一个节点，"员工修改"也是一个节点。
• 边（Edge）：是路，是节点之间的连接。它定义了任务执行完以后，下一步该往哪儿走。
• 条件边（Conditional Edge）：是红绿灯，是根据实时情况动态决定方向的开关。比如"AI初审"节点执行完后，如果金额小于5000且票据合规，走"自动通过"这条路；否则走"人工复核"这条路。
• 状态（State）：是车里拉的货，是在整个交通网中流转的数据。它从起点装载，经过每个节点时被加工、被补充，最终带着完整的信息到达终点。

这种图结构的表达能力，远超线性链。理论上，图编排模型等价于图灵机，而链式编排模型只等价于有限状态自动机。换句话说，图能表达任何可计算的流程，而链不能。

三、LangGraph的四大核心概念

理解了定位，我们来看LangGraph的四大核心概念。掌握这四个概念，你就掌握了LangGraph的骨架。

概念一：StateGraph——状态图，整个交通网的蓝图

StateGraph是LangGraph中定义工作流的起点。它相当于一张空白图纸，你在上面画节点、连边，最终拼成一张完整的交通网。

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

# 第一步：定义状态结构
class ApprovalState(TypedDict):
    expense_data: dict      # 报销单原始数据
    ai_review: str          # AI初审意见
    risk_tags: list         # AI打的风险标签
    human_decision: str     # 人工复核结果
    iteration: int          # 循环次数，防死循环

# 第二步：创建状态图
workflow = StateGraph(ApprovalState)

ApprovalState这个TypedDict定义了整辆"货车"能拉什么货。所有节点共享同一个State，每个节点读取自己需要的字段，加工后写回新的字段。这种设计让数据流变得异常清晰——你知道每个阶段产生了什么数据，数据在节点之间怎么传递。

概念二：Node——节点，路口上的执行单元

节点是图中最基础的执行单元，本质上就是一个Python函数。它接收当前State，执行业务逻辑，返回一个字典来更新State。

def ai_review_node(state: ApprovalState) -> dict:
    """AI初审节点：读取报销单，输出初审意见和风险标签"""
    expense = state["expense_data"]

    # 调用LLM进行初审
    review = llm.invoke(f"请审核以下报销单：{expense}")

    # 解析出风险标签
    tags = extract_risk_tags(review)

    return {
        "ai_review": review,
        "risk_tags": tags,
        "iteration": state.get("iteration", 0) + 1
    }

节点的设计原则是单一职责。一个节点只做一件事：要么读取数据，要么调用模型，要么调用工具，要么做判断。不要把"初审+判断+通知"三件事塞在一个节点里，那样图结构就失去了意义。

概念三：Edge——边，连接节点的路

边定义了节点之间的执行顺序。最简单的边是无条件边：A节点执行完，固定去B节点。

# 普通边：AI初审完后，固定进入人工复核节点
workflow.add_edge("ai_review", "human_review")

但真实世界中，大部分流程都需要根据状态动态决定下一步。这就需要条件边。

概念四：Conditional Edge——条件边，交通网上的红绿灯

条件边是LangGraph的灵魂。它允许你根据当前State的内容，动态选择下一个节点。

def route_after_ai_review(state: ApprovalState) -> str:
    """根据AI初审结果，决定下一步走向"""
    tags = state.get("risk_tags", [])
    amount = state["expense_data"].get("amount", 0)
    iteration = state.get("iteration", 0)

    # 硬规则：金额超过5000，必须人工复核
    if amount > 5000:
        return "human_review"

    # 如果AI打了高风险标签，也转人工
    if "高风险" in tags or "票据异常" in tags:
        return "human_review"

    # 如果已经循环超过3次还没过，强制转人工
    if iteration >= 3:
        return "human_review"

    # 其他情况，自动通过
    return "auto_approve"

# 添加条件边：从ai_review节点出发，根据route_after_ai_review的返回值决定去向
workflow.add_conditional_edges(
    "ai_review",
    route_after_ai_review,
    {
        "human_review": "human_review",   # 返回值"human_review"，跳转到human_review节点
        "auto_approve": "auto_approve"    # 返回值"auto_approve"，跳转到auto_approve节点
    }
)

条件边的强大之处在于：路由逻辑完全由你控制。不是让模型在黑盒里猜，而是你用代码写死规则。金额阈值、风险标签、循环次数，这些业务规则清清楚楚地写在route_after_ai_review函数里，可审计、可测试、可修改。

四、实战：搭建一个"报销审批流"

把上面的概念串起来，我们搭建一个完整的报销审批工作流。

from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict
from langchain_openai import ChatOpenAI

# ===== 1. 定义状态 =====
class ApprovalState(TypedDict):
    expense_data: dict
    ai_review: str
    risk_tags: list
    human_decision: str
    iteration: int
    final_status: str

# ===== 2. 定义节点 =====
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def ai_review_node(state: ApprovalState) -> dict:
    """AI初审：审核报销单，输出意见和标签"""
    expense = state["expense_data"]
    prompt = f"""你是一位财务初审AI。请审核以下报销单，判断票据合规性和金额合理性。
    输出格式：
    意见：<通过/需复核>
    标签：<合规/票据异常/金额超限/类别不符>

    报销单：{expense}"""

    response = llm.invoke(prompt)
    content = response.content

    tags = []
    if "票据异常" in content:
        tags.append("票据异常")
    if "金额超限" in content:
        tags.append("金额超限")
    if "类别不符" in content:
        tags.append("类别不符")

    return {
        "ai_review": content,
        "risk_tags": tags,
        "iteration": state.get("iteration", 0) + 1
    }

def human_review_node(state: ApprovalState) -> dict:
    """人工复核：模拟财务人员的判断（实际项目中这里是人工审批接口）"""
    tags = state.get("risk_tags", [])

    # 模拟人工判断逻辑
    if "票据异常" in tags:
        return {"human_decision": "打回修改", "final_status": "rejected"}
    elif "金额超限" in tags:
        return {"human_decision": "通过但备注", "final_status": "approved_with_note"}
    else:
        return {"human_decision": "通过", "final_status": "approved"}

def auto_approve_node(state: ApprovalState) -> dict:
    """自动通过：AI初审无风险，直接审批"""
    return {"final_status": "approved", "human_decision": "AI自动通过"}

def notify_node(state: ApprovalState) -> dict:
    """通知节点：发送审批结果通知"""
    status = state.get("final_status", "unknown")
    print(f"【通知】报销单审批结果：{status}")
    return {}

# ===== 3. 定义条件路由 =====
def route_after_ai(state: ApprovalState) -> str:
    """AI初审后的路由逻辑"""
    tags = state.get("risk_tags", [])
    amount = state["expense_data"].get("amount", 0)
    iteration = state.get("iteration", 0)

    # 硬规则：金额>5000必须人工复核
    if amount > 5000:
        return "human"

    # 有风险标签，转人工
    if tags:
        return "human"

    # 循环超限，强制人工
    if iteration >= 3:
        return "human"

    return "auto"

# ===== 4. 组装图 =====
workflow = StateGraph(ApprovalState)

# 注册节点
workflow.add_node("ai_review", ai_review_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("auto_approve", auto_approve_node)
workflow.add_node("notify", notify_node)

# 设置入口
workflow.add_edge(START, "ai_review")

# 条件边：AI初审后分流
workflow.add_conditional_edges(
    "ai_review",
    route_after_ai,
    {
        "human": "human_review",
        "auto": "auto_approve"
    }
)

# 人工复核后，统一进入通知节点
workflow.add_edge("human_review", "notify")

# 自动通过后，也进入通知节点
workflow.add_edge("auto_approve", "notify")

# 通知后结束
workflow.add_edge("notify", END)

# ===== 5. 编译并运行 =====
# 加上MemorySaver，状态持久化到内存（生产环境可换成Redis/SQLite）
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 测试：一张3000元的合规报销单
result = app.invoke(
    {
        "expense_data": {"amount": 3000, "category": "差旅", "receipt": "合规"},
        "iteration": 0
    },
    config={"configurable": {"thread_id": "expense_001"}}
)
print(result["final_status"])  # 输出：approved

# 测试：一张8000元的大额报销单
result = app.invoke(
    {
        "expense_data": {"amount": 8000, "category": "招待", "receipt": "合规"},
        "iteration": 0
    },
    config={"configurable": {"thread_id": "expense_002"}}
)
print(result["final_status"])  # 输出：approved_with_note 或 approved

这段代码的精妙之处在于：

1. 流程可视化：你把节点和边画出来，就是一张清晰的流程图。AI初审→条件判断→人工/自动→通知→结束。任何人看一眼代码就能理解业务逻辑。

2. 状态全链路追踪：thread_id让每次审批都有独立的会话状态。如果审批流程中断（比如服务重启），只要用相同的thread_id重新调用，LangGraph会从断点继续执行，而不是从头开始。

3. 规则硬编码：金额超过5000转人工，这不是让模型"看着办"，而是代码里写死的if amount > 5000。业务规则可控、可审计。

4. 循环天然支持：如果未来需求变成"人工打回修改→员工重新提交→AI再审"，只需要在human_review_node返回时增加一个"修改"分支，连一条边回到ai_review节点，就是一个完整的循环。不需要在外层包while，不需要手动维护状态。

五、LangGraph不是银弹：什么时候该用，什么时候不该用

写到这里，有必要再冷静复盘一下。LangGraph很强大，但不是所有场景都需要上交通网。

适合用LangGraph的场景：

• 流程有循环或回溯。比如"生成初稿→审核→不达标→修改→再审"，这种循环用Chain几乎无法实现，用Agent又太黑盒。
• 需要人工介入（Human-in-the-Loop）。比如审批、审核、确认环节，流程必须停下来等人，人操作完再继续。LangGraph的interrupt机制专门支持这种场景。
• 状态需要持久化和恢复。比如一个审批流程可能持续三天，中间服务重启了无数次，状态必须能存能取。LangGraph的Checkpointer支持SQLite、Redis、PostgreSQL等多种后端。
• 需要并行执行。比如"同时查天气、查汇率、查股价"，三个节点互不依赖，可以并行跑，最后汇总结果。图结构天然支持这种并行编排。

不适合用LangGraph的场景：

• 纯线性的简单任务。比如"翻译→润色→输出"，用LCEL的prompt | llm | parser一行搞定，上LangGraph是杀鸡用牛刀。
• 对延迟极度敏感的单轮问答。图编排有额外的调度和状态管理开销，简单场景下不如裸调API快。
• 团队还没掌握LangChain基础。LangGraph的学习曲线比Chain陡，如果团队连PromptTemplate和LCEL都没搞熟，直接上LangGraph会事倍功半。

六、写在最后：从"搭积木"到"画电路图"

LangChain 1.0之后，整个框架的定位发生了本质变化：LangChain退居幕后，成为提供积木的零件库；LangGraph走到台前，成为搭建乐高的电路图。

这个变化反映了一个深层趋势：AI应用开发正在从"单点调用"走向"系统工程"。早期的开发者关心的是"怎么调模型"，现在的开发者关心的是"怎么编排一个可靠的、可审计的、可恢复的工作流"。

Chain和Agent解决了"让AI动起来"的问题，LangGraph解决的是"让AI工作流可控、可观测、可维护"的问题。

如果你今天还在用Chain硬凑循环、用Agent黑盒赌运气，不妨花一个周末学学LangGraph。当你第一次画出一张StateGraph，看到节点和边清清楚楚地拼成一张流程图，看到状态在节点之间有序流转，看到服务重启后流程从断点无缝恢复——你会明白：这不是又一个框架，这是AI工作流编排的成人礼。