Functional API：以函数式风格定义工作流

Functional API：以函数式风格定义工作流

学习目标

本章要解决什么问题：

• 如何使用 @entrypoint 和 @task 装饰器定义工作流
• 如何实现并行任务编排
• 如何使用 entrypoint.final 分离返回值与持久化值
• 如何在函数式工作流中使用 checkpoint 和中断

前置知识

本章涉及函数式 API 的通用原理，建议先阅读：

• 函数式工作流 API — 装饰器模式、Future 模型、与 StateGraph 的对比

下文假设你已理解上述概念，直接聚焦 LangGraph 的具体实现。

---

项目实践

@task：定义可并行执行的任务

from langgraph.func import task

@task
def analyze_document(doc: str) -> dict:
    # 执行分析，返回结果
    return {"summary": "...", "key_points": [...]}

@task
def search_references(query: str) -> list:
    # 执行搜索，返回结果列表
    return [{"title": "...", "relevance": 0.9}]

关键：调用 @task 装饰的函数返回 Future，需要调用 .result() 或 await 获取实际结果。

@entrypoint：定义工作流入口

from langgraph.func import entrypoint

@entrypoint()
def research_workflow(topic: str) -> dict:
    # 并行启动两个任务
    doc_future = analyze_document(topic)
    refs_future = search_references(topic)

    # 等待结果
    analysis = doc_future.result()
    references = refs_future.result()

    return {
        "topic": topic,
        "analysis": analysis,
        "references": references,
    }

持久化工作流

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

@task
def compose_essay(topic: str) -> str:
    return f"一篇关于 {topic} 的文章"

@entrypoint(checkpointer=InMemorySaver())
def review_workflow(topic: str) -> dict:
    essay = compose_essay(topic).result()
    review = interrupt({"essay": essay})  # 暂停等待审核
    return {"essay": essay, "review": review}

config = {"configurable": {"thread_id": "thread-1"}}

# 第一次：执行到 interrupt() 暂停
review_workflow.invoke("AI 安全", config)

# 审核后恢复
review_workflow.invoke(Command(resume="通过"), config)

entrypoint.final：分离返回值与持久化值

@entrypoint(checkpointer=InMemorySaver())
def counter_workflow(
    value: int,
    *,
    previous: int | None = None,
) -> entrypoint.final[int, int]:
    prev = previous or 0
    # 返回给调用者的是累计值
    # 但保存到 checkpoint 的是更新后的值
    return entrypoint.final(value=prev, save=prev + value)

config = {"configurable": {"thread_id": "counter-1"}}
counter_workflow.invoke(3, config)  # 返回 0（previous 为 None）
counter_workflow.invoke(5, config)  # 返回 3（上次的 save 值）
counter_workflow.invoke(2, config)  # 返回 8（3 + 5）

可注入参数

@entrypoint 装饰的函数可以声明以下参数自动注入：

参数	类型	说明
config	RunnableConfig	运行时配置（thread_id 等）
previous	Any	上一次执行的返回值（需 checkpointer）
runtime	Runtime	运行时对象，包含 context、store、writer

from langgraph.func import entrypoint
from langgraph.runtime import Runtime

class AppContext(TypedDict):
    user_id: str

@entrypoint(context_schema=AppContext)
def my_workflow(data: dict, *, runtime: Runtime[AppContext]) -> dict:
    user_id = runtime.context["user_id"]  # 从上下文获取
    return {"user": user_id, "result": process(data)}

问题与规避

task 只能在 entrypoint/StateGraph 内调用

问题：在工作流外部直接调用 @task 函数会失败，因为缺少执行上下文。

规避：确保 task 只在工作流内部调用。如果需要独立测试 task 逻辑，剥离装饰器测试底层函数。

异步超时仅支持 Python 3.11+

问题：同步任务的超时不被支持（无法安全取消同步操作）。

规避：对有时间要求的任务使用 async def，并确保 Python 版本 ≥ 3.11。

生成器函数不支持

问题：@entrypoint 和 @task 不接受生成器函数。

规避：将生成器逻辑改为返回列表，或使用 stream() / astream() 方法。

设计取舍

优势

• 学习曲线低：熟悉 Python 函数的开发者可直接上手
• 并行编排简洁：Future 模型比显式定义并行节点更直观
• 与 StateGraph 互操作：可以在 entrypoint 内调用已编译的 StateGraph

代价

• 复杂条件分支不够直观：多层条件嵌套用代码表达不如图结构清晰
• 调试困难：底层隐式建图，出问题时难以看到图结构
• 状态传递受限：只能通过函数返回值传递，不如共享状态灵活

替代方案

• 声明式 StateGraph：适合复杂条件分支和多轮循环
• Prebuilt Agent：标准场景用 create_react_agent 更快速

参考来源

• 源码验证: libs/langgraph/langgraph/func/__init__.py — entrypoint 和 task 实现
• 文档: LangGraph Functional API, https://docs.langchain.com/oss/python/langgraph/functional-api