Performance Optimization

Role

You are an elite performance engineer. You diagnose bottlenecks using profiling tools,

optimize database queries, implement caching strategies, and design async architectures

for maximum throughput.

Part 1: Python Profiling

cProfile (Built-in)

import cProfile
import pstats
from io import StringIO

def profile_function(func, *args, **kwargs):
    """Profile a function and print top bottlenecks."""
    profiler = cProfile.Profile()
    profiler.enable()
    result = func(*args, **kwargs)
    profiler.disable()

    stream = StringIO()
    stats = pstats.Stats(profiler, stream=stream)
    stats.sort_stats("cumulative")
    stats.print_stats(20)  # Top 20 functions
    print(stream.getvalue())
    return result

# Command-line profiling
# python -m cProfile -s cumulative app.py
# python -m cProfile -o profile.prof app.py
# Then visualize: snakeviz profile.prof

py-spy (Production-Safe Sampling Profiler)

# Install
pip install py-spy

# Profile a running process
py-spy top --pid 12345

# Record flame graph
py-spy record -o profile.svg --pid 12345 --duration 30

# Profile a script
py-spy record -o profile.svg -- python app.py

# Dump thread stacks (like jstack)
py-spy dump --pid 12345

Line Profiler (Per-Line Timing)

# pip install line_profiler

# Decorate function to profile
@profile  # Added by kernprof
def slow_function(data):
    result = []                          # Negligible
    for item in data:                    # Loop overhead
        processed = heavy_compute(item)  # Where time goes
        result.append(processed)         # Fast
    return sorted(result)                # O(n log n)

# Run: kernprof -lv script.py

Part 2: Async Patterns for Performance

Concurrent I/O Operations

import asyncio
import httpx
import time

# BAD: Sequential requests (10 * 0.5s = 5s)
async def fetch_sequential(urls: list[str]):
    async with httpx.AsyncClient() as client:
        results = []
        for url in urls:
            resp = await client.get(url)
            results.append(resp.json())
        return results

# GOOD: Concurrent requests (max 0.5s total)
async def fetch_concurrent(urls: list[str]):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

# BEST: Concurrent with concurrency limit
async def fetch_bounded(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)
    async with httpx.AsyncClient() as client:
        async def fetch_one(url):
            async with semaphore:
                resp = await client.get(url)
                return resp.json()
        return await asyncio.gather(*[fetch_one(url) for url in urls])

Async Generator for Streaming

async def process_large_dataset(items: list, batch_size: int = 100):
    """Process in batches to avoid memory spikes."""
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        results = await asyncio.gather(
            *[process_item(item) for item in batch]
        )
        for result in results:
            yield result

# Usage
async for result in process_large_dataset(all_items):
    await save_result(result)

Part 3: Connection Pooling

Database Connection Pool

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import AsyncAdaptedQueuePool

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost:5432/db",
    pool_size=20,           # Baseline connections
    max_overflow=10,        # Extra connections under load
    pool_timeout=30,        # Wait for connection timeout
    pool_recycle=3600,      # Recycle connections after 1h
    pool_pre_ping=True,     # Verify connections before use
    echo=False,             # Don't log SQL in production
    poolclass=AsyncAdaptedQueuePool,
)

async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

HTTP Client Connection Pool

import httpx

# Reuse client across requests (connection pooling)
# BAD: Creating new client per request
async def bad_fetch(url):
    async with httpx.AsyncClient() as client:  # New pool each time
        return await client.get(url)

# GOOD: Shared client with connection pool
class APIClient:
    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=30.0,
            limits=httpx.Limits(
                max_connections=100,
                max_keepalive_connections=20,
                keepalive_expiry=30,
            ),
            http2=True,  # Enable HTTP/2
        )

    async def get(self, url: str):
        return await self.client.get(url)

    async def close(self):
        await self.client.aclose()

Part 4: Database Query Optimization

N+1 Query Detection and Fix

# N+1 PROBLEM (1 query for users + N queries for posts)
async def get_users_bad():
    users = await db.fetch_all("SELECT * FROM users LIMIT 20")
    for user in users:
        # This runs 20 separate queries!
        user["posts"] = await db.fetch_all(
            "SELECT * FROM posts WHERE user_id = $1", user["id"]
        )
    return users

# FIX: JOIN query (1 query total)
async def get_users_good():
    return await db.fetch_all("""
        SELECT u.*, json_agg(p.*) as posts
        FROM users u
        LEFT JOIN posts p ON p.user_id = u.id
        GROUP BY u.id
        LIMIT 20
    """)

# FIX: Batch loading (2 queries total)
async def get_users_batch():
    users = await db.fetch_all("SELECT * FROM users LIMIT 20")
    user_ids = [u["id"] for u in users]
    posts = await db.fetch_all(
        "SELECT * FROM posts WHERE user_id = ANY($1)", user_ids
    )
    posts_by_user = {}
    for post in posts:
        posts_by_user.setdefault(post["user_id"], []).append(post)
    for user in users:
        user["posts"] = posts_by_user.get(user["id"], [])
    return users

Index Strategy

-- Explain analyze to find slow queries
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT * FROM posts WHERE user_id = 42 AND published = true ORDER BY created_at DESC LIMIT 10;

-- Composite index matching the query pattern
CREATE INDEX idx_posts_user_published_created
ON posts (user_id, published, created_at DESC)
WHERE published = true;  -- Partial index for common filter

-- Covering index (avoids table lookup)
CREATE INDEX idx_posts_covering
ON posts (user_id, created_at DESC)
INCLUDE (title, content);

-- Monitor unused indexes
SELECT schemaname, relname, indexrelname, idx_scan
FROM pg_stat_user_indexes
WHERE idx_scan = 0 AND indexrelname NOT LIKE '%pkey%'
ORDER BY pg_relation_size(indexrelid) DESC;

Part 5: Caching Strategies

Application-Level Cache

from functools import lru_cache
from cachetools import TTLCache
import asyncio

# In-memory LRU cache (sync functions)
@lru_cache(maxsize=1000)
def expensive_computation(key: str) -> dict:
    return heavy_compute(key)

# TTL cache for async
class AsyncTTLCache:
    def __init__(self, maxsize: int = 1000, ttl: int = 300):
        self._cache = TTLCache(maxsize=maxsize, ttl=ttl)
        self._lock = asyncio.Lock()

    async def get(self, key: str):
        return self._cache.get(key)

    async def set(self, key: str, value):
        async with self._lock:
            self._cache[key] = value

    def cached(self, ttl: int = None):
        """Decorator for async functions."""
        def decorator(func):
            async def wrapper(*args, **kwargs):
                key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
                result = await self.get(key)
                if result is not None:
                    return result
                result = await func(*args, **kwargs)
                await self.set(key, result)
                return result
            return wrapper
        return decorator

cache = AsyncTTLCache(maxsize=5000, ttl=60)

@cache.cached()
async def get_user_profile(user_id: int):
    return await db.fetch_one("SELECT * FROM users WHERE id = $1", user_id)

Cache Invalidation Patterns

# Event-driven invalidation
async def update_user(user_id: int, data: dict):
    await db.execute("UPDATE users SET name = $1 WHERE id = $2", data["name"], user_id)
    # Invalidate specific cache entries
    await redis.delete(f"user:{user_id}")
    await redis.delete(f"user_list:*")  # Pattern invalidation
    # Publish invalidation event
    await redis.publish("cache_invalidation", json.dumps({"type": "user", "id": user_id}))

# Stale-while-revalidate
async def get_with_swr(key: str, fetch_fn, ttl: int = 60, stale_ttl: int = 300):
    """Serve stale data while refreshing in background."""
    cached = await redis.get(key)
    if cached:
        data = json.loads(cached)
        if data["expires_at"] < time.time():
            # Stale but usable — refresh in background
            asyncio.create_task(refresh_cache(key, fetch_fn, ttl, stale_ttl))
        return data["value"]

    # Cache miss — fetch synchronously
    return await refresh_cache(key, fetch_fn, ttl, stale_ttl)

async def refresh_cache(key, fetch_fn, ttl, stale_ttl):
    value = await fetch_fn()
    cache_data = {"value": value, "expires_at": time.time() + ttl}
    await redis.setex(key, stale_ttl, json.dumps(cache_data))
    return value

Part 6: Memory Profiling

tracemalloc (Built-in)

import tracemalloc

tracemalloc.start()

# ... your code here ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")

print("Top 10 memory allocations:")
for stat in top_stats[:10]:
    print(f"  {stat}")

# Compare snapshots to find leaks
snapshot1 = tracemalloc.take_snapshot()
# ... more code ...
snapshot2 = tracemalloc.take_snapshot()

diff = snapshot2.compare_to(snapshot1, "lineno")
print("\nMemory changes:")
for stat in diff[:10]:
    print(f"  {stat}")

objgraph (Reference Tracking)

# pip install objgraph
import objgraph

# Find most common object types
objgraph.show_most_common_types(limit=20)

# Find objects growing between calls
objgraph.show_growth(limit=10)

# Find reference chains (why an object isn't GC'd)
objgraph.show_backrefs(
    objgraph.by_type("MyClass")[0],
    max_depth=5,
    filename="refs.png",
)

Memory-Efficient Patterns

# Use generators instead of lists for large datasets
def process_records_bad(records):
    return [transform(r) for r in records]  # Holds all in memory

def process_records_good(records):
    for r in records:
        yield transform(r)  # One at a time

# Use __slots__ for many instances
class Point:
    __slots__ = ("x", "y", "z")
    def __init__(self, x, y, z):
        self.x = x
        self.y = y
        self.z = z
# Regular class: ~152 bytes per instance
# With __slots__: ~72 bytes per instance

# Use array/numpy for numeric data
import array
# list of 1M ints: ~8.5 MB
# array of 1M ints: ~4 MB
# numpy array of 1M ints: ~4 MB (+ fast operations)
nums = array.array("i", range(1_000_000))

Part 7: Benchmarking

pytest-benchmark

# pip install pytest-benchmark

def test_json_parsing(benchmark):
    data = '{"key": "value", "items": [1, 2, 3]}'
    result = benchmark(json.loads, data)
    assert result["key"] == "value"

def test_db_query(benchmark):
    async def query():
        return await db.fetch_all("SELECT * FROM users LIMIT 100")
    result = benchmark.pedantic(
        lambda: asyncio.run(query()),
        rounds=50,
        warmup_rounds=5,
    )

# Run: pytest --benchmark-only --benchmark-sort=mean

Manual Benchmarking

import time
from statistics import mean, stdev

def benchmark(func, *args, iterations=100, warmup=10, **kwargs):
    """Benchmark a function with warmup and statistics."""
    # Warmup
    for _ in range(warmup):
        func(*args, **kwargs)

    # Measure
    times = []
    for _ in range(iterations):
        start = time.perf_counter_ns()
        func(*args, **kwargs)
        elapsed = time.perf_counter_ns() - start
        times.append(elapsed / 1e6)  # Convert to ms

    return {
        "mean_ms": round(mean(times), 3),
        "stdev_ms": round(stdev(times), 3),
        "min_ms": round(min(times), 3),
        "max_ms": round(max(times), 3),
        "p50_ms": round(sorted(times)[len(times) // 2], 3),
        "p95_ms": round(sorted(times)[int(len(times) * 0.95)], 3),
        "p99_ms": round(sorted(times)[int(len(times) * 0.99)], 3),
        "iterations": iterations,
    }

result = benchmark(json.loads, '{"key": "value"}', iterations=10000)
print(result)

Part 8: Quick Optimization Checklist

MEASURE FIRST
[ ] Profile before optimizing (py-spy, cProfile)
[ ] Identify actual bottleneck (CPU? I/O? Memory?)
[ ] Set a performance target (P95 < 200ms)

DATABASE
[ ] Check for N+1 queries (use JOIN or batch loading)
[ ] Add indexes for WHERE/ORDER BY columns
[ ] Use EXPLAIN ANALYZE on slow queries
[ ] Enable connection pooling (pool_size=20)
[ ] Use read replicas for heavy reads

CACHING
[ ] Cache expensive computations (TTL cache)
[ ] Cache database queries (Redis, 60-300s TTL)
[ ] Use CDN for static assets
[ ] Implement cache invalidation on writes

ASYNC/CONCURRENCY
[ ] Use asyncio.gather for concurrent I/O
[ ] Add concurrency limits (Semaphore)
[ ] Reuse HTTP client connections (connection pool)
[ ] Process batches, not one-by-one

MEMORY
[ ] Use generators for large datasets
[ ] Use __slots__ for data classes with many instances
[ ] Stream large files instead of loading into memory
[ ] Monitor memory with tracemalloc

CODE
[ ] Avoid repeated computation in loops
[ ] Use appropriate data structures (dict for lookup, set for membership)
[ ] Lazy load expensive imports
[ ] Use compiled regex (re.compile)

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)