工作流编排: LangGraph状态机-北京尧图网络

【摘要】编排式范式对比, LangGraph概念, 官方API, 使用方法和注意事项一、编排范式对比范式写法能力边界适用线性 ChainLCEL管道符 \固定顺序 A→B→C有状态图StateGraph节点条件边能循环/分支 / 多路检索→判断→回答、Agent 循环并行RunnableParallel多路同时触发结果汇聚多源检索、多模型对比判断标准流程里如果要根据中间结果决定下一步用图就比较适合。三步以内的固定流程使用 LCEL 管道符更适合用图会增加复杂度。二、LangGraph概念LangGraph 是用于构建可循环、可记忆工作流的框架它解决了 LangChain 中链式调用和状态管理的痛点。其核心是状态图(StateGraph), 通过节点(Nodes)‍和状态流转实现循环与持久化状态。LangGraph 把工作流抽象成一张有向图本质是状态驱动有向图, 有三个核心和一个规约State: 状态, 整张图的共享上下文相当于图的内存每一步节点读写都操作这个状态, 保存对话历史、工具返回值、中间结果等。它是不可变的, 每一步执行完不会去改原来的状态而是生成一份全新的状态快照。状态不可变带来的好处, 能快照/回放, 多节点并行跑完能确定性地合并结果。Node: 节点, 图里面一个个业务处理函数做实际计算逻辑干动作, 一个节点只负责一件事(大模型调用、工具调用、解析输出、校验逻辑等)。节点函数不改变状态只返回一个状态增量字典 {key:value}交给框架合并到全局 State。Edge: 边, 节点之间的连线, 定义工作流的方向和流转路径, 分为普通边(固定跳转)和条件边(根据当前state判断动态选择下一个节点)。Reducer: 规约, 负责合并。属于State的内部机制, 定义同一个字段, 新旧更新如何合并。Node 返回增量字典框架不会直接赋值, 对每一个 key 调用该字段绑定的 reducer, 产出新state值。newState[key] reducer(oldState[key], value)。编译后的图内部是一个事件循环(Scheduler)取当前节点 → 执行它拿到增量 → 用 Reducer 合并成新状态 → 看它的出边(普通边走下一个节点,条件边调用路由函数决定走哪条边, 走哪个节点→ 把下一个节点入队 → 循环直到走到 END节点。简单理解State是一块共享白板每个Node工作完不能直接上去涂改只能说我改了什么。白板上已经有3个旧消息, 这1条消息怎么合并进去, 是直接替换掉旧的3条, 还是追加成第 4 条, 决定这件事的函数就是Reducer。# 结构示例 StateGraph |-- Nodes(节点) |-- START - node_a |-- node_a - node_b(条件边) |-- node_a - node_c(条件边) -- node_b - END三、LangGraph官方API安装 pip install -U langgraph官方提供两种 APIGraph API 与 Functional API。两者共享同一套运行时(持久化、流式、人机协同、记忆都通用)只是编程风格不同可在同一项目混用。维度Graph API声明式Functional API命令式写法先定义节点/边/共享 State再编译成图直接写标准 Python 控制流if/else、循环,用装饰器 task/entrypoint 把函数挂进运行时适合复杂分支、并行汇聚、需要可视化调试、团队协作线性流程、想最小改动旧代码、快速原型、函数内局部状态典型场景多 Agent 编排、检索→判断→回答、需画架构图交付在已有脚本里加 LangGraph 能力, essay 生成这类顺序简单分支Graph API 的官方最小范式# Graph API声明式节点边共享状态结构可视化 from langgraph.graph import StateGraph from typing import TypedDict # state class AgentState(TypedDict): messages: list current_tool: str retry_count: int # edge def should_continue(state): if state[”retry_count”] 3: return ”end” elif state[”current_tool”] ”search”: return ”process_search” else: return ”call_llm” # call_llm_node和search_node是假想的两个node workflow StateGraph(AgentState) workflow.add_node(”call_llm”, call_llm_node) workflow.add_node(”process_search”, search_node) workflow.add_conditional_edges(”call_llm”, should_continue)Functional API 的官方最小范式# Functional API命令式把 LangGraph 能力嵌进普通 Python 函数 from langgraph.func import entrypoint, task task def process_user_input(user_input: str) - dict: return {”processed”: user_input.lower().strip()} task def handle_urgent_request(data: dict) - str: pass task def handle_normal_request(data: dict) - str: pass # 假想checkpointer entrypoint(checkpointercheckpointer) def workflow(user_input: str) - str: processed process_user_input(user_input).result() # .result() 取任务结果 if ”urgent” in processed[”processed”]: # 标准 Python 分支 response handle_urgent_request(processed).result() else: response handle_normal_request(processed).result() return response四、LangGraph使用我们以Graph API 学习, 它的图结构能更清楚的看到Agent怎么决策。定义Statefrom typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 普通字段不写Annotated: 默认覆盖语义(新值替换旧值) current_tool: str retry_count: int # 列表字段用Annotated绑定reducer, 决定”增量怎么并进旧值” messages: Annotated[list, add_messages] # 官方内置add_messages,追加并且按消息ID去重 log: Annotated[list, operator.add] # 列表追加(不去重)当节点返回 {“messages”: [新消息]}时不会直接 state[“messages”] [新消息], 而是调用 add_messages(old_messages, [新消息]), 得到 [旧…, 新]防止历史消息重复堆叠。定义Node与Edgefrom langgraph.graph import StateGraph, START, END # llm.invoke是之前学过的大模型调用方法 def call_llm(state: AgentState) - dict: # 节点只返回增量messages 走 add_messages 追加 return {”messages”: [llm.invoke(state[”messages”])]} # run_search是假想的一个函数 def process_search(state: AgentState) - dict: return {”messages”: [run_search(state)], ”current_tool”: ”search”} # 路由函数返回 END 或节点名框架据此选下一个节点 # should_continue也可写成 add_conditional_edges(call_llm, should_continue, {process_search: process_search, END: END}) 显式映射 # 不传映射时路由返回值须直接是节点名或 END def should_continue(state: AgentState) - str: if state[”retry_count”] 3: return END # 结束 elif state[”current_tool”] ”search”: return ”process_search” else: return ”call_llm” workflow StateGraph(AgentState) workflow.add_node(”call_llm”, call_llm) workflow.add_node(”process_search”, process_search) workflow.add_edge(START, ”call_llm”) # START入口 workflow.add_conditional_edges(”call_llm”, should_continue) # 条件边路由返回 END/节点名 app workflow.compile() # 一次编译即可, 流程中会多次invoke并行图级 Fan-out / Fan-in, 图级多后继workflow.add_edge(START, ”fanout”) workflow.add_edge(”fanout”, ”search_kb”) # 同时出发 workflow.add_edge(”fanout”, ”search_web”) # 同时出发 workflow.add_edge(”search_kb”, ”merge”) # 都完成后才进 merge workflow.add_edge(”search_web”, ”merge”) # 汇聚点会等所有前驱就绪 workflow.add_edge(”merge”, ”decide”) workflow.add_edge(”decide”, END)调度器把 search_kb / search_web 同时入队并发执行merge在所有前驱完成后才触发。断点恢复Checkpointer执行完状态会丢失复杂任务可能执行一半停下来, 下次继续上次接着执行。from langgraph.checkpoint.memory import MemorySaver # MemorySaver内存保存, 持久化可以用SqliteSaver.from_conn_string(”db.sqlite”)或RedisSaver app workflow.compile(checkpointerMemorySaver()) config {”configurable”: {”thread_id”: ”user-1”}} # 隔离不同用户, 不同用户必须不同 id app.invoke({”question”: ”解释一下 Reducer”}, config) print(app.get_state(config).values) # 查看当前状态快照 app.invoke({”question”: ”operator.add 是什么”}, config) # 同一个config会自动带历史调试与防死循环循环必须有上限否则 Agent 陷入调工具→不够→再调死循环, 多少token也不够用# 用recursion_limit防止死循环 app.invoke(state, {”recursion_limit”: 10}) # 超出10步直接抛异常GraphRecursionError # 用draw_mermaid可以查看状态 print(app.get_graph().draw_mermaid()) # 导出mermaid,可以看结构图(用drawio等可以看图) app.get_graph().draw_mermaid_png(output_file_path”workflow.png”) for chunk in app.stream(input, stream_mode”updates”): # 每次只返回变化的节点 print(chunk)学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

工作流编排: LangGraph状态机

相关推荐

昆山瑜伽新手动作跟不上怎么办

综合布线工程师怎么考证？从报名学习到考试拿证，报考全攻略

easy-vibe 前端工程化全景指南：从构建原理到 Vite 实战配置

更多资讯

Graphcool 迁移至 Prisma 实战：Authentication 与 Authorization 架构升级指南

低空无人机消防AI识别系统：架构、模型与边缘部署实践

wired-link 手绘风格链接组件：使用指南与源码实现解析

3个核心维度拆解blm模型源码与最佳实践

Triton Inference Server 快速上手指南：从模型仓库搭建到推理请求发送

PandaMH源码解析：内存读取与SDL覆盖层构建全图调试工具

今日更新

2k显示屏性能优化踩坑：版本升级后API全变了，这份源码解析救了我

3步搞定美眉图实战项目，告别官方文档抓不住重点

3步搞定黄金大劫案项目搭建从入门到精通

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？