Maîtriser le SDK Python Yevi : Pipeline RAG & Custom Providers

Le SDK Python de Yevi Core v1.0 (Licence MIT) est conçu selon des principes de découplage strict. Le pipeline RAG ne dépend d’aucun fournisseur d’inférence propriétaire : tout backend LLM s’intègre via l’interface abstraite LLMProvider.

Ce module apprend aux développeurs à utiliser le SDK Python et à écrire leur propre fournisseur LLM personnalisé.

---

1. Exemple d’Ingestion & Requête RAG avec le SDK Python

from pathlib import Path
from core.index.store import VectorStore
from core.ingest.loader import load_and_chunk_corpus
from core.privacy.anonymize import Anonymizer
from core.rag.pipeline import RagPipeline
from ops.serve.llm_provider import DummyProvider

# 1. Chargement et découpage du corpus
chunks = load_and_chunk_corpus(Path("data/corpus_demo"))

# 2. Stockage dans ChromaDB + BM25
store = VectorStore()
store.add_chunks([
    {
        "chunk_id": c.chunk_id,
        "doc_id": c.doc_id,
        "chunk_index": c.chunk_index,
        "text": c.text,
        "source_path": c.source_path,
    } for c in chunks
])

# 3. Anonymisation PII de la question
anonymizer = Anonymizer()
anon_res = anonymizer.anonymize("Contrat signé avec Me votrenom")

# 4. Inférence RAG avec fournisseur Dummy / Ollama
pipeline = RagPipeline(store=store, llm=DummyProvider(), top_k=4)
rag_ans = pipeline.answer(anon_res.anonymized_text, domain="legal")

# 5. Restitution du texte avec démasquage PII
final_text = anonymizer.restore(rag_ans.answer, anon_res.mapping)
print("Réponse finale démasquée :", final_text)

---

2. Écrire un LLMProvider Personnalisé (ex. vLLM / TextGen API)

Pour connecter un serveur d’inférence spécifique (ex. vLLM ou serveur HTTP interne), dérivez la classe LLMProvider :

import requests
from ops.serve.llm_provider import LLMProvider

class CustomVLLMProvider(LLMProvider):
    def __init__(self, endpoint_url: str = "http://localhost:8000/v1/completions"):
        self.endpoint_url = endpoint_url

    def generate(self, prompt: str, *, max_tokens: int = 512, temperature: float = 0.1) -> str:
        payload = {
            "prompt": prompt,
            "max_tokens": max_tokens,
            "temperature": temperature,
        }
        response = requests.post(self.endpoint_url, json=payload, timeout=30)
        response.raise_for_status()
        return response.json()["choices"][0]["text"]

    def is_available(self) -> bool:
        try:
            res = requests.get(self.endpoint_url.replace("/completions", "/models"), timeout=3)
            return res.status_code == 200
        except Exception:
            return False

---

🛠️ Passer à l’Action

• 📄 Consulter la Référence de la CLI Yevi
• 📄 Voir la Spécification de l’API REST FastAPI
• 💻 Accéder au Dépôt GitHub Yevi Core v1.0 (Licence MIT)