Gestion et optimisation des coûts pour applications LangChain : tracking, budgets et ROI

# ❌ DANGER : Coûts non maîtrisés
from langchain_openai import ChatOpenAI

# GPT-4 : $0.03/1K tokens input, $0.06/1K tokens output
llm = ChatOpenAI(model="gpt-4")

# Requête avec contexte massif (10K tokens)
huge_context = "..." * 5000  # 10K tokens
result = llm.invoke(huge_context + "Résume ce texte")

# Coût: (10K × $0.03/1K) + (1K × $0.06/1K) = $0.36 par requête !
# 100K requêtes/jour = $36K/jour = $1M/mois ! 💸

BREAKDOWN DES COÛTS MENSUELS ($10,000)
├─ System prompts (20%)              : $2,000
│   → Prompts longs répétés chaque requête
│   → Solution: Caching system prompt
│
├─ Contexte RAG (35%)                : $3,500
│   → Documents récupérés (souvent trop longs)
│   → Solution: Compression documents, reranking
│
├─ Outputs longs (30%)               : $3,000
│   → Réponses verbeuses non nécessaires
│   → Solution: max_tokens, "Sois concis"
│
├─ Appels redondants (10%)           : $1,000
│   → Mêmes questions répétées
│   → Solution: Caching sémantique
│
└─ Debugging/développement (5%)      : $500
    → Tests, expérimentations
    → Solution: Environnement séparé

# cost_tracking/token_counter.py
from typing import Dict
import tiktoken

class TokenCounter:
    """Compte les tokens avec précision"""

    # Prix par modèle ($/1K tokens)
    PRICING = {
        "gpt-4": {"input": 0.03, "output": 0.06},
        "gpt-4-32k": {"input": 0.06, "output": 0.12},
        "gpt-3.5-turbo": {"input": 0.0015, "output": 0.002},
        "claude-3-opus": {"input": 0.015, "output": 0.075},
        "claude-3-sonnet": {"input": 0.003, "output": 0.015},
    }

    def __init__(self, model: str = "gpt-4"):
        self.model = model
        self.encoding = tiktoken.encoding_for_model(model)

    def count_tokens(self, text: str) -> int:
        """Compte les tokens d'un texte"""
        return len(self.encoding.encode(text))

    def estimate_cost(
        self,
        input_text: str,
        output_text: str
    ) -> Dict:
        """Estime le coût d'une requête"""
        input_tokens = self.count_tokens(input_text)
        output_tokens = self.count_tokens(output_text)

        pricing = self.PRICING.get(self.model, self.PRICING["gpt-4"])

        input_cost = (input_tokens / 1000) * pricing["input"]
        output_cost = (output_tokens / 1000) * pricing["output"]
        total_cost = input_cost + output_cost

        return {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "input_cost": input_cost,
            "output_cost": output_cost,
            "total_cost": total_cost,
            "model": self.model
        }

# cost_tracking/tracked_chain.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from cost_tracking.token_counter import TokenCounter
import time

class CostTrackedChain:
    """Chain avec tracking automatique des coûts"""

    def __init__(self, model: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model, temperature=0)
        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")
        self.chain = self.prompt | self.llm
        self.counter = TokenCounter(model)

        # Stats
        self.total_requests = 0
        self.total_cost = 0.0
        self.total_tokens = 0

    def invoke(self, query: str, metadata: Dict = None) -> Dict:
        """Invoke avec tracking des coûts"""
        start = time.time()

        # Préparer l'input
        input_text = self.prompt.format(query=query)

        # Invoquer
        result = self.chain.invoke({"query": query})
        output_text = result.content

        # Calculer coûts
        costs = self.counter.estimate_cost(input_text, output_text)

        # Mettre à jour stats
        self.total_requests += 1
        self.total_cost += costs["total_cost"]
        self.total_tokens += costs["total_tokens"]

        return {
            "response": output_text,
            "costs": costs,
            "latency_ms": (time.time() - start) * 1000,
            "metadata": metadata or {}
        }

    def get_stats(self) -> Dict:
        """Statistiques globales"""
        return {
            "total_requests": self.total_requests,
            "total_cost": self.total_cost,
            "total_tokens": self.total_tokens,
            "avg_cost_per_request": self.total_cost / self.total_requests if self.total_requests > 0 else 0,
            "avg_tokens_per_request": self.total_tokens / self.total_requests if self.total_requests > 0 else 0
        }

# cost_tracking/user_cost_tracker.py
from collections import defaultdict
from datetime import datetime
from typing import Dict

class UserCostTracker:
    """Track coûts par utilisateur"""

    def __init__(self):
        self.user_costs = defaultdict(lambda: {
            "total_cost": 0.0,
            "total_requests": 0,
            "total_tokens": 0,
            "requests": []
        })

    def track_request(
        self,
        user_id: str,
        costs: Dict,
        metadata: Dict = None
    ):
        """Enregistre une requête pour un utilisateur"""
        user_data = self.user_costs[user_id]

        user_data["total_cost"] += costs["total_cost"]
        user_data["total_requests"] += 1
        user_data["total_tokens"] += costs["total_tokens"]

        user_data["requests"].append({
            "timestamp": datetime.now().isoformat(),
            "cost": costs["total_cost"],
            "tokens": costs["total_tokens"],
            "metadata": metadata or {}
        })

    def get_user_costs(self, user_id: str) -> Dict:
        """Coûts d'un utilisateur"""
        return self.user_costs[user_id]

    def get_top_users(self, limit: int = 10) -> list:
        """Top utilisateurs par coût"""
        sorted_users = sorted(
            self.user_costs.items(),
            key=lambda x: x[1]["total_cost"],
            reverse=True
        )
        return sorted_users[:limit]

    def get_total_cost(self) -> float:
        """Coût total tous utilisateurs"""
        return sum(
            data["total_cost"]
            for data in self.user_costs.values()
        )

# cost_tracking/feature_cost_tracker.py
from collections import defaultdict

class FeatureCostTracker:
    """Track coûts par feature"""

    def __init__(self):
        self.feature_costs = defaultdict(lambda: {
            "total_cost": 0.0,
            "total_requests": 0,
            "avg_cost": 0.0
        })

    def track(self, feature_name: str, cost: float):
        """Track coût pour une feature"""
        feature = self.feature_costs[feature_name]
        feature["total_cost"] += cost
        feature["total_requests"] += 1
        feature["avg_cost"] = feature["total_cost"] / feature["total_requests"]

    def get_breakdown(self) -> Dict:
        """Breakdown des coûts par feature"""
        total = sum(f["total_cost"] for f in self.feature_costs.values())

        breakdown = {}
        for feature, data in self.feature_costs.items():
            breakdown[feature] = {
                **data,
                "percentage": (data["total_cost"] / total * 100) if total > 0 else 0
            }

        return breakdown

    def get_most_expensive(self, limit: int = 5) -> list:
        """Features les plus coûteuses"""
        sorted_features = sorted(
            self.feature_costs.items(),
            key=lambda x: x[1]["total_cost"],
            reverse=True
        )
        return sorted_features[:limit]

# cost_tracking/budget_limiter.py
from datetime import datetime, timedelta
from typing import Optional

class BudgetLimiter:
    """Enforce budgets par période"""

    def __init__(
        self,
        daily_budget: float = 100.0,
        monthly_budget: float = 3000.0,
        per_user_daily_limit: float = 10.0
    ):
        self.daily_budget = daily_budget
        self.monthly_budget = monthly_budget
        self.per_user_daily_limit = per_user_daily_limit

        self.daily_spent = 0.0
        self.monthly_spent = 0.0
        self.last_reset_day = datetime.now().date()
        self.last_reset_month = datetime.now().month

        self.user_daily_spent = {}

    def can_process(
        self,
        cost: float,
        user_id: Optional[str] = None
    ) -> tuple[bool, Optional[str]]:
        """
        Vérifie si on peut traiter la requête
        Returns: (can_process, error_message)
        """
        self._reset_if_needed()

        # Check budget global journalier
        if self.daily_spent + cost > self.daily_budget:
            return False, f"Budget journalier dépassé (${self.daily_budget})"

        # Check budget global mensuel
        if self.monthly_spent + cost > self.monthly_budget:
            return False, f"Budget mensuel dépassé (${self.monthly_budget})"

        # Check budget par utilisateur
        if user_id:
            user_spent = self.user_daily_spent.get(user_id, 0.0)
            if user_spent + cost > self.per_user_daily_limit:
                return False, f"Limite utilisateur dépassée (${self.per_user_daily_limit}/jour)"

        return True, None

    def record_cost(self, cost: float, user_id: Optional[str] = None):
        """Enregistre un coût"""
        self.daily_spent += cost
        self.monthly_spent += cost

        if user_id:
            self.user_daily_spent[user_id] = self.user_daily_spent.get(user_id, 0.0) + cost

    def _reset_if_needed(self):
        """Reset les compteurs si période écoulée"""
        now = datetime.now()

        # Reset journalier
        if now.date() > self.last_reset_day:
            self.daily_spent = 0.0
            self.user_daily_spent = {}
            self.last_reset_day = now.date()

        # Reset mensuel
        if now.month != self.last_reset_month:
            self.monthly_spent = 0.0
            self.last_reset_month = now.month

    def get_remaining(self) -> Dict:
        """Budget restant"""
        return {
            "daily_remaining": self.daily_budget - self.daily_spent,
            "monthly_remaining": self.monthly_budget - self.monthly_spent,
            "daily_percentage_used": (self.daily_spent / self.daily_budget * 100),
            "monthly_percentage_used": (self.monthly_spent / self.monthly_budget * 100)
        }

# chains/budget_controlled_chain.py
from cost_tracking.budget_limiter import BudgetLimiter
from cost_tracking.tracked_chain import CostTrackedChain

class BudgetControlledChain:
    """Chain avec enforcement de budget"""

    def __init__(self, model: str = "gpt-4"):
        self.chain = CostTrackedChain(model)
        self.budget_limiter = BudgetLimiter(
            daily_budget=100.0,
            monthly_budget=3000.0,
            per_user_daily_limit=10.0
        )

    def invoke(self, query: str, user_id: str) -> Dict:
        """Invoke avec vérification budget"""
        # Estimer le coût (approximatif)
        estimated_cost = len(query) * 0.0001  # Rough estimate

        # Vérifier budget
        can_process, error = self.budget_limiter.can_process(
            estimated_cost,
            user_id
        )

        if not can_process:
            return {
                "error": error,
                "budget_exceeded": True,
                "remaining": self.budget_limiter.get_remaining()
            }

        # Traiter
        result = self.chain.invoke(query, metadata={"user_id": user_id})

        # Enregistrer coût réel
        self.budget_limiter.record_cost(
            result["costs"]["total_cost"],
            user_id
        )

        result["remaining_budget"] = self.budget_limiter.get_remaining()
        return result

# optimization/semantic_cache.py
from langchain.cache import SQLiteCache
from langchain.embeddings import OpenAIEmbeddings
import numpy as np
from typing import Optional

class SemanticCache:
    """Cache sémantique pour requêtes similaires"""

    def __init__(self, similarity_threshold: float = 0.95):
        self.embeddings = OpenAIEmbeddings()
        self.cache = {}  # {query_hash: (embedding, response, cost)}
        self.similarity_threshold = similarity_threshold
        self.hits = 0
        self.misses = 0

    def get(self, query: str) -> Optional[Dict]:
        """Cherche dans le cache par similarité"""
        # Calculer embedding de la query
        query_embedding = self.embeddings.embed_query(query)

        # Chercher similarité avec queries en cache
        for cached_query, (cached_embedding, response, cost) in self.cache.items():
            similarity = self._cosine_similarity(query_embedding, cached_embedding)

            if similarity >= self.similarity_threshold:
                self.hits += 1
                return {
                    "response": response,
                    "cached": True,
                    "similarity": similarity,
                    "cost_saved": cost
                }

        self.misses += 1
        return None

    def set(self, query: str, response: str, cost: float):
        """Ajoute au cache"""
        embedding = self.embeddings.embed_query(query)
        self.cache[query] = (embedding, response, cost)

    def _cosine_similarity(self, a, b) -> float:
        """Similarité cosinus"""
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def get_stats(self) -> Dict:
        """Statistiques du cache"""
        total = self.hits + self.misses
        hit_rate = (self.hits / total * 100) if total > 0 else 0

        total_saved = sum(cost for _, _, cost in self.cache.values()) * self.hits

        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": hit_rate,
            "cache_size": len(self.cache),
            "total_cost_saved": total_saved
        }

# optimization/redis_cost_cache.py
import redis
import json
import hashlib

class RedisCostCache:
    """Cache Redis avec tracking des économies"""

    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.cost_saved = 0.0

    def get(self, query: str) -> Optional[Dict]:
        """Récupère du cache"""
        key = self._hash_query(query)
        cached = self.redis.get(key)

        if cached:
            data = json.loads(cached)
            self.cost_saved += data.get("cost", 0)
            return {
                **data,
                "cached": True
            }

        return None

    def set(self, query: str, response: str, cost: float, ttl: int = 3600):
        """Ajoute au cache avec TTL"""
        key = self._hash_query(query)
        data = {
            "response": response,
            "cost": cost,
            "timestamp": time.time()
        }
        self.redis.setex(key, ttl, json.dumps(data))

    def _hash_query(self, query: str) -> str:
        """Hash de la query pour clé Redis"""
        return hashlib.sha256(query.encode()).hexdigest()

    def get_savings(self) -> float:
        """Total des économies"""
        return self.cost_saved

# optimization/prompt_optimizer.py
from langchain.prompts import ChatPromptTemplate

class PromptOptimizer:
    """Optimise les prompts pour réduire tokens"""

    @staticmethod
    def compress_context(text: str, max_tokens: int = 2000) -> str:
        """Compresse le contexte pour réduire tokens"""
        # Supprimer whitespace excessif
        text = " ".join(text.split())

        # Estimer tokens (rough)
        estimated_tokens = len(text.split()) * 1.3

        if estimated_tokens <= max_tokens:
            return text

        # Tronquer intelligemment
        words = text.split()
        target_words = int(max_tokens / 1.3)

        # Garder début et fin
        half = target_words // 2
        compressed = words[:half] + ["[...]"] + words[-half:]

        return " ".join(compressed)

    @staticmethod
    def create_efficient_prompt(
        task: str,
        context: Optional[str] = None,
        max_context_tokens: int = 2000
    ) -> str:
        """Crée un prompt efficace"""
        # Compresser contexte si nécessaire
        if context:
            context = PromptOptimizer.compress_context(
                context,
                max_context_tokens
            )
            return f"{task}\n\nContexte: {context}"

        return task

    @staticmethod
    def estimate_cost(
        prompt: str,
        expected_output_tokens: int = 500,
        model: str = "gpt-4"
    ) -> float:
        """Estime le coût d'un prompt"""
        from cost_tracking.token_counter import TokenCounter

        counter = TokenCounter(model)
        input_tokens = counter.count_tokens(prompt)

        pricing = counter.PRICING[model]
        input_cost = (input_tokens / 1000) * pricing["input"]
        output_cost = (expected_output_tokens / 1000) * pricing["output"]

        return input_cost + output_cost

# optimization/model_router.py
from enum import Enum

class QueryComplexity(Enum):
    SIMPLE = "simple"        # GPT-3.5
    MODERATE = "moderate"    # GPT-3.5 ou GPT-4
    COMPLEX = "complex"      # GPT-4

class ModelRouter:
    """Route vers le modèle optimal selon complexité"""

    MODEL_COSTS = {
        "gpt-3.5-turbo": 0.002,  # $/1K tokens (approx)
        "gpt-4": 0.045,           # $/1K tokens (approx)
    }

    @staticmethod
    def assess_complexity(query: str) -> QueryComplexity:
        """Évalue la complexité d'une query"""
        # Heuristiques simples
        word_count = len(query.split())
        has_context = "context" in query.lower() or "document" in query.lower()
        has_analysis = any(kw in query.lower() for kw in ["analyze", "explain", "compare"])

        if word_count > 100 or (has_context and has_analysis):
            return QueryComplexity.COMPLEX
        elif word_count > 30 or has_analysis:
            return QueryComplexity.MODERATE
        else:
            return QueryComplexity.SIMPLE

    @classmethod
    def route(cls, query: str) -> tuple[str, float]:
        """
        Route vers le modèle optimal
        Returns: (model_name, estimated_cost)
        """
        complexity = cls.assess_complexity(query)

        if complexity == QueryComplexity.SIMPLE:
            model = "gpt-3.5-turbo"
        elif complexity == QueryComplexity.MODERATE:
            # Préférer GPT-3.5 si possible
            model = "gpt-3.5-turbo"
        else:
            model = "gpt-4"

        # Estimer coût
        tokens_estimate = len(query.split()) * 1.5  # Rough
        cost_estimate = (tokens_estimate / 1000) * cls.MODEL_COSTS[model]

        return model, cost_estimate

# analytics/cost_forecaster.py
from datetime import datetime, timedelta
from typing import List
import statistics

class CostForecaster:
    """Prédictions de coûts"""

    def __init__(self):
        self.daily_costs = []  # [(date, cost), ...]

    def record_daily_cost(self, date: datetime.date, cost: float):
        """Enregistre coût journalier"""
        self.daily_costs.append((date, cost))

    def forecast_monthly(self) -> Dict:
        """Prévision du coût mensuel"""
        if len(self.daily_costs) < 7:
            return {"error": "Pas assez de données (min 7 jours)"}

        # Moyenne des 7 derniers jours
        recent_costs = [cost for _, cost in self.daily_costs[-7:]]
        avg_daily = statistics.mean(recent_costs)
        std_daily = statistics.stdev(recent_costs) if len(recent_costs) > 1 else 0

        # Projection mensuelle
        days_in_month = 30
        forecasted_monthly = avg_daily * days_in_month

        # Intervalle de confiance (± 2 std)
        monthly_std = std_daily * (days_in_month ** 0.5)
        confidence_low = forecasted_monthly - (2 * monthly_std)
        confidence_high = forecasted_monthly + (2 * monthly_std)

        return {
            "avg_daily_cost": avg_daily,
            "forecasted_monthly": forecasted_monthly,
            "confidence_interval": (confidence_low, confidence_high),
            "trend": self._calculate_trend()
        }

    def _calculate_trend(self) -> str:
        """Calcule la tendance (croissante, stable, décroissante)"""
        if len(self.daily_costs) < 14:
            return "insufficient_data"

        recent = [cost for _, cost in self.daily_costs[-7:]]
        older = [cost for _, cost in self.daily_costs[-14:-7]]

        recent_avg = statistics.mean(recent)
        older_avg = statistics.mean(older)

        change = ((recent_avg - older_avg) / older_avg) * 100

        if change > 10:
            return "increasing"
        elif change < -10:
            return "decreasing"
        else:
            return "stable"

    def should_alert(self, threshold: float = 20.0) -> tuple[bool, Optional[str]]:
        """Détermine si on doit alerter"""
        if len(self.daily_costs) < 7:
            return False, None

        forecast = self.forecast_monthly()

        if forecast["trend"] == "increasing":
            recent_avg = statistics.mean([cost for _, cost in self.daily_costs[-7:]])
            older_avg = statistics.mean([cost for _, cost in self.daily_costs[-14:-7]])
            increase = ((recent_avg - older_avg) / older_avg) * 100

            if increase > threshold:
                return True, f"Coûts en hausse de {increase:.1f}% sur 7 jours"

        return False, None

# analytics/cost_reporter.py
from datetime import datetime, timedelta
from typing import Dict, List

class CostReporter:
    """Génère rapports de coûts"""

    def __init__(
        self,
        user_tracker,
        feature_tracker,
        budget_limiter
    ):
        self.user_tracker = user_tracker
        self.feature_tracker = feature_tracker
        self.budget_limiter = budget_limiter

    def generate_daily_report(self) -> Dict:
        """Rapport journalier"""
        return {
            "date": datetime.now().date().isoformat(),
            "total_cost": self.user_tracker.get_total_cost(),
            "budget_remaining": self.budget_limiter.get_remaining(),
            "top_users": self.user_tracker.get_top_users(5),
            "top_features": self.feature_tracker.get_most_expensive(5),
            "feature_breakdown": self.feature_tracker.get_breakdown()
        }

    def generate_monthly_report(self) -> Dict:
        """Rapport mensuel complet"""
        return {
            "month": datetime.now().strftime("%Y-%m"),
            "total_cost": self.budget_limiter.monthly_spent,
            "budget": self.budget_limiter.monthly_budget,
            "budget_used_percentage": (
                self.budget_limiter.monthly_spent /
                self.budget_limiter.monthly_budget * 100
            ),
            "total_users": len(self.user_tracker.user_costs),
            "total_requests": sum(
                data["total_requests"]
                for data in self.user_tracker.user_costs.values()
            ),
            "avg_cost_per_request": (
                self.user_tracker.get_total_cost() /
                sum(data["total_requests"] for data in self.user_tracker.user_costs.values())
            )
        }

    def export_csv(self, filename: str = "cost_report.csv"):
        """Exporte rapport en CSV"""
        import csv

        with open(filename, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(["User ID", "Total Cost", "Requests", "Avg Cost/Req"])

            for user_id, data in self.user_tracker.user_costs.items():
                writer.writerow([
                    user_id,
                    f"${data['total_cost']:.4f}",
                    data['total_requests'],
                    f"${data['total_cost'] / data['total_requests']:.4f}"
                ])

# analytics/roi_calculator.py
class ROICalculator:
    """Calcule le ROI des features IA"""

    @staticmethod
    def calculate_roi(
        feature_cost: float,
        revenue_generated: float,
        time_saved_hours: float = 0,
        hourly_rate: float = 50.0
    ) -> Dict:
        """
        Calcule ROI d'une feature

        Args:
            feature_cost: Coût total LLM de la feature
            revenue_generated: Revenus directs générés
            time_saved_hours: Heures économisées
            hourly_rate: Taux horaire pour valoriser temps
        """
        # Valeur du temps économisé
        time_value = time_saved_hours * hourly_rate

        # Bénéfice total
        total_benefit = revenue_generated + time_value

        # ROI
        roi = ((total_benefit - feature_cost) / feature_cost) * 100 if feature_cost > 0 else 0

        return {
            "feature_cost": feature_cost,
            "revenue_generated": revenue_generated,
            "time_saved_hours": time_saved_hours,
            "time_value": time_value,
            "total_benefit": total_benefit,
            "net_profit": total_benefit - feature_cost,
            "roi_percentage": roi,
            "roi_multiple": total_benefit / feature_cost if feature_cost > 0 else 0
        }

    @staticmethod
    def breakeven_analysis(
        monthly_cost: float,
        revenue_per_user: float,
        conversion_rate: float = 0.05
    ) -> Dict:
        """Analyse de seuil de rentabilité"""
        # Users nécessaires pour breakeven
        users_needed = monthly_cost / (revenue_per_user * conversion_rate)

        return {
            "monthly_cost": monthly_cost,
            "revenue_per_user": revenue_per_user,
            "conversion_rate": conversion_rate,
            "users_needed_for_breakeven": users_needed,
            "monthly_revenue_needed": users_needed * revenue_per_user * conversion_rate
        }

# System prompt long (2000 tokens)
system_prompt = """Tu es un expert en code Python...""" * 100

# Requête 1 : 2000 tokens input @ $0.03/1k = $0.06
# Requêtes 2-N : 2000 tokens @ $0.015/1k = $0.03  (-50%)

from openai import OpenAI
client = OpenAI()

# Créer batch de 1000 requêtes
batch = client.batches.create(
    input_file_id="file-xyz",
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

# Coût : -50% vs API temps réel
# 1000 requêtes @ $0.045 → $0.0225/req
# Économie : $22.50 pour 1000 requêtes

from openai import OpenAI

client = OpenAI()
stream = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Ce texte est-il positif ou négatif ?"}],
    stream=True,
    max_tokens=10  # Arrêt précoce
)

# Génère seulement "Positif" (7 tokens) au lieu de 500 tokens
# Économie : 98% sur output

from llmlingua import PromptCompressor

compressor = PromptCompressor()

long_context = """..."""  # 10k tokens
compressed = compressor.compress_prompt(
    long_context,
    target_token=2000,  # Réduire à 2k tokens
    condition_in_question="What is the main topic?",
    rate=0.8  # Taux compression 80%
)

# Résultat : 2k tokens au lieu de 10k
# Économie : 80% sur input RAG
# Accuracy maintenue : ~95%