Functional API 概览
Functional API（函数式 API）是 LangGraph 3.0+ 引入的新编程模型，允许你用标准 Python 代码（if/for/while）来定义工作流，同时享受 LangGraph 的全部能力：持久化、记忆、人机协同、流式输出。
为什么需要 Functional API
Graph API 虽然强大，但要声明节点和边，对于简单场景略显繁琐。Functional API 让你：
- 写普通函数，不需要思考图结构
- 用
if/while/for做控制流 - 减少代码量，逻辑更直观
核心概念
@task（任务）
@task 装饰器标记一个函数为可执行的任务单元：
python
from langgraph.func import task
@task
def fetch_data(query: str) -> list:
"""API 调用或数据处理"""
return api_call(query)
@task
def process_data(data: list) -> str:
"""数据后处理"""
return summarize(data)- 任务可以同步或异步
- 任务返回一个 future-like 对象，调用
.result()获取结果 - 任务结果会被自动 checkpoint，失败恢复时无需重算
@entrypoint（入口点）
@entrypoint 装饰器标记工作流的入口函数：
python
from langgraph.func import entrypoint
@entrypoint(checkpointer=checkpointer)
def my_workflow(input_data: dict) -> dict:
# 调用任务
result1 = fetch_data(input_data["query"]).result()
result2 = process_data(result1).result()
return {"output": result2}- 入口函数必须接受一个位置参数
- 如果需要传递多个值，使用字典
- 装饰后返回一个
Pregel实例，支持.invoke()、.stream()
完整示例：论文写作 + 人工审批
python
import time
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt
@task
def write_essay(topic: str) -> str:
"""写一篇关于指定主题的文章"""
time.sleep(1) # 模拟长任务
return f"关于 {topic} 的文章"
@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
"""写文章并请求人工审核"""
essay = write_essay(topic).result()
# 中断等待人工审批
is_approved = interrupt({
"essay": essay,
"action": "请审批/拒绝这篇文章"
})
return {
"essay": essay,
"is_approved": is_approved,
}
# 执行
from langchain_core.utils.uuid import uuid7
thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}
for item in workflow.stream("猫", config):
print(item)
# 恢复：提供审批结果
from langgraph.types import Command
for item in workflow.stream(Command(resume=True), config):
print(item)Functional API vs Graph API
| 特性 | Graph API | Functional API |
|---|---|---|
| 控制流 | 节点+边（声明式） | if/while/for（命令式） |
| 代码量 | 较多 | 较少 |
| 状态管理 | 需定义 State + Reducer | 函数作用域自动管理 |
| 可视化 | 支持图可视化 | 不支持（动态生成） |
| Checkpoint 粒度 | 每 super-step 一个 | 每个 task 结果保存 |
| 适用场景 | 复杂编排、多分支 | 流程相对线性的场景 |
何时使用哪种 API
选择 Graph API 当：
- 工作流有复杂的条件分支和并行
- 需要可视化图结构
- 需要精细控制每个 step 的状态
选择 Functional API 当：
- 流程相对线性
- 希望用熟悉的 Python 写法
- 需要快速将现有代码转为 LangGraph 工作流
两种 API 可以混合使用。