Speculative Caching — Giảm time-to-first-token với cache dự đoán

Trong các ứng dụng conversational AI, time-to-first-token (TTFT) là metric quan trọng nhất ảnh hưởng đến user experience. Khi user gửi message, họ không muốn nhìn màn hình trống 3-5 giây. Speculative Caching là kỹ thuật pre-warm cache với context được dự đoán — để khi request thực sự đến, cache đã sẵn sàng.

Prompt Caching — Nền tảng cần hiểu trước

Anthropic Prompt Caching cho phép bạn cache phần đầu của prompt (system prompt + documents) và tái sử dụng ở các requests tiếp theo:

import anthropic

client = anthropic.Anthropic()

# Không có caching — mỗi request trả tiền đầy đủ input tokens
response_no_cache = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1000,
    system="You are a helpful assistant with extensive knowledge of Vietnamese law...",  # 5000 tokens
    messages=[{"role": "user", "content": "Giải thích Điều 32 Bộ luật Dân sự"}]
)

# Với caching — system prompt được cache sau lần đầu
response_with_cache = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1000,
    system=[{
        "type": "text",
        "text": "You are a helpful assistant with extensive knowledge of Vietnamese law...",
        "cache_control": {"type": "ephemeral"}  # Mark for caching
    }],
    messages=[{"role": "user", "content": "Giải thích Điều 32 Bộ luật Dân sự"}]
)

# Lần 2 trở đi: system prompt tokens được đọc từ cache
# Cost giảm 90% cho cached tokens
# TTFT giảm vì không cần process lại system prompt

Cache tồn tại tối thiểu 5 phút và được extend mỗi khi hit. Đây là foundation của Speculative Caching.

Speculative Caching là gì?

Standard caching reactive: cache được warm khi user thực sự gọi API. Speculative Caching proactive: bạn dự đoán user sẽ cần gì và warm cache trước khi họ hỏi.

Ví dụ: User đang xem product page cho "iPhone 15 Pro". Khả năng cao họ sẽ hỏi về specs, giá, hoặc so sánh. Bạn có thể pre-warm cache với product context ngay khi họ load page — trước khi họ gõ bất cứ điều gì.

Pattern 1: Page-Based Prefetching

import anthropic
import asyncio
import time
from typing import Optional

client = anthropic.Anthropic()
async_client = anthropic.AsyncAnthropic()

class SpeculativeCacheManager:
    def __init__(self):
        self.cache_registry = {}  # track what's been pre-warmed

    def warm_cache_for_context(self, context_id: str, context_content: str,
                                system_role: str = "helpful assistant") -> dict:
        """
        Pre-warm cache với một context cụ thể.
        Gọi hàm này ngay khi user bắt đầu interact với một page/resource.
        """
        if context_id in self.cache_registry:
            age_seconds = time.time() - self.cache_registry[context_id]["warmed_at"]
            if age_seconds < 240:  # Cache still warm (5 min TTL - 1 min buffer)
                return {"status": "already_warm", "age_seconds": int(age_seconds)}

        # Gửi request nhỏ để warm cache
        start = time.time()
        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=1,  # Minimal output — chỉ cần warm cache
            system=[
                {
                    "type": "text",
                    "text": f"You are a {system_role}.",
                    "cache_control": {"type": "ephemeral"}
                },
                {
                    "type": "text",
                    "text": context_content,
                    "cache_control": {"type": "ephemeral"}
                }
            ],
            messages=[{"role": "user", "content": "Ready."}]
        )

        warm_time = int((time.time() - start) * 1000)
        cache_tokens = getattr(response.usage, 'cache_creation_input_tokens', 0) or 0

        self.cache_registry[context_id] = {
            "warmed_at": time.time(),
            "cache_tokens": cache_tokens,
            "warm_time_ms": warm_time
        }

        print(f"Cache warmed: {context_id} | {cache_tokens} tokens | {warm_time}ms")
        return {"status": "warmed", "cache_tokens": cache_tokens, "warm_time_ms": warm_time}

    def query_with_cache(self, context_id: str, context_content: str,
                          system_role: str, user_message: str) -> dict:
        """Query Claude using pre-warmed cache"""
        start = time.time()

        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=2000,
            system=[
                {
                    "type": "text",
                    "text": f"You are a {system_role}.",
                    "cache_control": {"type": "ephemeral"}
                },
                {
                    "type": "text",
                    "text": context_content,
                    "cache_control": {"type": "ephemeral"}
                }
            ],
            messages=[{"role": "user", "content": user_message}]
        )

        latency_ms = int((time.time() - start) * 1000)
        cache_read = getattr(response.usage, 'cache_read_input_tokens', 0) or 0
        cache_written = getattr(response.usage, 'cache_creation_input_tokens', 0) or 0
        cache_hit = cache_read > 0

        return {
            "response": response.content[0].text,
            "latency_ms": latency_ms,
            "cache_hit": cache_hit,
            "cache_read_tokens": cache_read,
            "cache_write_tokens": cache_written,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens
        }

cache_manager = SpeculativeCacheManager()

Pattern 2: User Journey Prediction

class JourneyBasedSpeculator:
    """
    Dự đoán next context dựa trên user journey patterns.
    Pre-warm cache cho predicted next step.
    """

    # Định nghĩa journey maps: current_page -> likely_next_pages
    JOURNEY_GRAPH = {
        "product_listing": ["product_detail", "category_filter"],
        "product_detail": ["checkout", "comparison", "reviews"],
        "checkout_cart": ["checkout_shipping", "product_detail"],
        "checkout_shipping": ["checkout_payment"],
        "checkout_payment": ["order_confirmation"],
        "user_profile": ["order_history", "settings", "wishlist"],
        "order_history": ["order_detail", "return_request"],
    }

    def __init__(self, content_loader: callable):
        self.content_loader = content_loader  # Function to load page content
        self.cache_manager = SpeculativeCacheManager()
        self.prefetch_tasks = {}

    async def prefetch_predicted_contexts(self, current_page: str, current_entity_id: str):
        """Async prefetch predicted next pages"""
        predicted_pages = self.JOURNEY_GRAPH.get(current_page, [])

        async def warm_single(page: str):
            content = await self.async_load_content(page, current_entity_id)
            if content:
                return await self.async_warm_cache(f"{page}_{current_entity_id}", content)

        tasks = [warm_single(page) for page in predicted_pages]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        print(f"Pre-warmed {len([r for r in results if not isinstance(r, Exception)])} contexts for '{current_page}' journey")
        return results

    async def async_warm_cache(self, context_id: str, content: str):
        """Async version của cache warming"""
        response = await async_client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=1,
            system=[{
                "type": "text",
                "text": content,
                "cache_control": {"type": "ephemeral"}
            }],
            messages=[{"role": "user", "content": "Ready."}]
        )
        return context_id

    async def async_load_content(self, page: str, entity_id: str) -> Optional[str]:
        """Load content cho một page — implement với real data source"""
        # Mock implementation
        content_map = {
            "product_detail": f"Product ID: {entity_id}
Name: Sample Product
Price: $99
Specs: ...",
            "checkout": f"Cart for user: {entity_id}
Items: 3
Total: $297",
            "reviews": f"Reviews for product {entity_id}
5 stars: 80%
4 stars: 15%
Latest: ..."
        }
        return content_map.get(page)

Pattern 3: Time-Based Refresh

class AutoRefreshCacheManager:
    """
    Tự động refresh cache cho contexts có high access frequency.
    Đảm bảo cache luôn warm cho popular content.
    """

    def __init__(self, refresh_interval: int = 240):  # 4 minutes (before 5-min TTL)
        self.contexts = {}  # context_id -> {content, system_role, last_refresh}
        self.refresh_interval = refresh_interval
        self.running = False

    def register_hot_context(self, context_id: str, content: str, system_role: str = "assistant"):
        """Đăng ký một context để auto-refresh"""
        self.contexts[context_id] = {
            "content": content,
            "system_role": system_role,
            "last_refresh": 0,
            "hit_count": 0
        }
        # Warm immediately
        self._warm_cache(context_id)

    def _warm_cache(self, context_id: str):
        ctx = self.contexts[context_id]
        client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=1,
            system=[{
                "type": "text",
                "text": ctx["content"],
                "cache_control": {"type": "ephemeral"}
            }],
            messages=[{"role": "user", "content": "Ready."}]
        )
        self.contexts[context_id]["last_refresh"] = time.time()

    async def run_refresh_loop(self):
        """Background loop để refresh expiring caches"""
        self.running = True
        while self.running:
            now = time.time()
            for context_id, ctx in self.contexts.items():
                age = now - ctx["last_refresh"]
                if age >= self.refresh_interval:
                    print(f"Auto-refreshing cache: {context_id}")
                    self._warm_cache(context_id)

            await asyncio.sleep(60)  # Check every minute

    def stop(self):
        self.running = False

Đo lường hiệu quả Speculative Caching

def benchmark_speculative_caching(context_content: str, test_queries: list[str]) -> dict:
    """So sánh latency: cold cache vs warm cache"""

    manager = SpeculativeCacheManager()

    print("Benchmarking Speculative Caching...")
    print("=" * 50)

    # Test 1: Cold cache (no pre-warming)
    cold_latencies = []
    for query in test_queries[:3]:
        start = time.time()
        result = manager.query_with_cache(
            f"cold_{time.time()}", context_content, "assistant", query
        )
        cold_latencies.append(result["latency_ms"])
        print(f"Cold: {result['latency_ms']}ms | Cache hit: {result['cache_hit']}")

    # Test 2: Pre-warm then query
    warm_latencies = []
    ctx_id = "benchmark_warm"

    # Pre-warm
    warm_result = manager.warm_cache_for_context(ctx_id, context_content)
    print(f"
Pre-warm: {warm_result['warm_time_ms']}ms | {warm_result['cache_tokens']} tokens cached")

    # Now query with warm cache
    for query in test_queries[:3]:
        start = time.time()
        result = manager.query_with_cache(ctx_id, context_content, "assistant", query)
        warm_latencies.append(result["latency_ms"])
        print(f"Warm: {result['latency_ms']}ms | Cache hit: {result['cache_hit']}")

    avg_cold = sum(cold_latencies) / len(cold_latencies) if cold_latencies else 0
    avg_warm = sum(warm_latencies) / len(warm_latencies) if warm_latencies else 0
    improvement = ((avg_cold - avg_warm) / avg_cold * 100) if avg_cold > 0 else 0

    print(f"
Results:")
    print(f"  Avg Cold TTFT: {avg_cold:.0f}ms")
    print(f"  Avg Warm TTFT: {avg_warm:.0f}ms")
    print(f"  TTFT Improvement: {improvement:.1f}%")

    return {
        "avg_cold_ms": avg_cold,
        "avg_warm_ms": avg_warm,
        "ttft_improvement_pct": improvement
    }

# Benchmark
test_context = "Vietnamese Legal Code Content:
" + "Article text... " * 100  # Long context
test_queries = [
    "Điều 32 quy định gì về quyền cá nhân?",
    "Hình phạt cho vi phạm Điều 15?",
    "Giải thích khái niệm 'pháp nhân' theo bộ luật"
]

benchmark_results = benchmark_speculative_caching(test_context, test_queries)

Chi phí và ROI của Speculative Caching

Với bất kỳ context nào được hit 2+ lần, Speculative Caching tiết kiệm cả chi phí lẫn latency.

Khi nào dùng Speculative Caching

• Documentation chatbots — Pre-warm khi user mở doc page
• Customer support — Pre-warm product/order context khi user login
• E-commerce — Pre-warm product catalog khi user browse category
• Legal/Medical AI — Pre-warm case/patient file khi practitioner opens it
• Code assistants — Pre-warm codebase context khi dev opens project

Tổng kết

Speculative Caching là kỹ thuật high-impact, low-complexity. Bằng cách dự đoán user journey và pre-warm cache, bạn giảm TTFT từ giây xuống còn milliseconds — cải thiện UX đáng kể mà không cần thay đổi model hay infrastructure.

Key implementation steps:

• Add cache_control: {type: "ephemeral"} vào system prompt và documents dài
• Gọi warm cache API ngay khi user bắt đầu session hoặc navigate đến page
• Refresh cache mỗi 4 phút cho hot contexts để tránh expiry
• Monitor cache hit rate để đánh giá prediction accuracy

Đọc thêm: Usage & Cost API để theo dõi cache hit rates và tối ưu chiến lược caching.

Bài viết liên quan

• Contextual Retrieval — Nâng cấp RAG với embeddings ngữ cảnh
• Best Practices cho Vision — Tối ưu hình ảnh gửi Claude
• Agent Workflows — Chaining, Routing, Parallelization
• Context Engineering — Nghệ thuật quản lý context cho Claude
• Claude cho Engineering: Standup và báo cáo tiến độ