""" prompt_injection.py — Sliding Pipeline · Pernod Ricard ======================================================= Génère les blocs de contraintes injectés dans les prompts des agents. Ce script lit les 3 YAML (theme, components, layouts) et produit : 1. agent_constraints.md — fichier de référence complet (debug/doc) 2. prompt_the_designer_injected.md — prompt Designer avec blocs remplis 3. prompt_the_encoder_injected.md — prompt Encoder avec schémas remplis Usage : python prompt_injection.py python prompt_injection.py --theme theme.yaml --components components.yaml --layouts layouts.yaml --designer prompt_the_designer.md --encoder prompt_the_encoder.md Les fichiers injectés sont prêts à être copiés dans Mistral Studio. """ import argparse import os import sys from pathlib import Path import yaml # ───────────────────────────────────────────────────────────────────────────── # DONNÉES ÉDITORIALES (non générables depuis les YAML — savoirs métier) # ───────────────────────────────────────────────────────────────────────────── SEQUENCING_RULES = """### Règles de fluidité - Maximum 2 layouts texte consécutifs : `default_bullets`, `two_cols_text` - Maximum 2 tableaux consécutifs : `comparison_table`, `benchmark`, `raci_table`, `boxes_grid` - Après 3 slides denses → intercaler une respiration : `big_stat`, `key_message`, `section_divider` - Jamais 2 `section_divider` consécutifs - Le dernier slide de contenu avant `end_slide` doit être narratif (pas un tableau, pas un gantt) ### Règles de choix de layout - 1 seul chiffre décisif → `big_stat` (jamais `kpi_grid` avec 1 item) - 2 à 6 indicateurs chiffrés → `kpi_grid` - Transformation conceptuelle (avant/après) → `from_to` (pas `two_cols_text`) - Plus de 4 étapes avec timing → `process_arrow` ou `phases_timeline` (pas `numbered_steps`) - Recommandation unique et précise → `recommendation_card` - Données à comparer sur plusieurs critères avec plusieurs acteurs → `benchmark` - Données à comparer en tableau structuré → `comparison_table` - Si le contenu ne rentre dans aucun layout spécialisé → `default_bullets` - Citation ou message à marteler seul → `key_message` - Organigramme de gouvernance → `org_chart` - Responsabilités par rôle → `raci_table` ### Règles de quantité - Présentation 20 min → 10 à 15 slides max - Présentation 10 min → 6 à 10 slides max - Fusionner si > 15 slides : regrouper les slides proches thématiquement - Un `###` du Markdown avec un seul chiffre fort → envisager `big_stat` séparé - Un `##` du Markdown = 1 `section_divider` (sauf présentation < 6 slides)""" NARRATIVE_PATTERNS = """### Pattern "Problem → Solution → Proof" Adapté aux présentations de recommandation stratégique. ``` cover_split executive_summary (synthèse SCR dès le début) section_divider ("Le problème") big_stat (chiffre choc) default_bullets ou comparison_table section_divider ("Notre réponse") from_to ou numbered_steps kpi_grid ou chart_callout (preuve que ça marche) recommendation_card (ce qu'on demande) end_slide ``` ### Pattern "Roadmap Deck" Adapté aux présentations de planification / lancement de projet. ``` cover_split executive_summary (où on va et pourquoi) kpi_grid (état des lieux chiffré) phases_timeline ou gantt_timeline numbered_steps (comment on s'organise) org_chart ou raci_table (qui fait quoi) recommendation_card (décisions à prendre) end_slide ``` ### Pattern "Data Storytelling" Adapté aux présentations de revue de performance ou data governance. ``` cover_split big_stat (chiffre choc d'entrée) default_bullets (contexte et enjeux) kpi_grid (panorama des indicateurs) chart_callout (analyse d'un graphique clé) from_to (implication / transformation attendue) yearly_timeline (historique ou prospective) end_slide ``` ### Pattern "Executive Briefing" Adapté aux présentations courtes (< 10 slides) pour un CODIR. ``` cover_split executive_summary key_message (le So What en 1 slide) kpi_grid ou big_stat recommendation_card end_slide ``` ### Règles d'assemblage des patterns - Les patterns sont des points de départ, pas des contraintes rigides - Hybrider 2 patterns est possible si le contenu le justifie - Toujours préserver : cover_split en premier, end_slide en dernier - Les section_dividers sont optionnels pour les patterns courts (< 8 slides)""" # ───────────────────────────────────────────────────────────────────────────── # GÉNÉRATEURS DE BLOCS # ───────────────────────────────────────────────────────────────────────────── def generate_layouts_catalogue(layouts: dict) -> str: """ Génère le bloc LAYOUTS_CATALOGUE pour le prompt du Designer. Format compact, optimisé pour la lecture LLM. """ lines = [] current_famille = None for layout_name, cfg in layouts.items(): famille = cfg.get("famille", "") if famille != current_famille: lines.append(f"\n#### {famille}") current_famille = famille layout_id = cfg.get("id", "") roles = " / ".join(cfg.get("roles_narratifs", [])) hint = cfg.get("agent_hint", "").strip().replace("\n", " ").replace(" ", " ") # Tronquer le hint à 120 caractères pour rester compact if len(hint) > 120: hint = hint[:117] + "..." lines.append(f"\n`{layout_name}` ({layout_id}) — Rôles : {roles}") lines.append(f" → {hint}") return "\n".join(lines) def generate_yaml_schemas(layouts: dict) -> str: """ Génère le bloc YAML_SCHEMAS pour le prompt de l'Encoder. Pour chaque layout : champs requis / optionnels avec types et contraintes. """ lines = [] for layout_name, cfg in layouts.items(): layout_id = cfg.get("id", "") schema = cfg.get("json_schema", {}) constraints = cfg.get("constraints", {}) if not schema: continue lines.append(f"\n### `{layout_name}` ({layout_id})") # Champs requis required = schema.get("required", []) if required: lines.append(f"**Requis :** {', '.join(required)}") # Champs optionnels optional = schema.get("optional", []) if optional: lines.append(f"**Optionnels :** {', '.join(optional)}") # Contraintes importantes constraint_lines = [] for key, val in constraints.items(): if isinstance(val, dict): # Schéma imbriqué — on extrait les infos clés sub_req = val.get("required", []) sub_opt = val.get("optional", []) sub_min = val.get("min_items") or val.get(f"{key}_min") sub_max = val.get("max_items") or val.get(f"{key}_max") if sub_req or sub_opt: parts = [] if sub_req: parts.append(f"requis: {', '.join(str(x) for x in sub_req)}") if sub_opt: parts.append(f"optionnels: {', '.join(str(x) for x in sub_opt)}") constraint_lines.append(f" {key} : {' | '.join(parts)}") elif key.endswith("_min") or key.endswith("_max"): field = key.rsplit("_", 1)[0] bound = key.rsplit("_", 1)[1] constraint_lines.append(f" {field} : {bound} = {val}") elif key.endswith("_max_chars"): field = key.replace("_max_chars", "") constraint_lines.append(f" {field} : max {val} caractères") elif key == "items_min": constraint_lines.append(f" items : min {val}") elif key == "items_max": constraint_lines.append(f" items : max {val}") if constraint_lines: lines.append("**Contraintes :**") lines.extend(constraint_lines) # Exemple YAML minimal lines.append("**Exemple minimal :**") lines.append("```yaml") lines.append(f"layout: {layout_name}") lines.append(f'titre: "Votre titre affirmatif"') # Génère quelques champs d'exemple selon le layout example_fields = _generate_example_fields(layout_name, required, constraints) lines.extend(example_fields) lines.append("```") return "\n".join(lines) def _generate_example_fields(layout_name: str, required: list, constraints: dict) -> list: """Génère des champs d'exemple YAML pour un layout donné.""" examples = { "cover_split": ['sous_titre: "Présentation au CODIR — juin 2026"'], "section_divider": ["numero_section: 1"], "agenda": [ "items:", ' - numero: 1', ' titre: "Contexte et enjeux"', ' - numero: 2', ' titre: "Notre proposition"', ], "content_marker": ["current_index: 2"], "end_slide": [ 'message: "Merci pour votre attention"', "next_steps:", ' - texte: "Valider le modèle — juillet"', ' niveau: 1', ], "default_bullets": [ "bullets:", ' - texte: "Premier argument clé"', " niveau: 1", ' - texte: "Détail ou preuve"', " niveau: 2", ], "two_cols_text": [ "left:", ' titre: "Titre colonne gauche"', ' contenu: "Texte de la colonne gauche..."', "right:", ' titre: "Titre colonne droite"', ' contenu: "Texte de la colonne droite..."', ], "key_message": ['message: "Le message clé en une phrase forte."'], "executive_summary": [ 'situation: "État des lieux factuel..."', 'complication: "Le problème ou la tension..."', 'resolution: "La réponse proposée..."', ], "kpi_grid": [ "items:", ' - titre: "Indicateur 1"', ' valeur: "85%"', ' sous_titre: "Contexte de la valeur"', ' - titre: "Indicateur 2"', ' valeur: "+25%"', ], "big_stat": [ 'valeur: "2 400"', 'label: "jours/homme de réconciliation par an"', 'source: "Estimation interne 2026"', ], "comparison_table": [ "headers:", ' - "Critère"', ' - "Option A"', ' - "Option B"', "rows:", ' - ["Coût", "Élevé", "Moyen"]', ' - ["Délai", "3 mois", "6 mois"]', ], "chart_callout": [ "chart_type: bar", "data:", ' - label: "T1"', " valeur: 40", ' - label: "T2"', " valeur: 65", 'insight: "La croissance s\'accélère au T2 grâce au pilote."', ], "benchmark": [ "criteria:", ' - "Coût"', ' - "Délai"', "actors:", ' - "PR"', ' - "Concurrent A"', "scores:", " - [80, 60]", " - [70, 85]", ], "matrix_2x2": [ "axis_x:", ' label: "Effort"', "axis_y:", ' label: "Impact"', "items:", ' - label: "Initiative A"', " x: 20", " y: 80", " taille: 3", ], "pyramid": [ "levels:", ' - label: "Vision"', ' description: "Callout explicatif optionnel"', ' - label: "Stratégie"', ' - label: "Opérations"', ], "circular_diagram": [ "segments:", ' - label: "Segment 1"', ' description: "Description courte"', ' - label: "Segment 2"', ' description: "Description courte"', ' - label: "Segment 3"', ' description: "Description courte"', ], "from_to": [ "pairs:", ' - from: "Situation actuelle"', ' to: "Situation cible"', ' - from: "Processus manuel"', ' to: "Processus automatisé"', ], "boxes_grid": [ "columns:", ' - "Colonne 1"', ' - "Colonne 2"', "rows:", " - label: \"Ligne A\"", ' kpi: "xx%"', ' contents: ["Contenu 1", "Contenu 2"]', ], "numbered_steps": [ "steps:", " - numero: 1", ' titre: "Première étape"', ' description: "Ce que ça implique concrètement"', " - numero: 2", ' titre: "Deuxième étape"', ], "process_arrow": [ "phases:", ' - label: "Phase 1"', ' duree: "Juin"', " actif: false", ' bullets: ["Livrable A", "Livrable B"]', ' - label: "Phase 2"', ' duree: "Juil-Sept"', " actif: true", ], "gantt_timeline": [ "period:", ' start: "2026-06"', ' end: "2026-12"', "workstreams:", ' - label: "Workstream 1"', " tasks:", ' - start: "2026-06"', ' end: "2026-08"', ], "yearly_timeline": [ "milestones:", ' - annee: "2024"', ' label: "Lancement du projet"', ' - annee: "2025"', ' label: "Pilote Suède"', " actif: true", ' - annee: "2026"', ' label: "Déploiement nordique"', ], "phases_timeline": [ "phases:", ' - label: "PREP"', ' periode: "Juin"', ' items: ["Brief équipe", "Setup outil"]', ' - label: "PROD"', ' periode: "Juil-Oct"', ' items: ["Développement", "Tests"]', ], "org_chart": [ "root:", ' label: "Data Gov Leader"', " children:", ' - label: "Data Owner Finance"', ' children:', ' - label: "Data Steward"', ' - label: "Data Owner Supply"', ], "raci_table": [ "roles:", ' - "Data Owner"', ' - "Data Steward"', ' - "IT"', "tasks:", ' - label: "Définir les règles qualité"', ' raci: ["A", "R", "C"]', ' - label: "Exécuter les contrôles"', ' raci: ["A", "R", "I"]', ], "decision_tree": [ 'question: "Faut-il déployer le pilote en Suède ?"', "branches:", " yes:", ' label: "Engagement DG confirmé"', ' options: ["Démarrer en juin", "Allouer 0.5 ETP"]', " no:", ' label: "Engagement DG manquant"', ' options: ["Reporter à septembre", "Choisir une autre filiale"]', ], "recommendation_card": [ "numero: 1", ' headline: "TROIS DÉCISIONS AVANT FIN JUIN"', "bullets:", ' - texte: "Valider le modèle avec les DG locaux"', " niveau: 1", ' - texte: "Nommer les Data Owners"', " niveau: 1", ' - texte: "Allouer 0.5 ETP par filiale"', " niveau: 1", 'cta: "Décider en réunion du 30 juin"', ], } return examples.get(layout_name, []) # ───────────────────────────────────────────────────────────────────────────── # GÉNÉRATION DES FICHIERS # ───────────────────────────────────────────────────────────────────────────── def generate_agent_constraints(layouts: dict, output_path: str): """Génère agent_constraints.md — fichier de référence complet.""" lines = [ "# agent_constraints.md", "# Généré automatiquement par prompt_injection.py", "# Ne pas modifier manuellement.\n", "=" * 70, "## SECTION A — CATALOGUE DES LAYOUTS (pour The Designer)", "=" * 70, generate_layouts_catalogue(layouts), "\n" + "=" * 70, "## SECTION B — RÈGLES DE SÉQUENÇAGE (pour The Designer)", "=" * 70, SEQUENCING_RULES, "\n" + "=" * 70, "## SECTION C — PATTERNS NARRATIFS (pour The Designer)", "=" * 70, NARRATIVE_PATTERNS, "\n" + "=" * 70, "## SECTION D — SCHÉMAS YAML (pour The Encoder)", "=" * 70, generate_yaml_schemas(layouts), ] with open(output_path, "w", encoding="utf-8") as f: f.write("\n".join(lines)) print(f"✓ agent_constraints.md généré : {output_path}") def inject_prompt(template_path: str, layouts: dict, output_path: str): """ Lit un template de prompt, remplace les placeholders {{...}} et écrit le prompt injecté. """ with open(template_path, encoding="utf-8") as f: template = f.read() replacements = { "{{LAYOUTS_CATALOGUE}}": generate_layouts_catalogue(layouts), "{{SEQUENCING_RULES}}": SEQUENCING_RULES, "{{NARRATIVE_PATTERNS}}": NARRATIVE_PATTERNS, "{{YAML_SCHEMAS}}": generate_yaml_schemas(layouts), } injected = template for placeholder, content in replacements.items(): injected = injected.replace(placeholder, content) with open(output_path, "w", encoding="utf-8") as f: f.write(injected) # Vérifie qu'il ne reste pas de placeholders non résolus remaining = [p for p in replacements if p in injected] if remaining: print(f" ⚠ Placeholders non résolus dans {output_path} : {remaining}") else: print(f"✓ Prompt injecté : {output_path}") # ───────────────────────────────────────────────────────────────────────────── # RAPPORT DE COHÉRENCE # ───────────────────────────────────────────────────────────────────────────── def check_coherence(layouts: dict, components: dict): """Vérifie la cohérence entre layouts et components.""" print("\n── Rapport de cohérence ──────────────────────────────────") layout_to_components = components.get("layout_to_components", {}) all_comp_ids = {v["id"] for v in components.get("components", {}).values()} errors = 0 for layout_name in layouts: if layout_name not in layout_to_components: print(f" ⚠ Layout '{layout_name}' absent de layout_to_components") errors += 1 else: comp_ids = layout_to_components[layout_name] for cid in comp_ids: if cid not in all_comp_ids: print(f" ⚠ Composant '{cid}' (layout {layout_name}) introuvable") errors += 1 if errors == 0: print(f" ✓ {len(layouts)} layouts × {len(all_comp_ids)} composants — aucune erreur") else: print(f" ✗ {errors} erreur(s) détectée(s)") print() # ───────────────────────────────────────────────────────────────────────────── # CLI # ───────────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Sliding prompt_injection — Génère les prompts enrichis des agents") parser.add_argument("--theme", default="theme.yaml") parser.add_argument("--components", default="components.yaml") parser.add_argument("--layouts", default="layouts.yaml") parser.add_argument("--designer", default="prompt_the_designer.md") parser.add_argument("--encoder", default="prompt_the_encoder.md") parser.add_argument("--output-dir", default=".", help="Dossier de sortie des fichiers générés") args = parser.parse_args() # Vérification des fichiers sources for path in [args.theme, args.components, args.layouts, args.designer, args.encoder]: if not os.path.exists(path): print(f"✗ Fichier introuvable : {path}") sys.exit(1) out = Path(args.output_dir) out.mkdir(parents=True, exist_ok=True) # Chargement des YAML with open(args.theme, encoding="utf-8") as f: theme = yaml.safe_load(f) with open(args.components, encoding="utf-8") as f: components_data = yaml.safe_load(f) with open(args.layouts, encoding="utf-8") as f: layouts_data = yaml.safe_load(f) layouts = layouts_data["layouts"] components = components_data print(f"\n── Sliding prompt_injection ──────────────────────────────") print(f" Layouts : {len(layouts)}") print(f" Composants: {len(components.get('components', {}))}") print() # Rapport de cohérence check_coherence(layouts, components) # Génération des fichiers generate_agent_constraints( layouts, str(out / "agent_constraints.md") ) inject_prompt( args.designer, layouts, str(out / "prompt_the_designer_injected.md") ) inject_prompt( args.encoder, layouts, str(out / "prompt_the_encoder_injected.md") ) print() print("── Fichiers générés ──────────────────────────────────────") for f in ["agent_constraints.md", "prompt_the_designer_injected.md", "prompt_the_encoder_injected.md"]: path = out / f if path.exists(): size = path.stat().st_size print(f" {f} ({size:,} octets)") print() print("✓ Injection terminée. Copiez les prompts _injected.md") print(" dans le champ 'Instructions' de chaque agent Mistral Studio.") if __name__ == "__main__": main()