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. |