第 9 章 · Functional API 与任务重放

已有 Python 流程不一定要全部改成显式图。Functional API 把持久化执行边界放在函数上，让业务控制流继续使用普通 Python。

工作机制与设计选择

@entrypoint 定义可调用的工作流入口；@task 包装需要记录结果的任务。任务返回 future，同步调用 .result() 取值。工作流重放会利用已记录任务结果，但不是让任意外部副作用天然恰好执行一次。入口内的非确定性行为应进入任务边界；任务自身对外写入仍需要幂等键。入口输入通过一个参数传递，多字段可使用字典。

示例

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

@task
def normalize(text: str) -> str:
    return " ".join(text.split()).lower()

@task
def count_words(text: str) -> int:
    return len(text.split())

@entrypoint(checkpointer=InMemorySaver())
def analyze(inputs: dict) -> dict:
    text = normalize(inputs["text"]).result()
    count = count_words(text).result()
    return {"text": text, "words": count}

result = analyze.invoke(
    {"text": "  Functional   API  "},
    {"configurable": {"thread_id": "functional-001"}},
)
assert result == {"text": "functional api", "words": 2}
print(result)

运行与验收

直接运行，应得到规范化文本及 words=2。选择 Graph API 或 Functional API 主要取决于控制流表达和维护方式，不应把二者误认为两个不同的模型 Provider。

容易踩的坑

不要在一个任务里混合多个无关外部写操作，否则重试和审计粒度过粗。改变持久化工作流的任务顺序或中断顺序后，应测试旧检查点兼容性。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. task 调用返回什么？

2. 入口输入如何组织？

3. 持久化等于外部恰好一次吗？

4. Functional API 替代模型 Provider 吗？

动手实践

1. 将 normalize 故意抛错，观察错误如何传播到入口。
2. 为多个文本创建任务 future，再收集结果。
3. 说明为什么数据库写入仍需业务幂等键。

官方参考

• use-functional-api

核验日期：2026-09-17。