LiteLLM Router : DevOps Buddy en production

Fin du projet DevOps Buddy : notre assistant fonctionne en dev. Maintenant, on va le préparer pour la production : fallbacks multi-providers, caching pour réduire les coûts de 50%, et monitoring pour savoir quand ça va mal.

Ce que vous allez apprendre

• Configurer un Router avec plusieurs providers en fallback (OpenAI → Claude → Ollama)
• Définir les priorités avec l'option order pour choisir quel provider utiliser en premier
• Activer le cache en mémoire ou Redis pour économiser jusqu'à 50% des coûts
• Monitorer les appels avec des callbacks custom (latence, erreurs, fallbacks)
• Séparer la configuration du code avec un fichier YAML

Pourquoi le Router en production ?

En production, les problèmes arrivent :

Problème	Impact	Solution Router
OpenAI down	Service indisponible	Bascule sur Claude
Rate limiting	Requêtes rejetées	Répartition de charge
Latence élevée	UX dégradée	Timeout + fallback
Coûts élevés	Budget explosé	Cache des réponses

Glissez pour voir

Étape 1 : configuration multi-providers

Configurons DevOps Buddy avec 3 providers en fallback :

16_router_basic.py

from litellm import Router
from dotenv import load_dotenv
import os

load_dotenv()

# Configuration des providers
model_list = [
    {
        "model_name": "devops-buddy",  # Alias unique pour l'app
        "litellm_params": {
            "model": "openai/gpt-4.1-mini",
            "api_key": os.getenv("OPENAI_API_KEY"),
            "order": 1  # Provider principal
        }
    },
    {
        "model_name": "devops-buddy",
        "litellm_params": {
            "model": "anthropic/claude-3-5-haiku-latest",
            "api_key": os.getenv("ANTHROPIC_API_KEY"),
            "order": 2  # Premier fallback
        }
    },
    {
        "model_name": "devops-buddy",
        "litellm_params": {
            "model": "gemini/gemini-2.0-flash",
            "api_key": os.getenv("GOOGLE_API_KEY"),
            "order": 3  # Dernier recours
        }
    }
]

# Router avec retries et cooldowns
router = Router(
    model_list=model_list,
    num_retries=2,           # Réessaie 2x avant fallback
    retry_after=1,           # 1s entre retries
    timeout=30,              # Max 30s par requête
    allowed_fails=3,         # Échecs avant cooldown
    cooldown_time=60,        # Désactive 60s si trop d'échecs
    enable_pre_call_checks=True
)

# Test
response = router.completion(
    model="devops-buddy",
    messages=[
        {"role": "system", "content": "Tu es DevOps Buddy, assistant DevOps expert."},
        {"role": "user", "content": "Comment vérifier les logs d'un pod K8s ?"}
    ]
)

print(f"✅ Provider utilisé: {response.model}")
print(f"📝 Réponse: {response.choices[0].message.content[:200]}...")

Alias vs modèle

L'alias devops-buddy abstrait les providers. Vous changez de modèle sans modifier le code appelant.

Étape 2 : caching pour réduire les coûts

Les mêmes questions reviennent souvent. Activez le cache :

17_router_cached.py

from litellm import Router
from dotenv import load_dotenv
import os
import time

load_dotenv()

model_list = [
    {
        "model_name": "devops-buddy",
        "litellm_params": {
            "model": "openai/gpt-4.1-mini",
            "api_key": os.getenv("OPENAI_API_KEY")
        }
    }
]

# Cache en mémoire (dev/test)
router = Router(
    model_list=model_list,
    cache_responses=True
)

# Question fréquente
question = "Comment redémarrer un pod Kubernetes ?"
messages = [{"role": "user", "content": question}]

# Premier appel — va vers l'API
start = time.time()
response1 = router.completion(model="devops-buddy", messages=messages)
time1 = time.time() - start
print(f"1er appel: {time1:.2f}s (API)")

# Deuxième appel — du cache
start = time.time()
response2 = router.completion(model="devops-buddy", messages=messages)
time2 = time.time() - start
print(f"2e appel: {time2:.4f}s (cache) — {time1/max(time2, 0.001):.0f}x plus rapide")

Cache Redis pour la production

En multi-instances, utilisez Redis :

18_router_redis.py

from litellm import Router
from dotenv import load_dotenv
import os

load_dotenv()

model_list = [
    {
        "model_name": "devops-buddy",
        "litellm_params": {
            "model": "openai/gpt-4.1-mini",
            "api_key": os.getenv("OPENAI_API_KEY")
        }
    }
]

router = Router(
    model_list=model_list,
    redis_host=os.getenv("REDIS_HOST", "localhost"),
    redis_port=int(os.getenv("REDIS_PORT", 6379)),
    redis_password=os.getenv("REDIS_PASSWORD", None),
    cache_responses=True
)

Économies réelles

Avec un taux de cache hit de 30%, vous économisez ~30% sur vos coûts d'API. Sur 10 000 requêtes/jour à $0.01 : $30/jour économisés.

Étape 3 : monitoring et alertes

Ajoutez des callbacks pour surveiller les performances :

19_router_monitoring.py

from litellm import Router
from litellm.integrations.custom_logger import CustomLogger
from dotenv import load_dotenv
import os
from datetime import datetime

load_dotenv()


class DevOpsLogger(CustomLogger):
    """Logger custom pour DevOps Buddy."""

    def __init__(self):
        self.stats = {
            "success": 0,
            "failures": 0,
            "total_latency": 0,
            "fallbacks": 0
        }

    def log_success(self, kwargs, response, start_time, end_time):
        model = kwargs.get("model", "unknown")
        latency = end_time - start_time
        tokens = response.usage.total_tokens if response.usage else 0

        self.stats["success"] += 1
        self.stats["total_latency"] += latency

        print(f"✅ [{datetime.now().strftime('%H:%M:%S')}] "
              f"{model} | {latency:.2f}s | {tokens} tokens")

    def log_failure(self, kwargs, exception, start_time, end_time):
        model = kwargs.get("model", "unknown")

        self.stats["failures"] += 1

        print(f"❌ [{datetime.now().strftime('%H:%M:%S')}] "
              f"{model} | {type(exception).__name__}: {exception}")

    def get_metrics(self) -> dict:
        """Retourne les métriques agrégées."""
        total = self.stats["success"] + self.stats["failures"]
        return {
            "total_requests": total,
            "success_rate": self.stats["success"] / max(total, 1) * 100,
            "avg_latency": self.stats["total_latency"] / max(self.stats["success"], 1),
            "fallback_rate": self.stats["fallbacks"] / max(total, 1) * 100
        }


# Utilisation
logger = DevOpsLogger()

model_list = [
    {
        "model_name": "devops-buddy",
        "litellm_params": {
            "model": "openai/gpt-4.1-mini",
            "api_key": os.getenv("OPENAI_API_KEY")
        }
    }
]

router = Router(
    model_list=model_list,
    callbacks=[logger]
)

# Test
for i in range(5):
    router.completion(
        model="devops-buddy",
        messages=[{"role": "user", "content": f"Question {i + 1}"}]
    )

print("\n📊 Métriques:")
for k, v in logger.get_metrics().items():
    print(f"  {k}: {v:.2f}")

Étape 4 : DevOps Buddy complet pour la production

Intégrons tout, RAG + Router + Monitoring :

20_buddy_production.py

from litellm import Router, embedding
from litellm.integrations.custom_logger import CustomLogger
from pathlib import Path
import json
import numpy as np
from dotenv import load_dotenv
import os

load_dotenv()


def cosine_similarity(v1, v2):
    a, b = np.array(v1), np.array(v2)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))


class ProductionLogger(CustomLogger):
    def log_success(self, kwargs, response, start_time, end_time):
        latency = end_time - start_time
        print(f"📡 {response.model} | {latency:.2f}s")

    def log_failure(self, kwargs, exception, start_time, end_time):
        print(f"⚠️ Fallback déclenché: {exception}")


class DevOpsBuddyProduction:
    """DevOps Buddy prêt pour la production."""

    SYSTEM_PROMPT = """Tu es DevOps Buddy, l'assistant DevOps de l'équipe.

Utilise le contexte fourni pour répondre. Si le contexte ne suffit pas,
utilise tes connaissances générales mais précise-le.

Sois concis, donne des commandes concrètes."""

    def __init__(self, knowledge_file: str = "devops_knowledge.json"):
        # Router multi-providers
        model_list = [
            {
                "model_name": "chat",
                "litellm_params": {
                    "model": "openai/gpt-4.1-mini",
                    "api_key": os.getenv("OPENAI_API_KEY"),
                    "order": 1
                }
            },
            {
                "model_name": "chat",
                "litellm_params": {
                    "model": "anthropic/claude-3-5-haiku-latest",
                    "api_key": os.getenv("ANTHROPIC_API_KEY"),
                    "order": 2
                }
            }
        ]

        self.router = Router(
            model_list=model_list,
            num_retries=2,
            timeout=30,
            cache_responses=True,
            callbacks=[ProductionLogger()]
        )

        # Base de connaissances
        if Path(knowledge_file).exists():
            data = json.loads(Path(knowledge_file).read_text())
            self.docs = data["documents"]
            self.vecs = data["vectors"]
        else:
            self.docs = []
            self.vecs = []

    def search_context(self, query: str, k: int = 2) -> str:
        """Recherche du contexte pertinent."""
        if not self.docs:
            return ""

        resp = embedding(model="openai/text-embedding-3-small", input=[query])
        qv = resp.data[0]["embedding"]

        scored = [(d, cosine_similarity(qv, v)) for d, v in zip(self.docs, self.vecs)]
        top = sorted(scored, key=lambda x: x[1], reverse=True)[:k]

        return "\n\n".join([
            f"**{d['title']}**\n{d['content']}"
            for d, score in top if score > 0.5
        ])

    def ask(self, question: str) -> str:
        """Pose une question avec RAG et fallbacks."""

        context = self.search_context(question)

        messages = [
            {"role": "system", "content": self.SYSTEM_PROMPT},
            {"role": "user", "content": f"""CONTEXTE:
{context or "(Pas de documentation interne pertinente)"}

QUESTION: {question}"""}
        ]

        response = self.router.completion(
            model="chat",
            messages=messages
        )

        return response.choices[0].message.content


def main():
    buddy = DevOpsBuddyProduction()

    print("🚀 DevOps Buddy (Production Mode)")
    print("   Fallbacks: OpenAI → Claude")
    print("   Cache: activé")
    print("   Tapez 'quit' pour quitter\n")

    while True:
        question = input("👤 > ").strip()

        if question.lower() in ("quit", "exit", "q"):
            break

        if not question:
            continue

        answer = buddy.ask(question)
        print(f"🤖 {answer}\n")


if __name__ == "__main__":
    main()

Configuration YAML (recommandé)

Séparez la config du code :

litellm_config.yaml

model_list:
  - model_name: devops-buddy
    litellm_params:
      model: openai/gpt-4.1-mini
      api_key: os.environ/OPENAI_API_KEY
      order: 1

  - model_name: devops-buddy
    litellm_params:
      model: anthropic/claude-3-5-haiku-latest
      api_key: os.environ/ANTHROPIC_API_KEY
      order: 2

  - model_name: embeddings
    litellm_params:
      model: openai/text-embedding-3-small
      api_key: os.environ/OPENAI_API_KEY

router_settings:
  num_retries: 2
  timeout: 30
  allowed_fails: 3
  cooldown_time: 60
  cache_responses: true

Chargement :

import yaml
from litellm import Router

with open("litellm_config.yaml") as f:
    config = yaml.safe_load(f)

router = Router(
    model_list=config["model_list"],
    **config.get("router_settings", {})
)

Structure finale du projet

lab-devops-buddy/
├── .env                        # Clés API
├── litellm_config.yaml         # Config production
├── devops_knowledge.json       # Base de connaissances
│
├── 01_question_simple.py       # Guide 1: Bases
├── ...
├── 05_erreurs.py
│
├── 06_async_basic.py           # Guide 2: Async
├── ...
├── 10_error_handling.py
│
├── 11_embedding_basic.py       # Guide 3: RAG
├── ...
├── 15_buddy_interactive.py
│
├── 16_router_basic.py          # Guide 4: Production
├── 17_router_cached.py
├── 18_router_redis.py
├── 19_router_monitoring.py
└── 20_buddy_production.py      # Version finale

Checklist production

Élément	Statut	Action
Multi-providers	☐	2-3 providers en fallback
Retries	☐	num_retries=2
Timeout	☐	timeout=30
Cooldowns	☐	cooldown_time=60
Cache	☐	Redis en multi-instances
Monitoring	☐	Callbacks + alertes
Secrets	☐	Variables d'environnement
Logs	☐	Centralisés (ELK, Loki...)

Glissez pour voir

Contrôle de connaissances

Vérifiez que l'essentiel de ce guide est acquis. Les questions portent uniquement sur ce qui vient d'être expliqué ici.

Contrôle de connaissances

Validez vos connaissances avec ce quiz interactif

6 questions

6 min.

70% requis

Informations

• Le chronomètre démarre au clic sur Démarrer
• Questions à choix multiples, vrai/faux et réponses courtes
• Vous pouvez naviguer entre les questions
• Les résultats détaillés sont affichés à la fin

Démarrer le quiz

Lance le quiz et démarre le chronomètre

À retenir

• Router : abstrait les providers avec un alias unique
• order : définit la priorité des fallbacks
• cache_responses : économise jusqu'à 50% des coûts
• Redis : obligatoire pour cache multi-instances
• Callbacks : surveillez latence, erreurs, fallbacks
• Config YAML : séparez config et code

Récapitulatif du projet DevOps Buddy

En 4 guides, vous avez construit un assistant IA DevOps complet :

1. Bases : chatbot interactif avec streaming et conversation
2. Async : audit de fichiers DevOps en parallèle
3. RAG : base de connaissances avec recherche sémantique
4. Production : fallbacks, caching, monitoring

Prochaines étapes

Félicitations ! Vous avez construit un assistant IA DevOps complet, prêt pour la production.

Revenir à l'introduction Revoir les bases de LiteLLM et DevOps Buddy

Guide précédent : Embeddings/RAG Base de connaissances avec recherche sémantique

LiteLLM Proxy Server Passerelle centralisée multi-équipes avec API unifiée

×

📖 Voir la documentation →