Prompt templating and multi-turn conversation management

Why this matters

Prompts that are hardcoded strings are fine in a prototype. In production, prompts need to be parameterized (so the same structure works for different users or contexts), versioned (so you can roll back), and composed (so shared instruction blocks do not get copy-pasted across features). Conversation state management determines whether your multi-turn feature feels natural or forgets context at the worst moment.

This lesson covers the patterns that make both tractable at production scale.

Core concepts

Prompt templating. The simplest template system is Python string formatting with explicit variable names. The critical practice is keeping the template and the variables separate — never interpolate untrusted user input directly into the system prompt:

from dataclasses import dataclass
from typing import Any
import re

@dataclass
class PromptTemplate:
    name: str
    version: int
    system_template: str
    user_template: str
    required_vars: list[str]

    def render_system(self, variables: dict[str, Any]) -> str:
        self._validate_vars(variables)
        return self.system_template.format(**variables)

    def render_user(self, user_input: str, variables: dict[str, Any]) -> str:
        # user_input is always kept separate — never interpolated into system
        return self.user_template.format(user_input=user_input, **variables)

    def _validate_vars(self, variables: dict[str, Any]) -> None:
        missing = [v for v in self.required_vars if v not in variables]
        if missing:
            raise ValueError(f"Template '{self.name}' missing variables: {missing}")


# Template definition (lives in config or database, not scattered in route handlers)
SUPPORT_TEMPLATE = PromptTemplate(
    name="customer-support",
    version=3,
    system_template=(
        "You are a support agent for {company_name}. "
        "Respond helpfully and concisely. "
        "Your customer tier is: {customer_tier}. "
        "Tone: {tone}."
    ),
    user_template="{user_input}",
    required_vars=["company_name", "customer_tier", "tone"],
)

# Usage
system_prompt = SUPPORT_TEMPLATE.render_system({
    "company_name": "Acme Corp",
    "customer_tier": "enterprise",
    "tone": "professional but friendly",
})

The separation means the template can be stored in a database, fetched at call time, and updated without a code deploy. Variables like company_name come from your application data, not from user input.

Variable sanitization for template interpolation. When any variable comes from user-controlled data (even indirectly), sanitize before injection:

import html

def sanitize_for_prompt(value: str, max_length: int = 200) -> str:
    # Remove control characters, escape HTML entities, enforce length
    value = re.sub(r"[\x00-\x1f\x7f]", "", value)
    value = html.escape(value)
    return value[:max_length]

Never trust that user-facing data is safe to inject into system prompts even if it came from your own database — a stored XSS attack can become a stored prompt injection.

Multi-turn conversation state. A conversation is a list of message dicts. The naive approach keeps the full list in memory and passes it every call. The problems:

1. The list grows without bound, eventually exceeding the context window
2. Early turns become stale and irrelevant
3. Old tokens still cost money even when they add no value

The right approach is a ConversationManager that enforces a token budget:

import tiktoken
from typing import TypedDict

enc = tiktoken.get_encoding("cl100k_base")

class Message(TypedDict):
    role: str
    content: str

class ConversationManager:
    def __init__(self, max_history_tokens: int = 4000) -> None:
        self._history: list[Message] = []
        self._max_tokens = max_history_tokens
        self._summary: str | None = None

    def add(self, role: str, content: str) -> None:
        self._history.append({"role": role, "content": content})

    def _token_count(self, messages: list[Message]) -> int:
        return sum(len(enc.encode(m["content"])) + 4 for m in messages)

    def get_history(self) -> list[Message]:
        """Return the most recent turns that fit within the token budget."""
        result: list[Message] = []
        budget = self._max_tokens

        # Walk backwards from most recent to oldest
        for msg in reversed(self._history):
            tokens = len(enc.encode(msg["content"])) + 4
            if tokens <= budget:
                result.insert(0, msg)
                budget -= tokens
            else:
                break  # stop adding older messages

        return result

    def get_context_block(self) -> str | None:
        """Return a summary of older context if overflow has occurred."""
        return self._summary

    def inject_summary(self, summary: str) -> None:
        """Replace dropped history with a summary injected into system prompt."""
        self._summary = summary
        # Trim history to fit budget after receiving summary
        while self._token_count(self._history) > self._max_tokens and self._history:
            self._history.pop(0)

    def is_approaching_limit(self, threshold: float = 0.8) -> bool:
        """Returns True when history is consuming > threshold of the budget."""
        return self._token_count(self._history) > self._max_tokens * threshold

Summarize-on-overflow pattern. When is_approaching_limit() is True, call the model to produce a compact summary of the conversation so far, then replace the oldest turns with the summary:

async def maybe_summarize_history(
    manager: ConversationManager,
    client: ProviderClient,
) -> None:
    if not manager.is_approaching_limit():
        return

    history = manager.get_history()
    summary_prompt = (
        "Summarize the following conversation in 3-5 bullet points, "
        "preserving key facts, decisions, and open questions:\n\n"
        + "\n".join(f"{m['role'].upper()}: {m['content']}" for m in history)
    )

    result = await client.generate(
        system="You are a concise conversation summarizer.",
        messages=[{"role": "user", "content": summary_prompt}],
        model="claude-haiku-4-5",  # use cheap model for summarization
        max_tokens=512,
    )

    manager.inject_summary(result.text)


# In your request handler:
async def handle_message(
    user_input: str,
    manager: ConversationManager,
    client: ProviderClient,
    template: PromptTemplate,
) -> str:
    # Summarize if needed
    await maybe_summarize_history(manager, client)

    # Build messages
    history = manager.get_history()
    context_block = manager.get_context_block()

    system = template.render_system({"company_name": "Acme", "customer_tier": "standard", "tone": "helpful"})
    if context_block:
        system += f"\n\n## Earlier conversation summary\n{context_block}"

    messages = [*history, {"role": "user", "content": user_input}]

    result = await client.generate(system=system, messages=messages, model="claude-haiku-4-5")

    # Update history
    manager.add("user", user_input)
    manager.add("assistant", result.text)

    return result.text

Multi-turn state storage. For production features, conversation history must survive server restarts and scale across multiple instances. Use Redis or a database:

import json
import redis.asyncio as redis

class PersistentConversationManager:
    def __init__(self, redis_client, session_id: str, ttl_seconds: int = 3600) -> None:
        self._redis = redis_client
        self._key = f"conv:{session_id}"
        self._ttl = ttl_seconds

    async def load(self) -> list[Message]:
        raw = await self._redis.get(self._key)
        return json.loads(raw) if raw else []

    async def save(self, messages: list[Message]) -> None:
        await self._redis.setex(self._key, self._ttl, json.dumps(messages))

    async def append(self, role: str, content: str) -> None:
        messages = await self.load()
        messages.append({"role": role, "content": content})
        await self.save(messages)

The TTL ensures idle conversations expire automatically without a background cleanup job.

Common mistakes

f-strings for prompt construction. f"You are helping {user.name}" inlines untrusted data directly into the system prompt without sanitization. Use a template class that validates variables and sanitizes values from user-controlled sources.

Unbounded conversation history. Passing the full history list forever will eventually hit the context window limit at the worst possible moment — mid-conversation with an important user. Enforce a budget proactively.

Saving full history to a JWT or cookie. Conversation history grows unboundedly. Storing it client-side means your token size will eventually exceed cookie limits and the history will be silently truncated. Use server-side storage with a session ID.

Using the expensive model for summarization. The summarization call during overflow is a compression task, not a reasoning task. Use the cheapest model available. Using GPT-4o or Claude Sonnet for summarization is 10–20x more expensive than necessary.

No version tracking for templates. When a prompt change degrades quality, you need to know which version was active and when. Store version identifiers with every request trace.

Try it yourself

1. Build a ConversationManager with a token budget of 3000 tokens. Feed it a synthetic conversation of 20 turns. Verify that get_history() returns only the most recent turns that fit within the budget.
2. Implement the maybe_summarize_history function and write a test that verifies it calls the model when history exceeds 80% of the budget, and does not call it when below that threshold.
3. Extend PromptTemplate with a from_file class method that loads template text and variable definitions from a YAML file. This is the first step toward database-backed prompt management.