Dynamic Workflows¶

Use these capability snippets inside your application’s native runtime. Installation is through PyFly Agentic extras.

pyfly.agentic.workflows is a code-defined orchestration DSL for fanning out isolated sub-agents and reducing their results in plain Python. The plan lives in ordinary Python control flow, making orchestration inspectable. A journal can replay completed agent calls when the call sequence remains stable. Model outputs and external side effects retain their own execution semantics.

It is a peer of, not a replacement for, two existing layers:

Reasoning patterns drive a single agent's reason → act → observe loop. A workflow provides code-defined orchestration across many agents.
Pipelines are declarative DAGs with checkpointing and HITL. A workflow is an imperative code-first façade for fan-out/gather/verify patterns that read more naturally as a function than as a graph.

Quick start¶

Configure PYFLY_AGENTIC_DEFAULT_MODEL and provider credentials using the model guide. This example summarizes supplied source passages, reviews claims against those passages, and synthesizes a report. External retrieval is a separate tool integration; the example only uses the text supplied by the caller.

from pydantic import BaseModel, Field
from pyfly.agentic import get_config
from pyfly.agentic.workflows import (
    workflow, agent, parallel, pipeline, phase, run_workflow, WorkflowBudget,
)

class ResearchArgs(BaseModel):
    passages: list[str]
    model: str = Field(default_factory=lambda: get_config().default_model)

@workflow(name="deep_research", args_schema=ResearchArgs)
async def deep_research(args, ctx):
    with phase("extract"):
        hits = await parallel([
            (lambda passage=passage: agent(f"Extract the key claims from this source: {passage}", model=args.model))
            for passage in args.passages
        ])

    candidates = [
        {"source": source, "claims": claims, "model": args.model}
        for source, claims in zip(args.passages, hits, strict=True) if claims is not None
    ]
    with phase("verify"):
        verified = await pipeline(candidates, verify_stage, score_stage)

    with phase("synthesize"):
        notes = "\n".join(str(note) for note in verified if note is not None)
        if not notes:
            raise ValueError("No source passage completed review")
        return await agent(f"Write a concise report using only these reviewed notes:\n{notes}", model=args.model)

async def verify_stage(prev, item, index):
    return await agent(
        f"Keep only claims supported by source {index + 1}. Include the source number.\n"
        f"Source: {item['source']}\nClaims: {prev['claims']}",
        model=item["model"],
    )

async def score_stage(prev, item, index):
    return await agent(
        f"Rank these source-grounded claims by relevance to reliability engineering. "
        f"Preserve their source numbers and return the ranked notes:\n{prev}",
        model=item["model"],
    )

args = ResearchArgs(passages=[
    "Bounded retries stop after three attempts. Backoff increases the delay between attempts.",
    "A circuit breaker stops requests after repeated failures and probes for recovery later.",
])
report = await run_workflow(
    "deep_research", args,
    budget=WorkflowBudget(max_concurrent_agents=8, max_tokens=50_000),
)

A @workflow-decorated function can be invoked directly (await deep_research(args)) or by name through run_workflow("deep_research", args).

Primitives¶

Primitive	Signature	Semantics
`agent`	`await agent(prompt, *, label, model, output_type, instructions, deps, tools, toolsets)`	Run one isolated sub-agent; returns its `output` (a `str` or a validated `output_type`). A sub-agent can use `tools=`/`toolsets=` (e.g. `ToolKit.as_toolset()` or an MCP server) just like a top-level agent. Honours the budget, the concurrency gate, and the resume journal.
`parallel`	`await parallel(thunks)`	Barrier. Run zero-arg async thunks concurrently; a thunk that raises resolves to `None` (the call never propagates). Returns a list aligned to `thunks`.
`pipeline`	`await pipeline(items, *stages)`	No inter-stage barrier. Each item flows through every stage independently (item A can be in stage 3 while B is in stage 1). Each stage receives `(prev, item, index)` — declare only the params you need. A stage that raises drops that item to `None`.
`stream`	`async with stream(prompt, ...) as s:`	Stream one sub-agent token-by-token: iterate `s.text()` for deltas; `s.output` holds the full output after the block. Same budget/journal/cost accounting as `agent`. Requires a streaming runner. See Streaming.
`phase`	`with phase("title"):`	Group enclosed work for telemetry (`phase.start` / `phase.end` events).
`log`	`log("message")`	Emit a narrator line to the run's event handler.

Reduce in Python, not with a model. The strength of the DSL is that the glue between agent calls — dedup, rank, filter, branch — is ordinary deterministic Python. Reach for an agent() only when you genuinely need a model.

Type safety¶

The DSL is statically typed end-to-end. agent(output_type=T) is typed to return T (not Any), and @workflow produces a Workflow[OutputT] inferred from the function's return annotation — so the awaited result is typed too:

from pydantic import BaseModel
from pyfly.agentic.workflows import workflow, agent

class Report(BaseModel):
    summary: str
    sources: list[str]

@workflow(name="research")
async def research(args, ctx) -> Report:
    return await agent("write the report", output_type=Report)   # typed Report

report: Report = await research(args)   # research is Workflow[Report]; result is Report

Without output_type, agent() returns Any (the raw string output). Pyright infers all of the above with no casts.

Budgets¶

Every run is bounded by a WorkflowBudget:

WorkflowBudget(
    max_concurrent_agents=8,   # asyncio.Semaphore; default min(16, cpu - 2)
    max_agents_total=1000,     # hard kill-switch (runaway-loop backstop)
    max_tokens=500_000,        # optional output-token ceiling for the whole run
    max_cost_usd=2.50,         # optional USD ceiling (priced via genai-prices)
    max_wall_seconds=120,      # optional wall-clock ceiling for the whole run
)

Exceeding any ceiling raises WorkflowBudgetError — including inside a parallel/pipeline fan-out (a kill-switch aborts the run; only ordinary branch failures resolve to None). Token and dollar cost are fed by each AgentRunner call's reported usage, priced through the same genai-prices cost stack the rest of the framework uses; read live spend with ctx.tokens_spent / ctx.cost_spent_usd and the remaining headroom with ctx.remaining_tokens() / ctx.remaining_cost_usd().

Resume with a call journal¶

Every agent() call is keyed by its sequence number and recorded in a Journal. Re-running the workflow with a populated journal serves completed calls from cache and runs only new calls live:

from pyfly.agentic.workflows import Journal

journal = Journal()
await run_workflow("deep_research", args, journal=journal)   # full run, populates journal
# Run again with the same arguments and call sequence:
await run_workflow("deep_research", args, journal=journal)   # cached calls replay instantly

Persist journal.to_dict() (and rebuild with Journal.from_dict(...)) to resume manually. To persist completed calls, attach a JournalBackend; the journal attempts a flush after every completed call. Backend errors are logged and suppressed, so durability depends on successful storage:

from pyfly.agentic.workflows import Journal, FileJournalBackend

backend = FileJournalBackend("/var/lib/firefly/workflows")
journal = Journal(backend=backend, run_key="research-job-42")   # loads if present
await run_workflow("deep_research", args, journal=journal)      # auto-flushes each call
# ... process crashes; in a fresh process, the same run_key reloads progress:
await run_workflow("deep_research", args,
                   journal=Journal(backend=backend, run_key="research-job-42"))

FileJournalBackend stores each run as JSON (strings/numbers/lists/dicts round-trip exactly; structured outputs round-trip as their serialized form). Implement the JournalBackend protocol (load/save) for Postgres/Redis/S3.

Replay contract. Journal keys are call sequence numbers, not hashes of the prompt, model, or arguments. Resume with the same inputs and call sequence; code changes or timing-dependent branches can assign a cached output to a different call. Sequence numbers are assigned before the first await inside agent(), but calls launched after earlier asynchronous work can still depend on completion order. Keep replay-sensitive control flow stable and use a new journal when the inputs or sequence change. External side effects outside journaled calls may run again, and a crash before a successful journal write may repeat a model call.

The agent runner¶

agent() delegates LLM execution to a pluggable AgentRunner:

class AgentRunner(Protocol):
    async def run(self, prompt, *, model=None, output_type=None, instructions=None,
                  deps=None, tools=None, toolsets=None, using=None) -> AgentCall: ...

ManagedAgentRunner (default) runs each call through a Agent, so sub-agents inherit the full framework stack — see below.
DefaultAgentRunner (lightweight alternative) runs each call as a fresh bare pydantic_ai.Agent with no shared message history — isolation by construction and zero coupling. Choose it explicitly (runner=DefaultAgentRunner()) when you want the minimal path with none of the Agent middleware.
Custom runners make workflows trivially testable: inject a deterministic fake that returns canned AgentCall(output=..., tokens=...) and no network is touched.

await deep_research(args, runner=MyFakeRunner())

`ManagedAgentRunner` — sub-agents with the full Agent stack¶

This is the default runner: every workflow sub-agent runs through a Agent. Ephemeral agents get default logging and conditional observability; configured agents retain their middleware, tools, and model. Guards, caching, explainability, and validation remain opt-in. Rate-limit retry and usage tracking follow the same configuration as top-level agents. Construct the runner explicitly to select a default model or a specific agent source:

from pyfly.agentic import get_config
from pyfly.agentic.workflows import workflow, agent, ManagedAgentRunner

# The default — no runner needed; sub-agents are Agent instances already:
await my_workflow(args)

# Give every ephemeral sub-agent a default model (built auto_register=False, memory=None):
await my_workflow(args, runner=ManagedAgentRunner(default_model=get_config().default_model))

# Or reuse one configured agent for every call (keeps its model/tools/middleware):
await my_workflow(args, runner=ManagedAgentRunner(my_configured_agent))

The agent= source decides how each call's agent is obtained: None → a fresh ephemeral agent per call (isolated, mirrors DefaultAgentRunner); a Agent instance → reuse it; a registry name (str) → resolve it per call; a zero-arg factory → build a fresh agent per call.

Per-call targeting with using= — a single workflow can route different calls to different configured agents (multi-model sub-agents and per-task cost optimisation):

@workflow(name="triage")
async def triage(args, ctx):
    # cheap, fast model for the easy classification...
    label = await agent("classify the ticket", using="cheap-classifier")
    # ...escalate the hard synthesis to a configured premium agent
    return await agent("write the resolution", using="premium-writer")

await triage(args, runner=ManagedAgentRunner())   # both agents come from the registry

using accepts a Agent instance or a registry name and overrides the runner's source for that one call. DefaultAgentRunner rejects using= with a clear error (it has no agent registry to resolve against); SmartRoutingRunner forwards it to its base_runner, so using= works when the base is a ManagedAgentRunner.

Notes:

Separate usage ledgers. Per-call tokens/cost feed the per-run WorkflowBudget; Agent also records completed calls in the active runtime tracker when cost tracking is enabled. A journal replay makes no new model request.
Isolation. The ephemeral path sets auto_register=False and memory=None. A factory controls those settings itself; return a fresh agent with those options for the same isolation. The runner supplies no conversation_id. A reused agent may still share state through its cache, tools, middleware, or working memory.
tools= vs toolsets=. Per-call tools= only applies on the ephemeral path (pydantic-ai has no per-call tools parameter); for a reused agent, configure tools on the agent or pass toolsets=.
Composes with everything: SmartRoutingRunner(base_runner=ManagedAgentRunner(...)), cascade(), budgets and the journal all work unchanged.

Streaming¶

Stream a single sub-agent's output token-by-token with the stream() context manager (requires a streaming-capable runner — DefaultAgentRunner and ManagedAgentRunner both are). It honours the budget, concurrency gate, journal and cost accounting exactly like agent():

from pyfly.agentic.workflows import workflow, stream

@workflow(name="live_report")
async def live_report(args, ctx):
    async with stream("write a cited report", model=args.model) as s:
        async for delta in s.text():
            print(delta, end="", flush=True)   # surface tokens as they arrive
    return s.output                            # the full text, available after the block

On resume, a cached stream() call yields its full output once (the model is not re-invoked). A runner that doesn't support streaming raises WorkflowError.

Structured output. stream() accepts output_type= like agent(). Token deltas are inherently a text concept, so for a structured (non-text) run s.text() transparently falls back to yielding stringified partial-object snapshots (whole objects, not token deltas) instead of crashing; s.output still holds the validated model after the block.

Model routing & cost optimization¶

The cheapest model that can do the job should do the job. Two complementary tools give a workflow automatic per-task model selection — the workflow body is unchanged; only which model runs each agent() call changes.

`SmartRoutingRunner` — cheapest capable model per call (a-priori)¶

A drop-in AgentRunner: give it model tiers ordered cheap → expensive and a selection strategy. Each agent() call with no explicit model= is routed; an explicit model= always wins. A transient error escalates up the ladder.

from pyfly.agentic.workflows import (
    SmartRoutingRunner, ComplexityHeuristicStrategy, CostFloorStrategy,
)

runner = SmartRoutingRunner(
    tiers=["anthropic:claude-haiku-4-5", "anthropic:claude-sonnet-4-5"],
    strategy=ComplexityHeuristicStrategy(),   # training-free: routes by prompt length + cues
    fallback=True,                            # escalate to the next tier on a transient error
)
report = await run_workflow("deep_research", args, runner=runner)

ComplexityHeuristicStrategy (default) — no extra LLM call; short/simple prompts route to the cheapest tier, long or reasoning-heavy prompts escalate.
CostFloorStrategy — always start at the genai-prices cheapest tier (priced from the actual prompt length), leaving escalation to fallback.
Custom — implement ModelSelectionStrategy.select(prompt, tiers) -> int (e.g. an embedding router or a cheap classifier model).

Every decision emits a structured event (route.select, route.escalate), so a cheap-model quality regression is observable, not silent.

`cascade()` — escalate only on low confidence (a-posteriori)¶

The FrugalGPT trade-off: run the cheapest tier, score its answer, and escalate only when confidence is below threshold. Typically the strongest cost lever.

from pyfly.agentic.workflows import cascade

result = await cascade(
    "Summarise this contract clause: ...",
    tiers=["anthropic:claude-haiku-4-5", "anthropic:claude-sonnet-4-5"],
    threshold=0.7,            # accept the first tier scoring >= 0.7
    # confidence=my_async_scorer(output)->float, or judge_model=... (defaults to the cheapest tier)
    output_type=None,         # constrain each tier to a pydantic model (like agent())
    instructions=None,        # per-call system instructions for every tier
    max_escalations=None,     # cap how many tiers to climb (default: all of them)
)
result.output, result.tier, result.model, result.confidence, result.escalations

When no confidence callable is given, a judge model rates each answer 0–1 (defaulting to the cheapest tier as judge). Each tier emits a cascade.tier event. Cascades go through agent(), so they honour the run's budget and journal.

Pricing a model directly¶

from pyfly.agentic.workflows import price_model
usd = price_model("anthropic:claude-haiku-4-5", input_tokens=1000, output_tokens=500)

Verify combinators¶

Built on the primitives, these encode the "fan-out → reduce → decide" patterns that make multi-agent output trustworthy:

from pyfly.agentic import get_config
from pyfly.agentic.workflows import adversarial_verify, judge_panel, loop_until_dry

# Spawn N skeptics prompted to REFUTE; survives only if fewer than a majority refute.
ok = await adversarial_verify("claim: X causes Y", model=get_config().default_model, n=3)

# A panel of DIFFERENT models votes; returns a structured Verdict.
verdict = await judge_panel("X causes Y", judges=["openai-responses:gpt-6-luna", "anthropic:claude-sonnet-5"])
verdict.survived, verdict.support, verdict.votes   # bool, fraction, ((model, yes), ...)

# Keep producing until K consecutive rounds surface nothing new (catches the tail).
items = await loop_until_dry(produce_batch, dry_rounds=2, max_rounds=8, key=lambda x: x.id)

A heterogeneous-model panel (judge_panel) is a stronger signal than asking one model repeatedly, and its Verdict doubles as a confidence oracle for cascade().

Sub-workflows¶

Compose workflows by running one inline as a step of another with subworkflow. The child inherits the parent's budget, concurrency gate, journal and runner — its agent() calls count against the same budget and resume from the same journal (one sequence stream across the whole nested run):

from pyfly.agentic.workflows import workflow, agent, subworkflow

@workflow(name="summarize")
async def summarize(args, ctx):
    return await agent(f"summarize: {args}")

@workflow(name="report")
async def report(args, ctx):
    bullets = await subworkflow("summarize", args)   # by name, or pass the Workflow object
    return await agent(f"turn into a report: {bullets}")

To run a workflow in isolation (its own fresh budget), call other_wf.run(...) directly instead.

Human-in-the-loop¶

human(prompt) pauses a run for external input. It raises WorkflowInterrupt; you collect the answer, provide() it, and re-run with the same journal to resume past the pause. The answer is journaled under its sequence number, so the same replay requirements apply:

from pyfly.agentic.workflows import workflow, agent, human, Journal, WorkflowInterrupt

@workflow(name="triage")
async def triage(args, ctx):
    draft = await agent(f"draft a reply to: {args}")
    decision = await human(f"Approve this reply?\n\n{draft}")
    if decision.lower().startswith("y"):
        return draft
    return await agent(f"revise per feedback '{decision}': {draft}")

journal = Journal()
try:
    result = await triage.run(ticket, journal=journal)
except WorkflowInterrupt as it:
    it.provide(input(it.prompt))                         # a human reviews and answers
    result = await triage.run(ticket, journal=journal)   # resumes past human()

Pair human() with a JournalBackend for durable approvals that survive a process restart.

Telemetry¶

Pass events=callable to a run to receive structured events: workflow.start / workflow.end (with agents, tokens, cost_usd), phase.start / phase.end, agent.start / agent.end (with label, phase, seq, tokens, cost_usd; streamed calls also carry stream: True), route.select / route.escalate, cascade.tier, subworkflow.start / subworkflow.end, human.pause, and log. Wire this into the observability layer for live per-phase token/cost/agent/time counters.

API reference¶

Symbol	Purpose
`workflow(name=None, *, args_schema=None, description="", register=True)`	Decorator → a registered, runnable `Workflow`.
`Workflow.run(args, *, budget, runner, journal, events, run_id)`	Execute; returns the body's return value.
`run_workflow(name, args, **opts)`	Look up a registered workflow by name and run it.
`subworkflow(name_or_wf, args)`	Run another workflow inline, inheriting the parent's budget/journal/runner.
`agent` / `parallel` / `pipeline` / `phase` / `log`	The DSL primitives.
`stream(prompt, ...)`	Async context manager → a `StreamHandle` (`.text()` deltas, `.output`); streams one sub-agent.
`StreamHandle` / `StreamingAgentRunner`	The streaming result handle and the runner protocol a runner implements to support `stream`.
`human(prompt)`	Pause for external input (raises `WorkflowInterrupt`; resume via the same journal).
`map_agents(items, fn, *, strict=False)`	Run `fn(item)` per item concurrently — sugar over `parallel` (no late-binding lambda).
`WorkflowBudget`	Concurrency / agent-count / token / USD cost / wall-clock ceilings.
`Journal` / `JournalBackend` / `FileJournalBackend`	Sequence-keyed resume cache with optional best-effort persistence.
`AgentRunner` / `AgentCall`	The runner seam (`AgentCall` carries `output`, `tokens`, `cost_usd`).
`ManagedAgentRunner`	Default runner: each call runs through a `Agent` with its configured middleware, tools, and model. Pair with `agent(..., using=)` for multi-model sub-agents.
`DefaultAgentRunner`	Lightweight alternative: each call is a bare `pydantic_ai.Agent`. Pass it explicitly for the zero-coupling path.
`SmartRoutingRunner` / `ComplexityHeuristicStrategy` / `CostFloorStrategy` / `ModelSelectionStrategy`	Cheapest-capable model per call (with fallback).
`cascade` / `CascadeResult`	Cheap-first, escalate on low confidence.
`price_model(model, *, input_tokens, output_tokens)`	USD price for a model at a token shape.
`adversarial_verify` / `judge_panel` / `Verdict` / `loop_until_dry`	Quality combinators.
`workflow_registry`	The active runtime’s `WorkflowRegistry`.
`current_workflow()`	The active `WorkflowContext` (raises outside a run).

Exceptions live in pyfly.agentic.exceptions: WorkflowError (base), WorkflowNotFoundError, WorkflowBudgetError, WorkflowContextError, and WorkflowInterrupt (the human-in-the-loop pause signal).