从图论到 LangGraph
随着大模型能力的爆发,整个行业都在做同一件事:把LLM这个“通用大脑”装进工程系统,让更多业务流程实现自动化、智能化。
于是市面上涌现出了形形色色的Agent框架:有的主打角色化协作,有的主打对话式交互,有的主打低代码搭建。很多人追着新框架学,却很少有人停下来想一个问题:这些框架的底层,到底有没有统一的逻辑?
答案是有。所有LLM Agent框架,本质上都在做同一件事——用工程化的方法管理大模型的流程与状态。而这套方法,早在70年前的图灵时代就已经有了成熟的解决方案:有限状态机 + 有向图。
当年它们是为了解决工业自动化里「系统元素关系混乱、状态不可控、流程不可追溯」的问题;今天LLM Agent遇到的「流程乱跑、状态零散、无法上生产」,本质是同一个痛点的新场景再现。
一、从Chain到Graph:LangChain为什么一定要进化出图结构
在LangGraph之前,LangChain的核心范式是「链(Chain)」,也就是用管道式的线性结构串联大模型、提示词、解析器。
但做复杂Agent的时候,线性链的短板暴露无遗:
- 表达不了循环:工具调用、多轮推理本质就是“思考→执行→再思考”的循环,线性链天生做不到
- 表达不了分支:大模型判断“需要查资料就调用工具,不需要就直接回答”这类条件逻辑,线性链很难实现
- 状态不可控:数据顺着链路单向传递,中间状态无法持久化,做不了断点续跑、人工介入
而「图(Graph)」结构天然就解决了这些问题——它就是为了描述节点之间的任意连接关系而生的。从链式结构升级为图式结构,本质是LangChain从“线性管道工具”向“通用流程编排引擎”的跃迁,也让它第一次从设计上完全对齐了状态机与图论的经典模型。
这也是它和CrewAI、AutoGen等框架最核心的区别:
- CrewAI、AutoGen走的是上层封装路线,把图结构藏在“角色、对话、任务”的业务概念里,上手快但可控性弱
- LangGraph走的是底层编排路线,直接把状态机的原生能力暴露给开发者,自由度更高、流程更可控,更适合生产级复杂系统
二、LangGraph五要素:状态机+有向图的工程化落地
很多人学LangGraph,死记硬背“状态、节点、边、编译、执行”五个步骤,但其实这五件套不是官方凭空规定的,而是「静态结构定义 + 动态运行驱动」的必然拆分。
- 静态结构三要素(状态、节点、边):定义系统长什么样,解决“元素之间是什么关系”的问题
- 动态运行两步骤(编译、执行):让系统从蓝图变成运行中的程序,解决“怎么让系统动起来”的问题
动静结合,就构成了一个完整的、可控的智能体系统。
(一)静态结构:定义智能体的骨架
1. State(状态):全局共享的数据底座
解决的痛点:在普通函数调用或者线性链里,数据是顺着链路往下透传的。流程短还好,一旦节点多了、分支多了,就会变成“参数地狱”——后面的节点要用到前面的数据,就得一路传十几个参数,改一个字段全链路都要改。更麻烦的是,没有统一状态就做不了持久化、断点续跑,流程一断所有中间数据全丢。
本质是什么:
State是整个工作流的全局共享数据层,是所有节点的唯一数据源。它承载了三类核心信息:
- 对话上下文:也就是Agent的记忆,比如用户的历史提问、大模型的历史回复
- 中间计算结果:比如检索到的文档、工具返回的结果、大模型生成的中间数据
- 业务控制字段:比如当前执行到哪一步、是否需要人工介入、循环次数计数
所有节点只从State里读取数据,执行完成后只返回State的增量更新,不用关心数据要传给谁。
标准写法:
from typing import TypedDict
from langgraph.graph import StateGraph, END
# 定义状态结构:明确整个工作流共享哪些数据
class AgentState(TypedDict):
user_input: str # 用户输入
intermediate_result: str # 中间处理结果
final_answer: str # 最终答案
反证:如果没有统一State
你写的就是普通Python脚本,参数靠手动传递,状态靠全局变量维护,复杂流程很快就会失控,也根本做不了持久化和回溯。
2. Node(节点):业务逻辑的最小执行单元
解决的痛点:如果把业务逻辑和流程控制写在一起,想调整执行顺序、增加一个分支,就得改业务代码本身;框架也没法统一做日志、重试、超时控制这些通用能力。
本质是什么:
Node是业务逻辑的标准化封装,它遵守一个极简约定:输入是当前的完整State,输出是State的更新字典(只返回要修改的字段,不是完整State)。
- 节点只关心自己的业务逻辑:是调用大模型、执行工具,还是做数据处理,完全由开发者决定
- 节点之间完全解耦,可以单独测试、单独替换、重复复用
- 框架可以在节点层统一注入通用能力:错误重试、Token统计、日志埋点,不用侵入业务代码
标准写法:
# 定义一个节点:处理用户输入,生成中间结果
def process_input(state: AgentState) -> dict:
# 只从state里读数据
user_input = state["user_input"]
# 执行业务逻辑(这里可以替换成大模型调用、工具执行等)
processed = f"已处理输入:{user_input}"
# 只返回状态的增量更新
return {"intermediate_result": processed}
# 定义另一个节点:生成最终答案
def generate_answer(state: AgentState) -> dict:
result = state["intermediate_result"]
final = f"最终结论:{result},任务完成。"
return {"final_answer": final}
反证:如果没有节点抽象
流程和业务代码完全耦合,改一步动全身,所谓的“编排”就成了空话,框架也就失去了存在的意义。
3. Edge(边):状态流转的规则
解决的痛点:线性结构只能按固定顺序执行,表达不了Agent最核心的两个场景:条件分支(满足条件走A路径,不满足走B路径)和循环(反复执行某个步骤直到达成目标)。
本质是什么:
Edge是节点之间的流转规则,定义了“执行完当前节点,下一步该去哪”。
- 普通边:固定指向下一个节点,对应线性顺序执行
- 条件边:根据当前State里的字段动态判断下一个节点,对应分支逻辑
- 循环的本质:就是一条条件边指回了前面的节点——比如工具执行完,判断任务没完成就回到大模型思考节点
它是图结构的灵魂,也是LangGraph比线性链强大的根本原因。
标准写法:
# 初始化图,绑定状态结构
graph = StateGraph(AgentState)
# 向图中添加节点
graph.add_node("process", process_input)
graph.add_node("answer", generate_answer)
# 设置图的入口节点(从哪里开始)
graph.set_entry_point("process")
# 添加普通边:process执行完,就执行answer
graph.add_edge("process", "answer")
# 设置终止节点:answer执行完,流程结束
graph.add_edge("answer", END)
反证:如果没有边
流程只能是固定的线性顺序,做不了分支、做不了循环,根本实现不了真正的智能体。
(二)动态运行:让静态的图真正跑起来
4. Compile(编译):从蓝图到可执行实例
解决的痛点:你用add_node、add_edge写出来的只是一张“设计蓝图”,描述了图的拓扑结构,但还不能直接运行。如果每次执行都要重新解析结构、校验合法性,既低效,也没法统一注入通用能力。
本质是什么:
Compile是定义期和运行期的分界线,它做了四件核心的事:
- 拓扑校验:检查有没有孤立节点、有没有无法到达的终点、有没有死循环风险,提前发现结构错误
- 结构优化:整理执行顺序,合并可并行节点,优化执行效率
- 能力注入:挂载持久化、中间件、人工介入、回调钩子等运行时能力
- 生成引擎:产出一个可执行的运行时实例,后续直接调用即可
编译之后,你得到的就不再是一堆零散的节点和边,而是一个整装待发的完整程序。
标准写法:
# 编译图,生成可执行的智能体实例
agent = graph.compile()
反证:如果没有编译步骤
每次调用都要重新解析图结构,性能差、错误发现晚,通用能力也只能侵入到每个节点里去写。
5. Invoke(执行):驱动状态机运转
解决的痛点:编译好的图是一个静态的可执行对象,需要一个统一的入口来接收初始参数、驱动整个流程运转、管理执行生命周期、返回最终结果。
本质是什么:
Invoke是状态机的启动触发器,它负责完整的执行生命周期:
- 接收初始输入,初始化第一个State
- 按照编译好的拓扑结构,从入口节点开始一步步执行节点、判断边、更新状态
- 遇到终止节点(END)时自动停止
- 返回最终的完整State结果
除了同步调用的invoke,它还有流式输出(stream)、异步调用(ainvoke)等变体,本质只是执行模式不同,核心都是驱动状态流转。
标准写法:
# 传入初始状态,启动执行
result = agent.invoke({"user_input": "我想了解LangGraph的核心组件"})
# 打印最终结果
print(result["final_answer"])
反证:如果没有统一执行入口
你就得自己写循环去遍历节点、判断边、更新状态,相当于自己手写了一遍图执行引擎。
三、最小完整实战
把上面五部分拼起来,就是一个可直接运行的最小工作流:
from typing import TypedDict
from langgraph.graph import StateGraph, END
# 1. 定义状态
class AgentState(TypedDict):
user_input: str
final_answer: str
# 2. 定义节点
def echo_node(state: AgentState) -> dict:
"""简单的回显节点:把用户输入包装成最终答案"""
return {"final_answer": f"收到你的消息:{state['user_input']}"}
# 3. 构建图:添加节点、连接边
graph = StateGraph(AgentState)
graph.add_node("echo", echo_node)
graph.set_entry_point("echo")
graph.add_edge("echo", END)
# 4. 编译
agent = graph.compile()
# 5. 执行
result = agent.invoke({"user_input": "Hello LangGraph"})
print(result["final_answer"])
更多推荐


所有评论(0)