Chapter 1 · Python Async Programming and FastAPI Production

Chapter goals:

• Master asyncio core concepts: event loop, Task, gather, async with, async for
• Understand aiohttp async HTTP client for LLM API calls
• Implement production-grade rate limiting, retry (tenacity), and circuit breaker patterns
• Distinguish the appropriate boundaries between BackgroundTasks and message queues

1.1 asyncio Core Concepts

Python's asyncio is a single-threaded concurrency model that schedules coroutines through an event loop. Understanding the following concepts is foundational for building high-concurrency Agent services:

import asyncio

async def fetch_data(name: str, delay: float) -> str:
    """Simulate async data retrieval"""
    await asyncio.sleep(delay)
    return f"{name}: {delay}s"

async def main():
    # Create multiple tasks for concurrent execution
    tasks = [fetch_data("A", 0.5), fetch_data("B", 0.3), fetch_data("C", 0.8)]
    
    # gather runs all tasks concurrently, returns result list
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r)
    
    # Single task creation and waiting
    task = asyncio.create_task(fetch_data("D", 1.0))
    result = await task
    print(result)

asyncio.run(main())

Key APIs:

• asyncio.run(coro): Run top-level coroutine
• asyncio.create_task(coro): Schedule coroutine for concurrent execution
• asyncio.gather(*coros): Run multiple coroutines concurrently, collect results
• asyncio.wait_for(coro, timeout): Wait with timeout
• async with / async for: Async context management and iteration

1.2 aiohttp: Async HTTP Client

Agent services need to call LLM APIs asynchronously (e.g., Vercel AI Gateway); aiohttp is the preferred choice:

import aiohttp
import asyncio

async def call_llm_async(prompt: str) -> str:
    """Asynchronously call LLM (using httpbin as example instead of actual API)"""
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://httpbin.org/post",
            json={"prompt": prompt},
            timeout=aiohttp.ClientTimeout(total=30)
        ) as resp:
            data = await resp.json()
            return data.get("json", {}).get("prompt", "")

# Concurrently call multiple LLMs
async def main():
    prompts = ["Hello", "Explain quantum computing", "Write a poem"]
    tasks = [call_llm_async(p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for r in results:
        print(r if isinstance(r, str) else f"Error: {r}")

asyncio.run(main())

1.3 Production-Grade Rate Limiting: Token Bucket Implementation

import asyncio
import time
from typing import Dict

class RateLimiter:
    """Token Bucket rate limiter"""
    
    def __init__(self, rate: float, burst: int):
        self.rate = rate  # Allowed requests per second
        self.burst = burst  # Maximum burst capacity
        self._tokens = burst
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()
    
    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(self.burst, self._tokens + elapsed * self.rate)
            self._last_refill = now
            
            if self._tokens < 1:
                wait_time = (1 - self._tokens) / self.rate
                await asyncio.sleep(wait_time)
                self._tokens = 0
            else:
                self._tokens -= 1

async def limited_llm_call(prompt: str, limiter: RateLimiter) -> str:
    await limiter.acquire()  # Wait for token
    return f"response to: {prompt}"

async def main():
    limiter = RateLimiter(rate=5.0, burst=10)  # 5 req/s, max burst 10
    tasks = [limited_llm_call(f"prompt-{i}", limiter) for i in range(20)]
    results = await asyncio.gather(*tasks)
    print(f"Completed {len(results)} requests")

asyncio.run(main())

1.4 Retry Strategy: tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import asyncio

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((asyncio.TimeoutError, ConnectionError)),
    before_sleep=lambda retry_state: print(f"Retrying attempt {retry_state.attempt_number}...")
)
async def call_llm_with_retry(prompt: str, timeout: float = 5.0) -> str:
    """LLM call with exponential backoff retry"""
    try:
        return await asyncio.wait_for(_actual_call(prompt), timeout=timeout)
    except asyncio.TimeoutError:
        raise  # Timeout triggers retry

async def _actual_call(prompt: str) -> str:
    await asyncio.sleep(0.1)
    return f"result for: {prompt}"

async def main():
    try:
        result = await call_llm_with_retry("hello")
        print(result)
    except Exception as e:
        print(f"Final failure: {e}")

asyncio.run(main())

1.5 Circuit Breaker Pattern

import asyncio
import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal
    OPEN = "open"          # Tripped
    HALF_OPEN = "half_open" # Half-open

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self._state = CircuitState.CLOSED
        self._failures = 0
        self._last_failure_time = 0.0
        self._lock = asyncio.Lock()
    
    @property
    def state(self) -> CircuitState:
        return self._state
    
    async def call(self, func, *args, **kwargs):
        async with self._lock:
            if self._state == CircuitState.OPEN:
                if time.monotonic() - self._last_failure_time > self.reset_timeout:
                    self._state = CircuitState.HALF_OPEN
                    print("Circuit breaker: half-open state, allowing test request")
                else:
                    raise RuntimeError("Circuit breaker: open, service unavailable")
        
        try:
            result = await func(*args, **kwargs)
            async with self._lock:
                if self._state == CircuitState.HALF_OPEN:
                    self._state = CircuitState.CLOSED
                    self._failures = 0
                    print("Circuit breaker: restored to closed state")
            return result
        except Exception as e:
            async with self._lock:
                self._failures += 1
                self._last_failure_time = time.monotonic()
                if self._failures >= self.failure_threshold:
                    self._state = CircuitState.OPEN
                    print(f"Circuit breaker: reached {self.failure_threshold} failures, tripping open")
            raise

async def main():
    cb = CircuitBreaker(failure_threshold=3, reset_timeout=2.0)
    
    async def flaky_service():
        raise ConnectionError("Service unavailable")
    
    for i in range(5):
        try:
            await cb.call(flaky_service)
        except Exception as e:
            print(f"Request {i+1} failed: {e}")
        await asyncio.sleep(0.1)

asyncio.run(main())

1.6 BackgroundTasks vs Message Queue Boundaries

Scenario	Solution	Reason
Lightweight async tasks (sending notification emails)	FastAPI BackgroundTasks	In-process, simple and fast
Long-running LLM calls	Celery Worker + Redis Queue	Process isolation, retryable, persistent
High-concurrency Agent tasks	SQS + Fargate Worker	Horizontal scaling, distributed coordination

💡 See FastAPI ch16 for complete explanation of BackgroundTasks.

Chapter Summary

• asyncio is a single-threaded concurrency model; key APIs include gather, create_task, sleep
• aiohttp is used for async HTTP calls, suitable for concurrent LLM API requests
• Production-grade rate limiting uses Token Bucket, retry uses tenacity (exponential backoff), circuit breaker uses three-state machine
• Distinguish usage boundaries: BackgroundTasks (lightweight) vs message queues (heavy tasks)

🛠️ Hands-on Practice

1. Implement an async LLM call service using aiohttp + custom RateLimiter, limiting to max 3 requests per second
2. Add tenacity retry to the above service (max 3 attempts, exponential backoff 1-8 seconds)
3. Implement CircuitBreaker that opens after 3 consecutive failures, auto-transitions to half-open after 30 seconds

🧪 随堂测验

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

1. What is the main purpose of asyncio.gather()?

2. In the Token Bucket rate limiter, what does the burst parameter represent?

3. How does tenacity's wait_exponential implement backoff?

4. What scenario is the CircuitBreaker HALF_OPEN state used for?