函数式 API 概述

函数式 API 允许你以对现有代码的最小改动，将 LangGraph 的关键特性（持久化、记忆、人在回路 和 流式传输）添加到你的应用中。

它旨在将这些特性集成到可能使用标准语言原语进行分支和控制流的现有代码中，例如 if 语句、for 循环和函数调用。与许多要求将代码重构为显式管道或 DAG 的数据编排框架不同，函数式 API 允许你在不强制执行僵化执行模型的情况下融入这些能力。

函数式 API 使用两个关键的构建块：

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

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

TIP

有关如何使用函数式 API 的信息，请参见使用函数式 API。

函数式 API 与图 API

对于偏好更具声明式方法的用户，LangGraph 的图 API 允许你使用图范式定义工作流。两个 API 共享相同的底层运行时，因此你可以在同一个应用中将它们一起使用。

以下是一些关键差异：

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

详细解释

此工作流将撰写一篇关于主题 "cat" 的文章，然后暂停以获取人工审核。在提供审核之前，工作流可以被无限期地中断。

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

import time

from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt

@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}}
stream = workflow.stream_events("cat", config, version="v3")
_ = stream.output
print({"write_essay": stream.interrupts[0].value["essay"]})
print({"__interrupt__": stream.interrupts})
# {'write_essay': 'An essay about topic: cat'}
# {
#   '__interrupt__': [
#     Interrupt(
#       value={
#           'essay': 'An essay about topic: cat',
#           'action': 'Please approve/reject the essay'
#       },
#       id='369d44b3d93d4a631ae583367ac6b5cc'
#     )
#   ]
# }

文章已经写好并可供审核。一旦提供了审核，我们就可以恢复工作流：

# 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

resumed_stream = workflow.stream_events(Command(resume=human_review), config, version="v3")
print(resumed_stream.output)
# {'essay': 'An essay about topic: cat', 'is_approved': True}

工作流已完成，审核结果已添加到文章中。

Entrypoint

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

定义

entrypoint 通过使用 @entrypoint 装饰器装饰一个函数来定义。

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

使用 entrypoint 装饰一个函数会生成一个 Pregel 实例，它有助于管理工作流的执行（例如处理流式传输、恢复和检查点）。

你通常会希望向 @entrypoint 装饰器传入一个检查点器，以启用持久化并使用人在回路等特性。

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

WARNING

序列化 entrypoint 的 inputs 和 outputs 必须是 JSON 可序列化的，以支持检查点。更多详情请参见序列化一节。

可注入参数

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

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

WARNING

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

请求可注入参数

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 方法执行它。

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

恢复

在中断之后恢复执行，可以通过向 Command 原语传入一个 resume 值来完成。

from langgraph.types import Command

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

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

在错误后恢复

要在错误后恢复，请使用 None 和相同的线程 id（config）运行 entrypoint。

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

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

my_workflow.invoke(None, config)

短期记忆

当 entrypoint 使用 checkpointer 定义时，它会在检查点中存储同一 线程 id 上连续调用之间的信息。

这允许使用 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

WARNING

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

执行

任务 只能从 entrypoint、另一个 任务 或状态图节点内调用。

任务不能直接从主应用代码调用。

当你调用一个 任务 时，它会立即返回一个 future 对象。future 是一个占位符，表示稍后才可用的结果。

要获取 任务 的结果，你可以同步等待它（使用 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

何时使用任务

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

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

序列化

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

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

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

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

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

确定性

当你恢复一次工作流运行时，代码不会从执行停止的同一行代码处恢复。执行会回到一个检查点边界，工作流会向前重放，直到再次到达暂停处。

对于函数式 API，重放从 entrypoint 的开头开始，同时 LangGraph 会从检查点器恢复已完成的 任务 和 子图 结果，而不是重新计算它们。这保留了跨暂停的已记录步骤顺序，包括长时间运行或不确定性的 任务 输出。

要使用 人在回路 等特性，你必须将不确定性的工作（例如随机值）和副作用（例如文件写入或 API 调用）放入 任务 中。

工作流的不同运行可能产生不同的结果，但恢复一个特定线程时，应当重放出相同的已持久化的 任务 和 子图 结果。

为确保你的工作流具有确定性并能够被一致地重放，请遵循以下准则：

• 避免重复工作：在一个 entrypoint 中，如果你串联了多个副作用（例如日志记录、文件写入或网络调用），请为每一个赋予自己的 任务，这样恢复时会从检查点器恢复它们的输出，而不是再次运行它们。
• 封装不确定性操作：将可能在多次尝试之间变化的值（例如随机数或墙钟读取）保留在 任务 内，以便重放与检查点记录的内容一致。
• 使用幂等操作：关于部分任务失败和重试，请参见幂等性。

幂等性

幂等性确保多次运行同一操作会产生相同的结果。这有助于在某个步骤因失败而被重新运行时，防止重复的 API 调用和冗余处理。始终将 API 调用放在 任务 函数内以便检查点，并将它们设计为在重新执行时具有幂等性。 这对于会导致数据写入的操作尤为重要。 当工作流恢复时，LangGraph 会从检查点重放已完成的 任务 结果。一个已经开始但未完成的 任务 可能会在该次恢复时再次运行，因此请将副作用设计为幂等的。使用幂等键或验证现有结果，以避免意外的重复。

常见陷阱

处理副作用

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

在此示例中，副作用（写入文件）被直接包含在工作流中，因此当恢复工作流时它会被第二次执行。

@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

不确定性控制流

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

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

在使用带有多次中断调用的 人在回路 工作流时，这一点尤为重要。LangGraph 会为每个任务/入口点维护一个 resume 值列表。当遇到中断时，它会与对应的 resume 值进行匹配。这种匹配严格基于索引，因此 resume 值的顺序应与中断的顺序一致。

如果在恢复时未保持执行顺序，一次 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
    }

了解更多

• 如何使用函数式 API
• Graph API 概念性概述
• 在 Graph API 与函数式 API 之间进行选择

---