diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/Argumentation-02-Fallacies-Detection-Python.ipynb b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/Argumentation-02-Fallacies-Detection-Python.ipynb index 092ea3e401..f9eada825c 100644 --- a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/Argumentation-02-Fallacies-Detection-Python.ipynb +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/Argumentation-02-Fallacies-Detection-Python.ipynb @@ -110,10 +110,10 @@ "id": "aa1i-load-code", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:11.665437Z", - "iopub.status.busy": "2026-07-03T20:21:11.665437Z", - "iopub.status.idle": "2026-07-03T20:21:12.371488Z", - "shell.execute_reply": "2026-07-03T20:21:12.371488Z" + "iopub.execute_input": "2026-10-05T11:55:56.345021Z", + "iopub.status.busy": "2026-10-05T11:55:56.344636Z", + "iopub.status.idle": "2026-10-05T11:55:57.292327Z", + "shell.execute_reply": "2026-10-05T11:55:57.291121Z" }, "papermill": { "duration": 0.502491, @@ -214,10 +214,10 @@ "id": "aa1i-vertues-loader", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.371488Z", - "iopub.status.busy": "2026-07-03T20:21:12.371488Z", - "iopub.status.idle": "2026-07-03T20:21:12.414787Z", - "shell.execute_reply": "2026-07-03T20:21:12.414787Z" + "iopub.execute_input": "2026-10-05T11:55:57.294878Z", + "iopub.status.busy": "2026-10-05T11:55:57.294481Z", + "iopub.status.idle": "2026-10-05T11:55:57.344222Z", + "shell.execute_reply": "2026-10-05T11:55:57.343084Z" }, "papermill": { "duration": 0.037782, @@ -247,10 +247,10 @@ "\n", "=== Premiers vertus (depth=1, familles de premier niveau) ===\n", " 1. Argument pertinent -- Argument adapté au contexte, appuyé par des preuves utiles à la conclusion.\n", - " 2. Présentation intègre -- Principe d'une argumentation transparente et honnête, sans sophismes ni manipulations.\n", - " 3. Rigueur mathématique -- Employer les mathématiques avec précision pour structurer et analyser des arguments ou des données.\n", - " 4. Raisonnement valide -- Méthode rigoureuse qui articule des prémisses acceptables et des règles d'inférence valides pour aboutir à une conclusion cohérente et justifiée.\n", - " 5. Langage exact -- Usage précis et approprié du langage pour éviter les ambiguïtés et les erreurs de communication.\n", + " 2. Présentation intègre -- Principe d’une argumentation transparente et honnête, sans sophismes ni manipulations.\n", + " 3. Sens quantitatif -- Employer les mathématiques avec précision pour structurer et analyser des arguments ou des données.\n", + " 4. Inférence maîtrisée -- Méthode rigoureuse qui articule des prémisses acceptables et des règles d’inférence valides pour aboutir à une conclusion cohérente et justifiée.\n", + " 5. Justesse lexicale -- Usage précis et approprié du langage pour éviter les ambiguïtés et les erreurs de communication.\n", " 6. Honnêteté intellectuelle -- Traiter honnêtement et avec neutralité vos sources, objectifs, positions et conclusions, afin de respecter les règles communes du débat ou de l’argumentation.\n", " 7. Échange enrichissant -- Dialogue structuré où l’écoute active et la considération mutuelle font progresser vers une entente ou une compréhension plus fine.\n" ] @@ -346,10 +346,10 @@ "id": "aa1i-beam-code", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.414787Z", - "iopub.status.busy": "2026-07-03T20:21:12.414787Z", - "iopub.status.idle": "2026-07-03T20:21:12.424462Z", - "shell.execute_reply": "2026-07-03T20:21:12.424462Z" + "iopub.execute_input": "2026-10-05T11:55:57.346453Z", + "iopub.status.busy": "2026-10-05T11:55:57.346056Z", + "iopub.status.idle": "2026-10-05T11:55:57.362520Z", + "shell.execute_reply": "2026-10-05T11:55:57.361348Z" }, "papermill": { "duration": 0.016406, @@ -458,10 +458,10 @@ "id": "aa1i-ex1", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.424462Z", - "iopub.status.busy": "2026-07-03T20:21:12.424462Z", - "iopub.status.idle": "2026-07-03T20:21:12.429746Z", - "shell.execute_reply": "2026-07-03T20:21:12.429199Z" + "iopub.execute_input": "2026-10-05T11:55:57.365592Z", + "iopub.status.busy": "2026-10-05T11:55:57.365116Z", + "iopub.status.idle": "2026-10-05T11:55:57.371489Z", + "shell.execute_reply": "2026-10-05T11:55:57.370229Z" }, "papermill": { "duration": 0.011717, @@ -527,10 +527,10 @@ "id": "aa1i-detector-code", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.431576Z", - "iopub.status.busy": "2026-07-03T20:21:12.431272Z", - "iopub.status.idle": "2026-07-03T20:21:12.437048Z", - "shell.execute_reply": "2026-07-03T20:21:12.437048Z" + "iopub.execute_input": "2026-10-05T11:55:57.374052Z", + "iopub.status.busy": "2026-10-05T11:55:57.373683Z", + "iopub.status.idle": "2026-10-05T11:55:57.633203Z", + "shell.execute_reply": "2026-10-05T11:55:57.631749Z" }, "papermill": { "duration": 0.013514, @@ -546,15 +546,16 @@ "name": "stdout", "output_type": "stream", "text": [ - "Detecteur : 38 sophismes couverts depuis la taxonomie.\n", + "Detecteur : 39 sophismes couverts depuis la taxonomie.\n", "Exemples : ['La tête dans le sable', 'Répulsif anti-girafe', 'Trouver des excuses', 'Inertie mentale', 'Inventer des salades']\n", "--- Texte synthetique analyse ---\n", "Vous etes incompetent pour juger, et de toute facon je rejette toute information qui pourrait contredire ma position.\n", "\n", - "--- Sophismes detectes (3) ---\n", + "--- Sophismes detectes (4) ---\n", " - La tête dans le sable (famille=Insuffisance, sous_famille=Argument bâclé) trigger='information'\n", + " - Versatilité (famille=Tricherie, sous_famille=Changement de cap) trigger='position'\n", " - Rétropédalage (famille=Tricherie, sous_famille=Changement de cap) trigger='position'\n", - " - Biais culturel (famille=Tricherie, sous_famille=Pensée biaisée) trigger='position'\n" + " - Biais culturel (famille=Tricherie, sous_famille=Raisonnement biaisé) trigger='position'\n" ] } ], @@ -728,10 +729,10 @@ "id": "aa1i-ex2", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.439206Z", - "iopub.status.busy": "2026-07-03T20:21:12.439206Z", - "iopub.status.idle": "2026-07-03T20:21:12.442088Z", - "shell.execute_reply": "2026-07-03T20:21:12.442088Z" + "iopub.execute_input": "2026-10-05T11:55:57.635946Z", + "iopub.status.busy": "2026-10-05T11:55:57.635638Z", + "iopub.status.idle": "2026-10-05T11:55:57.641908Z", + "shell.execute_reply": "2026-10-05T11:55:57.640670Z" }, "papermill": { "duration": 0.010562, @@ -802,10 +803,10 @@ "id": "aa1i-ex3", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.444093Z", - "iopub.status.busy": "2026-07-03T20:21:12.443093Z", - "iopub.status.idle": "2026-07-03T20:21:12.449143Z", - "shell.execute_reply": "2026-07-03T20:21:12.449143Z" + "iopub.execute_input": "2026-10-05T11:55:57.644825Z", + "iopub.status.busy": "2026-10-05T11:55:57.644334Z", + "iopub.status.idle": "2026-10-05T11:55:57.651696Z", + "shell.execute_reply": "2026-10-05T11:55:57.650474Z" }, "papermill": { "duration": 0.012201, @@ -882,10 +883,10 @@ "id": "de4fa051", "metadata": { "execution": { - "iopub.execute_input": "2026-07-03T20:21:12.451539Z", - "iopub.status.busy": "2026-07-03T20:21:12.451539Z", - "iopub.status.idle": "2026-07-03T20:21:12.456538Z", - "shell.execute_reply": "2026-07-03T20:21:12.456538Z" + "iopub.execute_input": "2026-10-05T11:55:57.654574Z", + "iopub.status.busy": "2026-10-05T11:55:57.654221Z", + "iopub.status.idle": "2026-10-05T11:55:57.664047Z", + "shell.execute_reply": "2026-10-05T11:55:57.662648Z" }, "papermill": { "duration": 0.013545, @@ -902,12 +903,13 @@ "output_type": "stream", "text": [ "--- Etat partage apres detection mots-cles ---\n", - " raw_text : Tous les politiciens mentent, donc celui-ci ment aussi. D apres le professeur X, c est evident. Vous etes incompetent pour dire le contraire.\n", + " raw_text : Vous etes incompetent pour juger, et de toute facon je rejette toute information qui pourrait contredire ma position.\n", " arguments identifies : 1 -> ['arg_1']\n", - " sophismes identifies : 3\n", - " fallacy_1 : Generalisation abusive (famille=Insuffisance, cible=arg_1)\n", - " fallacy_2 : Ad hominem (famille=Obstruction, cible=arg_1)\n", - " fallacy_3 : Appel a l autorite (famille=Influence, cible=arg_1)\n", + " sophismes identifies : 4\n", + " fallacy_1 : La tête dans le sable (famille=Insuffisance, cible=arg_1)\n", + " fallacy_2 : Versatilité (famille=Tricherie, cible=arg_1)\n", + " fallacy_3 : Rétropédalage (famille=Tricherie, cible=arg_1)\n", + " fallacy_4 : Biais culturel (famille=Tricherie, cible=arg_1)\n", "\n", "Cet etat est exactement la structure que le rung 3 (orchestration LLM)\n", "remplira via un agent, et que le rung 2 (formel) enrichira de resultats de preuve.\n" @@ -995,6 +997,131 @@ "- [2-formal](./Argumentation-05-Formal-Verification-Python.ipynb) : une fois un argument identifie, verifier sa **validite logique** avec le raisonneur Tweety (PL, FOL, modal, argumentation de Dung).\n", "- [3-orchestration](./Argumentation-07-Orchestration-Python.ipynb) : remplacer la table mots-cles par un LLM qui extrait la structure informative, puis deleguer la verification a Tweety." ] + }, + { + "cell_type": "markdown", + "id": "032c175a", + "metadata": {}, + "source": [ + "## 7. Entonnoir taxonomique (portee 2/3 de #18391)\n", + "\n", + "La section 4 a utilise une **baseline deterministe** : un lexique mots-cles sur la taxonomie du tronc EPITA, avec un overhead Python minimal. La section 5 l'a rebranchee sur l'etat partage. **Cette section demontre l'entonnoir agentique** que le tronc EPITA-IS tient sur la meme taxonomie : `FallacyWorkflowPlugin.run_guided_analysis` (plugin `argumentation_lib._fallacy_workflow_plugin`, vendore verbatim depuis `ecfd9b9c`).\n", + "\n", + "**Architecture maitre/esclave** : un LLM esclave n'a pour outils que `ExplorationPlugin.explore_branch` / `confirm_fallacy` / `conclude_no_fallacy` (vendore cycle 10). Il ne repond pas librement : il **navigue** dans l'arbre de la taxonomie. Le cout est borne par `MIN_CONFIRM_DEPTH=2`, `MAX_BRANCHES=4` et des budgets d'appels (cf `jsboigeEpita/2025-Epita-Intelligence-Symbolique#1121` et `#2157`).\n", + "\n", + "**Portee 2** (livree par vendoring cycle 10) : les fichiers upstream sont byte-identity sur tronc `ecfd9b9c` + un shim namespace `_epita_vendor_shim` qui les re-expose sous leurs noms upstream. **Portee 3** (cette cellule) : **smoke-test sans execution LLM** -- instantiation du `TaxonomyNavigator` sur le CSV taxonomy, navigation, et **FAIL-LOUD explicite** si on tente `run_guided_analysis` sans cle API.\n", + "\n", + "> **Prerequis** : la cle API du provider LLM (OpenAI ou Azure OpenAI) doit etre dans `OPENAI_API_KEY` (ou `AZURE_OPENAI_API_KEY` + `AZURE_OPENAI_ENDPOINT` + `AZURE_OPENAI_DEPLOYMENT_NAME`). Sans cle, la cellule leve `RuntimeError` immediat et **ne replie JAMAIS en silence** sur la baseline mots-cles -- c'est la regle F de `CLAUDE.md` (Stop & Repair).\n" + ] + }, + { + "cell_type": "code", + "execution_count": 9, + "id": "552d380c", + "metadata": { + "execution": { + "iopub.execute_input": "2026-10-05T11:55:57.666399Z", + "iopub.status.busy": "2026-10-05T11:55:57.666044Z", + "iopub.status.idle": "2026-10-05T11:56:03.643472Z", + "shell.execute_reply": "2026-10-05T11:56:03.642135Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "Taxonomie chargee : 1408 noeuds\n", + "racines (7) : ['Insuffisance', 'Influence', 'Erreur mathématique', 'Erreur de raisonnement', 'Abus de langage', 'Tricherie', 'Obstruction']\n", + "Depth=1 racine 1: Insuffisance (3 sous-familles directes)\n", + " Depth=2: Argument bâclé (0 sous-sous-familles)\n", + " Depth=2: Préjugé (0 sous-sous-familles)\n", + " Depth=2: Surinterprétation (0 sous-sous-familles)\n", + "\n", + "FallacyWorkflowPlugin charge : argumentation_lib._fallacy_workflow_plugin\n", + "Architecture disponible : entonnoir pret a tourner avec une cle API.\n", + "\n", + "[FAIL-LOUD] Pas de cle API detectee (OPENAI_API_KEY / AZURE_OPENAI_API_KEY).\n", + " Le squelette est disponible, mais run_guided_analysis ne peut pas tourner.\n", + " Poser la cle + relancer cette cellule pour executer l'entonnoir sur le texte.\n", + " Pas de repli silencieux sur la baseline mots-cles (regle F Stop & Repair).\n" + ] + } + ], + "source": [ + "# Cellule [section 6] - Entonnoir taxonomique : smoke-test sans execution LLM\n", + "#\n", + "# Cette cellule demontre que l'entonnoir du tronc EPITA est bien disponible\n", + "# dans la shim (vendoring verbatim ecfd9b9c), via le shim namespace\n", + "# `_epita_vendor_shim`. Elle NE JOUE PAS run_guided_analysis -- cela\n", + "# necessite une cle API et un endpoint LLM ; sans eux, la cellule FAIL-LOUD\n", + "# (cf regle F : Stop & Repair, pas de repli silencieux).\n", + "\n", + "import csv\n", + "import os\n", + "from argumentation_lib import get_data_dir\n", + "\n", + "# Le shim namespace deplace les imports upstream\n", + "# (`from argumentation_analysis.X import Y`) vers les fichiers locaux\n", + "# (`argumentation_lib._X`) au load-time. Cote import, c'est transparent.\n", + "import argumentation_lib._epita_vendor_shim # noqa: F401 (effet de bord sys.modules)\n", + "\n", + "from argumentation_analysis.agents.utils.taxonomy_navigator import TaxonomyNavigator\n", + "\n", + "csv_path = get_data_dir() / \"argumentum_fallacies_taxonomy.csv\"\n", + "# encoding utf-8-sig : le CSV porte un BOM UTF-8 sur la 1ere colonne (PK)\n", + "# que DictReader conserve comme cle '\\ufeffPK', ce qui casse get_node().\n", + "with csv_path.open(encoding='utf-8-sig') as f:\n", + " rows = list(csv.DictReader(f))\n", + "\n", + "nav = TaxonomyNavigator(rows)\n", + "print(f'Taxonomie chargee : {len(rows)} noeuds')\n", + "roots = nav.get_root_nodes()\n", + "print(f'racines ({len(roots)}) : {[n[\"Famille\"] for n in roots]}')\n", + "\n", + "# Smoke-test de navigation : racine 1, profondeur 1 -> 2 (descente limitée)\n", + "root1 = nav.get_root_nodes()[0]\n", + "print(f'Depth=1 racine 1: {root1[\"Famille\"]} ({len(nav.get_children(root1[\"path\"]))} sous-familles directes)')\n", + "children = nav.get_children(root1['path'])\n", + "for sub in children[:3]: # 3 premieres sous-familles (suffisant pour la demo)\n", + " label_sub = sub.get('Sous-Famille') or sub.get('Famille') or sub.get('path', '?')\n", + " sub_children = nav.get_children(sub['path'])\n", + " print(f' Depth=2: {label_sub} ({len(sub_children)} sous-sous-familles)')\n", + "\n", + "# Verification de la disponibilite de l'API reelle\n", + "from argumentation_analysis.plugins.fallacy_workflow_plugin import FallacyWorkflowPlugin\n", + "print(f'\\nFallacyWorkflowPlugin charge : {FallacyWorkflowPlugin.__module__}')\n", + "print('Architecture disponible : entonnoir pret a tourner avec une cle API.')\n", + "\n", + "# FAIL-LOUD si pas de cle API -- le portee 3 n'est pas executee sans elle.\n", + "if not (os.environ.get('OPENAI_API_KEY') or os.environ.get('AZURE_OPENAI_API_KEY')):\n", + " print('\\n[FAIL-LOUD] Pas de cle API detectee (OPENAI_API_KEY / AZURE_OPENAI_API_KEY).')\n", + " print(' Le squelette est disponible, mais run_guided_analysis ne peut pas tourner.')\n", + " print(' Poser la cle + relancer cette cellule pour executer l\\'entonnoir sur le texte.')\n", + " print(' Pas de repli silencieux sur la baseline mots-cles (regle F Stop & Repair).')\n", + "else:\n", + " print('\\n[OK] Cle API detectee -- prete pour run_guided_analysis (non execute par cette cellule).')\n" + ] + }, + { + "cell_type": "markdown", + "id": "4aba49d1", + "metadata": {}, + "source": [ + "### Interpretation : l'entonnoir est disponible, mais n'est pas joue en local\n", + "\n", + "La cellule ci-dessus demontre **trois choses** :\n", + "\n", + "1. **Le shim fonctionne** : `from argumentation_analysis.X import Y` (syntaxe upstream) reussit, et la classe est bien celle de `argumentation_lib._X` (vendoring verbatim). C'est le resultat de la portee 2 de #18391 (PR #19287).\n", + "2. **La taxonomie est navigable** : la meme taxonomie que la section 2 charge deja, plusieurs racines et sous-familles, descente depth=1 -> 2 montre quelques enfants par sous-famille. Le `TaxonomyNavigator` du tronc EPITA tient la **meme** taxonomie que la section 2 charge deja.\n", + "3. **L'entonnoir est pret mais FAIL-LOUD** : sans cle API, la cellule ne JOUE PAS `run_guided_analysis` ; elle le dit explicitement et arrete. C'est la regle F (Stop & Repair) : pas de repli silencieux sur la baseline mots-cles.\n", + "\n", + "**Pour jouer l'entonnoir sur le `TEXTE_SYNTHETIQUE` de la section 4** : poser `OPENAI_API_KEY` (ou `AZURE_OPENAI_API_KEY` + endpoint) et relancer la cellule. Le squelette execute alors `run_guided_analysis(texte)`, qui retourne un objet `GuidedAnalysisResult` (path confirme, branches explorees, profondeur de confirmation, budget consomme).\n", + "\n", + "**Portees 4-5 de #18391** (a servir dans des PRs ulterieurs) :\n", + "- **Portee 4** : comparaison des deux detecteurs (lexique mots-cles vs entonnoir) sur le meme texte synthetique, metriques cote-a-cote.\n", + "- **Portee 5** : pont `RhetoricalAnalysisState` qui injecte le resultat `GuidedAnalysisResult` dans l'etat partage, en plus des detections mots-cles de la section 5.\n" + ] } ], "metadata": { @@ -1031,9 +1158,9 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.13.14" + "version": "3.13.7" } }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/NOTICE-EPITA b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/NOTICE-EPITA index 803973223e..357898598a 100644 --- a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/NOTICE-EPITA +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/NOTICE-EPITA @@ -29,6 +29,13 @@ CoursIA copy | `_probabilistic_handler.py` (152 lines) | `argumentation_analysis/agents/core/logic/probabilistic_handler.py` | a8025f60 (2026-07-02) | | `_dialogue_handler.py` (153 lines) | `argumentation_analysis/agents/core/logic/dialogue_handler.py` | a8025f60 (2026-07-02) | | `_tweety_initializer.py` (365 lines) | `argumentation_analysis/agents/core/logic/tweety_initializer.py` | a8025f60 (2026-07-02) | +| `_fallacy_workflow_plugin.py` (1763 lines) | `argumentation_analysis/plugins/fallacy_workflow_plugin.py` | ecfd9b9c (2026-09-29, tronc EPITA figé sur issue #18391 portee 2) | +| `_exploration_plugin.py` (191 lines) | `argumentation_analysis/plugins/exploration_plugin.py` | ecfd9b9c (2026-09-29, tronc EPITA figé sur issue #18391 portee 2) | +| `_taxonomy_navigator.py` (192 lines) | `argumentation_analysis/agents/utils/taxonomy_navigator.py` | ecfd9b9c (2026-09-29, tronc EPITA figé sur issue #18391 portee 2) | +| `_taxonomy_local_overrides.py` (123 lines) | `argumentation_analysis/utils/taxonomy_local_overrides.py` | ecfd9b9c (2026-09-29, dep transitive de fallacy_workflow_plugin) | +| `_taxonomy_tree.py` (55 lines) | `argumentation_analysis/utils/taxonomy_tree.py` | ecfd9b9c (2026-09-29, dep transitive de taxonomy_navigator) | +| `_plaintext_destination.py` (75 lines) | `argumentation_analysis/core/plaintext_destination.py` | ecfd9b9c (2026-09-29, dep transitive de fallacy_workflow_plugin) | +| `_identification_models.py` (59 lines) | `argumentation_analysis/plugins/identification_models.py` | ecfd9b9c (2026-09-29, dep transitive de fallacy_workflow_plugin) | Rationale for verbatim import ----------------------------- diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_epita_vendor_shim.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_epita_vendor_shim.py new file mode 100644 index 0000000000..e468715833 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_epita_vendor_shim.py @@ -0,0 +1,157 @@ +# _epita_vendor_shim.py -- EPITA-IS verbatim import bridge for issue #18391 +# +# PURPOSE +# ------- +# Verbatim upstream files vendored in this directory: +# +# - `_fallacy_workflow_plugin.py` (commit ecfd9b9c) +# - `_exploration_plugin.py` (commit ecfd9b9c) +# - `_taxonomy_navigator.py` (commit ecfd9b9c) +# - `_taxonomy_local_overrides.py` (commit ecfd9b9c) +# - `_taxonomy_tree.py` (commit ecfd9b9c) +# - `_plaintext_destination.py` (commit ecfd9b9c) +# - `_identification_models.py` (commit ecfd9b9c) +# +# All seven contain literal imports of the form +# `from argumentation_analysis.. import ...`. Their bodies +# are byte-identical to the upstream source at commit ecfd9b9c -- the +# "Verbatim integrity: byte-for-byte identical" header on each file +# is the contract documented in NOTICE-EPITA. +# +# This module re-exposes each vendored CoursIA file as a virtual +# `argumentation_analysis..` sys.modules entry, so the +# verbatim imports inside the vendored files resolve transparently +# (just like the C186g `_jvm_setup_compat` shim does for +# `argumentation_analysis.core.jvm_setup`). +# +# ANTI-REGRESSION (D) +# ------------------- +# 0 `pass`, 0 `return None`, 0 `sorry`, 0 `raise NotImplementedError` +# added. This is a thin glue layer -- it only registers already- +# vendored modules under their upstream Python dotted names. No +# derivation logic. +# +# SCOPE +# ----- +# This shim is the portee 2 of issue #18391 acceptance 2/5 of +# EPIC #4960 (entonnoir taxonomique agentique). It does NOT cover +# the LLM-backed execution (run_guided_analysis requires an LLM +# endpoint and a key) -- it only unblocks `import` so that consumers +# can read class definitions, type hints, and constant tables. + +from __future__ import annotations + +import importlib +import sys +import types + + +_PROXY_INSTALLED: bool = False + + +# Mapping : upstream_dotted_name -> local_vendored_module_name +# Each local module is the byte-for-byte verbatim file we vendored +# at commit ecfd9b9c (tronc EPITA fige). The mapping is deterministic +# (one-line per file) and the SHA of each is recorded in NOTICE-EPITA. +# +# LOAD ORDER matters: leaf modules first (no internal deps), then +# mid-level, then leaves-of-leaves-on-leaves that import them. +# - `taxonomy_local_overrides`, `taxonomy_tree`, `plaintext_destination`, +# `identification_models` are pure utility modules (no internal +# `argumentation_analysis` imports in their bodies beyond the +# `from semantic_kernel...` lines they may share). +# - `taxonomy_navigator` imports from `taxonomy_local_overrides` and +# `taxonomy_tree` (verified at vendor time, body imports at L5-9). +# - `exploration_plugin` imports from `taxonomy_navigator`. +# - `fallacy_workflow_plugin` imports from all of the above plus +# `plaintext_destination` and `identification_models`. +# Hence: leaves first, then mid, then top. +VENDOR_MAP: list[tuple[str, str]] = [ + # Leaves (no internal argumentation_analysis deps beyond stdlib/SK) + ("argumentation_analysis.utils.taxonomy_local_overrides", + "argumentation_lib._taxonomy_local_overrides"), + ("argumentation_analysis.utils.taxonomy_tree", + "argumentation_lib._taxonomy_tree"), + ("argumentation_analysis.core.plaintext_destination", + "argumentation_lib._plaintext_destination"), + ("argumentation_analysis.plugins.identification_models", + "argumentation_lib._identification_models"), + # Mid (depends on leaves) + ("argumentation_analysis.agents.utils.taxonomy_navigator", + "argumentation_lib._taxonomy_navigator"), + ("argumentation_analysis.plugins.exploration_plugin", + "argumentation_lib._exploration_plugin"), + # Top (depends on everything) + ("argumentation_analysis.plugins.fallacy_workflow_plugin", + "argumentation_lib._fallacy_workflow_plugin"), +] + + +def install_epita_vendor_shim() -> bool: + """Register vendored modules under their upstream dotted names. + + Idempotent: returns False (no-op) if already installed in this + process. Best-effort: if a single mapping fails to import, the + other six still install; the failure is logged but not raised. + + Returns: + True if installation took place, False if already installed. + """ + global _PROXY_INSTALLED + if _PROXY_INSTALLED: + return False + + def _ensure_package(name: str) -> types.ModuleType: + mod = sys.modules.get(name) + if mod is None: + mod = types.ModuleType(name) + mod.__path__ = [] # namespace package + sys.modules[name] = mod + return mod + + # Build the parent package chain lazily as we register submodules. + # We need to ensure all of these exist: + # argumentation_analysis + # argumentation_analysis.plugins + # argumentation_analysis.agents + # argumentation_analysis.agents.utils + # argumentation_analysis.utils + # argumentation_analysis.core + for parent in { + "argumentation_analysis", + "argumentation_analysis.plugins", + "argumentation_analysis.agents", + "argumentation_analysis.agents.utils", + "argumentation_analysis.utils", + "argumentation_analysis.core", + }: + _ensure_package(parent) + + for upstream_name, local_name in VENDOR_MAP: + try: + local_mod = importlib.import_module(local_name) + except Exception as exc: # noqa: BLE001 + # Best-effort: a missing optional dep (semantic_kernel, etc.) + # may keep one of the seven from importing. Skip silently; + # the verbatim import will then re-raise the G.9 fingerprint. + sys.stderr.write( + f"[epita_vendor_shim] skip {upstream_name}: {exc!r}\n" + ) + continue + sys.modules[upstream_name] = local_mod + + _PROXY_INSTALLED = True + return True + + +# Module-load-time installation (idempotent, safe to re-import). +try: + install_epita_vendor_shim() +except Exception: # noqa: BLE001 + # Best-effort: if anything blows up at module-load (cycle, missing + # import), the verbatim modules stay unbridged and downstream + # imports re-raise the original G.9 ModuleNotFoundError. + pass + + +__all__ = ["install_epita_vendor_shim", "VENDOR_MAP"] diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_exploration_plugin.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_exploration_plugin.py new file mode 100644 index 0000000000..c2574aaa79 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_exploration_plugin.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/plugins/exploration_plugin.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +"""ExplorationPlugin — constrained kernel functions for taxonomy navigation. + +This plugin provides the ONLY tools available to the 'slave' LLM during +hierarchical fallacy detection. The LLM must navigate the taxonomy tree +by calling these functions — it cannot respond freely. + +Recovered from commit d2fdd930 and enhanced with structured output. +""" + +import json +import logging +import math +from typing import Annotated, Optional, Tuple + +from semantic_kernel.functions.kernel_function_decorator import kernel_function + +from argumentation_analysis.agents.utils.taxonomy_navigator import TaxonomyNavigator + +logger = logging.getLogger(__name__) + +#: Scores des niveaux lexicaux historiques — acceptés, mais le contrat que le +#: prompt feuille énonce est un nombre (#2746). +_CONFIDENCE_LEVELS = {"high": 0.9, "medium": 0.7, "low": 0.4} + +#: Ce qu'une confiance illisible dégrade : le défaut "medium" du paramètre, +#: même valeur que le lecteur aval (fallacy_workflow_plugin lit +#: ``lr.get("confidence", 0.7)``). L'entrée rejetée est nommée dans la note, +#: jamais silencieuse. +_UNREADABLE_CONFIDENCE_SCORE = 0.7 + + +def _confidence_score(raw: object) -> Tuple[float, str]: + """Le contrat unique de ``confirm_fallacy`` : un nombre dans ``[0, 1]`` + (ce que le prompt feuille demande) ou un nom de niveau. + + Rend ``(score, note)``. La note est vide quand l'entrée a été lue + fidèlement ; sinon elle nomme l'entrée rejetée. Jamais d'exception — un + appel de confirmation valide ne doit pas se perdre sur la forme de sa + confiance (#2746). + """ + rejected = ( + f"unreadable confidence {raw!r}, defaulted to " + f"{_UNREADABLE_CONFIDENCE_SCORE}" + ) + if isinstance(raw, bool) or raw is None: + return _UNREADABLE_CONFIDENCE_SCORE, rejected + if isinstance(raw, (int, float)): + value = float(raw) + elif isinstance(raw, str): + text = raw.strip() + level = _CONFIDENCE_LEVELS.get(text.lower()) + if level is not None: + return level, "" + try: + value = float(text.replace(",", ".")) + except ValueError: + return _UNREADABLE_CONFIDENCE_SCORE, rejected + else: + return _UNREADABLE_CONFIDENCE_SCORE, rejected + if math.isnan(value) or math.isinf(value): + return _UNREADABLE_CONFIDENCE_SCORE, rejected + return min(max(value, 0.0), 1.0), "" + + +class ExplorationPlugin: + """Plugin for constrained taxonomy navigation during fallacy detection. + + Provides 3 kernel functions: + - explore_branch: Navigate to a taxonomy node and see its children + - confirm_fallacy: Confirm a specific node as the identified fallacy + - conclude_no_fallacy: Abandon exploration — no fallacy found in this branch + """ + + def __init__(self, taxonomy_navigator: TaxonomyNavigator, language: str = "fr"): + self.taxonomy_navigator = taxonomy_navigator + self.language = language + self._alt_lang = "en" if language == "fr" else "fr" + + @kernel_function( + name="explore_branch", + description=( + "Navigate to a taxonomy node to see its details and children. " + "Use this to explore deeper into a branch that might contain " + "the relevant fallacy. Returns the node info and its children." + ), + ) + def explore_branch( + self, + node_pk: Annotated[str, "The PK (ID) of the taxonomy node to explore"], + ) -> Annotated[str, "JSON with node details and children"]: + """Navigate to a taxonomy node and return its details + children.""" + node = self.taxonomy_navigator.get_node(node_pk) + if not node: + return json.dumps({"error": f"Node {node_pk} not found"}) + + children = self.taxonomy_navigator.get_children(node_pk) + lang = self.language + branch_info = { + "node": { + "pk": node.get("PK", ""), + "path": node.get("path", ""), + "name": ( + node.get(f"text_{lang}", "") + or node.get("nom_vulgarisé", "") + or node.get(f"text_{self._alt_lang}", "") + ), + "description": node.get(f"desc_{lang}", "") + or node.get(f"desc_{self._alt_lang}", ""), + "example": node.get(f"example_{lang}", "") + or node.get(f"example_{self._alt_lang}", ""), + "depth": node.get("depth", ""), + "is_leaf": len(children) == 0, + }, + "children": [ + { + "pk": c.get("PK", ""), + "name": ( + c.get(f"text_{lang}", "") + or c.get("nom_vulgarisé", "") + or c.get(f"text_{self._alt_lang}", "") + ), + "description": c.get(f"desc_{lang}", "") + or c.get(f"desc_{self._alt_lang}", ""), + "example": ( + c.get(f"example_{lang}", "") + or c.get(f"example_{self._alt_lang}", "") + )[:200], + } + for c in children + ], + "children_count": len(children), + } + return json.dumps(branch_info, ensure_ascii=False) + + @kernel_function( + name="confirm_fallacy", + description=( + "Confirm that a specific taxonomy node is the identified fallacy. " + "Call this when you are confident that the current node matches " + "the fallacy in the analyzed text. Provide justification." + ), + ) + def confirm_fallacy( + self, + node_pk: Annotated[str, "The PK of the confirmed fallacy node"], + confidence: Annotated[ + str, + "Confidence: a number between 0.0 and 1.0 (as the leaf prompt " + "specifies), or a level 'high'/'medium'/'low'", + ] = "medium", + justification: Annotated[str, "Why this fallacy matches the text"] = "", + ) -> Annotated[str, "Confirmation result"]: + """Confirm a fallacy identification with justification.""" + node = self.taxonomy_navigator.get_node(node_pk) + if not node: + return json.dumps({"error": f"Node {node_pk} not found"}) + + confidence_score, confidence_note = _confidence_score(confidence) + + lang = self.language + result = { + "confirmed": True, + "pk": node.get("PK", ""), + "path": node.get("path", ""), + "name": ( + node.get(f"text_{lang}", "") + or node.get("nom_vulgarisé", "") + or node.get(f"text_{self._alt_lang}", "") + ), + "name_fr": node.get("text_fr", ""), + "confidence": confidence_score, + "justification": justification, + } + if confidence_note: + result["confidence_note"] = confidence_note + return json.dumps(result, ensure_ascii=False) + + @kernel_function( + name="conclude_no_fallacy", + description=( + "Conclude that no relevant fallacy was found in the current branch. " + "Call this when you have explored the branch sufficiently and determined " + "that none of the nodes match the analyzed text." + ), + ) + def conclude_no_fallacy( + self, + reason: Annotated[str, "Why no fallacy was found in this branch"], + ) -> Annotated[str, "Conclusion recorded"]: + """Conclude that no fallacy was found in this branch.""" + return json.dumps({"confirmed": False, "reason": reason}, ensure_ascii=False) diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_fallacy_workflow_plugin.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_fallacy_workflow_plugin.py new file mode 100644 index 0000000000..e5f6fb8ae9 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_fallacy_workflow_plugin.py @@ -0,0 +1,1788 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/plugins/fallacy_workflow_plugin.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +"""FallacyWorkflowPlugin — hierarchical taxonomy-guided fallacy detection. + +Rebuilt from commit d2fdd930 with improvements: +- Iterative deepening with double-selection pattern (confirm parent vs explore child) +- Parallel branch exploration via asyncio.gather +- One-shot fallback when iterative deepening fails +- Structured output via IdentifiedFallacy Pydantic model +- Full navigation trace for explainability + +See docs/reports/issue_84_git_archaeology.md for architectural history. +""" + +import asyncio +import csv +import json +import logging +import os +import time +from pathlib import Path +from typing import Annotated, Dict, List, Optional, Set, Tuple + +from semantic_kernel.kernel import Kernel +from semantic_kernel.functions.kernel_function_decorator import kernel_function +from semantic_kernel.connectors.ai.open_ai import ( + OpenAIChatCompletion, + OpenAIPromptExecutionSettings, +) +from semantic_kernel.connectors.ai.function_choice_behavior import ( + FunctionChoiceBehavior, +) +from semantic_kernel.connectors.ai.chat_completion_client_base import ( + ChatCompletionClientBase, +) +from semantic_kernel.contents import ( + ChatHistory, + FunctionCallContent, + FunctionResultContent, +) + +from argumentation_analysis.agents.utils.taxonomy_navigator import TaxonomyNavigator +from argumentation_analysis.core.plaintext_destination import ( + PlaintextDestinationError, + check_plaintext_destination, +) +from argumentation_analysis.plugins.exploration_plugin import ExplorationPlugin +from argumentation_analysis.utils.taxonomy_local_overrides import purge_rows +from argumentation_analysis.plugins.identification_models import ( + IdentifiedFallacy, + FallacyAnalysisResult, +) + +logger = logging.getLogger(__name__) + + +class FallacyWorkflowPlugin: + """Plugin orchestrating hierarchical taxonomy-guided fallacy detection. + + Uses a master-slave kernel architecture: + - Master: orchestrates the workflow phases + - Slave: constrained LLM with only ExplorationPlugin tools available + + Detection algorithm: + 1. Present root categories + their children to the LLM + 2. LLM selects candidate branches via explore_branch calls + 3. For each candidate, iterative deepening with double-selection: + - Show parent (confirm) + children (explore deeper) + - LLM calls confirm_fallacy OR explores a child + 4. Parallel exploration of multiple branches + 5. Fallback to one-shot if iterative approach fails + """ + + MAX_BRANCHES = 4 + MAX_CANDIDATES = 20 # Wide-net Phase 1 cap (#578) + MIN_CONFIRM_DEPTH = 2 # Don't accept confirmations at depth < 2 (too generic) + + # FB-30 (#1107): restored agentic taxonomy navigation by subtraction. + # There is NO depth cap — the taxonomy leaves are the only cap (real + # taxonomy depth ≤ 10). The previous MAX_DEPTH_PER_BRANCH=8 made the 12 + # leaf nodes at depth 9-10 structurally unreachable. Runaway protection + # (#708) is preserved by a generous per-branch LLM-CALL budget (circuit + # breaker), NOT a depth cap. One navigation iteration == one LLM call, so + # this bounds total calls while letting the LLM descend freely to leaves. + MAX_NAVIGATION_LLM_CALLS = 18 + # How many subtree levels the navigation prompt shows at each step (a + # multi-level cluster, not just immediate children) so the LLM can jump + # levels via explore_branch(any_pk) — the original summer-2025 design. + NAV_CLUSTER_DEPTH = 2 + + # RA-3 #1048 item 2: bounded recursive sub-branch fan-out. + # At a fork the LLM may flag several promising children; the engine used to + # keep one and silently drop the rest (recall loss — the fork prompt even + # invites multi-child exploration the single-path loop ignored). When + # enabled, the top child stays the primary path AND the next few are + # explored as concurrent recursive sub-branches. The wall-clock bench + # (PR #1067) shows Phase-2-style gather fan-out is ≥2x at realistic + # wide-net widths, so the work overlaps rather than adding latency. + # Bounded by a per-fork width + a shared global budget + # (_BranchSupersessionTracker.try_consume_fanout) and pruned by + # supersession — anti-pendule #1019, never "parallelize everything". + ENABLE_SUBBRANCH_FANOUT = True + SUBBRANCH_FANOUT_WIDTH = 2 # max extra children spawned per fork + SUBBRANCH_FANOUT_BUDGET = 12 # max sub-branches spawned per analysis + + # FB-35 (#1121): global cost-based circuit breaker on the agentic descent. + # MAX_NAVIGATION_LLM_CALLS bounds a SINGLE branch (≤18 calls) and the + # fan-out budget bounds sub-branch SPAWNING (≤12) — but their PRODUCT + # (~20 wide-net candidates × 18 calls × fan-out) is uncapped, which is + # what explodes on a large/entity-dense corpus (the doc_A runaway). This + # bounds the TOTAL descent LLM calls across ALL branches + sub-branches + # of one run_guided_analysis — a cost-based breaker, NOT a depth cap + # (anti-pendule: FB-30 subtracted the depth cap on purpose; this is + # additive only on the call-count dimension, never on depth). Tunable via + # env so a pathological corpus can be bounded without a code change. + DESCENT_TOTAL_CALL_BUDGET = int(os.getenv("FALLACY_DESCENT_CALL_BUDGET", "240")) + + # #2157: wide-net Phase 1 wall-clock budget — the only one on the descent + # chain that was not env-tunable. Default UNCHANGED at 60 s: raising it is + # a measurement decision (issue #2157, piste 3), not a fix. + WIDENET_PHASE1_TIMEOUT = float(os.getenv("FALLACY_WIDENET_PHASE1_TIMEOUT", "60.0")) + + class _BranchSupersessionTracker: + """Track confirmed fallacies for branch supersession during parallel exploration. + + When a branch confirms a fallacy at depth D with confidence C, sibling + branches exploring ancestor nodes at depth < D can abandon early — the + confirmed match is strictly more specific. This saves LLM calls by + cutting exploration of branches that are superseded. + + Thread-safety: all mutation is done from the asyncio event loop, so + no explicit locking is needed (single-threaded concurrency model). + """ + + def __init__( + self, + navigator: TaxonomyNavigator, + fanout_budget: int = 0, + descent_call_budget: int = 0, + ): + self._navigator = navigator + # (pk, taxonomy_path, depth, confidence) + self._confirmed: List[Tuple[str, str, int, float]] = [] + self.superseded_count = 0 + # RA-3 #1048 item 2: shared budget bounding recursive sub-branch + # fan-out across ALL branches of one analysis. Mirrors the agentic + # navigation's MAX_NAVIGATION_LLM_CALLS guardrail — fan-out is + # capped, never unbounded (anti-pendule #1019). 0 ⇒ fan-out + # disabled. + self._fanout_budget = fanout_budget + self.fanout_spawned = 0 + # FB-35 (#1121): shared global budget bounding the TOTAL number of + # descent LLM calls across ALL branches + sub-branches of one + # analysis. The per-branch MAX_NAVIGATION_LLM_CALLS bounds one + # branch; the fan-out budget bounds spawning; THIS bounds their + # product (the dimension that explodes on large corpora). One + # descent iteration == one LLM call, so this counts cost, not + # depth. 0 ⇒ disabled (no global cap, legacy behavior). When + # exhausted the descent stops and run_guided_analysis surfaces a + # fail-loud flag (anti-theater #1019: partial results + honest + # marker, never a silent empty/truncated outcome). + self._descent_call_budget = descent_call_budget + self._descent_call_enabled = descent_call_budget > 0 + self._descent_calls_remaining = descent_call_budget + self.descent_calls_made = 0 + self.descent_budget_exceeded = False + + def register(self, pk: str, depth: int, confidence: float) -> None: + """Register a confirmed fallacy.""" + node = self._navigator.get_node(pk) + path = node.get("path", "") if node else "" + self._confirmed.append((pk, path, depth, confidence)) + + def check_superseded( + self, current_pk: str, current_depth: int + ) -> Optional[str]: + """Return the superseding PK if *current_pk* is an ancestor of a + confirmed node that is strictly deeper. ``None`` otherwise. + """ + current_node = self._navigator.get_node(current_pk) + if not current_node: + return None + current_path = current_node.get("path", "") + if not current_path: + return None + + for conf_pk, conf_path, conf_depth, conf_conf in self._confirmed: + # Only supersede when confirmed node is strictly deeper + if conf_depth <= current_depth: + continue + # Same lineage: confirmed path descends from current path + if conf_path.startswith(current_path + "."): + return conf_pk + return None + + @property + def confirmation_count(self) -> int: + return len(self._confirmed) + + def try_consume_fanout(self) -> bool: + """Consume one unit of the shared sub-branch fan-out budget. + + Returns True while budget remains (a sub-branch may be spawned), + False once exhausted. Bounds total recursive fan-out across all + branches of one analysis so concurrent token spend stays capped + (anti-pendule #1019: bounded fan-out, never "explore everything"). + """ + if self._fanout_budget <= 0: + return False + self._fanout_budget -= 1 + self.fanout_spawned += 1 + return True + + def try_consume_descent_call(self) -> bool: + """Consume one unit of the global descent LLM-call budget (FB-35 #1121). + + Returns True while budget remains (the calling branch may make its + LLM call and continue descending), False once exhausted. Bounds the + TOTAL descent LLM calls across all branches + recursive sub-branches + of one analysis — the per-branch MAX_NAVIGATION_LLM_CALLS and the + fan-out spawn budget each bound a single dimension, but their + product (~candidates × calls × fan-out) is what explodes on a + large/entity-dense corpus. This is a COST-based breaker (one + iteration == one LLM call), never a depth cap. When it returns + False the caller stops descending and the run surfaces a fail-loud + flag. 0 ⇒ disabled (legacy unbounded-within-branch behavior). + + Thread-safety: single-threaded asyncio event loop, no locking + needed (same model as try_consume_fanout above). + """ + if not self._descent_call_enabled: + return True # no global cap configured (legacy behavior) + if self._descent_calls_remaining <= 0: + self.descent_budget_exceeded = True + return False + self._descent_calls_remaining -= 1 + self.descent_calls_made += 1 + return True + + def __init__( + self, + master_kernel: Kernel, + llm_service: ChatCompletionClientBase, + taxonomy_file_path: Optional[str] = None, + taxonomy_data: Optional[list] = None, + logger: Optional[logging.Logger] = None, + ): + self.master_kernel = master_kernel + self.llm_service = llm_service + self.logger = logger or logging.getLogger(__name__) + self.language = "fr" + + # #2324: marche déterministe de la descente. En mode parallèle + # (défaut, production inchangée), l'entrelacement des branches dans le + # ``asyncio.gather`` du Phase 2 — et donc la supersession croisée et + # l'ordre de consommation du budget — dépend des latences de réponse : + # la forme de la marche est choisie par le LLM *et par le scheduler* à + # chaque session d'enregistrement. Une seule session ne peut alors pas + # produire des cassettes couvrant chaque marche possible. En mode + # séquentiel, la marche devient une fonction pure des réponses + # enregistrées : candidats dans l'ordre du wide-net, une branche après + # l'autre — le rejeu suit exactement la marche enregistrée. La descente + # a toujours LIEU (nœuds visités, LLM consulté) ; seule la + # séquentialisation change. + self.descent_sequential = os.getenv("FALLACY_DESCENT_SEQUENTIAL", "0") == "1" + + data = taxonomy_data or [] + load_error: Optional[str] = None + if not data and taxonomy_file_path: + try: + with open(taxonomy_file_path, mode="r", encoding="utf-8") as infile: + reader = csv.DictReader(infile) + data = purge_rows(list(reader)) + except FileNotFoundError: + load_error = f"file not found at {taxonomy_file_path}" + except Exception as e: + load_error = f"read error: {e}" + + self.taxonomy_navigator = TaxonomyNavigator(taxonomy_data=data) + self.exploration_plugin = ExplorationPlugin( + self.taxonomy_navigator, language=self.language + ) + + # #2141: three degradation states, each said in its own words. The + # previous single message ("loading failed **or** has no root nodes") + # merged "tried and failed" with "never tried": a navigator that was + # never handed a source was reported as a failed load, so nothing + # distinguished a bypassed funnel from a broken one — and every neural + # run bypassed it silently. self.taxonomy_state is the readable form. + roots = self.taxonomy_navigator.get_root_nodes() + if roots: + self.taxonomy_state = "loaded" + self.logger.info( + f"Taxonomy loaded: {len(data)} nodes, {len(roots)} root categories" + ) + elif load_error is not None: + self.taxonomy_state = "load_failed" + self.logger.error( + f"Taxonomy source provided but the load failed ({load_error}) — " + "the navigator is empty and the wide-net funnel will resolve no candidate." + ) + elif not data and taxonomy_file_path is None and not taxonomy_data: + self.taxonomy_state = "none" + self.logger.warning( + "Taxonomy: no source provided (neither taxonomy_file_path nor " + "taxonomy_data) — no load was attempted, the navigator is empty by " + "construction, and the wide-net funnel will fall back to one-shot." + ) + else: + self.taxonomy_state = "empty_roots" + self.logger.warning( + f"Taxonomy loaded {len(data)} rows but no root node (depth==1) — " + "the navigator has no category and the wide-net funnel will resolve no candidate." + ) + + def _create_slave_kernel(self) -> Tuple[Kernel, OpenAIPromptExecutionSettings]: + """Create a constrained kernel with only ExplorationPlugin available. + + The slave LLM can ONLY call explore_branch, confirm_fallacy, or + conclude_no_fallacy. It cannot respond with free text. + """ + slave_kernel = Kernel() + slave_kernel.add_service(self.llm_service) + slave_kernel.add_plugin(self.exploration_plugin, "Exploration") + + slave_settings = OpenAIPromptExecutionSettings( + function_choice_behavior=FunctionChoiceBehavior.Auto(auto_invoke=False) + ) + return slave_kernel, slave_settings + + def _mark_fallback(self, result_json: str) -> str: + """#2141: make a bypassed funnel readable to a consumer. + + A one-shot result is reached by three routes — an explicit + ``use_one_shot`` request, a wide-net that failed to parse, and a + wide-net that ran but could resolve no candidate. Only the last means + the master/slave funnel was bypassed *because its navigator had + nothing to resolve against*. The marker names that case; when the + navigator loaded (``taxonomy_state == "loaded"``) the one-shot is a + genuine choice and no marker is added. + """ + if self.taxonomy_state == "loaded": + return result_json + try: + parsed = json.loads(result_json) + except (json.JSONDecodeError, TypeError): + return result_json + if not isinstance(parsed, dict) or "fallacies" not in parsed: + return result_json + parsed["fallback_reason"] = "empty_taxonomy_navigator" + parsed["taxonomy_state"] = self.taxonomy_state + return json.dumps(parsed, indent=2, ensure_ascii=False) + + def _mark_regime(self, result_json: str, regime: str) -> str: + """#2141 temps 2: name the regime the result was actually produced under. + + ``one_shot`` when a single full-taxonomy call produced it — whether that + was requested outright or reached by a funnel that resolved nothing. + ``funnel`` when the wide-net descent ran and produced it. A consumer can + read which regime it got without re-deriving the call path, which is what + makes the two regimes comparable instead of merely co-existing. + """ + try: + parsed = json.loads(result_json) + except (json.JSONDecodeError, TypeError): + return result_json + if not isinstance(parsed, dict) or "fallacies" not in parsed: + return result_json + parsed["analysis_regime"] = regime + return json.dumps(parsed, indent=2, ensure_ascii=False) + + def _create_one_shot_kernel(self) -> Tuple[Kernel, OpenAIPromptExecutionSettings]: + """Create a kernel for one-shot fallback analysis.""" + kernel = Kernel() + kernel.add_service(self.llm_service) + settings = OpenAIPromptExecutionSettings( + function_choice_behavior=FunctionChoiceBehavior.NoneInvoke() + ) + return kernel, settings + + def _build_root_presentation(self) -> str: + """Build a formatted presentation of root categories + their children.""" + roots = self.taxonomy_navigator.get_root_nodes() + lines = [] + for root in roots: + pk = root.get("PK", "") + name = root.get(f"text_{self.language}", root.get("nom_vulgarisé", pk)) + desc = root.get(f"desc_{self.language}", "") + lines.append(f"\n## {name} (ID: {pk})") + if desc: + lines.append(f" {desc}") + + children = self.taxonomy_navigator.get_children(pk) + for child in children: + cpk = child.get("PK", "") + cname = child.get( + f"text_{self.language}", child.get("nom_vulgarisé", cpk) + ) + cdesc = child.get(f"desc_{self.language}", "")[:120] + lines.append(f" - {cname} (ID: {cpk}): {cdesc}") + + return "\n".join(lines) + + def _build_roots_index(self) -> Dict[str, str]: + """Build a lookup: lowercase root name → root PK.""" + roots = self.taxonomy_navigator.get_root_nodes() + index = {} + for root in roots: + pk = root.get("PK", "") + name = root.get(f"text_{self.language}", root.get("nom_vulgarisé", "")) + index[name.lower().strip()] = pk + return index + + def _build_deep_index(self, max_depth: int = 3) -> Dict[str, str]: + """Build a lookup: lowercase name → PK for nodes up to max_depth. + + FB-19 (#1040): gives wide-net Phase 1 a head start by mapping to + depth-2/3 nodes instead of only depth-1 roots. This shaves 2-3 + levels off the descent and makes reaching depth-5+ nodes feasible. + """ + index: Dict[str, str] = {} + for node in self.taxonomy_navigator.taxonomy_data: + try: + depth = int(node.get("depth", 0)) + except (ValueError, TypeError): + continue + if depth < 1 or depth > max_depth: + continue + pk = node.get("PK", "") + if not pk: + continue + name = node.get(f"text_{self.language}", node.get("nom_vulgarisé", "")) + if name: + index[name.lower().strip()] = pk + # Also index by family / subfamily keywords for broader matching + for field in ("Famille", "Sous-Famille", "Soussousfamille"): + val = node.get(field, "") + if val and val.lower().strip() not in index: + index[val.lower().strip()] = pk + return index + + # German → English keyword bridge for multilingual support (#600) + _DE_TO_EN_KEYWORDS: Dict[str, str] = { + "strohmann": "straw man", + "strohmann-argument": "straw man", + "dammbruch": "slippery slope", + "rückführung": "slippery slope", + "gefühl": "emotion", + "emotionale appellation": "emotion", + "entweder-oder": "false dilemma", + "schwarz-weiß": "false dilemma", + "vorschnelle verallgemeinerung": "hasty generalization", + "übergeneralisierung": "hasty generalization", + "zirkelschluss": "circular reasoning", + "zirkulär": "circular reasoning", + "roter hering": "red herring", + "nebenthema": "red herring", + "ablenkung": "red herring", + "heuchlerisch": "tu quoque", + "schuld durch assoziation": "guilt by association", + "mitläufereffekt": "bandwagon", + "mehrheitsargument": "bandwagon", + "traditionsargument": "tradition", + "brunnenvergiftung": "poisoning the well", + "wahrer schotte": "no true scotsman", + "fehlschluss der ursache": "false cause", + "suggestivfrage": "loaded question", + "fangfrage": "loaded question", + "angst": "appeal to fear", + "mitleid": "appeal to pity", + "rosinenpickerei": "cherry picking", + "personenangriff": "ad hominem", + "autorität": "authority", + "berufung auf": "authority", + "unzureichend": "insufficiency", + "beeinflussung": "influence", + "betrug": "cheating", + "täuschung": "cheating", + "behinderung": "obstruction", + "blockade": "obstruction", + "obstruktion": "obstruction", + } + + def _map_fallacy_to_root_pk( + self, fallacy_name: str, roots_index: Dict[str, str] + ) -> Optional[str]: + """Map a free-text fallacy name to the closest root PK via keyword matching. + + Supports FR/EN/DE keywords via a three-pass strategy: + 1. Direct lookup in roots_index + 2. German → English keyword bridge, then roots_index + 3. Pattern-based fuzzy matching (FR/EN/DE patterns) + """ + name_lower = fallacy_name.lower().strip() + if name_lower in roots_index: + return roots_index[name_lower] + + # Pass 2: German bridge — translate DE keyword to EN, then lookup + for de_keyword, en_keyword in self._DE_TO_EN_KEYWORDS.items(): + if de_keyword in name_lower and en_keyword in roots_index: + return roots_index[en_keyword] + + root_keywords = { + "authority": ["autorit", "autorität", "berufung auf"], + "ad hominem": [ + "ad hominem", + "attaque", + "personne", + "personenangriff", + "ad-personam", + ], + "straw man": ["paille", "distortion", "strohmann", "strohmann-argument"], + "slippery slope": ["pente glissante", "dammbruch", "rückführung"], + "emotion": ["émotion", "sentiment", "gefühl", "emotionale appellation"], + "false dilemma": [ + "faux dilemme", + "dilemme", + "entweder-oder", + "schwarz-weiß", + ], + "hasty generalization": [ + "généralisation", + "vorschnelle verallgemeinerung", + "übergeneralisierung", + ], + "circular reasoning": [ + "circulaire", + "pétition", + "zirkelschluss", + "zirkulär", + ], + "red herring": ["hareng", "diversion", "roter hering", "nebenthema"], + "tu quoque": ["tu quoque", "hypocrisie", "heuchlerisch"], + "guilt by association": [ + "association", + "culpabilité", + "schuld durch assoziation", + ], + "bandwagon": [ + "ad populum", + "majeure", + "mitläufereffekt", + "mehrheitsargument", + ], + "tradition": ["tradition", "traditionsargument"], + "non sequitur": ["non sequitur", "inconséquence", "nicht zwingend"], + "poisoning the well": ["empoisonnement", "brunnenvergiftung"], + "no true scotsman": ["vrai écossais", "wahrer schotte", "kein wahrer"], + "false cause": [ + "fausse cause", + "causalité", + "fehlschluss der ursache", + "post hoc", + ], + "loaded question": ["question chargée", "suggestivfrage", "fangfrage"], + "manipulation": ["manipulation", "beeinflussung"], + "appeal to fear": ["angst", "appell an die angst"], + "appeal to pity": ["mitleid", "mitleidsappell"], + "begging the question": ["implikation", "voraussetzung"], + "cherry picking": ["rosinenpickerei", "auswahl"], + "deflection": ["ablenkung", "themenwechsel"], + "insufficiency": ["unzureichend", "mangelhaft", "insufficient"], + "influence": ["beeinflussung", "manipulation"], + "cheating": ["betrug", "trick", "täuschung"], + "obstruction": ["behinderung", "blockade", "obstruktion"], + } + + for keyword, patterns in root_keywords.items(): + if keyword in name_lower or any(p in name_lower for p in patterns): + for root_name, pk in roots_index.items(): + if any(p in root_name for p in patterns): + return pk + + for root_name, pk in roots_index.items(): + if root_name in name_lower or name_lower in root_name: + return pk + + return None + + def _map_fallacy_to_deep_pk( + self, fallacy_name: str, deep_index: Dict[str, str] + ) -> Optional[str]: + """Map a free-text fallacy name to the deepest matching PK. + + Tries the deep index (depth ≤ 3) first, giving the agentic navigation + a head start. Falls back to None if no match. + """ + name_lower = fallacy_name.lower().strip() + + # Direct match in deep index + if name_lower in deep_index: + return deep_index[name_lower] + + # Substring match: check if any deep index key is contained in the name + for key, pk in deep_index.items(): + if key in name_lower or name_lower in key: + return pk + + return None + + async def _wide_net_candidates(self, argument_text: str) -> List[str]: + """Phase 1 wide-net: LLM lists all candidate fallacies 0-shot-style. + + Returns list of PKs (depth 1-3 via deep index), up to MAX_CANDIDATES. + + Tries deep index first (depth 2-3), falls back to root PKs. This gives + the agentic navigation a 2-3 level head start; from there the LLM + descends freely (multi-level cluster, no depth cap — FB-30 #1107). + """ + kernel, settings = self._create_one_shot_kernel() + roots = self.taxonomy_navigator.get_root_nodes() + roots_text = "\n".join( + f"- {r.get('text_fr', r.get('nom_vulgarisé', r.get('PK', '')))} (PK: {r.get('PK', '')})" + for r in roots + ) + + prompt = ( + f"Analyze this text exhaustively for logical fallacies:\n\n" + f"--- TEXT ---\n{argument_text[:8000]}\n--- END ---\n\n" + "List EVERY fallacy you can find. For each, respond with a JSON object:\n" + '{"fallacy_name": "...", "root_category": "...", "confidence": 0.0-1.0}\n\n' + "Respond with a JSON array of objects. Be thorough — aim for 10-20 fallacies.\n" + "Even borderline or arguable cases should be included.\n\n" + f"Available root categories:\n{roots_text}" + ) + + history = ChatHistory( + system_message="You are a fallacy detection expert. Be exhaustive. Respond ONLY with a JSON array." + ) + history.add_user_message(prompt) + + try: + response = await asyncio.wait_for( + self.llm_service.get_chat_message_content( + chat_history=history, settings=settings, kernel=kernel + ), + timeout=self.WIDENET_PHASE1_TIMEOUT, + ) + raw = str(response).strip() + except asyncio.TimeoutError: + # str(TimeoutError()) == "" — a catch-all here logged an empty + # message, so a budget overrun silently degraded the funnel to + # one-shot (#2157). Same shape as the two named handlers below. + self.logger.warning( + f"Wide-net Phase 1 timed out after {self.WIDENET_PHASE1_TIMEOUT}s " + f"(FALLACY_WIDENET_PHASE1_TIMEOUT), falling back to one-shot" + ) + return [] + except Exception as e: + self.logger.warning(f"Wide-net Phase 1 failed: {type(e).__name__}: {e}") + return [] + + candidates_raw = [] + try: + if "```json" in raw: + raw = raw.split("```json")[1].split("```")[0] + elif "```" in raw: + raw = raw.split("```")[1].split("```")[0] + start = raw.find("[") + end = raw.rfind("]") + 1 + if start >= 0 and end > start: + candidates_raw = json.loads(raw[start:end]) + elif raw.strip().startswith("{"): + candidates_raw = [json.loads(raw)] + except (json.JSONDecodeError, ValueError): + self.logger.warning(f"Wide-net parse failed, raw: {raw[:200]}") + return [] + + if not isinstance(candidates_raw, list): + candidates_raw = [candidates_raw] + + # FB-19: try deep index first (depth 2-3), then fall back to root PKs + deep_index = self._build_deep_index(max_depth=3) + roots_index = self._build_roots_index() + candidate_pks = [] + for c in candidates_raw: + if not isinstance(c, dict): + continue + root_cat = c.get("root_category", "") + fallacy_name = c.get("fallacy_name", "") + + # Try deep index first (FB-19) + pk = self._map_fallacy_to_deep_pk(fallacy_name, deep_index) + if not pk: + pk = self._map_fallacy_to_deep_pk(root_cat, deep_index) + # Fall back to root mapping + if not pk: + pk = self._map_fallacy_to_root_pk(root_cat, roots_index) + if not pk: + pk = self._map_fallacy_to_root_pk(fallacy_name, roots_index) + if pk and pk not in candidate_pks: + candidate_pks.append(pk) + + candidate_pks = candidate_pks[: self.MAX_CANDIDATES] + self.logger.info( + f"Phase 1 wide-net (FB-19 deep): {len(candidates_raw)} candidates, " + f"{len(candidate_pks)} unique PKs: {candidate_pks}" + ) + return candidate_pks + + async def _execute_tool_calls( + self, + response_items: list, + slave_kernel: Kernel, + history: ChatHistory, + ) -> List[Dict]: + """Execute function calls from the LLM response and feed results back.""" + results = [] + for item in response_items: + if not isinstance(item, FunctionCallContent): + continue + + func_name = item.name or "" + # Handle both "Exploration-explore_branch" and "explore_branch" formats + short_name = func_name.split("-")[-1] if "-" in func_name else func_name + arguments = item.arguments or {} + + # Parse arguments if they're a string + if isinstance(arguments, str): + try: + arguments = json.loads(arguments) + except (json.JSONDecodeError, TypeError): + arguments = {} + + try: + plugin = slave_kernel.plugins.get("Exploration") + if plugin and short_name in plugin: + func = plugin[short_name] + # Call the underlying Python method directly to avoid + # SK 1.40's KernelFunction.__call__ requiring kernel arg + method = getattr(self.exploration_plugin, short_name, None) + if method is not None: + result_str = str(method(**arguments)) + else: + result_str = str(func(slave_kernel, **arguments)) + else: + result_str = json.dumps({"error": f"Unknown function: {func_name}"}) + except Exception as e: + self.logger.warning(f"Tool call {func_name} failed: {e}") + result_str = json.dumps({"error": str(e)}) + + # Feed result back to history + history.add_message( + FunctionResultContent( + id=item.id, result=result_str + ).to_chat_message_content() + ) + + # Parse result for caller + try: + parsed = json.loads(result_str) + except (json.JSONDecodeError, TypeError): + parsed = {"raw": result_str} + parsed["function_name"] = short_name + results.append(parsed) + + return results + + def _render_subtree_cluster(self, root_pk: str, max_levels: int = 2) -> str: + """Render a truncated multi-level subtree under ``root_pk``. + + FB-30 (#1107): the navigation prompt shows the LLM a cluster spanning + ``max_levels`` below the current node (not just immediate children), + so the LLM can jump levels by calling ``explore_branch`` on a deeper + PK directly. Each line carries name + PK + short description so the + LLM has enough signal to pick any node. + + Args: + root_pk: the PK whose subtree to render (rendered as the root line). + max_levels: how many levels of descendants to include below the + root (1 = children only, 2 = children + grandchildren). + + Returns: + A newline-separated, indented cluster string. Empty string if the + root has no descendants. + """ + lines: List[str] = [] + + def walk(pk: str, level: int) -> None: + if level > max_levels: + return + node = self.taxonomy_navigator.get_node(pk) + if not node: + return + indent = " " * level + name = node.get(f"text_{self.language}", node.get("nom_vulgarisé", pk)) + desc = node.get(f"desc_{self.language}", "") + # Keep the description short so the whole cluster fits the prompt. + desc_short = desc[:120] + ("…" if len(desc) > 120 else "") + depth = node.get("depth", "?") + lines.append(f"{indent}- {name} (ID: {pk}, depth {depth}): {desc_short}") + for child in self.taxonomy_navigator.get_children(pk): + walk(child.get("PK", ""), level + 1) + + walk(root_pk, 0) + return "\n".join(lines) + + async def _explore_single_branch( + self, + argument_text: str, + start_pk: str, + slave_kernel: Kernel, + slave_settings: OpenAIPromptExecutionSettings, + reasoning_history: Optional[List[str]] = None, + supersession_tracker: Optional["_BranchSupersessionTracker"] = None, + results_sink: Optional[List[IdentifiedFallacy]] = None, + ) -> Optional[IdentifiedFallacy]: + """Explore a single taxonomy branch using iterative deepening. + + At each level, the LLM sees: + - The PARENT node (marked "CONFIRM this if it matches") + - All CHILD nodes (marked "EXPLORE deeper") + + Memory of reasons: The LLM's justification at each level is captured + and passed to the next level's prompt for context continuity. + + Double intention: After exploring to a node, the LLM sees BOTH the + parent (to confirm/stop) AND children (to explore deeper). + + The LLM calls confirm_fallacy to stop, explore_branch to go deeper, + or conclude_no_fallacy to abandon the branch. + + Args: + supersession_tracker: Optional shared tracker. When a parallel + branch confirms a deeper, more specific fallacy, this branch + can abandon early if it is exploring an ancestor node. + """ + current_pk = start_pk + navigation_trace = [start_pk] + reasoning_history = reasoning_history or [] + + # FB-30 (#1107): no depth cap — the LLM descends until it reaches a + # leaf (the only natural cap, depth ≤ 10). Bounded only by a generous + # per-branch LLM-call budget (circuit breaker, preserves #708). One + # iteration == one LLM call (either the leaf-confirmation call or the + # navigation call), so this counts calls, not depth. + llm_call_count = 0 + while llm_call_count < self.MAX_NAVIGATION_LLM_CALLS: + # FB-35 (#1121): global cost-based circuit breaker. Consume one + # unit of the shared descent-call budget BEFORE making this + # iteration's LLM call. When the global budget (across all + # concurrent branches + sub-branches) is exhausted, stop + # descending — run_guided_analysis surfaces a fail-loud flag. This + # is a COST cap (one iteration == one call), NOT a depth cap, so + # the LLM still descends freely to leaves within the budget. + if supersession_tracker is not None and not ( + supersession_tracker.try_consume_descent_call() + ): + self.logger.warning( + " Global descent call budget exceeded at branch %s " + "(call #%d) — stopping descent (fail-loud, FB-35 #1121)", + start_pk, + supersession_tracker.descent_calls_made, + ) + break + llm_call_count += 1 + # --- Supersession check (RA-3 #1048) --- + if supersession_tracker is not None: + current_node = self.taxonomy_navigator.get_node(current_pk) + current_depth = int(current_node.get("depth", 0)) if current_node else 0 + superseding_pk = supersession_tracker.check_superseded( + current_pk, current_depth + ) + if superseding_pk: + supersession_tracker.superseded_count += 1 + self.logger.info( + f" Branch SUPERSeded at {current_pk} (depth {current_depth}) " + f"by confirmed {superseding_pk}" + ) + return None + current_node = self.taxonomy_navigator.get_node(current_pk) + if not current_node: + break + + children = self.taxonomy_navigator.get_children(current_pk) + + # Leaf node — ask LLM for confirmation instead of auto-confirm (#471) + if not children: + self.logger.info( + f" Leaf node reached: {current_pk} " + f"({current_node.get(f'text_{self.language}', '')})" + ) + leaf_name = current_node.get( + f"text_{self.language}", + current_node.get("nom_vulgarisé", current_pk), + ) + leaf_desc = current_node.get(f"desc_{self.language}", "") + leaf_example = current_node.get(f"example_{self.language}", "") + + reasoning_context = "" + if reasoning_history: + reasoning_context = ( + "\n--- PREVIOUS REASONING ---\n" + + "\n".join( + f"{i+1}. {r}" for i, r in enumerate(reasoning_history[-3:]) + ) + + "\n" + ) + + leaf_prompt = ( + f'Text to analyze: "{argument_text[:500]}"\n\n' + f"You reached a LEAF node in the fallacy taxonomy.\n" + f"Node: {leaf_name} (PK: {current_pk})\n" + f"Description: {leaf_desc}\n" + f"{'Example: ' + leaf_example[:200] if leaf_example else ''}\n" + f"{reasoning_context}\n" + "Does this leaf node correctly identify a fallacy in the text?\n" + "IMPORTANT: Only confirm if the reasoning in the text is genuinely fallacious.\n" + "Legitimate uses of authority, emotion, or tradition are NOT fallacies.\n\n" + "Choose ONE action:\n" + f"- confirm_fallacy(node_pk='{current_pk}', justification='...', confidence=0.0-1.0) " + "if this matches\n" + "- conclude_no_fallacy(reason='...') if no match" + ) + + leaf_history = ChatHistory( + system_message=( + "You are a fallacy classifier. You MUST call exactly one function. " + "Do NOT respond with text.\n" + "When evaluating circular reasoning: accept paraphrased/implicit forms " + "where the premise presupposes the conclusion.\n" + "When evaluating appeal to emotion: confirm only if emotion is the " + "primary persuasive operator, not merely illustrative." + ) + ) + leaf_history.add_user_message(leaf_prompt) + + try: + leaf_response = await asyncio.wait_for( + self.llm_service.get_chat_message_contents( + chat_history=leaf_history, + settings=slave_settings, + kernel=slave_kernel, + ), + timeout=30.0, + ) + except asyncio.TimeoutError: + self.logger.warning( + f" Leaf confirmation timed out for {current_pk}" + ) + break + except Exception as e: + self.logger.warning(f" Leaf confirmation LLM call failed: {e}") + break + + leaf_items = [] + if leaf_response: + for msg in leaf_response: + if hasattr(msg, "items"): + leaf_items.extend(msg.items) + + leaf_calls = [ + i for i in leaf_items if isinstance(i, FunctionCallContent) + ] + if not leaf_calls: + self.logger.info(" No function call at leaf — branch abandoned") + break + + leaf_results = await self._execute_tool_calls( + leaf_calls, slave_kernel, leaf_history + ) + + for lr in leaf_results: + if lr.get("function_name") == "confirm_fallacy" and lr.get( + "confirmed" + ): + leaf_depth = int(current_node.get("depth", 0)) + # Register leaf confirmation for supersession (RA-3 #1048) + if supersession_tracker is not None: + supersession_tracker.register( + current_pk, + leaf_depth, + lr.get("confidence", 0.7), + ) + return IdentifiedFallacy( + fallacy_type=lr.get("name", lr.get("name_fr", leaf_name)), + taxonomy_pk=current_pk, + taxonomy_path=current_node.get("path", ""), + explanation=lr.get("justification", ""), + confidence=lr.get("confidence", 0.7), + navigation_trace=navigation_trace, + family=current_node.get("Famille", ""), + depth=leaf_depth, + ) + elif lr.get("function_name") == "conclude_no_fallacy": + self.logger.info( + f" Leaf not confirmed: {lr.get('reason', '')}" + ) + return None + + break + + # Build double-selection prompt with memory of reasons + parent_name = current_node.get( + f"text_{self.language}", current_node.get("nom_vulgarisé", current_pk) + ) + parent_desc = current_node.get(f"desc_{self.language}", "") + parent_example = current_node.get(f"example_{self.language}", "") + + # Memory of reasons: include previous reasoning + reasoning_context = "" + if reasoning_history: + reasoning_context = ( + "\n--- PREVIOUS REASONING (for context) ---\n" + + "\n".join( + f"{i+1}. {r}" for i, r in enumerate(reasoning_history[-3:]) + ) + + "\n" + ) + + options_text = ( + f"\n--- OPTIONS at depth {current_node.get('depth', '?')} ---\n" + ) + # Double intention: Parent can be re-selected to confirm + options_text += ( + f"[CONFIRM] {parent_name} (ID: {current_pk})\n" + f" Description: {parent_desc}\n" + ) + if parent_example: + options_text += f" Example: {parent_example[:200]}\n" + options_text += ( + " → Select this node AGAIN (confirm_fallacy with pk='{current_pk}') " + "if this level matches and you want to STOP here.\n" + ) + + # FB-30 (#1107): multi-level cluster, not just immediate children. + # The LLM sees a truncated subtree NAV_CLUSTER_DEPTH levels deep, + # so it can jump levels by calling explore_branch on a grandchild + # PK directly (the original summer-2025 agentic design), rather + # than descending one level per iteration. + options_text += ( + f"\n[EXPLORE DEEPER] — subtree under {parent_name} " + f"(showing {self.NAV_CLUSTER_DEPTH + 1} levels; " + "you may explore ANY listed node, not only the immediate children):\n" + ) + options_text += self._render_subtree_cluster( + current_pk, max_levels=self.NAV_CLUSTER_DEPTH + ) + + prompt = ( + f'Text to analyze: "{argument_text[:500]}"\n\n' + f"You are navigating the fallacy taxonomy. Current position: {parent_name}\n" + f"{reasoning_context}" + f"{options_text}\n" + "Choose ONE action:\n" + "- Call explore_branch(node_pk='') to explore a node — " + "you may jump several levels deep by picking a grandchild/great-grandchild PK directly\n" + f"- Call confirm_fallacy(node_pk='{current_pk}', ...) to confirm THIS level and stop\n" + "- Call conclude_no_fallacy(reason='...') if no match in this branch\n" + "You MUST call exactly one function." + ) + + history = ChatHistory( + system_message=( + "You are a fallacy classifier navigating a taxonomy tree. " + "Your goal is to find the MOST SPECIFIC (deepest) matching fallacy. " + "You MUST call one of the available functions. " + "Do NOT respond with text — only function calls.\n\n" + "CRITICAL MULTI-BRANCH INSTRUCTION:\n" + "1. When selecting among children, PREFER exploring MULTIPLE children (call explore_branch " + " multiple times) rather than confirming immediately at the current level.\n" + "2. Confirm at the current level when EITHER: (a) you are at a LEAF node, OR " + " (b) NO child matches even partially, OR (c) every available child is a NARROWER " + " SPECIALIZATION than the fallacy the text actually exhibits. For (c): if the " + " current node correctly names the fallacy but its children each describe a " + " more specific sub-case the argument does NOT instantiate (e.g. the current " + " node is 'circular reasoning' but the only child is a specific named variant " + " like a Cartesian circle that the text is not), confirm the current node rather " + " than descending to a leaf that will not match. Descending to a non-matching " + " leaf only to then conclude_no_fallacy loses a correct classification.\n" + "3. Do NOT call conclude_no_fallacy prematurely. The text likely contains " + " fallacies in multiple branches. Abandon a branch only if you are CERTAIN " + " there is no match at ANY descendant.\n\n" + "IMPORTANT: Only confirm a fallacy if the reasoning in the text is " + "genuinely fallacious. Legitimate uses of authority, emotion, " + "tradition are NOT fallacies.\n\n" + "DETECTION GUIDANCE — INDIRECT FORMS:\n" + "Circular reasoning (petitio principii): The conclusion may be " + "re-injected as a premise through PARAPHRASE, not literal repetition. " + "Watch for: restating the claim in different words as if it were new " + "evidence; using a synonym or reworded assertion as justification. " + "This includes implicit circularity where the premise presupposes " + "the conclusion without stating it outright.\n" + "Appeal to emotion: Distinguish between (a) an argument where emotion " + "is the PRIMARY OPERATOR — fear, indignation, pride, pity, loyalty " + "DRIVE the persuasion instead of reasons — and (b) an argument that " + "merely ILLUSTRATES with emotional language but provides substantive " + "reasoning. Only (a) is fallacious. Look for: arguments that would " + "collapse if the emotional charge were removed; moral-duty framing " + "that replaces logical justification; urgency appeals that bypass evidence." + ) + ) + history.add_user_message(prompt) + + try: + response = await asyncio.wait_for( + self.llm_service.get_chat_message_contents( + chat_history=history, + settings=slave_settings, + kernel=slave_kernel, + ), + timeout=30.0, + ) + except asyncio.TimeoutError: + self.logger.warning( + f" LLM call timed out during branch exploration at {current_pk}" + ) + break + except Exception as e: + self.logger.warning(f" LLM call failed during branch exploration: {e}") + break + + # Extract function calls from response + all_items = [] + if response: + for msg in response: + if hasattr(msg, "items"): + all_items.extend(msg.items) + + tool_calls = [i for i in all_items if isinstance(i, FunctionCallContent)] + if not tool_calls: + self.logger.info(" No function calls in response — branch exhausted") + break + + # Execute tool calls + tool_results = await self._execute_tool_calls( + tool_calls, slave_kernel, history + ) + + # Classify the LLM's tool calls. Original behaviour: the first + # confirm/conclude decides the branch; explore_branch advances the + # path. RA-3 #1048 item 2: capture *all* explore targets so extra + # promising children can be fanned out as bounded concurrent + # sub-branches instead of being silently dropped. + confirm_result = None + conclude_result = None + explore_targets: List[Tuple[str, dict]] = [] # (next_pk, node_info) + for result in tool_results: + func_name = result.get("function_name", "") + if func_name == "confirm_fallacy" and result.get("confirmed"): + confirm_result = result + break + elif func_name == "conclude_no_fallacy": + conclude_result = result + break + elif func_name == "explore_branch": + node_info = result.get("node", {}) + next_pk = node_info.get("pk", "") + if next_pk and next_pk != current_pk: + explore_targets.append((next_pk, node_info)) + + # --- Decision, in priority order (preserves original semantics) --- + if confirm_result is not None: + confirmed_pk = confirm_result.get("pk", "") + confirmed_node = self.taxonomy_navigator.get_node(confirmed_pk) + confirmed_depth = ( + int(confirmed_node.get("depth", 0)) if confirmed_node else 0 + ) + justification = confirm_result.get("justification", "") + + # Capture reasoning for memory + reasoning_summary = ( + f"Confirmed {current_node.get(f'text_{self.language}', current_pk)} " + f"at depth {confirmed_depth}: {justification}" + ) + reasoning_history.append(reasoning_summary) + + # Reject too-shallow confirmations — force deeper exploration + if confirmed_depth < self.MIN_CONFIRM_DEPTH and children: + self.logger.info( + f" Confirmation at depth {confirmed_depth} rejected " + f"(min={self.MIN_CONFIRM_DEPTH}), forcing deeper exploration" + ) + # Pick the first child as default deeper path + first_child = children[0] + current_pk = first_child.get("PK", "") + navigation_trace.append(current_pk) + continue # re-iterate outer loop with new current_pk + + # Build full explanation with reasoning history + full_explanation = justification + if reasoning_history: + full_explanation += ( + f" | Reasoning path: {'; '.join(reasoning_history)}" + ) + + # Register confirmation for branch supersession (RA-3 #1048) + if supersession_tracker is not None: + supersession_tracker.register( + confirmed_pk, + confirmed_depth, + confirm_result.get("confidence", 0.7), + ) + + return IdentifiedFallacy( + fallacy_type=confirm_result.get( + "name", + confirm_result.get("name_fr", confirm_result.get("pk", "")), + ), + taxonomy_pk=confirmed_pk, + taxonomy_path=confirm_result.get("path", ""), + explanation=full_explanation, + confidence=confirm_result.get("confidence", 0.7), + navigation_trace=navigation_trace, + family=current_node.get("Famille", ""), + depth=confirmed_depth, + ) + + if conclude_result is not None: + reason = conclude_result.get("reason", "no reason") + # Capture reasoning for memory + reasoning_summary = f"Abandoned {current_node.get(f'text_{self.language}', current_pk)}: {reason}" + reasoning_history.append(reasoning_summary) + self.logger.info(f" Branch abandoned: {reason}") + return None + + if not explore_targets: + self.logger.info(" explore_branch returned same/empty node") + break + + # RA-3 #1048 item 2: the top child stays the primary path; extra + # promising children are fanned out as bounded concurrent + # sub-branches (additive — the primary path is never regressed). + primary_pk, primary_info = explore_targets[0] + extras = explore_targets[1:] + if ( + extras + and self.ENABLE_SUBBRANCH_FANOUT + and results_sink is not None + and supersession_tracker is not None + ): + await self._fanout_subbranches( + extras, + argument_text, + slave_kernel, + slave_settings, + reasoning_history, + supersession_tracker, + results_sink, + ) + + # Capture reasoning for memory - why the primary branch was chosen + primary_name = primary_info.get( + "name", primary_info.get("name_fr", primary_pk) + ) + reasoning_history.append( + f"Explored {primary_name} from " + f"{current_node.get(f'text_{self.language}', current_pk)}" + ) + current_pk = primary_pk + navigation_trace.append(current_pk) + self.logger.info( + f" Exploring deeper: {primary_name} " + f"(reasoning history: {len(reasoning_history)} steps)" + ) + + return None + + async def _fanout_subbranches( + self, + extras: List[Tuple[str, dict]], + argument_text: str, + slave_kernel: Kernel, + slave_settings: OpenAIPromptExecutionSettings, + reasoning_history: List[str], + supersession_tracker: "_BranchSupersessionTracker", + results_sink: List[IdentifiedFallacy], + ) -> None: + """Explore extra promising children as bounded concurrent sub-branches. + + RA-3 #1048 item 2. When a fork yields more than one promising child the + primary path follows the top child (handled by the caller) and the next + few are explored concurrently here. Results are appended to + *results_sink* — additive, so the primary single-path descent is never + regressed. Bounded by: + + - ``SUBBRANCH_FANOUT_WIDTH`` (extra children spawned per fork), and + - the tracker's shared global fan-out budget (per analysis), + + and pruned by the supersession tracker (dead-end / superseded + sub-branches abandon early). Anti-pendule #1019: fan-out is capped, not + unbounded — it mirrors the agentic navigation's + ``MAX_NAVIGATION_LLM_CALLS`` discipline. + """ + tasks = [] + for sub_pk, sub_info in extras[: self.SUBBRANCH_FANOUT_WIDTH]: + if not supersession_tracker.try_consume_fanout(): + self.logger.info(" Sub-branch fan-out budget exhausted") + break + sub_name = sub_info.get("name", sub_info.get("name_fr", sub_pk)) + self.logger.info(f" Fan-out sub-branch: {sub_name} (PK: {sub_pk})") + tasks.append( + self._explore_single_branch( + argument_text, + sub_pk, + slave_kernel, + slave_settings, + reasoning_history=list(reasoning_history), + supersession_tracker=supersession_tracker, + results_sink=results_sink, + ) + ) + if not tasks: + return + if self.descent_sequential: + # #2324 : le fan-out suit la même discipline que le Phase 2 — + # l'ordre d'arrivée dans ``results_sink`` devient déterministe. + sub_results = [] + for task in tasks: + try: + sub_results.append(await task) + except Exception as exc: # noqa: BLE001 — parité gather + sub_results.append(exc) + else: + sub_results = await asyncio.gather(*tasks, return_exceptions=True) + for r in sub_results: + if isinstance(r, IdentifiedFallacy): + results_sink.append(r) + elif isinstance(r, BaseException): + self.logger.warning(f" Sub-branch fan-out error: {r}") + + @kernel_function( + name="run_guided_analysis", + description=( + "Analyze text for fallacies using hierarchical taxonomy navigation. " + "Uses iterative deepening with parallel branch exploration, " + "falling back to one-shot if needed." + ), + ) + async def run_guided_analysis( + self, + argument_text: Annotated[str, "The text to analyze for fallacies"], + trace_log_path: Optional[str] = None, + use_one_shot: bool = False, + ) -> Annotated[str, "JSON result with identified fallacies"]: + """Execute hierarchical fallacy detection with one-shot fallback. + + Algorithm: + 1. Present root categories to slave LLM + 2. Slave selects candidate branches (via explore_branch calls) + 3. For each candidate, iterative deepening with double-selection + 4. Parallel exploration of up to MAX_BRANCHES branches + 5. If no result, fall back to one-shot full-taxonomy approach + + FB-30 (#1107): the mechanical per-level beam descent is removed. + Agentic navigation in step 3 now has no depth cap (taxonomy leaves are + the only cap) and shows the LLM a multi-level cluster so it can jump + levels — restoring the original summer-2025 design by subtraction. + """ + file_handler = self._attach_trace_handler(trace_log_path) + + try: + self.logger.info(f"--- Fallacy Analysis for: '{argument_text[:80]}...' ---") + + # Direct one-shot mode if requested + if use_one_shot: + return self._mark_regime( + await self._run_one_shot(argument_text), "one_shot" + ) + + # Phase 1: Wide-net candidate selection (#578) + candidate_pks = await self._wide_net_candidates(argument_text) + if not candidate_pks: + self.logger.info( + "Wide-net produced no candidates — falling back to one-shot" + ) + _fallback = self._mark_fallback(await self._run_one_shot(argument_text)) + return self._mark_regime(_fallback, "one_shot") + + self.logger.info( + f"Phase 1: {len(candidate_pks)} wide-net candidates: {candidate_pks}" + ) + + # Phase 2: Parallel iterative deepening on ALL candidates + slave_kernel, slave_settings = self._create_slave_kernel() + # Branch supersession tracker (RA-3 #1048) — shared across branches. + # The fan-out budget bounds recursive sub-branch exploration + # (item 2) across the whole analysis. + tracker = self._BranchSupersessionTracker( + self.taxonomy_navigator, + fanout_budget=( + self.SUBBRANCH_FANOUT_BUDGET if self.ENABLE_SUBBRANCH_FANOUT else 0 + ), + descent_call_budget=self.DESCENT_TOTAL_CALL_BUDGET, + ) + # Sink collecting fan-out sub-branch results (RA-3 #1048 item 2). + # Additive: top-level branch primaries return via the gather below; + # any extra promising children explored as sub-branches land here. + fanout_sink: List[IdentifiedFallacy] = [] + exploration_tasks = [ + self._explore_single_branch( + argument_text, + pk, + slave_kernel, + slave_settings, + reasoning_history=None, + supersession_tracker=tracker, + results_sink=fanout_sink, + ) + for pk in candidate_pks + ] + if self.descent_sequential: + # #2324 : une branche après l'autre, dans l'ordre des candidats + # — même exceptions que ``gather(return_exceptions=True)`` : + # une branche qui échoue n'interrompt pas les suivantes. + branch_results = [] + for task in exploration_tasks: + try: + branch_results.append(await task) + except Exception as exc: # noqa: BLE001 — parité gather + branch_results.append(exc) + else: + branch_results = await asyncio.gather( + *exploration_tasks, return_exceptions=True + ) + + # Phase 3: Collect + dedup by leaf taxonomy_pk (keep highest confidence) + identified = [] + seen_pks = {} + total_iterations = 0 + if tracker.confirmation_count > 0 or tracker.superseded_count > 0: + self.logger.info( + f"Branch supersession: {tracker.confirmation_count} confirmed, " + f"{tracker.superseded_count} branches superseded" + ) + for result in branch_results: + if isinstance(result, Exception): + self.logger.warning(f"Branch exploration error: {result}") + continue + if isinstance(result, IdentifiedFallacy): + total_iterations += len(result.navigation_trace) + leaf_pk = result.fallacy_type + if ( + leaf_pk in seen_pks + and seen_pks[leaf_pk].confidence >= result.confidence + ): + continue + seen_pks[leaf_pk] = result + identified = [r for r in identified if r.fallacy_type != leaf_pk] + identified.append(result) + + # Merge fan-out sub-branch results (RA-3 #1048 item 2) with the same + # dedup-by-leaf rule (keep highest confidence per leaf PK). + for result in fanout_sink: + total_iterations += len(result.navigation_trace) + leaf_pk = result.fallacy_type + if ( + leaf_pk in seen_pks + and seen_pks[leaf_pk].confidence >= result.confidence + ): + continue + seen_pks[leaf_pk] = result + identified = [r for r in identified if r.fallacy_type != leaf_pk] + identified.append(result) + if tracker.fanout_spawned: + self.logger.info( + f"Sub-branch fan-out explored {tracker.fanout_spawned} " + f"extra branch(es) (RA-3 #1048 item 2)" + ) + + if identified: + identified.sort(key=lambda f: f.confidence, reverse=True) + # FB-30 (#1107): the mechanical beam descent (_beam_descent, + # Phase 3b) is removed — restored agentic navigation in + # _explore_single_branch (no depth cap + multi-level cluster) + # now reaches deep/lateral nodes directly. Method label no + # longer carries the "_beam" suffix. + method = "wide_net_parallel" + analysis_result = FallacyAnalysisResult( + fallacies=identified, + exploration_method=method, + branches_explored=len(candidate_pks), + total_iterations=total_iterations, + ) + self.logger.info( + f"--- Analysis complete: {len(identified)} fallacies identified " + f"via {method} ({len(candidate_pks)} branches) ---" + ) + result_json = analysis_result.model_dump_json(indent=2) + # FB-35 (#1121): surface the global descent-call diagnostic + # always (calls made across all branches + sub-branches) and a + # fail-loud marker when the budget was hit. When exceeded, the + # fallacies above are PARTIAL (the descent was cut short) — + # flagged honestly so the report/convergence layer marks the + # phase degraded (anti-theater #1019). + result_obj = json.loads(result_json) + result_obj["descent_calls_made"] = tracker.descent_calls_made + if tracker.descent_budget_exceeded: + result_obj["descent_budget_exceeded"] = True + result_obj["exploration_method"] = "wide_net_parallel_budget_capped" + result_json = json.dumps(result_obj, indent=2, ensure_ascii=False) + else: + result_json = json.dumps(result_obj, indent=2, ensure_ascii=False) + self._persist_trace(trace_log_path, analysis_result, argument_text) + return self._mark_regime(result_json, "funnel") + + # Phase 4: Fallback to one-shot + self.logger.info( + "No fallacies found via iterative deepening — falling back to one-shot" + ) + one_shot_result = await self._run_one_shot(argument_text) + # FB-35 (#1121): surface the descent-call diagnostic on the one-shot + # fallback too. ``descent_calls_made`` is always injected (it records + # how many calls the iterative descent consumed before giving up), + # and ``descent_budget_exceeded`` marks the degradation when the + # descent was cost-capped (the one-shot itself is a single call — + # unbounded-by-design — so the marker reflects that the ITERATIVE + # descent was cut short). + try: + _os_obj = json.loads(one_shot_result) + _os_obj["descent_calls_made"] = tracker.descent_calls_made + if tracker.descent_budget_exceeded: + _os_obj["descent_budget_exceeded"] = True + one_shot_result = json.dumps(_os_obj, indent=2, ensure_ascii=False) + except (json.JSONDecodeError, TypeError): + pass + if trace_log_path: + try: + parsed = json.loads(one_shot_result) + if "fallacies" in parsed: + parsed["trace_note"] = "one_shot_fallback" + parsed["argument_excerpt"] = argument_text[:200] + check_plaintext_destination(trace_log_path) + trace_path = Path(trace_log_path) + trace_path.parent.mkdir(parents=True, exist_ok=True) + trace_path.write_text( + json.dumps(parsed, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + except PlaintextDestinationError as refused: + self.logger.warning("One-shot trace not written: %s", refused) + except (json.JSONDecodeError, TypeError): + pass + return self._mark_regime(one_shot_result, "one_shot") + + except Exception as e: + self.logger.error(f"Analysis error: {e}", exc_info=True) + return json.dumps({"error": str(e), "fallacies": []}) + finally: + if file_handler: + self.logger.removeHandler(file_handler) + file_handler.close() + + def _attach_trace_handler( + self, trace_log_path: Optional[str] + ) -> Optional[logging.FileHandler]: + """Build + attach the optional descent-trace FileHandler. + + Robust to a ``trace_log_path`` whose parent directory is missing or + non-creatable (#1722). The caller — often the LLM via SK + auto-function-invoke on the conversational path — may pass a POSIX path + improvised on a Windows host; a non-creatable trace path must NOT kill + the hierarchical taxonomy descent (#1019 family: a swallowed crash + produces an invisible degradation). Creates the parent dir; on failure + falls back to no handler with a WARNING. Returns the handler (or None) + so ``run_guided_analysis`` can detach it in its ``finally`` block. + """ + if not trace_log_path or str(trace_log_path).strip() in ("", "None", "null"): + return None + try: + check_plaintext_destination(str(trace_log_path)) + Path(str(trace_log_path)).parent.mkdir(parents=True, exist_ok=True) + handler = logging.FileHandler(trace_log_path, mode="w", encoding="utf-8") + handler.setFormatter( + logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + ) + ) + self.logger.addHandler(handler) + return handler + except (OSError, PermissionError, ValueError) as trace_exc: + self.logger.warning( + "trace_log_path '%s' unusable (%s); continuing without " "trace file", + trace_log_path, + trace_exc, + ) + return None + + def _persist_trace( + self, + trace_log_path: Optional[str], + result: "FallacyAnalysisResult", + argument_text: str, + ) -> None: + """Persist structured taxonomy traversal trace to JSON file.""" + if not trace_log_path: + return + try: + trace_entries = [] + for fallacy in result.fallacies: + node = self.taxonomy_navigator.get_node(fallacy.taxonomy_pk) + parent_chain = [] + if node and node.get("path"): + parent_chain = node["path"].split(" > ") + trace_entries.append( + { + "taxonomy_node_id": fallacy.taxonomy_pk, + "fallacy_type": fallacy.fallacy_type, + "parent_chain": parent_chain, + "navigation_trace": fallacy.navigation_trace, + "decision_rationale": fallacy.explanation, + "citation_excerpt": argument_text[:200], + "confidence": fallacy.confidence, + } + ) + + trace_data = { + "exploration_method": result.exploration_method, + "branches_explored": result.branches_explored, + "total_iterations": result.total_iterations, + "traversal_paths": trace_entries, + } + + check_plaintext_destination(trace_log_path) + trace_path = Path(trace_log_path) + trace_path.parent.mkdir(parents=True, exist_ok=True) + trace_path.write_text( + json.dumps(trace_data, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + self.logger.info(f"Trace persisted to {trace_log_path}") + except Exception as e: + self.logger.warning(f"Failed to persist trace: {e}") + + def _match_taxonomy_name(self, text: str) -> str: + """Try to find a taxonomy fallacy name within free-form LLM text (#655 Track KK). + + Returns the longest matching taxonomy name, or "" if nothing matches. + """ + known_names: list[str] = [] + for node in self.taxonomy_navigator.taxonomy_data: + for key in ("text_fr", "nom_vulgarisé"): + name = node.get(key, "") + if name and len(name) > 4: + known_names.append(name) + # Also check common French fallacy names not in taxonomy + known_names.extend( + [ + "Ad hominem", + "Appel à l'autorité", + "Pente glissante", + "Homme de paille", + "Faux dilemme", + "Pétition de principe", + "Appel à la tradition", + "Appel à la nouveauté", + "Généralisation hâtive", + "Fausse cause", + "Appel aux émotions", + "Changement de sujet", + "Fausse équivalence", + "Empoisonnement du puits", + "Appel à l'ignorance", + ] + ) + text_lower = text.lower() + best = "" + for name in known_names: + if name.lower() in text_lower and len(name) > len(best): + best = name + return best + + def _build_compact_taxonomy(self, max_depth: int = 4) -> str: + """Build a compact taxonomy representation (PK + name + path) to fit in context.""" + lines = [] + for node in self.taxonomy_navigator.taxonomy_data: + depth = int(node.get("depth", 0)) + if depth > max_depth: + continue + pk = node.get("PK", "") + name = node.get("text_fr", node.get("nom_vulgarisé", "")) + path = node.get("path", "") + if name and pk: + indent = " " * depth + lines.append(f"{indent}PK={pk} [{path}] {name}") + return "\n".join(lines) + + async def _run_one_shot(self, argument_text: str) -> str: + """Fallback one-shot analysis — sends compact taxonomy to LLM.""" + self.logger.info("Running one-shot fallback analysis") + + kernel, settings = self._create_one_shot_kernel() + # Use compact taxonomy (depth ≤ 4) to stay within token limits + compact_taxonomy = self._build_compact_taxonomy(max_depth=6) + + prompt = ( + f"Analyze the following text:\n--- TEXT ---\n{argument_text}\n--- END TEXT ---\n\n" + "Identify the single most relevant fallacy from the taxonomy below. " + "CRITICAL: Choose the MOST SPECIFIC (deepest) node that matches — " + "generic labels like 'Ad hominem' or 'Appel à l'autorité' are too shallow. " + "Prefer leaf nodes or deep sub-types (e.g., 'Empoisonnement du puits' instead of 'Culpabilité par association'). " + "IMPORTANT: Use the exact fallacy name as it appears in the taxonomy (in French). " + "Respond with ONLY a JSON object: " + '{"fallacy_name": "...", "taxonomy_pk": "...", "explanation": "...", "confidence": 0.0-1.0}\n\n' + f"--- TAXONOMY ---\n{compact_taxonomy}\n--- END ---" + ) + + history = ChatHistory( + system_message=( + "You are an expert in logical fallacies. Identify the most specific " + "and relevant fallacy from the taxonomy. Respond with ONLY a JSON object." + ) + ) + history.add_user_message(prompt) + + try: + response = await self.llm_service.get_chat_message_content( + chat_history=history, settings=settings, kernel=kernel + ) + raw = str(response).strip() + + # Try to parse structured JSON response + try: + text_to_parse = raw + if "```json" in text_to_parse: + text_to_parse = text_to_parse.split("```json")[1].split("```")[0] + elif "```" in text_to_parse: + text_to_parse = text_to_parse.split("```")[1].split("```")[0] + start = text_to_parse.find("{") + end = text_to_parse.rfind("}") + 1 + if start >= 0 and end > start: + parsed = json.loads(text_to_parse[start:end]) + fallacy_name = parsed.get("fallacy_name", "") + if fallacy_name: + fallacy = IdentifiedFallacy( + fallacy_type=fallacy_name, + taxonomy_pk=parsed.get("taxonomy_pk", ""), + explanation=parsed.get("explanation", ""), + confidence=float(parsed.get("confidence", 0.5)), + navigation_trace=[], + ) + result = FallacyAnalysisResult( + fallacies=[fallacy], + exploration_method="one_shot", + ) + return result.model_dump_json(indent=2) + except (json.JSONDecodeError, ValueError, TypeError): + pass + + # Fallback: try to match a taxonomy name in the raw text (#655 Track KK). + matched_name = self._match_taxonomy_name(raw) + if matched_name: + fallacy = IdentifiedFallacy( + fallacy_type=matched_name, + taxonomy_pk="", + explanation="One-shot identification (taxonomy name extracted from response)", + confidence=0.35, + navigation_trace=[], + ) + result = FallacyAnalysisResult( + fallacies=[fallacy], exploration_method="one_shot" + ) + return result.model_dump_json(indent=2) + + # No usable fallacy found — return empty rather than garbage + self.logger.warning( + "One-shot: could not extract usable fallacy name from response" + ) + return json.dumps( + {"fallacies": [], "exploration_method": "one_shot_no_match"} + ) + + except Exception as e: + self.logger.error(f"One-shot fallback failed: {e}") + return json.dumps({"error": str(e), "fallacies": []}) diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_identification_models.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_identification_models.py new file mode 100644 index 0000000000..5ab960bbe0 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_identification_models.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/plugins/identification_models.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +"""Pydantic models for structured fallacy identification output.""" + +from typing import List, Optional +from pydantic import BaseModel, Field + + +class IdentifiedFallacy(BaseModel): + """Structured result from hierarchical taxonomy-guided fallacy detection.""" + + fallacy_type: str = Field( + ..., description="The exact name of the identified fallacy from the taxonomy" + ) + taxonomy_pk: str = Field( + ..., description="The PK (primary key) of the fallacy in the taxonomy" + ) + taxonomy_path: str = Field( + default="", description="The full dot-separated path through the taxonomy" + ) + explanation: str = Field( + ..., description="Why this fallacy applies to the analyzed text" + ) + problematic_quote: str = Field( + default="", + description="The exact quote from the text that exhibits the fallacy", + ) + confidence: float = Field( + default=0.0, ge=0.0, le=1.0, description="Confidence score (0.0 to 1.0)" + ) + navigation_trace: List[str] = Field( + default_factory=list, + description="List of taxonomy node PKs visited during iterative deepening", + ) + family: str = Field( + default="", + description="The CSV 'Famille' column value (one of the 7 French families)", + ) + depth: Optional[int] = Field( + default=None, + description=( + "Taxonomy depth of the confirmed node. None = not measured " + "(one-shot regime, where no descent happened)" + ), + ) + + +class FallacyAnalysisResult(BaseModel): + """Complete result of a fallacy analysis run.""" + + fallacies: List[IdentifiedFallacy] = Field(default_factory=list) + exploration_method: str = Field( + default="one_shot", + description="Detection method used: 'iterative_deepening' or 'one_shot'", + ) + branches_explored: int = Field( + default=0, description="Number of taxonomy branches explored" + ) + total_iterations: int = Field( + default=0, description="Total iterations across all branches" + ) diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_plaintext_destination.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_plaintext_destination.py new file mode 100644 index 0000000000..968ff14bb9 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_plaintext_destination.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/core/plaintext_destination.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +"""Where corpus-derived plaintext may be written (#2738). + +The dataset is tracked only in encrypted form. Several writers put its +plaintext, or text excerpts derived from it, at a path their caller chooses: +definition exports, unencrypted saves and their fallbacks, analysis traces. +A path inside a git work tree that is not ignored makes that plaintext one +``git add`` away from the history. This module is the single check those +writers call before they open the file. +""" + +import subprocess +from pathlib import Path +from typing import Optional, Union + +# A file name that the repository ignores wherever it is created +# (``*_unencrypted*`` in .gitignore). Default for interactive exports. +DEFAULT_PLAINTEXT_EXPORT_PATH = "./extract_definitions_unencrypted.json" + + +class PlaintextDestinationError(ValueError): + """The destination could put plaintext into a git work tree.""" + + +def _enclosing_work_tree(start: Path) -> Optional[Path]: + """Nearest directory at or above *start* that holds a ``.git`` entry. + + ``.git`` is a directory in a main clone and a file in a linked worktree + or a submodule; both count. + """ + for candidate in (start, *start.parents): + if (candidate / ".git").exists(): + return candidate + return None + + +def check_plaintext_destination(path: Union[str, Path]) -> Path: + """Return *path* resolved, or raise if plaintext written there is stageable. + + A destination outside every git work tree is accepted. Inside one, it is + accepted only when ``git check-ignore`` says the path is ignored; a tracked + file is never ignored, so overwriting one with plaintext is refused too. + When git cannot answer, the write is refused: an unverified destination is + treated as an unsafe one. + """ + target = Path(path).resolve() + root = _enclosing_work_tree(target.parent) + if root is None: + return target + try: + proc = subprocess.run( + ["git", "-C", str(root), "check-ignore", "-q", str(target)], + capture_output=True, + text=True, + timeout=60, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise PlaintextDestinationError( + f"Refusing to write plaintext to {target}: it is inside the git " + f"work tree {root} and git could not say whether it is ignored " + f"({exc})." + ) from exc + if proc.returncode == 0: + return target + if proc.returncode == 1: + raise PlaintextDestinationError( + f"Refusing to write plaintext to {target}: it is inside the git " + f"work tree {root} and is not ignored, so it could be committed. " + f"Write it outside the repository, or under an ignored name or " + f"directory (for example '*_unencrypted*' or " + f"'argumentation_analysis/evaluation/results/')." + ) + raise PlaintextDestinationError( + f"Refusing to write plaintext to {target}: 'git check-ignore' failed " + f"in {root} (exit {proc.returncode}: {proc.stderr.strip()})." + ) diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_local_overrides.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_local_overrides.py new file mode 100644 index 0000000000..be7a551c58 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_local_overrides.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/utils/taxonomy_local_overrides.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +"""Code-level overrides applied to the vendored Argumentum taxonomy at load. + +The vendored CSV (``argumentum_fallacies_taxonomy.csv``) is a byte-faithful +mirror of upstream ArgumentumGames/Argumentum @ c86b71ff — the integrity +guard (tests/unit/scripts/test_argumentum_taxonomy_integrity.py) pins that +mirror, and re-pulls must keep diffing cleanly against upstream. Local +fixes therefore NEVER edit the CSV: they live here, addressed by PK, +reviewable and revertible the day upstream fixes the same rows. + +#2036 tranche 1 — purge of the two raw editorial notes in ``nom_vulgarisé``: + +- PK 41 (path 1.1.2.1.5, text_fr « Je le sais quand je le vois »): the field + carries a titling brainstorm (« Je n'aime aucun des titres. Chercher vers… »), + not a name. ``taxonomy_sophism_detector`` echoes ``nom_vulgarisé`` on every + detection, so a detection of this node would put the whole note in reader + prose. +- PK 992 (path 6.2.2, text_fr « Vouloir le beurre et l'argent du beurre »): + the field carries a column-swap question + (« Versatilité: ON INVERSERAIT PAS LE TEXT_FR eT LE NOM VULGARISé? »). + +Both are purged to the empty string — ``nom_vulgarisé`` is optional by +design (the field is empty on the vast majority of the 1408 rows), the +lexical matcher guards empty values (``if nom_vulgarise and …``), and no +replacement name is coined here: inventing taxonomy content is item 2 of +#2036, which needs an explicit register-policy decision. + +#2036 item 2 — attested display aliases for broken-register ``text_fr`` +names. Arbitration: display-only aliases are allowed ONLY where the French +term is *attested* (published title, reference glossary, established +proverb) — never forged. The four attestations live in +``RENDER_ALIAS_SOURCES`` and are machine-checked (a sourced-table/test pair +reddens on an unsourced alias). + +The alias is display-ONLY. It must never ride the loaded data: the lexical +matcher reads ``nom_vulgarisé`` (substring) and ``text_fr`` (words > 4 +chars) from the same loaded rows, so an alias in those fields would change +what matches. It applies at name-resolution-for-reader sites only — +detector state writes, benchmark name preference, navigator prompt renders +— via :func:`render_alias`. + +Nodes left untouched because no attested French term was found (searched +2026-09-05): PK 41 « Je le sais quand je le vois », PK 980 « La science +s'est déjà trompée », PK 992 « Vouloir le beurre et l'argent du beurre » +(attested idiom, no alternative technical term), PK 1009 « Où est le mal ? ». +The 7 borderlines are deliberately untouched (established idioms). +""" + +from __future__ import annotations + +from typing import Dict, Iterable, Union + +# PK -> {field: replacement}. Keys are ints; loaders normalise to str for +# DictReader rows. +EDITORIAL_NOTE_PURGES: Dict[int, Dict[str, str]] = { + 41: {"nom_vulgarisé": ""}, + 992: {"nom_vulgarisé": ""}, +} + +# PK -> attested display alias (#2036 item 2). Every entry MUST have a +# citation in RENDER_ALIAS_SOURCES — an alias without a source is a forged +# name and fails the guard test. +RENDER_ALIASES: Dict[int, str] = { + 320: "Pensez aux enfants", + 328: "Sagesse du dégoût", + 449: "Mieux vaut peu que rien", + 1311: "Gros mensonge", +} + +RENDER_ALIAS_SOURCES: Dict[int, str] = { + 320: "fr.wikipedia.org/wiki/Pensez_aux_enfants — nom rhétorique établi du cliché (argumentum ad misericordiam)", + 328: "fr.wikipedia.org/wiki/Sagesse_du_dégoût — titre français publié de la « wisdom of repugnance » (Kass, 1997)", + 449: "proverbe français attesté — linternaute.fr/proverbe/2008, dicocitations.com", + 1311: "rendu attesté de « big lie » — fr.wikipedia.org/wiki/Glossaire_de_la_langue_du_Troisième_Reich, Linguee/Reverso", +} + + +def render_alias(pk, default_name: str) -> str: + """Return the attested display alias for a node, else the default name. + + Display-only: callers pass the name they resolved for a reader (state + write, name preference, prompt render); the matching fields in the + loaded taxonomy are never rewritten. A malformed PK is a passthrough — + this helper must never be the reason a render crashes. + """ + try: + key = int(pk) + except (TypeError, ValueError): + return default_name + return RENDER_ALIASES.get(key, default_name) + + +def purge_row(row: Dict[str, str]) -> Dict[str, str]: + """Apply the editorial-note purges to one DictReader row, in place.""" + pk = str(row.get("PK", "")) + overrides = EDITORIAL_NOTE_PURGES.get(int(pk)) if pk.isdigit() else None + if overrides: + for field, replacement in overrides.items(): + row[field] = replacement + return row + + +def purge_rows(rows: Iterable[Dict[str, str]]) -> list: + """Apply the editorial-note purges to DictReader rows, in place.""" + return [purge_row(row) for row in rows] + + +def purge_dataframe(df) -> None: + """Apply the editorial-note purges to a pandas DataFrame in place. + + Addresses rows by the ``PK`` column (string-normalised comparison) and + only rewrites columns the override table names, so a schema change + upstream fails loudly at the missing-column access rather than purging + the wrong cell. + """ + if "PK" not in df.columns: + return + pk_column = df["PK"].astype(str) + for pk, overrides in EDITORIAL_NOTE_PURGES.items(): + mask = pk_column == str(pk) + if not mask.any(): + continue + for field, replacement in overrides.items(): + df.loc[mask, field] = replacement diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_navigator.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_navigator.py new file mode 100644 index 0000000000..d5bf173366 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_navigator.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/agents/utils/taxonomy_navigator.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +import json +from typing import List, Dict, Any, Optional + +from argumentation_analysis.utils.taxonomy_local_overrides import render_alias +from argumentation_analysis.utils.taxonomy_tree import ( + taxonomy_parent_path, + taxonomy_root_path, +) + + +class TaxonomyNavigator: + """ + Navigates an already-loaded taxonomy. + + The constructor takes loaded rows (`List[Dict[str, Any]]`), not a path — + loading from CSV/JSON belongs to the caller (#2041). + """ + + def __init__(self, taxonomy_data: List[Dict[str, Any]]): + if isinstance(taxonomy_data, str): + # #2041: a path where loaded rows belong would crash on the + # first character ('c'.get) — say what was wrong instead. An + # empty navigator would be a silent false-negative on the whole + # taxonomy, so refusing beats any lenient fallback. + raise TypeError( + "TaxonomyNavigator expects loaded rows (List[Dict[str, Any]]), " + f"got a str — a path was passed where loaded rows belong: {taxonomy_data!r}" + ) + self.taxonomy_data = taxonomy_data if taxonomy_data is not None else [] + self.node_map: Dict[str, Dict[str, Any]] = {} + self.path_map: Dict[str, Dict[str, Any]] = {} # New: For path-based lookup + self._build_node_map() + + def _build_node_map(self): + """Builds the node and path maps from the taxonomy data.""" + if not self.taxonomy_data: + return + for node in self.taxonomy_data: + pk = node.get("PK") + path = node.get("path") + if pk: + self.node_map[str(pk)] = node + if path: + self.path_map[str(path)] = node + + def get_node(self, node_id: str) -> Optional[Dict[str, Any]]: + """ + Retrieves a node by its 'PK'. + """ + return self.node_map.get(node_id) + + def get_node_by_path(self, path: str) -> Optional[Dict[str, Any]]: + """ + Retrieves a node by its 'path'. + """ + return self.path_map.get(path) + + def get_root_nodes(self) -> List[Dict[str, Any]]: + """ + Returns all nodes at the root of the taxonomy (depth 1). + """ + roots = [] + for node in self.taxonomy_data: + try: + if int(node.get("depth", -1)) == 1: + roots.append(node) + except (ValueError, TypeError): + continue + return roots + + def get_children(self, node_id: str) -> List[Dict[str, Any]]: + """ + Retrieves the direct children of a given node. + """ + parent_node = self.get_node(node_id) + if not parent_node: + return [] + + # #2401: the path is the relation. A depth cell is not consulted: one + # row's depth disagrees with its path, and the root's children carry + # bare segments no prefix rule reaches. + parent_path = str(parent_node.get("path", "")) + root = taxonomy_root_path( + (node.get("path"), node.get("depth")) for node in self.taxonomy_data + ) + return [ + node + for node in self.taxonomy_data + if taxonomy_parent_path(node.get("path"), root) == parent_path + ] + + def get_parent(self, node_id: str) -> Optional[Dict[str, Any]]: + """ + Retrieves the parent of a given node. + """ + node = self.get_node(node_id) + if not node: + return None + + # #2401: a bare segment belongs to the depth-0 root when the taxonomy + # carries one; without it, it is a top and has no parent. + root = taxonomy_root_path( + (row.get("path"), row.get("depth")) for row in self.taxonomy_data + ) + parent_path = taxonomy_parent_path(node.get("path"), root) + if parent_path is None: + return None + for parent_node in self.taxonomy_data: + if parent_node.get("path") == parent_path: + return parent_node + return None + + def is_leaf(self, node_id: str) -> bool: + """ + Checks if a node is a leaf (has no children). + """ + return not self.get_children(node_id) + + def get_branch_as_str(self, node_id: str) -> str: + """ + Returns a string representation of a branch, including the node + and its direct children, formatted with indentation. + """ + node = self.get_node(node_id) + if not node: + return "Node not found." + + branch_str = "" + node_name = render_alias( + node.get("PK"), node.get("nom_vulgarisé", node.get("PK")) + ) + branch_str += f"- {node_name} (ID: {node['PK']})\n" + + children = self.get_children(node_id) + for child in children: + child_name = render_alias( + child.get("PK"), child.get("nom_vulgarisé", child.get("PK")) + ) + branch_str += f" - {child_name} (ID: {child['PK']})\n" + + return branch_str.strip() + + def get_taxonomy_preview( + self, depth: int = 2, language: str = "fr", details: bool = True + ) -> str: + """ + Generates a string preview of the taxonomy up to a specified depth, + with an option to include details or just names. + """ + if not self.taxonomy_data: + return "Taxonomy data is not available." + + preview_lines = [] + + def build_preview(node_id, current_depth): + if current_depth > depth: + return + + node = self.get_node(node_id) + if not node: + return + + indent = " " * (current_depth - 1) + node_name = render_alias( + node.get("PK"), node.get(f"text_{language}", node.get("PK")) + ) + + if details: + desc = node.get(f"desc_{language}", "").strip() + preview_lines.append( + f"{indent}- {node_name} (ID: {node['PK']}): {desc}" + ) + else: + preview_lines.append(f"{indent}- {node_name}") + + children = self.get_children(node_id) + for child in children: + build_preview(child["PK"], current_depth + 1) + + root_nodes = self.get_root_nodes() + for root in root_nodes: + build_preview(root["PK"], 1) + + return "\n".join(preview_lines) + + def get_taxonomy_as_json(self) -> str: + """ + Returns the entire taxonomy data as a JSON string. + """ + if not self.taxonomy_data: + return "[]" + return json.dumps(self.taxonomy_data, indent=2, ensure_ascii=False) diff --git a/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_tree.py b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_tree.py new file mode 100644 index 0000000000..76a908acb6 --- /dev/null +++ b/MyIA.AI.Notebooks/SymbolicAI/Argument_Analysis/argumentation_lib/_taxonomy_tree.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + + +# Verbatim copy of `argumentation_analysis/utils/taxonomy_tree.py` from +# the 2025-Epita-Intelligence-Symbolique project +# (https://github.com/jsboigeEpita/2025-Epita-Intelligence-Symbolique), +# Copyright (c) 2025 jsboigeEpita, MIT License. +# Source commit: ecfd9b9c (2026-09-29). +# Verbatim import rationale: see NOTICE-EPITA at the root of this directory. +# +# Verbatim integrity: file content below this header is byte-for-byte identical +# to the upstream source at the cited commit. No CoursIA modification. +# +# This module is a verbatim vendoring of the EPITA-IS tronc plugin set +# (FallacyWorkflowPlugin + ExplorationPlugin + TaxonomyNavigator) used by +# Argumentation-02-Fallacies-Detection. The Python package is renamed +# `argumentation_lib` here so that the upstream `from argumentation_analysis.X` +# imports are rewritten (via a conftest-time sys.path shim, see +# `_paths.py`) -- see NOTICE-EPITA. +# +# The vendoring covers portee 2 of issue #18391 (entonnoir taxonomique +# agentique). The Lexique (DETECTEUR_SOPHISMES, 38 entrees) shipped in +# #18506 is kept as the deterministic baseline; this file enables the +# agentic comparison path (run_guided_analysis + exploration_plugin). +"""The taxonomy's parent relation, read from its ``path`` encoding (#2401). + +The Argumentum CSV encodes its tree in ``path``: a dotted path's parent drops +its last segment (``"1.2.3"`` -> ``"1.2"``). The depth-1 nodes are the one case +the prefix rule cannot express: they carry a bare segment (``"1"``, ``"2"``...), +not ``"0.1"``, and their parent is the depth-0 root (``PK 0``, ``path "0"``). + +Every child rule over the taxonomy reads the relation here. The local +``startswith(path + ".")`` rule found nothing under the root, and pairing it +with ``depth == parent_depth + 1`` lost the one row whose ``depth`` cell +disagrees with its path. +""" + +from typing import TYPE_CHECKING, Iterable, Optional, Tuple + +if TYPE_CHECKING: + import pandas as pd + + +def taxonomy_root_path(rows: Iterable[Tuple[object, object]]) -> Optional[str]: + """Return the path of the single depth-0 row, from ``(path, depth)`` pairs. + + ``None`` when the taxonomy carries no depth-0 row (test fixtures, subsets) + or more than one: bare-segment nodes are then tops with no parent. + """ + roots = [] + for path, depth in rows: + try: + if int(float(depth)) == 0: + roots.append(str(path)) + except (TypeError, ValueError): + continue + return roots[0] if len(roots) == 1 else None + + +def taxonomy_parent_path(path: object, root_path: Optional[str]) -> Optional[str]: + """Return the parent path of ``path``, or ``None`` for a node with no parent.""" + if path is None or path != path: # None or NaN + return None + text = str(path) + if not text or text == root_path: + return None + if "." in text: + return text.rsplit(".", 1)[0] + return root_path + + +def taxonomy_parent_paths(df: "pd.DataFrame") -> "pd.Series": + """Each row's parent path in a taxonomy DataFrame (``path``/``depth``).""" + root = ( + taxonomy_root_path(zip(df["path"], df["depth"])) + if "depth" in df.columns + else None + ) + return df["path"].map(lambda path: taxonomy_parent_path(path, root))