How to route tasks between specialized agents

Task routing between agents is the core problem in any multi-agent orchestration that scales beyond a demo. A naive approach bolts a giant prompt onto one model; a mature system routes each unit of work to a specialized agent with a narrow job. This guide gives you an end-to-end pattern for task routing between agents using an LLM classifier and isolated worker agents, with code you can run against any OpenAI-compatible endpoint.

Step 1: Define agent boundaries and contracts

Before writing a router, you need explicit contracts for each specialized agent. An agent should own one capability: SQL generation, document summarization, code review, etc. If two agents overlap, your router will guess and your logs will lie.

Define a minimal typed interface. This makes dispatch and testing straightforward.

from typing import TypedDict, Literal, Callable

AgentName = Literal["sql", "summarize", "code"]

class RouteDecision(TypedDict):
    agent: AgentName
    reason: str
    params: dict

# Each agent is a callable that takes extracted params and returns a string
Agent = Callable[[dict], str]

Keep the params schema loose but documented. The router will extract them; the agent validates. If an agent needs strict input, enforce it with Pydantic inside the agent, not in the router.

Step 2: Build a lightweight router with an LLM classifier

The router is itself an LLM call with a constrained output. Use JSON mode (or function calling) to force a decision. Point the OpenAI client at any compatible gateway—n4n.ai provides one endpoint covering 240+ models with automatic fallback when a provider is degraded—or your own proxy.

from openai import OpenAI
import json

client = OpenAI(base_url="https://your-gateway/v1", api_key="sk-...")

ROUTER_PROMPT = """You route tasks to specialized agents.
Available agents: sql, summarize, code.
Return JSON: {"agent": str, "reason": str, "params": dict}
Extract any fields the agent needs (e.g., table name, text, language)."""

def route(task: str) -> RouteDecision:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": ROUTER_PROMPT},
            {"role": "user", "content": task}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    data = json.loads(resp.choices[0].message.content)
    return data  # type: ignore

The router should run on a fast, cheap model. Its only job is classification, not execution. Set temperature=0 to keep routing deterministic.

Handling ambiguous input

If the task is genuinely unclear, route to a default agent that asks a clarifying question. Don’t let the router hallucinate an agent name. Validate the returned agent key against your registry before dispatch.

Step 3: Implement specialized agents as isolated callables

Each agent is a plain function that calls the model best suited for its task. Isolation means you can swap models, add caching, or unit-test without touching the router.

def sql_agent(params: dict) -> str:
    table = params.get("table", "users")
    resp = client.chat.completions.create(
        model="mistralai/mixtral-8x7b-instruct",
        messages=[
            {"role": "system", "content": f"Write a SQL query for table {table}."},
            {"role": "user", "content": params.get("question", "")}
        ]
    )
    return resp.choices[0].message.content or ""

def summarize_agent(params: dict) -> str:
    text = params.get("text", "")
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Summarize the text in 3 bullets."},
            {"role": "user", "content": text}
        ]
    )
    return resp.choices[0].message.content or ""

AGENTS: dict[AgentName, Agent] = {
    "sql": sql_agent,
    "summarize": summarize_agent,
    # "code": code_agent omitted for brevity
}

Note the different models per agent. Task routing between agents lets you match model capability to task cost. A summarization agent can use a smaller model; a code agent may need a stronger one.

Step 4: Wire the dispatch loop with error handling and fallback

The orchestrator calls the router, then dispatches. Misclassification happens. Wrap agent execution in try/except and fall back to a safe default.

def default_agent(params: dict) -> str:
    return "I couldn't route that confidently. Please specify sql, summarize, or code."

def orchestrate(task: str) -> str:
    try:
        decision = route(task)
    except json.JSONDecodeError:
        return default_agent({})

    agent_name = decision.get("agent")
    agent = AGENTS.get(agent_name, default_agent)
    try:
        return agent(decision.get("params", {}))
    except Exception as e:
        # log e, then retry with explicit correction
        return default_agent({"error": str(e)})

If you want resilience against provider outages, use a gateway that honors client routing directives and forwards cache-control hints. Your client base URL can stay constant while the backend fails over. That’s a config concern, not application logic.

Retry on invalid params

Agents should raise ValueError on missing required params. Catch it in orchestrate, then re-call route with a hint: "Previous route to {agent} failed: missing 'text'. Re-extract." This closes the loop without human intervention.

Step 5: Add observability and token metering

You can’t improve task routing between agents without seeing decisions. Log every route: input hash, chosen agent, latency, and token usage.

import logging

def orchestrate_with_logging(task: str) -> str:
    decision = route(task)
    logging.info("route_decision", extra={"agent": decision["agent"], "task": task[:50]})
    # ... dispatch as before
    return result

If your gateway provides per-token usage metering, pipe resp.usage into your metrics store. This tells you which agent burns budget and whether the router sends hard tasks to expensive models too often.

Keep traces correlated: assign a trace_id at the entrypoint and pass it through route and agent calls. When a user complains about a bad answer, you can reconstruct the exact path.

Step 6: Verify the system end to end

Verification means proving the router sends known tasks to the right agent and that the fallback works. Write a pytest suite with three layers.

import pytest
from mymodule import route, orchestrate, AGENTS

def test_router_sql():
    d = route("Get all users from the customers table")
    assert d["agent"] == "sql"
    assert "table" in d["params"]

def test_router_summarize():
    d = route("Summarize this article: ...")
    assert d["agent"] == "summarize"

def test_fallback_on_bad_agent():
    # force a bad route by monkeypatching route
    import mymodule
    mymodule.route = lambda t: {"agent": "nonexistent", "params": {}}
    out = orchestrate("anything")
    assert "couldn't route" in out.lower()

Run pytest -q. Success criteria:

Router unit tests pass with deterministic agent selection on fixed inputs.
Dispatch returns agent output (not fallback) for at least 9 of 10 representative tasks from your domain.
Forcing an unknown agent name triggers default_agent without raising.

Beyond tests, run a manual smoke test with real tasks and inspect logs. If the router consistently picks code for SQL questions, adjust the system prompt or add few-shot examples. Task routing between agents is not set-and-forget; treat the router prompt as code and version it.

Closing practice notes

Keep the router stateless. If you need conversation context, pass a compressed summary as part of the task string—don’t let the router hold session memory. Specialized agents can hold their own state if required, but the routing decision should be derivable from the current task alone.

When you add a new agent, register it in AGENTS and append its name to the router prompt. That’s the entire change surface. This pattern scales to dozens of agents without rewriting your orchestration layer.