函数式工作流 API

函数式工作流 API

学习目标

读完本章后，你将能够：

• 理解函数式 API 与声明式 StateGraph 的设计差异
• 使用 @entrypoint 和 @task 装饰器定义工作流
• 掌握并行任务编排与 checkpoint 持久化的函数式模式
• 评估函数式 API 的适用场景与限制

---

1. 概念定义

函数式工作流 API 提供了一种不同于显式建图的工作流定义方式。开发者通过装饰器标注普通函数，框架在底层自动构建执行图。

对比两种风格：

维度	声明式 StateGraph	函数式 API
定义方式	显式构建图结构	装饰器标注函数
状态管理	共享状态对象	函数返回值传递
控制流	边和条件边定义	代码逻辑直接控制
适用场景	复杂条件分支、多轮循环	线性/并行流水线

2. 核心组件

2.1 @task 装饰器

@task 将普通函数标记为可并行执行的工作流任务：

• 被 @task 装饰的函数调用后返回 Future 而非结果
• Future 支持 .result() 阻塞等待或 await 异步等待
• 多个 task 可以并发启动，通过 Future 聚合结果

关键能力：

能力	说明
重试策略	retry_policy 配置自动重试次数、退避策略
缓存策略	cache_policy 缓存相同输入的输出结果
超时控制	timeout 限制单次执行时长

2.2 @entrypoint 装饰器

@entrypoint 定义工作流的入口点：

• 接受单个参数作为输入（多个参数用字典传递）
• 可访问 config（运行时配置）、previous（上次返回值）、runtime（运行时对象）
• 底层编译为 Pregel 图

2.3 entrypoint.final

entrypoint.final(value=..., save=...) 分离两个值：

• value：返回给调用者的值
• save：持久化到 checkpoint 的值（下次通过 previous 参数获取）

这种分离允许工作流返回简化结果给调用者，同时保存完整上下文供下次恢复使用。

3. 工作流模式

3.1 并行任务模式

@entrypoint
def workflow(input):
    future_a = task_a(input.x)  # 启动，不等待
    future_b = task_b(input.y)  # 启动，不等待
    result_a = future_a.result()  # 等待 A
    result_b = future_b.result()  # 等待 B
    return combine(result_a, result_b)

3.2 带人类审核的模式

@entrypoint(checkpointer=...)
def workflow(input):
    result = slow_task(input).result()
    review = interrupt({"result": result})  # 暂停等待人类审核
    return apply_review(result, review)

3.3 增量更新模式

@entrypoint(checkpointer=...)
def workflow(data, *, previous=None):
    history = previous or []
    new_result = process(data)
    return entrypoint.final(value=new_result, save=history + [new_result])

4. 陷阱与对策

陷阱 1：task 只能在 entrypoint/StateGraph 内调用

问题：@task 装饰的函数脱离了 entrypoint 或 StateGraph 上下文调用会失败。

对策：确保 task 只在工作流内部调用，不在工作流外部直接调用。

陷阱 2：异步超时仅支持 Python 3.11+

问题：同步任务不支持超时（无法安全取消），异步超时需要 Python 3.11+ 的任务组支持。

对策：对有时间要求的任务使用异步函数，确保 Python 版本 ≥ 3.11。

陷阱 3：生成器函数不支持

问题：@entrypoint 和 @task 不支持生成器函数（generator/async generator）。

对策：使用 stream() / astream() 方法代替生成器返回值。

5. 设计取舍

优势

• 学习曲线低：熟悉 Python 函数的开发者可以直接上手
• 控制流自然：使用 if/for/while 等原生语法，无需学习图的 DSL
• 并行编排简洁：Future 模型比显式定义并行节点更直观

代价

• 灵活性受限：复杂的条件分支和循环用代码表达不如图结构清晰
• 调试困难：底层隐式建图，出问题时难以定位图结构
• 与 StateGraph 的边界模糊：两者可以混合使用，但混合后难以理解整体架构

替代方案

• 声明式 StateGraph：适合复杂的条件分支、多轮循环场景
• 预构建 Agent：create_react_agent 等高层封装，适合标准 ReAct 模式

6. 参考来源

• 文档: LangGraph Functional API, https://docs.langchain.com/oss/python/langgraph/functional-api
• 文档: LangGraph Functional API Reference, https://reference.langchain.com/python/langgraph/