Flow

Agent Flow lets you define multiple agents in a single flow.yaml file, wire them together with delegate sinks, and run them all with one command. The agent topology is compiled into a pydantic-graph execution graph, where each agent is a graph step and fan-out delegation runs in parallel via Fork/Join. Delegate sinks route output from one agent to the next as an immutable envelope.

Agents start in tiers based on after. Each agent is a standalone unit connected to others via delegate sinks.

Related: enable Durability to make a flow resumable after a crash or interruption, and add a Blackboard tool when agents need a shared structured key-value store instead of passing prose between each other.

Quick Start

# flow.yaml
name: my-pipeline
description: Simple producer-consumer pipeline
agents:
  producer:
    use: roles/producer.yaml
    then:
      to: consumer
  consumer:
    use: roles/consumer.yaml
    after:
      - producer

# Validate
initrunner flow validate flow.yaml

# Start (foreground, Ctrl+C to stop)
initrunner flow up flow.yaml

The public document is flat. Old envelopes still load; convert them with initrunner doctor --fix PATH. See Envelope Migration.

A flow needs edges. Since v2026.8.6, flow validate on a composed document with no then or after says the file is a group of independent agents instead of crashing with an AttributeError.

Scaffold a Project

Use initrunner flow new to generate a complete multi-agent project with role files and a flow.yaml:

initrunner flow new my-pipeline                          # default: chain pattern
initrunner flow new my-pipeline --pattern fan-out        # dispatcher + parallel workers
initrunner flow new my-pipeline --pattern route          # intake with sense-based routing
initrunner flow new my-pipeline --agents 4               # customize agent count
initrunner flow new my-pipeline --shared-memory          # enable shared memory
initrunner flow new my-pipeline --list-patterns          # show available patterns

Three patterns are available:

Each pattern generates a ready-to-run project directory with role YAML files and a flow.yaml. Review the generated files, customize as needed, then run with initrunner flow up flow.yaml.

Flow Definition

name: my-pipeline             # required
description: A pipeline       # optional
spec_version: 3
agents:                       # required, at least one agent
  agent-a:
    use: roles/a.yaml
  agent-b:
    use: roles/b.yaml
    after:
      - agent-a
shared_memory:                # optional
  enabled: false
  store_path: null
  max_memories: 1000
  store_backend: lancedb
shared_documents:             # optional
  enabled: false
  store_path: null
  store_backend: lancedb
  embeddings:
    provider: ""
    model: ""

Top-Level Fields

Agent Configuration

Each entry in agents configures one agent.

agents:
  my-agent:
    use: roles/my-role.yaml         # path to a role file, or inline prompt:
    # prompt: You are ...
    then:                           # optional, graph edge
      to: other-agent
    after:                          # optional, startup ordering
      - dependency-agent
    environment: {}

Delegate Sinks

Route an agent's output to other agents via in-memory queues. For sinks that send output outside the flow (webhooks, files, or custom functions), see Sinks.

# Single target
then:
  to: consumer
  queue_size: 100
  timeout_seconds: 60

# Fan-out to multiple targets
then:
  to:
    - researcher
    - responder
  strategy: sense
  keep_existing_sinks: true

Only successful runs are forwarded. Failed runs are silently skipped.

Routing Strategy

When a delegate sink has multiple targets, the strategy field controls how messages are routed.

The keyword and sense strategies use the same two-pass Intent Sensing logic used by --sense in the CLI. They score the agent's output text against each target agent's name, description, and tags from its role definition.

Before (static fan-out): every message goes to ALL targets:

triager:
  use: roles/triager.yaml
  then:
    to: [researcher, responder, escalator]

After (sense picks the right target):

triager:
  use: roles/triager.yaml
  then:
    to: [researcher, responder, escalator]
    strategy: sense              # ← one line added

How routing works

1. The upstream agent's output is scored against each target's role metadata (name, description, tags) using keyword matching.
2. If the output doesn't produce a confident match, the original user prompt (preserved from the head of the delegation chain) is also scored.
3. For sense strategy, if both attempts are inconclusive, an LLM tiebreaker call selects the best target.
4. The message is forwarded to the selected target only (not fanned out).

Routing diagnostics are injected into the payload's trigger metadata as _flow_route_reason for audit visibility.

Optimizing roles for routing

The same tips from Intent Sensing: Writing Roles That Sense Well apply. Each target agent's role should have specific, non-overlapping tags and a clear description:

# roles/researcher.yaml
name: researcher
description: Researches topics in depth and gathers supporting evidence
tags: [research, analysis, investigation, evidence]

# roles/responder.yaml
name: responder
description: Responds directly to user queries with concise answers
tags: [response, chat, answer, reply]

# roles/escalator.yaml
name: escalator
description: Escalates complex issues to human operators
tags: [escalation, support, human, complex]

Dashboard configuration

The dashboard flow builder exposes routing strategy visually when creating a flow with the Route pattern. Three pill buttons (Broadcast / Keyword / Sense) appear below the slot picker with a "Recommended" badge on Sense. A collapsible detail section shows scoring weights (tags 3x, name 2x, description 1.5x) and per-slot quality indicators. The Route pattern supports variable agent counts (3-10) with semantic specialist names that directly feed into name-match scoring. The selected strategy is written into the generated flow.yaml.

Single target behavior

When only one target is specified, strategy has no effect. The message always goes to that target regardless of the strategy setting.

Ensemble Voting

The ensemble strategy fans the same prompt out to every target (like all), then a reducer picks one winning answer instead of concatenating the responses. Use it when you want several attempts at the same task and a single best result.

An ensemble sink needs at least two targets and an ensemble: block. The block is required when strategy is ensemble and is rejected for any other strategy.

drafter:
  use: roles/drafter.yaml
  then:
    to: [writer-a, writer-b, writer-c]
    strategy: ensemble
    ensemble:
      mode: majority

The three modes pick a winner differently:

• majority: the most frequent identical answer wins. Ties break on the lowest topology index, so the result is deterministic. No extra API calls.
• weighted: the answer from the highest-weight target wins. Requires a non-empty weights map whose keys are all target names. Ties break on index. No extra API calls.
• judge: an LLM judge scores each candidate against judge_criteria and the highest-scoring answer wins. This costs one judge call per candidate.

Weighted example:

drafter:
  use: roles/drafter.yaml
  then:
    to: [senior, junior]
    strategy: ensemble
    ensemble:
      mode: weighted
      weights:
        senior: 2.0
        junior: 1.0

Judge example:

drafter:
  use: roles/drafter.yaml
  then:
    to: [writer-a, writer-b, writer-c]
    strategy: ensemble
    ensemble:
      mode: judge
      judge_model: openai:gpt-4o-mini
      judge_criteria: [clarity, accuracy]

The winning answer flows downstream as a single envelope. Targets can be terminal, or they can feed a downstream agent. When a single ensemble source's targets all fan in to one downstream agent, the vote replaces concatenation and only the winning answer is passed along (mixed fan-ins keep concatenation).

Each vote is recorded on the audit chain with trigger_type ensemble_vote, storing the winning output and the full vote trace. Candidate strings are truncated to 1000 characters in the trace. Inspect votes with:

initrunner audit export --trigger-type ensemble_vote

Loop-Back Routing

A loop_back edge turns a forward delegation into a bounded refine loop. A writer delegates a draft to a critic, and the critic's output routes back to an upstream agent for another pass. This is the classic writer/critic refine pattern.

Flow graphs are otherwise acyclic. Every unmarked cycle is still rejected at validation time. Only an explicitly-marked loop_back edge may close a cycle.

writer:
  use: roles/writer.yaml
  then:
    to: critic
    loop_back:
      type: loop-back
      target: writer
      max_iterations: 4
      until:
        output: "contains:APPROVED"
critic:
  use: roles/critic.yaml
  after: [writer]

The until predicate supports only the output field. Its value is either:

• contains:<text>: exit when the output contains <text> (case-insensitive).
• <op><number>: exit when the first number parsed from the output satisfies the comparison, where <op> is one of >, >=, <, <=, ==. For example, ">0.8" exits on a self-reported confidence score above 0.8.

The loop is bounded two ways: max_iterations is a hard cap and until is an optional early exit. A per-edge iteration counter rides along on an immutable envelope, and the flow delegation depth limit of 20 agents stays in force as a backstop, so a misconfigured loop cannot run unbounded.

initrunner flow validate renders a loop-back edge in the Sink column as (loop-back: writer x4).

initrunner flow validate flow.yaml

A durable flow cannot use loop_back. Disable durability or remove the loop_back edge; the combination is rejected at validation.

Startup Order

Agents start in topological order based on after. Agents without dependencies start first, forming tiers of parallel startup. Shutdown happens in reverse order.

agents:
  inbox-watcher:
    use: roles/inbox-watcher.yaml
    then: { to: triager }
  triager:
    use: roles/triager.yaml
    after: [inbox-watcher]
    then: { to: [researcher, responder] }
  researcher:
    use: roles/researcher.yaml
    after: [triager]
  responder:
    use: roles/responder.yaml
    after: [triager]

Tier 0:  inbox-watcher          (no dependencies)
Tier 1:  triager                (depends on inbox-watcher)
Tier 2:  researcher, responder  (both depend on triager)

Process Supervision

Flat flow agents have no restart block. Each entry under agents accepts only use, prompt, model, tools, tools_mode, environment, triggers, then, after, and guardrails. Any other key, including restart and health_check, is rejected at validation:

[ERROR] document
  1 validation error for AgentDocument
agents.responder.restart
  Extra inputs are not permitted [type=extra_forbidden]

The deprecated envelope shape still parses restart and health_check, but records them as inert and never acts on them. Converting with initrunner doctor --fix PATH drops them.

Per-agent run and error counts are still tracked and available via agent_health(). In daemon mode, each trigger event spawns an independent graph run, and a failed run increments that agent's error counter.

For process-level supervision, install the flow as a systemd user service with initrunner flow install. The generated unit sets Restart=on-failure with RestartSec=10, so the flow process is restarted on failure. See Systemd Deployment.

Runtime Architecture

Graph-Based Execution

Since v2026.3.8, flow and team runners use pydantic-graph for orchestration instead of thread-per-agent. Agents are modeled as graph nodes with edges representing delegate sinks. Fan-out, routing, and delegation run as graph steps with native async agent execution.

Since v2026.4.8, tool call start/complete events and a usage event (with token counts and cost) are streamed via SSE for flow runs, matching the agent stream contract. The dashboard displays these in the unified bottom panel with live tool activity.

Flow YAML
  │
  └── pydantic-graph
        ├── Step: agent-a (entry)
        ├── Fork: agent-b, agent-c (then.to)
        ├── Join
        └── Step: agent-d (then.to)

Delegation edges come from each agent's then block. A single target becomes a linear step sequence, several targets under the default all strategy (or ensemble) become a fork/join pair whose branches run concurrently, and keyword or sense routing is resolved at the graph edge by a decision node. after sets daemon startup tiers and dependency validation, not graph edges: an agent reached only through after, with no inbound then, is never executed by the graph. When a child omits after, its startup dependencies are derived from the then edges pointing at it. Agent executions run as native async calls within each graph step.

One-Shot and Daemon Execution

run_once() builds the graph and runs it via anyio.run(), so fan-out branches execute concurrently as anyio tasks and each step runs the agent natively async. In daemon mode, start() runs an anyio event loop on a background thread; trigger events (cron, webhook, file watcher) are enqueued to a bounded threading.Queue(maxsize=32), which blocks trigger threads when full to provide backpressure. A dispatcher polls the queue and spawns an independent graph run per event, so multiple runs execute concurrently with no shared mutable state.

Shutdown Semantics

1. First Ctrl+C (or SIGTERM) sets the shutdown event, and the dispatcher stops accepting new trigger events.
2. In-flight graph runs complete naturally.
3. The daemon thread joins (30s timeout).

A second Ctrl+C force-exits immediately.

Shared Memory

When shared_memory.enabled is true, all agents in the flow share a single memory database. One agent's remember() calls become visible to every other agent's recall().

Configuration

shared_memory:
  enabled: true
  store_path: null          # default: ~/.initrunner/memory/{name}-shared.lance
  max_memories: 1000
  store_backend: lancedb

All agents sharing a memory store must use compatible embedding models (same dimensions). Keep memory.embeddings consistent across roles, or omit it to let all agents derive from their model provider defaults.

Shared Documents

When shared_documents.enabled is true, all agents in the flow share a single document store. This lets you ingest documents once (e.g. via one agent's ingest config) and have every agent's search_documents tool query the same store.

Unlike shared memory, shared documents requires explicit embedding configuration at the flow level. This prevents embedding model mismatches between roles querying the same store.

Configuration

shared_documents:
  enabled: true
  store_path: ./shared-docs.lance   # optional, default: ~/.initrunner/stores/{name}-shared.lance
  embeddings:
    provider: openai                # required when enabled
    model: text-embedding-3-small   # required when enabled
agents:
  researcher:
    use: roles/researcher.yaml     # has ingest config with sources
  writer:
    use: roles/writer.yaml         # no ingest config needed

How It Works

At startup, apply_shared_documents() patches each agent's role definition:

• Roles with ingest: configured: the existing store_path, store_backend, and embeddings are overridden with the shared values. All other ingest settings (sources, chunking) are preserved.
• Roles without ingest:: a minimal IngestConfig is injected with empty sources and the shared store settings. This registers the search_documents retrieval tool so the role can query the shared store without needing its own ingest config.

Shared documents is a flow-time config patch only. It does not run ingestion automatically. Run initrunner ingest against the role that has sources configured to populate the shared store.

Embedding Consistency

The flow definition owns the embedding configuration for the shared store. When shared_documents.enabled is true, both embeddings.provider and embeddings.model must be set explicitly. This is validated at parse time and prevents the situation where different roles derive different embedding models from their model provider.

Usage Pattern

1. Configure one role (e.g. researcher) with ingest.sources pointing at your documents.
2. Enable shared_documents with the same embedding model the researcher would use.
3. Run initrunner ingest roles/researcher.yaml to populate the shared store.
4. Start the flow. All agents can now query the shared documents via search_documents.

Coordinating with Shared State

When agents need to pass a named, structured value between each other rather than concatenated prose, add a Blackboard tool to a flow agent. The blackboard is a per-flow-run key-value store that any agent in the run can read and write, and its final state is recorded on the audit chain. See Blackboard for setup and limits.

Systemd Deployment

Install flow pipelines as systemd user services for production:

# Install the unit
initrunner flow install flow.yaml

# Start
initrunner flow start my-pipeline

# Enable on boot
systemctl --user enable initrunner-my-pipeline.service

# Monitor
initrunner flow status my-pipeline
initrunner flow logs my-pipeline -f

Environment Variables

Systemd services don't inherit shell exports. Provide secrets via environment files:

• {flow_dir}/.env for project-level secrets
• ~/.initrunner/.env for user-level defaults

Use --generate-env to create a template .env file:

initrunner flow install flow.yaml --generate-env

User Lingering

To keep services running after logout:

loginctl enable-linger $USER

Example: Email Pipeline

inbox-watcher ──> triager ──> researcher
                     │
                     └──────> responder

name: email-pipeline
description: Multi-agent email processing pipeline
agents:
  inbox-watcher:
    use: roles/inbox-watcher.yaml
    then:
      to: triager
  triager:
    use: roles/triager.yaml
    after: [inbox-watcher]
    then:
      to: [researcher, responder]
      circuit_breaker_threshold: 5
  researcher:
    use: roles/researcher.yaml
    after: [triager]
  responder:
    use: roles/responder.yaml
    after: [triager]

Agent Roles

Each agent points to a standalone role YAML. Here are the two key roles in this pipeline:

roles/triager.yaml routes emails to the right handler:

name: triager
description: Routes emails to the right handler
prompt: >
  You are an email triage agent. Analyze the email summary and
  determine if it needs research (technical questions, data requests)
  or a direct response (simple inquiries, acknowledgments).
  Output your decision and reasoning clearly.
model:
  provider: openai
  name: gpt-4o-mini
  temperature: 0.1
guardrails:
  max_tokens_per_run: 2000
  timeout_seconds: 30

roles/responder.yaml drafts email responses:

name: responder
description: Drafts email responses
prompt: >
  You are an email response agent. Given a triaged email that needs
  a direct response, draft a professional, helpful reply. Keep the
  tone friendly and concise.
model:
  provider: openai
  name: gpt-4o-mini
  temperature: 0.5
guardrails:
  max_tokens_per_run: 3000
  timeout_seconds: 30

Agent roles are minimal. They focus on a single task and don't need triggers or sinks (the flow file handles routing). This keeps each agent simple and testable independently.

Example: CI Pipeline

A webhook-driven pipeline that processes CI events, diagnoses build failures, and sends notifications.

webhook-receiver ──> build-analyzer ──> notifier

flow.yaml

name: ci-pipeline
description: CI event processing pipeline
agents:
  webhook-receiver:
    use: roles/webhook-receiver.yaml
    then:
      to: build-analyzer
  build-analyzer:
    use: roles/build-analyzer.yaml
    after: [webhook-receiver]
    then:
      to: notifier
  notifier:
    use: roles/notifier.yaml
    after: [build-analyzer]

roles/notifier.yaml

This agent combines Slack messaging with the GitHub commit status API:

name: ci-notifier
description: Sends Slack notifications and updates GitHub commit status
prompt: |
  You are a CI notification agent. You receive analyzed build events and:

  1. Send a formatted Slack notification:
     - Success: "✅ Build passed — [repo] @ [branch] ([sha])"
     - Failure: "❌ Build failed — [repo] @ [branch] ([sha])\n
       Diagnosis: [diagnosis]\nCategory: [category]"
     - Include the build URL as a link
     - Add a timestamp via get_current_time

  2. Update the GitHub commit status using the create_commit_status API
     endpoint:
     - state: "success" or "failure"
     - description: brief status message
     - context: "ci-pipeline/initrunner"

  Always send both the Slack message and the GitHub status update.
model:
  provider: openai
  name: gpt-4o-mini
  temperature: 0.0
tools:
  - slack:
      webhook_url: "${SLACK_WEBHOOK_URL}"
      default_channel: "#ci-alerts"
      username: CI Pipeline
      icon_emoji: ":construction_worker:"
  - api:
      name: github-status
      description: GitHub commit status API
      base_url: https://api.github.com
      headers:
        Accept: application/vnd.github.v3+json
      auth:
        Authorization: "Bearer ${GITHUB_TOKEN}"
      endpoints:
        - name: create_commit_status
          method: POST
          path: "/repos/{owner}/{repo}/statuses/{sha}"
          description: Create a commit status check
          parameters:
            - name: owner
              type: string
              required: true
            - name: repo
              type: string
              required: true
            - name: sha
              type: string
              required: true
            - name: state
              type: string
              required: true
              description: "pending, success, failure, or error"
            - name: description
              type: string
              required: false
            - name: context
              type: string
              required: false
              default: "ci-pipeline/initrunner"
          body_template:
            state: "{state}"
            description: "{description}"
            context: "{context}"
          timeout_seconds: 15
  - datetime
guardrails:
  max_tokens_per_run: 15000
  max_tool_calls: 10
  timeout_seconds: 60

Test the webhook

# Start the pipeline
initrunner flow up flow.yaml

# In another terminal, send a test event
curl -X POST http://localhost:9090/ci-webhook \
  -H "Content-Type: application/json" \
  -d '{
    "source": "github-actions",
    "repo": "myorg/myapp",
    "branch": "main",
    "sha": "abc12345",
    "status": "failure",
    "author": "dev@example.com",
    "message": "fix: update auth middleware",
    "url": "https://github.com/myorg/myapp/actions/runs/12345"
  }'

What to notice: The notifier combines two tool types: slack for human-readable alerts and api for machine-readable GitHub status updates. The webhook receiver uses a webhook trigger (port 9090), and the flow file wires all three agents together with delegate sinks.

Example: Support Desk

intake ──[sense]──> researcher | responder | escalator

A support desk pipeline where strategy: sense on the intake's delegate sink auto-routes each message to the best-matching handler, with no static fan-out.

flow.yaml

name: support-desk
description: >
  Support desk pipeline with intelligent auto-routing. An intake agent
  summarizes incoming requests, then sense routing automatically sends
  each request to the right handler -- researcher for technical issues,
  responder for quick answers, or escalator for urgent/complex cases.
  No static fan-out: each message goes to exactly one target.
agents:
  intake:
    use: roles/intake.yaml
    then:
      # strategy: sense uses keyword scoring + LLM tiebreak to pick the
      # best target for each message. Use "keyword" for zero API calls,
      # or "all" to fan out to every target (default).
      strategy: sense
      to:
        - researcher
        - responder
        - escalator

  researcher:
    use: roles/researcher.yaml
    after:
      - intake

  responder:
    use: roles/responder.yaml
    after:
      - intake

  escalator:
    use: roles/escalator.yaml
    after:
      - intake

Agent Roles

roles/intake.yaml receives and summarizes support requests:

name: intake
description: Receives support requests and summarizes them for triage
tags:
  - support
  - intake
prompt: >
  You are a support intake agent. When you receive a support request,
  produce a concise summary including: the customer's issue, urgency
  level, and the type of action needed (research, direct response,
  or human escalation). Be factual and brief.
model:
  provider: openai
  name: gpt-5-mini
  temperature: 0.1
guardrails:
  max_tokens_per_run: 1000
  timeout_seconds: 30

roles/researcher.yaml investigates technical issues:

name: researcher
description: Investigates technical issues and gathers diagnostic information
tags:
  - research
  - analysis
  - investigation
  - technical
  - diagnose
prompt: >
  You are a technical research agent for a support desk. When you
  receive a triaged support request that requires investigation,
  research the issue thoroughly. Produce a structured report with:
  root cause analysis, relevant documentation references, and
  recommended resolution steps.
model:
  provider: openai
  name: gpt-5-mini
  temperature: 0.3
guardrails:
  max_tokens_per_run: 4000
  timeout_seconds: 60

initrunner flow up flow.yaml

What to notice: The strategy: sense on the intake's delegate sink means each message is scored against the three targets' role metadata (name, description, tags). Because the tags are non-overlapping (researcher uses [research, analysis, investigation, technical, diagnose] while responder and escalator cover different domains), keyword scoring alone resolves most messages without an LLM call. See Routing Strategy for details.

Example: Content Pipeline

content-watcher ──> researcher ──> writer
                        │
                        └──────> reviewer

Uses process_existing: true on the file watch trigger to handle files already in the directory on startup. See Triggers for details.

name: content-pipeline
description: >
  Multi-agent content creation pipeline. A file watcher monitors ./drafts/
  for new markdown or text files, extracts the topic, and delegates to a
  researcher. The researcher fans out to a writer (polished output) and a
  reviewer (QA checks).
agents:
  content-watcher:
    use: roles/content-watcher.yaml
    then:
      to: researcher
  researcher:
    use: roles/researcher.yaml
    then:
      to:
        - writer
        - reviewer
    after:
      - content-watcher
  writer:
    use: roles/writer.yaml
    after:
      - researcher
  reviewer:
    use: roles/reviewer.yaml
    after:
      - researcher

Key patterns:

• process_existing: true on the file watch trigger lets content-watcher pick up drafts already in the directory, not just new changes. See Triggers.
• Fan-out delegation from researcher to both writer and reviewer runs the two downstream agents in parallel on the same input.

See also: Team Mode for single-file multi-agent collaboration. It is simpler than Flow when you need multiple perspectives on the same task rather than independent agents.