本文详细介绍了LangGraph框架的核心概念包括状态图、节点、边和规约等并通过Graph API和Functional API两种方式展示了LangGraph的使用方法。文章还涵盖了并行处理、断点恢复、调试与防死循环等实用技巧帮助初学者快速掌握大模型编程适合想要学习大模型开发的程序员收藏。一、编排范式对比范式写法能力边界适用线性 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是假想的两个nodeworkflow 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}) 显式映射# 不传映射时路由返回值须直接是节点名或 ENDdef 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 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取 小白程序员轻松入门大模型：LangGraph核心概念与实践
本文详细介绍了LangGraph框架的核心概念包括状态图、节点、边和规约等并通过Graph API和Functional API两种方式展示了LangGraph的使用方法。文章还涵盖了并行处理、断点恢复、调试与防死循环等实用技巧帮助初学者快速掌握大模型编程适合想要学习大模型开发的程序员收藏。一、编排范式对比范式写法能力边界适用线性 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是假想的两个nodeworkflow 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}) 显式映射# 不传映射时路由返回值须直接是节点名或 ENDdef 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 由于新岗位的生产效率要优于被取代岗位的生产效率所以实际上整个社会的生产效率是提升的。但是具体到个人只能说是“最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线科技企业深耕十二载见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事早已在效率与薪资上形成代际优势我意识到有很多经验和知识值得分享给大家也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套AI 大模型突围资料包✅ 从零到一的 AI 学习路径图✅ 大模型调优实战手册附医疗/金融等大厂真实案例✅ 百度/阿里专家闭门录播课✅ 大模型当下最新行业报告✅ 真实大厂面试真题✅ 2026 最新岗位需求图谱所有资料 ⚡️ 朋友们如果有需要《AI大模型入门进阶学习资源包》下方扫码获取~① 全套AI大模型应用开发视频教程包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点② 大模型系统化学习路线作为学习AI大模型技术的新手方向至关重要。 正确的学习路线可以为你节省时间少走弯路方向不对努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划带你从零基础入门到精通③ 大模型学习书籍文档学习AI大模型离不开书籍文档我精选了一系列大模型技术的书籍和学习文档电子版它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。④ AI大模型最新行业报告2025最新行业报告针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。⑤ 大模型项目实战配套源码学以致用在项目实战中检验和巩固你所学到的知识同时为你找工作就业和职业发展打下坚实的基础。⑥ 大模型大厂面试真题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我精心整理了一份大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。以上资料如何领取为什么大家都在学大模型最近科技巨头英特尔宣布裁员2万人传统岗位不断缩减但AI相关技术岗疯狂扩招有3-5年经验大厂薪资就能给到50K*20薪不出1年“有AI项目经验”将成为投递简历的门槛。风口之下与其像“温水煮青蛙”一样坐等被行业淘汰不如先人一步掌握AI大模型原理应用技术项目实操经验“顺风”翻盘这些资料真的有用吗这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。以上全套大模型资料如何领取