本文详细介绍了LangGraph框架的核心概念,包括状态图、节点、边和规约等,并通过Graph API和Functional API两种方式展示了LangGraph的使用方法。文章还涵盖了并行处理、断点恢复、调试与防死循环等实用技巧,帮助初学者快速掌握大模型编程,适合想要学习大模型开发的程序员收藏。
一、编排范式对比
| 范式 | 写法 | 能力边界 | 适用 |
|---|---|---|---|
| 线性 Chain | LCEL管道符 / | 固定顺序 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
官方提供两种 API:Graph 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(checkpointer=checkpointer) 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怎么决策。
- 定义State
from 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与Edge
from 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(checkpointer=MemorySaver()) 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大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目,无论你是小白还是有些技术基础的技术人员,这份资料都绝对能帮助你提升薪资待遇,转行大模型岗位。