> ## Documentation Index
> Fetch the complete documentation index at: https://cndoc-langchain.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 功能 API 概览

**功能 API** 允许您以对现有代码的最小改动，将 LangGraph 的关键功能（[持久化](/oss/python/langgraph/persistence)、[记忆](/oss/python/langgraph/add-memory)、[人机协作](/oss/python/langgraph/interrupts) 和 [流式处理](/oss/python/langgraph/streaming)）添加到您的应用程序中。

它旨在将这些功能集成到可能使用标准语言原语（如 `if` 语句、`for` 循环和函数调用）进行分支和控制流的现有代码中。与许多需要将代码重构为显式管道或 DAG 的数据编排框架不同，功能 API 允许您在不强制执行严格执行模型的情况下整合这些能力。

功能 API 使用两个关键构建块：

* **`@entrypoint`**：将函数标记为工作流的起点，封装逻辑并管理执行流程，包括处理长时间运行的任务和中断。
* **[`@task`](https://reference.langchain.com/python/langgraph/func/task)**：表示一个离散的工作单元，例如 API 调用或数据处理步骤，可以在入口点内异步执行。任务返回一个类似 future 的对象，可以被等待或同步解析。

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

<Tip>
  有关如何使用功能 API 的信息，请参阅[使用功能 API](/oss/python/langgraph/use-functional-api)。
</Tip>

## 功能 API 与图 API

对于更喜欢声明式方法的用户，LangGraph 的[图 API](/oss/python/langgraph/graph-api) 允许您使用图范式定义工作流。两种 API 共享相同的底层运行时，因此您可以在同一应用程序中结合使用它们。

以下是一些关键区别：

* **控制流**：功能 API 不需要考虑图结构。您可以使用标准 Python 构造来定义工作流。这通常会减少您需要编写的代码量。
* **短期记忆**：**图 API** 需要声明一个[**状态**](/oss/python/langgraph/graph-api#state)，并且可能需要定义[**归约器**](/oss/python/langgraph/graph-api#reducers)来管理图状态的更新。`@entrypoint` 和 `@task` 不需要显式的状态管理，因为它们的状态作用域限于函数内部，不会在函数间共享。
* **检查点机制**：两种 API 都会生成和使用检查点。在**图 API** 中，每个[超级步骤](/oss/python/langgraph/graph-api)后都会生成一个新的检查点。在**功能 API** 中，当任务执行时，其结果会保存到与给定入口点关联的现有检查点中，而不是创建新的检查点。
* **可视化**：图 API 可以轻松地将工作流可视化为图，这对于调试、理解工作流以及与他人共享非常有用。功能 API 不支持可视化，因为图是在运行时动态生成的。

## 示例

下面我们演示一个简单的应用程序，它撰写一篇文章并[中断](/oss/python/langgraph/interrupts)以请求人工审核。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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
    }
```

<Accordion title="详细说明">
  此工作流将撰写一篇关于主题“猫”的文章，然后暂停以获取人工审核。工作流可以中断任意长的时间，直到提供审核。

  当工作流恢复时，它会从最开始执行，但由于 `writeEssay` 任务的结果已经保存，任务结果将从检查点加载，而不是重新计算。

  ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  import time
  from langchain_core.utils.uuid import uuid7
  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(uuid7())
  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'
  # >        ),
  # >    )
  # > }
  ```

  一篇文章已经写好，准备接受审核。一旦提供审核，我们就可以恢复工作流：

  ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  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)
  ```

  ```pycon theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  {'workflow': {'essay': 'An essay about topic: cat', 'is_approved': False}}
  ```

  工作流已完成，审核已添加到文章中。
</Accordion>

## 入口点

[`@entrypoint`](https://reference.langchain.com/python/langgraph/func/entrypoint) 装饰器可用于从函数创建工作流。它封装工作流逻辑并管理执行流程，包括处理 *长时间运行的任务* 和[中断](/oss/python/langgraph/interrupts)。

### 定义

**入口点**通过使用 `@entrypoint` 装饰器装饰函数来定义。

该函数**必须接受一个位置参数**，作为工作流的输入。如果需要传递多条数据，请使用字典作为第一个参数的输入类型。

使用 `entrypoint` 装饰函数会生成一个 [`Pregel`](https://reference.langchain.com/python/langgraph/pregel/#langgraph.pregel.Pregel.stream) 实例，该实例有助于管理工作流的执行（例如，处理流式传输、恢复和检查点机制）。

您通常需要向 `@entrypoint` 装饰器传递一个 **checkpointer**，以启用持久化并使用**人机协作**等功能。

<Tabs>
  <Tab title="同步">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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
    ```
  </Tab>

  <Tab title="异步">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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
    ```
  </Tab>
</Tabs>

<Warning>
  **序列化**
  入口点的**输入**和**输出**必须是 JSON 可序列化的，以支持检查点机制。更多详情请参阅[序列化](#序列化)部分。
</Warning>

### 可注入参数

声明 `entrypoint` 时，您可以请求访问在运行时自动注入的额外参数。这些参数包括：

| 参数           | 描述                                                                                                                |
| ------------ | ----------------------------------------------------------------------------------------------------------------- |
| **previous** | 访问与给定线程的上一个 `checkpoint` 关联的状态。参见[短期记忆](#短期记忆)。                                                                   |
| **store**    | \[BaseStore]\[langgraph.store.base.BaseStore] 的一个实例。用于[长期记忆](/oss/python/langgraph/use-functional-api#长期记忆)。      |
| **writer**   | 在使用 Async Python \< 3.11 时用于访问 StreamWriter。详情参见[使用功能 API 进行流式处理](/oss/python/langgraph/use-functional-api#流式处理)。 |
| **config**   | 用于访问运行时配置。有关信息，请参阅 [RunnableConfig](https://python.langchain.com/docs/concepts/runnables/#runnableconfig)。        |

<Warning>
  请使用正确的名称和类型注解来声明参数。
</Warning>

<Accordion title="请求可注入参数">
  ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  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
  ) -> ...:
  ```
</Accordion>

### 执行

使用 [`@entrypoint`](#入口点) 会生成一个 [`Pregel`](https://reference.langchain.com/python/langgraph/pregel/#langgraph.pregel.Pregel.stream) 对象，可以使用 `invoke`、`ainvoke`、`stream` 和 `astream` 方法执行。

<Tabs>
  <Tab title="调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }
    my_workflow.invoke(some_input, config)  # Wait for the result synchronously
    ```
  </Tab>

  <Tab title="异步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }
    await my_workflow.ainvoke(some_input, config)  # Await result asynchronously
    ```
  </Tab>

  <Tab title="流式处理">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }

    for chunk in my_workflow.stream(some_input, config):
        print(chunk)
    ```
  </Tab>

  <Tab title="异步流式处理">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }

    async for chunk in my_workflow.astream(some_input, config):
        print(chunk)
    ```
  </Tab>
</Tabs>

### 恢复

在[中断](https://reference.langchain.com/python/langgraph/types/interrupt)后恢复执行，可以通过向 [`Command`](https://reference.langchain.com/python/langgraph/types/Command) 原语传递一个 **resume** 值来完成。

<Tabs>
  <Tab title="调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.types import Command

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

    my_workflow.invoke(Command(resume=some_resume_value), config)
    ```
  </Tab>

  <Tab title="异步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.types import Command

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

    await my_workflow.ainvoke(Command(resume=some_resume_value), config)
    ```
  </Tab>

  <Tab title="流式处理">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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)
    ```
  </Tab>

  <Tab title="异步流式处理">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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)
    ```
  </Tab>
</Tabs>

**错误后恢复**

要在错误后恢复，请使用 `None` 和相同的 **thread id** (config) 运行 `entrypoint`。

这假设底层的**错误**已解决，执行可以成功继续。

<Tabs>
  <Tab title="调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    my_workflow.invoke(None, config)
    ```
  </Tab>

  <Tab title="异步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    await my_workflow.ainvoke(None, config)
    ```
  </Tab>

  <Tab title="流式处理">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    for chunk in my_workflow.stream(None, config):
        print(chunk)
    ```
  </Tab>

  <Tab title="异步流式处理">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    async for chunk in my_workflow.astream(None, config):
        print(chunk)
    ```
  </Tab>
</Tabs>

### 短期记忆

当使用 `checkpointer` 定义 `entrypoint` 时，它会在同一 **thread id** 的连续调用之间，在[检查点](/oss/python/langgraph/persistence#checkpoints)中存储信息。

这允许使用 `previous` 参数访问上一次调用的状态。

默认情况下，`previous` 参数是上一次调用的返回值。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
@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`](https://reference.langchain.com/python/langgraph/func/entrypoint/final) 是一个特殊的原语，可以从入口点返回，并允许将**保存在检查点中的值**与**入口点的返回值**解耦。

第一个值是入口点的返回值，第二个值是将保存在检查点中的值。类型注解为 `entrypoint.final[return_type, save_type]`。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
@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 调用或数据处理步骤。它有两个关键特征：

* **异步执行**：任务设计为异步执行，允许多个操作并发运行而不阻塞。
* **检查点机制**：任务结果保存到检查点，从而可以从最后保存的状态恢复工作流。（更多详情请参阅[持久化](/oss/python/langgraph/persistence)）。

### 定义

任务使用 `@task` 装饰器定义，该装饰器包装一个普通的 Python 函数。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.func import task

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

<Warning>
  **序列化**
  任务的**输出**必须是 JSON 可序列化的，以支持检查点机制。
</Warning>

### 执行

**任务**只能从**入口点**、另一个**任务**或[状态图节点](/oss/python/langgraph/graph-api#nodes)内部调用。

任务\_不能\_直接从主应用程序代码调用。

当您调用一个**任务**时，它会\_立即\_返回一个 future 对象。Future 是一个占位符，代表稍后可用的结果。

要获取**任务**的结果，您可以同步等待（使用 `result()`）或异步等待（使用 `await`）。

<Tabs>
  <Tab title="同步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    @entrypoint(checkpointer=checkpointer)
    def my_workflow(some_input: int) -> int:
        future = slow_computation(some_input)
        return future.result()  # Wait for the result synchronously
    ```
  </Tab>

  <Tab title="异步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    @entrypoint(checkpointer=checkpointer)
    async def my_workflow(some_input: int) -> int:
        return await slow_computation(some_input)  # Await result asynchronously
    ```
  </Tab>
</Tabs>

## 何时使用任务

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

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

## 序列化

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

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

这些要求对于启用检查点机制和工作流恢复是必要的。使用 Python 原语（如字典、列表、字符串、数字和布尔值）来确保您的输入和输出是可序列化的。

序列化确保工作流状态（如任务结果和中间值）可以可靠地保存和恢复。这对于启用人机协作交互、容错和并行执行至关重要。

提供不可序列化的输入或输出将在配置了检查点机制的工作流中导致运行时错误。

## 确定性

要利用**人机协作**等功能，任何随机性都应封装在**任务**内部。这保证了当执行暂停（例如，为了人机协作）然后恢复时，它将遵循相同的\_步骤序列\_，即使**任务**结果是非确定性的。

LangGraph 通过在执行过程中持久化**任务**和[**子图**](/oss/python/langgraph/use-subgraphs)结果来实现此行为。精心设计的工作流确保恢复执行遵循\_相同的步骤序列\_，从而可以正确检索先前计算的结果，而无需重新执行它们。这对于长时间运行的**任务**或具有非确定性结果的**任务**特别有用，因为它避免了重复先前完成的工作，并允许从本质上相同的状态恢复。

虽然工作流的不同运行可能产生不同的结果，但恢复**特定**运行应始终遵循相同的已记录步骤序列。这使得 LangGraph 可以高效地查找在图中断之前执行的**任务**和**子图**结果，并避免重新计算它们。

## 幂等性

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

## 常见陷阱

### 处理副作用

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

<Tabs>
  <Tab title="错误">
    在此示例中，副作用（写入文件）直接包含在工作流中，因此在恢复工作流时将第二次执行。

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    @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:  # [!code highlight]
            f.write("Side effect executed")  # [!code highlight]
        value = interrupt("question")
        return value
    ```
  </Tab>

  <Tab title="正确">
    在此示例中，副作用被封装在一个任务中，确保在恢复时一致执行。

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.func import task

    @task  # [!code highlight]
    def write_to_file():  # [!code highlight]
        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
    ```
  </Tab>
</Tabs>

### 非确定性控制流

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

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

当使用具有多个中断调用的**人机协作**工作流时，这一点尤其重要。LangGraph 为每个任务/入口点维护一个恢复值列表。当遇到中断时，它会与相应的恢复值匹配。此匹配严格基于**索引**，因此恢复值的顺序应与中断的顺序匹配。

如果在恢复时未保持执行顺序，一个 [`interrupt`](https://reference.langchain.com/python/langgraph/types/interrupt) 调用可能会与错误的 `resume` 值匹配，导致结果不正确。

请阅读[确定性](#确定性)部分以获取更多详情。

<Tabs>
  <Tab title="错误">
    在此示例中，工作流使用当前时间来确定执行哪个任务。这是非确定性的，因为工作流的结果取决于其执行的时间。

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.func import entrypoint

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

        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
        }
    ```
  </Tab>

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

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    import time

    from langgraph.func import task

    @task  # [!code highlight]
    def get_time() -> float:  # [!code highlight]
        return time.time()

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

        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
        }
    ```
  </Tab>
</Tabs>

## 了解更多

* [如何使用功能 API](/oss/python/langgraph/use-functional-api)
* [图 API 概念概览](/oss/python/langgraph/graph-api)
* [在图 API 和功能 API 之间选择](/oss/python/langgraph/choosing-apis)

***

<div className="source-links">
  <Callout icon="terminal-2">
    [将这些文档连接](/use-these-docs)到 Claude、VSCode 等，通过 MCP 获取实时答案。
  </Callout>

  <Callout icon="edit">
    [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/functional-api.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。
  </Callout>
</div>