随着大模型能力的爆发,整个行业都在做同一件事:把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是定义期和运行期的分界线,它做了四件核心的事:

  1. 拓扑校验:检查有没有孤立节点、有没有无法到达的终点、有没有死循环风险,提前发现结构错误
  2. 结构优化:整理执行顺序,合并可并行节点,优化执行效率
  3. 能力注入:挂载持久化、中间件、人工介入、回调钩子等运行时能力
  4. 生成引擎:产出一个可执行的运行时实例,后续直接调用即可

编译之后,你得到的就不再是一堆零散的节点和边,而是一个整装待发的完整程序。

标准写法

# 编译图,生成可执行的智能体实例
agent = graph.compile()

反证:如果没有编译步骤
每次调用都要重新解析图结构,性能差、错误发现晚,通用能力也只能侵入到每个节点里去写。

5. Invoke(执行):驱动状态机运转

解决的痛点:编译好的图是一个静态的可执行对象,需要一个统一的入口来接收初始参数、驱动整个流程运转、管理执行生命周期、返回最终结果。

本质是什么
Invoke是状态机的启动触发器,它负责完整的执行生命周期:

  1. 接收初始输入,初始化第一个State
  2. 按照编译好的拓扑结构,从入口节点开始一步步执行节点、判断边、更新状态
  3. 遇到终止节点(END)时自动停止
  4. 返回最终的完整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"])
Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐