LangGraph vs Temporal: Checkpointer, Activity or Plugin

Shape 1 is a persistent checkpointer, idempotent tools, and some process that re-invokes stalled threads. Shape 2 wraps graph.ainvoke in a single Temporal Activity so Temporal owns scheduling and retries for the run as a whole. Shape 3 is the LangGraphPlugin.

Shape 2 looks like the cheap way to add Temporal. An Activity is the unit Temporal retries. If a worker dies at node 25 of 30, Temporal retries the Activity, and the Activity starts the graph again from the top, re-running 24 nodes and their side effects. The only way to avoid that is to keep a persistent checkpointer inside the Activity and resume the thread on retry, at which point you are running shape 1 with Temporal as the scheduler. The Activity also needs a start_to_close_timeout longer than your slowest run, and a human approval wait inside it means a worker slot held for days. Shape 2 is a migration step, not a destination.

For every node in a registered StateGraph, the plugin replaces the node's runnable with its own wrapper. A node marked execute_in: "activity" is registered as a Temporal Activity named <graph_name>.<node_name>, and the Workflow schedules it with workflow.execute_activity. A node marked execute_in: "workflow" runs inline in Workflow code. The plugin covers both the Graph API (StateGraph) and the Functional API (@entrypoint and @task).

The rules that bite on day one, all from _plugin.py and the README at tag python/langgraph/v0.1.0:

Here is a graph with model and tool nodes as Activities carrying Temporal timeouts and retries, a pure merge step in the Workflow, and an async conditional edge:

from datetime import timedelta
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from temporalio.common import RetryPolicy
from temporalio.langgraph import LangGraphPlugin
from temporalio.worker import Worker


class State(TypedDict):
    messages: list
    pending_tool_calls: list


async def call_model(state: State) -> dict: ...      # network call: must be an Activity
async def run_tools(state: State) -> dict: ...       # side effects: must be an Activity
def merge_findings(state: State) -> dict: ...        # pure, deterministic: can run in the Workflow


async def should_continue(state: State) -> str:      # conditional edges always run in the Workflow
    return "tools" if state["pending_tool_calls"] else "merge"


g = StateGraph(State)
g.add_node("model", call_model, metadata={
    "execute_in": "activity",
    "start_to_close_timeout": timedelta(minutes=2),
    "retry_policy": RetryPolicy(maximum_attempts=3),
})
g.add_node("tools", run_tools, metadata={
    "execute_in": "activity",
    "start_to_close_timeout": timedelta(minutes=5),
})
g.add_node("merge", merge_findings, metadata={"execute_in": "workflow"})
g.add_edge(START, "model")
g.add_conditional_edges("model", should_continue, ["tools", "merge"])
g.add_edge("tools", "model")
g.add_edge("merge", END)

plugin = LangGraphPlugin(
    graphs={"review-agent": g},
    default_activity_options={"start_to_close_timeout": timedelta(minutes=1)},
)

# worker = Worker(client, task_queue="agents", workflows=[ReviewWorkflow], plugins=[plugin])

There is no retry_policy= argument on add_node: that is the LangGraph retry policy, and the plugin refuses it.

The plugin's README is explicit: if your code requires a checkpointer, for example because it uses interrupts, use InMemorySaver; Temporal handles durability, so PostgreSQL or Redis checkpointers are not needed. The plugin's own interrupt and continue-as-new tests compile with InMemorySaver().

Under the plugin, LangGraph's Pregel loop, the code that decides which nodes run next and calls the checkpointer, runs in Workflow code. A Postgres checkpointer would issue network writes from inside the Workflow, which is exactly what replay forbids. Temporal's history is now the durable record, and the in-memory saver only exists so LangGraph's interrupt machinery has somewhere to keep state during a single replay.

That leaves a real gap if you used checkpoints as session history for a UI, an audit export or analytics. Issue 1803 on the Temporal Python SDK asks precisely this: the reporter ran agents with AsyncPostgresSaver for permanent session history and wanted the plugin for runs of up to an hour. The maintainer's reply opens with "Two untested ideas": stream LangGraph's updates and values streams through Temporal Workflow Streams to an Activity that writes to Postgres, or a custom saver that wraps InMemorySaver and mirrors each checkpoint write through an Activity. Treat both as suggestions, not supported patterns. The first also needs a manual astream() bridge, because streaming_topic only captures stream_mode="custom".

The approach the reporter of issue 1894 proposed as their leading fix is simpler and does not touch Workflow code: each node's Activity writes its own delta to Postgres before it returns. Because Temporal retries Activities, that write runs at least once, so key it on the Workflow run ID plus the Activity ID, both available from the Activity's info, and make it an upsert. A run that completes but silently lost its step history is a failure mode worth monitoring for; our post on detecting silent AI agent failure covers how to check the record rather than the run's own report.

If you stay on shape 1, none of this applies, and the work that matters is getting the saver right: connection pooling, PgBouncer, SSL and retention, covered in our LangGraph Postgres checkpointer production guide.

The mechanism comes straight from _activity.py: for every Activity node, the plugin builds an ActivityInput whose arguments are what LangGraph hands that node, which for a StateGraph node is its input state, plus a stripped, serializable copy of the config. The node's return value comes back as the Activity result. Both are written into Workflow history. So the size of your graph state sets the size of every node's payload, and the sum of those payloads sets how fast history approaches its limit. A 2 MB cap is per payload, and the 4 MB gRPC limit applies to the whole request, so a Workflow can hit it even when every individual payload is under 2 MB.

In issue 1894, the reporter's agents had 20 to 30 nodes. After dropping the Postgres checkpointer, they added one Activity at the end of the run that received the full accumulated final state to persist it. On a larger run that Activity failed to schedule:

BadScheduleActivityAttributes: ScheduleActivityTaskCommandAttributes.Input exceeds size limit.

The issue body says all 29 node Activities completed fine. The plugin's per-node payloads were not the problem in that run; the single end-of-run payload was. The reporter says the server terminated the Workflow (WORKFLOW_EXECUTION_TERMINATED), with all its completed work behind it. Temporal's docs describe termination as the behavior of older SDKs; for Python SDK 1.23.0 and later they say the SDK fails the Workflow Task with WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE and the Workflow stays open. Either way the run stops making progress, so alert on both outcomes.

The fixes, in the order to apply them:

External Storage at temporalio 1.35.0 is configured on the data converter:

import dataclasses

import aioboto3
from temporalio.client import Client
from temporalio.contrib.aws.s3driver import S3StorageDriver
from temporalio.contrib.aws.s3driver.aioboto3 import new_aioboto3_client
from temporalio.converter import DataConverter, ExternalStorage


async def main() -> None:
    session = aioboto3.Session()
    async with session.client("s3") as s3_client:
        driver = S3StorageDriver(
            client=new_aioboto3_client(s3_client),
            bucket="agent-payloads",
        )
        client = await Client.connect(
            "temporal.internal:7233",
            data_converter=dataclasses.replace(
                DataConverter.default,
                external_storage=ExternalStorage(drivers=[driver]),
            ),
        )
        ...  # run workers and start workflows here, while s3_client is open

Three caveats. ExternalStorage and the S3 driver are both marked experimental, and Temporal's docs say the APIs may change before general availability. The same data converter must be configured on every client and every worker, or a worker will find a reference where it expected a payload. And it requires the temporalio[aioboto3] extra. Compression is not a fix: Temporal's docs call it a short-term measure that stops working as payloads keep growing.

The maintainer's answer on issue 1894 combined these: External Storage to clear the limit, then an Activity that copies the stored data to Postgres, or a custom External Storage driver that writes to Postgres directly to avoid storing it twice. A custom driver means subclassing StorageDriver and implementing name(), store() and retrieve().

Everything in shape 3 can run inside your perimeter under MIT licenses: the Temporal server, the temporalio SDK, temporalio-langgraph and LangGraph. The LangGraph Agent Server, the langgraph-api package, declares Elastic-2.0, and you do not need it for any of the three shapes.

For a bank, insurer or hospital, the decision changes where agent state sits. With a checkpointer, it is rows in your Postgres, governed by the retention job and access controls you already have. With the plugin, every Activity node's input state and output is a payload in Temporal history, and anything offloaded is an object in the External Storage bucket. If a run touches patient records or loan files, those records are now in both places.

Self-hosting also gives you the limit knobs. limit.blobSize.error defaults to 2 MiB in the server's dynamic config at 1.32.2 and can be raised per namespace; Temporal Cloud fixes it at 2 MB. Cloud also places history outside your network unless every payload is encrypted client-side by your codec. For External Storage on-prem, the S3 driver accepts any client that implements its client interface, so you can point it at an S3-compatible store your team operates or write a custom StorageDriver against storage you already run. Test either against your own store before relying on it.

Our verdict: for long-running agents with side effects or approval waits, use the LangGraphPlugin and give up the Postgres checkpointer rather than trying to keep both; a Postgres saver called from Workflow code breaks replay anyway. For short agents, do not add Temporal at all; a checkpointer plus idempotent tools is the smaller system and fails in simpler ways. More patterns for building and running agents are collected on our AI agents pillar page.

This week, list every node in your graph. Mark each one that makes a network call, touches a file or writes to another system: those become Activities, and the rest can run in the Workflow. Mark every node with code before an interrupt() and check that it is safe to run twice. Then serialize the largest state object from a recent production run and measure it in bytes. If it is anywhere near 2 MB, decide between moving tool outputs out of state and enabling External Storage before you write a line of plugin code.