Scaling and Multi-Agent Architectures

Your agent works for 10 users. How do you make it run for 10,000? The challenges change: it’s no longer about instructions failing, but about provider rate limits, server memory, and your monthly bill. In this final lesson, you’ll learn how to scale your agents, design high-performance multi-agent architectures, and manage production constraints at scale.

Production Architecture

The Agent Backend with FastAPI

The first structural decision concerns the agent lifecycle. Building an Agent for every request wastes time and scatters configuration; FastAPI’s lifespan lets you instantiate them once at startup and keep them in a registry indexed by type. The route reads the requested type, cleanly rejects it if it doesn’t exist, then streams the response: run_streamed emits deltas as generation proceeds, making text appear on screen immediately instead of leaving the user staring at a loading spinner for eight seconds.

from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import StreamingResponse
from agents import Agent, Runner, function_tool
from contextlib import asynccontextmanager
import asyncio

# Pool of pre-configured agents
agents_registry = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize agents at startup
    agents_registry["sales"] = Agent(
        name="Sales Agent",
        instructions="You assist customers with their purchases.",
        model="gpt-5.6-terra",
        tools=[search_product, calculate_price],
    )
    agents_registry["support"] = Agent(
        name="Support Agent",
        instructions="You provide technical support.",
        model="gpt-5.6-terra",
        tools=[check_ticket, create_ticket],
    )
    yield
    agents_registry.clear()

app = FastAPI(lifespan=lifespan)

@app.post("/api/chat/{agent_type}")
async def chat(agent_type: str, request: Request):
    agent = agents_registry.get(agent_type)
    if not agent:
        raise HTTPException(404, f"Agent '{agent_type}' not found")

    body = await request.json()

    async def stream():
        result = Runner.run_streamed(agent, body["messages"])
        async for event in result.stream_events():
            if event.type == "raw_response_event":
                if hasattr(event.data, "delta") and event.data.delta:
                    yield f"data: {event.data.delta}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

Concurrency Management

Sharing Agent objects across all requests like this is safe because each request creates an independent Runner: the agent holds the configuration, the Runner holds the execution state. The SDK handles concurrency natively, and processing a batch of messages means stacking coroutines in asyncio.gather. The key detail here is return_exceptions=True: without it, a single failed request cancels the entire batch. With it, you get errors as values, match them to their original message, and process the ninety-nine valid results.

import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Concurrent Agent",
    instructions="You answer questions.",
    model="gpt-5.6-terra",
)

async def process_parallel_requests(messages: list[str]):
    """Process multiple requests in parallel."""
    tasks = [Runner.run(agent, msg) for msg in messages]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for msg, res in zip(messages, results):
        if isinstance(res, Exception):
            print(f"Error for '{msg}': {res}")
        else:
            print(f"'{msg}' -> {res.final_output[:50]}...")

    return results

Multi-Agent Patterns

Supervisor Pattern

Beyond a certain scale, a single agent with fifteen tools becomes unpredictable: its instructions bloat, its choices degrade. The supervisor pattern splits work among specialized agents—research, analysis, writing—and delegates routing to a dedicated agent via handoff. Each specialist keeps short instructions and a model suited to its task: analysis inherits a reasoning model, while the supervisor, which only picks a recipient, runs on a fast model since its decision hinges on a few words.

from agents import Agent, Runner, Handoff

search_agent = Agent(
    name="Search",
    instructions="You search for information on the web and in databases.",
    tools=[WebSearchTool(), search_crm],
    model="gpt-5.6-terra",
)

analysis_agent = Agent(
    name="Analysis",
    instructions="You analyze data and produce insights.",
    model="gpt-5.6-sol",
)

writing_agent = Agent(
    name="Writing",
    instructions="You write professional reports in English.",
    model="gpt-5.6-terra",
)

supervisor_agent = Agent(
    name="Supervisor",
    instructions="""You coordinate a team of specialized agents.
    For each request:
    1. Analyze what is needed
    2. Delegate to specialized agents via handoff
    3. The last agent produces the final response

    Available agents:
    - Search: to collect data
    - Analysis: to analyze and interpret
    - Writing: to produce the final deliverable""",
    handoffs=[
        Handoff(agent=search_agent, description="Collect data"),
        Handoff(agent=analysis_agent, description="Analyze data"),
        Handoff(agent=writing_agent, description="Write a report"),
    ],
    model="gpt-5.6-terra",  # Fast for routing
)

Pipeline Pattern

When the sequence is known in advance, letting a model decide the chain is an unnecessary expense. The pipeline pattern wires steps in your code: collect, analyze, write. A competitive analysis thus starts with two collections launched together—the web and CRM have no reason to wait for each other—then passes everything to analysis, whose output feeds writing. You get a reproducible workflow, traceable step by step, whose total duration is that of the slowest collection plus the two following steps.

async def competitive_analysis_pipeline(company: str):
    # Step 1: Parallel data collection
    web_task = Runner.run(search_agent, f"Search for information about {company}")
    crm_task = Runner.run(crm_agent, f"CRM data for {company}")
    results = await asyncio.gather(web_task, crm_task)

    data = f"""Web data: {results[0].final_output}
    CRM data: {results[1].final_output}"""

    # Step 2: Analysis
    analysis = await Runner.run(analysis_agent, f"Analyze this data: {data}")

    # Step 3: Report
    report = await Runner.run(
        writing_agent,
        f"Write a competitive analysis report: {analysis.final_output}"
    )

    return report.final_output

Map-Reduce Pattern

The third pattern addresses a volume problem rather than complexity. Two thousand customer verbatims don’t fit in a single request and can’t be analyzed sequentially in reasonable time. The MAP phase classifies each feedback in parallel with a fast model and typed output; the REDUCE phase aggregates these classifications, far more compact than the original texts, and entrusts the synthesis to a more capable model. Cost collapses compared to full analysis, because the bulk of the volume is processed by the cheapest model.

async def analyze_feedbacks(feedbacks: list[str]):
    # MAP: analyze each feedback in parallel
    classifier_agent = Agent(
        name="Classifier",
        instructions="Classify the feedback: positive, negative, or neutral. Identify the main theme.",
        model="gpt-5.6-terra",
        output_type=FeedbackClassification,
    )

    tasks = [Runner.run(classifier_agent, fb) for fb in feedbacks]
    classifications = await asyncio.gather(*tasks)

    # REDUCE: synthesize results
    summary_data = [
        {"feedback": fb, "class": c.final_output.dict()}
        for fb, c in zip(feedbacks, classifications)
    ]

    synthesizer_agent = Agent(
        name="Synthesizer",
        instructions="Synthesize the feedback classifications into a report with statistics and insights.",
        model="gpt-5.6-terra",
    )

    import json
    report = await Runner.run(
        synthesizer_agent,
        f"Synthesize these {len(feedbacks)} classified feedbacks: {json.dumps(summary_data, ensure_ascii=False)}"
    )

    return report.final_output

Rate Limit Management

Launching two thousand simultaneous calls doesn’t make them finish faster: the provider refuses you and you lose the batch. A semaphore bounds the number of concurrent executions, and the manager below adds exponential backoff capped at sixty seconds when a limit error occurs. The error counter resets as soon as a call succeeds, avoiding long-term penalty to the service for a transient incident.

import asyncio
from agents import Runner

class RateLimitManager:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.error_count = 0

    async def execute(self, agent, message, **kwargs):
        async with self.semaphore:
            try:
                result = await Runner.run(agent, message, **kwargs)
                self.error_count = 0
                return result
            except Exception as e:
                if "rate_limit" in str(e).lower():
                    self.error_count += 1
                    wait_time = min(2 ** self.error_count, 60)
                    await asyncio.sleep(wait_time)
                    return await self.execute(agent, message, **kwargs)
                raise

manager = RateLimitManager(max_concurrent=20)

Cost Optimization at Scale

Then there’s the bill. Two levers reduce it without degrading service. The first is not paying for a reasoning model to answer “what are your hours”: a lookup table maps complexity level to the suitable model, from the most economical for simple questions to the most capable for critical cases, with a default for anything unclassified. The second exploits repetition—in customer support, the same questions return hundreds of times a day, and a cache indexed on a message hash returns the answer instantly and free. Reserve this cache for questions whose answer depends neither on the user nor on the time: a return policy caches well, an order status never does.

# Model strategy by complexity
def choose_model(complexity: str) -> str:
    models = {
        "simple": "gpt-5.6-terra",      # Simple questions: cheap
        "medium": "gpt-5.6-terra",       # Most cases
        "complex": "gpt-5.6-terra",     # Fast reasoning
        "critical": "gpt-5.6-sol",      # Critical cases: maximum quality
    }
    return models.get(complexity, "gpt-5.6-terra")

# Cache for frequent responses
from functools import lru_cache
import hashlib

response_cache = {}

async def execute_with_cache(agent, message: str):
    key = hashlib.md5(message.encode()).hexdigest()
    if key in response_cache:
        return response_cache[key]

    result = await Runner.run(agent, message)
    response_cache[key] = result.final_output
    return result.final_output

Key Takeaways

Pre-configure your agents at startup and reuse them across requests
The SDK handles concurrency natively via asyncio.gather()
The supervisor pattern coordinates specialized agents via handoffs
The map-reduce pattern parallelizes large-volume processing
Manage rate limits with a semaphore and exponential backoff
Optimize costs with model selection by complexity and caching

Test Your Knowledge

Before scaling up, check your agent architecture knowledge.

1. What are the building blocks of the Agents SDK?

Answer: Agent (the role and its instructions), Runner (the execution loop), Tool (the capabilities), and Guardrail (input/output validations)—the anatomy of any SDK agent.

2. What are Guardrails for?

Answer: To validate what enters and leaves the agent: filter out-of-scope requests, control outputs before they act—security declared in the architecture, not hoped for from the prompt.

3. What is a handoff?

Answer: The transfer of a conversation from one agent to another, more specialized one: multi-agent orchestration where each agent does what it knows best, with context following.

4. Why is tracing indispensable in production?

Answer: Because an agent decides dynamically: without traces of steps, tools called, and decisions, debugging or evaluation is impossible—trace grading precisely scores these executions.

5. When should you move to a multi-agent architecture?

Answer: When a single agent accumulates too many roles and tools: you specialize (triage, expertise, execution) and orchestrate via handoffs—each agent becomes simple and testable again.

SDK, guardrails, handoffs, tracing: the scaffolding is complete—the scaling above tells you how to deploy it without weakening it.