随着大语言模型（LLM）应用的不断深入，我们发现单纯的“线性链式”逻辑（Chains）已经难以满足复杂的业务需求。Agent（智能体）的兴起要求系统具备循环、自我修正和记忆能力。

作为 LangChain 生态的最新重要成员，LangGraph 应运而生。本文带你理解从 LangChain (DAG) 到 LangGraph (State Graph) 的核心思想转变。

1. 范式转移：从 DAG 到 State Graph

在使用 LangChain（特别是 1.x 版本之前的概念）构建应用时，我们的思维模型通常是 DAG（有向无环图）。

• LangChain (The Chain): 就像一条流水线。数据从输入端进入，经过 PromptTemplate、LLM、OutputParser 等一系列处理，最终产出结果。它的特点是单向、无环。
• LangGraph (The Graph): 就像一个状态机。系统维护一个共享的“状态（State）”，不同的处理单元（Node）读取并更新这个状态，执行流可以根据条件在节点之间跳转，甚至**循环（Cycle）**回到之前的节点。

架构对比示意图

核心区别： LangGraph 允许“回路”。这对于 Agent 来说至关重要，因为 Agent 经常需要“思考 -> 执行 -> 观察 -> 再思考”的循环过程。

---

2. LangGraph 的三大核心概念

要掌握 LangGraph，必须理解三个核心原语：State（状态）、Node（节点） 和 Edge（边）。

2.1 State（状态）：系统的记忆

在 LangChain Expression Language (LCEL) 中，我们通常通过传递字典或字符串在 Chain 之间流动数据。而在 LangGraph 中，State 是所有节点共享的数据结构。

• 它定义了图的“Schema”（通常使用 TypedDict 或 Pydantic 模型）。
• 每个节点执行完后，会返回一个更新（Update），LangGraph 负责将这个更新**合并（Merge）**到当前状态中，而不是完全覆盖。

from typing import TypedDict, Annotated
import operator

# 定义 State
class AgentState(TypedDict):
    # messages 列表，新的消息会被 append 进去而不是覆盖 (由 operator.add 控制)
    messages: Annotated[list[str], operator.add]
    # 当前步骤计数
    steps: int

2.2 Node（节点）：执行单元

Node 就是具体的执行逻辑，通常是一个 Python 函数。

• 输入：当前的 State。
• 输出：对 State 的部分更新（Partial Update）。

def call_model_node(state: AgentState):
    # 获取历史消息
    messages = state['messages']
    # 调用模型 (假设 model 是已初始化的 LangChain 对象)
    response = model.invoke(messages)
    # 返回更新：将新回复追加到 messages 列表中，并更新步数
    return {
        "messages": [response.content], 
        "steps": state['steps'] + 1
    }

2.3 Edge（边）：控制流

Edge 定义了节点之间的连接关系。LangGraph 有两种边：

1. 普通边 (Normal Edge)：从 A 节点无条件流向 B 节点。
2. 条件边 (Conditional Edge)：使用一个路由函数（Router），根据当前的 State 决定下一个节点是谁。这是实现 Agent 动态决策的关键。

---

3. LangGraph vs. LangChain：如何选择？

很多开发者会问：“我现有的 LangChain 代码需要重构吗？”。答案是：不一定。LangGraph 是建立在 LangChain 之上的，它们不是互斥关系，而是互补关系。

对比分析表

特性	LangChain (LCEL / Chains)	LangGraph
核心模型	DAG (有向无环图)	State Machine (状态机)
数据流向	线性传递 (Input -> Output)	状态持久化与更新 (State Persistence)
循环能力	困难 (需要递归或硬编码循环)	原生支持 (Native Loops)
容错与纠正	一次性执行，失败即报错	可在循环中自我修正 (Self-Correction)
记忆管理	依赖外部 Memory 组件	State 本身就是短期记忆，支持 Checkpoint 持久化
适用场景	简单的 RAG、明确的问答管道	Agent、多步复杂任务、人机交互 (Human-in-the-loop)

什么时候必须使用 LangGraph？

1. 你需要循环（Loops）： 例如代码生成场景。LLM 生成代码 -> 运行测试 -> 报错 -> 将错误回传给 LLM -> LLM 修正代码。这个“修正”的过程在传统 Chain 中很难优雅实现。
2. 多 Agent 协作（Multi-Agent）： 你需要一个专门的“主管 Agent”来分发任务给“搜索 Agent”和“计算 Agent”，并汇总结果。
3. 人机协同（Human-in-the-loop）： 任务执行到一半，需要暂停，等待人类批准或修改输入，然后再继续执行。LangGraph 的 checkpointer 机制完美支持这一点。
4. 非确定性路径： 执行路径高度依赖 LLM 的实时判断，而非预设的硬编码逻辑。

---

4. 实战代码结构预览 (Python)

下面是一个使用 LangGraph 构建最基础的“ReAct”风格循环图的代码骨架。

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

# 1. 定义状态
class State(TypedDict):
    messages: Annotated[list, operator.add]

# 2. 初始化图
workflow = StateGraph(State)

# 3. 添加节点 (可以是 LangChain 的 Runnable 或普通函数)
workflow.add_node("agent", agent_node_function)
workflow.add_node("action", tool_execution_function)

# 4. 设置入口点
workflow.set_entry_point("agent")

# 5. 添加条件边
def should_continue(state):
    last_message = state['messages'][-1]
    if "tool_calls" in last_message:
        return "continue"
    return "end"

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)

# 6. 添加普通边 (工具执行完必须回到 Agent 进行下一次思考)
workflow.add_edge("action", "agent")

# 7. 编译图 (这一步会进行校验)
app = workflow.compile()

# 8. 运行
inputs = {"messages": [("user", "查询一下今天北京的天气")]}
for output in app.stream(inputs):
    # 实时流式输出每一个节点的执行结果
    print(list(output.keys()), output)

---

5. 进阶实战：构建带自我修正 (Self-Correction) 的 Agent

为了体现 LangGraph 的强大，我们构建一个具备“自我反思”能力的 Agent。它不仅仅是执行任务，还会检查执行结果。如果结果不满意，它会自我批评并重新规划。

工作流： Plan -> Execute -> Reflect -> (Decision)

• Plan: 生成行动计划。
• Execute: 执行计划（例如调用工具或生成代码）。
• Reflect: 评估执行结果。如果通过，结束；如果不通过，提出修改意见。
• Decision: 根据 Reflect 的结果，决定是结束任务还是带着反馈回到 Plan 阶段。

5.1 状态机图示

5.2 核心代码实现

这是一个简化的逻辑实现，展示如何利用 State 在循环中传递“反馈”。

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

# 1. 定义状态：包含任务、计划、结果和反思反馈
class AgentState(TypedDict):
    task: str
    plan: str
    result: str
    feedback: Optional[str]
    retry_count: int

# 2. 定义节点函数 (模拟)
def plan_node(state: AgentState):
    print("--- Planning ---")
    # 如果有反馈，说明是重试，根据反馈调整计划
    if state.get("feedback"):
        new_plan = f"Revised plan for '{state['task']}' based on feedback: {state['feedback']}"
    else:
        new_plan = f"Initial plan for '{state['task']}'"
    return {"plan": new_plan}

def execute_node(state: AgentState):
    print("--- Executing ---")
    plan = state["plan"]
    # 模拟执行
    result = f"Executed content for: {plan}" 
    return {"result": result}

def reflect_node(state: AgentState):
    print("--- Reflecting ---")
    # 模拟反思逻辑：这里简单的假设第一次总是失败，第二次成功
    retry_count = state.get("retry_count", 0)
    
    if retry_count < 1:
        # 第一次反思，提出改进意见
        return {
            "feedback": "Result is too simple, add more details.",
            "retry_count": retry_count + 1
        }
    else:
        # 通过测试，标记通过
        return {"feedback": "PASS", "retry_count": retry_count + 1}

# 3. 定义条件判断函数
def should_continue(state: AgentState):
    if state["feedback"] == "PASS":
        return "end"
    return "continue"

# 4. 构建图
workflow = StateGraph(AgentState)

workflow.add_node("planner", plan_node)
workflow.add_node("executor", execute_node)
workflow.add_node("reflector", reflect_node)

workflow.set_entry_point("planner")

# 建立连接
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", "reflector")

# 条件跳转
workflow.add_conditional_edges(
    "reflector",
    should_continue,
    {
        "continue": "planner", # 反馈不好，回到计划阶段
        "end": END             # 反馈通过，结束
    }
)

app = workflow.compile()

# 5. 运行
inputs = {"task": "Write a Python script", "retry_count": 0, "feedback": None}
for output in app.stream(inputs):
    # 打印每次状态更新的节点名
    print(f"Node finished: {list(output.keys())}")

这个模式展示了 LangGraph 最强大的特性：状态持久化与循环控制。在传统的 Chain 中，实现这种“带着反馈回到上一步”的逻辑是非常复杂的，但在 Graph 中，它只是一个指向旧节点的边。

---

6. 总结

从 LangChain 到 LangGraph，不仅仅是工具的更换，更是开发思维的升级：

• 我们不再只是设计“输入到输出”的管道。
• 我们开始设计系统如何维护状态，以及系统如何在不同状态间流转。

如果你正在构建简单的 RAG 应用，LangChain 的 LCEL 依然是最快、最简洁的选择。但如果你试图构建一个能够自主决策、自我修正、甚至长期运行的智能体，LangGraph 是目前 Python 生态中架构最清晰、扩展性最强的选择。

---