Define a LangGraph task using the task decorator.
The task decorator supports both sync and async functions. To use async
functions, ensure that you are using Python 3.11 or higher.
Tasks can only be called from within an entrypoint or
from within a StateGraph. A task can be called like a regular function with the
following differences:
StateGraph.task(
__func_or_none__: Callable[P, Awaitable[T]] | Callable[P, T] | None = None,
*,
name: str | None = None,
retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
**kwargs: Unpack[DeprecatedKwargs] = {}
) -> Callable[[Callable[P, Awaitable[T]] | Callable[P, T]], _TaskFunction[P, T]] | _TaskFunction[P, T]Sync Task:
from langgraph.func import entrypoint, task
@task
def add_one_task(a: int) -> int:
return a + 1
@entrypoint()
def add_one(numbers: list[int]) -> list[int]:
futures = [add_one_task(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]Async Task:
import asyncio
from langgraph.func import entrypoint, task
@task
async def add_one_task(a: int) -> int:
return a + 1
@entrypoint()
async def add_one(numbers: list[int]) -> list[int]:
futures = [add_one_task(n) for n in numbers]
return asyncio.gather(*futures)
# Call the entrypoint
await add_one.ainvoke([1, 2, 3]) # Returns [2, 3, 4]| Name | Type | Description |
|---|---|---|
name | str | None | Default: NoneAn optional name for the task. If not provided, the function name will be used. |
retry_policy | RetryPolicy | Sequence[RetryPolicy] | None | Default: NoneAn optional retry policy (or list of policies) to use for the task in case of a failure. |
cache_policy | CachePolicy[Callable[P, str | bytes]] | None | Default: NoneAn optional cache policy to use for the task. This allows caching of the task results. |