函数式 API 的全部公开符号只有两个，看模块开头 func/__init__.py:56：

# func/__init__.py:56
__all__ = ("task", "entrypoint")

对比一下同一个"加 1 流程"的两种写法：

# 声明式（前 46 天的写法）
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class S(TypedDict):
    n: int
g = StateGraph(S)
g.add_node("inc", lambda s: {"n": s["n"] + 1})
g.add_edge(START, "inc")
g.add_edge("inc", END)
app = g.compile()
app.invoke({"n": 1})            # {'n': 2}

# 命令式（今天的函数式 API）
from langgraph.func import entrypoint, task
@task
def inc(n: int) -> int:
    return n + 1
@entrypoint()
def app(n: int) -> int:
    return inc(n).result()       # 直接像调函数
app.invoke(1)                    # 2

task 是个可带参也可裸用的装饰器 func/__init__.py:132。核心是把你的函数塞进一个 _TaskFunction 对象 func/__init__.py:59：

# func/__init__.py:59
class _TaskFunction(Generic[P, T]):
    def __init__(self, func, *, retry_policy, cache_policy=None,
                 timeout=None, name=None):
        ...
        self.func = func                    # 你原始的函数
        self.retry_policy = retry_policy     # 失败重试策略（D52）
        self.cache_policy = cache_policy     # 结果缓存策略（D51）
        self.timeout = timeout
        functools.update_wrapper(self, func) # 把 __name__/__doc__ 抄过来，伪装成原函数

而 task(...) 本体只是"整理参数 + 造装饰器"，看 func/__init__.py:226：

# func/__init__.py:226
retry_policies: Sequence[RetryPolicy] = (
    ()                                       # 没给 → 空元组（不重试）
    if retry_policy is None
    else (retry_policy,)                     # 给了单个 → 包成 1 元组
    if isinstance(retry_policy, RetryPolicy)
    else retry_policy                        # 给了列表 → 原样
)
def decorator(func):
    ...
    return _TaskFunction(func, retry_policy=retry_policies,
                         cache_policy=cache_policy, timeout=timeout_policy, name=name)
if __func_or_none__ is not None:             # 裸用 @task（没括号）
    return decorator(__func_or_none__)
return decorator                             # 带参 @task(...)（有括号）

关键在 _TaskFunction.__call__ func/__init__.py:86——你以为在"调函数"，其实在"提交一个任务"：

# func/__init__.py:86
def __call__(self, *args, **kwargs) -> SyncAsyncFuture[T]:
    return _call_with_options(
        self.func, args, kwargs,
        retry_policy=self.retry_policy,
        cache_policy=self.cache_policy,
        timeout=self.timeout,
    )

_call_with_options 不当场执行你的函数，而是把它交给"当前正在跑的引擎"，返回一个 Future pregel/_call.py:276：

# pregel/_call.py:276
def _call_with_options(func, args, kwargs, *, retry_policy=None,
                       cache_policy=None, timeout=None) -> SyncAsyncFuture[T]:
    ...
    config = get_config()                    # 拿到"我现在在哪张图里跑"的上下文
    impl = config[CONF][CONFIG_KEY_CALL]     # 引擎放在 config 里的"提交任务"回调
    fut = impl(func, (args, kwargs), retry_policy=retry_policy,
               cache_policy=cache_policy, callbacks=config["callbacks"],
               timeout=timeout)
    return fut                               # 返回 Future，不是结果本身

而 SyncAsyncFuture 本身是个"既能 .result() 又能 await"的双面 Future pregel/_call.py:253：

# pregel/_call.py:253
class SyncAsyncFuture(Generic[T], concurrent.futures.Future[T]):
    def __await__(self) -> Generator[T, None, T]:
        yield cast(T, ...)

@entrypoint()
def add_one(numbers: list[int]) -> list[int]:
    futures = [add_one_task(n) for n in numbers]   # 一口气发起 N 个任务
    results = [f.result() for f in futures]         # 再统一收结果
    return results
add_one.invoke([1, 2, 3])                           # [2, 3, 4]

entrypoint 是个类（为了挂 .final），真正的转换在 __call__ func/__init__.py:516。它读你函数的签名，然后 手工组装一张只有一个节点的 Pregel 图 func/__init__.py:576：

# func/__init__.py:576
graph: Pregel = Pregel(
    nodes={
        func.__name__: PregelNode(              # 你的函数 = 唯一的节点
            bound=bound,                         # 包装后的可运行体
            triggers=[START],                    # 被 START 触发
            channels=START,                      # 从 START 通道读输入
            timeout=self.timeout,
            writers=[
                ChannelWrite([
                    ChannelWriteEntry(END, mapper=_pluck_return_value),  # 返回值 → END
                    ChannelWriteEntry(PREVIOUS, mapper=_pluck_save_value),# 存档值 → PREVIOUS
                ])
            ],
        )
    },
    channels={
        START: EphemeralValue(input_type),       # 输入通道（用完即弃）
        END: LastValue(output_type, END),        # 输出通道
        PREVIOUS: LastValue(save_type, PREVIOUS),# "上一次的结果"通道
    },
    input_channels=START, output_channels=END, stream_channels=END,
    stream_mode="updates", stream_eager=True,
    checkpointer=self.checkpointer, store=self.store, cache=self.cache,
    cache_policy=self.cache_policy, retry_policy=self.retry_policy or (),
    context_schema=self.context_schema,
)

那 bound（被包装的函数体）从哪来？看 func/__init__.py:531：

# func/__init__.py:526
if inspect.isgeneratorfunction(func) or inspect.isasyncgenfunction(func):
    raise NotImplementedError("Generators are not supported in the Functional API.")
bound = get_runnable_for_entrypoint(func)     # 把普通函数裹成 Runnable

StateGraph 里你自己定义 State 字段当通道；entrypoint 里通道是引擎硬编码的三条 func/__init__.py:593：

输入类型不是瞎猜的——entrypoint 读你函数第一个参数的类型注解 func/__init__.py:535：

# func/__init__.py:535
sig = inspect.signature(func)
first_parameter_name = next(iter(sig.parameters.keys()), None)
if not first_parameter_name:
    raise ValueError("Entrypoint function must have at least one parameter")
input_type = (
    sig.parameters[first_parameter_name].annotation
    if sig.parameters[first_parameter_name].annotation is not inspect.Signature.empty
    else Any
)

默认情况下，函数返回什么，就存什么、下次 previous 拿到的也是它。但有时你想"返回给调用方 A，却给下次存 B"。这就是 entrypoint.final func/__init__.py:475：

# func/__init__.py:475
@dataclass(**_DC_KWARGS)
class final(Generic[R, S]):
    value: R   # 返回给调用方的值（invoke 拿到这个）
    save: S    # 存进 checkpoint 的值（下次 previous 拿到这个）

拆分逻辑就是 L04 那两个 mapper func/__init__.py:546：

# func/__init__.py:546
def _pluck_return_value(value):    # 写 END 用：取 .value，否则原样
    return value.value if isinstance(value, entrypoint.final) else value
def _pluck_save_value(value):      # 写 PREVIOUS 用：取 .save，否则原样
    return value.save if isinstance(value, entrypoint.final) else value

@entrypoint(checkpointer=InMemorySaver())
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    return entrypoint.final(value=previous, save=2 * number)

config = {"configurable": {"thread_id": "some_thread"}}
my_workflow.invoke(3, config)  # 返回 0   （previous 为 None → 0；存 2*3=6）
my_workflow.invoke(1, config)  # 返回 6   （previous 为上次存的 6；存 2*1=2）

回看 L04 那行——entrypoint 明确拒绝生成器函数 func/__init__.py:526：

# func/__init__.py:526
if inspect.isgeneratorfunction(func) or inspect.isasyncgenfunction(func):
    raise NotImplementedError("Generators are not supported in the Functional API.")

✋ 10 分钟动手

# 1. 读三段核心源码
sed -n '59,106p'  libs/langgraph/langgraph/func/__init__.py   # _TaskFunction
sed -n '516,620p' libs/langgraph/langgraph/func/__init__.py   # entrypoint.__call__ 组装 Pregel

# 2. 亲手验证"返回值≠存档值"
python - <<'PY'
from typing import Any
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver
@entrypoint(checkpointer=InMemorySaver())
def wf(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    return entrypoint.final(value=previous, save=2 * number)
cfg = {"configurable": {"thread_id": "t"}}
print(wf.invoke(3, cfg))   # 0
print(wf.invoke(1, cfg))   # 6
PY