# === CELLULE DE VALIDATION FINALE (validation de contenu, issue #18395) ===
# Genere un rapport JSON structure avec cross-validation.
# Cette version verifie le CONTENU (et pas seulement la presence des champs) :
# * un argument doit etre une formulation reprise du texte source,
# jamais une cle de l'etat interne de l'agent ;
# * un sophisme doit porter un type, une justification et une cible ;
# * une requete PL ne compte comme significative qu'avec un verdict
# ACCEPTED ou REJECTED (pas une simple execution sans reponse).
import json
import os
import re
from datetime import datetime
from typing import Dict, Any, List, Set
# --- Constantes de validation de contenu ---
# Cles de l'etat partage qui ne sont PAS des arguments (defaut mesure #18395).
# Ces noms apparaissent comme valeurs de identified_arguments quand l'agent
# informel a confondu son etat interne avec le contenu du texte.
STATE_KEYS_BLACKLIST: Set[str] = {
"analysis_state", "arguments", "identified_arguments", "identified_fallacies",
"belief_sets", "query_log", "answers", "raw_text", "final_conclusion",
"shared_state", "task_context", "metadata", "state",
}
# Longueur minimale d'une description d'argument (en mots) pour qu'on
# la considere comme une phrase du texte, pas une etiquette.
MIN_ARGUMENT_WORDS: int = 4
# Longueur minimale de la sous-chaine commune (en caracteres) entre
# la description de l'argument et le texte source.
MIN_ARGUMENT_TEXT_OVERLAP: int = 12
def _resolve_chat_model_id() -> str:
"""Substitue les identifiants de modele retires (gpt-5-mini etc.).
Port minimal de resolve_chat_endpoint
(jsboigeEpita/2025-Epita-Intelligence-Symbolique core/llm_service.py L279).
La liste OBSOLETE_MODEL_SUBSTITUTIONS complete sera portee dans une PR
de suivi ; ici on couvre les trois modeles les plus frequemment retires
depuis 2026-09.
"""
OBSOLETE_MODEL_SUBSTITUTIONS = {
"gpt-5-mini": "gpt-4o-mini",
"gpt-5": "gpt-4o",
"gpt-5-nano": "gpt-4o-mini",
}
configured = os.getenv("OPENAI_CHAT_MODEL_ID", "<unset>")
return OBSOLETE_MODEL_SUBSTITUTIONS.get(configured, configured)
def _text_overlap(text: str, candidate: str) -> int:
"""Longueur du plus long suffixe/prefixe commun (proxy rapide).
On evite SequenceMatcher (trop lent) ; on cherche une sous-chaine
contigue d'au moins MIN_ARGUMENT_TEXT_OVERLAP caracteres en minuscule,
insensible aux espaces multiples et a la ponctuation simple.
"""
if not text or not candidate:
return 0
a = re.sub(r"\s+", " ", text.lower())
b = re.sub(r"\s+", " ", candidate.lower())
best = 0
n = len(b)
upper = min(n, 80)
for size in range(upper, MIN_ARGUMENT_TEXT_OVERLAP - 1, -1):
for i in range(0, n - size + 1):
sub = b[i:i + size]
if sub in a:
return size
return 0
def _argument_has_substance(arg_desc: str, raw_text: str) -> bool:
"""Verifie qu'une description d'argument est reellement extraite du texte.
Trois conditions, toutes requises :
1. Description non vide apres strip.
2. Description pas dans STATE_KEYS_BLACKLIST (defaut #18395 : l'agent
confond son etat interne avec les arguments du texte).
3. Description chevauche le texte source sur au moins
MIN_ARGUMENT_TEXT_OVERLAP caracteres (defaut : "analysis_state" /
"arguments" sont des mots-cles internes, ils n'apparaissent pas
dans le texte utilisateur sur la transition energetique).
"""
desc = str(arg_desc).strip()
if not desc:
return False
if desc.lower() in STATE_KEYS_BLACKLIST:
return False
if len(desc.split()) < MIN_ARGUMENT_WORDS:
return False
if _text_overlap(raw_text or "", desc) < MIN_ARGUMENT_TEXT_OVERLAP:
return False
return True
def _fallacy_has_substance(f_data: Dict[str, Any]) -> bool:
"""Verifie qu'un sophisme detecte est reel (defaut #18395 : Type Inconnu).
Conditions :
1. Type non vide, different de "Type Inconnu" / "Unknown" / "" / None.
2. Justification non vide (au moins 10 caracteres, pas la valeur par
defaut "Justification manquante").
3. target_argument_id non null (defaut : sophisme en l'air).
"""
if not isinstance(f_data, dict):
return False
ftype = str(f_data.get("type", "")).strip()
if not ftype or ftype.lower() in ("unknown", "type inconnu", ""):
return False
justification = str(f_data.get("justification", "")).strip()
if not justification or justification.lower() == "justification manquante":
return False
if len(justification) < 10:
return False
target = f_data.get("target_argument_id")
if target is None or target == "":
return False
return True
def generate_validated_analysis_report(state) -> Dict[str, Any]:
"""Genere rapport JSON structure avec cross-validation (controle de contenu).
Sortie : dict avec metadata, informal_analysis, formal_analysis,
cross_validation (avec checks_passed + issues + failed_reason),
conclusion, et resolved_model_id (#18395).
"""
raw_text = getattr(state, "raw_text", None) or ""
report = {
"metadata": {
"timestamp": datetime.now().isoformat(),
"version": "2.1-validated-content",
"text_length": len(raw_text) if raw_text else 0,
"text_snippet": (raw_text[:150] + "...") if raw_text and len(raw_text) > 150 else (raw_text or ""),
"resolved_model_id": _resolve_chat_model_id(),
},
"informal_analysis": {
"arguments": [],
"arguments_substantive_count": 0,
"fallacies": [],
"fallacies_substantive_count": 0,
"taxonomy_families_used": set(),
},
"formal_analysis": {
"belief_sets": [],
"query_results": [],
"queries_meaningful_count": 0,
"consistency_checked": False,
},
"cross_validation": {
"validation_status": "INCOMPLETE",
"confidence_score": 0.0,
"checks_passed": [],
"issues": [],
"failed_reason": None,
},
"conclusion": {
"summary": getattr(state, "final_conclusion", None),
"is_complete": hasattr(state, "final_conclusion") and state.final_conclusion is not None,
},
}
# --- Populate arguments ---
substantive_args: List[Dict[str, Any]] = []
if hasattr(state, "identified_arguments"):
for arg_id, arg_desc in state.identified_arguments.items():
has_fallacy = False
if hasattr(state, "identified_fallacies"):
has_fallacy = any(
f.get("target_argument_id") == arg_id
for f in state.identified_fallacies.values()
)
report["informal_analysis"]["arguments"].append({
"id": arg_id,
"description": str(arg_desc)[:200],
"has_fallacy": has_fallacy,
"is_substantive": _argument_has_substance(arg_desc, raw_text),
})
if _argument_has_substance(arg_desc, raw_text):
substantive_args.append({"id": arg_id, "description": str(arg_desc)})
report["informal_analysis"]["arguments_substantive_count"] = len(substantive_args)
# --- Populate fallacies ---
substantive_fallacies: List[Dict[str, Any]] = []
if hasattr(state, "identified_fallacies"):
for f_id, f_data in state.identified_fallacies.items():
fallacy_type = f_data.get("type", "Unknown") if isinstance(f_data, dict) else str(f_data)
justification = f_data.get("justification", "") if isinstance(f_data, dict) else ""
target_id = f_data.get("target_argument_id") if isinstance(f_data, dict) else None
severity = "HIGH" if any(kw in str(fallacy_type).lower() for kw in ["manipulation", "tromperie"]) else "MEDIUM"
report["informal_analysis"]["fallacies"].append({
"id": f_id,
"type": fallacy_type,
"justification": justification,
"target_id": target_id,
"severity": severity,
"is_substantive": _fallacy_has_substance(f_data),
})
if isinstance(fallacy_type, str) and "/" in fallacy_type:
report["informal_analysis"]["taxonomy_families_used"].add(fallacy_type.split("/")[0])
if _fallacy_has_substance(f_data):
substantive_fallacies.append({"id": f_id, "type": fallacy_type})
report["informal_analysis"]["fallacies_substantive_count"] = len(substantive_fallacies)
report["informal_analysis"]["taxonomy_families_used"] = list(
report["informal_analysis"]["taxonomy_families_used"]
)
# --- Populate belief sets ---
if hasattr(state, "belief_sets"):
for bs_id, bs_data in state.belief_sets.items():
content = bs_data.get("content", "") if isinstance(bs_data, dict) else str(bs_data)
report["formal_analysis"]["belief_sets"].append({
"id": bs_id,
"logic_type": bs_data.get("logic_type", "PL") if isinstance(bs_data, dict) else "PL",
"formula_count": content.count("\n") + 1 if content else 0,
"is_consistent": "NOT_CHECKED",
})
# --- Populate query results ---
meaningful_count = 0
if hasattr(state, "query_log"):
for qlog in state.query_log:
raw_result = qlog.get("raw_result", "") if isinstance(qlog, dict) else ""
status = "UNKNOWN"
if "ACCEPTED" in str(raw_result):
status = "ACCEPTED"
elif "REJECTED" in str(raw_result):
status = "REJECTED"
elif "FUNC_ERROR" in str(raw_result):
status = "ERROR"
if status in ("ACCEPTED", "REJECTED"):
meaningful_count += 1
report["formal_analysis"]["query_results"].append({
"log_id": qlog.get("log_id", "") if isinstance(qlog, dict) else "",
"belief_set_id": qlog.get("belief_set_id", "") if isinstance(qlog, dict) else "",
"query": qlog.get("query", "") if isinstance(qlog, dict) else "",
"status": status,
})
report["formal_analysis"]["queries_meaningful_count"] = meaningful_count
# --- CROSS-VALIDATION (contenu) ---
checks: List[str] = []
issues: List[str] = []
failed_reasons: List[str] = []
# Check 1 : arguments REELS (defaut : "analysis_state" / "arguments" passaient)
if substantive_args:
checks.append("ARGUMENTS_IDENTIFIED")
elif len(report["informal_analysis"]["arguments"]) > 0:
# Arguments presents mais aucun n'est une vraie sous-chaine du texte
issues.append(
f"{len(report['informal_analysis']['arguments'])} argument(s) detecte(s) "
f"mais aucun ne provient du texte (cles d'etat partage ou libelles internes)."
)
failed_reasons.append("ARGUMENTS_FORM_ONLY")
else:
issues.append("Aucun argument identifie.")
failed_reasons.append("ARGUMENTS_NONE")
# Check 2 : sophismes REELS (defaut : "Type Inconnu" passait via FALLACY_ANALYSIS_ATTEMPTED)
if substantive_fallacies:
checks.append("FALLACIES_ANALYZED")
elif len(report["informal_analysis"]["fallacies"]) > 0:
issues.append(
f"{len(report['informal_analysis']['fallacies'])} sophisme(s) detecte(s) "
f"mais aucun n'a de type/justification/cible valides."
)
failed_reasons.append("FALLACIES_FORM_ONLY")
elif hasattr(state, "answers") and any("sophisme" in str(v).lower() for v in state.answers.values()):
checks.append("FALLACY_ANALYSIS_ATTEMPTED")
else:
issues.append("Analyse sophismes non effectuee.")
failed_reasons.append("FALLACIES_NONE")
# Check 3 : belief set PL cree
if len(report["formal_analysis"]["belief_sets"]) > 0:
checks.append("BELIEF_SET_CREATED")
else:
issues.append("Aucun Belief Set PL cree.")
failed_reasons.append("BELIEF_SET_NONE")
# Check 4a : au moins une requete soumise
if len(report["formal_analysis"]["query_results"]) > 0:
checks.append("QUERIES_SUBMITTED")
else:
issues.append("Aucune requete PL soumise.")
failed_reasons.append("QUERIES_NONE")
# Check 4b : au moins une requete avec verdict (ACCEPTED ou REJECTED)
if meaningful_count > 0:
checks.append("QUERIES_MEANINGFUL")
else:
issues.append("Aucune requete PL n'a produit de verdict (ACCEPTED/REJECTED).")
failed_reasons.append("QUERIES_NOT_MEANINGFUL")
# Check 5 : conclusion generee
if report["conclusion"]["is_complete"]:
checks.append("CONCLUSION_GENERATED")
else:
issues.append("Conclusion finale non generee.")
failed_reasons.append("CONCLUSION_NONE")
# --- Score et statut ---
max_checks = 7 # ARGUMENTS, FALLACIES, BELIEF_SET, QUERIES_SUBMITTED, QUERIES_MEANINGFUL, CONCLUSION (+ split)
confidence = len(checks) / max_checks
report["cross_validation"]["confidence_score"] = round(confidence, 2)
report["cross_validation"]["checks_passed"] = checks
report["cross_validation"]["issues"] = issues
report["cross_validation"]["failed_reason"] = failed_reasons[0] if failed_reasons else None
# Determination du statut : un ARGUMENTS_FORM_ONLY ou FALLACIES_FORM_ONLY
# degrade systematiquement le statut, sans annuler entierement le
# pipeline (on garde un diagnostic STRUCTURED plutot que INCOMPLETE
# binaire). Justification : la validation precedente disait 67%
# PARTIAL_VALIDATED sur un run dont les 2 arguments etaient des
# mots-cles d'etat -- c'est exactement le bug corrige ici.
if any(fr in ("ARGUMENTS_FORM_ONLY", "FALLACIES_FORM_ONLY") for fr in failed_reasons):
report["cross_validation"]["validation_status"] = "INVALIDATED_FORM"
elif confidence >= 0.8:
report["cross_validation"]["validation_status"] = "COMPLETE_VALIDATED"
elif confidence >= 0.5:
report["cross_validation"]["validation_status"] = "PARTIAL_VALIDATED"
elif confidence >= 0.3:
report["cross_validation"]["validation_status"] = "MINIMAL"
else:
report["cross_validation"]["validation_status"] = "INCOMPLETE"
return report
def display_validation_summary(report: Dict[str, Any]) -> str:
"""Affiche resume lisible du rapport de validation."""
print("\n" + "=" * 70)
print(" RAPPORT D'ANALYSE RHETORIQUE VALIDEE")
print("=" * 70)
cv = report["cross_validation"]
status_symbols = {
"COMPLETE_VALIDATED": "[OK]",
"PARTIAL_VALIDATED": "[PARTIEL]",
"MINIMAL": "[MINIMAL]",
"INCOMPLETE": "[INCOMPLET]",
"INVALIDATED_FORM": "[FORME_INVALIDE]",
}
print(f"\n STATUT: {status_symbols.get(cv['validation_status'], '?')} {cv['validation_status']}")
print(f" CONFIANCE: {cv['confidence_score'] * 100:.0f}%")
if cv["failed_reason"]:
print(f" ECHEC CONTENU: {cv['failed_reason']}")
print(f"\n [ANALYSE INFORMELLE]")
print(f" Arguments: {len(report['informal_analysis']['arguments'])} "
f"(substantifs: {report['informal_analysis']['arguments_substantive_count']})")
print(f" Sophismes: {len(report['informal_analysis']['fallacies'])} "
f"(substantifs: {report['informal_analysis']['fallacies_substantive_count']})")
print(f"\n [ANALYSE FORMELLE]")
print(f" Belief Sets: {len(report['formal_analysis']['belief_sets'])}")
print(f" Requetes: {len(report['formal_analysis']['query_results'])} "
f"(significatives: {report['formal_analysis']['queries_meaningful_count']})")
if report["metadata"].get("resolved_model_id"):
print(f"\n [MODELE]")
print(f" Modele resolu: {report['metadata']['resolved_model_id']}")
print(f"\n [VALIDATIONS PASSEES]")
for check in cv["checks_passed"]:
print(f" [+] {check}")
if cv["issues"]:
print(f"\n [PROBLEMES DETECTES]")
for issue in cv["issues"]:
print(f" [-] {issue}")
print(f"\n [CONCLUSION]")
if report["conclusion"]["is_complete"]:
conclusion_preview = str(report["conclusion"]["summary"])[:200]
print(f" {conclusion_preview}...")
else:
print(" (Non generee)")
print("\n" + "=" * 70)
return cv["validation_status"]
# === EXECUTION DE LA VALIDATION ===
print("\n--- Generation du Rapport d'Analyse Validee ---")
print(f"Modele resolu : {_resolve_chat_model_id()}")
# Chercher l'etat local_state defini par l'orchestration
if "local_state" in dir() and local_state is not None:
# Generate report
final_report = generate_validated_analysis_report(local_state)
# Display summary
validation_status = display_validation_summary(final_report)
# Export JSON
output_path = "output/analysis_report.json"
try:
os.makedirs("output", exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
json.dump(final_report, f, indent=2, ensure_ascii=False, default=str)
print(f"\nRapport JSON exporte vers: {output_path}")
except Exception as e:
print(f"Erreur export JSON: {e}")
# Afficher JSON complet
print("\n--- RAPPORT JSON COMPLET ---")
print(json.dumps(final_report, indent=2, ensure_ascii=False, default=str))
# Final verdict
print("\n--- VERDICT FINAL ---")
if validation_status == "COMPLETE_VALIDATED":
print("[OK] Analyse COMPLETE et validee.")
elif validation_status == "PARTIAL_VALIDATED":
print("[PARTIEL] Analyse partielle : certains controles de contenu ont echoue.")
elif validation_status == "INVALIDATED_FORM":
print("[INVALIDE] Sortie degeneree : les champs sont presents mais le contenu est vide/incorrect.")
print(" Voir failed_reason et issues pour le diagnostic detaille.")
else:
print(f"[{validation_status}] Verifier les problemes detectes ci-dessus.")
else:
print("Etat 'local_state' non disponible : orchestrateur non execute.")
print("La validation ne peut pas tourner sans un etat issu d'un run agentique.")