LiteLLM Embeddings : base de connaissances DevOps

Suite du projet DevOps Buddy : notre assistant analyse des fichiers et répond aux questions. Mais il ne connaît que ses connaissances de base. Dans ce guide, on va lui ajouter une base de connaissances DevOps pour qu’il réponde avec du contexte pertinent — c’est le principe du RAG (Retrieval-Augmented Generation).

Ce que vous allez apprendre

Comprendre les embeddings — comment transformer du texte en vecteur numérique
Calculer la similarité cosinus — mesurer la proximité sémantique entre deux textes
Construire un index de documents DevOps (runbooks, best practices)
Implémenter le RAG — Retrieval-Augmented Generation pour enrichir les réponses
Créer une interface interactive qui cite ses sources

Ce qu’on va construire

Un système RAG qui indexe vos runbooks et documentations DevOps, puis les utilise pour enrichir les réponses :

👤 "Comment déployer sur prod avec ArgoCD ?"

🔍 Recherche dans la base de connaissances...
   → runbook-argocd-deploy.md (score: 0.89)
   → best-practices-gitops.md (score: 0.76)

🤖 DevOps Buddy (avec contexte):
"D'après notre documentation interne, le déploiement production
suit ces étapes :
1. Merger dans la branche `main`
2. ArgoCD détecte automatiquement le changement
3. Synchroniser via la commande `argocd app sync prod`
..."

Qu’est-ce qu’un embedding ?

Un embedding transforme du texte en vecteur (liste de nombres). Des textes similaires auront des vecteurs proches :

"déployer avec ArgoCD"      →  [0.12, -0.34, 0.56, ...]
"ArgoCD deployment prod"    →  [0.11, -0.33, 0.55, ...]  ← proches !
"cuisiner des pâtes"        →  [0.87, 0.23, -0.45, ...]  ← différent

Étape 1 : générer un embedding

from litellm import embedding
from dotenv import load_dotenv

load_dotenv()

response = embedding(
    model="openai/text-embedding-3-small",
    input=["Comment configurer un pipeline GitLab CI ?"]
)

vector = response.data[0]["embedding"]
print(f"📐 Dimensions: {len(vector)}")  # 1536
print(f"📊 Échantillon: {vector[:5]}")

Étape 2 : calculer la similarité

La similarité cosinus mesure la proximité entre deux vecteurs (1 = identiques, 0 = sans rapport) :

from litellm import embedding
from dotenv import load_dotenv
import numpy as np

load_dotenv()


def cosine_similarity(v1: list, v2: list) -> float:
    """Similarité cosinus entre deux vecteurs."""
    a = np.array(v1)
    b = np.array(v2)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))


# Documents DevOps à comparer
docs = [
    "ArgoCD déploie automatiquement depuis Git",
    "GitOps synchronise le cluster avec le repo",
    "Docker build crée une image de conteneur"
]

response = embedding(
    model="openai/text-embedding-3-small",
    input=docs
)

vectors = [item["embedding"] for item in response.data]

# Chercher le doc le plus proche de la question
question = "comment automatiser les déploiements k8s"
q_response = embedding(
    model="openai/text-embedding-3-small",
    input=[question]
)
q_vector = q_response.data[0]["embedding"]

print(f"🔍 Question: {question}\n")
for doc, vec in zip(docs, vectors):
    score = cosine_similarity(q_vector, vec)
    print(f"  [{score:.3f}] {doc}")

Résultat :

🔍 Question: comment automatiser les déploiements k8s

  [0.847] ArgoCD déploie automatiquement depuis Git
  [0.812] GitOps synchronise le cluster avec le repo
  [0.523] Docker build crée une image de conteneur

Étape 3 : indexer une base documentaire

Créons un index de runbooks DevOps :

from litellm import embedding
from pathlib import Path
import json
import numpy as np
from dotenv import load_dotenv

load_dotenv()


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


class KnowledgeBase:
    """Base de connaissances DevOps avec recherche sémantique."""

    def __init__(self, model: str = "openai/text-embedding-3-small"):
        self.model = model
        self.documents: list[dict] = []
        self.vectors: list[list[float]] = []

    def add_document(self, title: str, content: str, tags: list[str] = None):
        """Ajoute un document à la base."""
        # Créer le texte à indexer
        text = f"{title}\n\n{content}"

        # Générer l'embedding
        response = embedding(model=self.model, input=[text])
        vector = response.data[0]["embedding"]

        self.documents.append({
            "title": title,
            "content": content,
            "tags": tags or []
        })
        self.vectors.append(vector)

    def search(self, query: str, top_k: int = 3) -> list[dict]:
        """Recherche les documents les plus pertinents."""
        # Embedding de la requête
        response = embedding(model=self.model, input=[query])
        query_vector = response.data[0]["embedding"]

        # Calculer les scores
        scored = []
        for doc, vec in zip(self.documents, self.vectors):
            score = cosine_similarity(query_vector, vec)
            scored.append({"score": score, **doc})

        # Trier par pertinence
        return sorted(scored, key=lambda x: x["score"], reverse=True)[:top_k]

    def save(self, path: str):
        """Sauvegarde l'index sur disque."""
        data = {
            "documents": self.documents,
            "vectors": self.vectors
        }
        Path(path).write_text(json.dumps(data))

    def load(self, path: str):
        """Charge l'index depuis le disque."""
        data = json.loads(Path(path).read_text())
        self.documents = data["documents"]
        self.vectors = data["vectors"]


# Créer et remplir la base
kb = KnowledgeBase()

kb.add_document(
    title="Runbook: Déploiement ArgoCD",
    content="""Pour déployer une application avec ArgoCD:
1. Pusher les manifests dans le repo Git (branche main)
2. ArgoCD détecte automatiquement le changement
3. Vérifier le statut: argocd app get <app-name>
4. Forcer la sync si nécessaire: argocd app sync <app-name>
5. Rollback si problème: argocd app rollback <app-name>""",
    tags=["argocd", "deploy", "gitops"]
)

kb.add_document(
    title="Runbook: Debug pods Kubernetes",
    content="""Diagnostic d'un pod en erreur:
1. Vérifier le status: kubectl get pods -n <namespace>
2. Voir les événements: kubectl describe pod <pod-name>
3. Logs du conteneur: kubectl logs <pod-name> -f
4. Shell dans le pod: kubectl exec -it <pod-name> -- /bin/sh
5. Métriques: kubectl top pod <pod-name>""",
    tags=["kubernetes", "debug", "pods"]
)

kb.add_document(
    title="Runbook: Pipeline GitLab CI bloqué",
    content="""Si un pipeline GitLab CI est bloqué:
1. Vérifier les runners disponibles: Settings > CI/CD > Runners
2. Consulter les logs du job: cliquer sur le job en échec
3. Variables manquantes: Settings > CI/CD > Variables
4. Relancer le job: bouton "Retry"
5. Cache corrompue: vider via Settings > CI/CD > Clear Runner Caches""",
    tags=["gitlab", "ci", "pipeline"]
)

kb.add_document(
    title="Bonnes pratiques Dockerfile",
    content="""Optimisations Dockerfile:
1. Utiliser des images légères (alpine, distroless)
2. Multi-stage builds pour réduire la taille finale
3. Un seul RUN pour apt-get update && install && clean
4. USER non-root pour la sécurité
5. HEALTHCHECK pour la supervision
6. Copier les dépendances AVANT le code (cache)""",
    tags=["docker", "dockerfile", "best-practices"]
)

# Sauvegarder
kb.save("devops_knowledge.json")
print(f"✅ {len(kb.documents)} documents indexés")

# Tester la recherche
results = kb.search("mon pod kubernetes ne démarre pas")
print("\n🔍 Recherche: 'mon pod kubernetes ne démarre pas'\n")
for r in results:
    print(f"  [{r['score']:.3f}] {r['title']}")

Étape 4 : RAG — répondre avec contexte

On combine la recherche avec la génération :

from litellm import embedding, completion
from pathlib import Path
import json
import numpy as np
from dotenv import load_dotenv

load_dotenv()


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


class DevOpsBuddy:
    """Assistant DevOps avec base de connaissances RAG."""

    SYSTEM_PROMPT = """Tu es DevOps Buddy, un assistant DevOps expert.

Tu as accès à une base de connaissances interne. Utilise UNIQUEMENT
les informations fournies dans le contexte pour répondre. Si le contexte
ne contient pas l'information nécessaire, dis-le clairement.

Réponds de manière concise et actionnable avec des commandes concrètes."""

    def __init__(self, knowledge_file: str = "devops_knowledge.json"):
        self.model = "openai/text-embedding-3-small"
        self.llm = "gpt-4.1-mini"

        # Charger la base de connaissances
        data = json.loads(Path(knowledge_file).read_text())
        self.documents = data["documents"]
        self.vectors = data["vectors"]

    def search(self, query: str, top_k: int = 2) -> list[dict]:
        """Recherche dans la base de connaissances."""
        response = embedding(model=self.model, input=[query])
        query_vector = response.data[0]["embedding"]

        scored = []
        for doc, vec in zip(self.documents, self.vectors):
            score = cosine_similarity(query_vector, vec)
            if score > 0.5:  # Seuil de pertinence
                scored.append({"score": score, **doc})

        return sorted(scored, key=lambda x: x["score"], reverse=True)[:top_k]

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

        # Rechercher le contexte pertinent
        context_docs = self.search(question)

        # Construire le contexte
        if context_docs:
            context = "\n\n---\n\n".join([
                f"**{d['title']}**\n{d['content']}"
                for d in context_docs
            ])
            sources = [d['title'] for d in context_docs]
        else:
            context = "(Aucun document pertinent trouvé)"
            sources = []

        # Appeler le LLM avec le contexte
        response = completion(
            model=self.llm,
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": f"""CONTEXTE (documentation interne):
{context}

QUESTION: {question}"""}
            ]
        )

        answer = response.choices[0].message.content

        # Ajouter les sources
        if sources:
            answer += f"\n\n📚 Sources: {', '.join(sources)}"

        return answer


# Test
buddy = DevOpsBuddy()

questions = [
    "Mon pod Kubernetes est en CrashLoopBackOff, que faire ?",
    "Comment déployer avec ArgoCD ?",
    "Le pipeline GitLab est bloqué depuis 30 minutes"
]

for q in questions:
    print(f"👤 {q}")
    print(f"🤖 {buddy.ask(q)}")
    print("\n" + "="*60 + "\n")

Exemple de sortie :

👤 Mon pod Kubernetes est en CrashLoopBackOff, que faire ?

🤖 Pour diagnostiquer un pod en CrashLoopBackOff :

1. Voir les événements du pod :
   ```bash
   kubectl describe pod <pod-name> -n <namespace>

Consulter les logs du dernier crash :
Fenêtre de terminal
```
kubectl logs <pod-name> --previous
```
Si le pod démarre brièvement, ouvrir un shell :
Fenêtre de terminal
```
kubectl exec -it <pod-name> -- /bin/sh
```

📚 Sources: Runbook: Debug pods Kubernetes

## Étape 5 : interface interactive

```python title="15_buddy_interactive.py"
from litellm import embedding, completion
from pathlib import Path
import json
import numpy as np
from dotenv import load_dotenv

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 DevOpsBuddy:
    """Assistant DevOps avec RAG."""

    SYSTEM = """Tu es DevOps Buddy. Réponds en utilisant le contexte fourni.
Si le contexte ne suffit pas, dis-le. Sois concis et donne des commandes."""

    def __init__(self, kb_file: str = "devops_knowledge.json"):
        data = json.loads(Path(kb_file).read_text())
        self.docs = data["documents"]
        self.vecs = data["vectors"]

    def search(self, query: str, k: int = 2):
        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)]
        return sorted(scored, key=lambda x: x[1], reverse=True)[:k]

    def ask(self, question: str) -> str:
        results = self.search(question)
        context = "\n\n".join([f"**{d['title']}**\n{d['content']}" for d, _ in results])

        resp = completion(
            model="gpt-4.1-mini",
            messages=[
                {"role": "system", "content": self.SYSTEM},
                {"role": "user", "content": f"CONTEXTE:\n{context}\n\nQUESTION: {question}"}
            ]
        )
        return resp.choices[0].message.content


def main():
    buddy = DevOpsBuddy()

    print("🤖 DevOps Buddy (tapez 'quit' pour quitter)\n")

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

        if question.lower() in ("quit", "exit", "q"):
            print("👋 À bientôt !")
            break

        if not question:
            continue

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


if __name__ == "__main__":
    main()

Structure du projet

lab-devops-buddy/
├── .env
├── devops_knowledge.json       # Index sauvegardé
├── 01-10_*.py                  # Guides précédents
├── 11_embedding_basic.py       # Premier embedding
├── 12_similarity.py            # Similarité cosinus
├── 13_knowledge_base.py        # Indexation
├── 14_rag_devops.py            # RAG complet
└── 15_buddy_interactive.py     # Interface CLI

Bonnes pratiques

Chunking des longs documents

Pour les docs de plus de 500 tokens, découpez-les :

def chunk_document(text: str, max_chars: int = 1500) -> list[str]:
    """Découpe un document en morceaux."""
    paragraphs = text.split("\n\n")
    chunks = []
    current = ""

    for p in paragraphs:
        if len(current) + len(p) < max_chars:
            current += f"\n\n{p}"
        else:
            if current:
                chunks.append(current.strip())
            current = p

    if current:
        chunks.append(current.strip())

    return chunks

Seuil de pertinence

Filtrez les résultats avec un score trop bas :

def search(self, query: str, min_score: float = 0.6):
    # ...
    return [d for d, score in scored if score >= min_score]

À retenir

embedding() : transforme du texte en vecteur
Similarité cosinus : mesure la proximité sémantique (0 à 1)
RAG : Retrieval + Generation = réponses contextualisées
Chunking : découpez les longs documents pour de meilleurs résultats
Seuil : filtrez les résultats peu pertinents (< 0.5)

Prochaines étapes

DevOps Buddy a maintenant une base de connaissances. Dans le dernier guide, on va le préparer pour la production : fallbacks multi-modèles, caching, et monitoring.

Guide suivant : Production avec Router Fallbacks, load balancing, caching et monitoring

Guide précédent : Analyse parallèle Utiliser acompletion() pour auditer plusieurs fichiers

Revenir à l'introduction Les bases de LiteLLM et DevOps Buddy