Workflows | CriogenAI

A workflow is a graph of steps (nodes) over a shared state. Each node is a function (or an agent) that reads the state and returns the keys it changes; edges say what runs next. Use a workflow when you know the steps in advance: intake, then checks in parallel, then a summary, then a person signs off. Use an agent when the model should choose the steps. The two mix: agents can be nodes, and a pattern of agents is a workflow.

wf = Workflow()

@wf.node
def classify(state: dict) -> dict:
    return {"category": "billing" if "charge" in state["text"] else "other"}

@wf.node
def approve(state: dict, ctx: NodeContext) -> object:
    # Pause for a person; continue when someone answers.
    if ctx.resume is None:
        return suspend({"refund": state["amount"]})
    return {"approved": ctx.resume == "yes"}

@wf.node
def answer(state: dict) -> dict:
    return {"reply": "A human will follow up."}

wf.start("classify")
wf.route("classify", lambda s: s["category"], {"billing": "approve", "other": "answer"})
wf.edge("approve", END).edge("answer", END)
app = wf.compile(storage=Storage("sqlite://app.db"))  # every graph problem, reported at once

# Durable: the pause waits in the database, for days if needed.
turn = await app.thread("ticket-7").run({"text": "I was charged twice", "amount": 40})
if turn.suspended:
    # Later, from any process:
    await app.thread("ticket-7", actor="user:boss").resume({"approve": "yes"})

const app = new Workflow()
  .node("classify", (state) => ({
    category: String(state.text).includes("charge") ? "billing" : "other",
  }))
  .node("approve", (state, ctx) =>
    // Pause for a person; continue when someone answers.
    ctx.resume === undefined ? suspend({ refund: state.amount }) : { approved: ctx.resume === "yes" },
  )
  .node("answer", () => ({ reply: "A human will follow up." }))
  .start("classify")
  .route("classify", (state) => String(state.category), { billing: "approve", other: "answer" })
  .edge("approve", END)
  .edge("answer", END)
  .compile({ storage: new Storage({ url: "sqlite://app.db" }) }); // every graph problem, reported at once

// Durable: the pause waits in the database, for days if needed.
const turn = await app.thread("ticket-7").run({ text: "I was charged twice", amount: 40 });
if (turn.suspended) {
  // Later, from any process:
  await app.thread("ticket-7", { actor: "user:boss" }).resume({ approve: "yes" });
}

use criogenai_core::{
    END, NodeOutput, State, Update, WorkflowBuilder, node_fn, node_with_context, router_fn,
};
use criogenai_storage::Storage;
use serde_json::json;

let mut builder = WorkflowBuilder::new();
builder
    .add_node(
        "classify",
        node_fn(|state| async move {
            let text = state.get("text").and_then(|v| v.as_str()).unwrap_or("");
            let category = if text.contains("charge") {
                "billing"
            } else {
                "other"
            };
            Ok(Update::new().set("category", category))
        }),
    )
    .add_node(
        "approve",
        node_with_context(|ctx, state| async move {
            // Pause for a person; continue when someone answers.
            Ok(match ctx.resume() {
                None => {
                    NodeOutput::suspend(Update::new(), json!({"refund": state.get("amount")}))
                }
                Some(answer) => Update::new().set("approved", answer == "yes").into(),
            })
        }),
    )
    .add_node(
        "answer",
        node_fn(|_| async { Ok(Update::new().set("reply", "A human will follow up.")) }),
    )
    .set_entry("classify")
    .add_router(
        "classify",
        router_fn(|state| async move {
            Ok(state
                .get("category")
                .and_then(|v| v.as_str())
                .unwrap_or("other")
                .to_string())
        }),
        [("billing", "approve"), ("other", "answer")],
    )
    .add_edge("approve", END)
    .add_edge("answer", END);
let workflow = builder.build()?; // every graph problem, reported at once

// Durable: the pause waits in the database, for days if needed.
let storage = Storage::connect("sqlite://app.db").await?;
let input = State::new()
    .with("text", "I was charged twice")
    .with("amount", 40);
let turn = storage
    .workflow_thread(&workflow, "ticket-7")
    .run(input)
    .await?;
if turn.is_suspended() {
    // Later, from any process:
    storage
        .workflow_thread(&workflow, "ticket-7")
        .actor("user:boss")
        .resume([("approve", json!("yes"))])
        .await?;
}

How a run proceeds#

The engine runs a workflow in steps:

The start node runs with the input as the state.
After a step, every node whose edges are satisfied is ready. All ready nodes run at the same time in the next step.
When all nodes of a step have finished, their updates are merged into the state, and the step is saved (with storage). If any node of the step failed, none of that step's updates are kept.
The run ends when no node is ready (every path reached END).

compile() checks the graph before any run: unknown nodes, nodes that can never run, nodes with no way out. All problems are reported at once, as a ValidationError (invalid_workflow) whose details["issues"] lists them:

edge `a` -> `c` refers to unknown node `c`
node `b` has no outgoing edges; add an edge to END if it should finish the run
node `b` is not reachable from the entry node

Nodes#

A node receives the state (a dict / object) and, if it takes a second argument, a NodeContext. It returns one of:

Return	Meaning
a dict / object	The keys to change (the rest of the state is kept)
`None` / `undefined`	No change
`suspend(payload)`	Pause the run for a person (below)
`goto(targets, update)`	Choose the next nodes (below)

Nodes may be sync or async. The context has run_id, step, node, attempt, and resume (the answer after a pause). In Python, @wf.node adds a function under its own name; wf.add_node(name, fn) names it yourself. In TypeScript, .node(name, fn).

An exception in a node fails the run with node_failed; the exception is the error's cause. A router that throws fails it with router_failed.

Edges, routes and joins#

Method	Meaning
`start(node)`	The node a run begins with
`edge(a, b)`	After `a`, run `b` (`b` may be `END`)
`route(a, router, {label: node})`	After `a`, `router(state)` returns a label; run the node for it. A label with no entry fails with `unknown_route`
`join([a, b], c)`	Run `c` once all of `a` and `b` have finished
`goto(a, [targets])`	`a` chooses among these targets itself, by returning `goto(...)`

Several edges from one node start the targets in parallel. A router is a plain function of the state (not async in Python); keep it quick and without side effects.

Parallel steps and reducers#

from criogenai import END, Workflow

# Two checks run at the same time; "summary" waits for both.
wf = Workflow(reducers={"findings": "append"})  # both write "findings": append, not replace

@wf.node
def intake(state: dict) -> dict:
    return {"findings": []}

@wf.node
async def check_allergies(state: dict) -> dict:
    return {"findings": ["no known allergies"]}

@wf.node
async def check_interactions(state: dict) -> dict:
    return {"findings": ["no interactions"]}

@wf.node
def summary(state: dict) -> dict:
    return {"report": "; ".join(state["findings"])}

wf.start("intake")
wf.edge("intake", "check_allergies").edge("intake", "check_interactions")
wf.join(["check_allergies", "check_interactions"], "summary")
wf.edge("summary", END)

result = await wf.compile().run({"patient_ref": "P-1"})
print(result.state["report"], result.steps)  # 3 steps: intake, both checks, summary

// Two checks run at the same time; "summary" waits for both.
const checks = new Workflow({ reducers: { findings: "append" } }) // both write "findings": append, not replace
  .node("intake", () => ({ findings: [] }))
  .node("check_allergies", async () => ({ findings: ["no known allergies"] }))
  .node("check_interactions", async () => ({ findings: ["no interactions"] }))
  .node("summary", (state) => ({ report: (state.findings as string[]).join("; ") }))
  .start("intake")
  .edge("intake", "check_allergies")
  .edge("intake", "check_interactions")
  .join(["check_allergies", "check_interactions"], "summary")
  .edge("summary", END)
  .compile();

const checked = await checks.run({ patientRef: "P-1" });
console.log(checked.state.report, checked.steps); // 3 steps: intake, both checks, summary

When two nodes of the same step write the same key, the key's reducer decides how the writes combine:

Reducer	Python	TypeScript	Result
replace (default)	`"replace"`	`"replace"`	The new value; two writes in one step fail with `update_conflict`
append	`"append"`	`"append"`	Lists are concatenated
update by id	`"upsert_by_id"`	`"upsertById"`	Items with an `id` replace the item with the same id; others are added

A conflict is an error on purpose: without a reducer, which of two parallel writes wins would depend on timing.

Loops and choices#

A node can pick what runs next by returning goto(...), among targets declared with wf.goto(...). That makes loops (draft, review, draft again) explicit:

from criogenai import END, Workflow, goto

wf = Workflow()

@wf.node
def draft(state: dict) -> dict:
    return {"text": f"draft {state.get('round', 0) + 1}", "round": state.get("round", 0) + 1}

@wf.node
def review(state: dict) -> object:
    # A node can choose what runs next, among targets declared below.
    if state["round"] < 3:
        return goto("draft", {"feedback": "shorter"})
    return goto(END)

wf.start("draft").edge("draft", "review")
wf.goto("review", ["draft", END])

# max_steps bounds any loop; deadline (seconds) bounds the whole run.
result = await wf.compile().run({}, max_steps=20, deadline=30)
print(result.state["text"])  # "draft 3"

import { goto } from "criogenai";

const drafting = new Workflow()
  .node("draft", (state) => {
    const round = Number(state.round ?? 0) + 1;
    return { text: `draft ${round}`, round };
  })
  // A node can choose what runs next, among targets declared with .goto().
  .node("review", (state) => (Number(state.round) < 3 ? goto("draft", { feedback: "shorter" }) : goto(END)))
  .start("draft")
  .edge("draft", "review")
  .goto("review", ["draft", END])
  .compile();

// maxSteps bounds any loop; deadlineMs bounds the whole run.
const drafted = await drafting.run({}, { maxSteps: 20, deadlineMs: 30_000 });
console.log(drafted.state.text); // "draft 3"

Every loop needs a way out, and every run has bounds:

Option (Python / TypeScript)	Fails with
`max_steps` / `maxSteps`	`step_limit_exceeded`
`deadline` (seconds) / `deadlineMs`	`deadline_exceeded`

TypeScript runs also take a signal (an AbortSignal).

Agents as nodes#

An agent can be one node. It reads the conversation from the messages key (a list of messages, or a string taken as one user message), runs its loop, and writes the whole transcript back to messages:

from criogenai import END, Agent, Workflow

triage = Agent("openai:gpt-5-mini", instructions="Classify the request in one word.")
writer = Agent("openai:gpt-5-mini", instructions="Write a short, kind reply.")

wf = Workflow()
wf.add_node("triage", triage)  # reads "messages", writes the whole transcript back
wf.add_node("writer", writer)
wf.start("triage").edge("triage", "writer").edge("writer", END)

first = {"role": "user", "content": [{"type": "text", "text": "My parcel is late"}]}
result = await wf.compile().run({"messages": [first]})
print(result.state["messages"][-1]["content"][0]["text"])

import { Agent } from "criogenai";

const triage = new Agent({ model: "openai:gpt-5-mini", instructions: "Classify the request in one word." });
const writer = new Agent({ model: "openai:gpt-5-mini", instructions: "Write a short, kind reply." });

const desk = new Workflow()
  .addNode("triage", triage) // reads "messages", writes the whole transcript back
  .addNode("writer", writer)
  .start("triage")
  .edge("triage", "writer")
  .edge("writer", END)
  .compile();

const first = { role: "user", content: [{ type: "text", text: "My parcel is late" }] };
const answered = await desk.run({ messages: [first] });
const last = (answered.state.messages as { content: { text?: string }[] }[]).at(-1);
console.log(last?.content[0].text);

Keep the default reducer for messages: the agent returns the full list, so append or upsert_by_id would add the earlier messages twice. If an agent node needs approval, the workflow pauses; see Human approval.

Pause for a person#

A node returns suspend(payload) to pause the run: payload says what is asked (any JSON). The run stops with the pause recorded in result.interrupts (node, payload). Continuing it gives each paused node its answer: the node runs again, now with ctx.resume set to the answer. So write paused nodes as "if there is no answer yet, ask; otherwise use it", as in the first example. Other nodes that had finished are not run again.

Without storage, keep result.snapshot and call app.resume(snapshot, {node: answer}). With storage, use a thread (below).

Durable workflows#

A workflow compiled with storage gives threads the same guarantees as agent threads: progress saved every step, a pause that waits in the database and is answered from any process, retry() after a failure without re-running finished nodes, history, rewind and the audit trail.

Python	TypeScript	What it does
`app.thread(id, actor=None)`	`app.thread(id, { actor })`	A handle on a durable run history
`run(input)`	`run(input, options)`	Runs from the start node on the stored state plus `input`
`resume({node: value})`	`resume({ node: value })`	Answers the paused nodes
`interrupts()`	`interrupts()`	What the paused nodes asked for
`state()`	`state()`	The stored state
`retry()`	`retry()`	Continues a failed or interrupted run
`history()`, `rewind(id)`, `delete()`	same	As for agent threads

Each run on a thread starts from the start node, with the state the last run left plus the new input. A run while the thread is paused fails with thread_suspended; resume must answer every paused node, or it fails with resume_invalid. In Rust, the same is storage.workflow_thread(&workflow, id).

Errors#

Code	When
`invalid_workflow`	The graph does not validate (`details["issues"]`)
`node_failed`	A node raised (the exception is the cause)
`router_failed`, `unknown_route`	A router raised, or returned a label with no route
`missing_goto`	A node declared with `goto` did not return one
`update_conflict`	Two nodes wrote one key in one step, without a reducer
`step_limit_exceeded`, `deadline_exceeded`	A run bound was reached
`resume_invalid`	`resume` with a missing or unknown node
`workflow_mismatch`	A snapshot made by a different workflow

Next#

Patterns and specialists: ready-made workflows of agents.
How a run works: steps, merging and snapshots.
API: Python workflows, TypeScript workflows.