Optymalizacja kosztów w systemach LLM: gdzie naprawdę idzie pieniądze

Inwestuj tokeny tam, gdzie naprawdę mają znaczenie.

Page content

Koszty LLM rosną liniowo wraz ze wzrostem wykorzystania. System przetwarzający 10 000 zapytań dziennie po cenie 0,01 USD za zapytanie kosztuje 100 USD dziennie — co daje 365 USD rocznie. W skali przedsiębiorczej kwota ta przekracza 10 000 USD.

Optymalizacja kosztów nie polega na oszczędzaniu w niewłaściwych miejscach. Chodzi o wydawanie tokenów tam, gdzie mają największe znaczenie.

Każdy token, który marnujesz, to token, który mógłbyś wydać na lepszą odpowiedź.

Strategie optymalizacji kosztów LLM

Budżetowanie tokenów

Najprostszym sposobem kontrolowania kosztów jest ustawienie limitów. Limitów na sesję, zadanie lub dzień.

Strategia 1: Budżety na sesję

Budżety na sesję są proste w zrozumieniu:

class SessionBudget:
    def __init__(self, budget_tokens: int = 10000):
        self.budget = budget_tokens
        self.used = 0

    def allocate(self, tokens: int) -> bool:
        if self.used + tokens <= self.budget:
            self.used += tokens
            return True
        return False

    def remaining(self) -> int:
        return self.budget - self.used

Strategia 2: Budżety na zadanie

Budżety na zadanie są bardziej przydatne. Różne zadania wymagają różnej ilości kontekstu:

task_budgets:
  classify:
    max_tokens: 100
    model: qwen2.5-1.5b
  summarize:
    max_tokens: 500
    model: qwen2.5-7b
  code_review:
    max_tokens: 2000
    model: qwen2.5-coder-7b
  reason:
    max_tokens: 4000
    model: qwen2.5-32b

Strategia 3: Budżety adaptacyjne

Budżety adaptacyjne dostosowują się do tego, co naprawdę się dzieje. Jeśli zadania klasyfikacyjne konsekwentnie zużywają 80 tokenów, przestań przydzielać 100:

class AdaptiveBudget:
    def __init__(self):
        self.task_history = {}

    def allocate(self, task_type: str) -> int:
        if task_type in self.task_history:
            return int(self.task_history[task_type] * 1.5)
        return 1000

    def record(self, task_type: str, tokens_used: int):
        if task_type not in self.task_history:
            self.task_history[task_type] = tokens_used
        else:
            self.task_history[task_type] = (
                0.9 * self.task_history[task_type] + 0.1 * tokens_used
            )

Średnia ruchoma wykładnicza (z wagą 0,9) oznacza, że ostatnie wykorzystanie ma większe znaczenie niż historia. Dostosuj wagę w zależności od zmienności Twoich obciążeń.

Interfejs API a wnioskowanie lokalne

Wnioskowanie lokalne jest tańsze w skali. Punkt przerwania (break-even) zależy od Twojego sprzętu i stawek API.

Model API (USD/miliona tokenów) Koszt lokalny/godzina Punkt przerwania
GPT-4o $2,50 / $10,00 — N/A
Claude Sonnet 4 $3,00 / $15,00 — N/A
Qwen2.5-72B $0,50 / $2,00 ~$0,50 ~4 godziny/dzień
Qwen2.5-32B $0,30 / $1,20 ~$0,20 ~2 godziny/dzień
Qwen2.5-7B $0,10 / $0,40 ~$0,05 ~1 godzina/dzień

Matematyka sprzętu:

Sprzęt Koszt początkowy Miesięczny koszt energii Punkt przerwania względem API
RTX 3090 (używany) $600 $15 ~4 miesiące
RTX 4090 $1 500 $20 ~6 miesięcy
RTX 5080 $1 000 $18 ~5 miesięcy
DGX Spark $2 000 $30 ~8 miesięcy

Przy umiarkowanym wykorzystaniu — godzinie lub więcej dziennie — wnioskowanie lokalne zwraca zainwestowane środki. Przy dużym obciążeniu oszczędności są drastyczne. Haczyk polega na kapitale początkowym. RTX 5080 kosztuje 1 000 USD. Fakturę za API można wstrzymać. Sprzętu nie.

Strategie awaryjne (fallback)

Gdy Twój preferowany model jest zbyt drogi lub zbyt wolny, przełącz się na coś tańszego. Kluczowe jest wiedza, kiedy jakość jest „wystarczająco dobra”.

Strategia 1: Awaryjne przełączanie oparte na jakości

Awaryjne przełączanie oparte na jakości próbuje modeli, dopóki wyjście nie osiągnie określonego progu:

class QualityFallback:
    def __init__(self, quality_threshold: float = 0.8):
        self.threshold = quality_threshold
        self.models = [
            {"model": "claude-sonnet-4", "cost": 0.015},
            {"model": "qwen2.5-72b", "cost": 0.002},
            {"model": "qwen2.5-32b", "cost": 0.001},
            {"model": "qwen2.5-7b", "cost": 0.0004},
        ]

    def route(self, prompt: str) -> str:
        for model_config in self.models:
            result = self.call_model(model_config["model"], prompt)
            if self.evaluate_quality(result) >= self.threshold:
                return result
        return self.call_model(self.models[0]["model"], prompt)

Problemem jest sama ocena. Jak zmierzyć jakość bez wywołania innego modelu? Niektóre systemy używają małego klasyfikatora. Inne stosują heurystyczne sprawdzenia — długość, strukturę, występowanie słów kluczowych. Żadne z tych rozwiązań nie jest idealne.

Strategia 2: Awaryjne przełączanie oparte na opóźnieniu

Awaryjne przełączanie oparte na opóźnieniu jest prostsze. Przekieruj do najszybszego modelu, który spełnia Twój budżet czasowy:

class LatencyFallback:
    def __init__(self, max_latency: float = 5.0):
        self.max_latency = max_latency
        self.models = [
            {"model": "qwen2.5-1.5b", "latency": 0.5},
            {"model": "qwen2.5-7b", "latency": 2.0},
            {"model": "qwen2.5-32b", "latency": 10.0},
            {"model": "claude-sonnet-4", "latency": 5.0},
        ]

    def route(self, prompt: str) -> str:
        for model_config in sorted(self.models, key=lambda x: x["latency"]):
            if model_config["latency"] <= self.max_latency:
                return self.call_model(model_config["model"], prompt)
        return self.call_model(self.models[0]["model"], prompt)

Buforowanie (Caching)

Buforowanie to najbardziej niedoceniona optymalizacja kosztów. Identyczne propty (prompty) występują częściej, niż myślisz — zapytania o klasyfikację, zapytania w stylu FAQ, powtarzające się wywołania narzędzi.

Strategia 1: Buforowanie promptów

Dokładne buforowanie promptów jest proste:

import hashlib

class PromptCache:
    def __init__(self, max_size: int = 1000):
        self.cache = {}
        self.max_size = max_size

    def get(self, prompt: str) -> str | None:
        key = hashlib.sha256(prompt.encode()).hexdigest()
        return self.cache.get(key)

    def set(self, prompt: str, response: str):
        key = hashlib.sha256(prompt.encode()).hexdigest()
        if len(self.cache) >= self.max_size:
            self.cache.pop(next(iter(self.cache)))
        self.cache[key] = response

Strategia 2: Buforowanie semantyczne

Buforowanie semantyczne jest bardziej przydatne. Łapie prompty, które są różne, ale oznaczają to samo:

from sentence_transformers import SentenceTransformer

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.cache = {}
        self.threshold = similarity_threshold

    def get(self, prompt: str) -> str | None:
        prompt_embedding = self.model.encode([prompt])[0]
        for cached_prompt, cached_response in self.cache.items():
            cached_embedding = self.model.encode([cached_prompt])[0]
            similarity = self.cosine_similarity(
                prompt_embedding, cached_embedding
            )
            if similarity >= self.threshold:
                return cached_response
        return None

    def set(self, prompt: str, response: str):
        self.cache[prompt] = response

Próg ma znaczenie. 0,95 jest agresywny — dopasowuje tylko bardzo podobne prompty. 0,85 jest bardziej wyrozumiały, ale ryzykuje zwrócenie błędnych odpowiedzi. Mierz swoją stopę trafień i dostosuj próg.

Buforowanie odpowiedzi dla typowych zapytań również się opłaca. Jeśli użytkownicy wielokrotnie pytają „jaka jest pogoda” lub „która jest godzina”, buforuj wzorzec, a nie tylko dokładny prompt:

class ResponseCache:
    def __init__(self):
        self.common_queries = {
            "what is the weather": "Sprawdź API pogody",
            "what is the time": "Sprawdź czas systemowy",
            "who is the president": "Sprawdź obecnego prezydenta",
        }

    def get(self, query: str) -> str | None:
        query_lower = query.lower()
        for common_query, response in self.common_queries.items():
            if common_query in query_lower:
                return response
        return None

To nie jest rozwiązanie zaawansowane, ale działa. Typowe zapytania są typowe z powodu.

Kiedy optymalizacja pomaga

Optymalizacja ma znaczenie, gdy przetwarzasz duże objętości danych, uruchamiasz mieszane obciążenia lub płacisz koszty API, które się sumują.

Nie ma znaczenia, gdy prototypujesz, używasz jednego modelu lub przetwarzasz małe objętości danych. Złożoność budżetowania, awaryjnego przełączania i buforowania nie jest warta wysiłku dla systemu, który generuje 100 zapytań dziennie.

Najpierw zadziałań podstawowy przepływ. Dodaj optymalizację, gdy przyjdzie faktura.

Zastosowania (Tradeoffs)

Strategia Koszt Jakość Złożoność
Brak optymalizacji Najwyższy Stała Najniższa
Budżetowanie tokenów Umiarkowany Zmienna Średnia
Modele awaryjne Niski-Średni Zmienna Średnia
Buforowanie Najniższy Wysoka (dla trafień w buforze) Średnia
Hybrydowa Zoptymalizowana Zoptymalizowana Najwyższa

Systemy produkcyjne zwykle działają w sposób hybrydowy. Budżetuj na sesję, używaj awaryjnego przełączania opartego na jakości lub opóźnieniu, buforuj to, co możesz. Złożoność jest realna, ale oszczędności też.

Powiązane

Subskrybuj

Otrzymuj nowe wpisy o systemach, infrastrukturze i inżynierii AI.