← Documentation API Reference

Pipeline RAG AISIA

Version : 4.15.0 Société : AISIA — Structure juridique en cours de création URL publique : https://aisia.fr/

---

Table des matières

1. Architecture RAG 2. Phase d'ingestion 3. Stockage vectoriel Qdrant 4. Recherche et retrieval 5. Mémoire conversationnelle Redis 6. Construction du prompt augmenté 7. Knowledge Manager 8. Tool Calling 9. Boucle d'apprentissage 10. Collections Qdrant 11. Configuration 12. Monitoring et diagnostics

---

1. Architecture RAG

Le pipeline RAG (Retrieval Augmented Generation) d'AISIA enrichit chaque requête LLM avec du contexte pertinent, améliorant la qualité et la pertinence des réponses.

Flux complet

Requête utilisateur
       │
       ▼
┌─────────────────────┐
│  1. Retrieval       │
│  (Qdrant search)    │ ← Cherche les connaissances pertinentes
└─────────────────────┘
       │
       ▼
┌─────────────────────┐
│  2. Memory          │
│  (Redis history)    │ ← Récupère l'historique de conversation
└─────────────────────┘
       │
       ▼
┌─────────────────────┐
│  3. Augmentation    │
│  (build_augmented   │ ← Assemble system + RAG + history + user
│   _prompt)          │
└─────────────────────┘
       │
       ▼
┌─────────────────────┐
│  4. Generation      │
│  (LLM call)         │ ← Appel au modèle avec le prompt enrichi
└─────────────────────┘
       │
       ▼
┌─────────────────────┐
│  5. Tool execution  │
│  (optional)         │ ← Exécute les outils invoqués par le LLM
└─────────────────────┘
       │
       ▼
┌─────────────────────┐
│  6. Storage         │
│  (save response)    │ ← Sauvegarde pour futur RAG
└─────────────────────┘
       │
       ▼
  Réponse enrichie

---

2. Phase d'ingestion

Chunking

Le texte brut est découpé en morceaux (chunks) pour le stockage vectoriel :

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    if len(text) <= chunk_size:
        return [text]
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start = end - overlap
    return chunks

Paramètres de chunking

Paramètre	Valeur	Description
`chunk_size`	500 caractères	Taille maximale d'un chunk
`overlap`	50 caractères	Chevauchement entre chunks consécutifs

Pourquoi le chevauchement ?

Le chevauchement de 50 caractères garantit que les informations à la frontière de deux chunks ne sont pas perdues. Exemple :

Texte : "AISIA utilise un algorithme de routage bandit multi-bras pour..."

Chunk 1 : "AISIA utilise un algorithme de routage band" [overlap] Chunk 2 : "de routage bandit multi-bras pour..."

Nettoyage du texte

Avant le chunking, le texte est nettoyé :

def clean_text(text: str) -> str:
    text = re.sub(r"<[^>]+>", " ", text)   # Supprime les balises HTML
    text = re.sub(r"\s+", " ", text)        # Normalise les espaces
    return text.strip()

Encodage vectoriel

Chaque chunk est encodé en vecteur de 384 dimensions :

Attribut	Valeur
Modèle	`all-MiniLM-L6-v2`
Dimensions	384
Normalisation	L2 (embeddings normalisés)
Distance	Cosinus
Backend prioritaire	`sentence-transformers` (PyTorch)
Backend fallback	`fastembed` (ONNX, plus léger)

def encode(self, text: str) -> list[float]:
    if _BACKEND == "sentence_transformers":
        return self._encoder.encode(text, normalize_embeddings=True).tolist()
    elif _BACKEND == "fastembed":
        return next(self._encoder.embed([text])).tolist()
    else:
        return [0.0] * 384  # vecteur zéro si pas d'encodeur

Sources d'ingestion

Source	Méthode	Endpoint
Texte brut	`ingest_text(text, source)`	`POST /admin/knowledge/ingest`
URL	`ingest_url(url)`	`POST /admin/knowledge/ingest`
Réponses LLM	Automatique via learning loop	Interne
Datasets HuggingFace	Via dataset manager	`POST /admin/datasets`

---

3. Stockage vectoriel Qdrant

Architecture Qdrant

Qdrant est une base de données vectorielle spécialisée dans la recherche par similarité sémantique.

Attribut	Valeur
Host	`aisia-core_qdrant` (Swarm DNS)
Port HTTP	6333
Port gRPC	6334
Algorithme d'indexation	HNSW (Hierarchical Navigable Small World)

Structure d'un point Qdrant

Chaque élément stocké dans Qdrant est un "point" avec :

{
  "id": 123456789,
  "vector": [0.012, -0.034, 0.056, ...],
  "payload": {
    "entry_id": "uuid",
    "prompt": "Question originale",
    "best_response": "Réponse optimale",
    "source_provider": "cerebras",
    "quality_score": 0.85,
    "timestamp": "2026-04-19T10:30:00Z",
    "run_id": "uuid",
    "used_for_training": false
  }
}

Identifiant de point

L'ID du point est dérivé du hash de l'entry_id :

point_id = abs(hash(entry.entry_id)) % (2**63)

Cela permet des upserts déterministes : ré-ingérer la même entrée met à jour le point existant plutôt que d'en créer un nouveau.

---

4. Recherche et retrieval

Recherche de contexte RAG

async def retrieve_context(
    qdrant_store,
    prompt: str,
    top_k: int = 3,
    min_score: float = 0.5,
) -> list[dict]:

Paramètres de recherche

Paramètre	Valeur par défaut	Description
`top_k`	3	Nombre de résultats retournés
`min_score`	0.5	Score minimum de similarité

Processus de recherche

1. Encodage : le prompt utilisateur est encodé en vecteur 384D 2. Recherche : Qdrant effectue une recherche ANN (Approximate Nearest Neighbor) dans la collection mtp_knowledge 3. Filtrage : les résultats avec un score < min_score sont éliminés 4. Extraction : le texte pertinent est extrait du payload (champs response, text, ou content) 5. Troncature : chaque résultat est tronqué à 500 caractères

Structure du résultat

context_items = [
    {
        "text": "Contenu pertinent (max 500 chars)...",
        "source": "cerebras",  # ou "knowledge_base"
        "score": 0.823,
    },
    ...
]

Recherche dans d'autres collections

Collection	Méthode	Usage
`mtp_knowledge`	`search_knowledge()`	RAG principal
`ai_responses`	`search_responses()`	Historique des réponses
`eval_corpus`	`search_eval()`	Prompts d'évaluation
`mtp_training_pairs`	Scroll filtré	Paires d'entraînement

---

5. Mémoire conversationnelle Redis

Principe

AISIA maintient un historique de conversation en temps réel dans Redis pour chaque utilisateur et chaque session de conversation.

Paramètres

Paramètre	Valeur	Description
`CONV_TTL_S`	3600 (1h)	Durée de vie de la session
`CONV_MAX_MESSAGES`	20	Messages maximum par conversation
Préfixe Redis	`aisia:conv:`	Préfixe des clés

Structure de la clé Redis

aisia:conv:{user_id}:{conv_id}

La valeur est un JSON contenant la liste des messages :

[
  {"role": "user", "content": "Bonjour", "ts": 1713520200.0},
  {"role": "assistant", "content": "Bonjour ! Comment...", "ts": 1713520201.5},
  {"role": "user", "content": "Explique le RAG", "ts": 1713520210.0}
]

Opérations

#### Récupérer l'historique

async def get_conversation(redis_client, user_id, conv_id="default"):
    key = f"aisia:conv:{user_id}:{conv_id}"
    data = await redis_client.get(key)
    messages = json.loads(data)
    return messages[-CONV_MAX_MESSAGES:]  # Garde les 20 derniers

#### Ajouter un message

async def append_to_conversation(redis_client, user_id, role, content, conv_id="default"):
    key = f"aisia:conv:{user_id}:{conv_id}"
    # Récupère, ajoute, tronque, sauvegarde
    messages.append({
        "role": role,
        "content": content[:2000],  # Tronque à 2000 chars
        "ts": time.time(),
    })
    messages = messages[-CONV_MAX_MESSAGES:]
    await redis_client.set(key, json.dumps(messages), ex=CONV_TTL_S)

#### Effacer l'historique

async def clear_conversation(redis_client, user_id, conv_id="default"):
    await redis_client.delete(f"aisia:conv:{user_id}:{conv_id}")

Persistance long terme

En complément du cache Redis (1h), les conversations sont sauvegardées en base de données MariaDB pour un accès à long terme :

Table	Contenu
`aisia_conversations`	Métadonnées (id, user_id, title, message_count, dates)
`aisia_messages`	Messages (role, content, provider_id, latency_ms, tokens_used)

---

6. Construction du prompt augmenté

Fonction build_augmented_prompt

def build_augmented_prompt(
    user_prompt: str,
    rag_context: list[dict] | None = None,
    conversation_history: list[dict] | None = None,
    system_prompt: str = "",
) -> str:

Structure du prompt final

Le prompt est assemblé dans cet ordre :

1. [System prompt] ← Instructions globales (guardrails) 2. [Contexte RAG] ← Connaissances pertinentes de Qdrant Contexte pertinent (base de connaissances AISIA) : - Texte pertinent 1 - Texte pertinent 2 - Texte pertinent 3 3. [Historique conversation] ← 6 derniers messages Historique de conversation : Utilisateur: Message précédent 1 AISIA: Réponse précédente 1 Utilisateur: Message précédent 2 AISIA: Réponse précédente 2

4. [Question utilisateur] ← Prompt actuel Question : Votre question ici

Troncature de l'historique

Seuls les 6 derniers messages de l'historique sont injectés dans le prompt, même si Redis en contient 20. Cela limite la taille du contexte tout en préservant la continuité conversationnelle.

if conversation_history:
    history_text = "\n".join(
        f"{'Utilisateur' if m['role'] == 'user' else 'AISIA'}: {m['content']}"
        for m in conversation_history[-6:]  # 6 derniers messages
    )

Injection du system prompt (guardrails)

Si un system prompt global est configuré dans les guardrails (system_prompt_global), il est injecté en tête du prompt :

def inject_system_prompt(prompt, rules):
    sp = rules.get("system_prompt_global", "").strip()
    if not sp:
        return prompt
    return f"[Instructions système : {sp}]\n\n{prompt}"

---

7. Knowledge Manager

Description

Le KnowledgeManager gère l'ingestion et la consultation de la base de connaissances Qdrant via l'API d'administration.

Endpoints

Méthode	Endpoint	Description
`GET`	`/admin/knowledge`	Statistiques de la base
`POST`	`/admin/knowledge/ingest`	Ingérer un document
`POST`	`/admin/knowledge/clear`	Vider la base

Statistiques

GET /admin/knowledge
Authorization: Bearer TOKEN

{
  "status": "available",
  "collections": [
    {"name": "mtp_knowledge", "points_count": 1500, "vectors_count": 1500},
    {"name": "ai_responses", "points_count": 5000, "vectors_count": 5000},
    {"name": "mtp_training_pairs", "points_count": 800, "vectors_count": 800},
    {"name": "eval_corpus", "points_count": 200, "vectors_count": 200},
    {"name": "router_embeddings", "points_count": 300, "vectors_count": 300}
  ]
}

Ingestion de texte

POST /admin/knowledge/ingest Authorization: Bearer TOKEN Content-Type: application/json

{ "text": "AISIA est une plateforme d'orchestration IA qui route les requêtes...", "source": "documentation_interne" }

Réponse :

{
  "status": "ingested",
  "chunks": 5,
  "stored": 5,
  "source": "documentation_interne",
  "elapsed_ms": 1200.3
}

Ingestion depuis une URL

POST /admin/knowledge/ingest
{
  "url": "https://aisia.fr/about"
}

Le contenu HTML est téléchargé, nettoyé (suppression des balises), découpé en chunks et indexé dans Qdrant.

Vider la base

POST /admin/knowledge/clear
Authorization: Bearer TOKEN

Supprime la collection mtp_knowledge de Qdrant. Elle sera recréée automatiquement au prochain accès.

---

8. Tool Calling

Principe

AISIA permet au LLM d'invoquer des outils pendant la génération de sa réponse. Le LLM écrit des patterns [TOOL:nom(paramètre)] dans sa réponse, et AISIA les détecte et les exécute.

Les 5 outils disponibles

Outil	Syntaxe	Description
`search`	`[TOOL:search(requête)]`	Recherche web DuckDuckGo
`calculate`	`[TOOL:calculate(expression)]`	Calcul mathématique sécurisé
`knowledge`	`[TOOL:knowledge(question)]`	Recherche knowledge base Qdrant
`datetime`	`[TOOL:datetime()]`	Date et heure actuelles UTC
`providers`	`[TOOL:providers()]`	Liste des providers actifs

Outil search

Effectue une recherche web via DuckDuckGo HTML (pas de clé API requise) :

async def tool_search(query: str) -> str:
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(
            "https://html.duckduckgo.com/html/",
            params={"q": query},
            headers={"User-Agent": "AISIA/4.12.0"},
        )
    # Extrait les 3 premiers snippets
    snippets = re.findall(r'class="result__snippet">(.*?)</a>', resp.text)
    return "\n".join(cleaned_snippets[:3])

Outil calculate

Calcul mathématique sécurisé avec des fonctions autorisées :

safe_funcs = {
    "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
    "tan": math.tan, "log": math.log, "log10": math.log10,
    "abs": abs, "round": round, "pi": math.pi, "e": math.e,
    "pow": pow, "max": max, "min": min,
}
Évaluation sécurisée (pas de __builtins__)
result = eval(expr, {"__builtins__": {}}, safe_funcs)

Caractères autorisés : 0-9 + - * / ( ) . ^ , et noms de fonctions. L'opérateur ^ est converti en ** (puissance Python).

Outil knowledge

Interroge la base de connaissances Qdrant via le pipeline RAG :

async def tool_knowledge(query: str, qdrant_store=None) -> str:
    results = await retrieve_context(qdrant_store, query, top_k=3)
    return "\n".join(f"- {r['text']}" for r in results)

Outil datetime

def tool_datetime() -> str:
    now = datetime.now(UTC)
    return now.strftime("%A %d %B %Y, %H:%M:%S UTC")

Outil providers

async def tool_providers() -> str:
    providers = get_active_providers()
    lines = [f"- {p['display_name']} ({len(p.get('models', []))} modèles)" for p in active[:10]]
    return f"{len(active)} providers actifs:\n" + "\n".join(lines)

Détection et exécution

TOOL_PATTERN = re.compile(r'\[TOOL:(\w+)\(([^)]*)\)\]')


async def execute_tools(text: str, qdrant_store=None) -> str:
    matches = TOOL_PATTERN.findall(text)
    for tool_name, raw_param in matches:
        # Exécute l'outil correspondant
        result = await tool_fn(param)
        # Remplace l'appel par le résultat
        text = text.replace(f"[TOOL:{tool_name}({raw_param})]", result, 1)
    return text

Prompt d'injection des outils

Le prompt suivant est injecté pour informer le LLM des outils disponibles :

Tu disposes des outils suivants. Pour les utiliser, écris
[TOOL:nom_outil(paramètre)] dans ta réponse.


Outils disponibles :
[TOOL:search(requête)] -- recherche web
[TOOL:calculate(expression)] -- calcul mathématique
[TOOL:knowledge(question)] -- cherche dans la base de connaissances
[TOOL:datetime()] -- date et heure actuelles
[TOOL:providers()] -- liste des providers IA disponibles

Tu peux utiliser ces outils pour enrichir tes réponses.

---

9. Boucle d'apprentissage

Principe

La boucle d'apprentissage (LearningLoop) collecte les meilleures réponses LLM et les stocke dans Qdrant pour améliorer continuellement le RAG.

Flux

1. Requête LLM → Réponse
       │
2. Scoring (reward_from_response)
       │
3. Si quality_score >= 0.6 → Stockage dans mtp_knowledge
       │
4. Stockage dans mtp_training_pairs
       │
5. Toutes les 100 requêtes → Analyse des pairs
       │
6. Génération de propositions d'amélioration

Paramètres

Variable	Défaut	Description
`LEARNING_LOOP_ENABLED`	true	Active la boucle
`LEARNING_LOOP_POLL_INTERVAL_S`	60	Intervalle de poll (secondes)
`LEARNING_TRIGGER_EVERY_N`	100	Déclenche après N requêtes
`MIN_DATASET_SIZE_FOR_TRAINING`	50	Taille min dataset
`MIN_QUALITY_SCORE_FOR_TRAINING`	0.6	Score qualité minimal

Stockage des paires d'entraînement

def store_training_pair(self, pair: TrainingPair) -> str:
    vector = self.encode(f"{pair.prompt} {pair.completion}")
    payload = {
        "pair_id": pair.pair_id,
        "prompt": pair.prompt,
        "completion": pair.completion,
        "quality_score": pair.quality_score,
        "source_provider": pair.source_provider,
        "timestamp": pair.timestamp.isoformat(),
        "trained": pair.trained,
    }
    self.client.upsert(collection_name=self.training_collection, ...)

Suivi des paires

État	Description
`trained: false`	Paire en attente de traitement
`trained: true`	Paire déjà utilisée pour l'apprentissage

Statistiques d'apprentissage

def get_learning_stats(self) -> dict:
    return {
        "knowledge_collection": "mtp_knowledge",
        "training_collection": "mtp_training_pairs",
        "knowledge_points": count_knowledge,
        "training_pairs_total": total,
        "training_pairs_pending": pending,
        "training_pairs_trained": total - pending,
    }

---

10. Collections Qdrant

Vue d'ensemble

Collection	Description	Remplissage	Lecture
`ai_responses`	Résultats bruts des appels	Automatique (chaque appel)	Recherche historique
`router_embeddings`	Décisions de routing	Automatique	Analyse
`eval_corpus`	Prompts d'évaluation	Manuel / bot	Benchmark
`mtp_knowledge`	Connaissances distillées	Learning loop + admin	RAG
`mtp_training_pairs`	Paires entraînement	Learning loop	Training

Création automatique

Les collections sont créées automatiquement au démarrage si elles n'existent pas, avec la configuration vectorielle standard :

VectorParams(size=384, distance=Distance.COSINE)

Nommage configurable

Les collections knowledge et training sont configurables :

Variable	Défaut
`QDRANT_KNOWLEDGE_COLLECTION`	`mtp_knowledge`
`QDRANT_TRAINING_COLLECTION`	`mtp_training_pairs`

---

11. Configuration

Variables d'environnement RAG

Variable	Défaut	Description
`QDRANT_URL`	-	URL complète Qdrant
`QDRANT_HOST`	localhost	Hôte Qdrant
`QDRANT_PORT`	6333	Port Qdrant
`QDRANT_KNOWLEDGE_COLLECTION`	mtp_knowledge	Collection knowledge
`QDRANT_TRAINING_COLLECTION`	mtp_training_pairs	Collection training
`REDIS_URL`	redis://localhost:6379/0	URL Redis
`CACHE_TTL_SECONDS`	3600	TTL cache Redis

Variables mémoire conversationnelle

Les paramètres de la mémoire sont codés en dur dans rag.py :

Constante	Valeur	Description
`CONV_PREFIX`	`aisia:conv:`	Préfixe clés Redis
`CONV_TTL_S`	3600	TTL session (1h)
`CONV_MAX_MESSAGES`	20	Messages max par conversation

---

12. Monitoring et diagnostics

Vérifier Qdrant

# Statut des collections
GET /admin/knowledge

Ou directement via l'API Qdrant
curl http://aisia-core_qdrant:6333/collections

Vérifier Redis

# Health check
GET /health
Vérifier redis.status == "ok"

Diagnostiquer le RAG

Si les réponses ne semblent pas utiliser le contexte RAG :

1. Qdrant est-il accessible ? Vérifiez /health → qdrant.status 2. L'encodeur est-il chargé ? Vérifiez les logs au démarrage (Encodeur sentence-transformers ou Encodeur fastembed) 3. La collection a-t-elle du contenu ? Vérifiez points_count > 0 via GET /admin/knowledge 4. Le score minimum est-il trop élevé ? Le défaut est 0.5, ajustable

Métriques

Les métriques RAG sont indirectement visibles via :

Latence des requêtes (ai_request_latency_seconds) : un RAG actif ajoute

quelques ms

Propositions d'apprentissage (ai_learning_proposals_total) : indique

que la boucle fonctionne

Performances typiques

Opération	Latence typique
Encodage d'un prompt (384D)	5-20ms
Recherche Qdrant (top-3)	2-10ms
Lecture Redis (historique)	<1ms
Construction du prompt augmenté	<1ms
Total overhead RAG	10-30ms

L'overhead du RAG est négligeable comparé à la latence du LLM (100ms - 5s).

---

Document AISIA v4.21.0