LangGraph 状态机驱动的复合 AI 系统设计模式
如果说 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 并不是简单地把函数穿起来,它在底层为你自动组装并打包了以下工业级红利:
- 内置并发支持: 如果图中有多个并列节点(没有边互相连接,但同时依赖上游),LangGraph 会在底层的线程池中自发并行执行它们,无需你写任何一行
asyncio.gather。 - 检查点挂载点: 编译时可以传入
checkpointer选项,让图天然具备了状态的内存/持久化落盘能力,这是迈向“长跑型智能体”和“时间旅行”的绝对大门。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)