agent-orchestrator

Agent Orchestrator — Production Best Practices

Patterns from a production multi-agent system. The body is decisions; the mechanisms are one file away, and it is held under a 4750-token budget — a body that grows absorbs the layer that should have been split, and this one did until 2026-08-16.

Architecture Overview

User Question
    ↓
OrchestratorAgent.run(AgentContext)
    ├─ Shape check → one loop, or a planned path (§5, references/pipeline.md)
    ├─ Context loading (a parallel layer: staleness, sources, KB — §13)
    ├─ History trimming, then context budget allocation
    ├─ System prompt built from the live capabilities (§10)
    └─ Execute:
        ├─ SIMPLE: LLM → tools → sub-agents → results → LLM → … → answer   (§2)
        └─ PLANNED: plan → dependency layers → checker → checkpoints → done (§5)

1. The Orchestrator Pattern

Shared Context Object

Pass a single immutable-ish context object to all sub-agents:

@dataclass
class AgentContext:
    project_id: str
    user_question: str
    chat_history: list[Message]
    llm_router: LLMRouter           # provider abstraction with retry/fallback
    tracker: WorkflowTracker        # SSE event emitter for real-time UI
    workflow_id: str                 # unique ID for this request
    connection_config: ... | None   # external resource config
    user_id: str | None
    preferred_provider: str | None  # e.g. "openrouter"
    model: str | None               # e.g. "<provider>/<model-id>"
    extra: dict[str, Any]           # pipeline_action, flags, overrides

Key principles:

• Sub-agents never modify context — they return typed results
• Provider/model preferences flow down from user → project defaults → app defaults
• extra dict carries pipeline state, flags like _skip_complexity, session IDs

Sub-Agent Protocol

Every sub-agent extends a base class:

class BaseAgent(ABC):
    @abstractmethod
    async def run(self, context: AgentContext, **kwargs) -> AgentResult: ...

    @property
    @abstractmethod
    def name(self) -> str: ...

    @staticmethod
    def accum_usage(total, usage): ...  # merge token counters