如果说 LCEL(LangChain 表达式语言)将大模型应用开发带入了“声明式流水线”时代,那么 LangGraph 的出现,则彻底将开发者从线性思维中解放出来,推向了“复合 AI 系统(Composite AI Systems)”的终极形态。

在 2026 年的今天,任何稍微复杂的商业级 Agent 都不可能只是一条单向流动的线。它需要反复的自我纠错、多步的工具路由、甚至在必要时停下来等待人类的介入。这一切的底层底座,正是 LangGraph。本文将带你剥开 LangGraph 的优雅外衣,直击其最核心的状态机图架构设计模式。
在这里插入图片描述

一、 图架构的三驾马车:State、Nodes 与 Edges

LangGraph 的本质是一个基于全局状态的有向图(Stateful Directed Graph)。要在这个框架里纵横驰骋,你必须精通它的三个核心原语:

1.1 State(集中式状态:图的灵魂)

在传统的 LangChain 链中,数据是像接力棒一样在组件间透传的,上游必须精确知道下游需要什么 Schema。而在 LangGraph 中,所有节点共享一个中央状态机。
状态的数学本质可以抽象为一个显式的状态更新函数:

St+1=δ(St,ΔN)S_{t+1} = \delta(S_t, \Delta N)St+1=δ(St,ΔN)

其中 StS_tSt 是当前图的全局状态,ΔN\Delta NΔN 是某个节点(Node)执行完毕后返回的局部更新字典,δ\deltaδ 则是状态缩减器(Reducer)。

💡 绝招:利用 Reducer 实现多渠道数据追加

默认情况下,新节点返回的 Key 会直接覆盖(Overwrite)旧状态。但通过 Annotated 和自定义 Reducer(如 operator.add),你可以让状态变成一个“只增不减”的滚动日志(如消息列表),这为多轮对话和并行 Agent 的数据合并奠定了基础。

1.2 Nodes(节点:纯粹的执行体)

在 LangGraph 中,节点本质上就是一个普通的 Python 函数(同步或异步皆可)

  • 它接收当前的全局 State 作为唯一输入。
  • 它内部可以调用大模型、读写数据库、甚至拉起另一个子图(Sub-graph)。
  • 它唯一的硬性要求是:必须返回一个 Dict,用以更新全局 State

1.3 Edges(边:控制流的方向盘)

边决定了数据在节点之间的流转路径。LangGraph 提供了两种边:

  • 普通边(Normal Edges): 强绑定关系。例如节点 A 结束后,无条件直接进入节点 B。
  • 条件边(Conditional Edges): 动态路由的核心。它接收一个“路由函数(Router Function)”,该函数根据当前 State 的内容进行推理,动态返回下一个节点的名称。通常,我们把大模型的 Function Calling 结果作为条件边的路由依据。

二、 零基础代码实战:构建一个带“自我纠错”的智能翻译 Agent

下面我们亲手用原生 LangGraph 编写一个“翻译 →\rightarrow 语法审查 →\rightarrow 动态打回重写”的闭环状态机。

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

# 1. 定义图的全局状态
class TranslationState(TypedDict):
    source_text: str
    target_lang: str
    current_translation: str
    review_feedback: str
    is_perfect: bool

# 2. 定义节点 A:翻译执行者
def translate_node(state: TranslationState):
    print("🤖 [Node: Translate] 正在生成初始翻译...")
    # 模拟大模型翻译行为
    return {"current_translation": "Hello, this is an AI agent workflow."}

# 3. 定义节点 B:严苛的语法审查者
def review_node(state: TranslationState):
    print("🧐 [Node: Review] 正在评估翻译质量...")
    # 模拟审查:如果翻译里包含 "agent" 但没包含 "智能体",就打回
    if "智能体" not in state["current_translation"]:
        return {
            "is_perfect": False,
            "review_feedback": "请将 'AI agent' 翻译为更地道的 'AI 智能体',而不是保留英文。"
        }
    return {"is_perfect": True, "review_feedback": "完美符合工业标准。"}

# 4. 定义节点 C:反思重写者
def rewrite_node(state: TranslationState):
    print(f"🔄 [Node: Rewrite] 收到驳回意见:{state['review_feedback']} 正在重构...")
    return {"current_translation": "你好,这是一个 AI 智能体工作流。"}

# 5. 定义条件边的路由算法
def route_after_review(state: TranslationState) -> str:
    if state["is_perfect"]:
        return "end"  # 走向终点
    return "rewrite"  # 回溯重修

# 6. 编排状态图
workflow = StateGraph(TranslationState)

# 添加节点
workflow.add_node("translator", translate_node)
workflow.add_node("reviewer", review_node)
workflow.add_node("rewriter", rewrite_node)

# 设置起点与常规边
workflow.set_entry_point("translator")
workflow.add_edge("translator", "reviewer")
workflow.add_edge("rewriter", "reviewer") # 重写后重新送审

# 设置动态条件边
workflow.add_conditional_edges(
    "reviewer",
    route_after_review,
    {
        "end": END,
        "rewrite": "rewriter"
    }
)

# 编译图
app = workflow.compile()

# 7. 触发运行
if __name__ == "__main__":
    initial_input = {"source_text": "Hello, this is an AI agent workflow.", "target_lang": "zh"}
    final_state = app.invoke(initial_input)
    print(f"\n🎉 [终产物]: {final_state['current_translation']}")


三、 现代编译心法:.compile() 的底牌

当你调用 workflow.compile() 时,LangGraph 并不是简单地把函数穿起来,它在底层为你自动组装并打包了以下工业级红利:

  1. 内置并发支持: 如果图中有多个并列节点(没有边互相连接,但同时依赖上游),LangGraph 会在底层的线程池中自发并行执行它们,无需你写任何一行 asyncio.gather
  2. 检查点挂载点: 编译时可以传入 checkpointer 选项,让图天然具备了状态的内存/持久化落盘能力,这是迈向“长跑型智能体”和“时间旅行”的绝对大门。
    在这里插入图片描述
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐