Planner & Orchestrator

Warning

Alpha. The ABI Swarm (Orchestrator + Planner + Builder + ephemeral agents) is under active development. APIs, generated structure, and behavior may change between releases. Not recommended for production yet.

The Orchestrator is the entry point for complex requests. The Planner breaks them into smaller tasks and assigns agents. Together they coordinate multi-agent work.

How they work together

User request
  → Orchestrator
    ├─ Step 1: Is this simple or complex? + Is it allowed?
    ├─ Step 2: Decision (answer directly | send to Planner | block)
    ├─ Step 3: Planner breaks it into tasks → assigns agents
    └─ Step 4: Execute tasks → combine results
  → Response to user

The Orchestrator

Receives every request. Its pipeline:

  1. classify_query — Is this simple (answer directly) or complex (needs planning)?

  2. guardian_validate — Is this request allowed by security policies? (runs in parallel with classify)

  3. gate_decision — Based on classification + security: respond directly, call planner, or block

  4. call_planner — Send to Planner via A2A for task decomposition

  5. build_workflow — Turn the plan into an AgentInteractionFlow with nodes for each agent

  6. execute — Run the workflow, collect results

  7. synthesize — Use LLM to combine all results into a coherent response

# Orchestrator DAG (from main.py)
@agent.step(name="classify_query", input_map={"query": "$input.query"})
async def classify_query(query):
    text = await invoke(config.LLM_CONFIG, TRIAGE_PROMPT.format(query=query))
    parsed = clean_llm_json(text)
    return {"classification": parsed.get("classification", "complex")}

@agent.step(name="guardian_validate", input_map={...})
async def guardian_validate(query, context_id):
    # Calls Guardian agent via A2A
    ...
    return {"status": "approved", "allowed": True}

@agent.step(name="gate_decision", depends_on=["classify_query", "guardian_validate"])
async def gate_decision(classification, guardian_result):
    if not guardian_result["allowed"]:
        return {"action": "blocked", "message": guardian_result["reason"]}
    if classification == "simple":
        return {"action": "respond_direct"}
    return {"action": "call_planner"}

The Planner

Receives a query and produces a structured plan:

  1. LLM decomposition — Calls the LLM with a chain-of-thought prompt to break the task into sub-tasks

  2. parse_plan — Extracts structured JSON from the LLM response

  3. assign_agents — For each task, searches the Semantic Layer for the right agent

Output:

{
  "status": "ready",
  "plan": {
    "objective": "Analyze Q4 sales and generate report",
    "execution_strategy": "sequential",
    "tasks": [
      {
        "task_id": "task-1",
        "type": "analysis",
        "description": "Analyze Q4 revenue data",
        "agents": [{"name": "analyst", "url": "http://..."}]
      },
      {
        "task_id": "task-2",
        "type": "generation",
        "description": "Generate PDF report from analysis",
        "agents": [{"name": "reporter", "url": "http://..."}],
        "depends_on": ["task-1"]
      }
    ]
  }
}

If the Planner needs more info, it returns {"status": "needs_clarification", "clarification": "..."} and the Orchestrator forwards it to the user.

Get orchestration in your project

abi-core create swarm --name my-system

This creates a complete project with Orchestrator, Planner, Builder, Semantic Layer, Guardian, and all infrastructure ready to run.

Next step

👉 Multi-Agent Workflows