التخزين المؤقت لبادئة المطالبة في OpenAI

ما التخزين المؤقت لبادئة المطالبة؟

كيف يعمل التخزين المؤقت للبادئة من الداخل

# No code changes needed to enable prefix caching!
# It is automatic on supported models.

# The API response shows you how many tokens were cached:
# response.usage.prompt_tokens_details.cached_tokens

# Example response usage:
# ChatCompletionUsage(
#   prompt_tokens=2048,
#   completion_tokens=256,
#   total_tokens=2304,
#   prompt_tokens_details=PromptTokensDetails(
#     cached_tokens=1984,   # these tokens were served from KV cache
#     audio_tokens=0,
#   )
# )

التحقق من إصابة التخزين المؤقت في الاستجابة

from openai import OpenAI

client = OpenAI()

SYSTEM_PROMPT = 'You are an expert AI engineer assistant. ' * 100  # long system prompt

def call_with_cache_check(user_message: str):
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {'role': 'system', 'content': SYSTEM_PROMPT},
            {'role': 'user', 'content': user_message},
        ],
    )
    usage = response.usage
    cached = usage.prompt_tokens_details.cached_tokens if usage.prompt_tokens_details else 0
    print(f'Total prompt tokens: {usage.prompt_tokens}')
    print(f'Cached tokens: {cached} ({100*cached//usage.prompt_tokens}%)')
    return response.choices[0].message.content

يجب أن تكون البادئة متطابقة تمامًا

# Optimal structure for prefix caching:
# [LONG STABLE SYSTEM PROMPT] [CACHED DOCUMENTS] [USER QUERY]
#         ↑                              ↑                ↑
#    always same               always same         varies per request
#    → cached at 50%           → cached at 50%     → not cached, full price

# BAD structure (defeats prefix caching):
# [USER QUERY] [CACHED DOCUMENTS] [LONG STABLE SYSTEM PROMPT]
#       ↑                                       ↑
#  changes every request                 never cached because
#  so prefix never matches               it comes after the query

تنظيم المطالبات لتحقيق أقصى كفاءة للتخزين المؤقت

def build_rag_prompt_for_caching(
    system_prompt: str,
    retrieved_docs: list[str],
    conversation_history: list[dict],
    user_query: str,
) -> list[dict]:
    # Order: stable → semi-stable → variable
    context_block = '\n\n'.join(
        f'[Document {i+1}]\n{doc}' for i, doc in enumerate(retrieved_docs)
    )
    return [
        # 1. Stable system prompt (always cached after first request)
        {'role': 'system', 'content': system_prompt},
        # 2. Context injection as a user message (cached when same docs retrieved)
        {'role': 'user', 'content': f'Context documents:\n{context_block}'},
        {'role': 'assistant', 'content': 'I have read the documents.'},
        # 3. Conversation history (semi-stable)
        *conversation_history,
        # 4. Current user query (always different → never cached prefix)
        {'role': 'user', 'content': user_query},
    ]

مدة التخزين المؤقت والإخلاء

النماذج والأسعار المدعومة

# Rough pricing reference (verify at platform.openai.com/pricing)
PRICING = {
    'gpt-4o': {
        'input_per_1M': 2.50,
        'cached_input_per_1M': 1.25,   # 50% off
        'output_per_1M': 10.00,
    },
    'gpt-4o-mini': {
        'input_per_1M': 0.15,
        'cached_input_per_1M': 0.075,  # 50% off
        'output_per_1M': 0.60,
    },
}

def estimate_cost_with_caching(prompt_tokens, cached_tokens, output_tokens, model):
    p = PRICING[model]
    uncached = (prompt_tokens - cached_tokens) * p['input_per_1M'] / 1_000_000
    cached_cost = cached_tokens * p['cached_input_per_1M'] / 1_000_000
    output_cost = output_tokens * p['output_per_1M'] / 1_000_000
    return uncached + cached_cost + output_cost

التخزين المؤقت للمطالبات في Anthropic

import anthropic

client = anthropic.Anthropic()

LONG_DOCUMENT = 'This is a very long reference document...' * 500  # 2000+ tokens

response = client.messages.create(
    model='claude-sonnet-4-5',
    max_tokens=1024,
    system=[
        {
            'type': 'text',
            'text': 'You are a helpful assistant.',
        },
        {
            'type': 'text',
            'text': LONG_DOCUMENT,
            'cache_control': {'type': 'ephemeral'},  # mark for caching
        }
    ],
    messages=[{'role': 'user', 'content': 'Summarize the document.'}],
)
print(response.usage.cache_read_input_tokens)   # tokens served from cache
print(response.usage.cache_creation_input_tokens)  # tokens written to cache

دمج التخزين المؤقت للبادئة مع التخزين المؤقت على مستوى التطبيق

# Three-layer cost optimization stack
#
# Layer 1: Exact cache (Redis, hash-based)
#   → Eliminates 100% of API cost for identical requests
#   → Miss rate: ~60-80% (most queries are unique)
#
# Layer 2: Semantic cache (vector similarity)
#   → Eliminates 100% of API cost for semantically similar requests
#   → Miss rate: ~40-60% of remaining queries
#
# Layer 3: OpenAI prefix caching (automatic)
#   → Reduces input token cost by 50% for long stable prefixes
#   → Applies to ALL remaining API calls that escape layers 1 and 2
#
# Combined effect: 60-80% cost reduction in FAQ/support applications

قياس كفاءة التخزين المؤقت

from dataclasses import dataclass, field
from typing import ClassVar

@dataclass
class CachingMetrics:
    total_prompt_tokens: int = 0
    total_cached_tokens: int = 0
    app_cache_hits: int = 0
    app_cache_misses: int = 0

    @property
    def prefix_cache_ratio(self) -> float:
        if self.total_prompt_tokens == 0:
            return 0
        return self.total_cached_tokens / self.total_prompt_tokens

    @property
    def app_cache_hit_rate(self) -> float:
        total = self.app_cache_hits + self.app_cache_misses
        return self.app_cache_hits / total if total > 0 else 0

    def report(self):
        print(f'App cache hit rate: {self.app_cache_hit_rate:.1%}')
        print(f'Prefix cache ratio: {self.prefix_cache_ratio:.1%}')
        savings_multiplier = (1 - self.app_cache_hit_rate) * (1 - 0.5 * self.prefix_cache_ratio)
        print(f'Effective cost vs no-cache: {savings_multiplier:.1%}')

متى لا يفيد التخزين المؤقت للبادئة

تحقق سريع

مراجعة الدرس