Multi-model routing in AutoGen: GPT-5 and Llama 4

Multi-model routing in AutoGen lets you assign different models to different agents without rewriting your agent logic. This tutorial shows how to wire GPT-5 for reasoning-heavy tasks and Llama 4 for cost-sensitive generation through one OpenAI-compatible endpoint, with automatic fallback and usage tracking built in.

Prerequisites

• Python 3.10+
• An n4n.ai API key (or any OpenAI-compatible gateway that serves GPT-5 and Llama 4)
• pip install autogen-agentchat autogen-ext openai python-dotenv

Create a .env file with your gateway credentials:

N4N_API_KEY=your_key_here
N4N_BASE_URL=https://api.n4n.ai/v1

The base URL points to the OpenAI-compatible endpoint. If you’re using a different gateway, swap the values accordingly.

Project structure

multimodel_autogen/
├── .env
├── config.yaml
├── main.py
└── requirements.txt

requirements.txt pins the versions used in this tutorial:

autogen-agentchat==0.2.0
autogen-ext==0.2.0
openai==1.35.0
python-dotenv==1.0.1
pyyaml==6.0.1

Model configuration

AutoGen’s OpenAIChatCompletionClient accepts any OpenAI-compatible endpoint. We’ll define two clients — one pinned to GPT-5, one to Llama 4 — and pass them to the agents that need them.

config.yaml:

gpt5:
  model: "gpt-5"
  base_url: "${N4N_BASE_URL}"
  api_key: "${N4N_API_KEY}"
  temperature: 0.2
  max_tokens: 4000

llama4:
  model: "llama-4-maverick"
  base_url: "${N4N_BASE_URL}"
  api_key: "${N4N_API_KEY}"
  temperature: 0.4
  max_tokens: 4000

The gateway handles model availability. If GPT-5 is rate-limited, the same endpoint can fall back to another provider serving GPT-5 without changing your code.

Loading configuration

main.py starts by loading the YAML and expanding environment variables:

import os
import yaml
from dotenv import load_dotenv
from autogen_ext.models.openai import OpenAIChatCompletionClient

load_dotenv()

def load_config(path: str = "config.yaml") -> dict:
    with open(path) as f:
        raw = f.read()
    # Expand ${VAR} references
    expanded = os.path.expandvars(raw)
    return yaml.safe_load(expanded)

cfg = load_config()

gpt5_client = OpenAIChatCompletionClient(**cfg["gpt5"])
llama4_client = OpenAIChatCompletionClient(**cfg["llama4"])

Run a quick sanity check:

# Quick test — remove after verifying
async def test_clients():
    from autogen_core import CancellationToken
    for name, client in [("gpt-5", gpt5_client), ("llama-4", llama4_client)]:
        resp = await client.create(
            [{"role": "user", "content": "Say 'ok' and nothing else."}],
            cancellation_token=CancellationToken()
        )
        print(f"{name}: {resp.content[:50]}")

import asyncio
asyncio.run(test_clients())

Expected output:

gpt-5: ok
llama-4: ok

Both clients reach the gateway and return responses. If one fails, the gateway’s automatic fallback kicks in before the error reaches your code.

Building the agent team

We’ll create a two-agent workflow: a planner that uses GPT-5 for structured reasoning, and a writer that uses Llama 4 for long-form generation. The planner emits a JSON plan; the writer executes it.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_core import CancellationToken

planner = AssistantAgent(
    name="planner",
    model_client=gpt5_client,
    system_message=(
        "You are a planning agent. Given a topic, produce a JSON outline with "
        "these keys: title, sections (array of {heading, key_points}), "
        "target_audience, tone. Output ONLY valid JSON."
    ),
)

writer = AssistantAgent(
    name="writer",
    model_client=llama4_client,
    system_message=(
        "You are a technical writer. Given a JSON outline from the planner, "
        "write a complete, well-structured article in Markdown. Follow the "
        "outline exactly. Do not add commentary."
    ),
)

team = RoundRobinGroupChat(
    participants=[planner, writer],
    termination_condition=MaxMessageTermination(max_messages=2),
)

RoundRobinGroupChat alternates between agents. The planner goes first, emits the outline, then the writer consumes it and produces the final article. Two messages total — one per agent.

Running the workflow

async def run_workflow(topic: str) -> str:
    task = f"Write a technical article about: {topic}"
    result = await team.run(task=task, cancellation_token=CancellationToken())
    # The last message is the writer's output
    return result.messages[-1].content

if __name__ == "__main__":
    topic = "How Rust's ownership model prevents data races at compile time"
    article = asyncio.run(run_workflow(topic))
    print(article)

Run it:

python main.py

Expected output (truncated):

# How Rust's Ownership Model Prevents Data Races at Compile Time

## Introduction
Rust's ownership system is the cornerstone of its memory safety guarantees...

## Core Concepts
### Ownership Rules
- Each value has a single owner
- When the owner goes out of scope, the value is dropped
- ...

### Borrowing and References
- Immutable references allow multiple readers
- Mutable references allow one writer, no readers
- ...

## Compile-Time Enforcement
The borrow checker analyzes...

The planner used GPT-5 (low temperature, structured output) to produce a tight outline. The writer used Llama 4 (higher temperature, larger context) to expand it into a full article. Each model did what it’s good at.

Adding usage metering

The gateway returns usage in the standard OpenAI usage field. AutoGen surfaces it via ChatCompletionClient.create() response metadata. Wrap the client to log per-call token counts:

from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ChatCompletionClient, ModelCapabilities
from typing import Any, AsyncIterator, List, Optional
from autogen_core import CancellationToken
from autogen_core.model_context import ChatCompletionContext

class MeteredClient(ChatCompletionClient):
    def __init__(self, inner: OpenAIChatCompletionClient, model_name: str):
        self._inner = inner
        self._model_name = model_name
        self.total_prompt_tokens = 0
        self.total_completion_tokens = 0

    async def create(
        self,
        messages: List[dict],
        *,
        cancellation_token: CancellationToken,
        **kwargs: Any
    ) -> Any:
        response = await self._inner.create(
            messages, cancellation_token=cancellation_token, **kwargs
        )
        usage = getattr(response, "usage", None)
        if usage:
            self.total_prompt_tokens += usage.prompt_tokens
            self.total_completion_tokens += usage.completion_tokens
            print(
                f"[{self._model_name}] "
                f"prompt={usage.prompt_tokens} "
                f"completion={usage.completion_tokens} "
                f"total={usage.total_tokens}"
            )
        return response

    # Delegate remaining interface methods
    def __getattr__(self, name: str) -> Any:
        return getattr(self._inner, name)

Wrap both clients:

gpt5_metered = MeteredClient(gpt5_client, "gpt-5")
llama4_metered = MeteredClient(llama4_client, "llama-4")

planner = AssistantAgent(name="planner", model_client=gpt5_metered, ...)
writer = AssistantAgent(name="writer", model_client=llama4_metered, ...)

Run again. Console output now includes:

[gpt-5] prompt=247 completion=183 total=430
[llama-4] prompt=1156 completion=2847 total=4003

You now have per-model token accounting without instrumenting every call site. The gateway meters at the provider level; this client wrapper surfaces it in your application logs.

Routing directives (advanced)

The gateway honors x-n4n-routing headers for per-request steering. AutoGen’s OpenAIChatCompletionClient passes extra headers via extra_create_args. Use this when you need explicit control — for example, forcing a specific provider for compliance:

from autogen_ext.models.openai import OpenAIChatCompletionClient

strict_gpt5 = OpenAIChatCompletionClient(
    model="gpt-5",
    base_url=cfg["gpt5"]["base_url"],
    api_key=cfg["gpt5"]["api_key"],
    extra_create_args={
        "extra_headers": {"x-n4n-routing": '{"provider": "openai"}'}
    },
)

This forces the request to OpenAI’s GPT-5 deployment even if other providers are available. Omit the header to let the gateway optimize for latency and cost.

Error handling and retries

The gateway returns standard HTTP codes. AutoGen’s client wraps them in ModelClientError. Add a retry policy at the team level:

from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core import CancellationToken
from autogen_core.exceptions import ModelClientError
import asyncio

async def run_with_retry(topic: str, max_retries: int = 2) -> str:
    for attempt in range(max_retries + 1):
        try:
            return await run_workflow(topic)
        except ModelClientError as e:
            if attempt == max_retries:
                raise
            wait = 2 ** attempt
            print(f"Model error: {e}. Retrying in {wait}s...")
            await asyncio.sleep(wait)

This handles transient gateway errors (5xx, 429) without leaking into your business logic.

Checkpoint: what we built

Next steps

• Add a critic agent (third model, e.g., Claude) to review the writer’s output before finalizing
• Persist usage to a database for cost dashboards
• Implement dynamic model selection based on task classification (code → GPT-5, summarization → Llama 4)
• Use SelectorGroupChat instead of RoundRobinGroupChat for conditional agent routing

The pattern scales: one gateway, multiple models, clean agent boundaries. Swap models by editing config.yaml — no agent code changes required.