Multi-Agent Workflows

Warning

Alpha. ABI Swarm orchestration (Orchestrator + Planner + Builder + ephemeral agents) is under active development and may change between releases.

AgentInteractionFlow is the engine that runs plans across multiple agents.

In the Orchestrator, this is what build_workflow constructs from the Planner’s plan — and it only runs after the user approves the plan (see Plan Confirmation), since building the workflow is also what invokes the Builder to create ephemeral agents.

What it does

Takes a plan (list of tasks with assigned agents) and runs them in the right order. Each task calls an agent and collects its response.

Creating a workflow

from abi_core.common.workflow import AgentInteractionFlow, InteractionFlowNode
from config import AGENT_CARD

# Create the flow
workflow = AgentInteractionFlow()

# Add a node (one task = one agent call)
node = InteractionFlowNode(
    task="Analyze Q4 revenue data",
    source_agent_card=AGENT_CARD,
    target_agent_card=analyst_card,  # AgentCard from discovery
    node_key="analyze",
    node_label="Revenue Analysis",
)
workflow.add_node(node)
workflow.set_source_card(AGENT_CARD)

# Execute and stream results
async for chunk in workflow.run_workflow():
    print(chunk)  # A2A response chunks from the target agent

Execution patterns

Sequential

Tasks run one after another. Each depends on the previous.

collect_data → analyze_data → write_report

The Planner expresses this with depends_on:

{
  "tasks": [
    {"task_id": "1", "description": "Collect data", "depends_on": []},
    {"task_id": "2", "description": "Analyze", "depends_on": ["1"]},
    {"task_id": "3", "description": "Report", "depends_on": ["2"]}
  ]
}

Parallel

Tasks with no dependencies between them run at the same time.

collect_sales ──┐
collect_costs ──┼→ analyze_all
collect_inventory─┘

{
  "tasks": [
    {"task_id": "1", "description": "Collect sales", "depends_on": []},
    {"task_id": "2", "description": "Collect costs", "depends_on": []},
    {"task_id": "3", "description": "Collect inventory", "depends_on": []},
    {"task_id": "4", "description": "Analyze all", "depends_on": ["1", "2", "3"]}
  ]
}

Hybrid

Mix of sequential and parallel.

classify ──→ analyze ──┐
guardian ──────────────┼→ synthesize
                       │
search_context ────────┘

Workflow state

from abi_core.common.workflow import Status

workflow.state  # Status.READY → RUNNING → COMPLETED

Each node also has its own state. The workflow tracks which nodes have completed and which are pending.

Heartbeat

Long-running workflows send heartbeat events to keep the SSE connection alive:

# In the Orchestrator's stream()
dag_result, heartbeats = await self._run_with_heartbeat(
    dag_coro, context_id, task_id, "Processing..."
)
for hb in heartbeats:
    yield hb  # Keeps the client connection alive

Next step

👉 Dependency Management