如何利用 LangGraph 的 StateGraph 实现带有条件循环的复杂多步工作流

LangGraph 的核心竞争力就在于它把 LLM 的调用从简单的「链式结构」变成了真正的「有向图」。很多朋友习惯了 LangChain 的 Linear Chain，但在处理需要反复修正、循环校验的复杂任务（比如：写代码 → 运行测试 → 报错 → 回到写代码）时，Linear Chain 根本跑不通，必须得用 StateGraph。

实现条件循环的关键在于 add_conditional_edges。它允许你定义一个路由函数，根据当前 State 的值来决定下一步跳转到哪个节点。

这里分享一个我最近在做的「自动化文档校对」工作流，逻辑是：生成内容 → 审查 → 如果不合格则打回重写 → 合格则结束。

核心实现逻辑：

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

# 1. 定义状态对象，所有节点共享这个 State
class AgentState(TypedDict):
    content: str
    review_feedback: str
    iterations: int

# 2. 定义节点逻辑
def generator(state: AgentState):
    # 模拟生成内容，实际这里调用 LLM
    return {"content": "这是生成的初稿", "iterations": state.get("iterations", 0) + 1}

def reviewer(state: AgentState):
    # 模拟审核，如果迭代次数少于2次就故意报错
    if state["iterations"] < 2:
        return {"review_feedback": "逻辑不够严谨，请修改"}
    return {"review_feedback": "通过"}

# 3. 定义路由函数（这是实现循环的灵魂）
def should_continue(state: AgentState):
    if state["review_feedback"] == "通过":
        return "end"
    return "retry"

# 4. 构建图
workflow = StateGraph(AgentState)
workflow.add_node("generate", generator)
workflow.add_node("review", reviewer)

workflow.set_entry_point("generate")
workflow.add_edge("generate", "review")

# 设置条件边：根据 should_continue 的返回值决定去向
workflow.add_conditional_edges(
    "review", 
    should_continue, 
    {
        "retry": "generate", 
        "end": END
    }
)

app = workflow.compile()

避坑指南与效率技巧：

防止死循环：在 State 中一定要加一个 iterations 计数器。LLM 有时候会陷入死循环（比如 A 说不对，B 改了，A 还是说不对），在路由函数里强制判断 if state["iterations"] > 5: return "end"，否则你的 Token 余额会被瞬间烧光。

状态更新陷阱：注意 StateGraph 的状态更新是「覆盖」还是「追加」。默认是覆盖，如果你需要记录所有历史记录，得在 TypedDict 中给字段加上 Annotated[list, operator.add]，这样新节点返回的列表才会 append 到旧列表后面，而不是直接替换掉。

调试技巧：不要直接运行整个图。建议先用 app.get_graph().print_ascii() 打印出拓扑结构，确认逻辑线没连错，再喂数据进去。

这种架构把「控制流」从 Prompt 里抽离了出来，交给 Python 代码控制，比在 Prompt 里写「如果你觉得不对请重新思考」要稳定得多。