Day 2: Advanced Python & Libraries 

[E] Implementation

Component Architecture:

• Async Core: asyncio.create_task() manages concurrent operations

• Decorator Layer: Wraps handlers with retry, rate limit, cache logic

• Pydantic Models: Define strict schemas for AgentRequest, AgentResponse, SystemMetrics

• FastAPI Server: Async routes leverage Python's event loop

• React Dashboard: WebSocket connection streams real-time metrics

Control Flow: Client request → FastAPI async endpoint → decorator chain (rate check → cache lookup → retry wrapper) → async business logic → Pydantic validation → response

Data Flow: JSON payload → Pydantic parse → dataclass transformation → async processing → Gemini API call → result aggregation → Pydantic serialize → JSON response

[F] Coding Highlights

Async Pattern (agents process requests without blocking):

async def process_agent_request(request: AgentRequest) -> AgentResponse:
    async with aiohttp.ClientSession() as session:
        tasks = [call_gemini(session, prompt) for prompt in request.prompts]
        results = await asyncio.gather(*tasks)
    return AgentResponse(results=results, latency_ms=...)

Production Decorator (automatic retries with exponential backoff):

def with_retry(max_attempts=3, backoff=2.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1: raise
                    await asyncio.sleep(backoff ** attempt)
        return wrapper
    return decorator

Pydantic Validation (catches invalid data before processing):

class AgentRequest(BaseModel):
    agent_id: str = Field(..., regex=r'^agent-[a-z0-9]{8}$')
    prompts: List[str] = Field(..., min_items=1, max_items=100)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(1024, ge=1, le=8192)

Production Considerations: Connection pooling limits concurrent API calls; semaphores prevent resource exhaustion; circuit breakers stop cascading failures; structured logging tracks request IDs through async operations.

[G] Validation

Verification Methods:

1. Load test with 1000 concurrent requests: python test.sh load_test

2. Monitor async task completion in dashboard at http://localhost:3000

3. Verify Pydantic rejects invalid payloads: python test.sh validation

4. Check decorator behavior under failures: python test.sh resilience

Success Criteria:

• Throughput: 500+ requests/second on single core

• Latency: P95 under 200ms for Gemini calls

• Error Rate: <0.1% after retry logic

• Validation: 100% of malformed requests rejected

Benchmarks: Compare synchronous vs async performance. Async should achieve 10x+ throughput improvement for I/O-bound workloads.

[H] Assignment

Challenge: Extend the async pipeline with batching logic. Accumulate requests for 100ms, then process in single Gemini API call using batch prompting. Implement a @batch_requests(window_ms=100) decorator.

Requirements:

• Maintain async semantics (non-blocking accumulation)

• Use Pydantic to validate batch schemas

• Add batch metrics to dashboard (batch size, processing time)

Deliverable: Modified server.py with batching decorator, updated Pydantic models for batch operations.

This builds toward L3 by introducing batching concepts—transformers process sequences in batches for efficiency, a pattern you'll implement in the next lesson.

[I] Solution Hints

• Use asyncio.Queue to accumulate requests during the window

• Create a background task that flushes the queue periodically

• Batch multiple prompts into a single Gemini request using newline delimiters

• Return results to individual requesters using asyncio.Future objects

• Update BatchMetrics(BaseModel) with average batch size, flush frequency

[J] Looking Ahead

L3 dives into transformer architecture—the engine behind Gemini's intelligence. Today's async patterns become critical: transformer inference is I/O-bound (GPU memory transfers), so async wrappers maximize throughput. The batching assignment prepares you for transformer sequence processing, where batching is essential for efficient matrix operations.

Your Pydantic models will evolve to define transformer configurations (numlayers, hiddendim, attention_heads). The decorator patterns extend to model checkpointing, gradient accumulation, and distributed training hooks.

Module Progress: 2/10 lessons complete in "Foundational VAIA Concepts". Next: deep dive into the architecture powering every modern LLM.