LangGraph 实战：打破线性枷锁，从流水线到状态机

LangGraph 实战（三）：打破线性枷锁，从流水线到状态机

1. 为什么我们需要 LangGraph？

在学习 LangChain 的过程中，你可能已经发现了一个局限性：Chain 太直了。

标准的 LangChain 工作流（LCEL）就像工厂的流水线： Prompt -> LLM -> OutputParser

不管中间发生什么，它必须按顺序走完。但是，如果我们想实现以下逻辑怎么办？

1. 循环（Cycles）：让 AI 写一篇文章，然后自己检查。如果写得不好，退回去重写，直到满意为止。
2. 分支（Branching）：根据 AI 的回答决定下一步是去“搜索网页”还是“直接回答”。
3. 持久化（Persistence）：在一个长达数天的任务中，保存当前的进度，明天接着跑。

用基础的 LangChain 实现这些非常痛苦，你可能需要写大量的 Python while 循环和 if-else 胶水代码。

LangGraph 的出现就是为了解决这个问题。它不是把步骤看作一条线，而是看作一个图（Graph）。更准确地说，它是一个状态机（State Machine）。

2. 核心概念：图的解剖学

在写代码之前，必须彻底理解 LangGraph 的三个核心概念。如果理解了这三个词，代码就很好懂了。

1. State (状态)

这是 LangGraph 的灵魂。 想象大家在一个会议室里开会。会议室的白板上写着当前讨论的所有内容。

• 每个参会者（Node）都可以看白板。
• 每个参会者都可以拿起笔，往白板上写新东西，或者修改旧东西。
• 当会议结束时，白板上的内容就是最终结果。

在代码中，State 就是一个 Python 字典（或 Pydantic 模型），它在整个图中流转，存储着对话历史、中间结果和决策信息。

2. Nodes (节点)

节点就是干活的人。 在 LangChain 里，节点通常是一个 LLM，或者一个工具（Tool），或者一个普通的 Python 函数。 节点接收当前的 State，干活（比如调用 GPT 生成文本），然后输出一个更新后的 State。

3. Edges (边)

边是连接节点的路线，决定了“下一步去哪儿”。

• 普通边：A 做完去 B。
• 条件边 (Conditional Edges)：这是最强大的地方。A 做完后，运行一个逻辑判断函数。如果结果是 "good"，去 B；如果结果是 "bad"，回 A。

3. 环境准备

LangGraph 是一个独立的库，虽然它构建在 LangChain 之上，但需要单独安装。

# 安装 LangGraph
pip install langgraph

# 确保你有 LangChain 核心库
pip install langchain langchain-openai

4. 实战：构建一个“甚至懂得自我反思”的循环系统

为了演示 LangGraph 的威力，我们不做简单的聊天机器人。我们要构建一个 “循环改进系统”。

任务目标：让 AI 写一个关于某个主题的短句。然后有一个“审稿人”来打分。如果分数太低，打回去重写。如果分数够高，结束。

这是一个典型的 Cyclic Graph（循环图）。

第一步：定义状态 (Define State)

我们需要一个结构来存储：当前的草稿、当前的修改次数。

import operator
from typing import Annotated, TypedDict, Union

# 我们定义一个字典结构，这就是我们的"白板"
# 这里的 State 继承自 TypedDict，只是为了让编辑器有代码提示
class AgentState(TypedDict):
    draft: str          # 当前生成的草稿
    critique: str       # 审稿意见
    revision_number: int # 当前是第几次修改
    content: str        # 最终输出的内容

第二步：定义节点 (Define Nodes)

我们需要两个节点（函数）：

1. Writer（作家）：负责写或者重写。
2. Critic（批评家）：负责评价。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# --- 节点 1: 作家 ---
def writer_node(state: AgentState):
    # 从状态中获取当前信息
    draft = state.get("draft", "")
    critique = state.get("critique", "")
    revision_number = state.get("revision_number", 0)
    
    # 如果是第一次写
    if revision_number == 0:
        prompt = ChatPromptTemplate.from_template(
            "请写一句关于 {topic} 的简短标语，要幽默一点。"
        )
        # 注意：这里的 topic 我们稍后会通过初始状态传入
        # 为了演示简单，这里我们硬编码一个 topic 或者假设它在 context 里
        # 在真实场景中，topic 应该也在 State 里
        res = prompt | model | StrOutputParser()
        new_draft = res.invoke({"topic": "程序员的头发"})
        print(f"--- [Writer] 初稿: {new_draft} ---")
    else:
        # 如果是重写，需要参考批评意见
        prompt = ChatPromptTemplate.from_template(
            "这是你之前的草稿: {draft}\n"
            "这是审稿人的意见: {critique}\n"
            "请根据意见重写一个新的版本。"
        )
        res = prompt | model | StrOutputParser()
        new_draft = res.invoke({"draft": draft, "critique": critique})
        print(f"--- [Writer] 第 {revision_number} 次修改稿: {new_draft} ---")

    # 返回更新后的状态（LangGraph 会自动合并这些字段到主 State 中）
    return {
        "draft": new_draft, 
        "revision_number": revision_number + 1
    }

# --- 节点 2: 批评家 ---
def critic_node(state: AgentState):
    draft = state.get("draft")
    
    # 让 AI 扮演严厉的编辑
    prompt = ChatPromptTemplate.from_template(
        "请评价这句标语: {draft}\n"
        "如果它很有趣且简短，回答 'PASS'。\n"
        "如果不够好，请给出简短的修改建议（不要包含 PASS 字样）。"
    )
    
    res = prompt | model | StrOutputParser()
    result = res.invoke({"draft": draft})
    
    print(f"--- [Critic] 意见: {result} ---")
    
    return {"critique": result}

第三步：定义条件边 (Conditional Edges)

这是一个逻辑判断函数。它不修改状态，只负责指路。

def should_continue(state: AgentState):
    critique = state.get("critique")
    revision_number = state.get("revision_number")
    
    # 限制最大循环次数，防止死循环烧干钱包
    if revision_number > 3:
        print("--- 达到最大修改次数，强制结束 ---")
        return "end"
    
    # 如果包含 PASS，说明通过了
    if "PASS" in critique:
        print("--- 审核通过！ ---")
        return "end"
    
    # 否则，继续回去修改
    return "continue"

第四步：构建图 (Build Graph)

这是 LangGraph 最独特的部分。我们需要实例化 StateGraph，添加节点，连接边，然后编译。

from langgraph.graph import StateGraph, END

# 1. 初始化图，传入我们定义的状态结构
workflow = StateGraph(AgentState)

# 2. 添加节点
# add_node("名字", 函数)
workflow.add_node("writer", writer_node)
workflow.add_node("critic", critic_node)

# 3. 设置入口点
# 告诉图，启动时先跑哪个节点
workflow.set_entry_point("writer")

# 4. 添加普通的边
# 作家写完，必须交给批评家看。这是固定的流向。
workflow.add_edge("writer", "critic")

# 5. 添加条件边
# 批评家看完后，根据 should_continue 的结果决定去哪。
# add_conditional_edges("起点", 判断函数, 路由字典)
workflow.add_conditional_edges(
    "critic", 
    should_continue,
    {
        "continue": "writer", # 如果返回 continue，回到 writer
        "end": END            # 如果返回 end，结束运行
    }
)

# 6. 编译
# 编译后的 app 就像一个普通的 LangChain Runnable，可以用 .invoke()
app = workflow.compile()

第五步：运行与观察

# 初始化输入状态
inputs = {"revision_number": 0}

print("开始运行循环优化系统...\n")

# app.invoke 会阻塞直到图运行结束（到达 END 节点）
final_output = app.invoke(inputs)

print("\n最终结果状态:")
print(f"最终草稿: {final_output['draft']}")
print(f"总共修改次数: {final_output['revision_number']}")

5. 深度解析：这和普通 Chain 有什么本质区别？

状态的持久性

在普通的 SequentialChain 中，数据像接力棒一样传递。如果链条断了，中间状态很难保存。 在 StateGraph 中，AgentState 始终存在。如果我们在第 2 次循环时程序崩溃了，理论上我们可以加载这个 State，直接从第 2 次循环继续跑，而不需要从头开始（这是 LangGraph 持久化功能的各种基础，我们将在后续章节详细讲）。

循环控制

注意看 should_continue 函数。在普通的 Chain 里，要实现“如果不满意就重做”，你需要在 Python 层面写 while True 循环来不断调用 Chain。 而在 LangGraph 里，循环是图结构的一部分。这意味着整个逻辑被封装在图内部，外部调用者只需要 app.invoke() 一次，不需要关心内部转了多少圈。

细粒度的控制

每个节点（Writer, Critic）都是独立的函数。这使得调试非常容易。你可以单独测试 writer_node，而不需要运行整个图。

6. 总结

在这一篇中，我们迈出了最重要的一步：从线到网。

我们构建了一个拥有自我纠错能力的简单系统。虽然它现在只是写写笑话，但请想象一下，把“写笑话”换成“写代码”，把“批评家”换成“运行单元测试”。如果测试不通过，就让 AI 看报错信息并修改代码，直到测试通过。这就是当下最火的 AI 编程助手（AI Coding Agent） 的基本原理。

但是，现在的图还是全自动运行的。在下一篇中，我们将通过引入 Tools（工具），构建一个真正的 ReAct Agent，让它能够自主决定是去搜索、计算，还是直接回答。我们将深入探讨 LangGraph 官方推荐的 Agent 架构。