跳到内容

功能性 API

名称 描述
entrypoint

使用 entrypoint 装饰器定义 LangGraph 工作流。

函数

名称 描述
task

使用 task 装饰器定义 LangGraph 任务。

entrypoint

使用 entrypoint 装饰器定义 LangGraph 工作流。

函数签名

装饰函数必须接受一个单一参数,该参数用作函数的输入。此输入参数可以是任何类型。使用字典可以向函数传递多个参数

可注入参数

装饰函数可以请求访问运行时自动注入的额外参数。这些参数包括

参数 描述
store BaseStore 的实例。对于长期记忆很有用。
writer 用于将自定义数据写入流的 StreamWriter 实例。
config 一个配置对象(也称为 RunnableConfig),用于存储运行时配置值。
previous 给定线程的先前返回值(仅在提供了检查点时可用)。

entrypoint 装饰器可以应用于同步函数或异步函数。

状态管理

previous 参数可用于访问同一线程 ID 上 entrypoint 上一次调用的返回值。仅在提供了检查点时,此值才可用。

如果您希望 previous 不同于返回值,可以使用 entrypoint.final 对象在返回一个值的同时将另一个不同的值保存到检查点。

参数

名称 类型 描述 默认值
checkpointer Optional[BaseCheckpointSaver]

指定一个检查点以创建可在多次运行中持久化其状态的工作流。

None
store Optional[BaseStore]

一个通用的键值存储。某些实现可能通过可选的 index 配置支持语义搜索功能。

None
config_schema Optional[type[Any]]

指定将传递给工作流的配置对象的模式。

None
使用 entrypoint 和 tasks
import time

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

@task
def compose_essay(topic: str) -> str:
    time.sleep(1.0)  # Simulate slow operation
    return f"An essay about {topic}"

@entrypoint(checkpointer=MemorySaver())
def review_workflow(topic: str) -> dict:
    """Manages the workflow for generating and reviewing an essay.

    The workflow includes:
    1. Generating an essay about the given topic.
    2. Interrupting the workflow for human review of the generated essay.

    Upon resuming the workflow, compose_essay task will not be re-executed
    as its result is cached by the checkpointer.

    Args:
        topic: The subject of the essay.

    Returns:
        dict: A dictionary containing the generated essay and the human review.
    """
    essay_future = compose_essay(topic)
    essay = essay_future.result()
    human_review = interrupt({
        "question": "Please provide a review",
        "essay": essay
    })
    return {
        "essay": essay,
        "review": human_review,
    }

# Example configuration for the workflow
config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# Topic for the essay
topic = "cats"

# Stream the workflow to generate the essay and await human review
for result in review_workflow.stream(topic, config):
    print(result)

# Example human review provided after the interrupt
human_review = "This essay is great."

# Resume the workflow with the provided human review
for result in review_workflow.stream(Command(resume=human_review), config):
    print(result)
访问先前的返回值

启用检查点后,函数可以访问同一线程 ID 上上一次调用的先前返回值。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint

@entrypoint(checkpointer=MemorySaver())
def my_workflow(input_data: str, previous: Optional[str] = None) -> str:
    return "world"

config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}
my_workflow.invoke("hello")
使用 entrypoint.final 保存值

entrypoint.final 对象允许您返回一个值,同时将另一个不同的值保存到检查点。只要使用相同的线程 ID,在下一次调用 entrypoint 时,可以通过 previous 参数访问此值。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint

@entrypoint(checkpointer=MemorySaver())
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": "some_thread"
    }
}

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

名称 描述
final

一个可以从 entrypoint 返回的原语。

方法

名称 描述
__init__

初始化 entrypoint 装饰器。

__call__

将函数转换为 Pregel 图。

final dataclass

Bases: Generic[R, S]

一个可以从 entrypoint 返回的原语。

此原语允许将一个与 entrypoint 返回值不同的值保存到检查点。

解耦返回值和保存值
from langgraph.checkpoint.memory import MemorySaver
from langgraph.func import entrypoint

@entrypoint(checkpointer=MemorySaver())
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)

属性

名称 类型 描述
value R

要返回的值。即使是 None,也会始终返回一个值。

save S

下一个检查点状态的值。

value instance-attribute

value: R

要返回的值。即使是 None,也会始终返回一个值。

save instance-attribute

save: S

下一个检查点状态的值。

即使是 None,也会始终保存一个值。

__init__

__init__(
    checkpointer: Optional[BaseCheckpointSaver] = None,
    store: Optional[BaseStore] = None,
    config_schema: Optional[type[Any]] = None,
) -> None

初始化 entrypoint 装饰器。

__call__

__call__(func: Callable[..., Any]) -> Pregel

将函数转换为 Pregel 图。

参数

名称 类型 描述 默认值
func Callable[..., Any]

要转换的函数。支持同步和异步函数。

required

返回

类型 描述
Pregel

一个 Pregel 图。

task

task(
    __func_or_none__: Optional[
        Union[Callable[P, Awaitable[T]], Callable[P, T]]
    ] = None,
    *,
    name: Optional[str] = None,
    retry: Optional[
        Union[RetryPolicy, Sequence[RetryPolicy]]
    ] = None
) -> Union[
    Callable[
        [
            Union[
                Callable[P, Awaitable[T]],
                Callable[P, T],
            ]
        ],
        Callable[P, SyncAsyncFuture[T]],
    ],
    Callable[P, SyncAsyncFuture[T]],
]

使用 task 装饰器定义 LangGraph 任务。

异步函数需要 Python 3.11 或更高版本

task 装饰器支持同步和异步函数。要使用异步函数,请确保您正在使用 Python 3.11 或更高版本。

任务只能从 entrypoint 或 StateGraph 中调用。任务可以像常规函数一样调用,但有以下几点不同

  • 启用检查点后,函数输入和输出必须可序列化。
  • 装饰函数只能从 entrypoint 或 StateGraph 中调用。
  • 调用该函数会产生一个 future。这使得并行化任务变得容易。

参数

名称 类型 描述 默认值
retry Optional[Union[RetryPolicy, Sequence[RetryPolicy]]]

任务失败时使用的可选重试策略。

None

返回

类型 描述
Union[Callable[[Union[Callable[P, Awaitable[T]], Callable[P, T]]], Callable[P, SyncAsyncFuture[T]]], Callable[P, SyncAsyncFuture[T]]]

用作装饰器时可调用的函数。

同步任务
from langgraph.func import entrypoint, task

@task
def add_one(a: int) -> int:
    return a + 1

@entrypoint()
def add_one(numbers: list[int]) -> list[int]:
    futures = [add_one(n) for n in numbers]
    results = [f.result() for f in futures]
    return results

# Call the entrypoint
add_one.invoke([1, 2, 3])  # Returns [2, 3, 4]
异步任务
import asyncio
from langgraph.func import entrypoint, task

@task
async def add_one(a: int) -> int:
    return a + 1

@entrypoint()
async def add_one(numbers: list[int]) -> list[int]:
    futures = [add_one(n) for n in numbers]
    return asyncio.gather(*futures)

# Call the entrypoint
await add_one.ainvoke([1, 2, 3])  # Returns [2, 3, 4]

评论