کش، محدودیت نرخ درخواست و بهینه‌سازی هزینه‌های مدل‌های زبانی

بسیاری از برنامه‌های هوش مصنوعی با یک دمو جذاب آغاز می‌شوند، اما با افزایش تعداد کاربران و صدور نخستین صورت‌حساب‌های مصرف توکن، توجیه اقتصادی خود را از دست می‌دهند. ارسال مکرر پرامپت‌های سیستمی حجیم، پاسخ‌گویی دوباره به پرسش‌های مشابه و استفاده‌ی غیرضروری از مدل‌های گران‌قیمت برای کارهای ساده، بودجه‌ی توسعه را به‌سرعت مصرف می‌کند. تسلط بر اقتصاد واحد (Unit Economics) و کنترل هزینه‌ی هر فراخوانی، شرط اصلی رساندن محصول به سودآوری پایدار است.

این موضوع برای توسعه‌دهندگان ایرانی اهمیتی دوچندان دارد؛ زیرا کارت‌های بانکی داخلی امکان پرداخت مستقیم به ارائه‌دهندگانی مانند OpenAI و Anthropic را ندارند و شارژ اعتبار API معمولاً از طریق واسطه‌ها یا گیت‌وی‌های ارزی انجام می‌شود. هر توکن هدررفته مستقیماً هزینه‌های ارزی پروژه را بالا می‌برد و تکیه بر استراتژی‌های کاهش مصرف توکن برای بقای سرویس حیاتی است.

در این درس یاد می‌گیرید چگونه با پیاده‌سازی کش معنایی (Semantic Cache)، کش تطابق دقیق و مسیریابی هوشمند بین مدل‌های گران و سبک، بیش از 70 درصد در مصرف توکن صرفه‌جویی کنید. همچنین با الگوریتم سطل توکن (Token Bucket) برای تعیین سهمیه کاربران و سازوکارهای مدارشکن (Circuit Breaker) آشنا می‌شوید تا از غافلگیری‌های مالی جلوگیری نمایید.

بیشتر استارتاپ‌های هوش مصنوعی به‌خاطر مدل‌های ضعیف شکست نمی‌خورند؛ دلیل اصلی شکست آن‌ها اقتصاد واحد (Unit Economics) نامناسب است. یک بار فراخوانی GPT-4o کسری از سنت هزینه دارد؛ اما اگر 10,000 کاربر روزی 10 درخواست بفرستند، فقط هزینه‌ی توکن‌های ورودی روزانه 250 دلار می‌شود—آن هم قبل از اینکه حتی یک دلار درآمد داشته باشید. شرکت‌هایی دوام می‌آورند که با هر فراخوانی API مثل یک تراکنش مالی رفتار کنند، نه یک فراخوانی تابع ساده در کد.

مرتبط: بخش 11 · درس 15 (کش کردن پرامپت یا Prompt Caching) — در این درس به کش کردن در سطح برنامه (کش معنایی، کش بر اساس هش دقیق و مسیریابی مدل) می‌پردازیم. درس 15 به کش کردن پرامپت در سطح ارائه‌دهنده اختصاص دارد (cache_control در Anthropic، کش خودکار در OpenAI و CachedContent در Gemini). ترکیب هر دو روش می‌تواند هزینه‌ها را بین 50 تا 95 درصد کاهش دهد.

اهداف یادگیری

• پیاده‌سازی کش معنایی (Semantic Cache) برای پاسخ‌گویی به پرسش‌های تکراری یا مشابه از حافظه‌ی کش، به‌جای فراخوانی مجدد API
• محاسبه‌ی هزینه‌ی هر درخواست در ارائه‌دهنده‌های مختلف و پیاده‌سازی محدودیت نرخ درخواست (Rate limit) مبتنی بر توکن و هشدارهای بودجه
• ساخت یک لایه‌ی بهینه‌سازی هزینه شامل فشرده‌سازی پرامپت، مسیریابی مدل (مدل گران در برابر ارزان) و کش کردن پاسخ‌ها
• طراحی استراتژی کش چندسطحی با تطابق دقیق (Exact Match)، شباهت معنایی و کش پیشوند (Prefix Caching) برای انواع مختلف پرسش‌ها

صورت مسئله

یک چت‌بات RAG می‌سازید. عالی کار می‌کند و کاربران عاشقش هستند.

سپس صورت‌حساب سر می‌رسد.

هزینه‌ی GPT-5 به ازای هر یک میلیون توکن، 1.25 دلار برای ورودی و 10 دلار برای خروجی است. Claude Opus 4.7 به ازای ورودی 5 دلار و خروجی 25 دلار تمام می‌شود. Gemini 3.1 Pro برای ورودی 2 دلار و خروجی 12 دلار هزینه دارد (برای پرامپت‌های تا 200K توکن). مدل GPT-5-mini معادل 0.25 و 2 دلار قیمت دارد. مدل GPT-4o که در محاسبات زیر از آن استفاده می‌کنیم، 2.50 دلار ورودی و 10 دلار خروجی است. البته این قیمت‌ها صرفاً جهت مقایسه‌اند؛ همیشه صفحه‌ی رسمی تعرفه‌ی ارائه‌دهنده‌ها را بررسی کنید.

محاسبه‌ای که باعث زمین خوردن استارتاپ‌ها می‌شود:

• 10,000 کاربر فعال روزانه
• 10 پرسش به ازای هر کاربر در روز
• 1,000 توکن ورودی برای هر پرسش (پرامپت سیستم + زمینه + پیام کاربر)
• 500 توکن خروجی برای هر پاسخ

هزینه‌ی ورودی روزانه: 10,000 × 10 × 1,000 / 1,000,000 × $2.50 = 250 دلار در روز هزینه‌ی خروجی روزانه: 10,000 × 10 × 500 / 1,000,000 × $10.00 = 500 دلار در روز مجموع ماهانه: 22,500 دلار در ماه

این فقط هزینه‌ی LLM است. امبدینگ‌ها، میزبانی پایگاه‌داده‌ی برداری و زیرساخت را هم اضافه کنید؛ هزینه‌ی ماهانه‌ی یک چت‌بات ساده به 30,000 دلار می‌رسد.

بخش تلخ ماجرا اینجاست: 40 تا 60 درصد این پرسش‌ها تقریباً تکراری هستند. کاربران سؤال‌های یکسانی را با ادبیات کمی متفاوت مطرح می‌کنند. پرامپت سیستم شما—که در تمام درخواست‌ها کاملاً یکسان است—هر بار هزینه‌تراشی می‌کند. اسناد زمینه‌ای که سیستم RAG بازیابی می‌کند، بین کاربرانی که درباره‌ی یک موضوع مشترک می‌پرسند بارها و بارها تکرار می‌شوند.

در واقع هزینه‌ی کامل یک پردازش تکراری و بیهوده را می‌پردازید.

مفاهیم پایه

کالبدشکافی هزینه‌ی یک فراخوانی LLM

هر فراخوانی API شامل پنج بخش هزینه‌ای است.

پرامپت‌های سیستم قاتل خاموش بودجه‌اند. یک پرامپت سیستم 1,500 توکنی که با هر درخواست ارسال می‌شود، به ازای هر یک میلیون درخواست، فقط برای همین پیشوند 3.75 دلار هزینه برمی‌دارد. در 100K درخواست در روز، یعنی 375 دلار در روز—11,250 دلار در ماه—برای متنی که هرگز تغییر نمی‌کند!

کش ارائه‌دهنده: تخفیف‌های پیش‌فرض

در سال 2026 هر سه ارائه‌دهنده‌ی اصلی، قابلیت کش کردن پرامپت در سمت ارائه‌دهنده را ارائه می‌دهند، اما سازوکار آن‌ها متفاوت است. برای بررسی دقیق‌تر، به بخش 11 · درس 15 مراجعه کنید.

رویکرد Anthropic صریح است. بخش‌هایی از پرامپت را با cache_control: {"type": "ephemeral"} مشخص می‌کنید. اولین درخواست 25% هزینه‌ی نوشتن اضافه دارد. درخواست‌های بعدی با همین پیشوند، 90% تخفیف می‌گیرند. یک پرامپت سیستم 2,000 توکنی که در حالت عادی 0.005 دلار هزینه دارد، در زمان اصابت به کش تنها 0.000625 دلار هزینه برمی‌دارد. در 100K درخواست، این یعنی روزانه 437.50 دلار صرفه‌جویی.

رویکرد OpenAI کاملاً خودکار است. هر پیشوندی از پرامپت که با درخواست‌های قبلی مطابقت داشته باشد، 50% تخفیف می‌گیرد و نیازی به نشانه‌گذاری نیست. نقطه‌ی ضعف: تخفیف کمتر و کنترل محدودتر، اما در عوض پیاده‌سازی آن هیچ زحمتی ندارد.

کش معنایی: لایه‌ی اختصاصی شما

کش ارائه‌دهنده فقط برای پیشوندهای کاملاً یکسان عمل می‌کند. کش معنایی حالت پیچیده‌تر را مدیریت می‌کند: پرسش‌های متفاوتی که معنای یکسانی دارند.

عبارت‌های «شرایط مرجوعی کالا چیست؟» و «چطور کالام رو پس بفرستم؟» رشته‌های متفاوتی هستند اما نیت (Intent) یکسانی دارند. کش معنایی هر دو پرسش را امبد می‌کند، شباهت کسینوسی (Cosine Similarity) آن‌ها را می‌سنجد و اگر شباهت از یک آستانه (Threshold)—معمولاً بین 0.92 تا 0.95—بیشتر باشد، پاسخ کش‌شده را برمی‌گرداند.

هزینه‌ی امبدینگ بسیار ناچیز است. مدل text-embedding-3-small در OpenAI به ازای هر یک میلیون توکن فقط 0.02 دلار هزینه دارد. بررسی حافظه‌ی کش در مقایسه با فراخوانی کامل یک LLM تقریباً رایگان است.

کش تطابق دقیق: هش و تطبیق

برای فراخوانی‌های قطعی (Deterministic) با دمای صفر (temperature=0)، مدل یکسان و پرامپت یکسان، کش تطابق دقیق بسیار ساده‌تر و سریع‌تر است. کافی است کل پرامپت را هش کنید، کش را بررسی کنید و در صورت وجود پاسخ، آن را برگردانید.

این روش برای موارد زیر عالی کار می‌کند:

• پرامپت سیستم + زمینه‌ی ثابت + پرسش‌های کاملاً یکسان کاربران
• صدا زدن توابع با تعاریف یکسان ابزارها
• پردازش‌های دسته‌ای که در آن‌ها یک سند بارها پردازش می‌شود

محدودیت نرخ درخواست: محافظت از بودجه

محدود کردن نرخ درخواست فقط برای برقراری عدالت بین کاربران نیست؛ مسئله‌ی بقای پروژه است.

الگوریتم سطل توکن (Token bucket): به هر کاربر سطلی از N توکن اختصاص داده می‌شود که با نرخ R توکن در ثانیه مجدداً پر می‌شود. هر درخواست تعدادی توکن از سطل مصرف می‌کند. اگر سطل خالی باشد، درخواست رد می‌شود. این سازوکار اجازه‌ی ارسال رگباری درخواست‌ها (استفاده‌ی یک‌باره از تمام سطل) را می‌دهد در حالی که نرخ میانگین را تحت کنترل نگه می‌دارد.

سهمیه‌ی هر کاربر: برای هر سطح کاربری، محدودیت روزانه/ماهانه برای توکن‌ها تعیین کنید.

مسیریابی مدل: انتخاب مدل مناسب برای کار مناسب

همه‌ی پرسش‌ها نیازی به GPT-4o ندارند.

پاسخ دادن به «فروشگاه ساعت چند بسته می‌شود؟» نیازی به یک مدل 10 دلاری به ازای هر میلیون توکن خروجی ندارد. مدل GPT-4o-mini با هزینه‌ی خروجی 0.60 دلار یا Claude Haiku با 1.25 دلار به ازای هر میلیون توکن خروجی این کار را به بهترین شکل انجام می‌دهند. یک دسته‌بند ساده می‌تواند پرسش‌های سبک را به مدل‌های ارزان و پرسش‌های پیچیده را به مدل‌های گران هدایت کند.

یک مسیریاب (Router) تنظیم‌شده و دقیق، به‌تنهایی 40 تا 70 درصد از هزینه‌های مدل را کاهش می‌دهد.

ردگیری هزینه‌ها: آگاهی از جریان مصرف بودجه

چیزی را که اندازه‌گیری نکنید، نمی‌توانید بهینه‌سازی کنید. هر فراخوانی API را با جزییات زیر لاگ کنید:

• برچسب زمان (Timestamp)
• نام مدل
• توکن‌های ورودی
• توکن‌های خروجی
• تاخیر (Latency) بر حسب میلی‌ثانیه
• هزینه‌ی محاسبه‌شده ($)
• شناسه‌ی کاربر
• اصابت/عدم اصابت به کش
• دسته‌بندی درخواست

این داده‌ها مشخص می‌کنند کدام قابلیت‌ها پرخرج هستند، کدام کاربران بیشترین منابع را مصرف می‌کنند و استفاده از کش در کدام بخش‌ها بیشترین بازدهی را دارد.

پردازش دسته‌ای: تخفیف‌های تجمیعی

قابلیت Batch API در OpenAI درخواست‌ها را به‌صورت ناهمگام و با 50% تخفیف پردازش می‌کند. می‌توانید دسته‌هایی تا سقف 50,000 درخواست ثبت کنید و خروجی را ظرف 24 ساعت تحویل بگیرید.

پردازش دسته‌ای برای این موارد مناسب است:

• پردازش شبانه‌ی اسناد
• دسته‌بندی انبوه داده‌ها
• ارزیابی‌ها
• پایپ‌لاین‌های غنی‌سازی داده

نامناسب برای: درخواست‌های آنی کاربران که تاخیر در آن‌ها اهمیت دارد.

هشدارهای بودجه و مدارشکن‌ها

مدارشکن در زمان رسیدن به سقف مشخصی از بودجه، جلوی هزینه را می‌گیرد. بدون این قابلیت، یک باگ یا سوءاستفاده‌ی کاربران می‌تواند کل بودجه‌ی ماهانه‌ی شما را ظرف چند ساعت بسوزاند.

سه آستانه تعیین کنید:

1. هشدار (70% از بودجه): ارسال پیام هشدار
2. کاهش سرعت (85% از بودجه): تغییر مسیر تمام درخواست‌ها فقط به مدل‌های ارزان‌تر
3. قطع کامل (95% از بودجه): رد درخواست‌های جدید و پاسخ‌دهی صرفاً از روی حافظه‌ی کش

پشته‌ی بهینه‌سازی هزینه

این روش‌ها را به ترتیب پیاده‌سازی کنید. هر لایه اثر لایه‌ی قبلی را تقویت می‌کند.

یک برنامه‌ی RAG با پیاده‌سازی لایه‌های 1 تا 5 معمولاً هزینه‌هایش را از 22,500 دلار در ماه به 4,000 تا 6,000 دلار می‌رساند. این همان تفاوت میان سوختن زودهنگام سرمایه و ساخت یک کسب‌وکار پایدار است.

ارقام واقعی: قبل و بعد از بهینه‌سازی

مقایسه‌ی واقعی ارقام برای یک چت‌بات RAG با 10,000 کاربر فعال روزانه:

هزینه‌ی امبدینگ برای کش معنایی (180 دلار در ماه)، در همان ساعت اول پاسخ‌گویی از طریق اصابت به کش جبران می‌شود.

پیاده‌سازی از صفر

گام 1: ماشین‌حساب محاسبه‌ی هزینه

یک ماشین‌حساب توکن بسازید که از قیمت‌های روز مدل‌های مطرح آگاه باشد.

import hashlib
import time
import json
import math
from dataclasses import dataclass, field


MODEL_PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00, "cached_input": 1.25},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60, "cached_input": 0.075},
    "gpt-4.1": {"input": 2.00, "output": 8.00, "cached_input": 0.50},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60, "cached_input": 0.10},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40, "cached_input": 0.025},
    "o3": {"input": 2.00, "output": 8.00, "cached_input": 0.50},
    "o3-mini": {"input": 1.10, "output": 4.40, "cached_input": 0.55},
    "o4-mini": {"input": 1.10, "output": 4.40, "cached_input": 0.275},
    "claude-opus-4": {"input": 15.00, "output": 75.00, "cached_input": 1.50},
    "claude-sonnet-4": {"input": 3.00, "output": 15.00, "cached_input": 0.30},
    "claude-haiku-3.5": {"input": 0.80, "output": 4.00, "cached_input": 0.08},
    "gemini-2.5-pro": {"input": 1.25, "output": 10.00, "cached_input": 0.3125},
    "gemini-2.5-flash": {"input": 0.15, "output": 0.60, "cached_input": 0.0375},
}


def calculate_cost(model, input_tokens, output_tokens, cached_input_tokens=0):
    if model not in MODEL_PRICING:
        return {"error": f"Unknown model: {model}"}
    pricing = MODEL_PRICING[model]
    non_cached = input_tokens - cached_input_tokens
    input_cost = (non_cached / 1_000_000) * pricing["input"]
    cached_cost = (cached_input_tokens / 1_000_000) * pricing["cached_input"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    total = input_cost + cached_cost + output_cost
    return {
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cached_input_tokens": cached_input_tokens,
        "input_cost": round(input_cost, 6),
        "cached_input_cost": round(cached_cost, 6),
        "output_cost": round(output_cost, 6),
        "total_cost": round(total, 6),
    }

گام 2: کش تطابق دقیق

کل پرامپت را هش کنید و برای درخواست‌های کاملاً مشابه، پاسخ کش‌شده را برگردانید.

class ExactCache:
    def __init__(self, max_size=1000, ttl_seconds=3600):
        self.cache = {}
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    def _hash(self, model, messages, temperature):
        key_data = json.dumps({"model": model, "messages": messages, "temperature": temperature}, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()

    def get(self, model, messages, temperature=0.0):
        if temperature > 0:
            self.misses += 1
            return None
        key = self._hash(model, messages, temperature)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                self.hits += 1
                entry["access_count"] += 1
                return entry["response"]
            del self.cache[key]
        self.misses += 1
        return None

    def put(self, model, messages, temperature, response):
        if temperature > 0:
            return
        if len(self.cache) >= self.max_size:
            oldest_key = min(self.cache, key=lambda k: self.cache[k]["timestamp"])
            del self.cache[oldest_key]
        key = self._hash(model, messages, temperature)
        self.cache[key] = {
            "response": response,
            "timestamp": time.time(),
            "access_count": 1,
        }

    def stats(self):
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 4) if total > 0 else 0,
            "cache_size": len(self.cache),
        }

گام 3: کش معنایی

پرسش‌ها را امبد کنید و در صورتی که شباهت از حد آستانه گذشت، پاسخ کش‌شده را برگردانید.

def simple_embed(text):
    words = text.lower().split()
    vocab = {}
    for w in words:
        vocab[w] = vocab.get(w, 0) + 1
    norm = math.sqrt(sum(v * v for v in vocab.values()))
    if norm == 0:
        return {}
    return {k: v / norm for k, v in vocab.items()}


def cosine_similarity(a, b):
    if not a or not b:
        return 0.0
    all_keys = set(a) | set(b)
    dot = sum(a.get(k, 0) * b.get(k, 0) for k in all_keys)
    return dot


class SemanticCache:
    def __init__(self, similarity_threshold=0.85, max_size=500, ttl_seconds=3600):
        self.entries = []
        self.threshold = similarity_threshold
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0

    def get(self, query):
        query_embedding = simple_embed(query)
        now = time.time()
        best_match = None
        best_sim = 0.0
        for entry in self.entries:
            if now - entry["timestamp"] > self.ttl:
                continue
            sim = cosine_similarity(query_embedding, entry["embedding"])
            if sim > best_sim:
                best_sim = sim
                best_match = entry
        if best_match and best_sim >= self.threshold:
            self.hits += 1
            best_match["access_count"] += 1
            return {"response": best_match["response"], "similarity": round(best_sim, 4), "original_query": best_match["query"]}
        self.misses += 1
        return None

    def put(self, query, response):
        if len(self.entries) >= self.max_size:
            self.entries.sort(key=lambda e: e["timestamp"])
            self.entries.pop(0)
        self.entries.append({
            "query": query,
            "embedding": simple_embed(query),
            "response": response,
            "timestamp": time.time(),
            "access_count": 1,
        })

    def stats(self):
        total = self.hits + self.misses
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": round(self.hits / total, 4) if total > 0 else 0,
            "cache_size": len(self.entries),
        }

گام 4: محدودکننده‌ی نرخ درخواست

یک محدودکننده‌ی نرخ درخواست بر اساس الگوریتم سطل توکن و سهمیه‌ی اختصاصی هر کاربر بسازید.

class TokenBucketRateLimiter:
    def __init__(self):
        self.buckets = {}
        self.tiers = {
            "free": {"capacity": 50_000, "refill_rate": 500, "max_requests_per_min": 10},
            "pro": {"capacity": 500_000, "refill_rate": 5_000, "max_requests_per_min": 60},
            "enterprise": {"capacity": 5_000_000, "refill_rate": 50_000, "max_requests_per_min": 300},
        }

    def _get_bucket(self, user_id, tier="free"):
        if user_id not in self.buckets:
            tier_config = self.tiers.get(tier, self.tiers["free"])
            self.buckets[user_id] = {
                "tokens": tier_config["capacity"],
                "capacity": tier_config["capacity"],
                "refill_rate": tier_config["refill_rate"],
                "last_refill": time.time(),
                "request_timestamps": [],
                "max_rpm": tier_config["max_requests_per_min"],
                "tier": tier,
                "total_tokens_used": 0,
            }
        return self.buckets[user_id]

    def _refill(self, bucket):
        now = time.time()
        elapsed = now - bucket["last_refill"]
        refill = int(elapsed * bucket["refill_rate"])
        if refill > 0:
            bucket["tokens"] = min(bucket["capacity"], bucket["tokens"] + refill)
            bucket["last_refill"] = now

    def check(self, user_id, tokens_needed, tier="free"):
        bucket = self._get_bucket(user_id, tier)
        self._refill(bucket)
        now = time.time()
        bucket["request_timestamps"] = [t for t in bucket["request_timestamps"] if now - t < 60]
        if len(bucket["request_timestamps"]) >= bucket["max_rpm"]:
            return {"allowed": False, "reason": "rate_limit", "retry_after_seconds": 60 - (now - bucket["request_timestamps"][0])}
        if bucket["tokens"] < tokens_needed:
            deficit = tokens_needed - bucket["tokens"]
            wait = deficit / bucket["refill_rate"]
            return {"allowed": False, "reason": "token_limit", "tokens_available": bucket["tokens"], "retry_after_seconds": round(wait, 1)}
        return {"allowed": True, "tokens_available": bucket["tokens"]}

    def consume(self, user_id, tokens_used, tier="free"):
        bucket = self._get_bucket(user_id, tier)
        bucket["tokens"] -= tokens_used
        bucket["request_timestamps"].append(time.time())
        bucket["total_tokens_used"] += tokens_used

    def get_usage(self, user_id):
        if user_id not in self.buckets:
            return {"error": "User not found"}
        b = self.buckets[user_id]
        return {
            "user_id": user_id,
            "tier": b["tier"],
            "tokens_remaining": b["tokens"],
            "capacity": b["capacity"],
            "total_tokens_used": b["total_tokens_used"],
            "utilization": round(b["total_tokens_used"] / b["capacity"], 4) if b["capacity"] else 0,
        }

گام 5: ردگیر هزینه

تمامی فراخوانی‌ها را لاگ کنید و مجموع هزینه‌ها را لحظه‌به‌لحظه محاسبه نمایید.

class CostTracker:
    def __init__(self, monthly_budget=1000.0):
        self.logs = []
        self.monthly_budget = monthly_budget
        self.alerts = []

    def log_call(self, model, input_tokens, output_tokens, cached_input_tokens=0, latency_ms=0, user_id="anonymous", cache_status="miss"):
        cost = calculate_cost(model, input_tokens, output_tokens, cached_input_tokens)
        entry = {
            "timestamp": time.time(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cached_input_tokens": cached_input_tokens,
            "latency_ms": latency_ms,
            "cost": cost["total_cost"],
            "user_id": user_id,
            "cache_status": cache_status,
        }
        self.logs.append(entry)
        self._check_budget()
        return entry

    def _check_budget(self):
        total = self.total_cost()
        pct = total / self.monthly_budget if self.monthly_budget > 0 else 0
        if pct >= 0.95 and not any(a["level"] == "stop" for a in self.alerts):
            self.alerts.append({"level": "stop", "message": f"Budget 95% consumed: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})
        elif pct >= 0.85 and not any(a["level"] == "throttle" for a in self.alerts):
            self.alerts.append({"level": "throttle", "message": f"Budget 85% consumed: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})
        elif pct >= 0.70 and not any(a["level"] == "warning" for a in self.alerts):
            self.alerts.append({"level": "warning", "message": f"Budget 70% consumed: ${total:.2f}/${self.monthly_budget:.2f}", "timestamp": time.time()})

    def total_cost(self):
        return round(sum(e["cost"] for e in self.logs), 6)

    def cost_by_model(self):
        by_model = {}
        for e in self.logs:
            m = e["model"]
            if m not in by_model:
                by_model[m] = {"calls": 0, "cost": 0, "input_tokens": 0, "output_tokens": 0}
            by_model[m]["calls"] += 1
            by_model[m]["cost"] = round(by_model[m]["cost"] + e["cost"], 6)
            by_model[m]["input_tokens"] += e["input_tokens"]
            by_model[m]["output_tokens"] += e["output_tokens"]
        return by_model

    def cache_savings(self):
        cache_hits = [e for e in self.logs if e["cache_status"] == "hit"]
        if not cache_hits:
            return {"saved": 0, "cache_hits": 0}
        saved = 0
        for e in cache_hits:
            full_cost = calculate_cost(e["model"], e["input_tokens"], e["output_tokens"])
            saved += full_cost["total_cost"]
        return {"saved": round(saved, 4), "cache_hits": len(cache_hits)}

    def summary(self):
        if not self.logs:
            return {"total_calls": 0, "total_cost": 0}
        total_latency = sum(e["latency_ms"] for e in self.logs)
        cache_hits = sum(1 for e in self.logs if e["cache_status"] == "hit")
        return {
            "total_calls": len(self.logs),
            "total_cost": self.total_cost(),
            "avg_cost_per_call": round(self.total_cost() / len(self.logs), 6),
            "avg_latency_ms": round(total_latency / len(self.logs), 1),
            "cache_hit_rate": round(cache_hits / len(self.logs), 4),
            "cost_by_model": self.cost_by_model(),
            "cache_savings": self.cache_savings(),
            "budget_remaining": round(self.monthly_budget - self.total_cost(), 2),
            "budget_utilization": round(self.total_cost() / self.monthly_budget, 4) if self.monthly_budget > 0 else 0,
            "alerts": self.alerts,
        }

گام 6: مسیریاب مدل

پرسش‌ها را به ارزان‌ترین مدلی که توانایی پاسخ‌گویی به آن‌ها را دارد هدایت کنید.

SIMPLE_KEYWORDS = ["what time", "hours", "address", "phone", "price", "return policy", "hello", "hi", "thanks", "yes", "no"]
COMPLEX_KEYWORDS = ["analyze", "compare", "explain why", "write code", "debug", "architect", "design", "trade-off", "evaluate"]


def classify_complexity(query):
    q = query.lower()
    if len(q.split()) <= 5 or any(kw in q for kw in SIMPLE_KEYWORDS):
        return "simple"
    if any(kw in q for kw in COMPLEX_KEYWORDS):
        return "complex"
    return "medium"


def route_model(query, tier="pro"):
    complexity = classify_complexity(query)
    routing_table = {
        "simple": {"free": "gpt-4.1-nano", "pro": "gpt-4o-mini", "enterprise": "gpt-4o-mini"},
        "medium": {"free": "gpt-4o-mini", "pro": "claude-sonnet-4", "enterprise": "claude-sonnet-4"},
        "complex": {"free": "gpt-4o-mini", "pro": "gpt-4o", "enterprise": "claude-opus-4"},
    }
    model = routing_table[complexity].get(tier, "gpt-4o-mini")
    return {"query": query, "complexity": complexity, "model": model, "tier": tier}

گام 7: اجرای نسخه‌ی نمایشی

def simulate_llm_call(model, query):
    input_tokens = len(query.split()) * 4 + 500
    output_tokens = 150 + (len(query.split()) * 2)
    latency = 200 + (output_tokens * 2)
    return {
        "model": model,
        "response": f"[Simulated {model} response to: {query[:50]}...]",
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "latency_ms": latency,
    }


def run_demo():
    print("=" * 60)
    print("  Caching, Rate Limiting & Cost Optimization Demo")
    print("=" * 60)

    print("\n--- Model Pricing ---")
    for model, pricing in list(MODEL_PRICING.items())[:6]:
        cost_1k = calculate_cost(model, 1000, 500)
        print(f"  {model}: ${cost_1k['total_cost']:.6f} per 1K in + 500 out")

    print("\n--- Cost Comparison: 100K Requests ---")
    for model in ["gpt-4o", "gpt-4o-mini", "claude-sonnet-4", "claude-haiku-3.5"]:
        cost = calculate_cost(model, 1000 * 100_000, 500 * 100_000)
        print(f"  {model}: ${cost['total_cost']:.2f}")

    print("\n--- Anthropic Cache Savings ---")
    no_cache = calculate_cost("claude-sonnet-4", 2000, 500, 0)
    with_cache = calculate_cost("claude-sonnet-4", 2000, 500, 1500)
    saving = no_cache["total_cost"] - with_cache["total_cost"]
    print(f"  Without cache: ${no_cache['total_cost']:.6f}")
    print(f"  With 1500 cached tokens: ${with_cache['total_cost']:.6f}")
    print(f"  Savings per call: ${saving:.6f} ({saving/no_cache['total_cost']*100:.1f}%)")

    exact_cache = ExactCache(max_size=100, ttl_seconds=300)
    semantic_cache = SemanticCache(similarity_threshold=0.75, max_size=100)
    rate_limiter = TokenBucketRateLimiter()
    tracker = CostTracker(monthly_budget=100.0)

    print("\n--- Exact Cache ---")
    messages_1 = [{"role": "user", "content": "What is the return policy?"}]
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.0)
    print(f"  First lookup: {'HIT' if result else 'MISS'}")
    exact_cache.put("gpt-4o-mini", messages_1, 0.0, "You can return items within 30 days.")
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.0)
    print(f"  Second lookup: {'HIT' if result else 'MISS'} -> {result}")
    result = exact_cache.get("gpt-4o-mini", messages_1, 0.7)
    print(f"  With temp=0.7: {'HIT' if result else 'MISS (non-deterministic, skip cache)'}")
    print(f"  Stats: {exact_cache.stats()}")

    print("\n--- Semantic Cache ---")
    test_queries = [
        ("What is the return policy?", "Items can be returned within 30 days with receipt."),
        ("How do I return an item?", None),
        ("What are your store hours?", "We are open 9am-9pm Monday through Saturday."),
        ("When does the store open?", None),
        ("Tell me about quantum computing", "Quantum computers use qubits..."),
        ("Explain quantum mechanics", None),
    ]
    for query, response in test_queries:
        cached = semantic_cache.get(query)
        if cached:
            print(f"  '{query[:40]}' -> CACHE HIT (sim={cached['similarity']}, original='{cached['original_query'][:40]}')")
        elif response:
            semantic_cache.put(query, response)
            print(f"  '{query[:40]}' -> MISS (stored)")
        else:
            print(f"  '{query[:40]}' -> MISS (no match)")
    print(f"  Stats: {semantic_cache.stats()}")

    print("\n--- Rate Limiting ---")
    for i in range(12):
        check = rate_limiter.check("user_1", 1000, "free")
        if check["allowed"]:
            rate_limiter.consume("user_1", 1000, "free")
        status = "OK" if check["allowed"] else f"BLOCKED ({check['reason']})"
        if i < 5 or not check["allowed"]:
            print(f"  Request {i+1}: {status}")
    print(f"  Usage: {rate_limiter.get_usage('user_1')}")

    print("\n--- Model Routing ---")
    routing_queries = [
        "What time do you close?",
        "Summarize this quarterly earnings report",
        "Analyze the trade-offs between microservices and monoliths",
        "Hello",
        "Write code for a binary search tree with deletion",
    ]
    for q in routing_queries:
        route = route_model(q, "pro")
        print(f"  '{q[:50]}' -> {route['model']} ({route['complexity']})")

    print("\n--- Full Pipeline: Before vs After Optimization ---")
    queries = [
        "What is the return policy?",
        "How do I return something?",
        "What are your hours?",
        "When do you open?",
        "Explain the difference between TCP and UDP",
        "Compare TCP vs UDP protocols",
        "Hello",
        "What is your phone number?",
        "Write a Python function to sort a list",
        "Analyze the pros and cons of serverless architecture",
    ]

    print("\n  [Before: no caching, single model (gpt-4o)]")
    tracker_before = CostTracker(monthly_budget=1000.0)
    for q in queries:
        result = simulate_llm_call("gpt-4o", q)
        tracker_before.log_call("gpt-4o", result["input_tokens"], result["output_tokens"], latency_ms=result["latency_ms"], cache_status="miss")
    before = tracker_before.summary()
    print(f"  Total cost: ${before['total_cost']:.6f}")
    print(f"  Avg cost/call: ${before['avg_cost_per_call']:.6f}")
    print(f"  Avg latency: {before['avg_latency_ms']}ms")

    print("\n  [After: caching + routing + rate limiting]")
    exact_c = ExactCache()
    semantic_c = SemanticCache(similarity_threshold=0.75)
    tracker_after = CostTracker(monthly_budget=1000.0)

    for q in queries:
        messages = [{"role": "user", "content": q}]
        cached = exact_c.get("gpt-4o", messages, 0.0)
        if cached:
            tracker_after.log_call("gpt-4o-mini", 0, 0, latency_ms=5, cache_status="hit")
            continue
        sem_cached = semantic_c.get(q)
        if sem_cached:
            tracker_after.log_call("gpt-4o-mini", 0, 0, latency_ms=15, cache_status="hit")
            continue
        route = route_model(q)
        result = simulate_llm_call(route["model"], q)
        tracker_after.log_call(route["model"], result["input_tokens"], result["output_tokens"], latency_ms=result["latency_ms"], cache_status="miss")
        exact_c.put(route["model"], messages, 0.0, result["response"])
        semantic_c.put(q, result["response"])

    after = tracker_after.summary()
    print(f"  Total cost: ${after['total_cost']:.6f}")
    print(f"  Avg cost/call: ${after['avg_cost_per_call']:.6f}")
    print(f"  Avg latency: {after['avg_latency_ms']}ms")
    print(f"  Cache hit rate: {after['cache_hit_rate']:.0%}")

    if before["total_cost"] > 0:
        savings_pct = (1 - after["total_cost"] / before["total_cost"]) * 100
        print(f"\n  SAVINGS: {savings_pct:.1f}% cost reduction")
        print(f"  Latency improvement: {(1 - after['avg_latency_ms'] / before['avg_latency_ms']) * 100:.1f}% faster")

    print("\n--- Budget Alerts Demo ---")
    alert_tracker = CostTracker(monthly_budget=0.01)
    for i in range(5):
        alert_tracker.log_call("gpt-4o", 5000, 2000, latency_ms=500)
    print(f"  Total spent: ${alert_tracker.total_cost():.6f} / ${alert_tracker.monthly_budget}")
    for alert in alert_tracker.alerts:
        print(f"  ALERT [{alert['level'].upper()}]: {alert['message']}")

    print("\n--- Cost Breakdown by Model ---")
    multi_tracker = CostTracker(monthly_budget=500.0)
    for _ in range(50):
        multi_tracker.log_call("gpt-4o-mini", 800, 200, latency_ms=150)
    for _ in range(30):
        multi_tracker.log_call("claude-sonnet-4", 1500, 500, latency_ms=400)
    for _ in range(10):
        multi_tracker.log_call("gpt-4o", 2000, 800, latency_ms=600)
    for _ in range(10):
        multi_tracker.log_call("claude-opus-4", 3000, 1000, latency_ms=1200)
    breakdown = multi_tracker.cost_by_model()
    for model, data in sorted(breakdown.items(), key=lambda x: x[1]["cost"], reverse=True):
        print(f"  {model}: {data['calls']} calls, ${data['cost']:.6f}, {data['input_tokens']:,} in / {data['output_tokens']:,} out")
    print(f"  Total: ${multi_tracker.total_cost():.6f}")

    print("\n" + "=" * 60)
    print("  Demo complete.")
    print("=" * 60)


if __name__ == "__main__":
    run_demo()

استفاده از ابزارهای واقعی

کش کردن پرامپت در Anthropic

# وارد کردن anthropic
#
# client = anthropic.Anthropic()
#
# response = client.messages.create(
#     model="claude-sonnet-5",
#     max_tokens=1024,
#     system=[
#         {
#             "type": "text",
# "text": "شما یک ایجنت پشتیبانی مشتریان مفید برای شرکت Acme هستید...",
#             "cache_control": {"type": "ephemeral"},
#         }
#     ],
#     messages=[{"role": "user", "content": "What is the return policy?"}],
# )
#
# print(f"Input tokens: {response.usage.input_tokens}")
# print(f"Cache creation tokens: {response.usage.cache_creation_input_tokens}")
# print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")

اولین فراخوانی داده‌ها را در کش می‌نویسد (با 25% هزینه‌ی اضافی). تمام فراخوانی‌های بعدی با پیشوند پرامپت سیستم یکسان، اطلاعات را از کش می‌خوانند (با 90% تخفیف). کش تا 5 دقیقه معتبر می‌ماند و با هر بار اصابت به کش، این تایمر مجدداً تمدید می‌شود.

کش خودکار در OpenAI

# وارد کردن OpenAI از openai
#
# client = OpenAI()
#
# response = client.chat.completions.create(
#     model="gpt-4o",
#     messages=[
#         {"role": "system", "content": "You are a helpful customer support agent..."},
#         {"role": "user", "content": "What is the return policy?"},
#     ],
# )
#
# print(f"Prompt tokens: {response.usage.prompt_tokens}")
# print(f"Cached tokens: {response.usage.prompt_tokens_details.cached_tokens}")
# print(f"Completion tokens: {response.usage.completion_tokens}")

سرویس OpenAI کش کردن را به‌طور خودکار انجام می‌دهد. هر پیشوندی از پرامپت با بیش از 1,024 توکن که با درخواست‌های اخیر مطابقت داشته باشد، مشمول 50% تخفیف می‌شود. نیازی به تغییر کد نیست—تنها فیلد prompt_tokens_details.cached_tokens را در پاسخ بررسی کنید تا از کارکرد آن مطمئن شوید.

استفاده از Batch API در OpenAI

# وارد کردن json
# وارد کردن OpenAI از openai
#
# client = OpenAI()
#
# requests = []
# حلقه روی queries به همراه شمارنده
#     requests.append({
#         "custom_id": f"request-{i}",
#         "method": "POST",
#         "url": "/v1/chat/completions",
#         "body": {
#             "model": "gpt-4o-mini",
#             "messages": [{"role": "user", "content": query}],
#         },
#     })
#
# باز کردن فایل batch_input.jsonl برای نوشتن
# حلقه روی requests
#         f.write(json.dumps(r) + "\n")
#
# batch_file = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")
# batch = client.batches.create(input_file_id=batch_file.id, endpoint="/v1/chat/completions", completion_window="24h")
# print(f"Batch ID: {batch.id}, Status: {batch.status}")

سرویس Batch API یک تخفیف ثابت 50 درصدی روی تمام توکن‌ها ارائه می‌دهد. نتایج ظرف 24 ساعت آماده می‌شوند؛ بنابراین برای کارهایی که نیاز به پاسخ آنی ندارند عالی است: ارزیابی مدل‌ها، برچسب‌گذاری داده‌ها و خلاصه‌سازی انبوه.

راه‌اندازی کش معنایی با Redis در محیط پروداکشن

# وارد کردن redis
# وارد کردن numpy به صورت np
# وارد کردن OpenAI از openai
#
# r = redis.Redis()
# client = OpenAI()
#
# تعریف تابع get_embedding(text)
#     response = client.embeddings.create(model="text-embedding-3-small", input=text)
# بازگرداندن response.data[0].embedding
#
# def semantic_cache_lookup(query, threshold=0.95):
#     query_emb = np.array(get_embedding(query))
#     keys = r.keys("cache:emb:*")
#     best_sim, best_key = 0, None
# حلقه روی keys
#         stored_emb = np.frombuffer(r.get(key), dtype=np.float32)
#         sim = np.dot(query_emb, stored_emb) / (np.linalg.norm(query_emb) * np.linalg.norm(stored_emb))
# اگر sim بزرگ‌تر از best_sim بود
#             best_sim, best_key = sim, key
#     if best_sim >= threshold and best_key:
#         response_key = best_key.decode().replace("cache:emb:", "cache:resp:")
# بازگرداندن مقدار دیکود شده از r.get(response_key)
# بازگرداندن None

در محیط واقعی و عملیاتی، جست‌وجوی خطی را با یک نمایه‌ی برداری (مانند جست‌وجوی برداری Redis، سرویس Pinecone یا pgvector) جایگزین کنید. بررسی خطی فقط برای کمتر از 1,000 ورودی پاسخ‌گو است. برای مقیاس‌های بزرگ‌تر، حتماً از روش‌های جست‌وجوی تقریبی نزدیک‌ترین همسایه (ANN) برای جست‌وجوی $O(\log n)$ استفاده کنید.

خروجی این درس

این درس فایل outputs/prompt-cost-optimizer.md را تولید می‌کند—پرامپتی با قابلیت استفاده‌ی مجدد که برنامه‌ی LLM شما را تحلیل می‌کند و راهکارهای مشخصی را همراه با پیش‌بینی میزان صرفه‌جویی پیشنهاد می‌دهد.

همچنین فایل outputs/skill-cost-patterns.md ایجاد می‌شود—چارچوبی برای تصمیم‌گیری در انتخاب استراتژی کش مناسب، تنظیمات محدودیت نرخ درخواست و قواعد مسیریابی مدل بر اساس نیازهای پروژه‌تان.

تمرین‌ها

1. پیاده‌سازی سازوکار حذف LRU برای کش معنایی. سازوکار حذف قدیمی‌ترین ورودی (Oldest-first) را با کم‌استفاده‌ترین ورودی‌های اخیر (LRU) جایگزین کنید. زمان آخرین دسترسی به هر ورودی را ثبت کرده و وقتی ظرفیت کش تکمیل شد، داده‌ای را که قدیمی‌ترین زمان دسترسی را دارد حذف کنید. نرخ اصابت به کش را میان این دو استراتژی در طول 100 پرسش مقایسه نمایید.

2. ساخت ابزار پیش‌بینی هزینه. با استفاده از لاگ‌های فراخوانی‌های API ثبت‌شده توسط CostTracker، هزینه‌ی ماهانه را بر اساس میانگین متحرک 7 روز گذشته پیش‌بینی کنید. الگوهای مصرف روزهای کاری در مقایسه با تعطیلات را در نظر بگیرید. اگر پیش‌بینی هزینه‌ی ماهانه بیش از 20% از بودجه فراتر رفت، هشدار صادر کنید.

3. پیاده‌سازی کش معنایی چندسطحی. دو آستانه‌ی شباهت تعیین کنید: 0.98 برای اصابت‌های با اطمینان بالا (پاسخ آنی) و 0.90 برای اصابت‌های با اطمینان متوسط (ارائه‌ی پاسخ همراه با پیام رفع مسئولیت: «بر اساس سؤالی مشابه در گذشته...»). ثبت کنید که هر پاسخ از کدام سطح بازگردانده شده و تفاوت در رضایت کاربران را بسنجید.

4. ساخت یک دسته‌بند برای مسیریابی مدل. دسته‌بند مبتنی بر کلمات کلیدی را با یک دسته‌بند مبتنی بر امبدینگ جایگزین کنید. 50 پرسش برچسب‌دار (ساده/متوسط/پیچیده) را امبد کنید؛ سپس پرسش‌های جدید را بر اساس نزدیک‌ترین نمونه‌ی برچسب‌دار دسته‌بندی نمایید. دقت دسته‌بندی را روی یک مجموعه‌ی آزمایشی شامل 20 پرسش بسنجید.

5. پیاده‌سازی مدارشکن با سطوح مختلف کاهش کارایی. در 70% مصرف بودجه، یک اخطار لاگ کنید. در 85%، تمام درخواست‌ها را به‌صورت خودکار به ارزان‌ترین مدل (gpt-4o-mini) هدایت نمایید. در 95%، فقط به درخواست‌های موجود در کش پاسخ دهید و ورودی‌های جدید را رد کنید. برای ارزیابی، 1,000 درخواست را در برابر یک بودجه‌ی 1.00 دلاری شبیه‌سازی کنید و فعال شدن درست هر آستانه را بیازمایید.

اصطلاحات کلیدی

برای مطالعه‌ی بیشتر

• راهنمای کش کردن پرامپت در Anthropic -- مستندات رسمی برای نشانه‌های صریح cache_control، قیمت‌گذاری و رفتار طول عمر کش.
• کش کردن پرامپت در OpenAI -- راهنمای کش خودکار در OpenAI، نحوه‌ی بررسی اصابت به کش از طریق فیلدهای usage و حداقل طول پیشوند لازم.
• Batch API در OpenAI -- تخفیف 50 درصدی برای پردازش ناهمگام، فرمت JSONL، بازه‌ی تکمیل 24 ساعته و سقف 50K درخواست.
• GPTCache -- کتابخانه‌ی متن‌باز برای کش معنایی با پشتیبانی از بک‌اندهای مختلف امبدینگ، پایگاه‌های داده‌ی برداری و سیاست‌های تخلیه حافظه.
• مسیریاب مدل Martian -- سیستم مسیریابی مدل عملیاتی که ارزان‌ترین مدل متناسب با هر پرسش را انتخاب می‌کند.
• Not Diamond -- مسیریاب مبتنی بر یادگیری ماشین که الگوهای ترافیک شما را یاد می‌گیرد تا توازن هزینه و کیفیت را میان ارائه‌دهنده‌های مختلف برقرار سازد.
• Helicone -- پلتفرم مشاهده‌پذیری LLM به همراه ردگیری هزینه، کش، محدودیت نرخ درخواست و هشدارهای بودجه به‌عنوان یک لایه‌ی پروکسی.
• دین و باروسو، «دم توزیع در مقیاس بالا» (CACM 2013) -- مباحث مربوط به تاخیر، نرخ پردازش (Throughput)، صدک‌های TTFT/TPOT و درخواست‌های پوششی؛ الگوی هزینه‌ای که پشت رویکرد «انتخاب ارزان‌ترین مدلی که نیاز صدک 95 را برآورده می‌کند» قرار دارد.
• کوان و همکاران، «مدیریت حافظه‌ی بهینه برای ارائه‌ی مدل‌های زبانی بزرگ به کمک PagedAttention» (SOSP 2023) -- مقاله‌ی اصلی ابزار vLLM؛ چرایی برتری 24 برابری کش صفحه‌بندی‌شده‌ی KV به همراه دسته‌بندی پیوسته (Continuous Batching) بر سرورهای ساده در نرخ پردازش؛ لایه‌ی زیرساخت زیر مبحث «کش و بهینه‌سازی هزینه».
• دائو و همکاران، «FlashAttention-2: اتنشن سریع‌تر با موازی‌سازی و تقسیم کار بهتر» (ICLR 2024) -- کاهش هزینه‌ها در سطح کرنل (Kernel) در کنار کش کردن پرامپت؛ این مقاله را همراه با دیکودینگ حدسی (Speculative Decoding) و GQA بخوانید تا تصویر کاملی از منحنی بهینه‌سازی هزینه‌ها به دست آورید.