LLM Integration Patterns

Production patterns for LLM API integration. Streaming, RAG, model routing, and cost optimization.

When to Apply

• Working with OpenRouter API calls
• Building or modifying chat streaming
• Implementing RAG with Qdrant
• Optimizing model selection and costs
• Managing context windows

1. Streaming Response (CRITICAL)

import aiohttp
import json

async def stream_llm_response(
    messages: list[dict],
    model: str,
    on_chunk: callable,
    temperature: float = 0.7,
    max_tokens: int = 4096
):
    """Stream LLM response via OpenRouter."""
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://openrouter.ai/api/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {OPENROUTER_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": model,
                "messages": messages,
                "temperature": temperature,
                "max_tokens": max_tokens,
                "stream": True
            }
        ) as response:
            full_text = ""
            async for line in response.content:
                line = line.decode().strip()
                if not line.startswith("data: "):
                    continue
                data = line[6:]
                if data == "[DONE]":
                    break
                try:
                    chunk = json.loads(data)
                    content = chunk["choices"][0]["delta"].get("content", "")
                    if content:
                        full_text += content
                        await on_chunk(content)
                except (json.JSONDecodeError, KeyError):
                    continue
            return full_text

2. Model Routing (CRITICAL)

MODEL_ROUTING = {
    # Task -> Model mapping
    "orchestration": "meta-llama/llama-3.3-70b-instruct:free",  # Fast classification
    "smalltalk": "meta-llama/llama-3.3-70b-instruct:free",      # Quick responses
    "price_query": "meta-llama/llama-3.3-70b-instruct:free",    # Simple lookup
    "technical_analysis": "qwen/qwen3-30b-a3b:free",            # Analytical
    "best_setups": "qwen/qwen3-30b-a3b:free",                   # Complex reasoning
    "deep_research": "nousresearch/hermes-3-llama-3.1-405b:free", # Maximum quality
}

def select_model(task_type: str) -> str:
    """Select optimal model based on task complexity."""
    return MODEL_ROUTING.get(task_type, MODEL_ROUTING["smalltalk"])

Fallback chain

FALLBACK_MODELS = [
    "meta-llama/llama-3.3-70b-instruct:free",
    "qwen/qwen3-30b-a3b:free",
    "openai/gpt-oss-20b:free",
]

async def call_with_fallback(messages: list[dict], **kwargs) -> str:
    for model in FALLBACK_MODELS:
        try:
            return await call_llm(model=model, messages=messages, **kwargs)
        except Exception as e:
            logger.warning(f"Model {model} failed: {e}")
            continue
    raise Exception("All models failed")

3. RAG Pipeline (CRITICAL)

from qdrant_client import QdrantClient

async def retrieve_and_generate(query: str, collection: str = "telegram_news_v2") -> str:
    # 1. Generate query embedding
    query_embedding = await generate_embedding(query)
    
    # 2. Search vector DB
    results = qdrant_client.search(
        collection_name=collection,
        query_vector=query_embedding,
        limit=5,
        score_threshold=0.7,
        with_payload=["title", "text", "timestamp", "source"]
    )
    
    # 3. Build context from results
    context_parts = []
    for i, result in enumerate(results, 1):
        payload = result.payload
        context_parts.append(
            f"[{i}] {payload.get('title', 'No title')}\n"
            f"Source: {payload.get('source', 'unknown')}\n"
            f"Date: {payload.get('timestamp', 'unknown')}\n"
            f"{payload.get('text', '')[:500]}"
        )
    context = "\n\n".join(context_parts)
    
    # 4. Generate answer with context
    messages = [
        {"role": "system", "content": (
            "Answer based on the provided context. "
            "Cite sources with [n] notation. "
            "If context doesn't contain the answer, say so."
        )},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"}
    ]
    
    return await call_llm(messages=messages, model=select_model("deep_research"))

4. Context Window Management (HIGH)

import tiktoken

def estimate_tokens(text: str, model: str = "gpt-4") -> int:
    """Estimate token count."""
    try:
        enc = tiktoken.encoding_for_model(model)
        return len(enc.encode(text))
    except Exception:
        return len(text) // 4  # Rough estimate

def trim_context(context: str, max_tokens: int = 6000) -> str:
    """Trim context to fit token budget."""
    tokens = estimate_tokens(context)
    if tokens <= max_tokens:
        return context
    
    # Keep most recent/relevant content
    ratio = max_tokens / tokens
    target_chars = int(len(context) * ratio * 0.9)
    return context[-target_chars:]

# Context priority for market analysis
CONTEXT_PRIORITY = [
    "current_prices",      # Always include
    "technical_signals",   # High priority
    "market_sentiment",    # Medium
    "recent_news",         # Medium
    "economic_calendar",   # Lower
    "historical_patterns", # Lowest
]

5. Retry with Exponential Backoff (HIGH)

import asyncio

async def call_llm_with_retry(
    messages: list[dict],
    model: str,
    max_retries: int = 3,
    **kwargs
) -> str:
    for attempt in range(max_retries):
        try:
            return await call_llm(model=model, messages=messages, **kwargs)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait = min(2 ** attempt, 30)
            logger.warning(f"LLM call failed (attempt {attempt+1}): {e}, retrying in {wait}s")
            await asyncio.sleep(wait)

6. Response Validation (HIGH)

import re

def validate_llm_response(response: str, expected_format: str = "text") -> bool:
    """Validate LLM output before using."""
    if not response or len(response.strip()) < 10:
        return False
    
    if expected_format == "json":
        try:
            json.loads(response)
            return True
        except json.JSONDecodeError:
            return False
    
    # Check for hallucination markers
    hallucination_patterns = [
        r"as an ai",
        r"i don't have access",
        r"i cannot browse",
        r"my training data",
    ]
    for pattern in hallucination_patterns:
        if re.search(pattern, response.lower()):
            logger.warning(f"Potential hallucination detected: {pattern}")
    
    return True

7. Anti-Patterns

• Sending entire documents when only relevant chunks needed
• No retry logic for API calls
• Ignoring token usage (unexpected costs)
• Using expensive models for simple classification
• Not validating LLM output before downstream use
• Building RAG without evaluating retrieval quality
• Not caching embeddings for repeated queries

8. Checklist

• API calls wrapped with retry + exponential backoff
• Streaming used for user-facing responses
• Model selection based on task complexity
• RAG chunks sized 500-1000 tokens with overlap
• Token usage tracked and monitored
• LLM output validated before use
• Embeddings cached to avoid redundant API calls
• Fallback chain for model failures