跳到主要内容
功能性 API 允许您将 LangGraph 的关键功能——持久化内存人工干预流式传输——添加到您的应用程序中,并且对现有代码的修改最小。 它旨在将这些功能集成到可能使用标准语言原语进行分支和控制流(如 if 语句、for 循环和函数调用)的现有代码中。与许多要求将代码重构为显式管道或 DAG 的数据编排框架不同,功能性 API 允许您在不强制执行严格执行模型的情况下整合这些功能。 功能性 API 使用两个关键构建块:
  • @entrypoint – 将函数标记为工作流的起点,封装逻辑并管理执行流,包括处理长时间运行的任务和中断。
  • @task – 表示一个离散的工作单元,例如 API 调用或数据处理步骤,可以在入口点内异步执行。任务返回一个类似 Future 的对象,可以被等待或同步解析。
这为构建具有状态管理和流式传输功能的工作流提供了最少的抽象。
有关如何使用功能性 API 的信息,请参阅使用功能性 API

功能性 API 与图 API

对于喜欢更具声明性方法的开发人员,LangGraph 的图 API 允许您使用图范式定义工作流。这两个 API 共享相同的底层运行时,因此您可以在同一个应用程序中一起使用它们。 以下是一些主要区别:
  • 控制流:功能性 API 不需要考虑图结构。您可以使用标准 Python 构造来定义工作流。这通常会减少您需要编写的代码量。
  • 短期内存图 API 要求声明状态,并且可能需要定义化简器来管理对图状态的更新。@entrypoint@tasks 不需要显式状态管理,因为它们的状态作用域限于函数,并且不跨函数共享。
  • 检查点:两个 API 都生成并使用检查点。在图 API 中,每个超步骤之后都会生成一个新的检查点。在功能性 API 中,当任务执行时,它们的結果将保存到与给定入口点关联的现有检查点,而不是创建新的检查点。
  • 可视化:图 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
    }
此工作流将撰写一篇关于“猫”主题的论文,然后暂停以获得人工审查。工作流可以无限期中断,直到提供审查。当工作流恢复时,它从头开始执行,但由于 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 装饰器,以启用持久性并使用诸如人工干预之类的功能。
  • 同步
  • 异步
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
序列化 入口点的输入输出必须是 JSON 可序列化的,以支持检查点。有关更多详细信息,请参阅序列化部分。

可注入参数

在声明 entrypoint 时,您可以请求访问在运行时自动注入的附加参数。这些参数包括:
参数描述
previous访问与给定线程的先前 checkpoint 关联的状态。请参阅短期内存
store[BaseStore][langgraph.store.base.BaseStore] 的实例。对于长期内存很有用。
writer在 Async Python < 3.11 中使用 StreamWriter。有关详细信息,请参阅使用功能性 API 进行流式传输
配置用于访问运行时配置。有关信息,请参阅RunnableConfig
使用适当的名称和类型注解声明参数。
from langchain_core.runnables import RunnableConfig
from langgraph.func import entrypoint
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore

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

@entrypoint(
    checkpointer=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对象,可以使用invokeainvokestreamastream方法执行。
  • 调用
  • 异步调用
  • 流式处理
  • 异步流
config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
my_workflow.invoke(some_input, config)  # Wait for the result synchronously

恢复

中断后恢复执行可以通过将resume值传递给Command原语来完成。
  • 调用
  • 异步调用
  • 流式处理
  • 异步流
from langgraph.types import Command

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

my_workflow.invoke(Command(resume=some_resume_value), config)
错误后恢复 要在错误后恢复,请使用 None 和相同的线程 ID(配置)运行 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.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 可序列化的,以支持检查点。

执行

任务只能从入口点、另一个任务状态图节点内部调用。 任务不能直接从主应用程序代码中调用。 当您调用任务时,它会立即返回一个 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 基本类型,如字典、列表、字符串、数字和布尔值,以确保您的输入和输出是可序列化的。 序列化确保工作流状态(例如任务结果和中间值)可以可靠地保存和恢复。这对于启用人工干预、容错和并行执行至关重要。 提供不可序列化的输入或输出将在工作流配置了检查点时导致运行时错误。

确定性

为了利用人工干预等功能,任何随机性都应封装在任务内部。这保证了当执行暂停(例如,用于人工干预)然后恢复时,它将遵循相同的步骤序列,即使任务结果是非确定性的。 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

非确定性控制流

每次可能给出不同结果的操作(如获取当前时间或随机数)应封装在任务中,以确保在恢复时返回相同的结果。
  • 在一个任务中:获取随机数 (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
    }

以编程方式连接这些文档到 Claude、VSCode 等,通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.