En tant qu'ingénieur qui a surveillé des factures API de plusieurs milliers de dollars par mois, je peux vous dire que la gestion du rate limiting n'est pas une option — c'est une nécessité absolue. Quand j'ai découvert que notre équipe dépassait les 50 000$ mensuels en appels API non optimisés, j'ai immédiatement implémenté des algorithmes de token bucket. Aujourd'hui, je vous partage tout ce que j'ai appris sur l'implémentation de ces algorithmes pour les API d'IA.
Pourquoi le Token Bucket est essentiel pour vos API IA en 2026
Les tarifs des modèles de langage ont considérablement évolué. Voici les prix de sortie actuels vérifiés au premier trimestre 2026 :
| Modèle IA | Prix sortie (USD/MTok) | Prix HolySheep (USD/MTok) | Économie |
|---|---|---|---|
| GPT-4.1 | 8,00 $ | 8,00 $ | Via ¥ (85%+ moins cher) |
| Claude Sonnet 4.5 | 15,00 $ | 15,00 $ | Via ¥ (85%+ moins cher) |
| Gemini 2.5 Flash | 2,50 $ | 2,50 $ | Via ¥ (85%+ moins cher) |
| DeepSeek V3.2 | 0,42 $ | 0,42 $ | Via ¥ (85%+ moins cher) |
Comparatif de coûts pour 10 millions de tokens/mois
| Modèle | Coût standard/mois | Coût via HolySheep (¥) | Économie mensuelle |
|---|---|---|---|
| GPT-4.1 | 80 $ | ~12 $ | 68 $ |
| Claude Sonnet 4.5 | 150 $ | ~22,50 $ | 127,50 $ |
| Gemini 2.5 Flash | 25 $ | ~3,75 $ | 21,25 $ |
| DeepSeek V3.2 | 4,20 $ | ~0,63 $ | 3,57 $ |
Sans gestion du rate limiting, ces coûts peuvent augmenter de 30 à 45% en raison des dépassements, des retry inefficaces et des appels redondants. L'algorithme token bucket devient votre allié stratégique.
Comprendre l'algorithme Token Bucket
Le token bucket est un algorithme de contrôle de débit qui permet de gérer efficacement les requêtes API. Voici son fonctionnement :
- Un compartiment (bucket) contient des jetons représentant les requêtes autorisées
- Les jetons sont générés à un rythme constant (rate)
- Chaque requête consomme un ou plusieurs jetons
- Si le compartiment est vide, la requête est bloquée ou mise en attente
Implémentation Python du Token Bucket
"""
Token Bucket Rate Limiter pour API IA
Implémentation complète avec support multi-buckets
"""
import time
import threading
from typing import Dict, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class TokenBucket:
"""
Implémentation thread-safe du Token Bucket Algorithm.
Attributs:
capacity: Nombre maximum de jetons dans le bucket
refill_rate: Jetons ajoutés par seconde
tokens: Jetons actuels (calculé dynamiquement)
last_refill: Timestamp du dernier rechargement
"""
capacity: float
refill_rate: float
tokens: float = field(init=False)
last_refill: float = field(init=False)
_lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
def __post_init__(self):
self.tokens = self.capacity
self.last_refill = time.monotonic()
def _refill(self) -> None:
"""Rafraîchit les jetons basés sur le temps écoulé."""
now = time.monotonic()
elapsed = now - self.last_refill
# Ajoute les jetons générés depuis la dernière vérification
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def consume(self, tokens_needed: float = 1.0, block: bool = False) -> Tuple[bool, float]:
"""
Tente de consommer des jetons du bucket.
Args:
tokens_needed: Nombre de jetons à consommer
block: Si True, attend que les jetons soient disponibles
Returns:
Tuple[bool, float]: (succès, temps d'attente en secondes)
"""
with self._lock:
if block:
wait_time = 0.0
while self.tokens < tokens_needed:
self._refill()
if self.tokens < tokens_needed:
tokens_deficit = tokens_needed - self.tokens
wait_time = tokens_deficit / self.refill_rate
time.sleep(min(wait_time, 0.1)) # Petit sleep pour éviter le busy-wait
self._refill()
self.tokens -= tokens_needed
return True, wait_time
self._refill()
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return True, 0.0
# Temps nécessaire pour avoir assez de jetons
tokens_deficit = tokens_needed - self.tokens
wait_time = tokens_deficit / self.refill_rate
return False, wait_time
class MultiModelRateLimiter:
"""
Gestionnaire de rate limiting pour multiples modèles IA.
Chaque modèle a ses propres limites de débit.
"""
def __init__(self):
self.buckets: Dict[str, TokenBucket] = {}
self._lock = threading.Lock()
# Configuration par défaut pour 2026 (en jetons/seconde)
self.default_config = {
"gpt-4.1": {"capacity": 100, "rate": 50}, # 50 tok/s
"claude-sonnet-4.5": {"capacity": 80, "rate": 40}, # 40 tok/s
"gemini-2.5-flash": {"capacity": 200, "rate": 100}, # 100 tok/s
"deepseek-v3.2": {"capacity": 150, "rate": 75}, # 75 tok/s
}
def register_model(
self,
model_id: str,
capacity: float,
rate: float
) -> None:
"""Enregistre un nouveau modèle avec ses limites."""
with self._lock:
self.buckets[model_id] = TokenBucket(
capacity=capacity,
refill_rate=rate
)
def set_defaults(self) -> None:
"""Configure les limites par défaut pour tous les modèles courants."""
for model_id, config in self.default_config.items():
self.register_model(
model_id,
config["capacity"],
config["rate"]
)
def acquire(
self,
model_id: str,
tokens: float = 1.0,
block: bool = False
) -> Tuple[bool, float]:
"""
Tente d'acquérir des jetons pour un modèle spécifique.
Returns:
Tuple[bool, float]: (acquisition_réussie, temps_attente)
"""
with self._lock:
if model_id not in self.buckets:
# Auto-création avec configuration par défaut
config = self.default_config.get(
model_id,
{"capacity": 50, "rate": 25}
)
self.register_model(model_id, config["capacity"], config["rate"])
return self.buckets[model_id].consume(tokens, block)
def get_status(self, model_id: str) -> Optional[Dict]:
"""Retourne le statut actuel d'un modèle."""
with self._lock:
if model_id not in self.buckets:
return None
bucket = self.buckets[model_id]
return {
"model": model_id,
"tokens_available": round(bucket.tokens, 2),
"capacity": bucket.capacity,
"fill_percentage": round(100 * bucket.tokens / bucket.capacity, 1),
"refill_rate": bucket.refill_rate
}
Exemple d'utilisation
if __name__ == "__main__":
limiter = MultiModelRateLimiter()
limiter.set_defaults()
# Test d'acquisition
success, wait = limiter.acquire("deepseek-v3.2", tokens=10)
print(f"DeepSeek V3.2: Succès={success}, Attente={wait:.3f}s")
# Vérifier le statut
status = limiter.get_status("deepseek-v3.2")
print(f"Statut DeepSeek: {status}")
Intégration avec l'API HolySheep
Maintenant, voici comment intégrer ce rate limiter avec l'API HolySheep pour une latence inférieure à 50ms et des économies maximales :
"""
Client API IA avec Token Bucket et HolySheep
Optimisé pour la gestion des coûts et des limites de débit
"""
import requests
import time
import json
from typing import Dict, List, Optional, Any, Union
from concurrent.futures import ThreadPoolExecutor, as_completed
=== Configuration HolySheep ===
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"models": {
"gpt-4.1": {
"cost_per_mtok": 8.00, # USD
"rate_limit_tpm": 500_000, # tokens par minute
"context_window": 128_000,
},
"claude-sonnet-4.5": {
"cost_per_mtok": 15.00,
"rate_limit_tpm": 400_000,
"context_window": 200_000,
},
"gemini-2.5-flash": {
"cost_per_mtok": 2.50,
"rate_limit_tpm": 1_000_000,
"context_window": 1_000_000,
},
"deepseek-v3.2": {
"cost_per_mtok": 0.42,
"rate_limit_tpm": 600_000,
"context_window": 640_000,
},
},
"timeout": 30, # secondes
"max_retries": 3,
"retry_delay": 1.0, # secondes avec backoff exponentiel
}
class HolySheepAIClient:
"""
Client pour l'API HolySheep avec gestion intégrée du Token Bucket.
Avantages HolySheep:
- Taux de change ¥1=$1 (économie 85%+ vs prix USD directs)
- Paiement via WeChat/Alipay
- Latence moyenne < 50ms
- Crédits gratuits pour les nouveaux inscrits
- Support technique en français 24/7
"""
def __init__(
self,
api_key: str,
rate_limiter: Optional[Any] = None,
auto_configure_bucket: bool = True
):
"""
Initialise le client HolySheep.
Args:
api_key: Clé API HolySheep (格式: YOUR_HOLYSHEEP_API_KEY)
rate_limiter: Instance du rate limiter (optionnel)
auto_configure_bucket: Configure automatiquement les buckets
"""
self.api_key = api_key
self.base_url = HOLYSHEEP_CONFIG["base_url"]
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.rate_limiter = rate_limiter
self._usage_stats = defaultdict(lambda: {
"requests": 0,
"input_tokens": 0,
"output_tokens": 0,
"total_cost_usd": 0.0
})
if auto_configure_bucket and rate_limiter:
self._configure_rate_limits()
def _configure_rate_limits(self) -> None:
"""Configure les limites de débit basées sur les specs HolySheep."""
if self.rate_limiter is None:
return
for model_id, config in HOLYSHEEP_CONFIG["models"].items():
# Convertir TPM en jetons par seconde
tokens_per_second = config["rate_limit_tpm"] / 60
# Capacité initiale = 2 secondes de burst
capacity = tokens_per_second * 2
self.rate_limiter.register_model(model_id, capacity, tokens_per_second)
def _calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Calcule le coût USD de la requête."""
model_config = HOLYSHEEP_CONFIG["models"].get(model, {})
cost_per_mtok = model_config.get("cost_per_mtok", 1.0)
total_tokens = (input_tokens + output_tokens) / 1_000_000
return total_tokens * cost_per_mtok
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None,
stream: bool = False,
wait_if_limited: bool = True
) -> Dict[str, Any]:
"""
Envoie une requête de chat completion à HolySheep.
Args:
model: Identifiant du modèle
messages: Liste des messages de conversation
temperature: Température de génération (0-2)
max_tokens: Limite de tokens de sortie
stream: Mode streaming
wait_if_limited: Attend si rate limit atteint
Returns:
Réponse de l'API au format standard
"""
# Gestion du rate limiting
if self.rate_limiter:
estimated_tokens = max_tokens or 2048
acquired, wait_time = self.rate_limiter.acquire(
model,
tokens=estimated_tokens,
block=wait_if_limited
)
if not acquired and not wait_if_limited:
raise RateLimitExceeded(
f"Rate limit atteint pour {model}. "
f"Réessayez dans {wait_time:.2f}s"
)
# Préparation de la requête
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"stream": stream
}
if max_tokens:
payload["max_tokens"] = max_tokens
# Envoi avec retry automatique
last_error = None
for attempt in range(HOLYSHEEP_CONFIG["max_retries"]):
try:
start_time = time.time()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=HOLYSHEEP_CONFIG["timeout"],
stream=stream
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
# Mise à jour des statistiques
usage = result.get("usage", {})
input_tok = usage.get("prompt_tokens", 0)
output_tok = usage.get("completion_tokens", 0)
cost = self._calculate_cost(model, input_tok, output_tok)
self._usage_stats[model]["requests"] += 1
self._usage_stats[model]["input_tokens"] += input_tok
self._usage_stats[model]["output_tokens"] += output_tok
self._usage_stats[model]["total_cost_usd"] += cost
self._usage_stats[model]["last_latency_ms"] = latency_ms
result["_internal"] = {
"latency_ms": latency_ms,
"cost_usd": cost,
"holy_sheep_rate": True
}
return result
elif response.status_code == 429:
# Rate limit côté serveur
retry_after = int(response.headers.get("Retry-After", 60))
time.sleep(retry_after)
continue
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
last_error = e
if attempt < HOLYSHEEP_CONFIG["max_retries"] - 1:
sleep_time = HOLYSHEEP_CONFIG["retry_delay"] * (2 ** attempt)
time.sleep(sleep_time)
raise RuntimeError(f"Échec après {HOLYSHEEP_CONFIG['max_retries']} tentatives: {last_error}")
def batch_completion(
self,
model: str,
prompts: List[str],
max_workers: int = 5,
**kwargs
) -> List[Dict[str, Any]]:
"""
Traite plusieurs prompts en parallèle avec gestion du rate limiting.
Args:
model: Modèle à utiliser
prompts: Liste des prompts
max_workers: Nombre de threads parallèles
Returns:
Liste des réponses
"""
results = [None] * len(prompts)
def process_prompt(index: int, prompt: str) -> tuple:
messages = [{"role": "user", "content": prompt}]
response = self.chat_completion(model, messages, **kwargs)
return index, response
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(process_prompt, i, p): i
for i, p in enumerate(prompts)
}
for future in as_completed(futures):
try:
index, response = future.result()
results[index] = response
except Exception as e:
print(f"Erreur pour le prompt {futures[future]}: {e}")
return results
def get_cost_summary(self) -> Dict[str, Any]:
"""Retourne un résumé des coûts par modèle."""
summary = {}
for model, stats in self._usage_stats.items():
cost_usd = stats["total_cost_usd"]
# Convertir en coût ¥ pour HolySheep (85%+ moins cher)
cost_yuan = cost_usd * 0.15 # Taux approximatif
summary[model] = {
"requests": stats["requests"],
"total_tokens": stats["input_tokens"] + stats["output_tokens"],
"cost_usd": round(cost_usd, 4),
"cost_yuan_approx": round(cost_yuan, 2),
"savings_vs_direct": round(cost_usd - cost_yuan, 2),
"avg_latency_ms": stats.get("last_latency_ms", 0)
}
return summary
class RateLimitExceeded(Exception):
"""Exception levée quand le rate limit est atteint."""
pass
=== Exemple d'utilisation ===
if __name__ == "__main__":
# Initialisation avec rate limiter
limiter = MultiModelRateLimiter()
limiter.set_defaults()
# Configuration du client HolySheep
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY", # Remplacez par votre clé
rate_limiter=limiter,
auto_configure_bucket=True
)
# Exemple de requête
messages = [
{"role": "system", "content": "Tu es un assistant technique expert."},
{"role": "user", "content": "Explique l'algorithme token bucket en français."}
]
try:
response = client.chat_completion(
model="deepseek-v3.2",
messages=messages,
max_tokens=500
)
print("Réponse:", response["choices"][0]["message"]["content"])
print(f"Latence: {response['_internal']['latency_ms']:.1f}ms")
print(f"Coût: {response['_internal']['cost_usd']:.4f}$ USD")
except RateLimitExceeded as e:
print(f"Rate limit atteint: {e}")
# Afficher le résumé des coûts
print("\n=== Résumé des coûts ===")
for model, stats in client.get_cost_summary().items():
print(f"{model}: {stats['cost_yuan_approx']}¥ (~{stats['cost_usd']:.4f}$)")
Implémentation JavaScript/TypeScript pour Node.js
/**
* Token Bucket Rate Limiter - Implémentation TypeScript
* Compatible Node.js et navigateurs modernes
*/
interface BucketConfig {
capacity: number;
refillRate: number;
lastRefill: number;
tokens: number;
}
interface RateLimitConfig {
requestsPerSecond: number;
burstCapacity: number;
}
type ModelRateLimits = Record;
const DEFAULT_RATE_LIMITS: ModelRateLimits = {
'gpt-4.1': { requestsPerSecond: 50, burstCapacity: 100 },
'claude-sonnet-4.5': { requestsPerSecond: 40, burstCapacity: 80 },
'gemini-2.5-flash': { requestsPerSecond: 100, burstCapacity: 200 },
'deepseek-v3.2': { requestsPerSecond: 75, burstCapacity: 150 },
};
class TokenBucketLimiter {
private buckets: Map = new Map();
private locks: Map> = new Map();
private config: ModelRateLimits;
constructor(config: ModelRateLimits = DEFAULT_RATE_LIMITS) {
this.config = config;
}
/**
* Initialise les buckets pour tous les modèles configurés
*/
initialize(): void {
for (const [model, rateConfig] of Object.entries(this.config)) {
this.buckets.set(model, {
capacity: rateConfig.burstCapacity,
refillRate: rateConfig.requestsPerSecond,
lastRefill: Date.now(),
tokens: rateConfig.burstCapacity
});
}
}
/**
* Enregistre un nouveau modèle avec ses limites
*/
registerModel(
modelId: string,
requestsPerSecond: number,
burstCapacity?: number
): void {
const capacity = burstCapacity ?? requestsPerSecond * 2;
this.buckets.set(modelId, {
capacity,
refillRate: requestsPerSecond,
lastRefill: Date.now(),
tokens: capacity
});
}
/**
* Refill les tokens basés sur le temps écoulé
*/
private refillBucket(modelId: string): void {
const bucket = this.buckets.get(modelId);
if (!bucket) return;
const now = Date.now();
const elapsedSeconds = (now - bucket.lastRefill) / 1000;
// Ajoute les tokens générés depuis la dernière vérification
const newTokens = elapsedSeconds * bucket.refillRate;
bucket.tokens = Math.min(bucket.capacity, bucket.tokens + newTokens);
bucket.lastRefill = now;
}
/**
* Tente d'acquérir des tokens
* @returns true si l'acquisition réussit, false sinon
*/
async acquire(
modelId: string,
tokensNeeded: number = 1,
block: boolean = false
): Promise<{ success: boolean; waitMs: number }> {
// Création auto si non existant
if (!this.buckets.has(modelId)) {
const defaultConfig = this.config[modelId] ?? {
requestsPerSecond: 25,
burstCapacity: 50
};
this.registerModel(
modelId,
defaultConfig.requestsPerSecond,
defaultConfig.burstCapacity
);
}
const bucket = this.buckets.get(modelId)!;
if (block) {
// Mode bloquant - attend jusqu'à ce que les tokens soient disponibles
let attempts = 0;
const maxAttempts = 100; // Protection contre boucle infinie
while (bucket.tokens < tokensNeeded && attempts < maxAttempts) {
await this.sleep(50); // Attend 50ms
this.refillBucket(modelId);
attempts++;
}
if (bucket.tokens >= tokensNeeded) {
bucket.tokens -= tokensNeeded;
return { success: true, waitMs: attempts * 50 };
}
return { success: false, waitMs: -1 }; // Timeout
}
// Mode non-bloquant
this.refillBucket(modelId);
if (bucket.tokens >= tokensNeeded) {
bucket.tokens -= tokensNeeded;
return { success: true, waitMs: 0 };
}
// Calcule le temps d'attente nécessaire
const tokensDeficit = tokensNeeded - bucket.tokens;
const waitMs = (tokensDeficit / bucket.refillRate) * 1000;
return { success: false, waitMs };
}
/**
* Retourne le statut actuel d'un bucket
*/
getStatus(modelId: string): {
tokensAvailable: number;
capacity: number;
fillPercentage: number;
refillRate: number;
} | null {
const bucket = this.buckets.get(modelId);
if (!bucket) return null;
this.refillBucket(modelId);
return {
tokensAvailable: Math.round(bucket.tokens * 100) / 100,
capacity: bucket.capacity,
fillPercentage: Math.round(100 * bucket.tokens / bucket.capacity * 10) / 10,
refillRate: bucket.refillRate
};
}
/**
* Reset un bucket spécifique
*/
reset(modelId: string): void {
const bucket = this.buckets.get(modelId);
if (bucket) {
bucket.tokens = bucket.capacity;
bucket.lastRefill = Date.now();
}
}
private sleep(ms: number): Promise {
return new Promise(resolve => setTimeout(resolve, ms));
}
}
/**
* Client HolySheep avec Rate Limiting intégré
*/
interface HolySheepMessage {
role: 'system' | 'user' | 'assistant';
content: string;
}
interface HolySheepResponse {
id: string;
model: string;
choices: Array<{
message: { role: string; content: string };
finish_reason: string;
}>;
usage: {
prompt_tokens: number;
completion_tokens: number;
total_tokens: number;
};
_internal?: {
latencyMs: number;
costUsd: number;
holySheepRate: true;
};
}
class HolySheepClient {
private baseUrl = 'https://api.holysheep.ai/v1';
private apiKey: string;
private rateLimiter: TokenBucketLimiter;
private usageStats: Map = new Map();
private modelCosts: Record = {
'gpt-4.1': 8.00,
'claude-sonnet-4.5': 15.00,
'gemini-2.5-flash': 2.50,
'deepseek-v3.2': 0.42,
};
constructor(apiKey: string, rateLimiter?: TokenBucketLimiter) {
this.apiKey = apiKey;
this.rateLimiter = rateLimiter ?? new TokenBucketLimiter();
this.rateLimiter.initialize();
}
private calculateCost(model: string, inputTokens: number, outputTokens: number): number {
const costPerMtok = this.modelCosts[model] ?? 1.0;
const totalTokens = (inputTokens + outputTokens) / 1_000_000;
return totalTokens * costPerMtok;
}
async chatCompletion(
model: string,
messages: HolySheepMessage[],
options: {
temperature?: number;
maxTokens?: number;
waitIfLimited?: boolean;
} = {}
): Promise {
const { temperature = 0.7, maxTokens = 2048, waitIfLimited = true } = options;
// Rate limiting
const estimatedTokens = maxTokens;
const { success, waitMs } = await this.rateLimiter.acquire(
model,
estimatedTokens,
waitIfLimited
);
if (!success && !waitIfLimited) {
throw new Error(
Rate limit atteint pour ${model}. Réessayez dans ${waitMs}ms
);
}
// Préparation de la requête
const payload = {
model,
messages,
temperature,
max_tokens: maxTokens,
stream: false
};
const startTime = Date.now();
try {
const response = await fetch(${this.baseUrl}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
body: JSON.stringify(payload)
});
const latencyMs = Date.now() - startTime;
if (!response.ok) {
const error = await response.json().catch(() => ({}));
throw new Error(API Error ${response.status}: ${error.message || 'Unknown'});
}
const result: HolySheepResponse = await response.json();
// Mise à jour des statistiques
const usage = result.usage;
const cost = this.calculateCost(
model,
usage.prompt_tokens,
usage.completion_tokens
);
const stats = this.usageStats.get(model) ?? {
requests: 0,
inputTokens: 0,
outputTokens: 0,
totalCostUsd: 0
};
stats.requests++;
stats.inputTokens += usage.prompt_tokens;
stats.outputTokens += usage.completion_tokens;
stats.totalCostUsd += cost;
this.usageStats.set(model, stats);
result._internal = {
latencyMs,
costUsd: cost,
holySheepRate: true
};
return result;
} catch (error) {
console.error(Erreur lors de l'appel API:, error);
throw error;
}
}
getUsageSummary(): Record {
const summary: Record = {};
for (const [model, stats] of this.usageStats.entries()) {
const costUsd = stats.totalCostUsd;
const costYuan = costUsd * 0.15; // Taux approximatif pour HolySheep
summary[model] = {
requests: stats.requests,
totalTokens: stats.inputTokens + stats.outputTokens,
costUsd: Math.round(costUsd * 10000) / 10000,
costYuanApprox: Math.round(costYuan * 100) / 100,
savingsVsDirect: Math.round((costUsd - costYuan) * 100) / 100
};
}
return summary;
}
}
// === Exemple d'utilisation ===
async function main() {
const limiter = new TokenBucketLimiter();
limiter.initialize();
const client = new HolySheepClient('YOUR_HOLYSHEEP_API_KEY', limiter);
const messages: HolySheepMessage[] = [
{ role: 'system', content: 'Tu es un assistant technique.' },
{ role: 'user', content: 'Quelle est la différence entre token bucket et leaky bucket?' }
];
try {
const response = await client.chatCompletion('deepseek-v3.2', messages, {
maxTokens: 500
});
console.log('Réponse:', response.choices[0].message.content);
console.log('Latence:', response._internal?.latencyMs, 'ms');
console.log('Coût USD:', response._internal?.costUsd);
console.log('\n=== Résumé des coûts ===');
const summary = client.getUsageSummary();
for (const [model, stats] of Object.entries(summary)) {
console.log(${model}: ${stats.costYuanApprox}¥ (~${stats.costUsd}$));
}
} catch (error) {
console.error('Erreur:', error);
}
}
// Export pour