Implementation:Langchain ai Langgraph Entrypoint Final

Attribute	Value
API	`entrypoint.final` dataclass and `entrypoint.__init__` configuration
Workflow	Functional_API_Workflow
Type	API Doc
Repository	Langchain_ai_Langgraph
Source File	`libs/langgraph/langgraph/func/__init__.py`
Source Lines	L430-469 (`entrypoint.final`), L394-428 (`entrypoint.__init__`)

Overview

entrypoint.final is a dataclass nested within the entrypoint class that enables decoupling the return value of a workflow from the value persisted in the checkpoint. It has two fields: value (sent to the caller) and save (written to the checkpoint for use as previous on the next invocation). This page also covers the entrypoint.__init__ configuration parameters that control checkpointing, storage, caching, and retry behavior.

Description

entrypoint.final Dataclass

The entrypoint.final class (lines 430-469) is a generic dataclass parameterized by two type variables:

R -- the return value type (what the caller receives)
S -- the save value type (what is persisted in the checkpoint)

It is decorated with @dataclass(**_DC_KWARGS) and has two fields:

value: R -- the value returned to the caller. A value is always returned, even if it is None.
save: S -- the value saved to the checkpoint state. A value is always saved, even if it is None.

When the entrypoint's __call__ method builds the Pregel graph, it creates two internal mapper functions:

_pluck_return_value: If the returned object is an entrypoint.final instance, extracts .value; otherwise passes through unchanged.
_pluck_save_value: If the returned object is an entrypoint.final instance, extracts .save; otherwise passes through unchanged.

These mappers are attached to ChannelWriteEntry instances for the END and PREVIOUS channels respectively.

Return Type Annotation Handling

The entrypoint.__call__ method inspects the decorated function's return type annotation to determine the output and save types:

If the annotation is bare entrypoint.final (unparameterized), both types default to Any.
If the annotation is entrypoint.final[R, S], the two type arguments are extracted via get_args().
If the annotation is entrypoint.final with the wrong number of type arguments (not exactly 2), a TypeError is raised.
If the annotation is any other type (or absent), the same type is used for both output and save.

entrypoint.init Configuration

The __init__ method (lines 394-428) stores all decorator arguments for later use during graph construction:

Handles deprecated config_schema parameter by mapping it to context_schema with a deprecation warning.
Handles deprecated retry parameter by mapping it to retry_policy with a deprecation warning.
Stores all parameters as instance attributes: self.checkpointer, self.store, self.cache, self.cache_policy, self.retry_policy, self.context_schema.

Usage

from typing import Any
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def my_workflow(
    number: int,
    *,
    previous: Any = None,
) -> entrypoint.final[int, int]:
    previous = previous or 0
    # Return the old accumulated value; save the new one
    return entrypoint.final(value=previous, save=previous + number)

config = {"configurable": {"thread_id": "thread-1"}}
my_workflow.invoke(3, config)   # Returns 0, saves 3
my_workflow.invoke(7, config)   # Returns 3, saves 10
my_workflow.invoke(5, config)   # Returns 10, saves 15

Code Reference

Source Location

File	`libs/langgraph/langgraph/func/__init__.py`
`entrypoint.final` class	Lines 430-469
`entrypoint.__init__`	Lines 394-428
`_pluck_return_value`	Lines 501-503 (inner function in `__call__`)
`_pluck_save_value`	Lines 505-507 (inner function in `__call__`)

Signature

@dataclass
class entrypoint.final(Generic[R, S]):
    value: R
    save: S

class entrypoint(Generic[ContextT]):
    def __init__(
        self,
        checkpointer: BaseCheckpointSaver | None = None,
        store: BaseStore | None = None,
        cache: BaseCache | None = None,
        context_schema: type[ContextT] | None = None,
        cache_policy: CachePolicy | None = None,
        retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
    ) -> None

Import

from langgraph.func import entrypoint

# entrypoint.final is accessed as a class attribute
result = entrypoint.final(value="hello", save={"key": "data"})

I/O Contract

entrypoint.final Fields

Field	Type	Description
`value`	`R`	The value returned to the caller via `invoke()` or `stream()`. Always returned, even if `None`.
`save`	`S`	The value persisted in the checkpoint. Available as the `previous` parameter on the next invocation with the same thread ID. Always saved, even if `None`.

entrypoint.init Parameters

Parameter	Type	Default	Description
`checkpointer`	None	`None`	Enables state persistence across invocations. Required for `previous` parameter and interrupt/resume.
`store`	None	`None`	General-purpose key-value store for cross-session data. Accessible via `runtime` parameter.
`cache`	None	`None`	Cache backend for memoizing workflow/task results.
`context_schema`	None	`None`	Schema type for the static context object passed at invocation time.
`cache_policy`	None	`None`	Controls cache key generation and TTL.
`retry_policy`	Sequence[RetryPolicy] \| None	`None`	Automatic retry configuration for the workflow.

Internal State After init

Attribute	Type	Description
`self.checkpointer`	None	Stored checkpointer instance.
`self.store`	None	Stored store instance.
`self.cache`	None	Stored cache instance.
`self.cache_policy`	None	Stored cache policy.
`self.retry_policy`	Sequence[RetryPolicy] \| None	Stored retry policy.
`self.context_schema`	None	Stored context schema type.

Usage Examples

Basic Split Return/Save

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

@entrypoint(checkpointer=InMemorySaver())
def log_workflow(
    message: str,
    *,
    previous: list | None = None,
) -> entrypoint.final[str, list]:
    history = previous or []
    history.append(message)
    # Return latest message to caller, save full history
    return entrypoint.final(value=message, save=history)

config = {"configurable": {"thread_id": "log-1"}}
log_workflow.invoke("hello", config)   # Returns "hello", saves ["hello"]
log_workflow.invoke("world", config)   # Returns "world", saves ["hello", "world"]

Without entrypoint.final (Same Value for Return and Save)

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

@entrypoint(checkpointer=InMemorySaver())
def simple_counter(
    increment: int,
    *,
    previous: int | None = None,
) -> int:
    # Same value returned to caller AND saved to checkpoint
    return (previous or 0) + increment

config = {"configurable": {"thread_id": "counter-1"}}
simple_counter.invoke(5, config)  # Returns 5
simple_counter.invoke(3, config)  # Returns 8

Typed Return Annotation

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

# The type annotation entrypoint.final[str, dict] tells the runtime:
# - Return type is str
# - Save type is dict
@entrypoint(checkpointer=InMemorySaver())
def typed_workflow(
    query: str,
    *,
    previous: dict | None = None,
) -> entrypoint.final[str, dict]:
    state = previous or {"queries": [], "count": 0}
    state["queries"].append(query)
    state["count"] += 1
    summary = f"Processed {state['count']} queries"
    return entrypoint.final(value=summary, save=state)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment