Functional API 概述

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

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

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

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

Functional API 与 Graph API 对比

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

控制流：Functional API 不需要考虑图结构。您可以使用标准的 Python 构造来定义工作流。这通常会减少您需要编写的代码量。
短期记忆：Graph API 需要声明一个 State，并且可能需要定义 reducers 来管理图状态的更新。@entrypoint 和 @tasks 不需要显式的状态管理，因为它们的状态作用域限于该函数，并且不在函数之间共享。
检查点：两种 API 都会生成和使用检查点。在 Graph API 中，每个超步后都会生成一个新的检查点。在 Functional API 中，当任务执行时，它们的结果会保存到与给定入口点关联的现有检查点中，而不是创建一个新的检查点。
可视化：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

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:
    """撰写一篇关于给定主题的文章。"""
    time.sleep(1)  # 这是一个长时间运行任务的占位符。
    return f"An essay about topic: {topic}"


@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """一个简单的工作流，用于撰写文章并请求审核。"""
    essay = write_essay("cat").result()
    is_approved = interrupt(
        {
            # 作为参数传递给 interrupt 的任何可 JSON 序列化的负载。
            # 当从工作流流式传输数据时，它将在客户端侧作为中断显示。
            "essay": essay,  # 我们希望审核的文章。
            # 我们可以添加任何需要的额外信息。
            # 例如，引入一个名为“action”的键，并附带一些指示。
            "action": "Please approve/reject the essay",
        }
    )
    return {
        "essay": essay,  # 生成的文章
        "is_approved": is_approved,  # 来自 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'
#     )
#   ]
# }

一篇论文已撰写完成，准备审阅。提供审阅后，我们可以恢复工作流：

# 从用户处获取审阅（例如通过界面）
# 此处使用布尔值表示，但可以是任何可序列化为JSON的值。
human_review = True

resumed_stream = workflow.stream_events(Command(resume=human_review), config, version="v3")
print(resumed_stream.output)
# {'essay': '关于主题：猫的文章', 'is_approved': True}

工作流已完成，审阅已添加到论文中。

Entrypoint

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

定义

一个 entrypoint 通过使用 @entrypoint 装饰器装饰一个函数来定义。该函数必须接受一个位置参数，该参数用作工作流输入。如果需要传递多个数据，请对第一个参数使用字典作为输入类型。使用 entrypoint 装饰一个函数会生成一个 Pregel 实例，该实例有助于管理工作的执行（例如，处理流式传输、恢复和检查点）。通常，您会希望将一个 checkpointer 传递给 @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

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	在使用 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 方法执行。

调用
异步调用
流
异步流

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"
    }
}

stream = my_workflow.stream_events(some_input, config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

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

stream = await my_workflow.astream_events(some_input, config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

恢复

在 interrupt 之后恢复执行，可以通过将恢复值传递给 Command 原语来完成。

调用
异步调用
流
异步流

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"
    }
}

stream = my_workflow.stream_events(Command(resume=some_resume_value), config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

from langgraph.types import Command

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

stream = await my_workflow.astream_events(Command(resume=some_resume_value), config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

出错后恢复 要在出错后恢复，请使用 None 和相同的线程 ID（config）运行 entrypoint。这假设根本的错误已解决，并且执行可以成功继续。

调用
异步调用
流
异步流

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"
    }
}

stream = my_workflow.stream_events(None, config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

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

stream = await my_workflow.astream_events(None, config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

短期记忆

当使用 checkpointer 定义 entrypoint 时，它会在同一线程 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.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)

Task

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

异步执行：任务设计为可异步执行，允许多个操作并发运行而不阻塞。
检查点：任务结果被保存到检查点，使得工作流可以从最后保存的状态恢复。（更多详情参见持久性）。

定义

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

from langgraph.func import task

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

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

执行

任务只能从 entrypoint、另一个 task 或状态图节点内部调用。任务不能直接从主应用程序代码调用。当您调用一个 task 时，它会立即返回一个 future 对象。Future 是一个占位符，代表稍后可用的结果。要获取 task 的结果，您可以同步等待它（使用 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

何时使用 task

Tasks 在以下场景中很有用：

检查点：当您需要将长时间运行的操作结果保存到检查点，以便在恢复工作流时无需重新计算。
人机交互：如果您正在构建需要人工干预的工作流，您必须使用 tasks 来封装任何随机性（例如，API 调用），以确保工作流可以正确恢复。更多详情参见确定性部分。
并行执行：对于 I/O 密集型任务，tasks 支持并行执行，允许多个操作并发运行而不阻塞（例如，调用多个 API）。
可观察性：将操作包装在 tasks 中，提供了一种跟踪工作流进展和使用 LangSmith 监控单个操作执行的方法。
可重试工作：当需要重试工作以处理故障或不一致时，tasks 提供了一种封装和管理重试逻辑的方式。

序列化

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

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

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

确定性

当您恢复工作流运行时，代码不会从执行停止的同一行代码恢复。执行返回到一个检查点边界，并且工作流回放直到再次到达暂停点。对于 Functional API，回放从 entrypoint 的开头开始，而 LangGraph 从检查器恢复已完成的 task 和子图结果，而不是重新计算它们。这保留了跨暂停的记录步骤顺序，包括长时间运行或非确定性的 task 输出。要使用人机交互等功能，您必须将非确定性工作（例如，随机值）和副作用（例如，文件写入或 API 调用）放在 tasks 中。工作流的不同运行可能产生不同的结果，但恢复特定线程应该回放相同的持久化 task 和子图结果。为确保您的工作流是确定性的并且可以一致地回放，请遵循以下准则：

避免重复工作：在 entrypoint 中，如果您链接了多个副作用（例如，日志记录、文件写入或网络调用），请为每个副作用创建单独的 task，以便恢复时从检查点恢复它们的输出，而不是再次运行。
封装非确定性操作：将可能在两次尝试之间变化的值（例如，随机数或系统时间读取）放在 tasks 内部，以便回放与检查点的内容对齐。
使用幂等操作：对于部分任务失败和重试，请参阅幂等性。

幂等性

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

常见陷阱

处理副作用

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

错误
正确

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

@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

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

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

非确定性控制流

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

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

这在使用具有多个 interrupt 调用的人机交互工作流时尤其重要。LangGraph 为每个 task/entrypoint 维护一个恢复值列表。当遇到一个 interrupt 时，它会与相应的恢复值进行匹配。这种匹配是严格基于索引的，因此恢复值的顺序应与 interrupt 的顺序匹配。如果在恢复时执行顺序未被维持，一个 interrupt 调用可能会与错误的 resume 值匹配，导致错误的结果。请阅读确定性部分以了解更多细节。

错误
正确

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

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 来确定要执行哪个 task。这是确定性的，因为工作流的结果仅取决于输入。

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.

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

Functional API 与 Graph API 对比

示例

Entrypoint

定义

可注入参数

执行

恢复

短期记忆

`entrypoint.final`

Task

定义

执行

何时使用 task

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多

​Functional API 与 Graph API 对比

​示例

​Entrypoint

​定义

​可注入参数

​执行

​恢复

​短期记忆

​entrypoint.final

​Task

​定义

​执行

​何时使用 task

​序列化

​确定性

​幂等性

​常见陷阱

​处理副作用

​非确定性控制流

​了解更多

Functional API 与 Graph API 对比

示例

Entrypoint

定义

可注入参数

执行

恢复

短期记忆

`entrypoint.final`

Task

定义

执行

何时使用 task

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多