LangGraph 技术详解:基于图结构的 AI 工作流与多智能体编排框架

引言
随着大语言模型能力不断增强,AI 应用的复杂度迅速上升。
现实中的AI 系统,早已不再是 “一次 Prompt → 一次回答” 的简单模式,而是逐渐演变为:
•多步骤推理
•多Agent 协作
•带条件分支与循环的工作流
•需要长期状态管理的系统
在这种背景下,LangGraph 应运而生。
一、LangGraph 是什么
LangGraph 是一个基于图结构(Graph)的 AI 工作流框架,用于构建可控、多步骤、多智能体的 LLM 应用系统。
一句话概括:
LangGraph 让 AI 应用从 “线性调用” 升级为 “有状态、可回路、可控制的推理图”。
如果从架构层理解:
•LangChain 提供的是构建LLM 能力的 “组件与抽象”
•LangGraph 提供的是组织这些能力的“执行结构与控制模型”
LangGraph 的核心价值不在于 “调用模型”,而在于控制模型如何协作、何时执行、是否回退、是否重试、如何流转状态。
二、为什么需要LangGraph
在复杂AI 应用中,传统 LangChain 会逐渐暴露一些结构性问题:
1. 线性 Chain 难以表达真实流程
LLM 应用往往存在:
•条件分支
•失败重试
•多阶段决策
•回环修正
单向Chain 在表达这些逻辑时非常笨重。
2. 多 Agent 协作缺乏统一调度
当系统中存在多个Agent(检索、写作、审查、工具执行)时,需要一个统一的流程控制器。
3. 状态管理分散
上下文、阶段结果、中间结论往往散落在不同Chain 中,难以统一管理和追踪。
LangGraph 正是为了解决这些问题而设计的。
三、LangGraph 能做什么
LangGraph 非常适合以下类型的 AI 系统:
1. 多智能体协作系统
例如:
•Research Agent(检索资料)
•Writer Agent(生成内容)
•Reviewer Agent(审核与反馈)
•Tool Agent(调用外部服务)
这些Agent 在同一个图中共享状态、顺序执行、条件流转。
2. 复杂 AI 工作流
典型流程示例:
用户输入
→ 知识检索
→ 初步推理
→ 生成草稿
→ 审核判断
→ 通过 → 输出
→ 不通过 → 返回修改
这种流程天然是图结构,而不是线性结构。
3. 构建 “可控” 的 AI 应用
LangGraph 可以强制 AI 系统遵循:
•明确步骤
•严格顺序
•条件分支
•循环与终止条件
•工具调用时机
非常适合生产级AI 系统。
四、LangGraph 的核心概念
1. Node(节点)
Node 是图中的基本执行单元,通常代表:
•一次LLM 调用
•一次工具调用
•一次判断逻辑
|
def write_node(state): draft = llm.invoke(f"围绕主题写一篇文章:{state['topic']}") return {"draft": draft.content} |
2. Edge(边)
Edge 定义节点之间的执行关系与流转条件。
|
graph.add_edge("writer", "reviewer") graph.add_edge("reviewer", "writer", condition=needs_revision) |
3. State(全局状态)
State 是 Graph 内所有节点共享的数据结构。
|
{ "topic": "...", "draft": "...", "review": "...", "approved": false } |
每个节点都可以读取或写入State 的一部分。
4. Graph(图)
Graph 是整个 AI 工作流的核心控制结构。
|
graph = StateGraph(State) graph.add_node("writer", write_node) graph.add_node("reviewer", review_node) |
五、LangGraph 最简可运行示例
1. 安装
|
pip install langgraph langchain-openai |
2. 定义状态结构
|
from typing import TypedDict class State(TypedDict): question: str answer: str |
3. 定义节点
|
def answer_node(state: State): from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") result = llm.invoke(state["question"]) return {"answer": result.content} |
4. 构建并编译图
|
from langgraph.graph import StateGraph graph = StateGraph(State) graph.add_node("answer", answer_node) graph.set_entry_point("answer") app = graph.compile() |
5. 执行
|
app.invoke({"question": "什么是 LangGraph?"}) |
六、LangChain 与 LangGraph 的关系
简化对比如下:
|
维度 |
LangChain |
LangGraph |
|
定位 |
LLM 能力抽象层 |
AI 工作流控制层 |
|
核心结构 |
Chain |
Graph |
|
执行方式 |
线性 |
非线性(分支/ 循环) |
|
状态管理 |
可选 |
强制内建 |
|
多Agent |
支持但弱 |
原生支持 |
一句话总结:
LangChain 负责 “能力”,LangGraph 负责 “秩序”。
七、实用案例:多Agent 写作与审核工作流
目标
构建一个AI 写作系统,具备以下能力:
1.写作Agent 生成初稿
2.审核Agent 判断质量
3.不通过则返回修改
4.通过则输出最终内容
状态定义
|
class State(TypedDict): topic: str draft: str approved: bool |
写作节点
|
def writer(state: State): llm = ChatOpenAI() draft = llm.invoke(f"围绕主题写文章:{state['topic']}") return {"draft": draft.content} |
审核节点
|
def reviewer(state: State): llm = ChatOpenAI() result = llm.invoke(f"判断文章是否合格:{state['draft']}") approved = "合格" in result.content return {"approved": approved} |
构建流程图
|
graph = StateGraph(State) graph.add_node("writer", writer) graph.add_node("reviewer", reviewer) graph.set_entry_point("writer") graph.add_edge("writer", "reviewer") graph.add_edge("reviewer", "writer", condition=lambda s: not s["approved"]) |
执行
|
app = graph.compile() app.invoke({"topic": "LangGraph 的工程价值"}) |
这个案例已经具备真实生产AI 工作流的雏形。
结语
LangGraph 并不是 LangChain 的替代品,而是它的结构补全。
当AI 应用进入以下阶段:
•多Agent
•长流程
•强控制
•可回溯
•可演进
LangGraph 几乎是必然选择。
从0到1打造一款具备Ai聊天,AI写作,文生图,语音合成,语音识别功能的多模态全栈项目,多模态AI项目开发链接

更多推荐



所有评论(0)