Prompt Caching — Tiết kiệm 90% chi phí với cache thông minh

Bạn đang build ứng dụng với system prompt dài 5.000 token, hoặc đưa cả cuốn sách vào context? Mỗi lần gọi API, bạn phải trả tiền cho tất cả tokens đó — dù Claude đã "đọc" chúng hàng trăm lần trước. Prompt Caching giải quyết vấn đề này: Anthropic cache phần context tốn kém, bạn chỉ trả 10% chi phí cho cache hits.

Cơ chế hoạt động

Khi bật caching, Claude lưu trữ prefix của prompt (thường là system prompt và context dài) trên server. Các request tiếp theo reuse cache thay vì reprocess toàn bộ.

Cache tồn tại trong 5 phút (có thể gia hạn bằng cách access). Phù hợp nhất cho:

• System prompts dài (hàng nghìn tokens)
• Tài liệu tham khảo lớn (FAQ, documentation, sách)
• Few-shot examples phức tạp
• Multi-turn conversations với context dài

Demo với Pride and Prejudice (187k tokens)

Đây là benchmark từ Anthropic Cookbook chính thức — đưa toàn bộ cuốn sách vào context và đặt câu hỏi:

import anthropic
import time

client = anthropic.Anthropic()

# Load toàn bộ cuốn sách (tải từ Project Gutenberg)
with open("pride_and_prejudice.txt", "r", encoding="utf-8") as f:
    book_text = f.read()

print(f"Book length: {len(book_text):,} characters")

def test_no_cache(question: str):
    """Request không có cache."""
    start = time.time()

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=300,
        system=f"You are a literary expert. Here is the full text of Pride and Prejudice:

{book_text}",
        messages=[{"role": "user", "content": question}],
    )

    elapsed = time.time() - start
    return {
        "answer": response.content[0].text,
        "time": elapsed,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
        "cache_read": 0,
        "cache_write": 0,
    }

def test_with_cache(question: str):
    """Request với explicit cache breakpoint."""
    start = time.time()

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=300,
        system=[
            {
                "type": "text",
                "text": "You are a literary expert. Here is the full text of Pride and Prejudice:",
            },
            {
                "type": "text",
                "text": book_text,
                # Đặt cache breakpoint tại đây
                "cache_control": {"type": "ephemeral"},
            },
        ],
        messages=[{"role": "user", "content": question}],
    )

    elapsed = time.time() - start
    usage = response.usage

    return {
        "answer": response.content[0].text,
        "time": elapsed,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "cache_read": getattr(usage, "cache_read_input_tokens", 0),
        "cache_write": getattr(usage, "cache_creation_input_tokens", 0),
    }

# Benchmark
question = "Mô tả tính cách của Elizabeth Bennet và mối quan hệ của cô với Mr. Darcy."

print("=== Test 1: Không cache ===")
r1 = test_no_cache(question)
print(f"Time: {r1['time']:.2f}s | Input tokens: {r1['input_tokens']:,}")
print(f"Answer: {r1['answer'][:100]}...")

print("
=== Test 2: Cache Write (lần đầu) ===")
r2 = test_with_cache(question)
print(f"Time: {r2['time']:.2f}s | Cache write: {r2['cache_write']:,} | Input: {r2['input_tokens']:,}")

print("
=== Test 3: Cache Hit (lần tiếp) ===")
r3 = test_with_cache("Mr. Darcy thay đổi như thế nào trong suốt câu chuyện?")
print(f"Time: {r3['time']:.2f}s | Cache read: {r3['cache_read']:,} | Input: {r3['input_tokens']:,}")

# So sánh
speedup = r1['time'] / r3['time']
cost_reduction = 1 - (r3['input_tokens'] / r1['input_tokens'])
print(f"
Kết quả: {speedup:.1f}x nhanh hơn, giảm {cost_reduction:.0%} chi phí input tokens")

Kết quả thực tế từ benchmark

Từ lần thứ 3 trở đi, mỗi request chỉ tốn ~11% so với không cache. Break-even point: chỉ cần 2 requests là đã có lời!

Multi-turn Conversation Caching

Đây là ứng dụng thực tế nhất — cache lịch sử conversation:

def chat_with_caching(conversation_history: list, new_message: str,
                      system_prompt: str = "") -> dict:
    """
    Multi-turn chat với caching trên conversation history.
    Cache toàn bộ history, chỉ gửi message mới.
    """

    # Build messages với cache trên lịch sử
    messages = []

    for i, msg in enumerate(conversation_history):
        if i == len(conversation_history) - 1:
            # Đặt cache breakpoint ở message cuối cùng trong history
            messages.append({
                "role": msg["role"],
                "content": [
                    {
                        "type": "text",
                        "text": msg["content"],
                        "cache_control": {"type": "ephemeral"},
                    }
                ]
            })
        else:
            messages.append(msg)

    # Thêm message mới
    messages.append({"role": "user", "content": new_message})

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=500,
        system=system_prompt,
        messages=messages,
    )

    usage = response.usage
    answer = response.content[0].text

    # Cập nhật lịch sử
    conversation_history.append({"role": "user", "content": new_message})
    conversation_history.append({"role": "assistant", "content": answer})

    return {
        "answer": answer,
        "cache_read": getattr(usage, "cache_read_input_tokens", 0),
        "cache_write": getattr(usage, "cache_creation_input_tokens", 0),
        "regular_input": usage.input_tokens,
    }

# Ví dụ: Chatbot phân tích tài liệu
system = "Bạn là trợ lý phân tích tài liệu. Trả lời ngắn gọn, chính xác."
history = []

# Turn 1
r = chat_with_caching(history, "Xin chào, tôi cần phân tích hợp đồng.", system)
print(f"Turn 1 | Cache read: {r['cache_read']} | Answer: {r['answer'][:60]}...")

# Turn 2 (cache history từ turn 1)
r = chat_with_caching(history, "Các điều khoản nào cần chú ý nhất?", system)
print(f"Turn 2 | Cache read: {r['cache_read']} | Answer: {r['answer'][:60]}...")

# Turn 3 (cache history từ turn 1+2)
r = chat_with_caching(history, "Rủi ro pháp lý là gì?", system)
print(f"Turn 3 | Cache read: {r['cache_read']} | Answer: {r['answer'][:60]}...")

Automatic Caching (claude-3-7 và mới hơn)

Với các model mới nhất, bạn có thể bật automatic caching mà không cần đặt cache_control thủ công:

# Automatic caching — Anthropic tự quyết định cache gì
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1000,
    # Không cần cache_control — model tự xử lý
    system="System prompt rất dài của bạn...",
    messages=[{"role": "user", "content": "Câu hỏi ngắn"}],
    # Bật extended thinking để tận dụng tối đa caching
)

# Kiểm tra cache được dùng không
usage = response.usage
if hasattr(usage, "cache_read_input_tokens") and usage.cache_read_input_tokens > 0:
    print(f"Cache hit! Đọc {usage.cache_read_input_tokens:,} tokens từ cache")
else:
    print("Cache miss (hoặc lần đầu)")

Chiến lược tối ưu caching

1. Đặt cache breakpoint đúng chỗ

Cache breakpoint nên đặt sau phần ít thay đổi nhất:

# TỐT: Cache system prompt + context tĩnh
messages = [
    {
        "role": "user",
        "content": [
            {"type": "text", "text": "SYSTEM: " + static_instructions},
            {"type": "text", "text": "CONTEXT: " + large_document,
             "cache_control": {"type": "ephemeral"}},  # Breakpoint ở đây
            {"type": "text", "text": "QUESTION: " + dynamic_user_question},
        ]
    }
]

# KHÔNG TỐT: Cache ở giữa phần thay đổi
# Cache sẽ miss mỗi lần vì content sau breakpoint thay đổi

2. Batch requests trong 5 phút

import threading

class CachingBatcher:
    """Gom nhiều requests trong window 5 phút để tối đa hóa cache hits."""

    def __init__(self, shared_context: str, window_seconds: int = 240):
        self.context = shared_context
        self.window = window_seconds
        self._last_request = None
        self._lock = threading.Lock()

    def should_refresh_cache(self) -> bool:
        if self._last_request is None:
            return True
        return (time.time() - self._last_request) > self.window

    def query(self, question: str) -> str:
        with self._lock:
            self._last_request = time.time()

        response = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": self.context,
                        "cache_control": {"type": "ephemeral"},
                    },
                    {"type": "text", "text": question},
                ]
            }],
        )
        return response.content[0].text

Tính ROI của Prompt Caching

Ví dụ thực tế: Chatbot với system prompt 10.000 tokens, 1.000 users/ngày, mỗi user hỏi 5 câu:

• Không cache: 1.000 × 5 × 10.000 = 50M tokens/ngày
• Với cache: 1.000 × 10.000 (write) + 4.000 × 1.000 (read, 10%) = 10M + 4M = 14M tokens hiệu quả
• Tiết kiệm: ~72% chi phí input tokens từ ngày thứ 2 trở đi

Prompt Caching là một trong những optimizations ROI cao nhất bạn có thể làm cho ứng dụng Claude. Kết hợp với Batch Processing để tối ưu hoàn toàn chi phí vận hành.