Contextual Compression and Dynamic Prompt Assembly: How to Programmatically Build Prompts at Runtime from Structured Data Sources, User Inputs, and Business Rules

Introduction

Picture this: your team has built a customer-facing AI assistant for a financial services firm. It needs to answer questions about account balances, flag compliance concerns based on jurisdiction, adjust its tone for premium versus standard customers, incorporate the last three months of transaction history, and stay within a 4,000-token context window — all in under 300 milliseconds. You can't hand-write a prompt for every combination of these variables. There are millions of them. What you need is a system that builds the right prompt for each request automatically, assembling the relevant pieces at runtime from a constellation of data sources, rules, and user signals.

This is contextual compression and dynamic prompt assembly. It's the discipline of treating prompts not as static text files but as structured artifacts that get compiled on-the-fly from components — just like a database query gets compiled from parameters, or a UI gets rendered from state. When you do it well, your AI system behaves like it was briefed perfectly for every single request. When you do it poorly, you get irrelevant context flooding expensive token budgets, security vulnerabilities from injected inputs, and outputs that ignore half of what the model was given because the prompt was too long and incoherent.

By the end of this lesson, you'll be able to design and implement a full dynamic prompt assembly pipeline from scratch. You'll know how to compress and select context intelligently, apply business rules as prompt modifiers, and defend the system against adversarial inputs — all in production-grade code.

What you'll learn:

• How to architect a modular prompt assembly system with distinct layers for system context, business rules, retrieved data, and user input
• Techniques for contextual compression: extracting signal from large data sources without blowing up token budgets
• How to apply conditional business logic that modifies prompt content at runtime based on user attributes, data state, or organizational rules
• Security patterns for sanitizing user inputs before they enter a dynamically assembled prompt
• Performance optimization strategies including caching, lazy loading, and pre-computed prompt segments

---

Prerequisites

You should be comfortable with the fundamentals of prompt construction — if you haven't already, review Prompt Engineering Fundamentals for Data Professionals before continuing. You should also understand tokens and context windows at a mechanical level (the Tokens, Context Windows, and Input Limits article is the right primer). Basic Python proficiency is assumed throughout this lesson. Familiarity with REST APIs and JSON will help.

---

The Architecture of a Dynamic Prompt

Before writing a single line of code, you need a mental model for what a dynamically assembled prompt actually is. Think of it as a layered document with distinct zones of concern, each controlled by different parts of your application.

Here's a canonical four-layer architecture:

┌─────────────────────────────────────────────────────┐
│  LAYER 1: System Foundation                          │
│  Role definition, tone, output format constraints   │
│  Static or near-static. Changes rarely.             │
├─────────────────────────────────────────────────────┤
│  LAYER 2: Business Rules & Personalization          │
│  Conditionally injected based on user attributes,   │
│  org config, compliance requirements, feature flags │
├─────────────────────────────────────────────────────┤
│  LAYER 3: Retrieved & Compressed Context           │
│  Data from databases, documents, APIs               │
│  Compressed to fit token budget                     │
├─────────────────────────────────────────────────────┤
│  LAYER 4: User Input (Sanitized)                   │
│  The actual user query or instruction               │
│  Never trusted; always validated before insertion  │
└─────────────────────────────────────────────────────┘

This layering matters for two reasons. First, it maps responsibility clearly: your infrastructure team owns Layer 1, your product team owns Layer 2, your data pipeline team owns Layer 3, and your security team has veto power over Layer 4. Second, it gives you a principled framework for deciding what to include and what to compress or exclude when your token budget is under pressure — always preserve Layer 1 intact, compress Layer 3 aggressively, and never truncate Layer 4.

Key insight: A dynamic prompt assembly system is fundamentally a content management system for AI inputs. The same discipline you'd apply to managing database schemas and API contracts applies here. Treat your prompt layers as versioned artifacts with explicit contracts between them.

---

Building the Foundation: Prompt Templates with Injection Points

The first concrete step is moving from literal strings to template objects with clearly defined injection points. Python's string.Template is too limited for production use. Instead, build a lightweight template class that understands your layer structure and enforces contracts at construction time.

from dataclasses import dataclass, field
from typing import Optional, Dict, Any
import re

@dataclass
class PromptLayer:
    name: str
    content: str
    token_budget: int  # Maximum tokens this layer may consume
    required: bool = True
    priority: int = 0  # Higher = more important when space is tight

@dataclass
class PromptTemplate:
    system_foundation: str
    injection_points: Dict[str, PromptLayer] = field(default_factory=dict)
    
    def register_layer(self, layer: PromptLayer):
        self.injection_points[layer.name] = layer
    
    def validate(self) -> bool:
        """Check that all required injection points have content."""
        for name, layer in self.injection_points.items():
            if layer.required and not layer.content.strip():
                raise ValueError(f"Required layer '{name}' has no content.")
        return True

This gives you a structured container. But the real intelligence is in the assembler that takes these layers and produces a final prompt string. Notice that each layer carries a token_budget — that's not a guideline, it's a hard constraint enforced during assembly.

For counting tokens accurately, use the tiktoken library for OpenAI models or the anthropic SDK's built-in token counting for Claude. Never approximate by character count — the variance is too high for structured data.

import tiktoken

class TokenCounter:
    def __init__(self, model: str = "gpt-4o"):
        self.encoder = tiktoken.encoding_for_model(model)
    
    def count(self, text: str) -> int:
        return len(self.encoder.encode(text))
    
    def fits_within(self, text: str, budget: int) -> bool:
        return self.count(text) <= budget

---

Contextual Compression: The Core Technical Problem

Here's the painful reality: the data that would make a prompt most useful is often enormous. A customer's transaction history might be 2,000 rows. A product knowledge base might contain 50,000 words. A regulatory document might run 400 pages. You can't inject all of it, and you can't just truncate it arbitrarily — truncation cuts equally from signal and noise.

Contextual compression is the discipline of reducing a large data source to its most query-relevant, most token-efficient representation. There are four main techniques, and the right choice depends on your data type and latency requirements.

Technique 1: Structured Extraction with Relevance Scoring

For tabular or semi-structured data, you pre-process the source at query time to extract only the fields and rows that are relevant to the specific user request.

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def compress_transaction_history(
    transactions_df: pd.DataFrame,
    user_query: str,
    max_rows: int = 10,
    max_tokens: int = 800
) -> str:
    """
    Given a full transaction history DataFrame and a user query,
    return a compressed, formatted string of the most relevant transactions.
    """
    # Combine relevant fields into a searchable text representation
    transactions_df = transactions_df.copy()
    transactions_df['search_text'] = (
        transactions_df['merchant_name'] + " " +
        transactions_df['category'] + " " +
        transactions_df['description'].fillna("")
    )
    
    # Vectorize and score relevance to query
    vectorizer = TfidfVectorizer(stop_words='english')
    corpus = transactions_df['search_text'].tolist() + [user_query]
    tfidf_matrix = vectorizer.fit_transform(corpus)
    
    query_vector = tfidf_matrix[-1]
    doc_vectors = tfidf_matrix[:-1]
    
    scores = cosine_similarity(query_vector, doc_vectors).flatten()
    transactions_df['relevance'] = scores
    
    # Always include recent transactions regardless of relevance (recency bias)
    recent_indices = transactions_df.nlargest(3, 'date').index
    relevant_indices = transactions_df.nlargest(max_rows, 'relevance').index
    
    selected_indices = list(set(recent_indices) | set(relevant_indices))
    selected = transactions_df.loc[selected_indices].sort_values('date', ascending=False)
    
    # Format as a compressed string, not a full table
    lines = []
    for _, row in selected.iterrows():
        lines.append(
            f"{row['date'].strftime('%Y-%m-%d')}: {row['merchant_name']} "
            f"${row['amount']:.2f} ({row['category']})"
        )
    
    return "Recent relevant transactions:\n" + "\n".join(lines)

This approach gives you control over what gets included (query-relevant rows) and how it's formatted (compact lines, not JSON blobs with every field). The combination of recency bias and relevance scoring reflects how a human analyst would brief someone: "Here are the three most recent transactions, plus anything that looks related to what you're asking about."

Tip: When compressing tabular data for prompt injection, never include column headers twice or repeat data type annotations. Instead, use a consistent one-line-per-record format. A header row that says "date | merchant | amount | category" costs you 8 tokens and saves 0 tokens over the naturally readable "2024-01-15: Amazon $47.99 (Shopping)" format.

Technique 2: Semantic Chunking with Vector Retrieval

For document-based context — policies, product specs, knowledge base articles — you need a retrieval layer before compression. This is the RAG pattern, and it's covered in depth in Retrieval-Augmented Generation in Practice. The compression angle here is specific: after retrieval, you still need to compress what you retrieved.

from typing import List

def compress_retrieved_chunks(
    chunks: List[str],
    query: str,
    max_tokens: int = 1200,
    counter: TokenCounter = None
) -> str:
    """
    Given retrieved document chunks (already ranked by vector similarity),
    pack as many as fit within the token budget, preferring earlier (higher-ranked) chunks.
    """
    if counter is None:
        counter = TokenCounter()
    
    header = "Relevant context from knowledge base:\n"
    used_tokens = counter.count(header)
    selected_chunks = []
    
    for i, chunk in enumerate(chunks):
        # Add a separator and chunk index for citation purposes
        chunk_text = f"[Source {i+1}]: {chunk.strip()}\n"
        chunk_tokens = counter.count(chunk_text)
        
        if used_tokens + chunk_tokens > max_tokens:
            # Don't include partial chunks — they're often worse than nothing
            break
        
        selected_chunks.append(chunk_text)
        used_tokens += chunk_tokens
    
    if not selected_chunks:
        return ""
    
    return header + "\n".join(selected_chunks)

Notice the design decision: we don't include partial chunks. A half-truncated policy document that cuts off mid-sentence is actively dangerous — the model might complete the sentence incorrectly or treat the truncation as meaningful. Better to include fewer complete chunks.

Technique 3: LLM-Assisted Summarization as a Pre-Pass

For narrative data — call transcripts, support ticket histories, meeting notes — neither keyword matching nor vector retrieval gives you clean compression. Instead, you run a fast, cheap summarization pass before the main prompt assembly.

import openai

def summarize_for_context(
    raw_text: str,
    focus_query: str,
    model: str = "gpt-4o-mini",  # Use a cheaper model for pre-processing
    max_summary_tokens: int = 300
) -> str:
    """
    Use a lightweight LLM call to compress a narrative document
    into a focused summary relevant to the user's query.
    """
    client = openai.OpenAI()
    
    compression_prompt = f"""Summarize the following text in under {max_summary_tokens} tokens.
Focus specifically on information relevant to: {focus_query}
Omit anything not relevant to that focus. Use concise, factual language.

Text to summarize:
{raw_text[:8000]}  # Hard cap on input to this sub-call
"""
    
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": compression_prompt}],
        max_tokens=max_summary_tokens,
        temperature=0.1  # Low temperature for consistent compression
    )
    
    return response.choices[0].message.content.strip()

Warning: LLM-assisted summarization adds latency and cost. It's appropriate when the source data is narrative and query-dependent, but it's overkill for structured data where rule-based extraction is faster and more reliable. Never default to this technique — use it only when structure isn't available.

The other risk is summarization hallucination — the summarizing model adds details not present in the source. Always use low temperature (0.0–0.1) for compression passes, and consider validating that key entities (names, numbers, dates) in the summary actually appear in the source text.

Technique 4: Template-Driven Compression

For data sources you control (your own databases, CRMs, data warehouses), the most efficient approach is designing a database query that already returns prompt-ready output rather than fetching raw rows and formatting in Python.

-- Instead of fetching all account data and formatting it in Python:
-- SELECT * FROM accounts WHERE customer_id = ?
-- 
-- Write a query that returns a prompt-ready summary:

SELECT 
    CONCAT(
        'Account tier: ', account_tier, '\n',
        'Open since: ', DATE_FORMAT(created_at, '%B %Y'), '\n', 
        'YTD spend: $', FORMAT(ytd_spend, 2), '\n',
        'Outstanding balance: $', FORMAT(current_balance, 2), '\n',
        'Last contact: ', DATE_FORMAT(last_contact_date, '%Y-%m-%d'), 
            ' (', last_contact_channel, ')\n',
        'Active products: ', GROUP_CONCAT(product_name ORDER BY product_name SEPARATOR ', ')
    ) as prompt_context
FROM accounts a
JOIN account_products ap ON a.id = ap.account_id
WHERE a.customer_id = ?
GROUP BY a.id;

This approach pushes the formatting work to the database, reduces network transfer, and means you can adjust the format by changing one SQL query rather than hunting through Python code. The trade-off is that your SQL layer now carries prompt-formatting responsibility, which some teams find uncomfortable from an architecture perspective. Document it clearly.

---

Business Rules as Prompt Modifiers

Static system prompts can't carry the full weight of business logic. A compliance prompt for a user in the European Union is different from one for a user in Canada. A prompt for a premium-tier customer should give different guidance than one for a free-tier user. A prompt during an active outage incident should acknowledge the situation. These are runtime conditions that must modify the prompt dynamically.

The right pattern for this is a rule engine that evaluates a context object and returns a list of prompt segments to inject.

from typing import List, Callable
from dataclasses import dataclass
from enum import Enum

class RulePriority(Enum):
    CRITICAL = 1   # Compliance, legal — always included, trim others first
    HIGH = 2       # Product tier, regional settings
    MEDIUM = 3     # Personalization, feature flags
    LOW = 4        # Nice-to-have context

@dataclass
class PromptModifier:
    segment: str          # The text to inject
    priority: RulePriority
    layer: str            # Which layer this belongs to ("rules", "context", etc.)
    token_estimate: int = 0

class BusinessRuleEngine:
    def __init__(self):
        self.rules: List[Callable] = []
    
    def register(self, rule_fn: Callable):
        """Decorator to register a rule function."""
        self.rules.append(rule_fn)
        return rule_fn
    
    def evaluate(self, context: dict) -> List[PromptModifier]:
        """
        Run all registered rules against the context dict.
        Return a list of PromptModifiers to inject.
        """
        modifiers = []
        for rule_fn in self.rules:
            result = rule_fn(context)
            if result is not None:
                if isinstance(result, list):
                    modifiers.extend(result)
                else:
                    modifiers.append(result)
        
        return sorted(modifiers, key=lambda m: m.priority.value)

# Instantiate the engine
rules = BusinessRuleEngine()

@rules.register
def compliance_jurisdiction_rule(ctx: dict) -> Optional[PromptModifier]:
    """Inject jurisdiction-specific compliance language."""
    jurisdiction = ctx.get("user_jurisdiction", "US")
    
    compliance_text = {
        "EU": (
            "REGULATORY REQUIREMENT: This user is subject to GDPR. "
            "Do not recommend storing personal data without explicit consent. "
            "If asked about data retention, cite the right to erasure."
        ),
        "CA": (
            "REGULATORY REQUIREMENT: This user is subject to PIPEDA and provincial privacy laws. "
            "Financial recommendations must include risk disclosure language."
        ),
        "US": (
            "Recommendations involving financial products must include standard risk disclosures. "
            "Do not provide specific investment advice."
        )
    }
    
    text = compliance_text.get(jurisdiction, compliance_text["US"])
    return PromptModifier(
        segment=text,
        priority=RulePriority.CRITICAL,
        layer="rules",
        token_estimate=len(text.split()) + 10  # rough estimate
    )

@rules.register
def account_tier_rule(ctx: dict) -> Optional[PromptModifier]:
    """Adjust response guidance based on customer tier."""
    tier = ctx.get("account_tier", "standard")
    
    if tier == "premium":
        return PromptModifier(
            segment=(
                "This is a Premium account holder. Offer concierge-level detail "
                "and proactively suggest account optimization opportunities. "
                "You may reference our advisor team as an escalation option."
            ),
            priority=RulePriority.HIGH,
            layer="rules"
        )
    elif tier == "trial":
        return PromptModifier(
            segment=(
                "This user is on a 30-day trial. Keep responses focused on core features. "
                "When relevant, mention that full access unlocks with a paid subscription."
            ),
            priority=RulePriority.HIGH,
            layer="rules"
        )
    return None  # No modifier needed for standard tier

@rules.register  
def active_incident_rule(ctx: dict) -> Optional[PromptModifier]:
    """Inject incident awareness if there's an active system issue."""
    incident = ctx.get("active_incident")
    if not incident:
        return None
    
    return PromptModifier(
        segment=(
            f"SYSTEM STATUS: There is currently a known issue with {incident['affected_service']}. "
            f"Estimated resolution: {incident['eta']}. "
            "If the user reports problems, acknowledge this known issue rather than troubleshooting."
        ),
        priority=RulePriority.CRITICAL,
        layer="rules"
    )

Key insight: Business rules should return None when they don't apply, not empty strings. Your assembler can then cleanly filter out non-applicable modifiers without inserting blank lines or empty sections into the prompt — both of which waste tokens and can confuse the model's parsing of the prompt structure.

---

The Assembler: Putting It All Together

Now you have all the components. The assembler is the orchestration layer that:

1. Collects all prompt segments
2. Evaluates the token budget
3. Prioritizes segments when the budget is tight
4. Assembles the final prompt string in the correct order

class PromptAssembler:
    def __init__(
        self, 
        total_token_budget: int,
        counter: TokenCounter,
        model: str = "gpt-4o"
    ):
        self.total_budget = total_token_budget
        self.counter = counter
        self.model = model
        # Reserve tokens for the response itself
        self.response_reservation = 1000
        self.available_budget = total_token_budget - self.response_reservation
    
    def assemble(
        self,
        system_foundation: str,
        modifiers: List[PromptModifier],
        context_segments: List[dict],  # [{"text": "...", "priority": RulePriority.MEDIUM}]
        user_input: str
    ) -> dict:
        """
        Assemble the final prompt, respecting token budgets.
        Returns a dict suitable for the OpenAI messages API.
        """
        # Layer 1: System foundation is sacred
        foundation_tokens = self.counter.count(system_foundation)
        user_tokens = self.counter.count(user_input)
        
        if foundation_tokens + user_tokens > self.available_budget:
            raise ValueError(
                f"System foundation ({foundation_tokens}) + user input ({user_tokens}) "
                f"already exceeds available budget ({self.available_budget}). "
                "Reduce foundation size or increase context window."
            )
        
        remaining = self.available_budget - foundation_tokens - user_tokens
        
        # Layer 2: Business rules (sorted by priority, CRITICAL first)
        system_content = system_foundation + "\n\n"
        
        for modifier in sorted(modifiers, key=lambda m: m.priority.value):
            segment_tokens = self.counter.count(modifier.segment)
            
            if modifier.priority == RulePriority.CRITICAL:
                # CRITICAL rules are always included even if over budget
                # (they should be kept short by design)
                system_content += modifier.segment + "\n\n"
                remaining -= segment_tokens
            elif segment_tokens <= remaining:
                system_content += modifier.segment + "\n\n"
                remaining -= segment_tokens
            else:
                # Log that a non-critical modifier was dropped
                print(f"[PromptAssembler] Dropped modifier due to budget: {modifier.segment[:50]}...")
        
        # Layer 3: Context segments (pack by priority until budget is exhausted)
        context_parts = []
        for seg in sorted(context_segments, key=lambda s: s.get("priority", RulePriority.LOW).value):
            seg_tokens = self.counter.count(seg["text"])
            if seg_tokens <= remaining:
                context_parts.append(seg["text"])
                remaining -= seg_tokens
        
        if context_parts:
            system_content += "\n".join(context_parts)
        
        return {
            "system": system_content.strip(),
            "user": user_input  # Already sanitized by the time it arrives here
        }

One subtlety here: CRITICAL modifiers are always included even if they push over budget. This is intentional. A compliance rule that gets dropped because you included too much transaction history is a regulatory liability, not an acceptable trade-off. Design your CRITICAL modifiers to be extremely concise (under 100 tokens) precisely because they'll always consume budget.

---

Sanitizing User Input: The Security Layer

Dynamic prompt assembly has a significant attack surface. Because user input gets incorporated into the prompt that the model executes, a malicious user can attempt to override your system instructions — a class of attack called prompt injection.

The mitigation strategy operates at multiple levels:

import re
from typing import Tuple

class InputSanitizer:
    # Patterns that suggest injection attempts
    INJECTION_PATTERNS = [
        r"ignore (all |previous |prior |above |the )?(instructions?|prompt|system|rules?)",
        r"you are now",
        r"new instructions?:",
        r"override (system|instructions?|rules?)",
        r"(pretend|act|behave) (like|as if) you",
        r"disregard (all |previous |your )?",
        r"\/\/ ?system",  # Attempts to inject system markers
        r"<\|system\|>",  # OpenAI-style delimiter injection
    ]
    
    # Patterns that indicate the user is trying to include structured prompt syntax
    STRUCTURAL_PATTERNS = [
        r"\[INST\]",      # Llama instruction format
        r"<s>",           # BOS token
        r"Human:",        # Chat format injection
        r"Assistant:",
        r"<\|im_start\|>",
    ]
    
    def __init__(self, max_input_length: int = 2000):
        self.max_length = max_input_length
        self.injection_regex = re.compile(
            "|".join(self.INJECTION_PATTERNS),
            re.IGNORECASE
        )
        self.structural_regex = re.compile(
            "|".join(self.STRUCTURAL_PATTERNS),
            re.IGNORECASE
        )
    
    def sanitize(self, user_input: str) -> Tuple[str, List[str]]:
        """
        Clean and validate user input.
        Returns (sanitized_input, list_of_warnings).
        Raises ValueError if input is categorically unsafe.
        """
        warnings = []
        
        # 1. Length check
        if len(user_input) > self.max_length:
            user_input = user_input[:self.max_length]
            warnings.append(f"Input truncated to {self.max_length} characters.")
        
        # 2. Injection pattern detection
        injection_matches = self.injection_regex.findall(user_input)
        if injection_matches:
            # Log for security monitoring but don't reveal detection to user
            warnings.append(f"SECURITY: Potential injection attempt detected: {injection_matches}")
            # Option 1: Reject outright
            raise ValueError("Input contains patterns that cannot be processed.")
            # Option 2: Strip and continue (weaker, use only for low-risk contexts)
            # user_input = self.injection_regex.sub("[removed]", user_input)
        
        # 3. Structural pattern stripping (less severe — just clean it up)
        if self.structural_regex.search(user_input):
            warnings.append("Structural prompt markers stripped from input.")
            user_input = self.structural_regex.sub("", user_input)
        
        # 4. Wrap in explicit demarcation
        # This doesn't prevent injection but makes structural attacks harder
        sanitized = f'User question: "{user_input.strip()}"'
        
        return sanitized, warnings

Warning: Wrapping user input in demarcation quotes (User question: "...") is a defense-in-depth measure, not a complete solution. A sufficiently motivated attacker can escape quotes. The real defense is a well-structured system prompt that explicitly instructs the model to ignore contradictory instructions from the user section — combined with output validation that detects when the model's behavior has been altered.

The sanitize method returns warnings as a separate list rather than raising on every issue. This lets your calling code decide whether to proceed (low-risk application), escalate (security monitoring), or hard-fail (high-compliance context). Don't bake the error-handling policy into the sanitizer itself.

---

A Complete End-to-End Example

Let's wire everything together with a realistic scenario: an internal analytics assistant for a SaaS company's customer success team. The assistant helps CSMs understand account health, draft renewal talking points, and get product usage insights.

import openai
from datetime import datetime

# --- Configuration ---
TOTAL_CONTEXT_WINDOW = 128000  # gpt-4o
RESPONSE_RESERVATION = 2000
PROMPT_BUDGET = TOTAL_CONTEXT_WINDOW - RESPONSE_RESERVATION

SYSTEM_FOUNDATION = """You are an AI assistant for Meridian's Customer Success team.
Your role is to help Customer Success Managers (CSMs) understand account health, 
prepare for renewal conversations, and surface product usage insights.

Core behaviors:
- Be concise and action-oriented. CSMs are busy; get to the point.
- When citing data, reference the time period it covers.
- If data is missing or ambiguous, say so explicitly rather than inferring.
- Never promise features that aren't confirmed in the product context provided.
- Format responses with clear headers when covering multiple topics."""

def build_csm_assistant_prompt(
    user_query: str,
    csm_context: dict,        # Who is the CSM, their permissions, region
    account_data: dict,        # Account attributes, health scores, etc.
    usage_df: pd.DataFrame,    # Product usage events
    knowledge_chunks: List[str],  # Retrieved KB articles
    incident_context: dict    # Any active incidents
) -> dict:
    """
    Build the complete prompt for the CSM assistant.
    """
    counter = TokenCounter(model="gpt-4o")
    assembler = PromptAssembler(
        total_token_budget=PROMPT_BUDGET,
        counter=counter,
        model="gpt-4o"
    )
    sanitizer = InputSanitizer(max_input_length=1500)
    rule_engine = BusinessRuleEngine()
    
    # Register rules (in production, these would be auto-registered at startup)
    rule_engine.rules = [
        compliance_jurisdiction_rule,
        account_tier_rule,
        active_incident_rule
    ]
    
    # --- Step 1: Sanitize user input ---
    try:
        clean_query, warnings = sanitizer.sanitize(user_query)
        if any("SECURITY" in w for w in warnings):
            # Log to SIEM and return a safe error
            log_security_event(warnings, csm_context)
            raise ValueError("Query could not be processed.")
    except ValueError as e:
        return {"error": str(e)}
    
    # --- Step 2: Evaluate business rules ---
    rule_context = {
        "user_jurisdiction": csm_context.get("region", "US"),
        "account_tier": account_data.get("tier", "standard"),
        "active_incident": incident_context if incident_context.get("active") else None
    }
    modifiers = rule_engine.evaluate(rule_context)
    
    # --- Step 3: Compress context layers ---
    
    # 3a: Account summary (structured extraction from dict)
    account_summary = f"""Account: {account_data['company_name']} (ID: {account_data['id']})
Health Score: {account_data['health_score']}/100 (trend: {account_data['health_trend']})
Contract value: ${account_data['arr']:,.0f} ARR | Renewal: {account_data['renewal_date']}
Plan: {account_data['plan_name']} | Seats: {account_data['active_seats']}/{account_data['licensed_seats']}
Primary contact: {account_data['primary_contact_name']} ({account_data['primary_contact_title']})"""

    # 3b: Compressed usage data
    usage_summary = compress_transaction_history(
        transactions_df=usage_df,
        user_query=user_query,
        max_rows=8,
        max_tokens=600
    )
    
    # 3c: Retrieved knowledge base chunks
    kb_context = compress_retrieved_chunks(
        chunks=knowledge_chunks,
        query=user_query,
        max_tokens=800,
        counter=counter
    )
    
    context_segments = [
        {
            "text": "## Account Overview\n" + account_summary,
            "priority": RulePriority.HIGH
        },
        {
            "text": "## Usage Activity\n" + usage_summary,
            "priority": RulePriority.MEDIUM
        }
    ]
    
    if kb_context:
        context_segments.append({
            "text": "## Product Knowledge\n" + kb_context,
            "priority": RulePriority.MEDIUM
        })
    
    # --- Step 4: Assemble ---
    messages = assembler.assemble(
        system_foundation=SYSTEM_FOUNDATION,
        modifiers=modifiers,
        context_segments=context_segments,
        user_input=clean_query
    )
    
    # --- Step 5: Call the API ---
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": messages["system"]},
            {"role": "user", "content": messages["user"]}
        ],
        max_tokens=2000,
        temperature=0.3
    )
    
    return {
        "response": response.choices[0].message.content,
        "tokens_used": response.usage.total_tokens,
        "prompt_tokens": response.usage.prompt_tokens,
        "warnings": warnings
    }

This is production-representative code. Notice that the function signature makes dependencies explicit — there's no hidden global state, every data source is a parameter. This makes the function testable in isolation, which is critical for a system you'll need to debug at 2am when a CSM can't get the account data they need for a renewal call.

---

Performance Optimization: Making It Fast

Dynamic prompt assembly adds latency. Every step — database queries, vector retrieval, compression passes, rule evaluation — happens in the critical path of the user's request. At scale, this compounds badly. Here are the techniques that actually move the needle.

Parallel Retrieval

Most of your context-gathering steps are IO-bound. Run them concurrently.

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

async def gather_context_async(account_id: str, query: str) -> dict:
    """
    Fetch all context sources in parallel rather than sequentially.
    """
    with ThreadPoolExecutor() as executor:
        loop = asyncio.get_event_loop()
        
        # These three IO operations run concurrently
        account_future = loop.run_in_executor(
            executor, fetch_account_data, account_id
        )
        usage_future = loop.run_in_executor(
            executor, fetch_usage_data, account_id
        )
        chunks_future = loop.run_in_executor(
            executor, retrieve_kb_chunks, query
        )
        
        account_data, usage_data, kb_chunks = await asyncio.gather(
            account_future, usage_future, chunks_future
        )
    
    return {
        "account": account_data,
        "usage": usage_data, 
        "kb_chunks": kb_chunks
    }

In practice, sequential retrieval of three data sources that each take 100ms costs you 300ms. Parallel retrieval costs you the maximum of the three — typically 100-120ms. For a user-facing system, this is the single biggest latency win.

Pre-Computing Static Segments

Your system foundation and most business rule segments don't change between requests for the same user class. Pre-compute and cache them.

import functools
import hashlib

@functools.lru_cache(maxsize=256)
def get_cached_rule_modifiers(
    jurisdiction: str,
    account_tier: str
) -> str:
    """
    Cache rule modifier output for common (jurisdiction, tier) combinations.
    The cache key is deterministic — same inputs always produce same modifiers
    (as long as no active incident is involved).
    """
    context = {"user_jurisdiction": jurisdiction, "account_tier": account_tier}
    # Note: incident rules are NOT evaluated here — they're always dynamic
    modifiers = evaluate_static_rules(context)
    return "\n\n".join(m.segment for m in modifiers)

Tip: For the cost optimization of your AI API usage, consider using OpenAI's prompt caching feature (available for prefixes over 1,024 tokens) or Anthropic's equivalent. If your system foundation and business rules are consistent across requests, you can get a cache hit on those tokens and pay only for the variable context and user input. This can reduce prompt token costs by 50-70% for high-volume systems.

Budget-Aware Early Termination

Don't retrieve more data than you can use. If you know your KB retrieval will return 10 chunks but your budget only allows 3, don't retrieve 10. Add token awareness to your retrieval queries.

def estimate_available_context_budget(
    foundation_tokens: int,
    modifier_tokens: int,
    user_input_tokens: int,
    total_budget: int,
    response_reservation: int = 1000
) -> int:
    """
    Calculate how many tokens remain for retrieved context
    before making retrieval calls.
    """
    used = foundation_tokens + modifier_tokens + user_input_tokens + response_reservation
    return max(0, total_budget - used)

Pass this budget value into your retrieval functions so they limit result set size at the query level — not after you've already paid the retrieval cost.

---

Monitoring and Observability

A dynamic prompt assembly system is opaque by nature. The prompt that caused a bad output was assembled at runtime and may never be seen again with the same combination of inputs. You need to instrument it deliberately.

Log the following for every request:

@dataclass
class PromptAssemblyTrace:
    request_id: str
    timestamp: datetime
    user_id: str
    account_id: str
    
    # What went into the prompt
    foundation_tokens: int
    modifier_segments_included: List[str]
    modifier_segments_dropped: List[str]
    context_segments_included: List[str]
    context_segments_dropped: List[str]
    user_input_tokens: int
    
    # Final token accounting
    total_prompt_tokens: int
    response_tokens: int
    
    # Quality signals
    compression_ratio: float  # (raw context size) / (included context size)
    budget_utilization: float  # total_prompt_tokens / available_budget
    security_flags: List[str]

The compression_ratio and budget_utilization metrics tell you whether your assembly system is working well. A compression ratio consistently above 10:1 means you're throwing away a lot of potentially useful context — investigate whether your retrieval is returning relevant chunks. A budget utilization consistently below 0.5 means you're wasting capacity — you could include richer context. These metrics, tracked over time alongside your output quality scores from an AI evaluation framework, will guide your tuning.

---

Hands-On Exercise

Build a dynamic prompt assembly system for a support ticket triage assistant. The assistant should help a support agent understand a customer's issue, review relevant past tickets, and suggest resolution approaches.

Setup:

1. Create a mock customer database with at least 5 customer records, each with a different account tier (free, starter, business, enterprise) and different geographic regions.
2. Create a mock ticket history with 20-30 past tickets across these customers, each with a subject, description, status, and resolution (if resolved).
3. Create a small knowledge base of 10-15 support articles in plain text format.

Build the following components:

Part 1: Context Compression Write a compress_ticket_history function that takes a customer's ticket history and a current ticket description, and returns a compressed string of the most relevant past tickets (max 500 tokens). Use TF-IDF relevance scoring similar to the transaction history example above.

Part 2: Business Rules Implement at least three business rules as modifiers:

• Enterprise customers should get a modifier indicating escalation paths to a named technical account manager
• Free-tier customers should get a modifier reminding the agent about feature limitations before suggesting solutions
• Any customer with more than 3 open tickets should trigger a "high-engagement account" modifier

Part 3: Security Test your InputSanitizer with at least 5 adversarial inputs. Document which patterns your sanitizer catches and which it misses. For the ones it misses, propose a mitigation strategy.

Part 4: Assembly and Tracing Wire the components together using the PromptAssembler class. Add tracing that logs which modifier segments were included versus dropped, and what the final token budget utilization was.

Stretch Goal: Implement a simple cache for the static rule modifiers and measure the latency difference between cached and uncached assembly for 100 sequential requests.

---

Common Mistakes & Troubleshooting

Mistake 1: Formatting Context as JSON Blobs

A common error is injecting retrieved data as raw JSON:

# Bad — wastes tokens on syntax, hard for model to read
context = json.dumps(account_data)

# Good — structured prose with clear labels
context = f"Account health: {account_data['health_score']}/100 ({account_data['health_trend']} trend)"

JSON formatting uses roughly 30-40% more tokens than equivalent human-readable text for the same information. The model doesn't need the syntax — it needs the content. If you genuinely need structured output from the model, ask for it in the output format instructions, not by providing JSON-formatted input.

Note: The exception is when you're passing data that will be processed programmatically inside a tool call or function call pipeline, where precise key names matter. In those cases, a minimal JSON subset is appropriate. For human-readable context, always prefer prose.

Mistake 2: Treating All Context as Equally Important

When the budget runs out and you're dropping segments, the order in which you include segments matters enormously. If you include low-priority personalization text before high-priority usage data, you'll drop the usage data — which is exactly backwards.

Always sort segments by priority before packing. And revisit your priority assignments quarterly — priorities drift as your product evolves.

Mistake 3: Ignoring the Model's Attention Mechanics

Long prompts don't get uniform attention. Research consistently shows that large language models tend to weight the beginning and end of context more heavily than the middle — the "lost in the middle" problem. Structure your prompt so that the most important instructions appear at the top (system foundation) and the most important context appears just before the user query, not buried in the middle.

This means you should order your context injection like this:

[System Foundation]      ← High attention (beginning)
[Business Rules]         ← High attention (near beginning)
[Less critical context]  ← Lower attention (middle)
[Most critical context]  ← High attention (near end)
[User Query]            ← High attention (end)

Mistake 4: Over-Engineering the Compression Logic

It's tempting to build elaborate multi-stage compression pipelines with LLM pre-passes, reranking, and semantic deduplication. In practice, for most applications, well-designed TF-IDF retrieval plus a token-budget packing algorithm gets you 80% of the benefit at 10% of the complexity. Start simple, instrument carefully, and only add compression sophistication when you can demonstrate from traces that context relevance is a real problem.

Mistake 5: Forgetting to Version Your Prompt Components

Your system foundation, business rule segments, and compression prompts are software artifacts that should live in version control. When a prompt change causes a regression in output quality, you need to be able to identify which component changed and revert precisely. Treat prompt components like database schema migrations — versioned, tested, and deployed deliberately.

Troubleshooting: The Model Ignores Injected Context

If you're seeing outputs that ignore data you injected, check:

1. Token count: Verify the context actually made it into the prompt. Add logging of final prompt token counts.
2. Formatting: Is the context visually distinct enough from other sections? Add clear headers.
3. Contradictions: Does your injected context contradict the system foundation? The model will often resolve contradictions by ignoring the newer instruction if it conflicts with a strong earlier directive.
4. Relevance: Did your compression algorithm actually select relevant chunks? Log which chunks were selected and manually verify they're germane to the test query.

---

Summary & Next Steps

You've now built a complete mental model and implementation framework for dynamic prompt assembly. The core principles:

• Treat prompts as compiled artifacts, not static text. Design with layers, injection points, and token budgets from the start.
• Compress context intelligently using the right technique for each data type: structured extraction for tabular data, semantic chunking for documents, LLM pre-passes for narrative data, and SQL-level formatting for owned databases.
• Apply business rules as modifiers evaluated at runtime from a rule engine that separates CRITICAL compliance constraints from MEDIUM personalization concerns — and never drops the former.
• Sanitize user input as a first-class concern, not an afterthought. Wrap, validate, and log every piece of user-provided text before it enters the prompt.
• Instrument everything: compression ratios, budget utilization, and segment inclusion/exclusion traces are your debugging tools when something goes wrong in production.

For your next skill development, explore how these assembly patterns integrate with multi-turn conversation management — specifically how you maintain state across turns without re-injecting the full context every time. The LLM Memory Architecture for Enterprise Applications article covers exactly this. You should also look at how dynamic prompt assembly combines with prompt chaining to build complex multi-step workflows where the output of one assembled prompt becomes the context input for the next.

If your organization is evaluating whether to build dynamic prompt assembly pipelines at all versus alternatives like fine-tuning, the Fine-Tuning vs. RAG vs. Prompt Engineering decision framework will help you make the right architectural choice for your specific constraints.

Finally, as you move dynamic prompt assembly into production, ensure your system is covered by proper guardrails — the Embedding AI Guardrails in Production Workflows article gives you the input validation and output filtering patterns that complement what you've built here.

Dynamic prompt assembly is, at its heart, the discipline of treating AI inputs with the same engineering rigor you'd apply to any other critical production system. The teams that get this right don't just get better AI outputs — they get predictable, auditable, maintainable AI outputs. That's the difference between a demo and a product.