Skip to content

Lesson 13 — workflows

What you will build

The sequence, written down: steps that each do one thing, edges that say what may follow, branches that choose, and deadlines that bound the whole thing. A model call is one step among several, which is what stops a slow provider from becoming an unbounded request.

xyberos.workflow is preview tier (stability guide), and its guarantees are specific: a validated graph, bounded retries, per-step and whole-run deadlines, structured events, and no orphan tasks. It also promises no parallel steps and no loops — execution is sequential and conditional, and cycles are refused at validation.

What you need

Nothing new.

examples/learn_ai/ai13_workflows.py
"""Lesson 13 — workflows: explicit steps, edges, branches, and deadlines.

Run it with:

    python -m learn_ai.ai13_workflows
"""

from __future__ import annotations

import asyncio
from typing import Any

from xyberos import Retry, StepContext, Workflow, seconds
from xyberos.errors import WorkflowValidationError


def build() -> Workflow:
    """Define the graph once, so it can be rendered, tested, and reviewed without running it."""

    async def classify(value: dict[str, str], context: StepContext) -> dict[str, str]:
        text = value["request"].lower()
        intent = "other" if "refund" not in text else "refund"
        return {"request": value["request"], "intent": intent}

    async def search(value: dict[str, str], context: StepContext) -> dict[str, str]:
        return {**value, "passage": "a refund takes three business days"}

    async def answer(value: dict[str, str], context: StepContext) -> str:
        return f"About your {value['intent']}: {value['passage']}"

    async def escalate(value: dict[str, str], context: StepContext) -> str:
        return "This needs a person."

    return (
        Workflow("triage", version="1", input_type=dict)
        .step("classify", classify, retry=Retry(max_attempts=2), timeout=seconds(5))
        .step("search", search, timeout=seconds(5))
        .step("answer", answer)
        .step("escalate", escalate)
        .edge("classify", "search", when=lambda value: value["intent"] != "other")
        .edge("classify", "escalate", when=lambda value: value["intent"] == "other")
        .edge("search", "answer")
    )


def cyclic() -> Workflow:
    """A graph with a cycle, which validation refuses before anything runs."""

    async def step(value: str, context: StepContext) -> str:
        return value

    return (
        Workflow("loop", version="1", input_type=str)
        .step("a", step)
        .step("b", step)
        .edge("a", "b")
        .edge("b", "a")
    )


async def main() -> None:
    """Render the graph, run both branches, and watch validation refuse a loop."""
    workflow = build()
    print(workflow.render().splitlines()[0])
    print("steps:", workflow.step_ids)
    print("entries:", [step.id for step in workflow.entries])
    print("exits:  ", [step.id for step in workflow.exits])

    for request in ("how long does a refund take?", "the printer is on fire"):
        result: Any = await workflow.run(input={"request": request}, timeout=seconds(10))
        print(f"{request!r} -> {result.value}")
        statuses = {step_id: outcome.status.value for step_id, outcome in result.steps.items()}
        print("   steps:", statuses)

    try:
        cyclic().validate()
    except WorkflowValidationError as error:
        print("a cycle is refused:", error)


if __name__ == "__main__":
    asyncio.run(main())

The graph is a value. build() returns a Workflow that can be rendered, validated, and inspected without executing anything, which is why the same definition is safe in a log line, a test, and a page like this one.

How to run it

$env:PYTHONPATH = "src;examples"
python -m learn_ai.ai13_workflows

PYTHONPATH=src:examples python -m learn_ai.ai13_workflows

What it prints

workflow 'triage' version='1'
steps: ('classify', 'search', 'answer', 'escalate')
entries: ['classify']
exits:   ['answer', 'escalate']
'how long does a refund take?' -> About your refund: a refund takes three business days
   steps: {'classify': 'succeeded', 'search': 'succeeded', 'escalate': 'skipped', 'answer': 'succeeded'}
'the printer is on fire' -> This needs a person.
   steps: {'classify': 'succeeded', 'search': 'skipped', 'escalate': 'succeeded', 'answer': 'skipped'}
a cycle is refused: workflow 'loop' contains a cycle: a -> b -> a; sequential and conditional execution does not support loops

Read the two steps: lines: the branch that was not taken is skipped, not missing. That is what makes a workflow inspectable — a run records what happened, what was skipped, how many attempts each step took, and how long it ran, so an incident review does not have to guess.

The last line is validation doing its job: a cycle is refused when the definition is validated, before any step runs, with the path named (a -> b -> a).

Common mistakes

  • Putting a whole handler in one step. A step is the unit of retry and observation; a step that does five things can only be retried as five.
  • Retrying a step that has side effects. A step that sends an email and fails after sending will send again. Keep the effect in its own step, and make it idempotent or approved (the hybrid path's subject).
  • Expecting parallel execution. Edge-independent steps still run in declaration order; the engine is sequential by contract.
  • Long deadlines "just in case". A deadline is a promise to the caller; a five-minute budget on a request that used to take one second is an outage waiting for a slow provider.
  • Reading state from outside the step. A step receives its input and returns the next value; a workflow that closes over mutable module state cannot be replayed, inspected, or tested.
  • Building the graph inline in a request. Define it once, validate it at startup, and keep the definition next to the code it sequences.

Next

Lesson 14 composes tools with a workflow: a bounded agent, a budget, and the escalation path when the answer is not the model's to give.