Worked Examples

Three complete, runnable checkpoint programs — crash-resume across a multi-node flow, a human-in-the-loop approval that survives a process exit, and time travel through a Graph Orchestrator's checkpoint chain.

The examples catalog links out to the scripts in the dynamiq repository. This page keeps three complete programs inline so you can read a whole checkpoint scenario end to end without leaving the docs. Each one builds on the API described in Checkpoints; read that page first for the CheckpointConfig reference and backend table.

Every block below is self-contained — copy it into a file and run it. The two flows built from Python nodes need no API keys; the Graph Orchestrator example calls OpenAI and reads a PostgreSQL connection from the environment.

1. Crash and resume across a multi-node flow

A four-node ETL flow — input → extract → transform → output — writes a checkpoint after every node completes. We run it once, then rewind the checkpoint to the state it would hold if the worker had been killed mid-transform, and resume. On resume the completed nodes are skipped and their saved outputs feed the remaining work; passing input_data=None reuses the checkpoint's original_input.

from dynamiq.checkpoints import CheckpointBehavior, CheckpointConfig, CheckpointStatus
from dynamiq.checkpoints.backends import FileSystem
from dynamiq.flows import Flow
from dynamiq.nodes.node import NodeDependency
from dynamiq.nodes.tools import Python
from dynamiq.nodes.utils import Input, Output

backend = FileSystem(base_path=".checkpoints")

inp = Input(id="input", name="Input")
extract = Python(
    id="extract",
    name="extract",
    code="def run(input_data): return {'rows': input_data['batch'] * 100}",
    depends=[NodeDependency(inp)],
)
transform = Python(
    id="transform",
    name="transform",
    code=(
        "def run(input_data):\n"
        "    rows = input_data['extract']['output']['content']['rows']\n"
        "    return {'clean_rows': rows - 7}\n"
    ),
    depends=[NodeDependency(extract)],
)
out = Output(id="output", name="Output", depends=[NodeDependency(transform)])

flow = Flow(
    id="etl-flow",
    nodes=[inp, extract, transform, out],
    checkpoint=CheckpointConfig(
        enabled=True,
        backend=backend,
        behavior=CheckpointBehavior.APPEND,
    ),
)

# 1. Run the pipeline. Every node completes, and each completion is checkpointed.
flow.run_sync(input_data={"batch": 12})

# 2. Simulate a crash *after* `extract` but *before* `transform` finished: take the
#    latest checkpoint and roll the downstream nodes back to "not yet run". This is
#    the state a durable checkpoint holds if the worker is killed mid-`transform`.
cp = backend.get_latest_by_flow(flow.id)
for node_id in ["transform", "output"]:
    cp.node_states.pop(node_id, None)
    if node_id in cp.completed_node_ids:
        cp.completed_node_ids.remove(node_id)
cp.status = CheckpointStatus.ACTIVE
backend.save(cp)

print(cp.completed_node_ids)  # ['input', 'extract'] - the work that survived the crash

# 3. Resume. input_data=None reuses the checkpoint's original_input ({"batch": 12}).
#    `input` and `extract` are skipped and their saved outputs feed `transform`,
#    which re-runs together with `output`.
result = flow.run_sync(input_data=None, resume_from=cp.id)

print(result.status)                                     # RunnableStatus.SUCCESS
print(result.output["transform"]["output"]["content"])   # {'clean_rows': 1193}

On resume the flow loads the checkpoint, marks input and extract as already done, and re-hydrates their outputs into the run so transform can read input_data['extract']['output']['content']['rows'] without recomputing it. Only transform and output execute the second time. In a real deployment you would not rewind the checkpoint by hand — the process would simply die, leaving the last durable checkpoint with transform still pending, and the resume call would pick up from exactly there.

2. Human-in-the-loop approval that survives a process exit

When a HumanFeedbackTool in ask mode waits on a streaming input connection and no one replies within the configured timeout, the flow saves a PENDING_INPUT checkpoint and the run ends. The process can exit entirely. Later — from any process — you deliver the reply and resume; the stored approval is handed to the waiting node instead of prompting again.

The queue below stands in for the connection the platform holds open to a human. The isolated checkpoint config (only the input-timeout trigger enabled) keeps the pending-input snapshot as the latest checkpoint.

from queue import Queue

from dynamiq.checkpoints import CheckpointConfig
from dynamiq.checkpoints.backends import InMemory
from dynamiq.flows import Flow
from dynamiq.nodes.tools.human_feedback import (
    HFStreamingInputEventMessage,
    HFStreamingInputEventMessageData,
    HumanFeedbackAction,
    HumanFeedbackTool,
)
from dynamiq.types.feedback import FeedbackMethod
from dynamiq.types.streaming import StreamingConfig

FLOW_ID = "approval-flow"
NODE_ID = "approval-gate"

# The queue models the connection the platform holds open to a human. While it is
# empty the tool's input read blocks; after `timeout` seconds the wait gives up.
input_queue = Queue()

approval_gate = HumanFeedbackTool(
    id=NODE_ID,
    action=HumanFeedbackAction.ASK,
    input_method=FeedbackMethod.STREAM,
    output_method=FeedbackMethod.STREAM,
    streaming=StreamingConfig(enabled=True, input_queue=input_queue, timeout=2.0),
)

backend = InMemory()
flow = Flow(
    id=FLOW_ID,
    nodes=[approval_gate],
    checkpoint=CheckpointConfig(
        enabled=True,
        backend=backend,
        checkpoint_on_input_timeout_enabled=True,
        # Isolate the input-timeout trigger so the pending-input snapshot is the one
        # retained (an after-node or on-failure save would otherwise supersede it).
        checkpoint_after_node_enabled=False,
        checkpoint_on_failure_enabled=False,
    ),
)

# 1. Nobody answers within the timeout. The input wait times out, the flow saves a
#    PENDING_INPUT checkpoint, and the run ends as FAILURE - the process is now free
#    to exit; the checkpoint is what survives.
first = flow.run_sync(input_data={"input": "Approve refund of $120 to customer 88?"})
print(first.status)  # RunnableStatus.FAILURE

paused = backend.get_latest_by_flow(FLOW_ID)
print(paused.status)                       # CheckpointStatus.PENDING_INPUT
print(list(paused.pending_inputs.keys()))  # ['approval-gate']

# 2. The approval finally arrives - possibly days later, from a different process.
#    Deliver it onto the queue exactly as the streaming transport would.
input_queue.put(
    HFStreamingInputEventMessage(
        entity_id=NODE_ID,
        data=HFStreamingInputEventMessageData(content="approved"),
    ).model_dump_json()
)

# 3. Resume from the saved checkpoint. The waiting node receives the delivered
#    approval and completes instead of re-prompting.
second = flow.run_sync(
    input_data={"input": "Approve refund of $120 to customer 88?"},
    resume_from=paused.id,
)
print(second.status)  # RunnableStatus.SUCCESS

The first run fails on purpose: the timeout is the signal that no human is currently attached. What matters is the checkpoint it leaves behind — status = pending_input, with the waiting node recorded under pending_inputs. On the deployed platform this is the same state a run reaches when it is listed under GET /v1/runs?status=awaiting_input (see Human in the loop and the Runs API). When the reply is delivered and you resume, the HumanFeedbackTool reads the approval off the queue and the run completes.

3. Time travel through a Graph Orchestrator's checkpoint chain

With checkpoint_mid_agent_loop_enabled=True, a GraphOrchestrator saves an APPEND snapshot at every state transition, building a parent-linked chain you can walk with get_chain and resume from at any point. This example wires a two-state graph (draft → edit) backed by PostgreSQL, then resumes from the snapshot taken after draft so only edit re-runs.

Set POSTGRESQL_HOST/PORT/DATABASE/USER/PASSWORD and an OPENAI_API_KEY before running.

from dynamiq.checkpoints import CheckpointBehavior, CheckpointConfig
from dynamiq.checkpoints.backends import PostgreSQL as PostgresCheckpointBackend
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.connections import PostgreSQL as PostgresConn
from dynamiq.flows import Flow
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.agents.orchestrators.graph import END, START, GraphOrchestrator
from dynamiq.nodes.agents.orchestrators.graph_manager import GraphAgentManager
from dynamiq.nodes.llms import OpenAI

ORCH_ID = "copy-graph"

llm = OpenAI(connection=OpenAIConnection(), model="gpt-4o", temperature=0.2)

drafter = Agent(name="drafter", llm=llm, role="Draft a short product announcement.")
editor = Agent(name="editor", llm=llm, role="Tighten the draft into two crisp sentences.")

orchestrator = GraphOrchestrator(
    id=ORCH_ID,
    name="Copy graph",
    manager=GraphAgentManager(llm=llm),
)
orchestrator.add_state_by_tasks("draft", [drafter])
orchestrator.add_state_by_tasks("edit", [editor])
orchestrator.add_edge(START, "draft")
orchestrator.add_edge("draft", "edit")
orchestrator.add_edge("edit", END)

# PostgreSQL keeps the checkpoint chain durable across process restarts. The
# connection reads POSTGRESQL_HOST/PORT/DATABASE/USER/PASSWORD from the environment.
backend = PostgresCheckpointBackend(
    connection=PostgresConn(),
    table_name="flow_checkpoints",
    create_if_not_exist=True,
)

# Checkpointing is a Flow-level feature, so the orchestrator runs as a Flow node.
flow = Flow(
    id="copy-graph-flow",
    nodes=[orchestrator],
    checkpoint=CheckpointConfig(
        enabled=True,
        backend=backend,
        behavior=CheckpointBehavior.APPEND,
        checkpoint_mid_agent_loop_enabled=True,
    ),
)

try:
    # Each state transition inside the orchestrator saves an APPEND snapshot, so the
    # backend accumulates a chain: START -> draft -> edit -> END.
    flow.run_sync(input_data={"input": "Announce our new checkpoint API."})

    latest = backend.get_latest_by_flow(flow.id)
    chain = backend.get_chain(latest.id)  # newest first, following parent_checkpoint_id

    def state_of(cp):
        node = cp.node_states.get(ORCH_ID)
        if not node:
            return None
        iteration = node.internal_state.get("iteration") or {}
        return iteration.get("iteration_data", {}).get("current_state_id")

    for cp in chain:
        print(cp.id[:8], cp.status.value, "next_state=", state_of(cp))

    # Time travel: find the snapshot taken after `draft` (its next state is `edit`)
    # and resume from it. input_data=None reuses the original input; the orchestrator
    # restores its chat history and current_state_id and continues from `edit`.
    after_draft = next(cp for cp in chain if state_of(cp) == "edit")
    resumed = flow.run_sync(input_data=None, resume_from=after_draft.id)
    print(resumed.status)  # RunnableStatus.SUCCESS
finally:
    backend.close()

The orchestrator implements the IterativeCheckpointMixin, so each snapshot carries the loop's iteration state — including current_state_id, the state the orchestrator would enter next. Walking the chain from newest to oldest gives you every decision point; resuming from any of them rebuilds the orchestrator's chat history and position and continues from there. Because PostgreSQL persists the chain, this time-travel works across process restarts, not just within one run.

1. Crash and resume across a multi-node flow

2. Human-in-the-loop approval that survives a process exit

3. Time travel through a Graph Orchestrator's checkpoint chain

Next steps

Checkpoints

Examples catalog

Human in the loop (platform)

On this page