Functional API 概述

Functional API 允许你在对现有代码进行最小更改的情况下，将 LangGraph 的关键功能（持久化、内存、人机回环和流式传输）添加到你的应用程序中。它旨在将这些功能集成到可能使用标准语言原语进行分支和流程控制的现有代码中，例如 if 语句、for 循环和函数调用。与许多要求将代码重构为显式管道或 DAG 的数据编排框架不同，Functional API 允许你整合这些能力，而无需强制执行僵化的执行模型。 Functional API 使用两个关键构建块：

@entrypoint：将函数标记为工作流的起点，封装逻辑并管理执行流，包括处理长时间运行的任务和中断。
@task：代表一个离散的工作单元，例如 API 调用或数据处理步骤，可以在 entrypoint 内异步执行。任务返回一个类似 Future 的对象，可以等待或同步解析。

这为构建具有状态管理和流式传输的工作流提供了最小的抽象。

有关如何使用 Functional API 的信息，请参阅 Use Functional API。

Functional API 与 Graph API

对于更喜欢声明式方法的用户，LangGraph 的 Graph API 允许你使用图范式定义工作流。两种 API 共享相同的底层运行时，因此你可以在同一个应用程序中一起使用它们。以下是一些关键区别：

控制流：Functional API 不需要考虑图结构。你可以使用标准的 Python 构造来定义工作流。这通常会减少你需要编写的代码量。
短期记忆：Graph API 需要声明 State，并且可能需要定义 reducers 来管理图状态的更新。@entrypoint 和 @tasks 不需要显式的状态管理，因为它们的状态限定在函数内，且不跨函数共享。
检查点：两种 API 都会生成和使用检查点。在 Graph API 中，每个 superstep 之后都会生成一个新的检查点。在 Functional API 中，当任务执行时，其结果会保存到与给定 entrypoint 关联的现有检查点中，而不是创建新的检查点。
可视化：Graph API 使得将工作流可视化为图变得容易，这对于调试、理解工作流和与他人分享很有用。Functional API 不支持可视化，因为图是在运行时动态生成的。

示例

下面我们演示一个简单的应用程序，该程序撰写一篇文章并中断以请求人工审查。

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

@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1) # A placeholder for a long-running task.
    return f"An essay about topic: {topic}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt({
        # Any json-serializable payload provided to interrupt as argument.
        # It will be surfaced on the client side as an Interrupt when streaming data
        # from the workflow.
        "essay": essay, # The essay we want reviewed.
        # We can add any additional information that we need.
        # For example, introduce a key called "action" with some instructions.
        "action": "Please approve/reject the essay",
    })

    return {
        "essay": essay, # The essay that was generated
        "is_approved": is_approved, # Response from HIL
    }

详细解释

此工作流将撰写一篇关于主题“猫”的文章，然后暂停以获取人工审查。工作流可以无限期地中断，直到提供审查。当工作流恢复时，它将从头开始执行，但由于 writeEssay 任务的结果已经保存，任务结果将从检查点加载，而不是重新计算。

import time
import uuid
from langgraph.func import entrypoint, task
from langgraph.types import interrupt
from langgraph.checkpoint.memory import InMemorySaver


@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1)  # This is a placeholder for a long-running task.
    return f"An essay about topic: {topic}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt(
        {
            # Any json-serializable payload provided to interrupt as argument.
            # It will be surfaced on the client side as an Interrupt when streaming data
            # from the workflow.
            "essay": essay,  # The essay we want reviewed.
            # We can add any additional information that we need.
            # For example, introduce a key called "action" with some instructions.
            "action": "Please approve/reject the essay",
        }
    )
    return {
        "essay": essay,  # The essay that was generated
        "is_approved": is_approved,  # Response from HIL
    }


thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
for item in workflow.stream("cat", config):
    print(item)
# > {'write_essay': 'An essay about topic: cat'}
# > {
# >     '__interrupt__': (
# >        Interrupt(
# >            value={
# >                'essay': 'An essay about topic: cat',
# >                'action': 'Please approve/reject the essay'
# >            },
# >            id='b9b2b9d788f482663ced6dc755c9e981'
# >        ),
# >    )
# > }

文章已写好，准备审查。一旦提供审查，我们可以恢复工作流：

from langgraph.types import Command

# Get review from a user (e.g., via a UI)
# In this case, we're using a bool, but this can be any json-serializable value.
human_review = True

for item in workflow.stream(Command(resume=human_review), config):
    print(item)

{'workflow': {'essay': 'An essay about topic: cat', 'is_approved': False}}

工作流已完成，审查已添加到文章中。

入口点

@entrypoint 装饰器可用于从函数创建工作流。它封装工作流逻辑并管理执行流，包括处理 长时间运行的任务 和中断。

定义

入口点 通过 @entrypoint 装饰器装饰函数来定义。该函数 必须接受单个位置参数，作为工作流输入。如果需要传递多个数据片段，请使用字典作为第一个参数的输入类型。使用 entrypoint 装饰函数会产生一个 Pregel 实例，有助于管理工作流的执行（例如，处理流式传输、恢复和检查点）。你通常希望向 @entrypoint 装饰器传递一个 checkpointer 以启用持久化并使用 人机回环 等功能。

同步
异步

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: dict) -> int:
    # some logic that may involve long-running tasks like API calls,
    # and may be interrupted for human-in-the-loop.
    ...
    return result

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: dict) -> int:
    # some logic that may involve long-running tasks like API calls,
    # and may be interrupted for human-in-the-loop
    ...
    return result

序列化 入口点的输入和输出必须是 JSON 可序列化的，以支持检查点。有关更多详细信息，请参阅序列化部分。

可注入参数

声明 entrypoint 时，可以请求访问将在运行时自动注入的其他参数。这些参数包括：

参数	描述
previous	访问给定线程的前一个 `checkpoint` 关联的状态。参见短期记忆。
store	[BaseStore][langgraph.store.base.BaseStore] 的实例。适用于长期记忆。
writer	用于在使用 Async Python < 3.11 时访问 StreamWriter。详见 Functional API 流式传输详情。
config	用于访问运行时配置。有关信息请参见 RunnableConfig。

请使用适当的名称和类型注解声明参数。

请求可注入参数

from langchain_core.runnables import RunnableConfig
from langgraph.func import entrypoint
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import StreamWriter

in_memory_checkpointer = InMemorySaver(...)
in_memory_store = InMemoryStore(...)  # An instance of InMemoryStore for long-term memory

@entrypoint(
    checkpointer=in_memory_checkpointer,  # Specify the checkpointer
    store=in_memory_store  # Specify the store
)
def my_workflow(
    some_input: dict,  # The input (e.g., passed via `invoke`)
    *,
    previous: Any = None, # For short-term memory
    store: BaseStore,  # For long-term memory
    writer: StreamWriter,  # For streaming custom data
    config: RunnableConfig  # For accessing the configuration passed to the entrypoint
) -> ...:

执行

使用 @entrypoint 会产生一个 Pregel 对象，可以使用 invoke、ainvoke、stream 和 astream 方法执行。

Invoke
Async Invoke
Stream
Async Stream

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
my_workflow.invoke(some_input, config)  # Wait for the result synchronously

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
await my_workflow.ainvoke(some_input, config)  # Await result asynchronously

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

for chunk in my_workflow.stream(some_input, config):
    print(chunk)

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

async for chunk in my_workflow.astream(some_input, config):
    print(chunk)

恢复

在 interrupt 后恢复执行可以通过向 Command 原语传递 resume 值来完成。

Invoke
Async Invoke
Stream
Async Stream

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(Command(resume=some_resume_value), config)

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

await my_workflow.ainvoke(Command(resume=some_resume_value), config)

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

for chunk in my_workflow.stream(Command(resume=some_resume_value), config):
    print(chunk)

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

async for chunk in my_workflow.astream(Command(resume=some_resume_value), config):
    print(chunk)

错误后恢复 要在错误后恢复，请使用 None 和相同的 thread id (config) 运行 entrypoint。这假设底层的 error 已解决，执行可以成功继续。

Invoke
Async Invoke
Stream
Async Stream

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(None, config)

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

await my_workflow.ainvoke(None, config)

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

for chunk in my_workflow.stream(None, config):
    print(chunk)

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

async for chunk in my_workflow.astream(None, config):
    print(chunk)

短期记忆

当 entrypoint 使用 checkpointer 定义时，它会在相同 thread id 的连续调用之间存储信息到 checkpoints。这允许使用 previous 参数访问前一次调用的状态。默认情况下，previous 参数是前一次调用的返回值。

@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> int:
    previous = previous or 0
    return number + previous

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(1, config)  # 1 (previous was None)
my_workflow.invoke(2, config)  # 3 (previous was 1 from the previous invocation)

`entrypoint.final`

entrypoint.final 是一个特殊原语，可以从 entrypoint 返回，允许解耦保存在 检查点 中的值和 entrypoint 的返回值。第一个值是 entrypoint 的返回值，第二个值是将保存在检查点中的值。类型注解为 entrypoint.final[return_type, save_type]。

@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    # This will return the previous value to the caller, saving
    # 2 * number to the checkpoint, which will be used in the next invocation
    # for the `previous` parameter.
    return entrypoint.final(value=previous, save=2 * number)

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

my_workflow.invoke(3, config)  # 0 (previous was None)
my_workflow.invoke(1, config)  # 6 (previous was 3 * 2 from the previous invocation)

任务

任务代表一个离散的工作单元，例如 API 调用或数据处理步骤。它具有两个关键特征：

异步执行：任务设计为异步执行，允许多个操作并发运行而不阻塞。
检查点：任务结果保存到检查点，使工作流能够从上次保存的状态恢复。（有关更多详细信息，请参阅持久化）。

定义

任务使用 @task 装饰器定义，该装饰器包装常规 Python 函数。

from langgraph.func import task

@task()
def slow_computation(input_value):
    # Simulate a long-running operation
    ...
    return result

序列化 任务的输出必须是 JSON 可序列化的，以支持检查点。

执行

任务只能从 入口点、另一个任务或状态图节点内部调用。任务不能直接从主应用程序代码中调用。当你调用任务时，它会立即返回一个未来对象。未来是稍后将可用的结果的占位符。要获取任务的结果，你可以同步等待（使用 result()）或异步等待（使用 await）。

同步调用
异步调用

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: int) -> int:
    future = slow_computation(some_input)
    return future.result()  # Wait for the result synchronously

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: int) -> int:
    return await slow_computation(some_input)  # Await result asynchronously

何时使用任务

任务在以下场景中很有用：

检查点：当你需要将长时间运行的操作结果保存到检查点时，这样在恢复工作流时就不需要重新计算它。
人机回环：如果你正在构建需要人工干预的工作流，你必须使用任务来封装任何随机性（例如 API 调用），以确保工作流可以正确恢复。有关更多详细信息，请参阅确定性部分。
并行执行：对于 I/O 绑定任务，任务支持并行执行，允许多个操作并发运行而不阻塞（例如，调用多个 API）。
可观测性：将操作包装在任务中提供了一种跟踪工作流进度并使用 LangSmith 监控单个操作执行的方法。
可重试工作：当工作需要重试以处理失败或不一致时，任务提供了一种封装和管理重试逻辑的方法。

序列化

LangGraph 中的序列化有两个关键方面：

entrypoint 的输入和输出必须是 JSON 可序列化的。
task 的输出必须是 JSON 可序列化的。

这些要求对于启用检查点和工作流恢复是必要的。使用 Python 原语如字典、列表、字符串、数字和布尔值，以确保你的输入和输出是可序列化的。序列化确保工作流状态（如任务结果和中间值）可以可靠地保存和恢复。这对于启用人机回环交互、容错性和并行执行至关重要。提供不可序列化的输入或输出将在配置了 checkpointer 的工作流运行时导致错误。

确定性

为了利用 人机回环 等功能，任何随机性都应封装在任务内部。这保证了当执行被中止（例如，用于人机回环）然后恢复时，它将遵循相同的 步骤序列，即使任务结果是非确定性的。 LangGraph 通过在执行时持久化任务和子图结果来实现此行为。设计良好的工作流确保恢复执行遵循 相同的步骤序列，允许正确检索先前计算的结果而无需重新执行它们。这对于长时间运行的任务或具有非确定性结果的任务特别有用，因为它避免了重复之前完成的工作，并允许从本质上相同的地方恢复。虽然工作流的不同运行可能会产生不同的结果，但恢复特定运行应始终遵循相同的记录步骤序列。这允许 LangGraph 高效查找在图被中断之前执行的任务和子图结果，并避免重新计算它们。

幂等性

幂等性确保多次运行同一操作会产生相同的结果。这有助于防止如果步骤因失败而重新运行而导致重复的 API 调用和冗余处理。始终将 API 调用放在任务函数中以进行检查点，并设计它们在重新执行时是幂等的。如果任务开始但未成功完成，则可能会发生重新执行。然后，如果工作流恢复，任务将再次运行。使用幂等键或验证现有结果以避免重复。

常见陷阱

处理副作用

将副作用（例如，写入文件、发送邮件）封装在任务中，以确保在恢复工作流时不会多次执行它们。

不正确
正确

在此示例中，副作用（写入文件）直接包含在工作流中，因此在恢复工作流时将再次执行。

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # This code will be executed a second time when resuming the workflow.
    # Which is likely not what you want.
    with open("output.txt", "w") as f:
        f.write("Side effect executed")
    value = interrupt("question")
    return value

在此示例中，副作用被封装在任务中，确保在恢复时执行一致。

from langgraph.func import task

@task
def write_to_file():
    with open("output.txt", "w") as f:
        f.write("Side effect executed")

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # The side effect is now encapsulated in a task.
    write_to_file().result()
    value = interrupt("question")
    return value

非确定性控制流

每次可能给出不同结果的操作（如获取当前时间或随机数）应封装在任务中，以确保在恢复时返回相同的结果。

在任务中：获取随机数 (5) → 中断 → 恢复 → (再次返回 5) → …
不在任务中：获取随机数 (5) → 中断 → 恢复 → 获取新随机数 (7) → …

这在具有多个中断调用的 人机回环 工作流中尤为重要。LangGraph 为每个任务/入口点保留恢复值列表。遇到中断时，它将与相应的恢复值匹配。此匹配严格 基于索引，因此恢复值的顺序应与中断的顺序匹配。如果在恢复时未保持执行顺序，一个 interrupt 调用可能与错误的 resume 值匹配，导致不正确的结果。请阅读确定性部分以了解更多信息。

不正确
正确

在此示例中，工作流使用当前时间来确定执行哪个任务。这是非确定性的，因为工作流的结果取决于执行时的时间。

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = time.time()

    delta_t = t1 - t0

    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")

    return {
        "result": result,
        "value": value
    }

在此示例中，工作流使用输入 t0 来确定执行哪个任务。这是确定性的，因为工作流的结果仅取决于输入。

import time

from langgraph.func import task

@task
def get_time() -> float:
    return time.time()

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = get_time().result()

    delta_t = t1 - t0

    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")

    return {
        "result": result,
        "value": value
    }

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Get started

Capabilities

Production

Frontend

LangGraph APIs

Functional API 与 Graph API

示例

入口点

定义

可注入参数

执行

恢复

短期记忆

`entrypoint.final`

任务

定义

执行

何时使用任务

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多

​Functional API 与 Graph API

​示例

​入口点

​定义

​可注入参数

​执行

​恢复

​短期记忆

​entrypoint.final

​任务

​定义

​执行

​何时使用任务

​序列化

​确定性

​幂等性

​常见陷阱

​处理副作用

​非确定性控制流

​了解更多

Functional API 与 Graph API

示例

入口点

定义

可注入参数

执行

恢复

短期记忆

`entrypoint.final`

任务

定义

执行

何时使用任务

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多