Python API: Patterns

Ready-made multi-agent designs: sequential, parallel, loops, routers, debates and more. Generated from the docstrings and type hints of criogenai 0.1.7. See also: Python API overview, the same module in TypeScript.

Ready-made designs for agentic systems: chains, fan-outs, loops, reviews, hand-offs and human review.

from criogenai import patterns

flow = patterns.sequential(["researcher", "writer"], model="openai:gpt-5-mini")
print((await flow.run("Write about solar panels")).output)

Steps can be an Agent, a ~criogenai.specialists.Specialist or the name it is registered under (a string), a plain function text -> text (sync or async), or a (name, step) tuple to name the step yourself. The model for specialists comes from model= or the CRIOGEN_MODEL environment variable. Every loop has a cap (at most 50 rounds), so nothing runs forever.

Workflow patterns return a Pattern: run / run_sync for the answer, compile(storage) for durable threads, approvals and an audit trail, workflow to extend the graph. react, coordinator and hierarchical return an Agent, so they nest as steps of other patterns.

from criogenai import patterns

patterns.sequential(...)

These are used as patterns.<name>.

Contents#

Name	Kind	Summary
`sequential`	function	Runs the steps in a fixed order; each step's output is the next step's input.
`parallel`	function	Runs the branches at the same time on the same input, then combines their answers.
`loop`	function	Repeats the steps until a condition holds or the round limit is reached.
`custom`	function	Wraps any `~criogenai.Workflow` as a `Pattern`, with the same `run` and `compile` as the ready-made ones.
`PatternResult`	class	What a pattern run produced.
`Pattern`	class	A workflow pattern: run it, compile it for durable threads, or extend it.
`iterative_refinement`	function	Improves a text round by round until a score is high enough.
`review_and_critique`	function	A generator writes, a critic reviews, and the generator revises until the critic approves.
`swarm`	function	Agents hand the work to each other, with no central coordinator.
`human_review`	function	Runs a step, then pauses so a person can approve, edit or reject its draft.
`quick`	function	A pattern from one line of registered specialist names.
`react`	function	A single agent in a reason-act loop (think, call a tool, read the result, repeat), with a hard cap on model calls.
`coordinator`	function	An agent that reads each request and hands it to the right specialist, which it calls as a tool.
`hierarchical`	function	A coordinator of coordinators: the root splits the work, each level hands it further down, and workers do it.

sequential#

def sequential(steps: Sequence[Step], *, model: Any = None) -> Pattern

Runs the steps in a fixed order; each step's output is the next step's input.

Parameters

steps (Sequence[Step]): The steps, in order (at least one).
model (Any, default None): The model for steps given as specialists (a model string or object). Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern; its output is the last step's text.

Raises

ValueError: No steps, or a specialist step and no model.

Example

from criogenai import patterns

def outline(text):
    return f"Outline: {text}"

flow = patterns.sequential([outline, "writer"], model="openai:gpt-5-mini")
result = await flow.run("Write about solar panels")

parallel#

def parallel(
    branches: Union[Sequence[Step], Mapping[str, Step]],
    synthesizer: Optional[Step] = None,
    *,
    model: Any = None,
) -> Pattern

Runs the branches at the same time on the same input, then combines their answers.

With a synthesizer step, it receives the original request and every branch's answer and writes the output. Without one, the output is the answers as a list with a ## <name> heading each, in branch order.

Parameters

branches (Union[Sequence[Step], Mapping[str, Step]]): The steps, as a list or as {name: step} (at least one).
synthesizer (Optional[Step], default None): A step that combines the answers. Default: none.
model (Any, default None): The model for steps given as specialists. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern.

Raises

ValueError: No branches, or a specialist step and no model.

Example

from criogenai import patterns

flow = patterns.parallel({
    "upper": str.upper,
    "length": lambda text: f"{len(text)} characters",
})
print(flow.run_sync("hello").output)   # "## upper", "HELLO", "## length", "5 characters"

loop#

def loop(
    steps: Sequence[Step],
    until: Callable[[str, int], bool],
    *,
    max_iterations: int = 5,
    model: Any = None,
) -> Pattern

Repeats the steps until a condition holds or the round limit is reached.

Each round runs the steps as a chain (each output feeds the next); the first step of the next round gets the last step's output.

Parameters

steps (Sequence[Step]): The steps of one round (at least one).
until (Callable[[str, int], bool]): until(text, round) is called after each round with the round's output and its number (from 1); True ends the loop (stop_reason="condition").
max_iterations (int, default 5): The most rounds (at most 50). Default 5; reaching it ends with stop_reason="max_iterations".
model (Any, default None): The model for steps given as specialists. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern.

Raises

ValueError: No steps, max_iterations below 1, or a specialist step and no model.

Example

from criogenai import patterns

flow = patterns.loop([lambda t: t + "!"], until=lambda text, n: text.count("!") >= 3)
result = flow.run_sync("Go")
result.output, result.stop_reason   # ("Go!!!", "condition")

custom#

def custom(workflow: Workflow, *, name: str = 'custom', max_steps: int = 64) -> Pattern

Wraps any ~criogenai.Workflow as a Pattern, with the same run and compile as the ready-made ones.

The workflow starts with the state {"input": text} and should write its answer to state["output"] (and a stop_reason if you like).

Parameters

workflow (Workflow): The workflow to run.
name (str, default 'custom'): The pattern's name. Default "custom".
max_steps (int, default 64): The step limit for Pattern.run. Default 64.

Returns: The pattern.

Example

from criogenai import END, Workflow, patterns

wf = Workflow()
wf.add_node("shout", lambda state: {"output": state["input"].upper()})
wf.start("shout").edge("shout", END)
patterns.custom(wf, name="shout").run_sync("hello").output   # "HELLO"

PatternResult#

@dataclass
class PatternResult:
    output: Optional[str]
    trace: List[Dict[str, Any]]
    state: Dict[str, Any]
    stop_reason: str
    suspended: bool = False
    waiting_for: Any = None
    raw: Optional[WorkflowResult] = field(default=None, repr=False)

What a pattern run produced.

print(result) shows the output; result.explain() shows each step.

Attributes

output (Optional[str]): The final answer (the last step's text), or None while suspended.
trace (List[Dict[str, Any]]): Each step in order: {"step": name, "output": text}.
state (Dict[str, Any]): The workflow's final state (input, output, trace and the pattern's own keys).
stop_reason (str): Why it ended: "done", "condition", "approved", "max_iterations", "max_rounds", "final", "max_handoffs", "unknown_handoff", "rejected" or "suspended".
suspended (bool, default False): True while a human_review run waits for a person; continue with Pattern.resume.
waiting_for (Any, default None): What the person is asked when suspended (for human_review: {"draft", "question"}), else None.
raw (Optional[WorkflowResult], default field(default=None, repr=False)): The underlying ~criogenai.WorkflowResult (snapshot, interrupts, run id).

Example

result = flow.run_sync("Write about solar panels")
print(result.output)
print(result.explain())

PatternResult.explain#

def explain() -> str

Each step on its own line: which step, and what it returned (cut to 140 characters).

Returns: The text, starting with ended: <stop_reason>.

Pattern#

class Pattern:
    def __init__(self, name: str, workflow: Workflow, budget: int)

A workflow pattern: run it, compile it for durable threads, or extend it.

Made by the pattern functions (sequential, parallel, loop, ...). Every run starts with the state {"input": text} and the answer is state["output"].

Parameters

name (str): The pattern's name, e.g. "sequential".
workflow (Workflow): The ~criogenai.Workflow it runs. Available as pattern.workflow: add nodes and edges to extend it before the first run.
budget (int): The step limit run uses unless given max_steps.

Example

from criogenai import patterns

flow = patterns.sequential([lambda t: f"Outline: {t}", lambda t: t + " (expanded)"])
result = flow.run_sync("solar panels")
result.output   # "Outline: solar panels (expanded)"

Pattern.compile#

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

The pattern as a compiled workflow, for durable threads, resume from another process and an audit trail.

Start a thread run with {"input": text}; the answer is state["output"]. Pass max_steps if the pattern needs more than the default 64 steps. A human_review pause is answered with thread.resume({"review": decision}).

Parameters

storage (Union[None, str, Storage], default None): A ~criogenai.Storage or a storage URL. Default: process memory.

Returns: A ~criogenai.CompiledWorkflow.

Example

app = flow.compile("sqlite://reviews.db")
paused = await app.thread("request-1").run({"input": "Can I change my booking?"})
done = await app.thread("request-1").resume({"review": "approve"})
done.state["output"]

Pattern.run#

async def run(text: str, **options: Any) -> PatternResult

Runs the pattern once on text and returns the result.

A human_review pattern returns suspended=True; continue it with resume.

Parameters

text (str): The input, given to the first step.
**options (Any): max_steps (default: the pattern's own budget, sized from its steps and caps) and deadline (seconds), as for CompiledWorkflow.run <criogenai.CompiledWorkflow.run>.

Returns: The output, each step's trace and why the run ended.

Pattern.run_sync#

def run_sync(text: str, **options: Any) -> PatternResult

run for scripts (not inside an event loop).

Parameters

text (str): The input.
**options (Any): As for run.

Pattern.resume#

async def resume(paused: PatternResult, decision: Any) -> PatternResult

Continues a human_review run that is waiting for a person.

Parameters

paused (PatternResult): The result of the run that paused (suspended=True).
decision (Any): "approve" (keep the draft), {"decision": "edit", "text": "..."} (use this text instead) or {"decision": "reject"} (empty output, stop_reason="rejected").

Returns: The finished result.

Raises

ValueError: paused is not a paused run of this pattern.

iterative_refinement#

def iterative_refinement(
    refine: Step,
    score: Callable[[str], float],
    *,
    threshold: float = 0.8,
    max_iterations: int = 5,
    model: Any = None,
) -> Pattern

Improves a text round by round until a score is high enough.

refine gets the current text and returns a better one; after each round, score(text) decides whether to stop. A loop with one step: stop_reason is "condition" or "max_iterations".

Parameters

refine (Step): The step that improves the text.
score (Callable[[str], float]): A function text -> float.
threshold (float, default 0.8): Stop when score(text) >= threshold. Default 0.8.
max_iterations (int, default 5): The most rounds (at most 50). Default 5.
model (Any, default None): The model if refine is a specialist. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern.

Example

from criogenai import patterns

flow = patterns.iterative_refinement(
    "writer",
    score=lambda text: min(len(text) / 500, 1.0),
    threshold=1.0,
    model="openai:gpt-5-mini",
)

review_and_critique#

def review_and_critique(
    generator: Step,
    critic: Step,
    *,
    max_rounds: int = 3,
    approved: Optional[Callable[[str], bool]] = None,
    model: Any = None,
) -> Pattern

A generator writes, a critic reviews, and the generator revises until the critic approves.

The critic approves when its reply starts with APPROVED (or when approved(reply) is true); otherwise its reply is passed back to the generator as feedback with the previous draft. The output is the last draft.

Parameters

generator (Step): The step that writes and revises.
critic (Step): The step that reviews (the "critic" specialist replies APPROVED or REVISE: ...).
max_rounds (int, default 3): The most reviews (at most 50). Default 3.
approved (Optional[Callable[[str], bool]], default None): A function reply -> bool that decides approval instead of the APPROVED prefix.
model (Any, default None): The model for steps given as specialists. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern; stop_reason is "approved" or "max_rounds".

Example

from criogenai import patterns

flow = patterns.review_and_critique("writer", "critic", max_rounds=3, model="openai:gpt-5-mini")
result = await flow.run("A two-line product description for a desk lamp")

swarm#

def swarm(
    agents: Mapping[str, Step],
    start: Union[str, Step, None] = None,
    *,
    max_handoffs: int = 6,
    model: Any = None,
) -> Pattern

Agents hand the work to each other, with no central coordinator.

Each agent is told who it can hand off to, and ends its reply with a line HANDOFF: <name> to pass the work on or FINAL when the answer is ready (a reply with neither counts as FINAL). Each agent sees the request and the work so far.

Parameters

agents (Mapping[str, Step]): {name: step}, at least two.
start (Union[str, Step, None], default None): The first agent's name, or a dispatcher step that replies with a name (an unknown name falls back to the first agent). Default: the first agent.
max_handoffs (int, default 6): The most hand-offs (at most 50). Default 6.
model (Any, default None): The model for steps given as specialists. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern; stop_reason is "final", "max_handoffs" or "unknown_handoff" (a hand-off to a name not in agents).

Raises

ValueError: Fewer than two agents, or max_handoffs below 1.

Example

from criogenai import patterns

flow = patterns.swarm({
    "triage": lambda text: "Looks like billing.\nHANDOFF: billing",
    "billing": lambda text: "Your invoice was sent again.\nFINAL",
})
result = flow.run_sync("I did not get my invoice")
result.stop_reason   # "final"

human_review#

def human_review(step: Step, *, model: Any = None) -> Pattern

Runs a step, then pauses so a person can approve, edit or reject its draft.

The run returns suspended=True with waiting_for set to {"draft", "question"}. Continue it with Pattern.resume. With compile(storage) the pause is durable: another process can answer it days later with thread.resume({"review": decision}).

Parameters

step (Step): The step that writes the draft.
model (Any, default None): The model if step is a specialist. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern; stop_reason is "approved" (also after an edit) or "rejected".

Example

from criogenai import patterns

flow = patterns.human_review(lambda text: f"Draft reply to: {text}")
paused = await flow.run("Can I change my booking?")
paused.waiting_for["draft"]
done = await flow.resume(paused, {"decision": "edit", "text": "Yes, from your account page."})

quick#

def quick(spec: str, *, model: Any = None) -> Pattern

A pattern from one line of registered specialist names.

"a > b" runs them one after the other (sequential), "a | b" at the same time (parallel), "writer <-> critic" writes, reviews and revises until approved (review_and_critique), and a single name runs one specialist. Use one kind of connector per line; combine patterns with the functions above.

Parameters

spec (str): The line, e.g. "researcher > writer".
model (Any, default None): The specialists' model. Default: the CRIOGEN_MODEL environment variable.

Returns: A Pattern.

Raises

ValueError: An empty line or name, mixed connectors, or <-> between more than two names.
KeyError: A name is not registered.

Example

from criogenai import patterns

flow = patterns.quick("researcher > writer", model="openai:gpt-5-mini")
result = await flow.run("Write about solar panels")

react#

def react(
    model: Any,
    tools: Sequence[Any],
    *,
    max_steps: int = 8,
    instructions: Optional[str] = None,
    **agent_options: Any,
) -> Agent

A single agent in a reason-act loop (think, call a tool, read the result, repeat), with a hard cap on model calls.

The agent loop already works this way; this adds instructions for it and the limit.

Parameters

model (Any): The agent's model.
tools (Sequence[Any]): Its tools.
max_steps (int, default 8): The most model calls (at most 50). Default 8.
instructions (Optional[str], default None): Your instructions, put before the reason-act ones.
**agent_options (Any): Other Agent keyword arguments (storage, approve, limits, ...).

Returns: An Agent.

Example

from criogenai import patterns, tool

@tool
def lookup_order(order_id: str) -> str:
    "Looks up an order's status."
    return "shipped"

agent = patterns.react("openai:gpt-5-mini", [lookup_order], max_steps=5)
result = await agent.run("Where is order A1?")

coordinator#

def coordinator(
    model: Any,
    specialists: Mapping[str, Any],
    *,
    instructions: Optional[str] = None,
    **agent_options: Any,
) -> Agent

An agent that reads each request and hands it to the right specialist, which it calls as a tool.

A model decides the routing: more flexible than a fixed workflow, and it costs more model calls.

Parameters

model (Any): The coordinator's model; also used to build specialists given as recipes or names.
specialists (Mapping[str, Any]): {name: specialist}, where each value is an Agent, a Specialist, a registered name, or (agent, "when to use it"). The description is what the coordinator reads to choose (a Specialist brings its own).
instructions (Optional[str], default None): The coordinator's instructions. Default: a short routing prompt.
**agent_options (Any): Other Agent keyword arguments.

Returns: An Agent.

Raises

ValueError: No specialists.
TypeError: A value is not an agent, a specialist or a name.

Example

from criogenai import Agent, patterns

orders = Agent("openai:gpt-5-mini", instructions="You answer order questions.")
triage = patterns.coordinator(
    "openai:gpt-5-mini",
    {"orders": (orders, "Order status questions"), "summaries": "summarizer"},
)

hierarchical#

def hierarchical(
    model: Any,
    tree: Mapping[str, Any],
    *,
    instructions: Optional[str] = None,
    **agent_options: Any,
) -> Agent

A coordinator of coordinators: the root splits the work, each level hands it further down, and workers do it.

Parameters

model (Any): The model of every coordinator (and of workers given as recipes or names).
tree (Mapping[str, Any]): {name: worker or subtree}. A worker is anything coordinator accepts; a nested mapping becomes a sub-coordinator, described by its "_description" key. Depth is bounded by the agents' nesting limit.
instructions (Optional[str], default None): The root coordinator's instructions.
**agent_options (Any): Other Agent keyword arguments for the root.

Returns: The root Agent.

Example

from criogenai import patterns

lead = patterns.hierarchical("openai:gpt-5-mini", {
    "research": {"_description": "Gathers facts", "web": "researcher", "check": "fact_checker"},
    "writing": "writer",
})