本文是 LangGraph 系列的第一篇，带你快速了解 LangGraph 是什么、能干什么、怎么用，以及它背后的核心原理。后续会有更深入的实战和源码解析，敬请期待。

一、LangGraph 是什么？

如果你最近在关注 AI Agent 的开发，应该听说过 LangChain。而 LangGraph 就是 LangChain 生态中专门用来构建 复杂 Agent 工作流 的框架。

简单来说：

• LangChain 提供了与 LLM 交互的标准接口，适合构建简单的链式调用（Chain）

• LangGraph 则专注于 多步骤、有状态、可循环 的 Agent 编排，适合构建复杂的 AI 应用

打个比方：

• LangChain 像是一条流水线，输入进去，按顺序经过几个步骤，输出结果

• LangGraph 像是一个工作流引擎，可以有分支、循环、条件判断，甚至多个 Agent 协作

为什么需要 LangGraph？

在实际的 AI 应用中，你会发现：

1. Agent 需要反复思考：不是一次调用就能得到结果，可能需要多次推理、工具调用、结果验证

2. 需要状态管理：每一步的输出要传递给下一步，甚至要记住之前所有的上下文

3. 需要条件分支：根据上一步的结果，决定下一步该做什么

4. 需要多 Agent 协作：复杂任务需要拆分给不同的专项 Agent 处理

这些需求用传统的链式调用很难满足，于是 LangGraph 应运而生。

二、LangGraph 能干什么？

LangGraph 特别适合构建以下类型的应用：

1. 对话式 AI 系统

• 多轮对话，带上下文记忆

• 根据用户意图动态调整对话流程

• 例如：智能客服、AI 助手

2. 研究和分析 Agent

• 自动搜索资料、提取信息、生成报告

• 多个 Agent 分工协作（一个负责搜索，一个负责总结，一个负责验证）

• 例如：市场调研助手、学术论文分析工具

3. 代码生成和调试

• 生成代码 → 运行测试 → 根据错误修复 → 再次测试

• 循环迭代直到通过测试

• 例如：AI 编程助手、自动化测试工具

4. 业务流程自动化

• 审批流、工单处理、数据处理流水线

• 根据不同条件走不同分支

• 例如：RPA + AI 的自动化系统

三、LangGraph 怎么用？

下面通过一个简单的例子，带你快速上手 LangGraph。

3.1 安装pip install langgraph langchain langchain-openai

3.2 最简单的例子：一个会思考的 Agent

假设我们要做一个简单的 Agent，它会：

1. 接收用户的问题

2. 调用 LLM 生成回答

3. 检查回答是否合理

4. 如果不合理，重新生成

用 LangGraph 实现如下：

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

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 消息历史
    question: str  # 用户问题
    answer: str  # 当前答案
    retry_count: int  # 重试次数

# 2. 定义节点：生成回答
def generate_answer(state: AgentState):
    llm = ChatOpenAI(model="gpt-4")
    messages = state["messages"]
    question = state["question"]
    
    response = llm.invoke([{"role": "user", "content": question}])
    
    return {
        "answer": response.content,
        "messages": [{"role": "assistant", "content": response.content}]
    }

# 3. 定义节点:检查回答
def check_answer(state: AgentState):
    answer = state["answer"]
    # 简单的检查逻辑
    if len(answer) < 10:
        return {"retry_count": state.get("retry_count", 0) + 1}
    return state

# 4. 定义路由逻辑
def should_retry(state: AgentState):
    if state.get("retry_count", 0) > 2:
        return"end"# 重试超过2次，结束
    if len(state.get("answer", "")) < 10:
        return"regenerate"# 回答太短，重新生成
    return"end"# 回答合格，结束

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

# 添加节点
workflow.add_node("generate", generate_answer)
workflow.add_node("check", check_answer)

# 添加边
workflow.set_entry_point("generate")  # 入口
workflow.add_edge("generate", "check")  # generate -> check
workflow.add_conditional_edges(
    "check",
    should_retry,
    {
        "regenerate": "generate",  # 重新生成
        "end": END  # 结束
    }
)

# 编译
app = workflow.compile()

# 6. 运行
result = app.invoke({
    "question": "什么是 LangGraph？",
    "messages": [],
    "answer": "",
    "retry_count": 0
})

print(result["answer"])

3.3 代码解析

这个例子虽然简单，但已经展示了 LangGraph 的核心概念：

1. State（状态）：AgentState 定义了整个流程中需要维护的状态

2. Node（节点）：generate_answer 和 check_answer 是两个处理节点

3. Edge（边）：定义了节点之间的流转关系

4. Conditional Edge（条件边）：should_retry 根据条件决定下一步走向

5. 循环：如果回答不合格，会回到 generate 重新生成

这就是一个最基础的 有状态、可循环 的 Agent 工作流。

四、LangGraph 的核心原理

4.1 核心概念

LangGraph 的设计思想是：把 Agent 的执行流程建模为一个有向图（Directed Graph）。

1. StateGraph（状态图）

• 这是 LangGraph 的核心类

• 它维护了一个 全局状态（State），这个状态会在各个节点之间传递和更新

• 每个节点可以读取状态、修改状态，然后传给下一个节点

2. Node（节点）

• 代表一个独立的计算单元或 Agent

• 每个节点是一个 Python 函数，接收当前状态，返回更新后的状态

• 节点可以调用 LLM、工具、或执行任何逻辑

3. Edge（边）

• 定义节点之间的连接关系

• 分为两种：

  • 普通边：固定的流转路径（A → B）

  • 条件边：根据状态动态决定下一步（A → B 或 A → C）

4. Cycle（循环）

• LangGraph 支持图中的循环（这是 LangGraph 区别于传统 DAG 框架的关键）

• Agent 可以反复执行某些步骤，直到满足条件

• 例如：生成代码 → 测试 → 修复 → 再测试

4.2 执行流程

当你调用 app.invoke(state) 时，LangGraph 会：

1. 从 entry_point 开始

2. 执行当前节点，更新状态

3. 根据边的定义，决定下一个节点

4. 重复 2-3，直到到达 END

整个过程中，状态会持续传递和更新，每个节点都能看到之前步骤的结果。

4.3 与传统 Chain 的区别

特性	LangChain (Chain)	LangGraph
执行方式	线性、顺序执行	图结构、支持分支和循环
状态管理	简单的输入输出传递	全局状态，自动管理
条件逻辑	需要手动编写	原生支持条件边
循环	不支持	原生支持
适用场景	简单的调用链	复杂的 Agent 工作流

五、从线性 Agent 到图结构 Multi-Agent 的演进

这是本文的重点，也是理解 LangGraph 价值的关键。

5.1 第一代：线性 Agent

最早的 Agent 很简单：

用户输入 → LLM → 输出结果

或者稍微复杂一点：

用户输入 → LLM(生成计划) → 调用工具 → LLM(总结) → 输出结果

这种线性结构的问题：

• 无法根据中间结果调整策略

• 无法处理复杂的多步推理

• 无法让多个 Agent 协作

5.2 第二代：带循环的单 Agent

引入了 ReAct 模式（Reason + Act）：

用户输入 
  → LLM(思考下一步) 
  → 调用工具 
  → LLM(分析结果) 
  → 判断是否结束
    - 没结束 → 回到"思考下一步"
    - 结束 → 输出结果

这解决了循环问题，但：

• 所有逻辑都在一个 Agent 里，容易混乱

• 无法并行处理多个任务

• 难以维护和扩展

5.3 第三代：图结构 Multi-Agent（LangGraph 的方案）

LangGraph 提出的方案是：把复杂任务拆分成多个专项 Agent，用图结构编排。

架构模式 1：Supervisor（监督者模式）

• Supervisor 接收任务，决定派给哪个 Agent

• 每个 专项 Agent 只负责自己擅长的事

• 最后 Aggregator 汇总结果

代码示例：

from langgraph.graph import StateGraph, END

# 定义状态
class MultiAgentState(TypedDict):
    task: str
    current_agent: str
    results: dict

# 监督者：决定下一步
def supervisor(state):
    task = state["task"]
    if"搜索"in task:
        return {"current_agent": "search_agent"}
    elif"分析"in task:
        return {"current_agent": "analyze_agent"}
    else:
        return {"current_agent": "general_agent"}

# 搜索 Agent
def search_agent(state):
    # 模拟搜索逻辑
    return {"results": {"search": "搜索结果..."}}

# 分析 Agent
def analyze_agent(state):
    # 模拟分析逻辑
    return {"results": {"analyze": "分析结果..."}}

# 路由逻辑
def route_to_agent(state):
    return state["current_agent"]

# 构建图
workflow = StateGraph(MultiAgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("search_agent", search_agent)
workflow.add_node("analyze_agent", analyze_agent)

workflow.set_entry_point("supervisor")
workflow.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "search_agent": "search_agent",
        "analyze_agent": "analyze_agent"
    }
)
workflow.add_edge("search_agent", END)
workflow.add_edge("analyze_agent", END)

app = workflow.compile()

架构模式 2：Collaborative（协作模式）

多个 Agent 共享同一个状态，互相传递信息：

每个 Agent 都能看到前面 Agent 的结果，并在此基础上继续工作。

代码示例：

class CollaborativeState(TypedDict):
    messages: Annotated[list, operator.add]
    research_data: dict
    draft_content: str
    final_output: str

def researcher(state):
    # 研究 Agent：收集信息
    return {"research_data": {"info": "..."}}

def writer(state):
    # 写作 Agent：根据研究结果起草
    research = state["research_data"]
    return {"draft_content": f"基于{research}的草稿..."}

def reviewer(state):
    # 审核 Agent：审核并优化
    draft = state["draft_content"]
    return {"final_output": f"优化后的{draft}"}

workflow = StateGraph(CollaborativeState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_edge("reviewer", END)

app = workflow.compile()

架构模式 3：LATS（Language Agent Tree Search）

更高级的模式，Agent 会 探索多个分支，评估结果,选择最优路径：

这适合需要"试错"的任务，比如代码生成（尝试多种实现，选最优）。

5.4 为什么图结构更强大？

对比一下：

维度	线性 Agent	图结构 Multi-Agent
复杂度管理	所有逻辑堆在一起	拆分成独立模块
可维护性	难以修改和扩展	每个 Agent 独立开发
并行能力	无法并行	可以并行执行多个 Agent
容错性	一步出错全盘皆输	可以局部重试或回退
可观测性	黑盒	每个节点可单独监控
灵活性	固定流程	动态路由、条件分支

最关键的优势：循环和反馈

传统 DAG（有向无环图）不允许循环，但 LangGraph 支持。这意味着：

• Agent 可以 反复思考（生成 → 验证 → 修正 → 再验证）

• Agent 可以 自我改进（根据之前的失败调整策略）

• 支持 Human-in-the-Loop（人工介入后继续执行)

六、LangGraph 的实际应用场景

1. 代码助手

2. 内容创作

3. 数据分析

七、总结

核心要点回顾

1. LangGraph 是什么？
   LangChain 生态中用于构建复杂 Agent 工作流的框架

2. 核心概念：
   State（状态）、Node（节点）、Edge（边）、Cycle（循环）

3. 关键优势：
   支持循环、条件分支、多 Agent 协作、状态管理

4. 架构演进：
   线性 Agent → 单 Agent 循环 → 图结构 Multi-Agent

什么时候用 LangGraph？

推荐使用：

• 需要多步推理和工具调用

• 需要根据中间结果动态调整流程

• 需要多个 Agent 分工协作

• 需要循环迭代（如生成-验证-修正）

不推荐使用：

• 简单的一次性 LLM 调用

• 固定的线性流程

• 对性能要求极高的场景（图执行有一定开销）

如果你对 LangGraph 感兴趣，可以先去官方文档(https://langchain-ai.github.io/langgraph/)看看更多例子。