aiflow

A small kit for running multiple agentic flows on the OpenAI Agents SDK, with evals wired in from day 0. A shared core/ (provider, structured output, delivery, guardrails) is reused by self-contained flows under flows/. Designed to run on a small VPS via a systemd timer; delivers to Telegram (or stdout).

aiflow digest                      # RSS  -> curate (agent)            -> ranked digest
aiflow research "<question>"       # plan -> PARALLEL lookups          -> cited answer
aiflow orchestrate "<task>"        # orchestrator -> fixed-role workers (guardrailed)
aiflow review [ref|--file f]       # lead -> specialist reviewers (as_tool) -> review
aiflow dispatch "<task>"           # coordinator DESIGNS its own sub-agents (guardrailed)

The flows

Flow	What it does	The interesting bit
digest	Watches RSS feeds, an agent curates a ranked, cited digest.	fetch_article tool — the agent decides what to deep-read.
research	Decomposes a question, researches sub-queries in parallel, synthesizes a cited answer.	asyncio.gather fan-out; one bad lookup degrades gracefully.
orchestrate	An orchestrator delegates to fixed-role workers (delegate tool).	Agents-as-tools + guardrails: cost, parallelism, turns.
review	A lead reviewer consults specialist agents-as-tools (security/correctness/style) and consolidates one verdict.	SDK-native Agent.as_tool; three lenses beat one prompt.
dispatch	A coordinator designs and spawns its own sub-agents — it writes their instructions — to solve a task.	Dynamic sub-agents (à la Claude's Task tool) + guardrails + tool allowlist.

Three patterns for multi-agent work, by who defines the sub-agent: orchestrate picks a fixed role, review wires fixed specialists as tools, dispatch lets the parent author each sub-agent's system prompt at runtime.

Why it's structured this way

• core/ is shared, flows are self-contained. Provider wiring, JSON output, Telegram delivery, and guardrails live once in core/. Each flow owns its agents, tools, schemas, and flow.py. Add a flow = add a folder + register it.
• Provider-agnostic structured output. Each agent is asked for JSON and we validate it into a Pydantic model ourselves (core/structured.py), with a repair retry — so it works on OpenAI and OpenAI-compatible providers that ignore strict json_schema.
• Guardrails are first-class (core/limits.py): a RunBudget carried as the SDK run context enforces cost (token budget), parallelism (semaphore), and delegation caps; max_turns bounds every agent loop.
• Evals from day 0 (evals/<flow>/): per-flow datasets + graders + gated harnesses, plus deterministic unit tests for graders and guardrails.

Setup

Requires Python 3.11+ and uv.

uv venv && source .venv/bin/activate
uv pip install -e ".[dev]"
cp .env.example .env        # add OPENAI_API_KEY (Telegram optional)

Using an OpenAI-compatible provider (Ollama Cloud, DeepSeek, vLLM, ...)

Set OPENAI_BASE_URL and the SDK targets that endpoint instead of OpenAI — it switches to chat-completions and disables tracing (the exporter only talks to platform.openai.com). OPENAI_API_KEY is then your provider's key:

OPENAI_API_KEY=<provider-key>
OPENAI_BASE_URL=https://ollama.com/v1
CURATOR_MODEL=deepseek-v4-flash:cloud   # applies to every flow

Run

aiflow digest --max-items 6
aiflow research "How does SQLite handle concurrent writes and what is WAL mode?"
aiflow orchestrate "Compare X and Y, research both, then recommend." \
    --max-parallel 2 --max-subagents 5 --token-budget 60000
aiflow review --file path/to/changed.py          # or: aiflow review HEAD~1  /  --staged
aiflow dispatch "Give a balanced recommendation on <hard question>." \
    --max-parallel 3 --token-budget 80000

Each run prints to stdout if Telegram isn't configured. With OpenAI proper, runs are traced at https://platform.openai.com/traces.

Choosing models

Model choice is a feature of every agent, at three levels:

# 1. Per-run — override the model for the whole run:
aiflow digest --model gpt-oss:120b

# 2. Named tiers (.env) — route cheap vs. hard work:
FAST_MODEL=deepseek-v4-flash:cloud
SMART_MODEL=deepseek-v4-pro

• Per-agent (in code): every builder takes a model, e.g. build_curator(model="smart").
• Per-call (by the LLM): delegate and spawn_agent expose a model arg, so an orchestrator/coordinator can send simple subtasks to "fast" and hard ones to "smart". review's specialists default to the "fast" tier while the lead can run stronger.

resolve_model() maps None/"default" → the default model, "fast"/"smart" → the configured tiers (which fall back to the default when unset), and anything else is treated as a literal model id.

Evals

pytest -q                                  # grader + guardrail unit tests (no key)
python evals/digest/run_evals.py --no-judge    # digest: structure + selection (cheap)
python evals/digest/run_evals.py               # + LLM-as-judge faithfulness
python evals/research/run_evals.py             # research: cited-answer invariants (live)
python evals/review/run_evals.py               # review: catches planted bugs (live)

Harnesses exit non-zero when a quality gate is missed, so they slot into CI / a pre-deploy check. Grow the datasets from real failures you see in traces.

Which multi-agent flow do I use?

Three of the flows are "a lead agent + worker sub-agents," which makes them look interchangeable. They aren't — they sit at three points on one spectrum: how much the parent decides at runtime.

   review            orchestrate              dispatch
fixed roster      fixed roles,            parent invents the
of specialists    dynamic tasks           roles AND writes them
   ──────────────────────────────────────────────────────────▶
              more decided by the parent at runtime

	review	orchestrate	dispatch
SDK mechanism	Agent.as_tool()	custom delegate(task, role) tool	dynamic Agent(...) in spawn_agent
What the lead sees	three named tools (review_security, …)	one tool with a role param	one tool; it writes instructions
Who defines the sub-agent	fixed specialists (prebuilt)	fixed roles, parent writes the task	parent writes the whole system prompt
Sub-agent count	exactly 3	dynamic	dynamic
Guardrails	none needed (3 cheap calls)	full RunBudget	full RunBudget + tool allowlist
Best when	a stable set of perspectives you always want	subtasks unknown until runtime	you can't predict what kinds of experts are needed

The mental models

• review — the synthesizer. The specialists are fixed and named, so the model sees review_security as a discrete capability and reasons about when to call it, like any other tool. Use as_tool for a stable roster.
• orchestrate — the planner. One generic delegate tool; the parent decides the decomposition and authors each subtask at runtime, picking from a fixed set of worker roles. Use a custom tool when the work is dynamic and you need a control point to wrap guardrails.
• dispatch — the manager. The parent goes one step further and writes each sub-agent's system prompt itself (à la Claude's Task tool). Use it when even the kinds of specialists can't be known ahead of time.

The honest caveat: the mechanisms partly overlap — you could build review on delegate, or orchestrate with as_tool. The deciding factor is whether your sub-agents are a fixed, named roster (as_tool reads cleanest) or a dynamic fan-out that needs budget/parallelism control (a custom tool + RunBudget).

Guardrails (orchestrate + dispatch flows)

Both spawning flows share core/limits.py. The RunBudget is passed as the SDK run context, so the spawn tool reads it via RunContextWrapper.

Guardrail	Flag	Mechanism
Max parallel sub-agents	--max-parallel	asyncio.Semaphore in RunBudget.slot()
Cost	--token-budget	token accounting; spawning refused once exhausted
Max total sub-agents	--max-subagents	counter checked before each spawn
Max turns	--max-turns, --worker-max-turns	passed to Runner.run(..., max_turns=)
Capability (dispatch)	tool allowlist	spawn_agent grants only vetted tools

A refused spawn returns a message (not an exception), so the coordinator adapts and answers with what it has. Telemetry prints at the end: subagents=3 peak_parallel=2/2 tokens=32676/60000 refusals=0. The guardrail logic has deterministic unit tests in tests/test_limits.py (no API needed).

Deploy to a VPS

docker build -t aiflow:latest -f deploy/Dockerfile .
sudo mkdir -p /etc/aiflow && sudo cp .env /etc/aiflow/.env
sudo cp deploy/systemd/digest.{service,timer} /etc/systemd/system/
sudo systemctl daemon-reload && sudo systemctl enable --now digest.timer

The timer runs aiflow digest daily at 07:30 (edit OnCalendar). For other flows, point a service's ExecStart at aiflow research ... / aiflow orchestrate ....

Layout

src/aiflow/
  cli.py  __main__.py          aiflow <flow> ... dispatcher
  core/
    config.py provider.py      settings; provider wiring (OpenAI / compatible)
    structured.py              run_json_agent: provider-agnostic typed output
    web.py tools.py            fetch_text + shared fetch_article tool
    limits.py                  Limits + RunBudget guardrails
    delivery/telegram.py       send_text (Telegram or stdout)
  flows/
    base.py __init__.py        Flow base + registry
    digest/      agent profile sources schemas flow
    research/    agent tools schemas flow         (parallel fan-out)
    orchestrate/ agent workers tools schemas flow (delegate to fixed roles + guardrails)
    review/      agent specialists diff schemas flow (Agent.as_tool specialists)
    dispatch/    agent tools schemas flow         (coordinator designs sub-agents + guardrails)
evals/digest/ research/ review/   datasets + graders + harnesses
tests/         grader + guardrail + review unit tests (no key)
deploy/        Dockerfile + systemd timer