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

À 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 →