LLM Context Window Strategies for Production: Summarization, Chunking, and RAG

import anthropic
 
client = anthropic.Anthropic()
 
class SlidingWindowConversation:
    def __init__(self, max_messages=20, summary_every=10):
        self.messages = []
        self.summary = ""
        self.max_messages = max_messages
        self.summary_every = summary_every
 
    def _summarize_old_messages(self):
        to_summarize = self.messages[:self.summary_every]
        self.messages = self.messages[self.summary_every:]
 
        messages_text = "\n".join(
            f"{m['role'].upper()}: {m['content']}" for m in to_summarize
        )
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=500,
            messages=[{"role": "user", "content": f"Summarize keeping key facts:\n\n{messages_text}"}]
        )
        self.summary = response.content[0].text
 
    def chat(self, user_message: str) -> str:
        self.messages.append({"role": "user", "content": user_message})
        if len(self.messages) > self.max_messages:
            self._summarize_old_messages()
 
        system = f"Previous summary:\n{self.summary}" if self.summary else ""
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1000,
            system=system,
            messages=self.messages
        )
        reply = response.content[0].text
        self.messages.append({"role": "assistant", "content": reply})
        return reply

def map_reduce_analysis(document: str, question: str) -> str:
    chunk_size = 50000
    chunks = [document[i:i+chunk_size] for i in range(0, len(document), chunk_size)]
 
    chunk_analyses = []
    for i, chunk in enumerate(chunks):
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=500,
            messages=[{"role": "user", "content": f"Analyze for: {question}\n\nSection {i+1}:\n{chunk}\n\nKey findings (2-3 bullet points):"}]
        )
        chunk_analyses.append(response.content[0].text)
 
    combined = "\n\n".join(chunk_analyses)
    final = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": f"Synthesize these into a final answer for: {question}\n\n{combined}"}]
    )
    return final.content[0].text

def analyze_with_cached_context(large_context: str, questions: list[str]) -> list[str]:
    system_messages = [
        {
            "type": "text",
            "text": f"You are analyzing this infrastructure:\n\n{large_context}",
            "cache_control": {"type": "ephemeral"}
        }
    ]
 
    answers = []
    for question in questions:
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=500,
            system=system_messages,
            messages=[{"role": "user", "content": question}]
        )
        answers.append(response.content[0].text)
    return answers

def count_tokens(text: str) -> int:
    response = client.messages.count_tokens(
        model="claude-sonnet-5",
        messages=[{"role": "user", "content": text}]
    )
    return response.input_tokens