Cache Invalidation: Avoiding Stale Cache When Updating System Prompt, Tools, FAQ

Sorun#

T=0  : SYSTEM_PROMPT_V1 deploy edildi, cache yazılıyor
T=1  : 1000 cache hit, harika
T=2  : Bug fix gerekli — SYSTEM_PROMPT_V2 deploy
T=3  : Cache hash değişti — tüm istekler cache miss, write yapıyor
T=4  : 1 saat boyunca %120 maliyet artışı (cache write storm)

Pattern 1 — Dual Write (Aşamalı Geçiş)#

ROLLOUT_PCT = int(os.environ.get("V2_ROLLOUT_PCT", "0"))

def get_system_prompt(user_id: str):
    if hash(user_id) % 100 < ROLLOUT_PCT:
        return SYSTEM_PROMPT_V2  # yeni
    return SYSTEM_PROMPT_V1  # eski (cache'lenmiş)

Rollout schedule#

Pattern 2 — Prompt Versioning#

# prompts/system_v1.txt
# prompts/system_v2.txt
# prompts/system_v3.txt

class PromptManager:
    def __init__(self):
        self.versions = {
            "v1": open("prompts/system_v1.txt").read(),
            "v2": open("prompts/system_v2.txt").read(),
            "v3": open("prompts/system_v3.txt").read(),
        }
        self.current = "v3"

    def get(self, version=None):
        return self.versions[version or self.current]

pm = PromptManager()

# Cache hit-rate metric'ı per-version
def chat(query, user_id):
    version = "v3" if hash(user_id) % 100 < 80 else "v2"  # gradual
    sys = pm.get(version)
    response = anthropic.messages.create(
        system=[{"type": "text", "text": sys, "cache_control": {"type": "ephemeral"}}],
        ...
    )
    log_metric(f"cache_hit_{version}", ...)
    return response

Langfuse Prompt Management#

from langfuse import Langfuse
langfuse = Langfuse()

prompt = langfuse.get_prompt("system-assistant", version=3, label="production")
SYSTEM_V3 = prompt.compile()

Pattern 3 — Emergency Invalidation#

Strateji#

1. Eski cache'in TTL'i biter (5dk veya 1h içinde)
2. Bu sırada maliyet sıçramasını kabul et (kısa süreli)
3. Yeni cache hızla ısınır

# Deploy yeni SYSTEM_V_FIXED
# Sleep 0 sec — tüm trafik anında yeni
# Cache write storm 5-60dk sürer, sonra normal

Pre-warm ile yumuşat#

# warmup_after_fix.py
async def warmup():
    SYSTEM = open("prompts/system_v_fixed.txt").read()
    for i in range(5):
        await anthropic.messages.create(
            model="claude-sonnet-4-6",
            system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
            messages=[{"role": "user", "content": "test"}],
            max_tokens=5,
        )

# Deploy:
# 1. Run warmup.py (yeni cache yazıldı)
# 2. Flip feature flag/env to V_FIXED
# 3. Production trafiği immediately cache hit yapıyor

Knowledge Base Update Pattern#

KB_VERSION = open("kb/version.txt").read().strip()  # "2026-05-13"

def get_kb_for_today():
    """KB'yi günlük yenile, cache'i otomatik refresh."""
    today = date.today().isoformat()
    if today != KB_VERSION:
        # Rebuild KB
        new_kb = build_kb_from_source()
        open("kb/system.txt", "w").write(new_kb)
        open("kb/version.txt", "w").write(today)

def chat(query):
    kb = open("kb/system.txt").read()
    response = anthropic.messages.create(
        system=[
            {"type": "text", "text": SYSTEM,
             "cache_control": {"type": "ephemeral"}},
            {"type": "text", "text": kb,
             "cache_control": {"type": "ephemeral"}},  # ← gün içi sabit
        ],
        ...
    )

Monitoring — Cache Update Sırasında Ne İzle?#

1. Cache write rate#

SELECT
    toStartOfMinute(ts) AS minute,
    sum(cache_creation_input_tokens) AS write_tokens,
    count() AS requests_with_cache_write
FROM llm_telemetry.requests
WHERE ts >= now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute

2. Cost per minute spike#

SELECT
    toStartOfMinute(ts) AS minute,
    sum(cost_total_usd) AS cost
FROM llm_telemetry.requests
WHERE ts >= now() - INTERVAL 2 HOUR
GROUP BY minute

3. Hit rate per minute#

SELECT
    toStartOfMinute(ts) AS minute,
    sum(cache_read_input_tokens) /
        (sum(cache_read_input_tokens) + sum(cache_creation_input_tokens) + sum(input_tokens)) * 100 AS hit_pct
FROM llm_telemetry.requests
WHERE ts >= now() - INTERVAL 2 HOUR
GROUP BY minute