Spaces:
Running
Running
| """ | |
| MODULE: SCHEMA EXTRACTOR (TYPE AWARE) | |
| ===================================== | |
| Responsabilité : Extraire le schéma réel DU GRAPHE pour informer l'Agent des types de données disponibles. | |
| """ | |
| class SchemaExtractor: | |
| def __init__(self, rdf_store): | |
| self.rdf_store = rdf_store | |
| def get_real_schema(self): | |
| """Récupère les classes, les propriétés et leurs types de données observés""" | |
| g = self.rdf_store.g | |
| schema = { | |
| "classes": {}, | |
| "predicates": [], | |
| "properties_by_class": {} | |
| } | |
| # 1. Compter les entités par classe | |
| q_classes = "SELECT ?type (COUNT(?s) as ?count) WHERE { ?s a ?type } GROUP BY ?type" | |
| for row in g.query(q_classes): | |
| cls_name = str(row[0]).split('#')[-1] | |
| schema["classes"][cls_name] = int(row[1]) | |
| # 2. Extraire les propriétés et deviner leur type | |
| for cls in schema["classes"]: | |
| # Cette requête regarde la donnée réelle (?o) et demande son type (datatype(?o)) | |
| q_props = f""" | |
| SELECT DISTINCT ?p ?datatype WHERE {{ | |
| ?s a vortex:{cls} . | |
| ?s ?p ?o . | |
| BIND(datatype(?o) AS ?datatype) | |
| FILTER(STRSTARTS(STR(?p), "http://vortex.ai/ontology#")) | |
| }} | |
| """ | |
| props = [] | |
| for row in g.query(q_props): | |
| pred = str(row[0]).split('#')[-1] | |
| # Si datatype est vide, c'est probablement un String ou une Relation | |
| dtype = str(row[1]).split('#')[-1] if row[1] else "string/obj" | |
| # On ajoute le prédicat à la liste globale | |
| if pred not in schema["predicates"]: | |
| schema["predicates"].append(pred) | |
| props.append(f"{pred} ({dtype})") | |
| schema["properties_by_class"][cls] = props | |
| return schema | |
| def generate_prompt_schema(self, ontology_rules=None): | |
| """Génère le texte de contexte pour le LLM""" | |
| schema = self.get_real_schema() | |
| lines = ["--- RÉALITÉ DU GRAPH RDF (TYPÉE) ---"] | |
| for cls, count in schema["classes"].items(): | |
| props = ", ".join(schema["properties_by_class"].get(cls, [])) | |
| lines.append(f"[LOT] {cls} ({count} items) : {props}") | |
| if ontology_rules: | |
| lines.append("\n--- DICTIONNAIRE MÉTIER DE LA FEUILLE ONTOLOGY ---") | |
| for rule in ontology_rules: | |
| subject = str(rule.get("SubjectClass", "")).strip() | |
| predicate = str(rule.get("Predicate", "")).strip() | |
| if not subject or not predicate: | |
| continue | |
| description = str(rule.get("Description_Metier", "")).strip() | |
| description = "" if description.lower() in {"nan", "none"} else description | |
| details = [ | |
| f"feuille={str(rule.get('Sheet', '')).strip()}", | |
| f"colonne_sujet={str(rule.get('SubjectCol', '')).strip()}", | |
| f"colonne_objet={str(rule.get('ObjectColOrConcept', '')).strip()}", | |
| f"type={str(rule.get('ObjectType', '')).strip()}", | |
| f"range={str(rule.get('Range', '')).strip()}", | |
| f"cardinalité={str(rule.get('Cardinality', '')).strip()}", | |
| ] | |
| if description: | |
| details.append(f"sens_métier={description}") | |
| lines.append(f"- {subject}.{predicate} ({'; '.join(details)})") | |
| return "\n".join(lines) |