LangGraph：当 AI 需要"反复思考"时

有段时间我在做一个代码生成工具：用户描述需求，AI 写代码，然后自动运行测试。听起来简单，但有个问题——代码第一次不一定能跑通，要能根据报错自动修改再试。

我当时的想法是用 LangChain 的普通 Chain 做：生成代码 → 运行测试 → 如果报错就再生成。写了半天发现根本实现不了，因为普通 Chain 是单向流水线，A → B → C 走完就结束了，没有"回头"这个操作。

然后发现了 LangGraph。

---

#Chain 和 LangGraph 的根本区别

普通 Chain 是高速公路：只能直行，有入口有出口，不能回头。

LangGraph 是城市路网：可以转弯、可以回头、可以根据路况选不同路线、可以在某个路口等待（人工审核）。

普通 Chain（单向）：

Prompt → LLM → Parser → 结束

LangGraph（可循环）：

          ┌─────────────────────┐
          │                     │
          ▼                     │ 不合格，重试
    ┌──────────┐          ┌──────────┐
    │  生成节点 │ ────────▶ │  审核节点 │
    └──────────┘  生成完成  └──────────┘
                                │ 合格
                                ▼
                             结束

所以什么时候该用 LangGraph？

一个判断标准：你的工作流里有没有"根据结果决定下一步"的逻辑？如果有，尤其是有循环（"不合格就重试"），就该用 LangGraph。没有的话，普通 Chain 更简单，没必要引入额外复杂度。

我见过很多人一上来就用 LangGraph 做所有事，然后发现 90% 的代码都是在管理节点和边，反而比直接写逻辑更复杂。先用最简单的方案，真的碰到"需要循环"时再考虑 LangGraph。

---

#三个核心概念

LangGraph 就三个东西：State、Node、Edge。

State（状态） 是节点之间共享的数据容器。所有节点都从这里读数据，也把结果写回这里。用 TypedDict 定义：

python复制
from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    question: str       # 普通字段：后写的会覆盖前写的
    answer: str
    quality_score: int

    # Annotated[list, operator.add]：每次追加而不是覆盖
    # 适合收集每一轮的结果
    revision_history: Annotated[list, operator.add]

把 State 想象成流水线上的托盘。每个工站（节点）都能读取托盘上的东西，处理完再把结果放回托盘，传给下一站。

Node（节点） 是普通的 Python 函数，接收 State，返回要更新的字段：

python复制
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def generate_answer(state: State) -> dict:
    response = llm.invoke(f"请回答：{state['question']}")
    return {
        "answer": response.content,
        "revision_history": [f"第 {len(state.get('revision_history', [])) + 1} 次生成"],
    }

def review_answer(state: State) -> dict:
    score_str = llm.invoke(
        f"给这个答案打分（1-10，只输出数字）：{state['answer']}"
    ).content.strip()
    return {"quality_score": int(score_str)}

Edge（边） 是节点之间的连接。有两种：固定边（总是走这条路）和条件边（根据逻辑决定走哪条路）：

python复制
from langgraph.graph import StateGraph, END

workflow = StateGraph(State)
workflow.add_node("generate", generate_answer)
workflow.add_node("review", review_answer)

# 固定边：生成完总是去审核
workflow.add_edge("generate", "review")

# 条件边：审核完根据分数决定
def should_retry(state: State) -> str:
    attempts = len(state.get("revision_history", []))
    if state["quality_score"] >= 7 or attempts >= 3:
        return "done"
    return "retry"

workflow.add_conditional_edges(
    "review",
    should_retry,
    {"retry": "generate", "done": END}
)

---

#完整示例：内容生成 + 自动质检循环

python复制
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

class SummaryState(TypedDict):
    article: str
    summary: str
    feedback: str
    quality_score: int
    attempts: Annotated[list, operator.add]

def generate_summary(state: SummaryState) -> dict:
    if state.get("feedback"):
        prompt = f"""根据以下意见修改摘要：
意见：{state['feedback']}
当前摘要：{state['summary']}
原文：{state['article'][:500]}

输出改进后的摘要（100 字以内）："""
    else:
        prompt = f"请为以下文章写一个 100 字以内的摘要：\n{state['article'][:500]}"

    response = llm.invoke(prompt)
    return {
        "summary": response.content,
        "attempts": [f"第 {len(state.get('attempts', [])) + 1} 次生成"],
    }

def review_summary(state: SummaryState) -> dict:
    prompt = f"""评审这个摘要（是否准确、简洁、完整）：
原文：{state['article'][:500]}
摘要：{state['summary']}

输出两行：
第一行：分数（1-10）
第二行：改进意见（分数 >= 7 则写"通过"）"""

    lines = llm.invoke(prompt).content.strip().split('\n')
    score = int(lines[0].strip())
    feedback = lines[1].strip() if len(lines) > 1 else "通过"

    return {
        "quality_score": score,
        "feedback": feedback if score < 7 else "",
    }

def should_revise(state: SummaryState) -> str:
    if state["quality_score"] >= 7 or len(state.get("attempts", [])) >= 3:
        return "done"
    return "revise"

# 构建图
workflow = StateGraph(SummaryState)
workflow.add_node("generate", generate_summary)
workflow.add_node("review", review_summary)

workflow.set_entry_point("generate")
workflow.add_edge("generate", "review")
workflow.add_conditional_edges("review", should_revise, {"revise": "generate", "done": END})

app = workflow.compile()

# 运行
article = "LangChain 是一个用于构建基于大型语言模型应用的开源框架..." * 20
result = app.invoke({
    "article": article,
    "summary": "",
    "feedback": "",
    "quality_score": 0,
    "attempts": [],
})

print(f"最终摘要（{len(result['attempts'])} 次生成）：")
print(result["summary"])
print(f"质量分：{result['quality_score']}")

---

#LangGraph 的三个独特能力

普通 Chain 没有这三个东西，这才是 LangGraph 真正的价值所在。

#1. 状态持久化（Checkpointer）

Checkpointer 让 LangGraph 支持断点恢复。用户关掉浏览器、服务器重启、程序中途崩溃，下次回来可以从中断的地方继续，不用重跑整个流程。

python复制
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()  # 开发用；生产换 SqliteSaver 或 PostgresSaver
app = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "session_abc123"}}

# 第一次调用
app.invoke(input_data, config=config)

# 下次继续（自动从上次中断处恢复）
app.invoke(next_input, config=config)

# 查看历史状态
history = list(app.get_state_history(config))

这个功能对长时间运行的 Agent 非常重要，比如一个需要几分钟完成多步骤分析的工作流。

#2. Human-in-the-Loop

在关键步骤暂停，等待人工确认再继续。做金融审批、内容发布等需要人工把关的场景，这个功能价值很大。

python复制
from langgraph.graph import interrupt

def human_approval_node(state: State) -> dict:
    # 到这个节点时，图会暂停，等待外部输入
    human_input = interrupt({
        "question": "是否批准此操作？",
        "current_result": state["answer"],
    })
    return {"approved": human_input == "yes"}

#3. 多 Agent 协作

把多个专业 Agent 定义为不同节点，让它们分工协作：

用户需求
   │
   ▼
研究员 Agent → 搜索相关资料
   │
   ▼
写手 Agent → 基于资料写初稿
   │
   ▼
审核 Agent → 检查质量
   │ 不合格
   └─────────→ 写手 Agent（修改）
   │ 合格
   ▼
输出给用户

每个 Agent 是一个节点，状态在它们之间共享和传递。

---

#常见坑

循环停不下来：条件边逻辑有 bug，永远走重试路径。必须加 attempts 上限作为强制退出条件，无论如何超过 N 次就结束。这个错误我见过真实项目里发生，花了几十美元才停下来。

State 字段越来越多：把所有临时变量都放进 State。State 只存节点间需要共享的数据，临时计算用本地变量。

节点报错整个图崩溃：节点里没有异常处理。在节点函数里加 try/except，把错误信息存进 State，让下游节点知道发生了什么。

过早引入 LangGraph：80% 的 LangChain 应用不需要 LangGraph。先用 LCEL Chain 跑通，只有当你真的遇到"需要根据结果决定下一步"时才引入。

忘记 .compile()：StateGraph 构建完必须调用 .compile() 才能运行。修改图结构后需要重新 compile。

---

#动手练习

练习 1：实现 AI 自动修复 Python 代码语法错误，最多重试 2 次：

python复制
from typing import TypedDict, Annotated
import operator

class CodeState(TypedDict):
    task: str         # 编程任务描述
    code: str         # 当前代码
    error: str        # 运行错误信息（空字符串表示无错误）
    attempts: Annotated[list, operator.add]

def write_code(state: CodeState) -> dict:
    # TODO：如果 state["error"] 非空，提示 AI 修复错误
    # 如果是第一次，让 AI 根据 state["task"] 写代码
    pass

def run_code(state: CodeState) -> dict:
    # TODO：用 exec() 运行 state["code"]
    # 成功：state["error"] = ""
    # 报错：把错误信息存到 state["error"]
    pass

def should_fix(state: CodeState) -> str:
    # TODO：有错误 AND 尝试次数 < 2 → "fix"
    # 否则 → "done"
    pass

练习 2：在上面的基础上加入 Human-in-the-Loop：代码运行成功后暂停，让人工审查代码，输入 "approve" 才结束，输入 "reject" 让 AI 重写。

---

#小结

1. LangGraph 解决 Chain 做不到的"循环"问题——需要"重试"或"根据结果走不同路径"时才用，不要过早引入。
2. 三要素：State（节点间共享的数据托盘）、Node（普通 Python 函数）、Edge（连接路径，条件边实现分支/循环）。
3. add_conditional_edges 是实现循环的核心，一定要给循环加最大次数上限，没有上限的循环是定时炸弹。
4. Checkpointer 实现断点恢复，是生产级 Agent 的必备能力——开发用 MemorySaver，生产用数据库持久化。
5. 不是所有任务都需要 LangGraph。先用最简单的 Chain，当真正碰到"需要循环"时再升级，而不是一开始就做最复杂的架构。

---

下一步：Agents 代理系统 — 了解 Agent 如何用 LangGraph 实现自主工具调用

可视化调试：LangGraph Studio↗ — 实时观察节点状态、手动回滚步骤