LLM App Patterns

LLM Application Patterns

Production patterns for building reliable, observable, and cost-effective LLM applications.

Workflow

When this skill activates:

1. Identify the task — RAG pipeline, agent architecture, prompt engineering, caching, observability, or evaluation.
2. Use the Architecture Decision Matrix to choose the right pattern before writing code.
3. Apply the relevant section — each section is self-contained.
4. For complete implementations of multi-agent systems, evaluation frameworks, streaming, and context window management, see references/advanced.md.
5. Flag prompt injection risks proactively — user input passed to LLMs without sanitization is a security issue in every LLM app.

Architecture Decision Matrix

Choose the right pattern before building:

Default starting point: Simple RAG or Function Calling. Add complexity only when simpler patterns demonstrably fail.

1. RAG Pipeline

Documents → Chunk → Embed → Store
Query ──────────────────── Retrieve → Rerank → Generate → Response + Citations

1.1 Chunking

CHUNK_CONFIG = {
    "chunk_size": 512,       # tokens — tune per document type
    "chunk_overlap": 50,     # preserve cross-chunk context
    "separators": ["\n\n", "\n", ". ", " "],
}

# Strategy selection:
# FIXED_SIZE     — simple; may break mid-sentence
# SEMANTIC       — split on paragraphs/sections; better quality
# RECURSIVE      — tries separators in order; good default
# DOCUMENT_AWARE — respects headers, lists, code blocks; best for structured docs

1.2 Vector Database & Embedding Selection

For embedding model options and dimension/cost tradeoffs, see Embedding Model Reference in references/advanced.md.

EMBEDDING_MODELS = {
    "openai/text-embedding-3-small": {
        "dimensions": 1536,
        "cost": "$0.02/1M tokens",
        "quality": "Good for most use cases",
    },
    "openai/text-embedding-3-large": {
        "dimensions": 3072,
        "cost": "$0.13/1M tokens",
        "quality": "Best for complex queries",
    },
    "local/bge-large-en-v1.5": {
        "dimensions": 1024,
        "cost": "Free (compute only)",
        "quality": "Comparable to OpenAI small",
    },
}

1.3 Retrieval Strategies

# Basic semantic search
def semantic_search(query: str, top_k: int = 5) -> list[Document]:
    return vector_db.similarity_search(embed(query), top_k=top_k)

# Hybrid search — semantic + keyword (BM25) via Reciprocal Rank Fusion
def hybrid_search(query: str, top_k: int = 5, alpha: float = 0.7) -> list[Document]:
    """alpha=1.0 pure semantic, alpha=0.0 pure BM25, 0.7 good default."""
    semantic = vector_db.similarity_search(embed(query), top_k=top_k * 2)
    keyword  = bm25_index.search(query, top_k=top_k * 2)
    return rrf_merge(semantic, keyword, alpha=alpha)[:top_k]

# Multi-query: generate variations to improve recall
def multi_query_retrieval(query: str) -> list[Document]:
    variations = llm.generate_variations(query, n=3)
    all_docs = [doc for q in variations for doc in semantic_search(q)]
    return deduplicate_by_id(all_docs)

1.4 Generation with Citations

RAG_PROMPT = """\
Answer the question using ONLY the context below.
If the context is insufficient, say "I don't have enough information."

Context:
{context}

Question: {question}
Answer:"""

def generate_with_rag(question: str) -> dict:
    docs = hybrid_search(question, top_k=5)

    # Respect context window — truncate if needed
    context = build_context(docs, max_tokens=4000)

    response = llm.generate(RAG_PROMPT.format(context=context, question=question))
    return {"answer": response, "sources": [d.metadata for d in docs]}

def build_context(docs: list[Document], max_tokens: int) -> str:
    """Build context string that fits within token budget."""
    parts, total = [], 0
    for doc in docs:
        tokens = count_tokens(doc.content)
        if total + tokens > max_tokens:
            break
        parts.append(doc.content)
        total += tokens
    return "\n\n---\n\n".join(parts)

2. Agent Architectures

2.1 Function Calling (Recommended Default)

Structured tool dispatch — the model returns typed tool calls, your code executes them:

TOOLS = [
    {
        "name": "search_web",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
]

class FunctionCallingAgent:
    MAX_ITERATIONS = 10   # always guard against infinite loops

    def run(self, question: str) -> str:
        messages = [{"role": "user", "content": question}]

        for _ in range(self.MAX_ITERATIONS):
            response = self.llm.chat(messages=messages, tools=TOOLS, tool_choice="auto")

            # No tool calls — final answer
            if not response.tool_calls:
                return response.content

            # CRITICAL: append the assistant message BEFORE tool results.
            # response.content may be None when tool calls are present — use [] as fallback.
            # Omitting this step causes a 400 error on the next API call.
            messages.append({
                "role": "assistant",
                "content": response.content or [],
            })

            for tool_call in response.tool_calls:
                result = self._execute_tool(tool_call.name, tool_call.arguments)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result),
                })

        raise RuntimeError(f"Agent exceeded {self.MAX_ITERATIONS} iterations")

2.2 ReAct Pattern

For models or workflows where native function calling isn't available:

REACT_PROMPT = """\
You are an assistant that uses tools to answer questions.

Available tools:
{tools_description}

Format:
Thought: [reasoning about what to do]
Action: tool_name(arguments)
Observation: [result — filled in by system]
... (repeat as needed)
Thought: I have enough information.
Final Answer: [response]

Question: {question}
"""

class ReActAgent:
    MAX_ITERATIONS = 10

    def run(self, question: str) -> str:
        messages = [
            {"role": "user", "content": REACT_PROMPT.format(
                tools_description=self._format_tools(),
                question=question,
            )}
        ]

        for _ in range(self.MAX_ITERATIONS):
            response = self.llm.chat(messages=messages)
            messages.append({"role": "assistant", "content": response.content})

            if "Final Answer:" in response.content:
                return self._extract_final_answer(response.content)

            action = self._parse_action(response.content)
            if action:
                observation = self._execute_tool(action)
                messages.append({"role": "user", "content": f"Observation: {observation}"})

        raise RuntimeError(f"Agent exceeded {self.MAX_ITERATIONS} iterations")

2.3 Plan-and-Execute

For complex long-horizon tasks with a predictable step structure:

class PlanAndExecuteAgent:
    def run(self, task: str) -> str:
        plan = self.planner.create_plan(task)   # ["Step 1: ...", "Step 2: ..."]
        results = []

        for i, step in enumerate(plan):
            result = self.executor.execute(step, context=results)
            results.append(result)

            # Replan if intermediate result changes scope
            if self._needs_replan(task, results):
                plan = self.planner.replan(
                    task, completed=results, remaining=plan[i+1:]
                )

        return self.synthesizer.summarize(task, results)

2.4 Multi-Agent Collaboration

For research or highly parallelizable tasks, consider orchestrating multiple specialized agents. See references/advanced.md for the full fault-tolerant implementation with per-agent error handling, partial-result synthesis, and revision loops.

class AgentTeam:
    """Coordination skeleton — see advanced.md for production version."""

    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "analyst":    AnalystAgent(),
            "writer":     WriterAgent(),
            "critic":     CriticAgent(),
        }
        self.coordinator = CoordinatorAgent()

    def solve(self, task: str) -> str:
        assignments = self.coordinator.decompose(task)
        results = {}

        for assignment in assignments:
            agent = self.agents.get(assignment.agent)   # .get() — safe on unknown agent
            if agent is None:
                continue
            result = agent.execute(assignment.subtask, context=results)
            results[assignment.id] = result

        critique = self.agents["critic"].review(results)
        if critique.needs_revision:
            return self._revise_with_feedback(task, results, critique)

        return self.coordinator.synthesize(results)

3. Prompt Engineering

3.1 Prompt Templates

from string import Formatter

class PromptTemplate:
    def __init__(self, template: str):
        self.template = template
        # Infer variables from template — no manual list needed
        self.variables = {
            fname for _, fname, _, _ in Formatter().parse(template) if fname
        }

    def format(self, **kwargs) -> str:
        missing = self.variables - set(kwargs)
        if missing:
            raise ValueError(f"Missing template variables: {missing}")
        return self.template.format(**kwargs)

    def with_examples(self, examples: list[dict]) -> "PromptTemplate":
        """Prepend few-shot examples to template."""
        shots = "\n\n".join(
            f"Input: {e['input']}\nOutput: {e['output']}" for e in examples
        )
        return PromptTemplate(f"{shots}\n\n{self.template}")

3.2 Prompt Versioning and A/B Testing

import hashlib
from datetime import datetime, timezone

class PromptRegistry:
    def register(self, name: str, template: str, version: str) -> None:
        self.db.save({
            "name": name,
            "template": template,
            "version": version,
            # datetime.utcnow() is deprecated in 3.12+ — always use timezone-aware form
            "created_at": datetime.now(timezone.utc).isoformat(),
            "metrics": {},
        })

    def get(self, name: str, version: str = "latest") -> str:
        return self.db.get(name, version)

    def ab_test(self, name: str, user_id: str) -> str:
        """Deterministic bucketing — stable across process restarts.
        Never use Python's hash() — it's randomized per process."""
        variants = self.db.get_all_versions(name)
        # md5 is fine here — not used for security, just stable bucketing
        bucket = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % len(variants)
        return variants[bucket]

    def record_outcome(self, prompt_id: str, outcome: dict) -> None:
        self.db.update_metrics(prompt_id, outcome)

3.3 Prompt Chaining

class PromptChain:
    def __init__(self, steps: list[dict]):
        """
        steps: [{"name": str, "prompt": str, "output_key": str,
                 "parser": Callable | None}]
        """
        self.steps = steps

    def run(self, initial_input: str) -> dict:
        context: dict = {"input": initial_input}
        results = []

        for step in self.steps:
            prompt = step["prompt"].format(**context)
            output = llm.generate(prompt)

            # Apply output parser if provided
            if step.get("parser"):
                output = step["parser"](output)

            context[step["output_key"]] = output
            results.append({"step": step["name"], "output": output})

        return {
            "final_output": context[self.steps[-1]["output_key"]],
            "intermediate_results": results,
        }

4. Production Patterns

4.1 Caching

Cache deterministic outputs (temperature=0). Include model and all parameters in the key:

import hashlib, json
from dataclasses import dataclass

@dataclass
class CacheStats:
    hits: int = 0
    misses: int = 0

class LLMCache:
    def __init__(self, redis_client, ttl_seconds: int = 3600):
        self.redis = redis_client
        self.ttl = ttl_seconds
        self.stats = CacheStats()

    def _cache_key(self, prompt: str, model: str, **kwargs) -> str:
        content = json.dumps({"model": model, "prompt": prompt, **kwargs}, sort_keys=True)
        return f"llm:{hashlib.sha256(content.encode()).hexdigest()}"

    def get_or_generate(self, prompt: str, model: str, **kwargs) -> str:
        # Only cache deterministic outputs
        if kwargs.get("temperature", 1.0) != 0:
            return llm.generate(prompt, model=model, **kwargs)

        key = self._cache_key(prompt, model, **kwargs)
        cached = self.redis.get(key)
        if cached:
            self.stats.hits += 1
            return cached.decode()

        self.stats.misses += 1
        response = llm.generate(prompt, model=model, **kwargs)
        self.redis.setex(key, self.ttl, response)
        return response

4.2 Rate Limiting and Retry

import threading, time
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception

class RateLimiter:
    """Thread-safe sliding-window rate limiter."""

    def __init__(self, requests_per_minute: int):
        self.rpm = requests_per_minute
        self.timestamps: list[float] = []
        self._lock = threading.Lock()

    def acquire(self) -> None:
        with self._lock:   # thread-safe — prevents race conditions
            now = time.time()
            self.timestamps = [t for t in self.timestamps if now - t < 60]
            if len(self.timestamps) >= self.rpm:
                sleep_for = 60 - (now - self.timestamps[0])
                time.sleep(max(0, sleep_for))
            self.timestamps.append(time.time())

def _is_retryable(exc: Exception) -> bool:
    """Only retry transient errors — not auth failures or bad requests."""
    if isinstance(exc, RateLimitError):
        return True
    if isinstance(exc, APIError):
        return getattr(exc, "status_code", 0) >= 500  # 5xx only
    return False

@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60),
    stop=stop_after_attempt(5),
    retry=retry_if_exception(_is_retryable),  # don't retry 400/401/403
)
def call_llm_with_retry(prompt: str) -> str:
    return llm.generate(prompt)

4.3 Model Fallback

class LLMWithFallback:
    def __init__(self, primary: str, fallbacks: list[str]):
        self.models = [primary] + fallbacks

    def generate(self, prompt: str, **kwargs) -> str:
        last_error: Exception | None = None
        for model in self.models:
            try:
                return llm.generate(prompt, model=model, **kwargs)
            except (RateLimitError, APIError) as e:
                if isinstance(e, APIError) and getattr(e, "status_code", 0) < 500:
                    raise   # 4xx errors won't be fixed by switching models
                logger.warning("Model %s failed: %s", model, e)
                last_error = e
        raise RuntimeError("All models failed") from last_error

5. LLMOps and Observability

5.1 Metrics

Track these at minimum:

LLM_METRICS = {
    # Performance
    "latency_p50_ms":       "50th percentile end-to-end latency",
    "latency_p99_ms":       "99th percentile — catches tail latency",
    "time_to_first_token":  "Streaming UX metric",

    # Quality
    "user_satisfaction":    "Thumbs up / thumbs down ratio",
    "task_completion_rate": "% tasks reaching a final answer",
    "hallucination_rate":   "% responses flagged by evaluator",

    # Cost
    "cost_usd_per_request": "Average $ per call",
    "tokens_per_request":   "Average input + output tokens",
    "cache_hit_rate":       "% served from cache — directly reduces cost",

    # Reliability
    "error_rate":           "% failed requests",
    "retry_rate":           "% requests needing at least one retry",
    "fallback_rate":        "% requests that hit a fallback model",
}

5.2 Structured Logging and Tracing

import json, logging
from datetime import datetime, timezone
from opentelemetry import trace

logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)

class LLMLogger:
    def log_request(self, request_id: str, model: str, prompt: str,
                    prompt_tokens: int, user_id: str | None = None) -> None:
        logger.info(json.dumps({
            "event": "llm.request",
            "request_id": request_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "model": model,
            "prompt_preview": prompt[:200],   # never log full prompt in prod
            "prompt_tokens": prompt_tokens,
            "user_id": user_id,
        }))

    def log_response(self, request_id: str, completion_tokens: int,
                     total_tokens: int, latency_ms: float,
                     finish_reason: str, cost_usd: float) -> None:
        logger.info(json.dumps({
            "event": "llm.response",
            "request_id": request_id,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens,
            "latency_ms": latency_ms,
            "finish_reason": finish_reason,
            "cost_usd": cost_usd,
        }))

@tracer.start_as_current_span("llm_call")
def call_llm_traced(prompt: str) -> str:
    span = trace.get_current_span()
    span.set_attribute("llm.prompt.tokens", count_tokens(prompt))
    response = llm.generate(prompt)
    span.set_attribute("llm.response.tokens", response.usage.total_tokens)
    span.set_attribute("llm.finish_reason", response.finish_reason)
    return response.content

5.3 Evaluation Framework (Skeleton)

Implement the scoring methods below, or load the full LLM-as-judge implementation from references/advanced.md:

class LLMEvaluator:
    """Evaluate LLM outputs for quality."""

    def evaluate_response(self,
                          question: str,
                          response: str,
                          context: str | None = None,
                          ground_truth: str | None = None) -> dict:
        scores = {}

        # Relevance: Does it answer the question?
        scores["relevance"] = self._score_relevance(question, response)

        # Coherence: Is it well-structured?
        scores["coherence"] = self._score_coherence(response)

        # Groundedness: Is it based on provided context (if any)?
        if context:
            scores["groundedness"] = self._score_groundedness(response, context)

        # Accuracy: Compare with ground truth
        if ground_truth:
            scores["accuracy"] = self._score_accuracy(response, ground_truth)

        # Safety: Check for harmful content
        scores["safety"] = self._score_safety(response)

        scores["overall"] = sum(scores.values()) / len(scores)
        return scores

    def run_benchmark(self, test_cases: list[dict]) -> dict:
        """Run evaluation on a test set."""
        results = []
        for case in test_cases:
            response = llm.generate(case["prompt"])
            scores = self.evaluate_response(
                question=case["prompt"],
                response=response,
                context=case.get("context"),
                ground_truth=case.get("expected"),
            )
            results.append(scores)
        return self._aggregate_scores(results)

    # Implement with custom logic or LLM-as-a-judge (see references/advanced.md)
    def _score_relevance(self, question: str, response: str) -> float: ...
    def _score_coherence(self, response: str) -> float: ...
    def _score_groundedness(self, response: str, context: str) -> float: ...
    def _score_accuracy(self, response: str, ground_truth: str) -> float: ...
    def _score_safety(self, response: str) -> float: ...
    def _aggregate_scores(self, results: list[dict]) -> dict: ...

6. Security — Prompt Injection

Every LLM application that passes user input to a model is vulnerable to prompt injection. Treat it as seriously as SQL injection.

# ❌ DANGEROUS: user content interpolated into system prompt
system = f"You are a helpful assistant. The user is {user_name}. Always be helpful."

# ✅ SAFE: user input isolated in its own message turn
messages = [
    {"role": "system", "content": "You are a helpful assistant. Be concise."},
    {"role": "user", "content": user_input},   # never in system prompt
]

# ✅ Validate and bound LLM outputs before acting on them
from pydantic import BaseModel

class AgentAction(BaseModel):
    tool: str
    arguments: dict

def parse_tool_call(raw: str) -> AgentAction:
    """Validate LLM output before executing — never eval() or exec() raw output."""
    data = json.loads(raw)
    action = AgentAction.model_validate(data)
    if action.tool not in ALLOWED_TOOLS:
        raise ValueError(f"Unknown tool: {action.tool}")
    return action

Rules:

• User input always goes in user role messages — never interpolated into system prompt
• Validate and schema-check LLM outputs before acting on them
• Never eval() or exec() LLM-generated code without sandboxing
• Scope tool access to the minimum needed for the task
• Log all prompts and responses for audit

For the full Prompt Injection Quick Reference table, see references/advanced.md.

For streaming responses, context window management, the full fault-tolerant multi-agent system, and the LLM-as-judge evaluation framework with implementations, see references/advanced.md.