从线性 Chain 到状态图：LangGraph 核心概念与极简实战

从线性 Chain 到状态图LangGraph 核心概念与极简实战 适用人群熟悉 LangChain/LLM 基础希望构建带状态、可循环、可人工干预的生产级 Agent 工作流的开发者。 为什么传统 Chain 不够用了LangChain 的Chain本质是线性管道适合单步 RAG 或固定流程。但真实业务中常遇到工具调用失败需要重试/降级生成内容需人工审核后再继续多模型/多任务需动态路由LangGraph 用有向图 状态机思想重构工作流让 Agent 具备可循环、可中断、可回溯的能力。 核心概念一图看懂概念作用类比State全局共享的记忆本节点间传递数据游戏存档Node执行单元LLM/工具/逻辑判断关卡机关Edge固定或条件流转路径传送门/岔路Graph节点边构成的完整工作流地图Checkpoint自动持久化状态支持中断恢复自动存档点否是STARTLLM生成校验通过?重试/修正END 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 不是“可选项”而是“必选项”。