Chapter 25 · Performance Optimization and Production Practice

Learning objectives: From model selection and concurrent throughput to cost control, evaluation regression, and containerized deployment — master the complete checklist for pushing an Agno application to production.

25.1 Model Tier Routing: Let the Right Model Do the Right Job

The biggest production cost comes from "using a large model to do a small model's work." Tier by task difficulty:

import os
from agno.agent import Agent
from agno.models.openai import OpenAIChat

def make_model(model_id: str) -> OpenAIChat:
    """Unified third-party model factory: all tiers go through the same OpenAI-compatible endpoint"""
    return OpenAIChat(
        id=model_id,
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    )

# Classification / extraction / formatting: small model — fast and cheap
classifier = Agent(
    name="Classifier",
    model=make_model("deepseek-chat"),
    output_schema=Category,           # Structured output is naturally suited to small models
    instructions=["Classify the user request as question / complaint / other"],
)

# Complex reasoning / creation: large or reasoning model
analyst = Agent(
    name="Analyst",
    model=make_model("deepseek-reasoner"),
    instructions=["Perform root-cause analysis on the user complaint and provide handling recommendations"],
)

Paired with the Router/Condition from Chapter 12, you get a funnel structure: "small model classifies first, large model picks up only the hard cases." Within a Team you can also equip only the leader with a large model while members inherit or explicitly use smaller models.

25.2 Concurrent Throughput: asyncio.gather + arun

Agent-level concurrency pairs arun() with standard asyncio:

import asyncio

async def analyze(symbol: str) -> str:
    result = await analyst.arun(f"Analyze the recent one-week trend risk for {symbol}")
    return f"{symbol}: {result.content}"

async def main():
    symbols = ["AAPL", "NVDA", "TSLA", "AMZN"]
    # Run 4 analyses concurrently; total time ≈ the slowest one, not the sum
    results = await asyncio.gather(*[analyze(s) for s in symbols])
    print("\n".join(results))

asyncio.run(main())

At the Workflow level, use Parallel steps for automatic concurrency (see Chapter 12). Note: the concurrency cap must match the provider's rate limit — use asyncio.Semaphore to control the number of in-flight requests.

25.3 Four Levers for Cost Control

1. Tiered routing (see 25.1) — often cuts cost by 50%+;
2. Control context size: with add_history_to_context=True, use num_history_runs to limit how many turns are carried, preventing the conversation from getting more expensive the longer it runs;
3. Cap tool loops: set max_tool_calls on Agents, max_iterations on Team Tasks mode, to prevent a single run from spiraling out of control;
4. Use metrics as a budget gate: after each run check metrics.total_tokens; if over budget, degrade to a smaller model or circuit-break outright.

result = agent.run("Long-running task...")
if result.metrics and result.metrics.total_tokens > 50_000:
    # Log and alert: single run token count exceeded budget
    print("WARN: tokens over budget", result.metrics.total_tokens)

25.4 Evals: Turning Quality into Regression-Testable Assertions

After changing prompts, swapping models, or tweaking tools, how do you know you didn't break the product? Agno's official Evals turn expected behaviors into repeatable checks you can run in development and CI:

from agno.agent import Agent
from agno.eval.accuracy import AccuracyEval

evaluation = AccuracyEval(
    name="calculator-accuracy",
    model=make_model("deepseek-chat"),
    agent=calc_agent,
    input="What is 10 factorial?",
    expected_output="3628800",
)

result = evaluation.run(print_results=True)
assert result is not None and result.avg_score >= 8

Four evaluation types to pick from as needed:

Question	Eval type
Does the response match the expected answer?	Accuracy
Does it meet a custom quality standard?	Agent as Judge
Did it call the expected tools with the right arguments?	Reliability
How does it perform on latency and memory?	Performance

Combine multiple cases into an Eval Suite; once hooked into CI, every commit runs an agent regression automatically.

25.5 Dockerizing AgentOS Deployment

AgentOS is just a FastAPI application — containerization is straightforward:

FROM python:3.12-slim AS base
WORKDIR /app

# Install deps first to leverage layer caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 7777
# No reload in production; worker count scaled to CPU cores
CMD ["uvicorn", "agent_os:app", "--host", "0.0.0.0", "--port", "7777", "--workers", "2"]

# Key snippet of docker-compose.yml
services:
  agentos:
    build: .
    environment:
      - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}   # Secrets via env vars, never baked into the image
    volumes:
      - agent-data:/app/tmp                     # Persist SqliteDb files
  db:
    image: postgres:16                          # Production recommends PostgresDb over SQLite
volumes:
  agent-data:

Production notes:

• Secrets management: API Keys injected via environment variables or a secrets manager at runtime; never hard-coded in code or baked into images;
• Database upgrade: SQLite single-file is fine for development; for production switch to PostgresDb (sessions, memories, and traces share or separate DBs as you prefer);
• Enable auth and tracing: authorization=True + tracing=True (Chapters 21–23);
• Health checks: configure liveness probes against the /info endpoint.

25.6 Pre-Launch Checklist

Check off items one by one before going live:

• [ ] All Agents / Teams / Workflows have explicit id set
• [ ] Third-party model keys go via env vars; no plaintext keys in the repo
• [ ] AgentOS has authorization=True enabled; public-facing entry has HTTPS and rate limiting
• [ ] Sessions / memories use a production-grade DB (e.g., PostgresDb) with a backup strategy
• [ ] tracing=True with a dedicated tracing DB; retention policy defined
• [ ] Core flows have an Eval Suite hooked into CI for regression
• [ ] Every execution path has runaway protection: max_iterations / token budgets
• [ ] Monitoring dashboards cover cost (tokens), latency (P95), and quality (tool failure rate)
• [ ] Validate with a small traffic rollout before full exposure

25.7 Chapter Summary

• Model tier routing is the highest-ROI optimization: small models for classification/extraction, large/reasoning models for creation;
• Agent concurrency via arun() + asyncio.gather (with Semaphore for rate limiting); Workflow concurrency via Parallel;
• Cost control = tiering + capped history rounds + iteration limits + metrics budget gates;
• Evals (Accuracy/Judge/Reliability/Performance) turn quality into CI-runnable assertions;
• Containerized AgentOS deployment is identical to FastAPI deployment: env-var secrets, SQLite → Postgres, auth + tracing enabled.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 「Let a small model classify the request first, then decide whether to hand it to a large model」— the best implementation combo is?

2. To verify tool-call behavior has not regressed after switching prompts, which eval type should you use?

3. Regarding API Key management in production, what is the correct approach?

4. Using SQLite for sessions in development — what does the official recommendation say for production?

🛠️ Hands-On Exercises

1. Wire AccuracyEval into the Chapter 11 writing pipeline: prepare 5 topics with expected highlights, run the eval, and hook it into GitHub Actions.
2. Use asyncio.gather + Semaphore(3) to analyze 10 topics concurrently; compare total elapsed time against serial execution (record via metrics.duration).
3. Go through the chapter's checklist item by item against the projects from earlier chapters; compile any non-compliant items into a remediation task list.

🎉 Production practice complete! Next come three capstone chapters: starting with Chapter 26: Capstone 1 — Intelligent Data Analysis Assistant, tying every capability learned here into a single rope.