LLM Context Engineering

Real-World Scenario

A customer support LLM handles queries about a 500-page technical manual. Stuffing the full manual into the context is expensive and causes the “lost in the middle” problem. Context engineering solves this: retrieve the 3 most relevant sections, format them clearly, inject them in the right order, and give the model explicit instructions about how to use them.

Structuring the System Prompt

import anthropic
from datetime import datetime

client = anthropic.Anthropic()

def build_system_prompt(
    persona: str,
    capabilities: list[str],
    constraints: list[str],
    knowledge_context: str = "",
    output_format: str = "",
) -> str:
    """Build a well-structured system prompt."""
    sections = [f"# Role\n{persona}"]

    if capabilities:
        cap_list = "\n".join(f"- {c}" for c in capabilities)
        sections.append(f"# Capabilities\n{cap_list}")

    if constraints:
        con_list = "\n".join(f"- {c}" for c in constraints)
        sections.append(f"# Constraints\n{con_list}")

    if knowledge_context:
        sections.append(f"# Context\n{knowledge_context}")

    if output_format:
        sections.append(f"# Output Format\n{output_format}")

    sections.append(f"# Current Date\n{datetime.now().strftime('%Y-%m-%d')}")

    return "\n\n".join(sections)


# Example: support agent system prompt
system = build_system_prompt(
    persona="You are a technical support specialist for CloudDB, an enterprise database product.",
    capabilities=[
        "Answer questions about CloudDB configuration and usage",
        "Diagnose connection and performance issues",
        "Guide users through step-by-step troubleshooting",
    ],
    constraints=[
        "Only provide information specific to CloudDB — do not compare to competitors",
        "Escalate billing questions to the sales team ([email protected])",
        "Never share internal pricing tiers or discount structures",
    ],
    knowledge_context="CloudDB version 4.2 was released on 2025-03-01. Major changes: new indexing engine, deprecated XML support.",
    output_format="Use numbered steps for procedures. Use code blocks for configuration. Be concise.",
)

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    system=system,
    messages=[{"role": "user", "content": "How do I optimize slow queries in CloudDB?"}]
)
print(response.content[0].text)

Managing Long Conversation History

import anthropic
from dataclasses import dataclass, field

client = anthropic.Anthropic()

@dataclass
class ContextManager:
    """Manages context window usage with summarization and trimming."""
    max_context_tokens: int = 100_000   # leave headroom below model limit
    messages: list[dict] = field(default_factory=list)
    summaries: list[str] = field(default_factory=list)

    def estimate_tokens(self, messages: list[dict]) -> int:
        """Rough token estimate: 4 characters ≈ 1 token."""
        total_chars = sum(len(str(m.get("content", ""))) for m in messages)
        return total_chars // 4

    def add_message(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})
        self._trim_if_needed()

    def _trim_if_needed(self) -> None:
        """Summarize oldest messages when approaching context limit."""
        if self.estimate_tokens(self.messages) < self.max_context_tokens * 0.8:
            return

        # Keep the most recent 10 messages, summarize the rest
        to_summarize = self.messages[:-10]
        self.messages = self.messages[-10:]

        if not to_summarize:
            return

        history = "\n".join(f"{m['role']}: {m['content']}" for m in to_summarize)
        resp = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{
                "role": "user",
                "content": f"Summarize this conversation, preserving key facts and decisions:\n\n{history}"
            }]
        )
        summary = resp.content[0].text
        self.summaries.append(summary)

    def get_messages(self) -> list[dict]:
        """Return messages with summary prepended if available."""
        if not self.summaries:
            return self.messages

        summary_msg = {
            "role": "user",
            "content": f"[Conversation history summary: {' '.join(self.summaries)}]"
        }
        ack_msg = {"role": "assistant", "content": "I have the context from our previous discussion."}
        return [summary_msg, ack_msg] + self.messages

    def token_usage(self) -> dict:
        tokens = self.estimate_tokens(self.messages)
        return {
            "current_tokens": tokens,
            "max_tokens":     self.max_context_tokens,
            "utilization":    f"{tokens / self.max_context_tokens:.1%}",
            "summaries":      len(self.summaries),
        }

Injecting Retrieved Context Effectively

import anthropic
from typing import Literal

client = anthropic.Anthropic()

def format_retrieved_context(
    chunks: list[dict],
    strategy: Literal["numbered", "xml", "markdown"] = "xml"
) -> str:
    """Format retrieved chunks for injection into the prompt."""

    if strategy == "xml":
        # XML tags help the model locate specific sources
        parts = ["<retrieved_context>"]
        for i, chunk in enumerate(chunks, 1):
            parts.append(f"""<source id="{i}" title="{chunk['title']}" relevance="{chunk.get('score', 0):.2f}">
{chunk['content']}
</source>""")
        parts.append("</retrieved_context>")
        return "\n".join(parts)

    elif strategy == "numbered":
        parts = []
        for i, chunk in enumerate(chunks, 1):
            parts.append(f"[{i}] {chunk['title']}\n{chunk['content']}")
        return "\n\n".join(parts)

    else:  # markdown
        parts = []
        for chunk in chunks:
            parts.append(f"### {chunk['title']}\n{chunk['content']}")
        return "\n\n".join(parts)


def rag_with_context_engineering(
    question: str,
    retrieved_chunks: list[dict],
    strategy: str = "xml",
) -> str:
    context = format_retrieved_context(retrieved_chunks, strategy=strategy)

    system = """You are a precise technical assistant.

Rules for using the retrieved context:
1. Answer ONLY from the provided sources
2. Cite the source ID [1], [2], etc. for each claim
3. If sources conflict, note the discrepancy
4. If the answer is not in the sources, say "This information is not in the provided documentation"
5. Never speculate or add information beyond what's provided"""

    # Critical ordering: question first, then context (reduces "lost in the middle")
    user_message = f"""Question: {question}

{context}

Please answer the question using only the sources above."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": user_message}]
    )
    return response.content[0].text


# Test
chunks = [
    {"title": "Installation Guide", "content": "Run pip install clouddb>=4.0. Python 3.10+ required.", "score": 0.92},
    {"title": "Configuration",      "content": "Set DB_HOST and DB_PORT env vars. Default port is 5432.", "score": 0.87},
    {"title": "Performance Tuning", "content": "Enable connection pooling with POOL_SIZE=10 for production.", "score": 0.71},
]

answer = rag_with_context_engineering(
    question="How do I install and configure CloudDB for production?",
    retrieved_chunks=chunks,
)
print(answer)

Document Chunking Strategies

import re
from typing import Iterator

def chunk_by_token_count(
    text: str,
    max_tokens: int = 512,
    overlap_tokens: int = 64,
) -> list[str]:
    """Split text by approximate token count with overlap."""
    words = text.split()
    # Approximate: 0.75 words per token
    words_per_chunk = int(max_tokens * 0.75)
    overlap_words   = int(overlap_tokens * 0.75)
    
    chunks = []
    start = 0
    while start < len(words):
        end = min(start + words_per_chunk, len(words))
        chunks.append(" ".join(words[start:end]))
        start += words_per_chunk - overlap_words
    
    return chunks


def chunk_by_markdown_headers(text: str) -> list[dict]:
    """Split markdown document at header boundaries — preserves section integrity."""
    header_pattern = re.compile(r'^(#{1,3})\s+(.+)$', re.MULTILINE)
    matches = list(header_pattern.finditer(text))
    
    if not matches:
        return [{"title": "Content", "content": text}]
    
    chunks = []
    for i, match in enumerate(matches):
        title = match.group(2).strip()
        start = match.end()
        end   = matches[i + 1].start() if i + 1 < len(matches) else len(text)
        content = text[start:end].strip()
        if content:
            chunks.append({"title": title, "content": content, "level": len(match.group(1))})
    
    return chunks


def chunk_by_sentences(text: str, max_sentences: int = 5, overlap: int = 1) -> list[str]:
    """Split text by sentences with overlap for dense content."""
    sentences = re.split(r'(?<=[.!?])\s+', text)
    chunks = []
    for i in range(0, len(sentences), max_sentences - overlap):
        chunk = " ".join(sentences[i:i + max_sentences])
        if chunk:
            chunks.append(chunk)
    return chunks


# Demonstrate chunking strategies
sample_doc = """# Introduction
CloudDB is a high-performance database system designed for cloud-native applications.
It supports both OLTP and OLAP workloads with automatic scaling.

## Installation
Install using pip: `pip install clouddb`. Requires Python 3.10 or later.
Set the DATABASE_URL environment variable before starting the server.

## Configuration
Connection pooling is enabled by default with a pool size of 5.
For production workloads, increase POOL_SIZE to 20 and enable read replicas.
"""

header_chunks = chunk_by_markdown_headers(sample_doc)
print("Header-based chunks:")
for chunk in header_chunks:
    print(f"  [{chunk['title']}]: {len(chunk['content'])} chars")