Python API: Workflows

Graphs of steps and agents: routes, joins, loops, pauses for people, and threads. Generated from the docstrings and type hints of criogenai 0.1.7. See also: Python API overview, the same module in TypeScript.

Workflows: a graph of Python functions and agents, run by the Rust engine.

You declare nodes and how they connect; the engine schedules them in steps (nodes that are ready together run concurrently), merges their updates into one state dict, checks the graph before the first run, and pauses a run for a person and continues it later.

from criogenai import END, Workflow

wf = Workflow()
wf.add_node("draft", lambda state: {"text": f"About {state['topic']}"})
wf.start("draft").edge("draft", END)
result = wf.compile().run_sync({"topic": "solar panels"})

from criogenai import (
    Workflow,
    CompiledWorkflow,
    WorkflowThread,
    END,
    NodeContext,
    suspend,
    goto,
    Interrupt,
    WorkflowResult,
)

Contents#

Name	Kind	Summary
`Workflow`	class	A graph of nodes (Python functions or agents), checked by `compile` and run by the Rust engine.
`CompiledWorkflow`	class	A checked workflow, ready to run; made by `Workflow.compile`.
`WorkflowThread`	class	One durable run history of a workflow, kept in storage.
`END`	constant	The end of a workflow: use it as the target of an edge, a route or `goto`.
`NodeContext`	class	What a node can know about its execution.
`suspend`	function	Return this from a node to pause the run until a person answers.
`goto`	function	Return this from a node to choose which nodes run next.
`Interrupt`	class	A paused node and what it asked for.
`WorkflowResult`	class	What a workflow run produced.

Workflow#

class Workflow:
    def __init__(self, reducers: Optional[Mapping[str, str]] = None)

A graph of nodes (Python functions or agents), checked by compile and run by the Rust engine.

A node gets the state (a dict) and returns an update (a dict of keys to change, or None), suspend or goto. After a node, the next nodes come from its edges (edge), a router (route), a join (join) or its own choice (goto). Nodes that are ready together run in the same step, concurrently, and their updates are merged into the state. State values must be JSON.

The builder methods return the workflow, so calls can be chained.

Parameters

reducers (Optional[Mapping[str, str]], default None): How writes to a state key combine, by key. "replace" (the default for every key): a write replaces the value, and two writes to the key in one step fail with update_conflict. "append": the value is a list and each write is a list appended to it. "upsert_by_id": a list of objects with a string id; a new id is appended, an existing one replaced in place, {"id": ..., "$remove": True} removes it. For both list reducers, {"$replace": [...]} replaces the whole list.

Example

from criogenai import END, Workflow

wf = Workflow(reducers={"notes": "append"})

@wf.node
def research(state):
    return {"notes": [f"facts about {state['topic']}"]}

@wf.node
def write(state):
    return {"text": "; ".join(state["notes"])}

wf.start("research").edge("research", "write").edge("write", END)
result = wf.compile().run_sync({"topic": "solar panels"})
result.state["text"]   # "facts about solar panels"

Workflow.node#

def node(
    func: Optional[Callable[..., Any]] = None,
    *,
    name: Optional[str] = None,
) -> Any

Decorator that adds a function as a node: @wf.node or @wf.node(name="...").

The function is returned unchanged.

Parameters

func (Optional[Callable[..., Any]], default None): The node function (given when used as @wf.node).
name (Optional[str], default None): The node's name. Default: the function's name.

Example

@wf.node
def summarize(state):
    return {"summary": state["text"][:100]}

Workflow.add_node#

def add_node(name: str, node: Union[Callable[..., Any], Agent]) -> 'Workflow'

Adds a node: a function or an Agent.

A function takes (state) or (state, ctx) (a NodeContext), sync or async, and returns an update dict, None, suspend or goto. An exception it raises fails the run (node_failed, with the exception as __cause__).

An Agent reads the conversation from the messages key (a list of messages, or a string taken as one user message), runs, and writes the whole transcript back to messages. If it waits for approval, the workflow suspends with {"kind": "agent_approval", "requests": [...]}.

Parameters

name (str): The node's name, unique in the workflow.
node (Union[Callable[..., Any], Agent]): The function or agent.

Raises

ValueError: A node with this name exists.
TypeError: node is not a function or an Agent.

Workflow.start#

def start(name: str) -> 'Workflow'

Sets the node every run starts with.

Parameters

name (str): The entry node's name.

Workflow.edge#

def edge(source: str, target: str) -> 'Workflow'

After source finishes, run target.

Several edges from one node run their targets in parallel.

Parameters

source (str): The node the edge leaves from.
target (str): The node to run next, or END to finish.

Workflow.route#

def route(
    source: str,
    router: Callable[[Dict[str, Any]], str],
    routes: Mapping[str, str],
) -> 'Workflow'

After source, a router function picks the next node from the state.

Parameters

source (str): The node the route leaves from.
router (Callable[[Dict[str, Any]], str]): A plain (not async) function state -> label that returns one of the labels in routes.
routes (Mapping[str, str]): Label to node name (or END).

Example

wf.route("classify", lambda s: s["kind"], {"refund": "refunds", "other": "general"})

Workflow.join#

def join(sources: Sequence[str], target: str) -> 'Workflow'

Runs target once, after every node in sources has finished.

Parameters

sources (Sequence[str]): The nodes to wait for (usually branches that run in parallel).
target (str): The node to run after all of them.

Example

wf.edge("begin", "a").edge("begin", "b").join(["a", "b"], "merge")

Workflow.goto#

def goto(source: str, targets: Sequence[str]) -> 'Workflow'

Declares that source chooses its next nodes itself, by returning goto.

The graph is checked with these targets, so a node can only jump where it said it might.

Parameters

source (str): The node that returns goto(...).
targets (Sequence[str]): Every node (or END) it may choose.

Workflow.compile#

def compile(storage: Union[None, str, Storage] = None) -> 'CompiledWorkflow'

Checks the graph in Rust and returns a workflow ready to run.

Every problem is reported at once (missing start node, edges to unknown nodes, unknown reducers, and so on).

Parameters

storage (Union[None, str, Storage], default None): Where CompiledWorkflow.thread keeps threads: a ~criogenai.Storage or a storage URL. Default: process memory. Not needed for run and resume.

Returns: The compiled workflow.

Raises

ValidationError: The graph is invalid (invalid_workflow); details["issues"] lists every problem.

CompiledWorkflow#

class CompiledWorkflow:
    def __init__(self, native: Any, storage: Optional[Storage])

A checked workflow, ready to run; made by Workflow.compile.

Use run and resume for one-off runs (a paused run is kept by you, as a snapshot), or thread for durable runs kept in storage.

Parameters

native (Any): The engine's workflow. Internal: use Workflow.compile.
storage (Optional[Storage]): The storage given to compile, or None.

CompiledWorkflow.run#

async def run(
    input: Optional[Mapping[str, Any]] = None,
    *,
    max_steps: Optional[int] = None,
    deadline: Optional[float] = None,
) -> WorkflowResult

Runs once from the start node, with input as the state.

Parameters

input (Optional[Mapping[str, Any]], default None): The starting state (a dict of JSON values). Default empty.
max_steps (Optional[int], default None): The most steps the run may take. Default 64; past it the run fails with step_limit_exceeded.
deadline (Optional[float], default None): The longest the run may take, in seconds. Default: no limit; past it the run fails with deadline_exceeded.

Returns: The final (or paused) state and how the run ended.

Raises

LimitExceededError: max_steps or deadline was reached.
CriogenError: A node raised (node_failed; __cause__ is the exception, snapshot the failed run).

CompiledWorkflow.run_sync#

def run_sync(
    input: Optional[Mapping[str, Any]] = None,
    **options: Any,
) -> WorkflowResult

run for scripts (not inside an event loop).

Parameters

input (Optional[Mapping[str, Any]], default None): The starting state.
**options (Any): max_steps and deadline, as for run.

CompiledWorkflow.resume#

async def resume(snapshot: Dict[str, Any], values: Mapping[str, Any]) -> WorkflowResult

Continues a paused run (without a thread) from its snapshot.

Parameters

snapshot (Dict[str, Any]): WorkflowResult.snapshot of the paused run.
values (Mapping[str, Any]): One answer per paused node, by node name; each node runs again with ctx.resume set to its value.

Returns: The result of the continued run.

Raises

ValidationError: The snapshot cannot be read or does not fit this workflow, or the values do not match the paused nodes (resume_invalid, workflow_mismatch).

CompiledWorkflow.thread#

def thread(thread_id: str, actor: Optional[str] = None) -> 'WorkflowThread'

A durable run history in the workflow's storage, created on first use.

Parameters

thread_id (str): The thread's id, e.g. one per request or case. Do not share ids between different workflows: another workflow's thread is refused (workflow_mismatch).
actor (Optional[str], default None): Who acts through this handle (a user id), written to the audit trail when the storage has audit=True.

Returns: A handle; nothing is read or written until you call it.

Example

app = wf.compile("sqlite://workflows.db")
thread = app.thread("request-1", actor="reviewer-7")
first = await thread.run({"text": "Hello"})
if first.suspended:
    done = await thread.resume({"approve": "yes"})

WorkflowThread#

class WorkflowThread:
    def __init__(self, app: CompiledWorkflow, thread_id: str, actor: Optional[str])

One durable run history of a workflow, kept in storage.

Each run starts from the state the last one ended with, plus the input. A run that suspends waits in storage for resume, from this process or another; a failed or interrupted run waits for retry. Only one call uses a thread at a time (see the storage's on_busy).

Parameters

app (CompiledWorkflow): The compiled workflow. Internal: use CompiledWorkflow.thread.
thread_id (str): The thread's id.
actor (Optional[str]): Who acts through this handle, for the audit trail.

WorkflowThread.run#

async def run(
    input: Optional[Mapping[str, Any]] = None,
    **options: Any,
) -> WorkflowResult

Runs from the start node, on the stored state plus input.

Parameters

input (Optional[Mapping[str, Any]], default None): State values to set before the run; a key here replaces the stored one.
**options (Any): max_steps (default 64) and deadline (seconds), as for CompiledWorkflow.run.

Returns: The result, with the checkpoint_id written and the thread's new version.

Raises

ThreadSuspendedError: Nodes wait for resume (thread_suspended).
CriogenError: The last run did not finish (thread_interrupted); call retry.
ThreadBusyError: Another call holds the thread.

WorkflowThread.resume#

async def resume(values: Mapping[str, Any]) -> WorkflowResult

Answers the paused nodes and continues the run.

Parameters

values (Mapping[str, Any]): One value per paused node, by node name; each node runs again with ctx.resume set to its value.

Raises

ValidationError: The values do not match the paused nodes (resume_invalid); nothing runs and the thread keeps waiting.

WorkflowThread.retry#

async def retry() -> WorkflowResult

Continues a failed or interrupted run from where it stopped.

Nodes that finished (and were saved) do not run again.

WorkflowThread.state#

async def state() -> Dict[str, Any]

The thread's current state (empty for a new thread).

WorkflowThread.interrupts#

async def interrupts() -> List[Interrupt]

What the paused nodes asked for (empty unless the thread is suspended).

WorkflowThread.history#

async def history(limit: int = 20) -> List[Dict[str, Any]]

The thread's checkpoints, newest first.

Parameters

limit (int, default 20): The most checkpoints to return. Default 20.

Returns: One dict per checkpoint: id, parent_id, version, source, status, run_id, step and created_at_ms (Unix milliseconds). Pass an id to rewind.

WorkflowThread.rewind#

async def rewind(checkpoint_id: str) -> Dict[str, Any]

Makes an older checkpoint current again.

The checkpoint is copied as the new head (the history is kept), so the next call continues from that point.

Parameters

checkpoint_id (str): An id from history.

Returns: The new head checkpoint, shaped like a history entry.

Raises

StorageError: The thread has no such checkpoint (checkpoint_not_found).

WorkflowThread.delete#

async def delete() -> bool

Deletes the thread and all its checkpoints.

Returns: True if the thread existed.

END#

END = '__end__'

The end of a workflow: use it as the target of an edge, a route or goto.

NodeContext#

@dataclass(frozen=True)
class NodeContext:
    run_id: str
    step: int
    node: str
    attempt: int
    resume: Any = None

What a node can know about its execution.

A node function that takes two parameters, (state, ctx), gets one.

Attributes

run_id (str): The run's id (the same for every node of one run).
step (int): The step this node runs in, counted from 0.
node (str): This node's name.
attempt (int): Which attempt this is, from 1.
resume (Any, default None): The value given to resume for this node after it suspended, else None.

Example

def review(state, ctx):
    if ctx.resume is None:
        return suspend({"question": "Approve this text?"})
    return {"approved": ctx.resume == "yes"}

suspend#

def suspend(payload: Any, update: Optional[Dict[str, Any]] = None) -> Any

Return this from a node to pause the run until a person answers.

The run ends with suspended=True and an Interrupt for this node. Continue it with CompiledWorkflow.resume(snapshot, {node: value}) or WorkflowThread.resume({node: value}): the node runs again, with ctx.resume == value.

Parameters

payload (Any): What the person is asked (any JSON value); it comes back as Interrupt.payload.
update (Optional[Dict[str, Any]], default None): State updates to apply now, before the pause.

Returns: A marker for the engine; return it from the node as is.

Example

from criogenai import END, Workflow, suspend

def approve(state, ctx):
    if ctx.resume is None:
        return suspend({"question": "Send this message?", "text": state["text"]})
    return {"approved": ctx.resume == "yes"}

wf = Workflow()
wf.add_node("approve", approve)
wf.start("approve").edge("approve", END)
app = wf.compile()
paused = await app.run({"text": "Hello"})
done = await app.resume(paused.snapshot, {"approve": "yes"})

goto#

def goto(
    targets: Union[str, Sequence[str]],
    update: Optional[Dict[str, Any]] = None,
) -> Any

Return this from a node to choose which nodes run next.

The node must be declared with Workflow.goto, and every target must be among the targets declared there.

Parameters

targets (Union[str, Sequence[str]]): A node name (or END), or several names to run next in parallel.
update (Optional[Dict[str, Any]], default None): State updates to apply with the choice.

Returns: A marker for the engine; return it from the node as is.

Example

from criogenai import END, Workflow, goto

def pick(state):
    return goto("big" if state["n"] > 10 else "small", {"checked": True})

wf = Workflow()
wf.add_node("pick", pick)
wf.add_node("big", lambda s: {"size": "big"})
wf.add_node("small", lambda s: {"size": "small"})
wf.goto("pick", ["big", "small"])
wf.start("pick").edge("big", END).edge("small", END)
wf.compile().run_sync({"n": 3}).state["size"]   # "small"

Interrupt#

@dataclass(frozen=True)
class Interrupt:
    node: str
    payload: Any

A paused node and what it asked for.

Attributes

node (str): The node that called suspend; use its name as the key when resuming.
payload (Any): What it asked: the payload given to suspend.

WorkflowResult#

@dataclass
class WorkflowResult:
    state: Dict[str, Any]
    steps: int
    suspended: bool
    interrupts: List[Interrupt]
    run_id: str
    snapshot: Optional[Dict[str, Any]] = field(repr=False, default=None)
    checkpoint_id: str = ''
    version: int = 0

What a workflow run produced.

Attributes

state (Dict[str, Any]): The state the run ended with (or paused at).
steps (int): How many steps ran.
suspended (bool): True if nodes wait for a person; see interrupts.
interrupts (List[Interrupt]): The waiting nodes and what they asked (empty unless suspended).
run_id (str): The run's id.
snapshot (Optional[Dict[str, Any]], default field(repr=False, default=None)): The paused run, to pass to CompiledWorkflow.resume (runs without a thread). None when the run finished.
checkpoint_id (str, default ''): Thread runs only: the checkpoint this call wrote ("" otherwise).
version (int, default 0): Thread runs only: the thread's version after this call (0 otherwise).