Spaces:
Running
Running
File size: 2,848 Bytes
790a141 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
"""
MODULE: QUERY PLANNER (OAG)
===========================
Responsabilité : Analyser la question, vérifier le schéma réel,
et décider de la meilleure stratégie (SPARQL vs FAISS vs Multi-étapes).
"""
import json
import re
class QueryPlanner:
def __init__(self, rdf_store, schema_info):
self.rdf = rdf_store
self.schema = schema_info # Provient de SchemaExtractor.get_real_schema()
def analyze_and_plan(self, user_query):
"""
Détermine si la question est :
1. Sémantique (Recherche d'un dossier client/prêt spécifique)
2. Analytique (Comptage, Somme, Croisement global)
3. Invalide (Donnée absente du schéma)
"""
# 1. ANALYSE DES MOTS CLÉS (Simplifiée)
is_analytical = any(word in user_query.lower() for word in [
'combien', 'total', 'somme', 'moyenne', 'liste tous', 'liste les', 'nombre'
])
# 2. VÉRIFICATION DU SCHÉMA (Prévention d'hallucination)
# On regarde si les termes de la question match avec nos prédicats réels
available_preds = self.schema.get("predicates", [])
found_preds = [p for p in available_preds if p.lower() in user_query.lower()]
# 3. DÉCISION DE LA STRATÉGIE
# CAS A : Recherche d'un individu ou d'un dossier précis
if not is_analytical and not found_preds:
return {
"strategy": "OG-RAG",
"tool": "search_semantic",
"reason": "La question semble porter sur un individu ou un cas spécifique non défini par un prédicat exact.",
"steps": [f"Rechercher le bloc contextuel pour : {user_query}"]
}
# CAS B : Question de masse ou croisement (Analytique)
if is_analytical:
return {
"strategy": "OAG-SPARQL",
"tool": "execute_sparql",
"reason": "La question nécessite une agrégation ou un filtrage sur l'ensemble de la base.",
"steps": ["Traduire en requête SPARQL optimisée", "Valider les prédicats contre le schéma réel"]
}
# CAS C : Hybride (Par défaut)
return {
"strategy": "HYBRID",
"tool": "execute_sparql",
"reason": "Usage de prédicats identifiés dans le schéma réel.",
"steps": ["Générer SPARQL avec les prédicats validés"]
}
def get_planning_logs(self, plan):
"""Génère un affichage propre pour le chat utilisateur"""
log = f"🎯 **Stratégie adoptée** : `{plan['strategy']}`\n"
log += f"🤔 **Raisonnement** : {plan['reason']}\n"
log += "📋 **Plan d'exécution** :\n"
for i, step in enumerate(plan['steps'], 1):
log += f" {i}. {step}\n"
return log |