从线性 Chain 到状态图:LangGraph 核心概念与极简实战
💡 适用人群:熟悉 LangChain/LLM 基础,希望构建带状态、可循环、可人工干预的生产级 Agent 工作流的开发者。
🔍 为什么传统 Chain 不够用了?
LangChain 的Chain本质是线性管道,适合单步 RAG 或固定流程。但真实业务中常遇到:
- 工具调用失败需要重试/降级
- 生成内容需人工审核后再继续
- 多模型/多任务需动态路由
LangGraph 用有向图 + 状态机思想重构工作流,让 Agent 具备可循环、可中断、可回溯的能力。
🧩 核心概念(一图看懂)
| 概念 | 作用 | 类比 |
|---|---|---|
State | 全局共享的记忆本,节点间传递数据 | 游戏存档 |
Node | 执行单元(LLM/工具/逻辑判断) | 关卡机关 |
Edge | 固定或条件流转路径 | 传送门/岔路 |
Graph | 节点+边构成的完整工作流 | 地图 |
Checkpoint | 自动持久化状态,支持中断恢复 | 自动存档点 |
🛠 5 分钟跑通最小示例
pip install langgraph langchain-openaifrom typing import TypedDict from langgraph.graph import StateGraph, START, END # 1. 定义状态 class WorkflowState(TypedDict): query: str response: str retries: int # 2. 定义节点 def llm_node(state: WorkflowState): # 此处替换为真实 LLM 调用 return {"response": f"已处理: {state['query']}"} def retry_node(state: WorkflowState): return {"retries": state["retries"] + 1} # 3. 构建图 builder = StateGraph(WorkflowState) builder.add_node("llm", llm_node) builder.add_node("retry", retry_node) builder.add_edge(START, "llm") # 条件边:控制循环与退出 def route(state: WorkflowState) -> str: # 模拟校验逻辑:重试超过2次则结束 if state.get("retries", 0) >= 2 or "错误" not in state.get("response", ""): return "end" return "retry" builder.add_conditional_edges("llm", route, {"retry": "retry", "end": END}) builder.add_edge("retry", "llm") # 4. 编译并运行 graph = builder.compile() result = graph.invoke({"query": "测试输入", "retries": 0}) print(result)🎯 核心目的与选型逻辑
📝 本篇博客写作目的
- 破除线性思维:帮开发者从
Chain的“单步管道”认知,升级到Graph的“状态机+循环”架构。 - 提供即插即用模板:不堆砌理论,直接给出可运行的最小示例与生产级模式(重试/人工干预/多路由)。
- 明确技术边界:用对比表格划清 LangGraph 的适用场景,避免“杀鸡用牛刀”或“该用时不用”。
🛠 技术实战目的(为什么选 LangGraph?)
| 业务诉求 | LangGraph 解决目的 | 关键实现 |
|---|---|---|
| 流程可循环 | 打破线性限制,支持自我修正/重试 | 条件边 + 计数器 + 安全退出机制 |
| 过程可干预 | 关键节点人工审核/确认后再继续 | interrupt_before+ 状态持久化 |
| 状态可追溯 | 复杂 Agent 的上下文不丢失、可回放 | 节点仅返回增量更新,自动合并 State |
| 路由可动态 | 根据 LLM 输出/工具结果智能分流 | 动态字符串路由至不同子图或END |
💡核心结论:当你的工作流需要确定性控制、状态可观测、异常可恢复时,LangGraph 不是“可选项”,而是“必选项”。