6616AI

004|LangGraph 入门:多 Agent 工作流框架

LangGraph 是 LangChain 团队出的多 Agent 工作流框架,它把状态、节点、边三件事封装好,让复杂 Agent 系统变得清晰。

核心结论:LangGraph(读作:兰格·格拉夫,/læŋˈɡræf/)是 LangChain(读作:兰·切恩)团队出的框架,专门用来把多个 LLM 调用或 Agent 组织成一个有状态、有分支、可循环的图。它解决的是复杂多 Agent 系统的流程编排问题。


一、LangGraph 是什么

LangGraph = Lang + Graph

  • Lang 来自 LangChain,代表 LLM 应用。
  • Graph 代表图结构,用节点和边组织流程。

一句话

LangGraph 让你用**图(Graph)**的方式,定义一个多步骤、多 Agent 的工作流。


二、LangGraph 和 LangChain 的关系

LangChain(兰·切恩)LangGraph(兰格·格拉夫)
做单次 LLM 调用做多个 LLM 调用组成的流程
像函数像状态机
链式调用(Linear)图式调用(Graph),有分支、循环、并行
适合简单任务适合复杂多 Agent 系统

LangChain 像乐高积木块,LangGraph 像用这些积木搭一个有复杂结构的模型。


三、为什么需要 LangGraph

实际做 Agent 时,流程往往不是直线。

比如数据分析 Agent:

用户问题
    ↓
意图识别
    ↓
生成 SQL
    ↓
执行 SQL
    ↓
执行成功?
  ├─ 是 → 归因分析
  └─ 否 → 修改 SQL → 重新执行
            ↓
        超过3次 → 返回失败
    ↓
风控检查
    ↓
输出

这个流程里有:

  • 分支(if / else)
  • 循环(重试)
  • 多个 Agent 节点

普通 LangChain 的链式调用很难表达这种结构。LangGraph 用图的方式,天然适合。


四、LangGraph 三个核心概念

1. State(斯泰特)—— 状态

State 是全局共享数据,所有节点都能读写。

class AgentState(TypedDict):
    question: str      # 用户问题
    intent: str        # 意图
    sql: str           # 生成的 SQL
    result: any        # 查询结果
    hypotheses: list   # 归因假设
    risk: dict         # 风控结果
    final_output: str  # 最终输出
    retry_count: int   # 重试次数

2. Node(诺德)—— 节点

每个节点是一个函数或一个 Agent。

def 数据查询节点(state):
    sql = 生成SQL(state["question"])
    result = 执行SQL(sql)
    return {"sql": sql, "result": result}

节点返回的字典会合并到 State 里。

3. Edge(艾吉)—— 边

边控制流程走向。

workflow.add_edge("意图识别", "数据查询")
workflow.add_conditional_edge("数据查询", 检查SQL状态)

普通边:固定流程
条件边:根据状态决定下一步


五、LangGraph 核心 API

API(读法)作用
StateGraph(斯泰特·格拉夫)创建图
add_node(艾德·诺德)添加节点
add_edge(艾德·艾吉)添加普通边
add_conditional_edge(艾德·肯·迪什·艾吉)添加条件边
set_entry_point(赛特·恩·吹·泡·恩特)设置入口节点
compile(康·派尔)编译图
invoke(因·沃克)执行图
stream(斯·特里姆)流式查看中间状态

六、完整伪代码示例

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

# 1. 定义状态
class AgentState(TypedDict):
    question: str
    intent: str
    sql: str
    result: any
    hypotheses: list
    risk_level: str
    final_output: str
    retry_count: int

# 2. 定义节点函数
def 意图识别节点(state):
    intent = 判断意图(state["question"])
    return {"intent": intent}

def 数据查询节点(state):
    sql = 生成SQL(state["question"])
    result = 执行SQL(sql)
    return {"sql": sql, "result": result}

def 重试节点(state):
    return {"retry_count": state["retry_count"] + 1}

def 归因节点(state):
    hypotheses = 生成假设(state["result"])
    return {"hypotheses": hypotheses}

def 风控节点(state):
    risk_level = 检查风险(state["result"], state["hypotheses"])
    return {"risk_level": risk_level}

def 输出节点(state):
    output = 生成回复(state)
    return {"final_output": output}

# 3. 路由函数
def 检查SQL后路由(state):
    if state["result"]["status"] == "成功":
        return "归因节点"
    elif state["retry_count"] < 3:
        return "重试节点"
    else:
        return "输出节点"

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

workflow.add_node("意图识别节点", 意图识别节点)
workflow.add_node("数据查询节点", 数据查询节点)
workflow.add_node("重试节点", 重试节点)
workflow.add_node("归因节点", 归因节点)
workflow.add_node("风控节点", 风控节点)
workflow.add_node("输出节点", 输出节点)

workflow.set_entry_point("意图识别节点")

workflow.add_edge("意图识别节点", "数据查询节点")
workflow.add_conditional_edge("数据查询节点", 检查SQL后路由)
workflow.add_edge("重试节点", "数据查询节点")
workflow.add_edge("归因节点", "风控节点")
workflow.add_edge("风控节点", "输出节点")
workflow.add_edge("输出节点", END)

# 5. 编译并运行
graph = workflow.compile()
result = graph.invoke({
    "question": "为什么广东销售额下降?",
    "retry_count": 0
})

print(result["final_output"])

七、LangGraph 的循环怎么实现

SQL 执行失败后的重试:

def 检查SQL后路由(state):
    if state["result"]["status"] == "成功":
        return "归因节点"
    if state["retry_count"] < 3:
        return "重试节点"
    return "输出节点"

workflow.add_conditional_edge("数据查询节点", 检查SQL后路由)
workflow.add_edge("重试节点", "数据查询节点")

流程:

数据查询节点 → 失败 → 重试节点 → 重试次数 +1 → 回到数据查询节点
           → 成功 → 归因节点
           → 重试 3 次还失败 → 输出节点

八、LangGraph 的并行怎么实现

workflow.add_edge("意图识别", "数据查询")
workflow.add_edge("意图识别", "归因分析")
workflow.add_edge("数据查询", "汇总")
workflow.add_edge("归因分析", "汇总")

数据查询和归因分析同时执行,汇总节点等两个都完成后再执行。


九、LangGraph 的调试

LangGraph 可以拿到每个节点的中间状态:

for event in graph.stream({"question": "为什么广东销售额下降?"}):
    print(event)

输出每个节点执行后的 state 变化,方便调试。


十、LangGraph 的优缺点

优点缺点
流程可控依赖 LangChain 生态
调试能力强学习成本较高
生态成熟节点太多时图会复杂
适合复杂系统异步支持有限

十一、LangGraph 的生态系统

工具读法作用
LangChain兰·切恩基础 LLM、工具、RAG
LangGraph兰格·格拉夫多 Agent 工作流
LangSmith兰·斯密斯监控、调试、评估
LangServe兰·瑟夫部署 API 服务

这四个通常一起用。


十二、一句话总结

LangGraph 把多个 LLM 调用或 Agent 组织成一个有状态、有分支、可循环的图。它让复杂 Agent 系统的流程变得清晰、可调试、可扩展。


下一步可以读

  • 005|主流多 Agent 框架对比:LangGraph / CrewAI / AutoGen / LlamaIndex(已更新)
  • 006|Agent 项目工程化:代码组织、测试、部署(待更新)