内容阅读

不画图写 LangGraph：@entrypoint、@task 与两种范式的选择

Agent•大约 14 小时前•阿新聊ai

TL;DR: LangGraph 有两套写法：StateGraph 画图，functional API 写普通函数。后者用 @entrypoint 标入口、@task 标步骤，靠「重放 + 跳过已完成 task」实现持久化。选型标准不是新旧，是控制流形态：分支路由复杂、需要可视化调试选图；步骤序列为主、想按普通代码方式推理选函数式。两者可以混用，不是二选一。

同一件事的两种写法

先感受差异。「取订单、生成周报」这个两步任务，StateGraph 版本长这样：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    user_id: str
    orders: list
    report: str

def fetch(state: S) -> dict:
    return {"orders": query_orders(state["user_id"])}

def report(state: S) -> dict:
    return {"report": make_report(state["orders"])}

g = StateGraph(S)
g.add_node("fetch", fetch)
g.add_node("report", report)
g.add_edge(START, "fetch")
g.add_edge("fetch", "report")
g.add_edge("report", END)
graph_app = g.compile()

函数式版本：

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def fetch_orders(user_id: str) -> list:
    return query_orders(user_id)

@task
def make_report(orders: list) -> str:
    return summarize(orders)

@entrypoint(checkpointer=InMemorySaver())
def weekly(user_id: str) -> str:
    orders = fetch_orders(user_id).result()
    return make_report(orders).result()

weekly.invoke("u42", config={"configurable": {"thread_id": "w-1"}})

同一个任务，图版本要三样东西：State schema、节点函数、边的连接。函数式版本只剩函数本身：两个业务函数挂上 @task，一个入口函数挂上 @entrypoint，调用顺序就是你眼睛看到的调用顺序。State schema 消失了（中间值在函数局部变量里传递），边消失了（顺序就是代码行序），节点命名也消失了（函数名就是步骤名）。对于「取数、加工、出报告」这种步骤链，函数式版本和手写 Python 几乎没有区别，唯一的仪式感是两个装饰器和一个 .result()。

这不是语法糖层面的差别，是思考方式的差别。写图版本时，你在设计一个状态机：先想 state 里有什么，再想每个节点贡献什么，最后想它们怎么连。写函数式版本时，你在写一个普通函数：数据从哪来、加工几步、返回什么。后者符合绝大多数工程师的默认思维方式，这也是函数式 API 存在的理由：不是图不好，是很多任务的形状本来就是一条线，为一条线建状态机是仪式负担。

代码量的差距随步骤数放大。两个步骤时图版本多出的东西还能忍受；十步的报表管道，图版本要为十步维护 schema 字段（很多字段只在相邻两步之间传递，却要进全局 state）、十条边、十个函数签名，而函数式版本只是十行函数调用。步骤越多、中间值越「过路」，函数式省得越多。反过来，中间值要被多个不相邻的步骤共享、要被外部检查时，显式的 state 又变成了优势：字段在 schema 里，看得见查得了。

注意两版保留的共同点：checkpointer 和 thread_id。函数式同样要挂 checkpointer 才有持久化，同样按 thread_id 区分会话，.invoke 的用法也和图一致。底座是同一套（第 03 篇讲的持久化机制），差别只在「持久化的单元怎么划分」。顺带一提，两版可以共存于同一个项目，编译好的图和 entrypoint 函数之间互相调用没有障碍，这给渐进式迁移留了后路。

注意 @task 装饰的函数调用后返回的不是结果，是一个 future，要 .result() 才拿值。这是函数式 API 最重要的语法细节，也是它实现持久化的钥匙，下一节展开。

future：为什么调用任务返回的不是结果

初见 fetch_orders(user_id).result() 多半会别扭：为什么不直接 fetch_orders(user_id)？这个别扭正是理解函数式持久化的入口。调用一个 @task 函数，函数体并不会立刻执行完并返回值，运行时把它登记成一个待执行单元，返回一个 future 句柄；.result() 才是「我要这个值」的时刻，如果它还没跑完，这里会等它。

为什么要这么绕？因为持久化需要明确的「步骤」边界，而普通函数调用没有边界：a(b(c())) 是一口气执行的一个表达式，崩在任何位置你都不知道该从哪重来。@task 把每个调用变成一个可单独记档、可单独跳过的单元：函数的输入是什么、输出是什么、执行没有执行，运行时全都知道。函数式 API 的所有持久化能力，都建立在这个单元化之上。

还有一层隐含的收获：把调用和取值分开，就给了并发一个入口。连续启动几个 task（都拿到 future），再逐个 .result() 取值，这几个 task 就有机会被运行时并发安排，而不是被迫写成严格串行。写串行代码，拿并行的机会，这是 future 风格的老传统。当然，任务之间有数据依赖时该串行还得串行，依赖关系由你取值的顺序决定。

拿一个真实形状的任务看这个入口值多少钱。周报任务要取三个数据源：订单库、工单库、用户调研表。串行写法是三个 task 依次取，总时长是三者之和；并发写法是三个调用全启动、最后一起取值，总时长约等于最慢的那个。三个外部查询各占两秒时，串行六秒，并发两秒出头，改的只是三行代码的顺序。多数据源聚合类任务（报表、风控快照、审批材料汇总）几乎都有这个形状，这是函数式 API 在线性任务里最实际的性能红利。

@task 函数的内部依然是普通 Python，你可以在里面写循环、异常处理、调任何库。被改变的只有它的边界：进来了什么、出去了什么、跑没跑过，这三件事归运行时管。

task 的失败语义也顺着边界来。task 执行出错，异常会在 .result() 取值那一刻抛到函数体里，你可以在函数体里 try/except 做兜底。注意这个 except 代码块本身在 task 外，重放时它每次都会执行，所以里面适合放「记录日志、返回降级文案」这类可以重复做的动作；如果 except 里要累计重试次数、写标记，那个计数就得是独立 task，不然恢复前后数字对不上。任务重试则交给运行时的重试配置或第 12 篇的错误分类策略，别在函数体里手写 while 循环硬包，那会把重放语义搅乱。

重放：函数式的持久化语义

图版本的恢复靠 superstep 边界快照；函数式没有图，它的恢复机制是重放：@entrypoint 挂了 checkpointer 之后，每次执行会把每个 task 的输入输出记档。中断后重新 invoke 同一个 thread，函数体从头执行，但已经跑完的 task 不再真的执行，直接取上次记录的结果，只有没跑到的地方才真正继续。

拿一个具体的中断走一遍，语义立刻清楚。第一次 invoke：fetch_orders 正常执行，结果记档；make_report 执行到一半进程挂了，它的输入记了档，输出没有。第二次 invoke 同一个 thread：函数体从第一行重新开始，fetch_orders(user_id) 这个调用点发现档里有这条记录，直接把上次的结果塞进 future，.result() 立刻返回，查询没有真的再发；make_report 没有档，真执行。整个过程对函数体代码完全透明，代码里没有任何「我是不是在恢复」的判断。

用户视角和运行时视角在这里值得分开说。用户视角里，这次任务是「跑了一次，中途断了，接着跑完了」；运行时视角里，同一 thread 实际发生了两次完整的函数体执行，只是第二次大部分步骤被档位短路。理解这层差异，就能理解为什么某些问题会出现：函数体里任何依赖「我只执行一次」假设的代码（自己加的防重标记、局部计数器）都会在第二次执行时露出破绽。把「函数体会被完整重跑」当成函数式世界的物理定律，这类问题就不会被写出来。

这套语义直接推出一条铁律：副作用必须放进 @task。写在 task 外面的代码，重放时会原样再跑一遍，请求发两次、邮件收两封。因为重放等于把函数体从头过一遍，task 外的每一行都会真实执行第二次，只有 task 内的执行被记档保护。换句话说，函数体里 task 外的区域是「每次都会跑」的区域，只能放纯计算和无副作用的准备动作；一切碰外部世界的事，发请求、写库、发消息、调支付，都必须包进某个 task。

反过来，把非确定值（随机数、当前时间、外部 API 响应）也放进 task，恢复后的执行才和正常执行一致：

@entrypoint(checkpointer=InMemorySaver())
def weekly(user_id: str) -> str:
    ts = datetime.now()          # 错：重放时会变成恢复时刻的时间
    orders = fetch_orders(user_id).result()
    ts2 = now().result()         # 对：时间也作为 task 结果被记录
    ...

错误的写法里，第一次执行 ts 是昨天下午三点，中断恢复后函数体重跑，ts 变成了今天上午十点，同一个 thread 前后两段执行拿着两个不同的「当前时间」，下游所有基于时间的判断都被污染。正确的写法把取时间也变成一个 task，它的值在第一次执行时被记档，恢复时返回的是档里的旧值，时间线保持一致。随机数、UUID 生成、任何「两次调用结果不同」的东西，同理都要进 task。

可以把铁律和它的逆命题合成一句话：task 是函数式 API 的「确定性边界」，副作用靠它不被重复，非确定值靠它不被改写。写函数体时的全部心思，就是划好这条边界：确定性纯计算放在外面随便跑，其余一律入 task。

和图版本的恢复对比一下，能看清两套机制的成本模型差异。图版本崩溃后恢复，运行时读链尾快照直接重建状态、按 snap.next 调度，没有执行的痕迹要补。函数式版本恢复，函数体要从第一行真的再跑一遍，已完成的 task 靠档位跳过。也就是说函数式的「跳过」省的是外部交互和重活，省不掉函数体本身的 CPU 执行。对大多数业务这无所谓（函数体本身是轻的，重的是外部调用），但如果函数体里有纯 CPU 的重计算，它没在 task 里的话恢复时会被再算一遍。这也再次呼应铁律：要么把重活放进 task，要么接受它重放时重来。

还有一点容易误解：重放只发生在同一个 thread 上。换个 thread_id 调用，档位是空的，所有 task 真实执行。档位跟着 thread 走，就像图的快照链跟着 thread 走一样，这是两套写法共享同一底座的又一处体现。

从图版本迁移：一张概念对照表

团队里已经有图版本代码、或者你已经读完第 02 篇，用一张对照表把两套概念接起来，学习成本能砍掉大半。

State schema 里的字段，对应函数体里的局部变量。不再需要为「中间值放哪」设计全局结构，作用域天然清晰。
图的节点，对应 @task 函数。都是「领走一段工作、交出一个结果」的单元，区别是节点的结果进 state，task 的结果进 future 再进局部变量。
图的边（执行顺序），对应函数体里的调用行序。写在哪一行就在哪一步执行，不用 add_edge。
条件边，对应普通的 if 和提前返回。分支逻辑回到熟悉的样子，不用 add_conditional_edges 写路由函数。
Send 动态扇出（第 08 篇），对应「连续启动多个 task 再逐个取结果」。并行的表达从「运行时指令」变成「代码写法」。
checkpointer、thread_id、恢复语义，完全相同。第 03 篇讲的东西原样适用。

这张表也反过来解释了选型：概念被替换掉的那些（schema、边、条件边）恰好是图的强项所在。如果你的任务用不到这些概念（没有共享状态要管理、没有运行时分支），替换就是白赚的简化；如果全用上了，翻译回普通代码反而丢失表达力。

真要做迁移，操作顺序也有讲究：先把每个图节点翻译成 @task（函数体基本原样搬），再把边的拓扑排序翻译成调用行序，条件边翻译成 if，最后把 state 字段的读写收窄成局部变量和参数。顺序对了，每一步都可独立验证；顺序反了（先写函数体再拆 task），很容易把副作用留在外面。翻译完跑一次「中断演练」：执行到一半杀进程，重新 invoke，确认已完成的 task 被跳过、副作用没有重复，这个演练通过，迁移才算闭环。

task 粒度：函数式的手艺活

重放语义给了第二条设计准则：task 划多粗，恢复就多划算或多么不划算。这是函数式 API 里最吃手艺的决定，值得单独一节。

task 切太粗，比如把「取数、清洗、生成」三步包成一个 task，三步里崩在最后一步，重放时前两步（两次外部查询加一段计算）全部真实重跑，记档机制形同虚设。task 切太细，比如把一次 HTTP 调用拆成「构造请求」「发送」「解析」三个 task，档位记录的开销上去了，恢复时跳过的粒度又细到没有意义，还让函数体变得琐碎难读。

实用的准绳：一个 task 对应一次外部交互，或者一个有名字的业务步骤。「查订单」「生成报告」「发送通知」是好的 task；「字符串拼接」「格式转换」不是 task，是函数体里的普通代码。评审时对每个 task 问一句「它重跑一次的代价是什么」：代价是可接受的一次查询，粒度合适；代价是重复付款，说明它外面还缺一层幂等（第 03 篇的幂等键在这里照常适用，task 记档防的是框架内的重放，防不了「task 真的执行了、进程在记档前挂了」的窗口）。

批处理是检验粒度准绳的好例子。几百条数据逐条调模型的离线任务，每条数据一个 task：跑到第 200 条崩溃，恢复后前 199 条全部跳过，只有第 200 条重做，中断的代价被钉死在单条。如果偷懒把整批包成一个 task，任何一次中断都从第一条重来，挂三次你就一晚上白跑。这个例子里 task 粒度不是风格问题，直接决定任务在故障下的期望完成时间。

还有一个和图版本互通的观察：函数式的「task 链」承担了图版本里 superstep 边界的角色。图版本的恢复粒度是节点，函数式的恢复粒度是 task，两者都遵循同一个原则：恢复单元的粒度，决定了重跑的浪费程度。第 02 篇讲过的「别把五个步骤塞进一个大节点」，在这里翻译成「别把五个外部调用塞进一个 task」。

怎么选

三条判断，按顺序问：

控制流是数据依赖的还是步骤序列的？ 有大量条件分支、动态路由、并行扇出，图的模型更贴合，add_conditional_edges 和 Send 就是干这个的。线性步骤为主，函数式更省事，不用为三个步骤定义 State schema。判断的实操方法：把流程讲一遍，如果每句话都是「然后」，函数式合适；出现「如果 X 则走 A 否则走 B」「这几件并行做」「看情况回到 Y」，用图。讲不全也没关系，讲你确定的部分，流程里「然后」的浓度是可以数的。
要不要图可视化？ functional API 不支持图可视化和 Studio 的图视图。调试靠日志。团队需要「给非工程同事看流程图」，选图。这条在多人协作和跨团队评审里权重比想象高：一张图是天然的对齐工具，函数体只有工程师能读。
团队熟悉度？ 函数式是普通 Python，新人上手快；图的 superstep、reducer 心智模型要先学（第 02 篇整篇在打这个地基）。接手人构成也要考虑：三年后的维护者读「带持久化的普通函数」和读「状态机」，哪个更容易接住。

灰色场景存在：主流程线性，中间一段是复杂子流程。两者可以混用，@entrypoint 里可以正常 invoke 一个编译好的图，图节点里也可以调 task 逻辑。按边界拆，别硬翻译成一种范式：主流程用函数式保住可读性，复杂子流程封成一个图（它内部的分支、并行、子图隔离各得其所），调用处就是一行 invoke。反过来硬翻的成本可以算：把分支逻辑翻译成条件边，把局部变量翻译成 state 字段，每个中间值都要在 schema 里注册，代码量翻倍且可读性下降。

选型时还有一个错误的排除法要防：不要因为「函数式更新」而选它，也不要因为「图更主流」而选它。两套写法底下是同一个运行时、同一套持久化（第 03 篇），它们不是新旧两代，是同一代的两副面孔。判断依据从头到尾只有一条：你的控制流长什么样。

补一条常被漏掉的判断维度：执行轨迹要不要「当数据用」。图版本的 checkpoint 链天然按节点切分，回放、轨迹评测（第 15 篇）、给非工程同事展示流程，都直接受益于这个结构。函数式的 task 档位同样记录了每一步，但围绕它的分析工具和惯例少得多，想要「把每次执行变成可分析的轨迹」时要多做一层自建。如果你的规划里评测和轨迹分析是重头，这条要计入选图的理由。

常见误用三则

误用一：把通知类副作用写在函数体里。 「报表生成完，发个钉钉通知」，顺手一行 send_dingtalk(...) 写在了 task 外面。正常执行一切正常，某次中断恢复后，函数体重跑，通知发了两遍，群里没人说话，但需求方已经在问「为什么收两条一模一样的日报」。这类事故的麻烦在于它不报错、不丢数据，只是让框架的承诺（不重复执行）在你手里失效。修法一行：包个 task。预防的办法是立一条团队规则：entrypoint 函数体里只允许出现纯计算、task 调用和 .result()，出现其他函数调用就打回。

误用二：task 依赖外部的可变状态。 函数体里读一个模块级全局变量或配置字典，task 内部也引用它。第一次执行和恢复重放时，如果这个全局量的值已经变了（有人改了配置、别的请求更新了缓存），同一个 task 在两个时刻读到不同的输入，输出随之漂移，档位记录的一致性承诺被绕过。修法是把 task 需要的一切通过参数显式传入：参数进档，恢复时原样取回。闭包和全局量在普通代码里是无害的习惯，在持久化语义下是隐雷。

误用三：把 entrypoint 函数当普通函数直接调用。 weekly("u42") 看起来和 weekly.invoke("u42", config=...) 差别不大，前者却完全绕过了运行时：没有 checkpointer 读取、没有档位记录，函数跑了，持久化静默失效。测试环境里因为总是一口气跑完，看不出区别；上生产就变成「每次都从头执行」，恢复能力为零。这不是假设的风险，是函数式 API 最常见的「看起来在用、实际没生效」。防御手段很简单：只暴露 invoke 入口（HTTP 路由、定时任务里都走 invoke），函数本身不导出。

三则误用共享同一个根源：函数式 API 长得太像普通代码，以至于框架的存在感容易被忘掉。图版本因为仪式感重（schema、边、compile），反而没人会忘记自己在用框架。享受函数式的简洁，就要把「这是带持久化语义的代码」这个意识主动带在身上，上面三条规则就是这份意识的具体化。

权衡

函数式换来的代价有三个方面，逐个摊开。

调试工具少。没有图视图，Studio 的图调试流用不上，执行过程靠日志和 checkpoint 检查恢复。图版本出问题，「打开图看节点状态」是直觉动作，函数式版本只能靠日志定位到「哪个 task 的档不对」。第 14 篇的可观测手段大部分建立在图形态上，函数式能复用的是 checkpoint 层，不是图视图层，这点在选型时要有预期。

生态集成少。大量教程、模板、第三方组件默认图形态，函数式出问题搜到的答案也少。这不是能力缺陷，是普及度差距，但它有真实的成本：新人入职后的学习资料、遇到冷门问题时的检索效率、复用社区组件时的适配工作，都要多出一截。选函数式等于选了一条人少的路，路本身没问题，补给站少，出发前把油加满（团队里先吃透官方的 functional API 文档和本系列第 02、03 篇的底座机制）。

恢复语义依赖你对 task 粒度的把握。图版本的恢复边界由 superstep 机械地保证，函数式的恢复边界由你的 task 划分决定，划错了（太粗）重放代价大，划太细记录开销大。上一节给了准绳，但准绳替代不了判断，这是函数式把一部分责任从框架转移到了写代码的人身上，责任转移是它换简洁付的价。

它适合的范围比图窄，但在适合的范围里，代码量的差距是压倒性的。我的用法：把 functional API 当成「LangGraph 提供的持久化函数库」，用在明确线性的任务上，比如批处理、报表、单文件 ETL、定时跑的数据管道；任何开始长出分支的需求，直接换 StateGraph，别等它腐烂。判断「开始长分支」的信号很具体：函数体里出现了第二个 if 去改执行路径、出现了「这两步其实可以同时做」、或者你需要用异常来跳转流程。出现任何一条，就是范式该换的时候。

到这里可以把两套写法的关系收拢成一句话：StateGraph 把控制流交给图，functional API 把控制流交还给代码，两者共享同一个持久化底座和同一个恢复哲学（把执行切成可记录的单元）。你在第 02、03 篇建立的通道、快照、恢复语义，在函数式世界里一个都没变，变的只是切单元的方式：图按 superstep 边界切，函数式按 task 切。理解了这一点，选型就不再是站队，而是给任务挑一件合身的衣服。

下一篇进入第二部分（控制）：interrupt() 和 Command(resume=...) 如何在图上实现人审，以及它落地的六个坑。

把正面清单和反面清单都摆出来，省得每次现想。正面：定时日报周报（取数、加工、渲染三步走）、单文件 ETL、批量调模型的离线任务（几百条数据逐条过，每条一个 task，中断后续跑）、告警聚合与通知分发。共同点全是「步骤序列 + 每步有明确输入输出」。反面：多轮对话 agent（状态在轮间流转，图的 messages 通道是主场，第 06 到 10 篇的能力全建立在图上）、工具调用有复杂循环的 ReAct（分支密集）、需要给产品方展示流程图的协作场景。共同点是「控制流本身是任务的一部分」。

混用时边界划分的原则一句话：按「状态是否需要共享」切。子流程要读写主流程的共享状态、或者子流程内部有分支路由，把它封成图，靠 invoke 传参进出；子流程自给自足（输入进来、结果出去，中间不与外界共享），封成一个 task 就够。边界划在数据流上，两种范式各管一段，谁也不别扭。

还没有评论

欢迎留下第一条评论，帮助这篇内容更快形成讨论。