Llm Api Scaffold

Generate complete, production-ready LLM API integration code with error handling, rate limiting, caching, monitoring, and best practices built-in.

What You'll Get

When you run this command, you'll receive:

1. Complete API client with retry logic and error handling
2. Rate limiting (token bucket algorithm)
3. Caching layer (in-memory + Redis)
4. Cost tracking and monitoring
5. Multi-provider support with fallback
6. Streaming responses implementation
7. Testing suite with mocks
8. Docker deployment configuration

Usage

/llm-api-scaffold <provider> [options]

Providers: anthropic, openai, multi (both)

Examples:

• /las anthropic - Generate Anthropic Claude integration
• /las openai - Generate OpenAI GPT integration
• /las multi - Generate multi-provider with fallback

Generated Output

Example: Anthropic Claude Integration

Input:

/las anthropic

Output:

1. Project Structure

llm-api-integration/
├── src/
│   ├── client/
│   │   ├── __init__.py
│   │   ├── base.py            # Base client interface
│   │   ├── anthropic_client.py # Anthropic implementation
│   │   └── rate_limiter.py     # Rate limiting
│   ├── cache/
│   │   ├── __init__.py
│   │   ├── memory_cache.py     # In-memory cache
│   │   └── redis_cache.py      # Redis cache
│   ├── monitoring/
│   │   ├── __init__.py
│   │   ├── metrics.py          # Prometheus metrics
│   │   └── cost_tracker.py     # Cost tracking
│   └── utils/
│       ├── __init__.py
│       └── retry.py            # Retry logic
…

2. Base Client (src/client/base.py)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, AsyncGenerator

@dataclass
class CompletionRequest:
    """Standardized completion request."""
    prompt: str
    max_tokens: int = 1024
    temperature: float = 1.0
    model: Optional[str] = None
    stream: bool = False

@dataclass
class CompletionResponse:
    """Standardized completion response."""
    content: str
    usage: dict
…

3. Anthropic Client (src/client/anthropic_client.py)

import time
import asyncio
from anthropic import AsyncAnthropic, RateLimitError, APIError
from .base import BaseLLMClient, CompletionRequest, CompletionResponse
from .rate_limiter import TokenBucket
from ..cache import CacheManager
from ..monitoring import MetricsCollector, CostTracker
from ..utils.retry import retry_with_backoff

class AnthropicClient(BaseLLMClient):
    """Production-ready Anthropic Claude client."""

    def __init__(
        self,
        api_key: str,
        model: str = "claude-3-haiku-20240307",
        requests_per_minute: int = 50,
        enable_cache: bool = True,
…

4. Rate Limiter (src/client/rate_limiter.py)

import time
import asyncio
from threading import Lock

class TokenBucket:
    """Thread-safe token bucket for rate limiting."""

    def __init__(self, capacity: int, refill_rate: float):
        """
        Args:
            capacity: Maximum tokens (e.g., 50 requests)
            refill_rate: Tokens per second (e.g., 50/60 = 0.833 req/s)
        """
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate
        self.last_refill = time.time()
        self.lock = Lock()
…

5. Cache Manager (src/cache/memory_cache.py)

import time
from typing import Optional, Any
from collections import OrderedDict

class MemoryCache:
    """In-memory LRU cache with TTL."""

    def __init__(self, max_size: int = 1000):
        self.max_size = max_size
        self.cache = OrderedDict()
        self.expiry = {}

    async def get(self, key: str) -> Optional[Any]:
        """Get cached value if not expired."""
        if key not in self.cache:
            return None

        # Check expiry
…

6. Cost Tracker (src/monitoring/cost_tracker.py)

import time
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class CostTracker:
    """Track LLM usage costs."""

    usage_history: list = field(default_factory=list)
    costs_by_model: dict = field(default_factory=lambda: defaultdict(float))

    PRICING = {
        "claude-3-opus-20240229": {"input": 0.015, "output": 0.075},
        "claude-3-sonnet-20240229": {"input": 0.003, "output": 0.015},
        "claude-3-haiku-20240307": {"input": 0.00025, "output": 0.00125},
        "gpt-4-turbo-preview": {"input": 0.01, "output": 0.03},
        "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015}
    }
…

7. Retry Utility (src/utils/retry.py)

import time
import random
import asyncio
from functools import wraps

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True
):
    """Async retry decorator with exponential backoff."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
…