Async Programming for High-Performance AI

In traditional software, wait times are measured in milliseconds. In LLM Engineering, wait times are measured in seconds. A single call to GPT-4o can take 5-10 seconds. If your agent needs to perform 5 reasoning steps, that’s nearly a minute of waiting.

If you write "Synchronous" (one-at-a-time) code, your application will feel broken. To build professional AI, you must master Asynchronous Programming (asyncio).

---

1. The Problem: Synchronous Blocking

In a synchronous script, the CPU sits idle while waiting for an API response.

# SYNCHRONOUS (Bad for AI)
def slow_ai_call():
    import time
    time.sleep(5) # Simulating an LLM call
    return "Done"

print("Starting Step 1")
slow_ai_call()
print("Starting Step 2") # This waits 5 seconds to start!
slow_ai_call()

If you have 100 users, and each one blocks the server for 5 seconds, your app will crash instantly.

---

2. The Solution: asyncio and await

Async allows your program to "pause" a task and work on something else while waiting for an external response.

Key Syntax:

• async def: Defines a function that is "awaitable."
• await: Tells the program: "Pause here, let someone else run, and come back when this API is done."
• asyncio.gather(): Runs multiple tasks in parallel.

sequenceDiagram
    participant App as Python App
    participant AI1 as LLM Call 1
    participant AI2 as LLM Call 2
    
    Note over App: Synchronous
    App->>AI1: Request
    Note over AI1: Waiting 5s...
    AI1-->>App: Response
    App->>AI2: Request
    Note over AI2: Waiting 5s...
    AI2-->>App: Response
    Note over App: Total Time: 10s
    
    Note over App: Asynchronous
    App->>AI1: Request
    App->>AI2: Request
    Note over AI1,AI2: Waiting 5s Simultaneously
    AI1-->>App: Response
    AI2-->>App: Response
    Note over App: Total Time: 5s

---

3. Parallel Retrieval: The Secret to Fast RAG

Imagine your agent needs to search a vector database, a Google Search API, and a local PDF.

• Sync: Search DB (2s) $\rightarrow$ Search Google (2s) $\rightarrow$ Read PDF (1s) = 5 seconds.
• Async: Search all 3 at once = 2 seconds.

Code Example: Parallel Tasks with asyncio.gather

import asyncio
import time

async def fetch_from_vector_db():
    await asyncio.sleep(2)
    return "Vector data"

async def fetch_from_google():
    await asyncio.sleep(2)
    return "Search results"

async def main():
    start = time.time()
    
    # Run both simultaneously
    results = await asyncio.gather(
        fetch_from_vector_db(),
        fetch_from_google()
    )
    
    end = time.time()
    print(f"Results: {results}")
    print(f"Total time: {end - start:.2f} seconds")

# Run the event loop
if __name__ == "__main__":
    asyncio.run(main())

---

4. Performance Considerations: Async in Web Servers

When building APIs (the primary delivery method for LLMs), we use FastAPI. FastAPI is built on asyncio.

Every endpoint in your AI service should be an async def. This ensures that while one user is waiting for Claude to generate a poem, another user can simultaneously ask for a summary of a document.

Performance Tip: Streaming Logs

When an agent is thinking, the user shouldn't see a spinner for 30 seconds. You should use Async Generators to stream logs to the frontend.

async def agent_step_generator():
    steps = ["Searching...", "Analyzing...", "Drafting..."]
    for step in steps:
        await asyncio.sleep(1) # Simulated work
        yield step

@app.get("/agent-stream")
async def stream():
    return StreamingResponse(agent_step_generator())

---

Summary

• Avoid blocking code: Never use time.sleep() in an AI app. Use await asyncio.sleep().
• Parallelize your I/O: Use asyncio.gather for multiple RAG sources.
• Async Frameworks: Stick to FastAPI and libraries that support httpx or aiohttp (async HTTP clients).
• Responsiveness: Use streaming to keep users engaged during long LLM generations.

In the next lesson, we will look at Data Manipulation, focusing on how to clean and prepare massive amounts of text for the models we are calling.

---

Exercise: Identify the Bottleneck

You are building an AI researcher that needs to:

1. Fetch 5 web pages (1.0s each).
2. Summarize each page (3.0s each).
3. Combine summaries into a final report (2.0s).

Calculate the total time needed for:

• A purely synchronous approach.
• An approach where step #1 and #2 are parallelized using asyncio.gather.

Answer Logic:

• Sync: $(5 \times 1.0) + (5 \times 3.0) + 2.0 = 22.0$ seconds.
• Async: The 5 fetches happen at once (1.0s), the 5 summaries happen at once (3.0s), plus 2.0s for the final part = $1.0 + 3.0 + 2.0 = 6.0$ seconds.

That is a 70% reduction in latency just by changing the architecture!