Test chunk

ol": {"type": "ephemeral"} } ] for doc in retrieved_documents: system.append({ "type": "text", "text": doc, "cache_control": {"type": "ephemeral"} }) messages = history_messages + [{"role": "user", "content": user_query}] return client.messages.create( model="claude-opus-4", system=system, messages=messages, max_tokens=2048 )

3.2 Cache TTL and Warming Strategies

class PromptCacheManager:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.cache_ttl = 300

    async def warm_cache_for_popular_queries(self, top_queries: list):
        for query in top_queries:
            await self.execute_warmup_request(query)

    def get_cache_metrics(self):
        return {
            "cache_hit_rate": self.calculate_hit_rate(),
            "cost_savings": self.estimated_savings_from_cache(),
        }

3.3 Input Token Truncation and Compression

class ContextCompressor:
    async def compress(self, query, documents, budget=8000, strategy='tiered'):
        if strategy == 'tiered':
            return await self._tiered_compression(query, documents, budget)

    async def _tiered_compression(self, query, docs, budget):
        scores = [(doc, await self.relevance_score(query, doc)) for doc in docs]
        scores.sort(key=lambda x: x[1], reverse=True)

        result = []
        remaining_budget = budget

        for i, (doc, score) in enumerate(scores):
            doc_tokens = self.count_tokens(doc)
            if i == 0 or score > 0.8:
                if doc_tokens <= remaining_budget:
                    result.append(doc)
                    remaining_budget -= doc_tokens
            elif score > 0.5:
                if remaining_budget > 200:
                    summary = await self.summarize(doc, max_tok