"""Helpers Jarvis (Phase 13) — construction de pré-prompts à partir de formulaires UI + exécution de flows agent avec budget. Pour ne pas surcharger app.py, on isole ici : - les fonctions `build_*_prompt(form_values)` qui composent le texte d'entrée envoyé à l'agent à partir des champs du formulaire - le générateur `run_jarvis_flow(prompt, model, api_key, budget_limit, drops_key)` qui pilote l'agent en mode streaming dans une bulle de chatbot Gradio (sans saisie utilisateur) L'agent lui-même est inchangé — c'est juste le harnais d'invocation qui change : (a) on lui injecte un budget via `budget_context`, (b) on lui passe un message utilisateur construit du formulaire, (c) on streame ses étapes dans le Chatbot. """ from __future__ import annotations from typing import Any, Generator, Optional def _get_app_module(): """Récupère le module app DÉJÀ chargé via sys.modules. Sur HF Spaces, app.py tourne comme __main__ (pas 'app'). Faire `from app import X` re-load app.py SOUS LA CLÉ 'app' et déclenche un bug Gradio (composants re-créés hors d'un contexte Blocks valide → 'Button' object has no attribute '_id'). Le pattern correct : lire app.X via sys.modules sans jamais ré-importer. Renvoie None si introuvable (= mode test isolé, p.ex. pytest).""" import sys m = sys.modules.get('__main__') # On vérifie qu'on a bien le bon module — pas par exemple le # __main__ d'un test pytest qui n'a pas nos symboles. if m is not None and hasattr(m, 'pick_unblown_gemini_key'): return m return sys.modules.get('app') def _content_to_text(content: Any) -> str: """Normalise un `AIMessage.content` LangChain en string plate (TEXTE PARLÉ uniquement, exclut les blocs thinking/reasoning). Selon le provider (Anthropic vs Gemini natif vs OpenAI), `m.content` peut être : - une str directe (cas le plus simple) - une liste de blocs dict {type, text, ...} (Anthropic, Gemini SDK natif quand reasoning_summary ou multimodal) - une liste de str (rare) - None (cas vide) On extrait UNIQUEMENT les blocs de type 'text' — pas les blocs 'thinking'/'reasoning' (cf. `_content_to_thoughts` pour ça). """ if content is None: return "" if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for block in content: if isinstance(block, str): parts.append(block) elif isinstance(block, dict): # On accepte UNIQUEMENT les blocs text (pas thinking). btype = block.get("type") if btype in (None, "text"): txt = block.get("text") if isinstance(txt, str): parts.append(txt) return "".join(parts) # Cas pathologique : on tente str() en garde-fou return str(content) def _content_to_thoughts(content: Any) -> str: """Extrait UNIQUEMENT le chain-of-thought / raisonnement exposé par le modèle dans `AIMessage.content`. Reconnaît les deux formats coexistant dans LangChain : - {"type": "thinking", "thinking": "..."} (Anthropic Extended Thinking, Gemini langchain-google-genai v0) - {"type": "reasoning", "reasoning": "..."} (OpenAI o1/o3, Gemini langchain-google-genai v1, standard LangChain Core 1.0) Renvoie "" si pas de thinking exposé (cas normal pour les modèles sans thinking, ou Gemini sans `include_thoughts=True`). """ if content is None or isinstance(content, str): return "" if not isinstance(content, list): return "" parts: list[str] = [] for block in content: if not isinstance(block, dict): continue btype = block.get("type") if btype == "thinking": txt = block.get("thinking") if isinstance(txt, str) and txt.strip(): parts.append(txt) elif btype == "reasoning": txt = block.get("reasoning") if isinstance(txt, str) and txt.strip(): parts.append(txt) return "\n".join(parts) # ---------- Narration lexicalisée des appels d'outils Jarvis ---------- # Pour les outils utilisés couramment dans les flux Jarvis, on remplace # l'affichage technique « 🔧 tool_name(args) » par une phrase en français # plus lisible. Fallback : si l'outil n'est pas dans la table, on garde # l'affichage technique actuel (zéro régression sur les ~30 autres outils). # # Chaque entrée : {"start": fn(args)->str, "done": fn(result_str)->str} # - `start` : phrase affichée AVANT l'exécution du tool (sur le tool_call) # - `done` : phrase affichée APRÈS (sur le ToolMessage de retour) # Si une fn lève une exception, on fait gracieusement fallback. def _truncate(s: str, n: int = 60) -> str: s = str(s or "").strip() return s if len(s) <= n else s[:n - 1] + "…" def _hi(text) -> str: """Wrap un terme-cle (mot JDM, cible candidate…) dans un span colore pour le faire ressortir dans les narrations Jarvis (couleur ambre sobre, distincte du gris du raisonnement). La classe `jarvis-term` est definie dans `_CHATBOT_CSS` cote app.py — si la classe est strippee par le sanitizer du Chatbot, le texte reste lisible (juste pas colore). HTML-escape le contenu pour eviter toute injection accidentelle si un terme JDM contient '<' '>' '&'. """ import html as _h return f'{_h.escape(str(text or ""))}' def strip_thinking_blocks(messages: list, keep_last: bool = True) -> list: """Filtre les blocs « thinking » / « reasoning » des AIMessage list-content. Gemini 3.x avec `include_thoughts=True` produit un content sous forme de liste `[{"type":"thinking", ...}, {"type":"text", ...}, ...]`. Ces résumés de raisonnement coûtent 30-50% des tokens accumulés et ne sont PAS nécessaires pour que l'agent continue son travail — seuls les tool_calls et tool results comptent pour la continuité d'action. `keep_last=True` : garde le DERNIER bloc thinking de la session, pour préserver le `thought_signature` que Gemini 3.x utilise entre tours (sans quoi l'API peut rejeter le retour des tool_calls enchaînés). """ if not messages: return messages # Trouver l'index du dernier AIMessage avec un thinking block last_thinking_idx = -1 if keep_last: for i, m in enumerate(messages): content = getattr(m, "content", None) if not isinstance(content, list): continue for blk in content: if isinstance(blk, dict) and blk.get("type") in ("thinking", "reasoning"): last_thinking_idx = i break out: list = [] for i, m in enumerate(messages): content = getattr(m, "content", None) if not isinstance(content, list): out.append(m) continue if i == last_thinking_idx: out.append(m) # garder tel quel pour la signature continue # Filtrer filtered = [ blk for blk in content if not (isinstance(blk, dict) and blk.get("type") in ("thinking", "reasoning")) ] if filtered == content: out.append(m) continue # Reconstruire (pydantic v2 model_copy) try: out.append(m.model_copy(update={"content": filtered})) except Exception: out.append(m) # fallback : on garde tel quel si copy échoue return out def build_relance_summary(messages: list, n_done: int, target: int, relance_num: int, max_relances: Optional[int] = None) -> str: """Construit un résumé condensé pour le HumanMessage de relance. Au lieu de ré-injecter `accumulated_messages` complet (50-200 messages, explosion des tokens), on liste : - les triplets déjà CONSOLIDÉS (à préserver) - les cibles déjà testées en échec (à ne pas reproposer, top 20) - les couples (term, relation) déjà pré-fetchés Le LLM reçoit cette synthèse + l'instruction « continue » et reprend avec un brief propre, sans bagage cognitif. """ consolidated: list[str] = [] failed: list[str] = [] prefetched: set[str] = set() for m in messages: name = getattr(m, "name", None) or "" content_str = str(getattr(m, "content", "") or "") parsed = _parse_tool_result(content_str) if not isinstance(parsed, dict): continue if name == "validate_candidate": t = parsed.get("term") or "?" r = parsed.get("relation") or "?" tg = parsed.get("target") or "?" triplet = f"{t} | {r} | {tg}" if parsed.get("ready_for_submission") is True or parsed.get("consolidation_status") == "consolidated": consolidated.append(triplet) else: failed.append(triplet) elif name == "list_existing_for_enrichment": t = parsed.get("term") r = parsed.get("relation") if t and r: prefetched.add(f"{t} | {r}") lines = [ f"⛔ STOP. Tu as consolidé **{n_done}/{target}** triplet(s) — il en manque " f"{target - n_done}.", ] if consolidated: lines.append("") lines.append("**Déjà consolidés (PRÉSERVE-les dans le fichier final)** :") for c in consolidated: lines.append(f" ✅ {c}") if failed: lines.append("") lines.append("**Cibles déjà testées sans succès** (NE PAS reproposer) :") cap = 20 for f in failed[:cap]: lines.append(f" ⏸️ {f}") if len(failed) > cap: lines.append(f" … (+ {len(failed) - cap} autres)") if prefetched: lines.append("") lines.append("**Pré-fetchs déjà effectués** (ne PAS rappeler " "`list_existing_for_enrichment` dessus, le registre " "côté Python te court-circuitera) :") for p in sorted(prefetched): lines.append(f" 📥 {p}") lines.append("") lines.append( f"RECOMMENCE avec de NOUVEAUX candidats sur d'AUTRES relations / " f"d'AUTRES cibles. Ne rends ta réponse finale QU'APRÈS avoir " f"atteint le compte cible. (Relance auto {relance_num}" + (f"/{max_relances}" if max_relances is not None else "") + ".)" ) return "\n".join(lines) # Seuil de condensation proactive de l'historique. Au-delà, l'historique # devient trop coûteux à ré-envoyer à chaque chunk → on remplace par # [HumanMessage initial, HumanMessage(summary+nudge)] avant que le LLM # n'explose le contexte. ~230k tokens ≈ 920k chars (heuristique simple # « 1 token ≈ 4 chars »). On vise 230k au lieu de 240k pour garder une # marge plus confortable sous le plafond effectif de ~250k tokens souvent # touché en pratique sur Gemini free tier (limite serveur, pas la limite # nominale 1M qui n'est pas toujours servie au free tier). Préfère # condenser un peu plus tôt et plus souvent que rater des appels. HISTORY_CONDENSE_THRESHOLD_CHARS = 920_000 # Nudges aléatoires injectés après un résumé condensé pour apporter de # la variété et de la nouveauté entre relances. Trois variantes # tirées au sort (random.choice) → l'agent ne reproduit pas le même # ton ni la même stratégie deux fois de suite. _CONDENSE_NUDGE_VARIANTS = [ "Continue en apportant plus de **variété et de nouveauté** dans tes " "propositions — change de relation, change d'angle, ne reprends pas " "ce qui a déjà été tenté.", "Continue : lorsque tu cibles un terme source ou une cible à " "**sens spécifique**, désambiguïse (`disambiguate`) puis propose / " "consolide avec le ou les **raffinements** plutôt qu'avec le " "terme générique.", "Continue : tu **n'as pas de limites**. Tu peux utiliser des termes " "techniques domain-specific, des néologismes, du vocabulaire de " "niche — la langue n'a pas de cloisons, du moment que tu passes " "par le flow d'enrichissement.", ] def _history_total_chars(messages: list) -> int: """Estimation grossière du poids de l'historique en chars. Sum(len(content)) sur tous les messages. Pas un comptage de tokens précis mais suffisant pour le seuil de condensation (heuristique 1 token ≈ 4 chars).""" return sum(len(str(getattr(m, "content", "") or "")) for m in messages) def condense_history_with_nudge( messages: list, *, consolidation_target: Optional[int], attempt: int = 0, ) -> Optional[list]: """Condense `messages` en `[HumanMessage initial, HumanMessage(summary + nudge aléatoire)]` SI son poids dépasse `HISTORY_CONDENSE_THRESHOLD_CHARS`. Retourne la nouvelle liste si la condensation a eu lieu, `None` sinon (poids sous le seuil ou erreur). Le summary est 100% déterministe (parcours Python — zéro appel LLM). Le nudge est tiré aléatoirement parmi `_CONDENSE_NUDGE_VARIANTS` pour apporter de la variété entre relances. La source de vérité du compteur est `count_consolidations()` (registry cumulatif, survit aux resets). Utilisé en DEUX endroits : 1. Après attente PerMinute (cf. except dans run_jarvis_flow) 2. Proactivement en cours de streaming dès que le seuil est atteint (évite que le contexte LLM n'explose même sans rate-limit) """ total_chars = _history_total_chars(messages) if total_chars <= HISTORY_CONDENSE_THRESHOLD_CHARS: return None try: from langchain_core.messages import HumanMessage as _HM from jdm_agent.enrich import count_consolidations import random as _random n_so_far = count_consolidations() summary = build_relance_summary( messages, n_so_far, consolidation_target or (n_so_far + 1), attempt, None, ) nudge = _random.choice(_CONDENSE_NUDGE_VARIANTS) return [ messages[0], # HumanMessage initial (le prompt original) _HM(content=summary + "\n\n" + nudge), ] except Exception: return None def count_consolidated_in_messages(messages: list) -> int: """Compte les triplets consolidés en parcourant les ToolMessages `validate_candidate` accumulés pendant l'invocation agent. Un triplet est compté comme consolidé si le retour contient soit `ready_for_submission=True`, soit `consolidation_status="consolidated"`. Robuste aux différents formats de sérialisation (JSON / Python repr). """ n = 0 for m in messages: name = getattr(m, "name", None) or "" if name != "validate_candidate": continue content = getattr(m, "content", "") or "" parsed = _parse_tool_result(str(content)) if not isinstance(parsed, dict): continue if parsed.get("ready_for_submission") is True: n += 1 elif parsed.get("consolidation_status") == "consolidated": n += 1 return n def is_invalid_api_key(exc) -> bool: """Détecte une clé API invalide (typo, révoquée, jamais activée pour Gemini, etc.). Google renvoie alors 400 INVALID_ARGUMENT avec `reason: 'API_KEY_INVALID'`. À traiter comme exclusion permanente (la clé ne deviendra pas valide en attendant) → bascule de clé.""" msg = str(exc) return "API_KEY_INVALID" in msg or "API key not valid" in msg def _extract_quota_model(exc) -> Optional[str]: """Extrait l'identifiant du modèle concerné par un quota épuisé, via `quotaDimensions.model` dans le message d'erreur. Ex : « 'quotaDimensions': {'location': 'global', 'model': 'gemini-3.1-flash-lite'} » → renvoie 'gemini-3.1-flash-lite'. Renvoie None si pas trouvé.""" import re msg = str(exc) m = re.search(r"'model'\s*:\s*'([^']+)'", msg) if m: return m.group(1) m = re.search(r'"model"\s*:\s*"([^"]+)"', msg) if m: return m.group(1) return None def is_per_day_quota_exhausted(exc, expected_model: Optional[str] = None) -> bool: """Détecte les quotas QUOTIDIENS épuisés sur Gemini. Logique STRICTE en deux temps : 1. PAS un 429 / RESOURCE_EXHAUSTED → False 2. PAS un PerDay → False : - si le message contient « PerMinute » sans « PerDay » dans un quotaId de violation → PerMinute, jamais PerDay. - on regarde UNIQUEMENT le quotaId à l'intérieur de la première « violations: [...] » de l'erreur (les autres quotaId qui pourraient apparaître ailleurs — limits configurées, messages de doc — sont ignorés). 3. si expected_model fourni : le quota doit cibler exactement ce modèle (extrait via quotaDimensions). Cible : éviter les faux positifs PerDay observés quand le payload PerMinute contenait par ailleurs une mention « PerDay » dans une description secondaire. """ import re msg = str(exc) if "RESOURCE_EXHAUSTED" not in msg and "429" not in msg: return False # Extrait le bloc 'violations: [...]' si présent. Le quotaId du # quota EFFECTIVEMENT violé y est. Hors de ce bloc on peut trouver # des quotaId informatifs (limits configurées, autres infos) qui # ne doivent PAS déclencher la détection. m = re.search(r"['\"]?violations['\"]?\s*:\s*\[(.*?)\]", msg, re.DOTALL) block = m.group(1) if m else msg # fallback : tout le msg si pas trouvé if not re.search(r"['\"]quotaId['\"]?\s*:\s*['\"][^'\"]*PerDay", block): return False # Sanity check final : si le bloc contient AUSSI un PerMinute # violation mais pas un PerDay isolé → c'est PerMinute, pas PerDay. has_perminute_violation = bool(re.search( r"['\"]quotaId['\"]?\s*:\s*['\"][^'\"]*PerMinute", block)) has_perday_violation = bool(re.search( r"['\"]quotaId['\"]?\s*:\s*['\"][^'\"]*PerDay", block)) if has_perminute_violation and not has_perday_violation: return False if not has_perday_violation: return False if expected_model is None: return True quota_model = _extract_quota_model(exc) if quota_model is None: return True # conservateur : on considère que ça concerne le modèle courant return quota_model == expected_model def detect_rate_limit_retry(exc) -> Optional[float]: """Détecte les erreurs 429 quota PerMinute Gemini (transients) et extrait le délai de retry recommandé par l'API. Ne renvoie un délai QUE pour les quotas PerMinute (fenêtres glissantes qui se régénèrent vite). Les quotas PerDay sont explicitement exclus → cf. `is_per_day_quota_exhausted` qui les intercepte en amont pour signaler une exhaustion réelle. Renvoie None si : - pas un 429 / RESOURCE_EXHAUSTED - pas un PerMinute (PerDay traité ailleurs) - pas de retryDelay parseable - délai > 120s """ import re msg = str(exc) if "RESOURCE_EXHAUSTED" not in msg and "429" not in msg: return None if "PerMinute" not in msg: return None # PerDay ou autre → pas de retry ici m = re.search(r"retry in ([\d.]+)s", msg) if not m: m = re.search(r"retryDelay['\"]?\s*:\s*['\"]?(\d+)s", msg) if not m: return None try: delay = float(m.group(1)) except ValueError: return None if delay > 120: return None return delay + 1.0 _PARSE_EMPTY = object() # sentinelle : contenu vide / blanc _PARSE_UNPARSABLE = object() # sentinelle : ni JSON ni Python literal valide def _parse_tool_result(content: str): """Parse défensif d'un retour de tool. Retourne : - `_PARSE_EMPTY` si le contenu est vide ou whitespace - l'objet parsé (dict / list / str / int / float / bool / None) si le contenu est du JSON valide - l'objet parsé via `ast.literal_eval` si le contenu est un Python literal valide (dict-repr avec single quotes, etc. — LangChain sérialise parfois ainsi) - `_PARSE_UNPARSABLE` si rien de tout ça ne marche Note : ne renvoie PLUS de dict vide `{}` pour signaler une erreur ; on utilise les sentinelles dédiées. """ import json import ast if not content or not str(content).strip(): return _PARSE_EMPTY s = str(content).strip() # 1) JSON canonique try: return json.loads(s) except Exception: pass # 2) Python literal (dict-repr avec single quotes, tuples, etc.) try: return ast.literal_eval(s) except Exception: pass return _PARSE_UNPARSABLE def _format_done(content: str, fmt_dict) -> str: """Helper commun pour les callbacks 'done' du TOOL_NARRATION. Parse le contenu, puis dispatche : - vide → '→ vide' - non parsable → '→ (résultat non parsable)' - dict → délègue à `fmt_dict(d)` - list → '→ N élément(s)' - autre (str/int) → '→ {valeur tronquée}' """ parsed = _parse_tool_result(content) if parsed is _PARSE_EMPTY: return "→ vide" if parsed is _PARSE_UNPARSABLE: return "→ (résultat non parsable)" if isinstance(parsed, dict): try: return fmt_dict(parsed) except Exception: return "→ (format inattendu)" if isinstance(parsed, list): return f"→ {len(parsed)} élément(s)" return f"→ {_truncate(str(parsed), 80)}" TOOL_NARRATION: dict[str, dict] = { "list_existing_for_enrichment": { "start": lambda a: ( f"📥 Je récupère ce qui existe déjà sur " f"« {_hi(_truncate(a.get('term')))} » pour la relation " f"`{a.get('relation_name') or a.get('relation') or '?'}`…" ), "done": lambda c: _format_done(c, lambda d: ( f"→ {d.get('count', '?')} triplet(s) existant(s) trouvé(s)." )), }, "validate_candidate": { "start": lambda a: ( f"🧪 Je teste le candidat « {_hi(_truncate(a.get('term')))} | " f"`{a.get('relation', '?')}` | {_hi(_truncate(a.get('target')))} »…" ), # Cascade d'affichage du verdict : # 1. consolidation_status (résultat de l'inférence) prime # 2. sinon validation_status (résultat de la validation structurelle) # Note : `not_consolidated` est la VRAIE valeur renvoyée par # consolidate_candidate quand l'inférence est silencieuse — l'ancien # `silent` ne matchait jamais et tombait dans le fallback `→ ok`. "done": lambda c: _format_done(c, lambda d: ( "✅ consolidé par inférence" if d.get("consolidation_status") == "consolidated" else "⏸️ non inférable à partir de JDM" if d.get("consolidation_status") in ("not_consolidated", "silent") else "❌ rejeté par inférence" if d.get("consolidation_status") == "rejected" # Pas de consolidation tentée → on lit le validation_status seul. else "❌ doublon" if d.get("validation_status") == "duplicate" else "❌ cible inconnue de JDM" if d.get("validation_status") == "unknown_term" else "❌ contradiction directe dans JDM" if d.get("validation_status") == "inconsistent" else "⏸️ non inférable à partir de JDM" if d.get("validation_status") == "ok" else f"→ {d.get('validation_status', '?')}" )), }, "disambiguate": { "start": lambda a: f"🔎 Je cherche les sens de « {_hi(_truncate(a.get('term')))} »…", "done": lambda c: _format_done(c, lambda d: ( f"→ {len(d.get('senses') or d.get('refinements') or []) or '?'} sens trouvés." )), }, "lookup_term": { "start": lambda a: f"📖 Je vérifie l'existence de « {_hi(_truncate(a.get('term')))} » dans JDM…", "done": lambda c: _format_done(c, lambda d: ( "→ trouvé." if d.get("found") or d.get("id") else "→ inconnu." )), }, "get_relations_of_type": { "start": lambda a: ( f"🔗 Je regarde les triplets « {_hi(_truncate(a.get('term')))} | " f"`{a.get('relation_name') or a.get('relation') or '?'}` »…" ), "done": lambda c: _format_done(c, lambda d: ( f"→ {d.get('count', len(d.get('triplets', [])) or '?')} relation(s) trouvée(s)." )), }, "write_submission_file": { "start": lambda a: ( f"💾 J'écris le fichier de soumission ({len(a.get('triplets') or [])} item(s))" + (" et je le pousse à JDM…" if a.get("upload") else "…") ), "done": lambda c: _format_done(c, lambda d: ( f"❌ {d['error']}" if d.get("error") else f"→ écrit dans `{d.get('path', '?')}` ({d.get('count', '?')} ligne(s))." )), }, } def _narrate_tool_call(name: str, args: dict) -> Optional[str]: """Renvoie une phrase narrative pour un tool_call si l'outil est connu de TOOL_NARRATION, sinon None (le caller fera fallback sur l'affichage technique).""" spec = TOOL_NARRATION.get(name) if not spec: return None try: return spec["start"](args or {}) except Exception: return None def _narrate_tool_result(name: str, content: str) -> Optional[str]: """Renvoie une phrase narrative pour un ToolMessage si l'outil est connu, sinon None.""" spec = TOOL_NARRATION.get(name) if not spec: return None try: return spec["done"](content) except Exception: return None # ---------- Construction des pré-prompts ---------- def _is_bounded_budget(budget_label: str) -> bool: """True si le label correspond à une limite finie (et donc qu'il faut prévenir le LLM du sentinel BUDGET_EXHAUSTED). Évite de polluer le prompt quand l'utilisateur a choisi 'illimité'.""" if not budget_label: return False s = str(budget_label).strip().lower() if s in ("illimité", "illimite", "unlimited", "none", "0", ""): return False return s.isdigit() and int(s) > 0 _RANDOM_TERM_INSTRUCTION = ( "Je n'ai pas précisé de terme — TIRE toi-même un mot français au " "hasard et VARIÉ (varie domaine, registre, longueur, niveau " "d'abstraction d'un essai à l'autre et d'une session à l'autre). " "Évite les taxonomies scolaires (animaux, plantes) où JDM est " "déjà dense. Vérifie d'abord qu'il existe via `lookup_term` ; " "si non, recommence avec un autre — jusqu'à un terme exploitable." ) def _norm_relations(rels) -> list[str]: """Normalise une entrée 'relations' qui peut être None, str ou liste.""" if rels is None: return [] if isinstance(rels, str): rels = [rels] return [str(r).strip() for r in rels if r and str(r).strip()] def build_enrich_prompt( term: str, relation=None, target_count: int = 10, vary_relations: bool = False, iterate: bool = False, budget_label: str = "25", upload: bool = False, ) -> str: """Compose le pré-prompt d'enrichissement à partir du formulaire. `relation` accepte str (rétro-compat), list ou None. """ term = (term or "").strip() rels = _norm_relations(relation) bounded = _is_bounded_budget(budget_label) parts: list[str] = [] if term: parts.append(f"Je veux ENRICHIR le terme « {term} » dans JDM.") else: parts.append("Je veux ENRICHIR un terme dans JDM.") parts.append(_RANDOM_TERM_INSTRUCTION) # Contrainte de portée selon ce que l'utilisateur a fourni : # - term + rels → tous les triplets ont CE terme et UNE de CES # relations ; seule la cible varie. (cas A) # - term seul → CE terme reste fixe ; relation ET cible varient. (cas B) # - rels seules → CES relations restent fixes ; le terme (source) # varie pour produire des triplets variés. (cas C) # - rien → tout est libre, comportement par défaut. (cas D) if term and rels: rel_str = ", ".join(f"`{r}`" for r in rels) parts.append( f"⚠️ IMPÉRATIF : tous les triplets proposés DOIVENT avoir " f"« {term} » comme SOURCE et l'une des relations {rel_str} " "comme PRÉDICAT. Tu varies les CIBLES uniquement. NE propose " "AUCUN triplet sur un autre terme ou une autre relation — " "même si elles paraissent intéressantes." ) elif term and not rels: parts.append( f"⚠️ IMPÉRATIF : tous les triplets proposés DOIVENT avoir " f"« {term} » comme SOURCE. Tu varies les RELATIONS et les " "CIBLES, mais pas le terme." ) elif rels and not term: rel_str = ", ".join(f"`{r}`" for r in rels) if len(rels) == 1: parts.append( f"⚠️ IMPÉRATIF : tous les triplets proposés DOIVENT " f"utiliser la relation {rel_str} comme PRÉDICAT. Tu " "varies les TERMES sources (et les cibles)." ) else: parts.append( f"⚠️ IMPÉRATIF : tous les triplets proposés DOIVENT " f"utiliser l'une des relations {rel_str} comme PRÉDICAT. " "Tu varies les TERMES sources (et les cibles)." ) # Cas D (ni term ni rels) : aucune contrainte injectée — l'agent # est libre, _RANDOM_TERM_INSTRUCTION (plus haut) suffit pour # orienter le tirage. if vary_relations and not rels: # 'Varier les relations' ne s'applique que si l'utilisateur n'a # PAS imposé de relations spécifiques (sinon contradictoire). parts.append( "Varie explicitement les TYPES de relations explorées — " "pas une seule, plusieurs angles." ) parts.append( f"Objectif : produire {int(target_count)} triplets candidats " "CONSOLIDÉS (ready_for_submission=true)." ) if iterate: # Persistance EXPLICITE : sans ce paragraphe, les LLM abandonnent # après 2-3 « non inférable » consécutifs en concluant que rien # n'est consolidable. Or c'est NORMAL d'en avoir beaucoup (le # moteur d'inférence est strict). La règle : tant que budget # restant, GÉNÉRER d'autres candidats avec d'AUTRES relations. if bounded: parts.append( "PERSISTANCE OBLIGATOIRE : itère jusqu'à atteindre le " "nombre cible OU jusqu'à épuisement du budget. Recevoir " "plusieurs « non inférable à partir de JDM » de suite " "est NORMAL — le moteur d'inférence est strict et ne " "consolide que ce qu'il peut prouver. Dans ce cas : " "essaie d'AUTRES relations, d'AUTRES cibles, ne te " "résigne pas. N'abandonne JAMAIS avant d'avoir épuisé " "le budget ou atteint le nombre cible de consolidés." ) else: parts.append( "PERSISTANCE ABSOLUE — N'ABANDONNE JAMAIS.\n" "Tu es en BUDGET ILLIMITÉ. La valeur de ce que tu " "produis est PROPORTIONNELLE au nombre de tentatives " "que tu endures avant de trouver des candidats " "consolidés. Recevoir 20, 50, 100 « non inférable à " "partir de JDM » consécutifs est NORMAL et ATTENDU — " "le moteur d'inférence est strict par construction.\n" "RÈGLES :\n" "1. NE JAMAIS s'arrêter après quelques échecs. Un " "résultat à 0 consolidé après 20 essais n'est PAS " "un échec : c'est la BASELINE attendue.\n" "2. Tant que le nombre cible de consolidés n'est PAS " "atteint, GÉNÈRE encore et encore des candidats — " "varie systématiquement les relations, varie les " "cibles, explore des angles obliques (relations " "moins évidentes, sens raffinés, inverses).\n" "3. Un seul triplet réellement consolidé après 50 " "tentatives vaut INFINIMENT plus qu'une réponse " "rapide « rien trouvé ». L'utilisateur a coché " "« itérer » EXPRÈS pour que tu persistes.\n" "4. NE rends ta réponse finale QUE quand tu as atteint " "le nombre cible de consolidés." ) if bounded: parts.append( f"Budget : {budget_label} appels d'outils maximum. Au-delà, " "tu recevras un sentinel BUDGET_EXHAUSTED — arrête alors " "immédiatement et compose ta réponse finale avec ce qui est " "déjà consolidé." ) if upload: parts.append( "Soumets directement le fichier d'enrichissement au " "endpoint LLMDrops à la fin (write_submission_file avec " "upload=True). La clé est dans l'env JDM_DROPS_API_KEY." ) else: parts.append( "Écris le fichier d'enrichissement à la fin " "(write_submission_file SANS upload=True) — l'utilisateur " "décidera ensuite de le soumettre ou non." ) parts.append( "Tu SUIVRAS `enrichment_workflow()` en TOUT PREMIER pour le " "flux canonique — c'est obligatoire." ) return "\n".join(parts) def build_audit_prompt( term: str, relation=None, budget_label: str = "50", upload: bool = False, ) -> str: """Compose le pré-prompt d'audit à partir du formulaire. `relation` accepte str (rétro-compat), list ou None. """ term = (term or "").strip() rels = _norm_relations(relation) parts: list[str] = [] if term: parts.append(f"Je veux AUDITER le terme « {term} » dans JDM.") else: parts.append( "Je veux AUDITER un terme POLYSÉMIQUE dans JDM (chercher des " "contaminations du générique par des sens non-premiers)." ) parts.append(_RANDOM_TERM_INSTRUCTION + ( " IMPORTANT : pour l'audit, le terme tiré doit être " "POLYSÉMIQUE (plusieurs sens dans disambiguate) ; sinon " "retire un autre mot." )) if len(rels) == 1: parts.append(f"Restreins l'audit à la relation `{rels[0]}`.") elif len(rels) > 1: parts.append( "Restreins l'audit à ces relations : " + ", ".join(f"`{r}`" for r in rels) + "." ) else: parts.append( "Pas de relation imposée : couvre un nombre suffisant de " "types de relations (variées) pour faire un audit représentatif." ) if _is_bounded_budget(budget_label): parts.append( f"Budget : {budget_label} appels d'outils maximum. Au-delà, " "arrête et compose ta synthèse avec ce que tu as déjà examiné." ) if upload: parts.append("Soumets ensuite le fichier .audit à JDM (LLMDrops).") else: parts.append( "Écris le fichier .audit (sans upload) — l'utilisateur " "décidera ensuite de la soumission." ) parts.append( "Tu SUIVRAS `audit_workflow()` en TOUT PREMIER. C'est obligatoire." ) return "\n".join(parts) def build_gap_prompt( term: str, relations: Optional[list[str]] = None, budget_label: str = "25", ) -> str: """Compose le pré-prompt de détection de trous à partir du formulaire.""" term = (term or "").strip() parts: list[str] = [] if term: parts.append(f"Je veux DÉTECTER les trous de JDM pour le terme « {term} ».") else: parts.append("Je veux DÉTECTER les trous de JDM pour un terme.") parts.append(_RANDOM_TERM_INSTRUCTION) rels = _norm_relations(relations) if rels: parts.append( "Relations cibles : " + ", ".join(f"`{r}`" for r in rels) + "." ) else: parts.append( "Pas de relation imposée : choisis-les toi-même (variées, " "couvre un nombre suffisant de types)." ) if _is_bounded_budget(budget_label): parts.append( f"Budget : {budget_label} appels d'outils maximum." ) parts.append( "Pour chaque gap identifié, propose explicitement les 3 actions " "(Enrichir / Auditer / Stats) avec le format `term | relation | " "type_de_gap` pour que je puisse les router." ) parts.append( "Tu SUIVRAS `gap_detection_workflow()` en TOUT PREMIER. Obligatoire." ) return "\n".join(parts) def build_signalement_prompt( term: str, relation=None, budget_label: str = "50", upload: bool = False, ) -> str: """Compose le pré-prompt de signalement à partir du formulaire. `relation` accepte str (rétro-compat), list ou None. """ term = (term or "").strip() rels = _norm_relations(relation) parts: list[str] = [] if term: parts.append( f"Je veux SIGNALER les triplets suspects de JDM pour « {term} »." ) else: parts.append("Je veux SIGNALER les triplets suspects de JDM pour un terme.") parts.append(_RANDOM_TERM_INSTRUCTION) if len(rels) == 1: parts.append(f"Restreins le scan à la relation `{rels[0]}` seule.") elif len(rels) > 1: parts.append( "Restreins le scan à ces relations : " + ", ".join(f"`{r}`" for r in rels) + "." ) else: parts.append( "Pas de relation imposée : choisis-les toi-même (variées, " "couvre un nombre suffisant de types)." ) parts.append( "Utilise TON JUGEMENT linguistique de francophone — pas besoin " "de vérifier chaque suspect par un outil, ta suspicion vaut. " "Suis la grille de signaux du workflow (sémantiques + structurels)." ) if _is_bounded_budget(budget_label): parts.append( f"Budget : {budget_label} appels d'outils maximum. Limite à ~20 " "suspects max pour éviter le bruit." ) else: parts.append("Limite à ~20 suspects max pour éviter le bruit.") if upload: parts.append("Soumets ensuite le fichier .err à JDM (LLMDrops).") else: parts.append("Écris le fichier .err sans upload.") parts.append( "Tu SUIVRAS `signalement_workflow()` en TOUT PREMIER. Obligatoire." ) return "\n".join(parts) def build_stats_prompt( term: str = "", relation=None, budget_label: str = "50", upload: bool = False, ) -> str: """Compose le pré-prompt de stats à partir du formulaire. `relation` accepte str, list ou None. Une seule relation passée → mode PAR_RELATION focalisé ; plusieurs → mode PAR_RELATION sur chacune ; aucune + terme → mode PAR_TERME (relations choisies par le LLM). """ term = (term or "").strip() rels = _norm_relations(relation) rel_label = ( f"`{rels[0]}`" if len(rels) == 1 else (", ".join(f"`{r}`" for r in rels) if rels else "") ) parts: list[str] = [] if term and rels: parts.append( f"Je veux des STATISTIQUES JDM sur le terme « {term} », " f"RESTREINTES à la/aux relation(s) {rel_label}." ) parts.append( "⚠️ Limite-toi STRICTEMENT à cette/ces relation(s) — n'en " "examine aucune autre, même par souci de couverture." ) elif term: parts.append( f"Je veux des STATISTIQUES JDM sur le terme « {term} » " "(mode PAR_TERME : couverture relation par relation)." ) elif rels: parts.append( f"Je veux des STATISTIQUES JDM sur la/les relation(s) " f"{rel_label} (mode PAR_RELATION : distribution sur " "termes-pivots variés)." ) parts.append( "⚠️ Limite-toi STRICTEMENT à cette/ces relation(s) — " "n'en examine aucune autre." ) else: parts.append( "Je veux des STATISTIQUES JDM mais je n'ai pas précisé " "le terme ni la relation — exécute le mode PAR_TERME sur " "un terme tiré au hasard." ) parts.append(_RANDOM_TERM_INSTRUCTION) if _is_bounded_budget(budget_label): parts.append( f"Budget : {budget_label} appels d'outils maximum." ) # La consigne 'couvre N types' ne s'applique QUE si aucune relation # n'est imposée — sinon contradictoire avec la restriction stricte. if not rels: parts.append( "Couvre un nombre SUFFISANT de types de relations (au moins " "8-12 différents) — qualité statistique." ) parts.append( "Rends DEUX vues complémentaires :\n" " 1) TABLEAU par RELATION : une ligne par relation (n_total, " "n_pos, n_neg, max_w, min_w, mean_w).\n" " 2) TABLEAU par TERMES RENCONTRÉS : agrège les cibles " "(targets) toutes relations confondues — top 20 par " "occurrence/poids — avec nb_relations_distinctes et " "poids_total. Permet de voir quels termes reviennent souvent.\n" "Plus 3-5 observations BRÈVES et FACTUELLES après les tableaux." ) if upload: parts.append( "Soumets directement le fichier `.stat` à JDM (LLMDrops) à " "la fin (`write_submission_file(..., upload=True)`)." ) else: parts.append( "Écris le fichier `.stat` à la fin " "(`write_submission_file(..., upload=False)`) — l'utilisateur " "décidera ensuite de le soumettre ou non." ) parts.append( "Tu SUIVRAS `stats_workflow()` en TOUT PREMIER. Obligatoire." ) return "\n".join(parts) # ---------- Exécution de flow agent (avec budget) ---------- # Mapping label dropdown → limite numérique. `"illimité"` → None. BUDGET_LABEL_TO_LIMIT: dict[str, Optional[int]] = { "10": 10, "25": 25, "50": 50, "100": 100, "illimité": None, } def _extract_submission_path(tool_message_content: str) -> Optional[str]: """Extrait le chemin du fichier produit par write_submission_file. Le ToolMessage contient un dict sérialisé en JSON (ou en repr Python). On regarde la clé `path` MAIS on retourne None si : - `error` est présent dans le dict (l'écriture a échoué) - `path` est vide - le fichier n'existe pas physiquement (filet anti-régression contre les chemins valides retournés sans fichier derrière) """ import json import re from pathlib import Path if not tool_message_content: return None def _validate(p: str) -> Optional[str]: if not p: return None try: if Path(p).exists(): return p except Exception: pass return None try: d = json.loads(tool_message_content) if isinstance(d, dict): if d.get("error"): return None p = d.get("path") if p: return _validate(str(p)) except Exception: pass # Fallback regex : si on ne peut pas parser le JSON, on extrait la # valeur de path mais on valide quand même l'existence du fichier. if "'error'" in tool_message_content or '"error"' in tool_message_content: return None m = re.search(r"['\"]path['\"]\s*:\s*['\"]([^'\"]+)['\"]", tool_message_content) if m: return _validate(m.group(1)) return None def _read_file_preview(path: Optional[str], max_chars: int = 6000) -> str: """Lit le contenu d'un fichier produit, tronqué à `max_chars` pour le preview UI.""" if not path: return "" try: from pathlib import Path text = Path(path).read_text(encoding="utf-8") except Exception as e: return f"⚠️ Impossible de lire {path} : {e}" if len(text) > max_chars: return text[:max_chars] + f"\n\n… [{len(text) - max_chars} caractères supplémentaires non affichés — télécharge le fichier pour tout voir]" return text def submit_existing_file( file_path: Optional[str], drops_key: str, model_name: str, current_chat: Optional[list[dict]] = None, ) -> list[dict]: """Soumet un fichier .enrich/.audit/.err déjà produit au LLMDrops JDM. À utiliser pour le bouton « 📤 Soumettre » post-hoc des sous-onglets Jarvis. Si une clé est fournie côté UI (`drops_key`), elle override temporairement `JDM_DROPS_API_KEY` le temps de l'appel. Renvoie la liste de messages mise à jour pour le `gr.Chatbot` (append d'un message assistant avec le verdict). """ import os from jdm_agent.enrich.uploader import submit_to_jdm chat = list(current_chat) if current_chat else [] if not file_path: chat.append({ "role": "assistant", "content": "⚠️ Aucun fichier produit à soumettre." }) return chat saved = os.environ.get("JDM_DROPS_API_KEY") if drops_key and drops_key.strip(): os.environ["JDM_DROPS_API_KEY"] = drops_key.strip() try: result = submit_to_jdm(file_path, model_name=(model_name or "").strip() or None) finally: if drops_key and drops_key.strip(): if saved is None: os.environ.pop("JDM_DROPS_API_KEY", None) else: os.environ["JDM_DROPS_API_KEY"] = saved if result.get("ok"): chat.append({ "role": "assistant", "content": ( f"✅ Fichier soumis à JDM (status {result.get('status_code')}) — " f"uploadé sous le nom `{result.get('uploaded_as')}`.\n\n" f"Réponse serveur : `{result.get('response')}`" ) }) else: chat.append({ "role": "assistant", "content": f"❌ Échec de soumission : {result.get('error', 'inconnu')}" }) return chat def has_drops_key(ui_key: str = "") -> bool: """True si une clé LLMDrops est disponible (UI override OU env).""" import os if ui_key and ui_key.strip(): return True return bool(os.environ.get("JDM_DROPS_API_KEY", "").strip()) def run_jarvis_flow( prompt: str, *, headline: str = "", model: str, api_key: str, budget_label: str, drops_key: str, build_llm_fn, build_agent_fn, get_client_fn, use_thinking: bool = True, consolidation_target: Optional[int] = None, max_persistence_relances: Optional[int] = None, auto_switch_on_perday: bool = False, resume_state: Optional[dict] = None, ) -> Generator[tuple, None, None]: """Générateur qui pilote un agent avec budget pour un sous-onglet Jarvis, et yield des tuples (messages_chatbot, file_path, file_preview) compatibles avec 3 composants Gradio : - `gr.Chatbot(type="messages")` - `gr.File` - `gr.Code`/`gr.Markdown`/`gr.Textbox` Le messaging modèle : - message 1 : user → headline court (PAS le prompt complet) - message 2 : assistant → contenu progressivement mis à jour pendant le streaming (tool calls + résultats partiels + réponse) `file_path` reste None jusqu'à ce qu'un `write_submission_file` soit détecté dans le stream, puis pointe sur le fichier produit. La 3e valeur est le preview texte (lecture tronquée) ou "" si pas de fichier. Args: prompt : pré-prompt construit par `build_*_prompt` (envoyé au LLM mais NON affiché à l'utilisateur) headline : résumé court 1-ligne affiché dans la bulle « user » model : modèle LLM api_key : clé visiteur (vide si modèle hébergé Space) budget_label : "10" / "25" / "50" / "100" / "illimité" drops_key : clé LLMDrops (override env) build_llm_fn : `_build_llm` (injection pour éviter import circulaire) build_agent_fn: `build_jdm_agent` idem get_client_fn : `get_client` idem Yields: (messages, file_path, file_preview) """ # Bulle user affichée — résumé léger, JAMAIS le prompt technique user_display = headline.strip() or "🚀 Demande envoyée." import os from jdm_agent.tools.budget import budget_context from jdm_agent.enrich.validators import exclusion_context from jdm_agent.enrich import count_consolidations from langchain_core.messages import AIMessage, HumanMessage, ToolMessage def _pending_line() -> str: """Ligne « génération en cours ». Le compteur cumulatif des triplets consolidés est affiché UNIQUEMENT pour l'enrichissement (= seul flow où consolidation_target est défini). Les autres flows (audit, gap, signalement, stats) n'ont pas la sémantique de « consolidation » → on n'expose pas un compteur trompeur. """ if consolidation_target: n = count_consolidations() return f"*⏳ Génération en cours… ({n}/{consolidation_target} consolidés)*" return "*⏳ Génération en cours…*" def _current_file_path() -> Optional[str]: """Renvoie canonical_path dès qu'il existe sur disque (auto-append l'aura créé au 1er triplet consolidé) — priorité absolue car c'est CE fichier qui contient l'historique complet du run en temps réel. Sinon fallback sur last_file_path (path écrit par le LLM via write_submission_file) qui peut être un .audit/.err legitimately différent.""" try: from pathlib import Path as _PathCheck if _PathCheck(canonical_path).exists(): return canonical_path except Exception: pass return last_file_path # Override env var pour LLMDrops si une clé est fournie côté UI saved_drops_key: Optional[str] = None if drops_key and drops_key.strip(): saved_drops_key = os.environ.get("JDM_DROPS_API_KEY") os.environ["JDM_DROPS_API_KEY"] = drops_key.strip() last_file_path: Optional[str] = None try: # LLM + agent — pool Gemini : on track la clé courante pour # pouvoir basculer sur quota PerDay (cf. retry plus bas). current_gemini_key: Optional[str] = None try: # Si modèle Gemini natif, on pick une clé du pool en explicite # pour pouvoir la marquer "blown" plus tard si nécessaire. try: _app = _get_app_module() if _app is None: raise RuntimeError("app module unavailable") GEMINI_MODELS = _app.GEMINI_MODELS pick_unblown_gemini_key = _app.pick_unblown_gemini_key _set_current_key = _app.set_current_gemini_key _set_current_model = _app.set_current_model # Pick pour TOUS les Gemini (3.1, 3.5, 2.5…), pas seulement # NATIVE_REQUIRED. Sinon 2.5 (qui passe par l'endpoint # OpenAI-compat) ne reçoit pas la clé du pool et tombe # sur l'env GOOGLE_API_KEY brut = CSV des 4 clés non # parsé = INVALID_KEY garanti. if model in GEMINI_MODELS: current_gemini_key = pick_unblown_gemini_key(model) # Fallback : pool vide → env GOOGLE_API_KEY (mais # _parse_google_keys → 1ère du CSV, pas CSV brut). if not current_gemini_key: keys = _app._parse_google_keys() if keys: current_gemini_key = keys[0] _set_current_key(current_gemini_key) # Annonce le modèle actif → préfixe ✅ dans le dropdown. _set_current_model(model) except Exception: pass # app pas importable (test mode) → comportement standard llm = build_llm_fn(model, api_key, use_thinking=use_thinking, gemini_key_override=current_gemini_key) except ValueError as e: yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": f"⚠️ {e}"}], None, "", ) return agent = build_agent_fn(client=get_client_fn(), llm=llm) limit = BUDGET_LABEL_TO_LIMIT.get(budget_label, 25) # Deux listes parallèles : # - progress_live : affichée pendant le streaming, thinking tronqué # pour éviter les blocs de texte massifs qui noient l'UI. # - progress_full : version complète sans troncature, montrée à la # FIN dans un
collapsible « Voir le raisonnement ». progress_live: list[str] = ["*🧠 Réflexion en cours…*"] progress_full: list[str] = [] final_answer: str = "" def _add_line(live: str, full: Optional[str] = None) -> None: """Ajoute une ligne aux 2 listes (full = live par défaut).""" progress_live.append(live) progress_full.append(full if full is not None else live) # OPTION C — path canonique unique pour CE run. # Timestamp + 6 premiers chars du hash du prompt → unicité même # si même prompt re-lancé. Tout va dans /tmp/jdm_outputs (la # seule destination fiable sur HF Spaces). # OPTION B — à la fin du flow on dumpe TOUT le registry de # consolidation dans ce path. Garantit que le fichier final # contient toutes les consolidations du run, même si le LLM # a écrit dans plusieurs paths intermédiaires ou si rien. import hashlib import time as _time_mod _ts = _time_mod.strftime("%Y%m%d_%H%M%S") _hash = hashlib.sha1((prompt or "").encode("utf-8")).hexdigest()[:6] canonical_path = f"/tmp/jdm_outputs/jdm_{_ts}_{_hash}.enrich" # Yield initial : user message + assistant placeholder, pas encore de fichier yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], None, "", ) # Retry sur quota PerMinute Gemini : on attend le délai annoncé # par l'API et on CONTINUE le travail en cours — on ne repart # PAS de zéro. Les messages déjà produits (HumanMessage, # AIMessage avec tool_calls, ToolMessage de retour) sont # accumulés et passés tel quel à la nouvelle invocation # agent.stream(), qui reprend là où la précédente s'est arrêtée # (langgraph create_agent supporte le restart par history). # # Le budget_context et exclusion_context vivent À L'EXTÉRIEUR # de la boucle de retry → compteur d'outils et registry # d'exclusion MAINTENUS à travers la pause. import time as _time # Si on est en mode RESUME (l'utilisateur a cliqué « Continuer # avec 3.1 » après un PerDay), on restaure l'état sauvé au lieu # de partir d'un HumanMessage frais. accumulated_messages a déjà # été passé par strip_thinking_blocks dans le state. if resume_state is not None and resume_state.get("accumulated_messages"): accumulated_messages = list(resume_state["accumulated_messages"]) if resume_state.get("progress_live"): progress_live = list(resume_state["progress_live"]) if resume_state.get("progress_full"): progress_full = list(resume_state["progress_full"]) if resume_state.get("last_file_path"): last_file_path = resume_state["last_file_path"] _add_line( "*▶️ Reprise du flow interrompu (PerDay) sur " f"`{model}` — l'agent continue exactement où il s'était arrêté.*" ) else: accumulated_messages = [HumanMessage(content=prompt)] # `persistence_relances` : nombre de relances déjà tentées via # nudge. Fix structurel du biais d'abandon du LLM — si le LLM # finalise prématurément (= consolidés < target alors qu'on a # demandé iterate), on injecte un HumanMessage qui le force à # reprendre, et on relance le streaming. Cap à # `max_persistence_relances` pour éviter les boucles infinies. # NB : chaque relance ouvre un NOUVEAU budget_context (compteur # reset par relance) — c'est volontaire : le budget label est # interprété « par session de streaming », pas « total absolu ». persistence_relances = 0 persistence_done = False budget = None # accessible hors du with pour le compteur final # exclusion_context AUTOUR du while persistance : sans ça le # registry de consolidation est wipé entre chaque relance car # le with exit + re-enter à chaque tour → count_consolidations() # repartirait de 0 et le fix cumulatif serait inopérant. Avec # exclusion_context ici, le registry persiste sur TOUT le run. # budget_context reste dans la boucle (compteur reset par relance # = comportement actuel volontaire, budget interprété par session # de streaming). # Entrée MANUELLE du context manager (__enter__/__exit__) pour # ne pas re-indenter tout le while → fermé dans le finally du try # principal (cf. plus bas). _excl_ctx = exclusion_context() _excl_ctx.__enter__() # Active l'auto-append : chaque register_consolidation écrit # immédiatement la ligne dans canonical_path. L'UI verra le # fichier grossir EN TEMPS RÉEL sans dépendre du LLM appelant # write_submission_file. Désactivé dans le finally. from jdm_agent.enrich import set_consolidation_output_path set_consolidation_output_path(canonical_path) while not persistence_done: rate_limit_attempts = 0 # Hits de rate limit CONSÉCUTIFS sans aucun chunk reçu # entre. Cap dur à 3 = quotas Google croisés bloqués, # inutile de boucler. Reset à 0 dès qu'on reçoit un chunk # (= du vrai progrès LLM s'est produit). consecutive_rate_limit_hits = 0 MAX_CONSECUTIVE_RATE_LIMIT = 3 proactive_condense_count = 0 with budget_context(limit=limit) as budget: # boucle retry quota : ILLIMITÉ tant que le délai # retry est court (cf. detect_rate_limit_retry, cap # interne à 120s par hit) ET qu'on fait du progrès # entre deux hits. Si quotas croisés (3 hits sans # progrès), on tombe en erreur finale. while True: _need_restart_after_condense = False try: for chunk in agent.stream( {"messages": accumulated_messages}, stream_mode="updates", ): # Reset du compteur de rate limit consécutifs : # on a reçu un chunk = vrai progrès LLM, donc # le quota a libéré quelque chose entre temps. consecutive_rate_limit_hits = 0 # chunk = dict {node_name: {"messages": [msg, ...]}} for _node, payload in chunk.items(): msgs = (payload or {}).get("messages") or [] for m in msgs: # Accumulation pour permettre la reprise # après pause quota (cf. retry plus bas). accumulated_messages.append(m) if isinstance(m, AIMessage): tcs = getattr(m, "tool_calls", []) or [] # 1) Chain-of-thought (Anthropic Extended, # Gemini avec include_thoughts, o1/o3). # Style : blockquote + + couleur # grisée + italique pour le distinguer # nettement des outils et du texte parlé, # et signaler son statut « pensée » plutôt # qu'action. Pas de troncature : Gemini # renvoie déjà une SYNTHÈSE côté API # (jamais les raw thoughts), inutile de # re-raboter. thoughts = _content_to_thoughts(m.content) if thoughts.strip(): t = thoughts.strip() # Le thinking contient souvent des # newlines markdown (\n\n) qui REFERMENT # le span/div HTML — d'où le bug observé # où seule la 1re ligne avait le style. # Fix : on convertit tous les retours en #
HTML pour rester inline-block, et # on enveloppe dans un
bloc (les # styles bloc s'appliquent au tout). # Markdown interne au thinking (genre # `code` ou *italique*) ne sera pas # rendu — acceptable pour un bloc déjà # marqué comme « discret ». t_html = ( t.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace("\n", "
") ) line = ( f"
" f"💭 {t_html}
" ) _add_line(line) # 2) Texte parlé entre 2 tool_calls (Claude/ # GPT le font ; Gemini souvent vide). # Blockquote normal pour le distinguer du # thinking (qui est plus discret). spoken = _content_to_text(m.content) if tcs and spoken.strip(): _add_line(f"> 💬 {spoken.strip()}") if tcs: for tc in tcs: name = tc.get("name", "?") tc_args = tc.get("args") or {} narrated = _narrate_tool_call(name, tc_args) if narrated: _add_line( f'
' f'{narrated}
' ) else: args_str = ", ".join( f"{k}={v!r}" for k, v in tc_args.items() ) _add_line( f'
' f'🔧 `{name}({args_str})`
' ) # Ajoute en bas un indicateur fugace # « génération en cours » pour qu'on # sache que ça tourne (le LLM peut # tarder à produire sa réponse finale # ou son prochain tool_call). Cette # ligne disparaît au prochain yield # ou au yield final. live_with_pending = ( "\n\n".join(progress_live) + "\n\n" + _pending_line() ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": live_with_pending}], last_file_path, _read_file_preview(last_file_path), ) else: # Pas de tool_calls → réponse finale final_answer = spoken elif isinstance(m, ToolMessage): content = _content_to_text(m.content) if m.name == "write_submission_file": p = _extract_submission_path(content) if p: last_file_path = p narrated_done = _narrate_tool_result(m.name, content) if narrated_done: _add_line( f'
' f'{narrated_done}
' ) else: preview = content[:120].replace("\n", " ") if len(content) > 120: preview += "…" _add_line( f'
' f'✓ *{m.name}* renvoie {len(content)} chars : `{preview}`' f'
' ) live_with_pending = ( "\n\n".join(progress_live) + "\n\n" + _pending_line() ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": live_with_pending}], _current_file_path(), _read_file_preview(_current_file_path()), ) # FIN DE CHUNK : check si historique # dépasse le seuil → condensation proactive # (mêmes conditions que post-PerMinute : # build_relance_summary + nudge random). # Si condensé, on BREAK le for chunk et on # CONTINUE le while True pour relancer # agent.stream avec les messages condensés. chars_before = _history_total_chars(accumulated_messages) if chars_before > HISTORY_CONDENSE_THRESHOLD_CHARS: condensed = condense_history_with_nudge( accumulated_messages, consolidation_target=consolidation_target, attempt=proactive_condense_count, ) if condensed is not None: proactive_condense_count += 1 accumulated_messages = condensed _add_line( f"*🗜️ Historique condensé " f"({chars_before // 1000}k chars → résumé, " f"relance {proactive_condense_count}) — " f"l'agent reprend avec un nudge frais.*" ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], last_file_path, _read_file_preview(last_file_path), ) _need_restart_after_condense = True break # sort du for chunk if _need_restart_after_condense: continue # relance agent.stream avec accumulated_messages condensé # Sortie normale de la boucle for chunk → quitter while break except Exception as e: # Quota PerMinute Gemini ET premier essai : on attend # le délai, on AFFICHE un message d'attente, et on # CONTINUE le travail en cours — on ne reset PAS les # progress lists, et on relance agent.stream() en # passant les `accumulated_messages` pour que langgraph # reprenne là où il en était (les messages déjà # produits = HumanMessage + AIMessages + ToolMessages). # 1) Quota QUOTIDIEN épuisé. # Si on a un POOL de clés (GOOGLE_API_KEYS CSV), # on marque la clé courante comme blown et on # tente de basculer sur la suivante non-blown. # Sinon (pool vide / toutes blown), on signale # et on stop. # 0) Clé API invalide (typo, révoquée). Marquer # la clé courante comme invalide pour la session # et basculer sur la suivante du pool. Même # logique que PerDay (rebuild LLM + agent + # continue), mais marquage différent (permanent). if is_invalid_api_key(e): # On NE MARQUE INVALIDE que si c'est le modèle # protégé (3.1) qui rejette la clé. Pour les # autres modèles (2.5, 3.5), un INVALID_KEY # peut être trompeur (endpoint OpenAI-compat # qui retourne ce code pour d'autres raisons # — modèle non dispo, version, etc.). Marquer # globalement invalide pourrait gâcher une # clé pourtant valide pour 3.1. _app = _get_app_module() _PROTECTED = (getattr(_app, 'GEMINI_POOL_PROTECTED_MODEL', "gemini-3.1-flash-lite") if _app else "gemini-3.1-flash-lite") if model != _PROTECTED: # Abort + DIAG complet pour pouvoir tracer # honnêtement (cas 2.5 qui renvoie INVALID # alors que clé valide pour 3.1). diag_lines = [ f"⚠️ **`API_KEY_INVALID` pour `{model}`** " f"alors que clé attendue valide pour `{_PROTECTED}`." f"\n\n**Diagnostic du dernier build LLM** :", ] try: db = getattr(_app, '_DEBUG_LAST_BUILD', {}) or {} if not db: diag_lines.append("- *(_DEBUG_LAST_BUILD vide)*") for k, v in db.items(): diag_lines.append(f"- `{k}` : `{v}`") except Exception as _ex: diag_lines.append(f"- *(diag indisponible : {_ex})*") diag_lines.append( "\n**Exception brute Gemini** (premiers 1500 chars) :" ) diag_lines.append(f"```\n{str(e)[:1500]}\n```") diag_lines.append( f"\n➡️ Pour l'instant, bascule sur " f"`{_PROTECTED}` ou un BYOK. La clé n'est " f"PAS marquée invalide globalement." ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n".join(diag_lines)}], last_file_path, _read_file_preview(last_file_path), ) return switched = False try: if _app is None: raise RuntimeError("app module unavailable") mark_gemini_key_invalid = _app.mark_gemini_key_invalid pick_unblown_gemini_key = _app.pick_unblown_gemini_key gemini_pool_size = _app.gemini_pool_size if current_gemini_key: mark_gemini_key_invalid(current_gemini_key) next_key = pick_unblown_gemini_key( model, skip=current_gemini_key ) if next_key: pool_n = gemini_pool_size() current_gemini_key = next_key try: _app.set_current_gemini_key(current_gemini_key) except Exception: pass llm = build_llm_fn( model, api_key, use_thinking=use_thinking, gemini_key_override=current_gemini_key, ) agent = build_agent_fn( client=get_client_fn(), llm=llm ) _add_line( f"*🔑 Clé Google invalide détectée — " f"bascule sur une autre clé du pool " f"(pool : {pool_n} clés).*" ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], last_file_path, _read_file_preview(last_file_path), ) switched = True except Exception: pass if switched: continue # Yield au chatbot avec DIAGNOSTIC : ce qui # a été parsé depuis GOOGLE_API_KEYS (4 chars # début + 4 chars fin + longueur de chaque # clé) pour vérifier que le parsing CSV # n'a rien tronqué. try: _app = _get_app_module() if _app is None: raise RuntimeError("app module unavailable") _parse_keys = _app._parse_google_keys _masked_key = _app._masked_key parsed = _parse_keys() diag = "\n".join( f" - {i+1}. {_masked_key(k)}" for i, k in enumerate(parsed) ) or " (aucune clé parsée)" except Exception: parsed = [] diag = " (diagnostic indisponible)" err_msg = ( "❌ **Toutes les clés Google du pool ont " "échoué**.\n\n" f"**Diagnostic** : {len(parsed)} clé(s) " f"parsée(s) depuis `GOOGLE_API_KEYS` :\n" f"{diag}\n\n" "Vérifie ci-dessus que chaque clé a la " "**longueur attendue (~39 chars)** et " "commence par `AIza`. Si une clé est " "tronquée → problème de parsing CSV.\n\n" "Sinon, causes possibles côté Google :\n" "1. Clés non activées pour l'**API " "Generative Language** (Google Cloud " "Console).\n" "2. Clés d'un projet sans accès aux " "modèles Gemini 3.x.\n" "3. Quotas PerDay tous épuisés (reset " "à minuit UTC).\n\n" "Bascule sur un modèle BYOK Claude / GPT." ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": err_msg}], last_file_path, _read_file_preview(last_file_path), ) return # PerDay : on TRACE TOUJOURS la clé courante # comme blown pour ce modèle (visuel dropdown # « épuisé »). On ne BASCULE de clé que si on # est sur le modèle protégé (gemini-3.1-flash- # lite, 500 req/jour). Pour les autres (quotas # ~20 req/jour), on remonte l'erreur après le # marquage. _app = _get_app_module() if _app is not None: _PROTECTED = getattr(_app, 'GEMINI_POOL_PROTECTED_MODEL', "gemini-3.1-flash-lite") _mark_blown_fn = getattr(_app, 'mark_gemini_key_blown', None) else: _PROTECTED = "gemini-3.1-flash-lite" _mark_blown_fn = None if is_per_day_quota_exhausted(e, expected_model=model): if _mark_blown_fn and current_gemini_key: _mark_blown_fn(current_gemini_key, model) # PerDay sur modèle non-protégé. # Deux modes : # - auto_switch_on_perday=True (option C) : on # bascule silencieusement sur _PROTECTED et # on continue le flow (state préservé via # accumulated_messages, strip_thinking pour # les tokens). # - sinon (option B, défaut) : ABORT, save # state stripped, yield un 5-tuple avec # state + marker → le wrapper affiche un # bouton « Continuer avec 3.1 ». if (model != _PROTECTED and is_per_day_quota_exhausted(e, expected_model=model) and _app is not None): try: _app.set_current_model(_PROTECTED) except Exception: pass if auto_switch_on_perday and current_gemini_key: # Option C : auto-retry silencieux try: accumulated_messages = strip_thinking_blocks( accumulated_messages, keep_last=True ) model = _PROTECTED llm = build_llm_fn( model, api_key, use_thinking=use_thinking, gemini_key_override=current_gemini_key, ) agent = build_agent_fn( client=get_client_fn(), llm=llm ) _add_line( f"*🔄 Quota épuisé — bascule auto " f"sur `{_PROTECTED}`, je continue.*" ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], last_file_path, _read_file_preview(last_file_path), ) continue except Exception: pass # fallback sur abort si erreur # Option B (défaut) : abort + save state + # yield 5-tuple pour activer bouton continuer. try: stripped = strip_thinking_blocks( accumulated_messages, keep_last=True ) except Exception: stripped = accumulated_messages saved_state = { "accumulated_messages": stripped, "progress_full": list(progress_full), "progress_live": list(progress_live), "last_file_path": last_file_path, "user_display": user_display, } switch_msg = ( f"⚠️ **Modèle `{model}` épuisé pour " f"aujourd'hui** (quota quotidien).\n\n" f"Le sélecteur est passé sur " f"`{_PROTECTED}` (500 req/j).\n\n" f"➡️ Clique sur **« ▶️ Continuer avec " f"`{_PROTECTED}` »** pour reprendre EXACTEMENT " f"où l'agent s'est arrêté (state préservé), " f"ou re-clique « Lancer » pour repartir de " f"zéro." ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": switch_msg}], last_file_path, _read_file_preview(last_file_path), saved_state, "show_continue_btn", ) return if (model == _PROTECTED and is_per_day_quota_exhausted(e, expected_model=model)): switched = False try: _app = _get_app_module() if _app is None: raise RuntimeError("app module unavailable") mark_gemini_key_blown = _app.mark_gemini_key_blown pick_unblown_gemini_key = _app.pick_unblown_gemini_key gemini_pool_size = _app.gemini_pool_size if current_gemini_key: mark_gemini_key_blown(current_gemini_key, model) next_key = pick_unblown_gemini_key( model, skip=current_gemini_key ) if next_key: # Rebuild LLM + agent avec la nouvelle clé. pool_n = gemini_pool_size() current_gemini_key = next_key try: _app.set_current_gemini_key(current_gemini_key) except Exception: pass llm = build_llm_fn( model, api_key, use_thinking=use_thinking, gemini_key_override=current_gemini_key, ) agent = build_agent_fn( client=get_client_fn(), llm=llm ) _add_line( f"*🔄 Quota quotidien atteint sur " f"cette clé Google — bascule sur " f"une autre clé du pool " f"(pool : {pool_n} clés).*" ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], last_file_path, _read_file_preview(last_file_path), ) switched = True except Exception: pass # bascule indisponible → on raise comme avant if switched: continue # reprend la boucle avec la nouvelle clé raise RuntimeError( "Quota quotidien Gemini free tier épuisé sur " "TOUTES les clés du pool (ou pool vide). Le " "quota se réinitialise à minuit UTC. Réessaie " "demain ou bascule sur un modèle BYOK " "(Claude / GPT)." ) from e # 2) Rate limit PerMinute → retry avec attente retry_delay = detect_rate_limit_retry(e) if retry_delay is not None: consecutive_rate_limit_hits += 1 # Filet : 3 hits PerMinute consécutifs sans # progrès = quotas glissants croisés bloqués. if consecutive_rate_limit_hits >= MAX_CONSECUTIVE_RATE_LIMIT: raise RuntimeError( f"Quotas Gemini free tier croisés " f"({consecutive_rate_limit_hits} hits " f"PerMinute consécutifs sans progrès). " f"Les fenêtres glissantes ne s'ouvrent " f"jamais en même temps. Réessaie dans " f"quelques minutes ou bascule sur un " f"modèle BYOK (Claude / GPT)." ) from e rate_limit_attempts += 1 wait_msg = ( f"*⏳ Quota Gemini free tier atteint — j'attends " f"{retry_delay:.0f}s puis je CONTINUE le travail " f"en cours (pas de redémarrage).*" ) current_progress = "\n\n".join(progress_live) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": current_progress + "\n\n" + wait_msg}], _current_file_path(), _read_file_preview(_current_file_path()), ) _time.sleep(retry_delay) # PAS de reset des progress / last_file_path. # MAIS strip des blocs thinking pour réduire # massivement les tokens ré-envoyés (le LLM # n'a pas besoin de ses propres pensées pour # continuer — juste des tool_calls et résultats). # On garde le DERNIER thinking pour préserver # le thought_signature Gemini 3.x. accumulated_messages = strip_thinking_blocks( accumulated_messages, keep_last=True ) # Condensation proactive si APRÈS strip # l'historique reste massif (>seuil). Le # helper renvoie None si pas nécessaire, # ou la nouvelle liste [initial, summary # + nudge random] sinon. Logique identique # à la condensation proactive en cours de # streaming (cf. fin du for chunk). chars_before = _history_total_chars(accumulated_messages) condensed = condense_history_with_nudge( accumulated_messages, consolidation_target=consolidation_target, attempt=rate_limit_attempts, ) if condensed is not None: accumulated_messages = condensed _add_line( f"*🗜️ Historique condensé " f"({chars_before // 1000}k chars → résumé) — " f"l'agent reprend avec un nudge frais.*" ) # Yield immédiat — sans ça la ligne n'est # poussée à Gradio QU'AU PROCHAIN chunk, # qui peut tarder (PerMinute en boucle) → # l'utilisateur ne voyait jamais le message # condensé, juste « j'attends Xs ». yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], last_file_path, _read_file_preview(last_file_path), ) continue # Pas un quota retryable, ou déjà tenté : erreur finale err_block = "" if progress_full: err_block = ( f"\n\n
🧠 Voir les étapes avant erreur " f"({len(progress_full)})\n\n" f"{(chr(10)*2).join(progress_full)}\n\n
" ) # PRÉSERVATION DU FICHIER SUR ERREUR : on yield # _current_file_path() (priorité canonical_path # si auto-append a écrit quelque chose) pour # que l'utilisateur garde son fichier de # consolidation même quand l'API LLM crashe. yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": f"❌ Erreur agent : {e}" + err_block}], _current_file_path(), _read_file_preview(_current_file_path()), ) return # Sortie normale du with → check persistance. # Si le LLM a finalisé prématurément (consolidés < target) # et qu'on a encore des relances disponibles, on injecte un # nudge et on relance le with (= nouveau budget_context, # mais accumulated_messages conservé donc l'agent reprend # avec tout son contexte). if consolidation_target is None: persistence_done = True continue # Source de vérité = registry GLOBAL des consolidations # (cumulatif depuis l'entrée dans exclusion_context, survit # aux RESET de accumulated_messages opérés par les relances # persistance). count_consolidated_in_messages() était # défaillant ici car il ne voyait que les ToolMessages du # tour COURANT — chaque relance était comptée from scratch # → boucle infinie possible si le LLM consolide < target # par tour. Cf. bug observé : « il a déjà ses 15 mais il # pense être à deux ». from jdm_agent.enrich import count_consolidations n_done = count_consolidations() if n_done >= consolidation_target: persistence_done = True continue # Pas de cap dur sur les relances persistance — on continue # tant que le LLM finalise sans avoir atteint le target. # Si l'utilisateur veut un cap, il passe max_persistence_relances # (par défaut None = illimité). if max_persistence_relances is not None and persistence_relances >= max_persistence_relances: persistence_done = True continue # On relance avec un nudge fort. persistence_relances += 1 # Construit un résumé condensé (consolidés / échecs / pré-fetchs) # à partir des accumulated_messages, PUIS reset à juste : # [HumanMessage initial, HumanMessage du résumé+nudge] # → drop massif des tokens (de ~50k à ~2k typiquement). # Le LLM reprend frais avec un état explicite plutôt que de # devoir digérer 50+ messages avec leurs raisonnements. summary = build_relance_summary( accumulated_messages, n_done, consolidation_target, persistence_relances, max_persistence_relances, ) initial_human = accumulated_messages[0] # HumanMessage(prompt) accumulated_messages = [ initial_human, HumanMessage(content=summary), ] _cap_label = ( f"/{max_persistence_relances}" if max_persistence_relances is not None else "" ) _add_line( f"*🔁 Relance automatique {persistence_relances}{_cap_label} — " f"{n_done}/{consolidation_target} consolidés, on continue.*" ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": "\n\n".join(progress_live)}], last_file_path, _read_file_preview(last_file_path), ) # Boucle continue → nouveau with budget_context + nouvelle # invocation agent.stream avec accumulated_messages enrichi. # Réponse finale : on remplace les progress_lines par la réponse # définitive du modèle, suivie d'un footer avec compteur (limite # n'est mentionnée que si elle est bornée). n = budget.count if budget.limit: footer = ( f"\n\n---\n*Budget : {n} appel{'s' if n > 1 else ''} " f"d'outils consommé{'s' if n > 1 else ''} / {budget.limit}.*" ) if budget.exhausted: footer += " ⚠️ **Budget atteint** — relance avec un budget plus large si besoin." else: footer = ( f"\n\n---\n*Budget illimité — {n} appel{'s' if n > 1 else ''} " f"d'outils consommé{'s' if n > 1 else ''}.*" ) # Bloc collapsible
avec la trace complète : résumé de # raisonnement (le « thought summary » de Gemini, déjà condensé # côté API — pas de version raw exposée) + texte parlé + # tool_calls + retours, dans l'ordre chronologique. Replié par # défaut pour ne pas polluer la réponse. # # Libellé adaptatif : # - raisonnement ON → « Voir le résumé du raisonnement (N étapes) » # - raisonnement OFF → « Voir les étapes (N étapes) » # (pas de raisonnement LLM dans le bloc, juste la narration des # appels d'outils et leurs résultats) reasoning_block = "" if progress_full: full_text = "\n\n".join(progress_full) n_steps = len(progress_full) plural = "s" if n_steps > 1 else "" if use_thinking: summary_label = ( f"🧠 Voir le résumé du raisonnement " f"({n_steps} étape{plural})" ) else: summary_label = f"🧠 Voir les étapes ({n_steps} étape{plural})" reasoning_block = ( f"\n\n
{summary_label}\n\n" f"{full_text}\n\n
" ) # OPTION B — FUSION FINALE depuis le registry de consolidation. # On dumpe TOUS les triplets consolidés du run (cumulatif via le # registry survivant grâce à exclusion_context wrappant le while # persistance) dans canonical_path. Garantit que le fichier # affiché contient TOUT, peu importe ce que le LLM a écrit dans # des paths intermédiaires ou si certaines écritures ont été # écrasées par des appels successifs. try: from jdm_agent.enrich import list_consolidations from jdm_agent.enrich.pipeline import write_submission as _write_sub from jdm_agent.enrich import Candidate as _Candidate from pathlib import Path as _Path entries = list_consolidations() if entries: _Path(canonical_path).parent.mkdir(parents=True, exist_ok=True) cands = [ _Candidate( term=e["term"], relation=e["relation"], target=e["target"], annotation="", consolidation_explanation=e.get("explanation") or "", confidence=0.8, source="agent", validation_status="ok", consolidation_status="consolidated", ) for e in entries ] _write_sub(canonical_path, cands, client=get_client_fn()) last_file_path = canonical_path _add_line( f"*📦 Fichier final fusionné : {len(entries)} triplets " f"consolidés écrits dans `{canonical_path}` " f"(garantit que rien n'est perdu).*" ) except Exception as _e: # Safety : si la fusion finale foire, on ne casse pas le flow, # on garde last_file_path tel que la dernière écriture LLM # l'avait laissé. _add_line( f"*⚠️ Fusion finale impossible : {_e}. Le fichier affiché " f"correspond à la dernière écriture du LLM.*" ) final_content = ( (final_answer or "*(réponse vide)*") + footer + reasoning_block ) yield ( [{"role": "user", "content": user_display}, {"role": "assistant", "content": final_content}], _current_file_path(), _read_file_preview(_current_file_path()), ) finally: # Désactive l'auto-append (path persistait globalement) try: from jdm_agent.enrich import set_consolidation_output_path set_consolidation_output_path(None) except Exception: pass # Ferme manuellement l'exclusion_context ouvert avant le # while persistance (cf. __enter__ plus haut). Try/except pour # supporter le cas où _excl_ctx n'a pas été initialisé (erreur # tres precoce dans le run). try: _excl_ctx.__exit__(None, None, None) except Exception: pass # Restore env var si on l'avait modifiée if drops_key and drops_key.strip(): if saved_drops_key is None: os.environ.pop("JDM_DROPS_API_KEY", None) else: os.environ["JDM_DROPS_API_KEY"] = saved_drops_key