Async Patterns for LLMs

import asyncio
from openai import AsyncOpenAI


async def async_completion(prompt: str) -> str:
    """Make an async completion request.
    
    Note: We use AsyncOpenAI (not OpenAI) because the synchronous
    client blocks the entire event loop during the HTTP call,
    preventing any other coroutines from making progress.
    """
    # Creating the client inside the function works for demos,
    # but in production, share one client across calls to reuse
    # the underlying HTTP connection pool.
    client = AsyncOpenAI()
    
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.choices[0].message.content


async def main():
    result = await async_completion("Explain async programming.")
    print(result)


# asyncio.run() creates an event loop, runs main(), then cleans up.
# Never call this inside an already-running loop (e.g., Jupyter notebooks
# use 'await main()' instead).
asyncio.run(main())

import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass


@dataclass
class CompletionResult:
    """Result from an async completion.
    
    Wrapping results in a dataclass lets us track success/failure
    per-item rather than having one failure blow up the entire batch.
    """
    prompt: str
    response: str
    success: bool
    error: str = None


async def process_prompt(
    client: AsyncOpenAI,
    prompt: str
) -> CompletionResult:
    """Process a single prompt.
    
    Notice: we catch exceptions and return them as data rather than
    letting them propagate. This is intentional -- with gather(), one
    unhandled exception would cancel or mask results from other tasks.
    """
    try:
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}]
        )
        
        return CompletionResult(
            prompt=prompt,
            response=response.choices[0].message.content,
            success=True
        )
    except Exception as e:
        return CompletionResult(
            prompt=prompt,
            response="",
            success=False,
            error=str(e)
        )


async def process_batch(prompts: list[str]) -> list[CompletionResult]:
    """Process multiple prompts in parallel.
    
    All prompts fire concurrently. For 4 prompts at 3 seconds each,
    total wall time is ~3 seconds (not 12).
    """
    # Share one client instance across all tasks -- this reuses the
    # underlying HTTP connection pool (httpx), which is much more
    # efficient than creating a new TLS connection per request.
    client = AsyncOpenAI()
    
    tasks = [process_prompt(client, prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks)
    
    return results


# Usage
prompts = [
    "Summarize machine learning in one sentence.",
    "What is deep learning?",
    "Explain neural networks briefly.",
    "Define natural language processing.",
]

results = asyncio.run(process_batch(prompts))

for result in results:
    if result.success:
        print(f"Q: {result.prompt[:30]}...")
        print(f"A: {result.response[:100]}...\n")
    else:
        print(f"Error: {result.error}")

import asyncio
import time
from dataclasses import dataclass, field


@dataclass
class TokenBucket:
    """Token bucket rate limiter.
    
    Think of this as a leaky bucket in reverse: tokens drip IN
    at a constant rate, and each request drains tokens OUT.
    """
    
    capacity: float       # Max tokens the bucket can hold (burst size)
    refill_rate: float    # Tokens added per second (sustained rate)
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)
    
    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()
    
    async def acquire(self, tokens: float = 1) -> float:
        """Acquire tokens, waiting if necessary. Returns wait time."""
        async with self._lock:
            await self._refill()
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return 0
            
            # Calculate wait time
            tokens_needed = tokens - self.tokens
            wait_time = tokens_needed / self.refill_rate
            
            await asyncio.sleep(wait_time)
            await self._refill()
            
            self.tokens -= tokens
            return wait_time
    
    async def _refill(self):
        """Refill tokens based on elapsed time."""
        now = time.monotonic()
        elapsed = now - self.last_refill
        
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.refill_rate
        )
        self.last_refill = now


class RateLimitedClient:
    """OpenAI client with rate limiting.
    
    LLM APIs enforce limits on TWO dimensions: request count and
    token count. You can hit the token limit with few large requests
    or the request limit with many small ones. This client guards
    against both.
    """
    
    def __init__(
        self,
        requests_per_minute: int = 60,
        tokens_per_minute: int = 90000
    ):
        from openai import AsyncOpenAI
        
        self.client = AsyncOpenAI()
        
        # Two separate rate limiters -- you must pass both gates
        # before a request is allowed through
        self.request_limiter = TokenBucket(
            capacity=requests_per_minute,
            refill_rate=requests_per_minute / 60  # Convert to per-second
        )
        self.token_limiter = TokenBucket(
            capacity=tokens_per_minute,
            refill_rate=tokens_per_minute / 60
        )
    
    async def complete(
        self,
        messages: list[dict],
        model: str = "gpt-4o-mini",
        estimated_tokens: int = 500
    ) -> str:
        """Make a rate-limited completion request."""
        # Acquire rate limit tokens
        await self.request_limiter.acquire(1)
        await self.token_limiter.acquire(estimated_tokens)
        
        response = await self.client.chat.completions.create(
            model=model,
            messages=messages
        )
        
        return response.choices[0].message.content


# Usage
async def main():
    client = RateLimitedClient(
        requests_per_minute=60,
        tokens_per_minute=90000
    )
    
    prompts = [f"Count to {i}" for i in range(1, 21)]
    
    async def process(prompt):
        return await client.complete([{"role": "user", "content": prompt}])
    
    results = await asyncio.gather(*[process(p) for p in prompts])
    print(f"Processed {len(results)} requests")


asyncio.run(main())

import asyncio
import time
from collections import deque
from dataclasses import dataclass, field


@dataclass
class SlidingWindowLimiter:
    """Sliding window rate limiter.
    
    Uses a deque of timestamps to track when requests were made.
    Old timestamps outside the window are pruned on each check.
    Memory usage is bounded by max_requests (one timestamp per request).
    """
    
    max_requests: int
    window_seconds: float
    timestamps: deque = field(default_factory=deque)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)
    
    async def acquire(self) -> float:
        """Acquire permission to make a request.
        
        Returns the time spent waiting (0 if no wait was needed).
        Useful for monitoring how much throttling is occurring.
        """
        async with self._lock:
            now = time.monotonic()
            
            # Evict timestamps that have fallen outside the window.
            # popleft() is O(1) on deque, so this stays fast.
            cutoff = now - self.window_seconds
            while self.timestamps and self.timestamps[0] < cutoff:
                self.timestamps.popleft()
            
            if len(self.timestamps) < self.max_requests:
                self.timestamps.append(now)
                return 0
            
            # Window is full -- wait until the oldest entry expires
            oldest = self.timestamps[0]
            wait_time = oldest + self.window_seconds - now
            
            if wait_time > 0:
                await asyncio.sleep(wait_time)
            
            # After waiting, re-clean (other requests may have completed)
            now = time.monotonic()
            cutoff = now - self.window_seconds
            while self.timestamps and self.timestamps[0] < cutoff:
                self.timestamps.popleft()
            
            self.timestamps.append(now)
            return wait_time


# Usage
limiter = SlidingWindowLimiter(max_requests=10, window_seconds=1.0)

async def rate_limited_request(i: int):
    wait = await limiter.acquire()
    if wait > 0:
        print(f"Request {i} waited {wait:.2f}s")
    print(f"Request {i} executing")
    return i

async def main():
    tasks = [rate_limited_request(i) for i in range(20)]
    await asyncio.gather(*tasks)

asyncio.run(main())

import asyncio
import random
from typing import TypeVar, Callable, Awaitable
from dataclasses import dataclass
from openai import RateLimitError, APIError, APIConnectionError

T = TypeVar("T")


@dataclass
class BackoffConfig:
    """Configuration for exponential backoff.
    
    The defaults here are reasonable for most LLM APIs:
    - Start with 1s delay (not too aggressive)
    - Cap at 60s (beyond this, the request is probably stale)
    - 5 retries gives ~30s of total retry window
    - Jitter prevents thundering herd on shared rate limits
    """
    
    initial_delay: float = 1.0
    max_delay: float = 60.0
    max_retries: int = 5
    exponential_base: float = 2.0
    jitter: bool = True  # Always use jitter in production


async def with_exponential_backoff(
    func: Callable[[], Awaitable[T]],
    config: BackoffConfig = None,
    retryable_exceptions: tuple = (RateLimitError, APIConnectionError)
) -> T:
    """Execute function with exponential backoff on failures."""
    config = config or BackoffConfig()
    
    last_exception = None
    
    for attempt in range(config.max_retries):
        try:
            return await func()
            
        except retryable_exceptions as e:
            last_exception = e
            
            if attempt == config.max_retries - 1:
                raise
            
            # Calculate delay
            delay = min(
                config.initial_delay * (config.exponential_base ** attempt),
                config.max_delay
            )
            
            # Add jitter
            if config.jitter:
                delay = delay * (0.5 + random.random())
            
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s")
            await asyncio.sleep(delay)
        
        except APIError as e:
            # Don't retry client errors
            if e.status_code and 400 <= e.status_code < 500:
                raise
            
            last_exception = e
            
            if attempt == config.max_retries - 1:
                raise
            
            delay = config.initial_delay * (config.exponential_base ** attempt)
            await asyncio.sleep(delay)
    
    raise last_exception


# Usage
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI()
    
    async def make_request():
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "Hello!"}]
        )
        return response.choices[0].message.content
    
    result = await with_exponential_backoff(
        make_request,
        BackoffConfig(initial_delay=1.0, max_retries=3)
    )
    print(result)

asyncio.run(main())

import asyncio
import time
from dataclasses import dataclass, field


@dataclass
class AdaptiveRateLimiter:
    """Rate limiter that adapts based on API responses.
    
    The asymmetry between increase and decrease is intentional:
    we increase slowly (1.1x) but decrease fast (0.5x). This is
    the AIMD pattern (Additive Increase, Multiplicative Decrease)
    borrowed from TCP congestion control -- it converges quickly
    to the actual limit without oscillating.
    """
    
    initial_rate: float  # Requests per second
    min_rate: float = 0.1
    max_rate: float = 100.0
    increase_factor: float = 1.1
    decrease_factor: float = 0.5
    
    current_rate: float = field(init=False)
    last_request: float = field(init=False)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)
    
    def __post_init__(self):
        self.current_rate = self.initial_rate
        self.last_request = 0
    
    async def acquire(self):
        """Wait for rate limit slot."""
        async with self._lock:
            now = time.monotonic()
            min_interval = 1.0 / self.current_rate
            elapsed = now - self.last_request
            
            if elapsed < min_interval:
                await asyncio.sleep(min_interval - elapsed)
            
            self.last_request = time.monotonic()
    
    def on_success(self):
        """Increase rate on successful request."""
        self.current_rate = min(
            self.current_rate * self.increase_factor,
            self.max_rate
        )
    
    def on_rate_limit(self):
        """Decrease rate on rate limit error."""
        self.current_rate = max(
            self.current_rate * self.decrease_factor,
            self.min_rate
        )
        print(f"Rate limited. Reducing to {self.current_rate:.2f} req/s")
    
    def on_error(self):
        """Slightly decrease rate on other errors."""
        self.current_rate = max(
            self.current_rate * 0.9,
            self.min_rate
        )


class AdaptiveClient:
    """OpenAI client with adaptive rate limiting."""
    
    def __init__(self, initial_rate: float = 10.0):
        from openai import AsyncOpenAI
        
        self.client = AsyncOpenAI()
        self.limiter = AdaptiveRateLimiter(initial_rate=initial_rate)
    
    async def complete(self, prompt: str) -> str:
        """Make a completion with adaptive rate limiting."""
        await self.limiter.acquire()
        
        try:
            response = await self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": prompt}]
            )
            
            self.limiter.on_success()
            return response.choices[0].message.content
            
        except RateLimitError:
            self.limiter.on_rate_limit()
            raise
            
        except Exception:
            self.limiter.on_error()
            raise

import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass


@dataclass
class ConcurrencyController:
    """Control concurrent request limits.
    
    Uses an asyncio.Semaphore as a gate: only max_concurrent
    coroutines can hold the semaphore at once. The rest queue up
    and wait their turn.
    """
    
    max_concurrent: int = 10
    
    def __post_init__(self):
        self.semaphore = asyncio.Semaphore(self.max_concurrent)
        self.active_count = 0  # For monitoring -- how many are in-flight?
    
    async def __aenter__(self):
        await self.semaphore.acquire()
        self.active_count += 1
        return self
    
    async def __aexit__(self, *args):
        self.active_count -= 1
        self.semaphore.release()


class ConcurrentBatchProcessor:
    """Process batches with controlled concurrency."""
    
    def __init__(self, max_concurrent: int = 10):
        self.client = AsyncOpenAI()
        self.controller = ConcurrencyController(max_concurrent)
    
    async def process_item(self, item: str) -> dict:
        """Process a single item with concurrency control."""
        async with self.controller:
            try:
                response = await self.client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=[{"role": "user", "content": item}]
                )
                
                return {
                    "input": item,
                    "output": response.choices[0].message.content,
                    "success": True
                }
            except Exception as e:
                return {
                    "input": item,
                    "output": None,
                    "success": False,
                    "error": str(e)
                }
    
    async def process_batch(
        self,
        items: list[str],
        progress_callback: callable = None
    ) -> list[dict]:
        """Process a batch of items."""
        results = []
        completed = 0
        
        async def process_with_progress(item):
            nonlocal completed
            result = await self.process_item(item)
            completed += 1
            
            if progress_callback:
                progress_callback(completed, len(items))
            
            return result
        
        tasks = [process_with_progress(item) for item in items]
        results = await asyncio.gather(*tasks)
        
        return results


# Usage
async def main():
    processor = ConcurrentBatchProcessor(max_concurrent=5)
    
    items = [f"Summarize the number {i}" for i in range(20)]
    
    def on_progress(done, total):
        print(f"Progress: {done}/{total}")
    
    results = await processor.process_batch(items, on_progress)
    
    successful = sum(1 for r in results if r["success"])
    print(f"Completed: {successful}/{len(items)} successful")

asyncio.run(main())

import asyncio
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable
from enum import IntEnum
import heapq


class Priority(IntEnum):
    """Lower number = higher priority (processed first).
    
    Map these to your business tiers:
    - HIGH: Real-time user-facing requests (chat, autocomplete)
    - NORMAL: Standard API requests
    - LOW: Background jobs (batch processing, analytics)
    """
    HIGH = 0
    NORMAL = 1
    LOW = 2


@dataclass(order=True)
class QueuedRequest:
    """A request in the priority queue."""
    priority: Priority
    sequence: int  # For FIFO ordering within same priority
    request: Any = field(compare=False)
    future: asyncio.Future = field(compare=False)


class PriorityRequestQueue:
    """Process LLM requests with priority ordering."""
    
    def __init__(
        self,
        processor: Callable[[Any], Awaitable[Any]],
        max_concurrent: int = 5
    ):
        self.processor = processor
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.queue: list[QueuedRequest] = []
        self.sequence = 0
        self._lock = asyncio.Lock()
        self._processing = False
    
    async def submit(
        self,
        request: Any,
        priority: Priority = Priority.NORMAL
    ) -> Any:
        """Submit a request and wait for result."""
        future = asyncio.Future()
        
        async with self._lock:
            queued = QueuedRequest(
                priority=priority,
                sequence=self.sequence,
                request=request,
                future=future
            )
            self.sequence += 1
            heapq.heappush(self.queue, queued)
        
        # Start processing if not already running
        if not self._processing:
            asyncio.create_task(self._process_queue())
        
        return await future
    
    async def _process_queue(self):
        """Process requests from the queue."""
        self._processing = True
        
        while True:
            async with self._lock:
                if not self.queue:
                    self._processing = False
                    return
                
                queued = heapq.heappop(self.queue)
            
            async with self.semaphore:
                try:
                    result = await self.processor(queued.request)
                    queued.future.set_result(result)
                except Exception as e:
                    queued.future.set_exception(e)


# Usage
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI()
    
    async def process_request(prompt: str) -> str:
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    
    queue = PriorityRequestQueue(process_request, max_concurrent=3)
    
    # Submit requests with different priorities
    tasks = [
        queue.submit("Low priority task 1", Priority.LOW),
        queue.submit("High priority task", Priority.HIGH),
        queue.submit("Normal task 1", Priority.NORMAL),
        queue.submit("Low priority task 2", Priority.LOW),
        queue.submit("Normal task 2", Priority.NORMAL),
    ]
    
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result[:50] + "...")

asyncio.run(main())

import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass, field
from typing import Optional
import time


@dataclass
class SessionStats:
    """Statistics for an async session.
    
    Track these metrics for every batch run. They tell you
    whether your rate limiting is too aggressive (low throughput),
    whether your prompts are too long (high token counts), and
    whether a provider is degraded (high latency, low success rate).
    """
    requests: int = 0
    successful: int = 0
    failed: int = 0
    total_tokens: int = 0
    total_latency: float = 0.0
    
    @property
    def avg_latency(self) -> float:
        return self.total_latency / self.requests if self.requests else 0
    
    @property
    def success_rate(self) -> float:
        return self.successful / self.requests if self.requests else 0


class AsyncLLMSession:
    """Managed async session for LLM operations."""
    
    def __init__(
        self,
        max_concurrent: int = 10,
        timeout: float = 30.0
    ):
        self.client: Optional[AsyncOpenAI] = None
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.timeout = timeout
        self.stats = SessionStats()
    
    async def __aenter__(self):
        self.client = AsyncOpenAI()
        self.stats = SessionStats()
        return self
    
    async def __aexit__(self, *args):
        self.client = None
    
    async def complete(
        self,
        prompt: str,
        model: str = "gpt-4o-mini"
    ) -> str:
        """Make a completion within the session."""
        if not self.client:
            raise RuntimeError("Session not initialized")
        
        async with self.semaphore:
            self.stats.requests += 1
            start = time.monotonic()
            
            try:
                response = await asyncio.wait_for(
                    self.client.chat.completions.create(
                        model=model,
                        messages=[{"role": "user", "content": prompt}]
                    ),
                    timeout=self.timeout
                )
                
                self.stats.successful += 1
                self.stats.total_tokens += response.usage.total_tokens
                self.stats.total_latency += time.monotonic() - start
                
                return response.choices[0].message.content
                
            except Exception as e:
                self.stats.failed += 1
                self.stats.total_latency += time.monotonic() - start
                raise
    
    async def batch_complete(
        self,
        prompts: list[str],
        model: str = "gpt-4o-mini"
    ) -> list[str]:
        """Complete multiple prompts."""
        tasks = [self.complete(p, model) for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)


# Usage
async def main():
    async with AsyncLLMSession(max_concurrent=5) as session:
        prompts = [f"What is {i} + {i}?" for i in range(10)]
        
        results = await session.batch_complete(prompts)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Error {i}: {result}")
            else:
                print(f"Result {i}: {result[:30]}...")
        
        print(f"\nStats:")
        print(f"  Requests: {session.stats.requests}")
        print(f"  Success rate: {session.stats.success_rate:.1%}")
        print(f"  Avg latency: {session.stats.avg_latency:.2f}s")
        print(f"  Total tokens: {session.stats.total_tokens}")

asyncio.run(main())