LangGraph Functional API 是 LangGraph 中一种以装饰器为核心的工作流构建方式,用 @entrypoint 和 @task 装饰器替代显式的图结构搭建,让开发者以更接近普通 Python 函数的方式组织工作流。
用途 该 API 适用于构建顺序工作流与条件分支、编排器-工作者(orchestrator-worker)模式、需要并行执行的任务、需要持久化与检查点的工作流、以及需要人工介入审批的流程。它被定位为显式 StateGraph 构建方式的一种更简单的替代方案。
主要能力
- 装饰器式工作流:@task 定义任务函数,@entrypoint 定义工作流入口并编排任务。
- 并行执行:任务返回 future,先启动多个任务再统一调用 .result() 阻塞取结果,即可实现并行;也可对集合进行 map 操作批量并行处理。
- 持久化与检查点:通过 checkpointer(如 InMemorySaver)启用,工作流状态在每个任务后自动保存,配合 thread_id 使用;入口函数可通过 previous 参数访问上一次返回值。
- 人工介入:使用 interrupt() 暂停等待人工审批,再用 Command(resume=...) 恢复执行。
- 条件逻辑:在入口函数中使用普通 if/else 分支,以及 while 循环实现循环直到完成(如带工具调用的 agent 循环)。
- 流式输出:支持 stream_mode 为 "updates"(任务完成事件)、"values"(每个任务后的完整状态)、"custom"(自定义事件)。
- TypeScript 支持:使用 task() 与 entrypoint(),并行通过 Promise.all 实现。
使用方式 基本模式为:从 langgraph.func 导入 entrypoint 和 task;用 @task 装饰任务函数;用 @entrypoint() 装饰工作流函数,在其中调用任务并对其返回值调用 .result();最后通过 workflow.invoke(input) 调用。启用检查点时,将 checkpointer 传入 @entrypoint(checkpointer=...),并在调用时传入 config = {"configurable": {"thread_id": "..."}}。人工审批流程中,初始运行会在 interrupt 处暂停,人工审核后以 Command(resume=True) 恢复。
关键规则
- @task 函数返回 future,必须调用 .result() 才能取得值。
- @entrypoint 是工作流入口,负责编排任务。
- 入口函数内部的任务会被追踪,以支持持久化与流式输出。
- 未加装饰器的普通函数按常规方式执行。
常见模式
- 编排器-工作者:编排器生成工作项,工作者并行处理,最后汇总结果(fan-out / fan-in)。
- 重试模式:在入口函数中循环调用不可靠任务,失败时重试,达到最大次数后抛出异常。
反模式
- 忘记调用 .result():任务返回的是 future。
- 在任务中阻塞:任务应保持专注,不要嵌套 entrypoint。
- 缺少 checkpointer:没有它就无法恢复被中断的工作流。
- 本可并行却写成顺序:应先启动任务,再阻塞取结果。
从 Graph API 迁移 原先使用 StateGraph、add_node、add_edge、compile 的写法,可改为用 @task 定义各步骤、用 @entrypoint 编排并依次调用 .result()。
适用场景与限制 适用于需要并行、持久化、人工审批和条件分支的 LangGraph 工作流开发。文档中未提供关于收费、价格、平台支持范围或性能效果的具体信息,这些内容暂未提供。