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

Inwestuj tokeny tam, gdzie naprawdę się liczą.

Page content

Koszty LLM rosną liniowo wraz z użyciem. System przetwarzający 10 000 zapytań dziennie po cenie 0,01 USD za zapytanie kosztuje 100 USD dziennie — czyli 365 USD rocznie. W skali przedsiębiorczej to ponad 10 000 USD.

Optymalizacja kosztów nie polega na oszczędzaniu w sposób krótkowzroczny. Chodzi o wydawanie tokenów tam, gdzie mają one 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 ustawianie limitów. Na sesję, na zadanie lub na dzień.

Strategia 1: Budżety na sesję

Budżety na sesję są proste:

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óżnych ilości kontekstu:

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

Strategia 3: Budżety adaptacyjne

Budżety adaptacyjne dostosowują się do tego, co naprawdę się dzieje. Jeśli zadania klasyfikacji konsekwentnie zużywają 80 tokenów, przestań alokować 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
            )

Eksponencjalnie ważona średnia ruchoma (waga 0,9) oznacza, że najnowsze użycie ma większe znaczenie niż historia. Dostosuj wagę w zależności od zmienności Twoich obciążeń.

Interfejs API vs wnioskowanie lokalne

Wnioskowanie lokalne jest tańsze w skali. Punkt bezstronności zależy od Twojego sprzętu i stawek API.

Model API (USD/mil. tokenów) Koszt lokalny/godz. Punkt bezstronności
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ń
qwen3-32b 0,30 / 1,20 ~0,20 ~2 godziny/dzień
qwen3-8b 0,10 / 0,40 ~0,05 ~1 godzina/dzień

Matematyka sprzętu:

Sprzęt Koszt początkowy Miesięczny rachunek za prąd Punkt bezstronności vs API
RTX 3090 (używany) 600 USD 15 USD ~4 miesiące
RTX 4090 1 500 USD 20 USD ~6 miesięcy
RTX 5080 1 000 USD 18 USD ~5 miesięcy
DGX Spark 2 000 USD 30 USD ~8 miesięcy

Przy umiarkowanym użyciu — godzinę lub więcej dziennie — wnioskowanie lokalne się zwraca. Przy dużym użyciu oszczędności są ogromne. Haczyk to kapitał początkowy. RTX 5080 kosztuje 1 000 USD. Rachunek za API możesz wstrzymać. Sprzętu nie.

Strategie rezerwowe

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

Strategia 1: Fallback oparty na jakości

Fallback oparty na jakości próbuje modeli, aż wyjście osiągnie próg jakości:

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": "qwen3-32b", "cost": 0.001},
            {"model": "qwen3-8b", "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 ewaluacja. Jak zmierzyć jakość bez wołania innego modelu? Niektóre systemy używają małego klasyfikatora. Inne stosują heurystyczne sprawdzenia — długość, struktura, obecność słów kluczowych. Żadna z tych metod nie jest idealna.

Strategia 2: Fallback oparty na opóźnieniu

Fallback oparty na opóźnieniu jest prostszy. 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": "qwen3-1.7b", "latency": 0.5},
            {"model": "qwen3-8b", "latency": 2.0},
            {"model": "qwen3-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)

Cache (buforowanie)

Cache to najbardziej niedoceniona optymalizacja kosztów. Identyczne prompty pojawiają się częściej, niż myślisz — zapytania klasyfikacyjne, pytania w stylu FAQ, powtarzające się wywołania narzędzi.

Strategia 1: Cache promptów

Dokładne cache’owanie 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: Cache semantyczny

Cache semantyczny jest bardziej przydatny. Chwytá 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 — tylko bardzo podobne prompty się dopasowują. 0,85 jest bardziej wyrozumiały, ale ryzykuje zwrócenie błędnych odpowiedzi. Mierz swoją stopę trafień i dostosuj.

Cache’owanie odpowiedzi dla popularnych zapytań też się opłaca. Jeśli użytkownicy wielokrotnie pytają „jaka jest pogoda” lub „która jest godzina”, buferuj wzorzec, a nie tylko dokładny prompt:

class ResponseCache:
    def __init__(self):
        self.common_queries = {
            "what is the weather": "Check weather API",
            "what is the time": "Check system time",
            "who is the president": "Check current president",
        }

    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 zaawansowane, ale działa. Popularne zapytania są popularne z powodu.

Kiedy optymalizacja pomaga

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

Nie ma znaczenia, gdy prototypujesz, używasz jednego modelu lub przetwarzasz małe objętości. Złożoność budżetowania, fallbacków i cache’owania nie jest warta wysiłku dla systemu, który wykonuje 100 zapytań dziennie.

Najpierw skonstruuj podstawowy przepływ. Dodaj optymalizację, gdy przyjdzie rachunek.

Kompromisy

Strategia Koszt Jakość Złożoność
Bez optymalizacji Najwyższy Stała Najniższa
Budżetowanie tokenów Umiarkowany Zmienna Średnia
Modele rezerwowe Niski-Średni Zmienna Średnia
Cache Najniższy Wysoka (dla trafień w cache) Średnia
Hybryda Optymalizowany Optymalizowany Najwyższa

Systemy produkcyjne zwykle działają hybrydowo. Budżetuj na sesję, korzystaj z fallbacku opartego na jakości lub opóźnieniu, buferuj, 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.