三亩地 三亩地SAN MU DI · CODE DIARY
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天,欢迎挑你感兴趣的翻一翻。

LangGraph核心:节点、边、状态的第一性原理

LangGraph核心:节点、边、状态的第一性原理

同一个 if/else,写进代码和画进一张图,区别在哪?LangGraph 把智能体逻辑还原成三个原语:节点=函数、边=条件判断、状态=共享内存,文末附已实跑验证的最小实现。

§1 为什么需要图:线性 Chain 表达不了「循环」

同一个 if/else,写进代码里和画进一张图里,有什么区别?

前者是一次性判断。后者是一条可循环、可回放、可在中途插手的路径。

你的 Agent 是调一次 LLM 就结束,还是自己决定要不要再来一次?

智能体的核心动作是循环推理。调 LLM,看结果,决定继续还是结束。LCEL 是线性 DAG,只能顺序执行 A→B→C,表达不了循环与分支。

  • 节点是操作
  • 是跳转规则
  • 状态是共享内存

代价:多轮对话一长,状态散落在外层变量里。用户三连问,每轮都要带上上一轮结论,状态一丢就得重说一遍。

官方把旧的 Agent Executor 重构到图上,正是看中图能表达的循环结构。LangGraph 用一张图解决。图里只有三个原语,加一个 StateGraph 容器就是全部。它也是 LangChain 生态里的独立模块。


§2 节点 Node:最小操作单元 = 函数

2.1 一个函数,一个节点

节点是一个 Python 函数。它接收 state 字典,返回部分更新字典,只需要写自己要改的键。

约束有两条。输入输出 schema 必须与 State 对齐。节点本身没有内部状态,所有记忆都在外部 State 里。

官方文档的最小节点长这样,函数名就是图里的节点名:

from typing_extensions import TypedDict class State(TypedDict): input: str results: str def node_with_runtime(state: State): # 只返回要改的键,不必拼整份 state return {"results": f"Hello, {state['input']}!"}

「只返回要改的键」是 LangGraph 的核心约定。图引擎会把返回值合并进共享状态,各节点不需要知道完整状态长什么样。

CSDN 实战示例把「部分更新」讲得更直白。一个 double_it 节点,只更新自己管的两个键,其余不动:

def double_it(state: dict) -> dict: value = state.get("value", 0) * 2 history = state.get("history", []) + [value] return {"value": value, "history": history} # 只更新两个键,其余不动

同步异步都支持。函数用 async def 定义,LangGraph 会在事件循环里调度它,写法上只差一个关键字。

2.2 为什么节点必须无状态

节点无状态是刻意设计。同一个函数在任何上下文里跑,行为都一样,图引擎因此能持久化、能回放任意一步。

注册方式有讲究。add_node 的两个参数,角色完全不同:

  • 第一个参数是字符串引用名,加边时用
  • 第二个参数才是函数本体,路由时不直接参与

名字错了,编译期就报错。这层间接很有用:想换实现,只改 add_node 绑定的函数,边和路由都不用动。贯穿示例里的 get_weather 也是这种函数,只返回 {“weather”: …} 一个键,§5 正式亮相。


§3 边 Edge:跳转规则 = 条件判断

3.1 两类边

边决定执行顺序,分两类。无条件边 add_edge(A, B),A 完成必去 B。条件边 add_conditional_edges 让 router 读 state、返回节点名,可指回上游。

START 是入口,END 是出口。END 不是真实节点,图走到 END 才停。


图3 条件边按 router 的返回值选路,node_c 可指回 node_a 形成循环。

条件边的第三个参数是映射表。router 的返回值是键,映射表告诉引擎该去哪。官方示例 {True: “node_b”, False: “node_c”},返回值与节点名解耦。

条件边也能从 START 出发挂,图一进来就先路由。

3.2 循环是怎么转起来的

ReAct 循环是典型用法。模型节点读完 state 后,router 决定继续调工具还是结束:

def should_continue(state) -> Literal["tool_node", END]: # router 读 state:有工具调用就继续,否则结束 if state["messages"][-1].tool_calls: return "tool_node" return END # 条件边:按 router 返回值映射到不同节点 builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END]) # 普通回边:工具执行完回到 LLM,循环由此形成 builder.add_edge("tool_node", "llm_call")

循环的实现拆开看。条件边负责选路,真正让图转起来的是一条指回上游的普通边。反复几轮直到模型不再要求调工具,条件边返回 END。

边把「if 结果正确就下一步,否则重试」从代码嵌套变成图的拓扑。

juejin 的示例把这条回边写得最清楚:tool_node 执行完工具,控制权交回模型,模型再走条件边决定下一步。

映射表本身也很灵活:

  • 可以省略,router 直接返回节点名
  • 返回值可以是节点名列表,多个分支并行执行

两类边一句话区分:

  • 无条件边add_edge(A, B):A 完成必去 B
  • 条件边add_conditional_edges:router 读 state 选路,可指回上游

§4 状态 State:全局数据 = 共享内存

4.1 共享 TypedDict 与 reducer

状态是贯穿全图的共享 TypedDict,节点靠它交互,彼此不直接调用。默认覆盖旧值,用 Annotated[list, operator.add] 做累加。

对话历史、中间结果这类数据,靠的就是累加。每次都覆盖,多轮上下文就只剩最后一轮。

State 的 schema 不只 TypedDict 一种。dataclass、Pydantic BaseModel 都可以,约束是节点收发都对齐这份 schema。


图4 两个节点只和 State 交互,彼此不直接互调。

reducer 的覆盖与累加,官方文档给过一段对比,已在本机实跑验证:

from operator import add from typing import Annotated from typing_extensions import TypedDict class State(TypedDict): foo: int bar: Annotated[list[str], add] # 累加而非覆盖 # 输入 {"foo":1,"bar":["hi"]} # 节点1 返回 {"foo":2} → 状态 {"foo":2,"bar":["hi"]} # 节点2 返回 {"bar":["bye"]} → 状态 {"foo":2,"bar":["hi","bye"]}

foo 被第二次写入覆盖成 2,bar 靠 operator.add 累加成两段。

4.2 消息场景用 add_messages

消息类状态直接用 add_messages,它按消息 ID 覆盖或追加。裸 operator.add 会把人工更新也误追加进去,这是官方文档特别提醒过的坑。

from langchain_core.messages import AnyMessage from langgraph.graph.message import add_messages class GraphState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] # 按 ID 覆盖或追加

理解 reducer 的关键在「谁决定合并」。声明了 reducer 的键,每次写入都先经过 reducer 再落进状态。合并策略是状态 schema 的一部分,跟着类型走,不散落在节点里。

reducer 决定键的合并方式:默认覆盖,Annotated 累加。

reducer 不只有 operator.add 和 add_messages。它就是一个普通函数,输入旧值和新值,输出合并结果。想自定义合并规则,写个函数声明上去就行,机制不变。

状态外置带来三个能力:

  • 可持久化:checkpointer 按 thread_id 存状态
  • 可回放:同一 thread_id 再调一次,上下文自动续上
  • 可人在回路:运行到断点停下等人介入

这是「共享内存」能累积上下文的关键,也是 LangGraph 相对手写循环的根本差异。


§5 贯穿示例:查天气 → 推荐穿搭

三个原语单独看都简单,合在一起才见功力。输入城市,输出天气和穿搭建议,把「图结构 ↔ 代码」双向映射讲透。


图5 查天气到推荐穿搭:两个节点、三条边、一个共享字段。

from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class OutfitState(TypedDict): # 状态=共享内存:三个字段 city: str weather: str recommendation: str def get_weather(state: OutfitState) -> dict: # 节点=函数 mock = {"北京": "晴,28°C", "上海": "多云,24°C", "广州": "雷阵雨,30°C"} return {"weather": mock.get(state["city"], "晴,25°C")} # 部分更新 def recommend_outfit(state: OutfitState) -> dict: w = state["weather"] # 读共享内存,不直接调 get_weather if "晴" in w: return {"recommendation": "短袖 + 防晒,紫外线强记得戴帽子"} if "雨" in w: return {"recommendation": "薄外套 + 随身雨具"} return {"recommendation": "薄外套,早晚温差带件方便穿脱的"} builder = StateGraph(OutfitState) # 图容器 builder.add_node("get_weather", get_weather) # 框 = add_node builder.add_node("recommend_outfit", recommend_outfit) builder.add_edge(START, "get_weather") # 箭头 = add_edge builder.add_edge("get_weather", "recommend_outfit") builder.add_edge("recommend_outfit", END) graph = builder.compile() # 必须编译后才能调用 result = graph.invoke({"city": "北京"}) print(result["weather"], "→", result["recommendation"])
  • 图里每个= 一个 add_node
  • 每条箭头= 一条 add_edge
  • weather字段= State 共享内存

get_weather 只写 weather,recommend_outfit 只读 weather,两者没有任何直接调用,信息全部经状态中转。这就是「共享内存」四个字的落点。

compile 做结构校验,缺边、不可达节点在这里就报错。invoke 传入初始状态 {“city”: “北京”},图引擎按边一路执行,返回最终状态。

晴,28°C → 短袖 + 防晒,紫外线强记得戴帽子

这段代码已在本机实跑验证(langgraph 1.2.10,exit 0),输出与预期逐字一致。

反向读也一样成立。拿到一段 LangGraph 代码,看 add_node 就是图里的框,看 add_edge 就是箭头,看 State 的字段就是共享数据。图能导出成 mermaid,代码能翻译回图,双向映射闭环。


§6 进阶:用条件边加「天气异常 → 人工确认」分支

给示例加一个分支:天气异常时走人工确认。在 recommend_outfit 后接条件边,router 读 weather 判断是否含「雨」。


图6 条件边在 recommend_outfit 后分叉,异常进人工确认,正常直达 END。

下面这段是演示变体,在 §5 脚手架上追加,已在本机实跑验证:

from typing import Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class OutfitState(TypedDict): city: str weather: str recommendation: str needs_confirm: bool # 是否需人工确认 def human_confirm(state: OutfitState) -> dict: return {"needs_confirm": True} def route(state: OutfitState) -> Literal["human_confirm", END]: # 含"雨"视为天气异常,需要人工确认 if "雨" in state["weather"]: return "human_confirm" return END builder.add_node("human_confirm", human_confirm) builder.add_conditional_edges( "recommend_outfit", route, {"human_confirm": "human_confirm", END: END}, ) builder.add_edge("human_confirm", END)

预期行为:北京「晴」直通 END,广州「雷阵雨」进 human_confirm。

  • 条件边把「要不要人看」变成图的拓扑,加分支不动已有节点
  • 纪律:循环和分支必须最终能结束,否则触发 recursion_limit

recursion_limit 是可以调大的,但调大只是拖延,不解决「循环没有出口」的设计缺陷。

调试建议:先用小 recursion_limit 快速暴露死循环,比一次跑满省时间。

往真实系统想一步:human_confirm 可以是发给运维的审批消息,节点停下来等人工回复,再带着回复继续走。人机协作不是另外一套机制,只是图里多了一个普通节点和一条条件边。


§7 边界与常见错误:什么时候别用图

compile 会做结构校验。不可达节点、缺 START 入口边、缺 END 出口边、节点名拼写不一致,都在编译期报错,不会拖到运行期才炸。


图7 正确骨架要求 START 入口边与 END 出口边齐全,缺一条都过不了 compile。

  • 节点名拼写不一致(“retreive” vs “retrieve”)→ 编译期报错
  • 漏了 START 或 END 边→ 编译期报错
  • 条件边里写死永真分支→ 循环不停,触发 recursion_limit

判断标准一句话:你的流程会不会回头。不会回头,线性链够用;会回头、要留状态、要人在中途插手,再上 LangGraph。

单次问答、无状态转换、纯顺序管道,LCEL 更轻。图的价值在「要循环、要共享状态、要中途介入」。compile 之后的图还能导出 mermaid,把代码画回图,调试和文档都靠它。

最小图的四步法,顺序固定:

  • 定义 State
  • add_node 注册节点
  • add_edge 连边、指定 START 入口
  • 最后 compile

漏一步都会在编译期暴露。


§8 最小实现

把上面的查天气图原样复制成一份独立脚本,存成 outfit.py,装好依赖就能跑。

pip install langgraph python outfit.py
from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END class OutfitState(TypedDict): city: str weather: str recommendation: str def get_weather(state: OutfitState) -> dict: mock = {"北京": "晴,28°C", "上海": "多云,24°C", "广州": "雷阵雨,30°C"} return {"weather": mock.get(state["city"], "晴,25°C")} def recommend_outfit(state: OutfitState) -> dict: w = state["weather"] if "晴" in w: return {"recommendation": "短袖 + 防晒,紫外线强记得戴帽子"} if "雨" in w: return {"recommendation": "薄外套 + 随身雨具"} return {"recommendation": "薄外套,早晚温差带件方便穿脱的"} builder = StateGraph(OutfitState) builder.add_node("get_weather", get_weather) builder.add_node("recommend_outfit", recommend_outfit) builder.add_edge(START, "get_weather") builder.add_edge("get_weather", "recommend_outfit") builder.add_edge("recommend_outfit", END) graph = builder.compile() result = graph.invoke({"city": "北京"}) print(result["weather"], "→", result["recommendation"])

预期输出:

晴,28°C → 短袖 + 防晒,紫外线强记得戴帽子

本示例已在 langgraph 1.2.10 下实跑验证,退出码 0。换上海输出「多云,24°C → 薄外套,早晚温差带件方便穿脱的」,换广州输出「雷阵雨,30°C → 薄外套 + 随身雨具」。


§9 总结:三原语收口

  • 节点是函数,图里最小的操作单元。
  • 是条件判断,决定下一步去哪,循环靠它实现。
  • 状态是共享内存,跨节点传数据,reducer 决定覆盖还是累加。
  • 起步建议:先跑通最小实现,再加条件边,再考虑持久化。

「图结构 ↔ 代码」的双向映射是 LangGraph 的心智模型。画图时想到 add_node 和 add_edge,写代码时想到节点和箭头。

学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%免费

← 返回列表