LLM 提示快取：降低 90% API 成本（涵蓋三大供應商）

from openai import OpenAI

client = OpenAI()

# This system prompt is ~2,000 tokens -- well above the 1,024 minimum
SYSTEM_PROMPT = """You are a senior Python developer specializing in async programming.
You follow PEP 8, use type hints, and write comprehensive docstrings.
When reviewing code, check for: race conditions, resource leaks, error handling,
and performance bottlenecks. Always suggest specific fixes with code examples.
[... imagine 1,800 more tokens of coding guidelines, examples, and rules ...]"""

def chat(user_message: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
    )

    # Check if caching kicked in
    usage = response.usage
    cached = usage.prompt_tokens_details.cached_tokens
    total_input = usage.prompt_tokens
    print(f"Cached: {cached}/{total_input} tokens ({cached/total_input*100:.0f}%)")

    return response.choices[0].message.content

# First call: cache miss (full price)
chat("Review this async function for race conditions...")

# Second call within 5-10 min: cache hit (50% off on cached tokens)
chat("Now optimize the same function for throughput...")

import anthropic

client = anthropic.Anthropic()

SYSTEM_PROMPT = """You are a senior Python developer specializing in async programming.
You follow PEP 8, use type hints, and write comprehensive docstrings.
When reviewing code, check for: race conditions, resource leaks, error handling,
and performance bottlenecks. Always suggest specific fixes with code examples.
[... imagine 1,800 more tokens of coding guidelines, examples, and rules ...]"""

def chat(user_message: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},  # Explicit breakpoint
            }
        ],
        messages=[
            {"role": "user", "content": user_message},
        ],
    )

    # Read cache metrics from the response
    usage = response.usage
    created = usage.cache_creation_input_tokens
    read = usage.cache_read_input_tokens
    standard = usage.input_tokens
    print(f"Cache write: {created}, Cache read: {read}, Standard: {standard}")

    return response.content[0].text

# First call: cache_creation_input_tokens = ~1920 (write at 1.25x)
chat("Review this async function for race conditions...")

# Second call: cache_read_input_tokens = ~1920 (read at 0.1x -- 90% off!)
chat("Now optimize the same function for throughput...")

from google import genai
from google.genai import types

client = genai.Client()

SYSTEM_PROMPT = """You are a senior Python developer specializing in async programming.
You follow PEP 8, use type hints, and write comprehensive docstrings.
When reviewing code, check for: race conditions, resource leaks, error handling,
and performance bottlenecks. Always suggest specific fixes with code examples.
[... imagine 1,800 more tokens of coding guidelines, examples, and rules ...]"""

# Step 1: Create a named cache object
cache = client.caches.create(
    model="gemini-2.5-flash",
    config=types.CreateCachedContentConfig(
        display_name="python-review-guidelines",
        system_instruction=SYSTEM_PROMPT,
        ttl="3600s",  # 1 hour
    ),
)

print(f"Cache created: {cache.name}, expires: {cache.expire_time}")

# Step 2: Use the cache in requests
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="Review this async function for race conditions...",
    config=types.GenerateContentConfig(
        cached_content=cache.name,
    ),
)

# Check cache usage in the response
metadata = response.usage_metadata
print(f"Cached tokens: {metadata.cached_content_token_count}")
print(f"Total input tokens: {metadata.prompt_token_count}")

# --- OpenAI: Zero config, just call the API ---
from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},  # Cached automatically
        {"role": "user", "content": user_message},
    ],
)
cached = response.usage.prompt_tokens_details.cached_tokens

# --- Anthropic: Explicit cache_control breakpoint ---
import anthropic
client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": SYSTEM_PROMPT,
        "cache_control": {"type": "ephemeral"},  # Mark cache boundary
    }],
    messages=[{"role": "user", "content": user_message}],
)
cached = response.usage.cache_read_input_tokens

# --- Gemini: Named cache object ---
from google import genai
from google.genai import types

client = genai.Client()
cache = client.caches.create(
    model="gemini-2.5-flash",
    config=types.CreateCachedContentConfig(
        system_instruction=SYSTEM_PROMPT,
        ttl="3600s",
    ),
)
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=user_message,
    config=types.GenerateContentConfig(cached_content=cache.name),
)
cached = response.usage_metadata.cached_content_token_count

# BAD: Cache misses every single request
system_prompt = f"""You are a helpful assistant.
Current time: {datetime.now().isoformat()}
Always be helpful and accurate."""

# GOOD: Move the timestamp to the user message
system_prompt = """You are a helpful assistant.
Always be helpful and accurate."""

user_message = f"[Current time: {datetime.now().isoformat()}]\n{user_query}"

# BAD: Unique prefix per user = zero cache reuse
messages = [
    {"role": "system", "content": f"User ID: {user_id}\nPreferences: {prefs}\n{GUIDELINES}"},
    {"role": "user", "content": query},
]

# GOOD: Static content first, user context at the end
messages = [
    {"role": "system", "content": GUIDELINES},  # Same for all users -> cached
    {"role": "user", "content": f"Context: User {user_id}, prefs: {prefs}\n{query}"},
]

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": SYSTEM_PROMPT,
        "cache_control": {"type": "ephemeral"},
    }],
    messages=[{"role": "user", "content": "Explain Python's GIL..."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    # After streaming completes, check cache metrics
    usage = stream.get_final_message().usage
    print(f"\nCache read: {usage.cache_read_input_tokens} tokens")

import logging

logger = logging.getLogger("cache_monitor")

def log_cache_metrics(provider: str, usage: dict) -> float:
    """Extract and log cache metrics from any provider's response. Returns hit rate."""

    if provider == "openai":
        cached = getattr(usage.prompt_tokens_details, "cached_tokens", 0)
        total = usage.prompt_tokens
    elif provider == "anthropic":
        cached = usage.cache_read_input_tokens
        created = usage.cache_creation_input_tokens
        total = cached + created + usage.input_tokens
    elif provider == "gemini":
        cached = getattr(usage, "cached_content_token_count", 0)
        total = usage.prompt_token_count
    else:
        raise ValueError(f"Unknown provider: {provider}")

    hit_rate = (cached / total * 100) if total > 0 else 0
    logger.info(f"[{provider}] Cache hit rate: {hit_rate:.1f}% ({cached}/{total} tokens)")

    if hit_rate < 50:
        logger.warning(f"[{provider}] Low cache hit rate! Check prompt structure.")

    return hit_rate