614 lines
24 KiB
Python
614 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
facilitator.py — Sliding Pipeline v3 · Pernod Ricard
|
||
=====================================================
|
||
Orchestre les 3 agents du nouveau pipeline :
|
||
|
||
The Narrator (Agent 1) → Markdown narratif
|
||
↓ approbation Bastien
|
||
The Designer (Agent 2) → Plan de présentation texte
|
||
↓ approbation optionnelle
|
||
The Encoder (Agent 3) → YAML valide
|
||
↓
|
||
render_engine.py → PPTX
|
||
|
||
Variables .env requises :
|
||
MISTRAL_API_KEY
|
||
NARRATOR_AGENT_ID
|
||
DESIGNER_AGENT_ID
|
||
ENCODER_AGENT_ID
|
||
OUTPUT_DIR (dossier de sortie des YAML/PPTX, défaut : ./output)
|
||
THEME_PATH (défaut : theme.yaml)
|
||
COMPONENTS_PATH (défaut : components.yaml)
|
||
LAYOUTS_PATH (défaut : layouts.yaml)
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import textwrap
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
import yaml
|
||
from dotenv import load_dotenv
|
||
from typing import Optional, Tuple, Dict, Union, List
|
||
|
||
load_dotenv()
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# CONFIGURATION
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
API_KEY = os.getenv("MISTRAL_API_KEY")
|
||
NARRATOR_ID = os.getenv("NARRATOR_AGENT_ID")
|
||
DESIGNER_ID = os.getenv("DESIGNER_AGENT_ID")
|
||
ENCODER_ID = os.getenv("ENCODER_AGENT_ID")
|
||
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "./output"))
|
||
THEME_PATH = os.getenv("THEME_PATH", "theme.yaml")
|
||
COMPONENTS_PATH = os.getenv("COMPONENTS_PATH", "components.yaml")
|
||
LAYOUTS_PATH = os.getenv("LAYOUTS_PATH", "layouts.yaml")
|
||
|
||
HEADERS = {
|
||
"Authorization": f"Bearer {API_KEY}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
VERSION = "3.0"
|
||
|
||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# UTILITAIRES D'AFFICHAGE
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def banner():
|
||
print("\n" + "═" * 62)
|
||
print(f" SLIDING PIPELINE v{VERSION} — Pernod Ricard")
|
||
print(f" {datetime.now().strftime('%d/%m/%Y %H:%M')}")
|
||
print("═" * 62 + "\n")
|
||
|
||
|
||
def section(title: str):
|
||
print(f"\n{'─' * 62}")
|
||
print(f" {title}")
|
||
print(f"{'─' * 62}\n")
|
||
|
||
|
||
def info(msg: str):
|
||
print(f" ℹ {msg}")
|
||
|
||
|
||
def ok(msg: str):
|
||
print(f" ✓ {msg}")
|
||
|
||
|
||
def warn(msg: str):
|
||
print(f" ⚠ {msg}")
|
||
|
||
|
||
def ask(prompt: str) -> str:
|
||
try:
|
||
return input(f"\n {prompt} ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\n Session interrompue.")
|
||
sys.exit(0)
|
||
|
||
|
||
def display_markdown(text: str, max_lines: int = 60):
|
||
"""Affiche le Markdown avec un max de lignes."""
|
||
lines = text.splitlines()
|
||
if len(lines) <= max_lines:
|
||
print(textwrap.indent(text, " "))
|
||
else:
|
||
preview = "\n".join(lines[:max_lines])
|
||
print(textwrap.indent(preview, " "))
|
||
print(f"\n ... [{len(lines) - max_lines} lignes supplémentaires — "
|
||
f"choisissez [4] pour tout voir]")
|
||
|
||
|
||
def display_plan(text: str, max_lines: int = 80):
|
||
"""Affiche le plan du Designer — extrait les lignes SLIDE pour un résumé rapide."""
|
||
slides = [l for l in text.splitlines() if l.strip().startswith("SLIDE")]
|
||
if slides:
|
||
print(" Résumé du plan :\n")
|
||
for s in slides:
|
||
print(f" {s.strip()}")
|
||
print(f"\n ({len(slides)} slides au total)")
|
||
else:
|
||
display_markdown(text, max_lines)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# APPEL AGENT MISTRAL (avec gestion PAUSE/continue)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def call_agent(agent_id: str, messages: list) -> str:
|
||
"""
|
||
Appelle un agent Mistral et gère la pagination PAUSE/continue.
|
||
Retourne la réponse complète concaténée.
|
||
"""
|
||
full_response = ""
|
||
current_messages = messages.copy()
|
||
|
||
while True:
|
||
resp = requests.post(
|
||
"https://api.mistral.ai/v1/agents/completions",
|
||
headers=HEADERS,
|
||
json={"agent_id": agent_id, "messages": current_messages},
|
||
timeout=120,
|
||
)
|
||
resp.raise_for_status()
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
full_response += content
|
||
|
||
# Détection pagination
|
||
if "PAUSE" in content and "FIN —" not in content:
|
||
info("Pagination détectée → continuation automatique...")
|
||
current_messages.append({"role": "assistant", "content": content})
|
||
current_messages.append({"role": "user", "content": "continue"})
|
||
else:
|
||
break
|
||
|
||
return full_response
|
||
|
||
|
||
def call_with_feedback(agent_id: str, initial_message: str,
|
||
feedback: str, previous_output: str) -> str:
|
||
"""Relance un agent avec le feedback utilisateur et la sortie précédente."""
|
||
messages = [
|
||
{"role": "user", "content": initial_message},
|
||
{"role": "assistant", "content": previous_output},
|
||
{"role": "user", "content": feedback},
|
||
]
|
||
return call_agent(agent_id, messages)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# VALIDATION YAML (The Encoder)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def extract_yaml(raw: str) -> str:
|
||
"""
|
||
Extrait le bloc YAML d'une réponse (enlève les ```yaml ... ``` si présents).
|
||
"""
|
||
# Bloc markdown yaml
|
||
match = re.search(r"```ya?ml\s*(.*?)```", raw, re.DOTALL | re.IGNORECASE)
|
||
if match:
|
||
return match.group(1).strip()
|
||
# Pas de bloc → retourne tel quel
|
||
return raw.strip()
|
||
|
||
|
||
def validate_yaml(raw: str, layouts: dict) -> tuple[bool, str, Optional[Dict]]:
|
||
"""
|
||
Valide la syntaxe YAML et vérifie les contraintes de base.
|
||
Retourne (is_valid, message_erreur, data_parsée).
|
||
"""
|
||
try:
|
||
content = extract_yaml(raw)
|
||
data = yaml.safe_load(content)
|
||
except yaml.YAMLError as e:
|
||
return False, f"Erreur de syntaxe YAML : {e}", None
|
||
|
||
if not isinstance(data, dict):
|
||
return False, "Le YAML doit être un dictionnaire à la racine.", None
|
||
|
||
slides = data.get("slides")
|
||
if not slides:
|
||
return False, "Clé 'slides' manquante ou vide.", None
|
||
|
||
errors = []
|
||
valid_layouts = set(layouts.keys())
|
||
|
||
for i, slide in enumerate(slides):
|
||
pos = slide.get("position", i + 1)
|
||
layout = slide.get("layout", "")
|
||
|
||
if not layout:
|
||
errors.append(f"Slide {pos} : clé 'layout' manquante")
|
||
continue
|
||
|
||
if layout not in valid_layouts:
|
||
errors.append(f"Slide {pos} : layout '{layout}' inconnu "
|
||
f"(valides : {', '.join(sorted(valid_layouts)[:5])}...)")
|
||
continue
|
||
|
||
# Vérification des champs requis
|
||
schema = layouts[layout].get("json_schema", {})
|
||
required_fields = schema.get("required", [])
|
||
for field in required_fields:
|
||
if field not in slide:
|
||
errors.append(f"Slide {pos} ({layout}) : champ requis '{field}' manquant")
|
||
|
||
if errors:
|
||
return False, "\n ".join(errors), data
|
||
|
||
return True, f"{len(slides)} slides valides.", data
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# SAUVEGARDE
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def save_output(content: str, suffix: str, ext: str) -> Path:
|
||
"""Sauvegarde un fichier de sortie avec timestamp."""
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
filename = OUTPUT_DIR / f"sliding_{timestamp}_{suffix}.{ext}"
|
||
filename.write_text(content, encoding="utf-8")
|
||
ok(f"Sauvegardé : {filename}")
|
||
return filename
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# CHARGEMENT DES LAYOUTS (pour validation)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def load_layouts() -> dict:
|
||
if not os.path.exists(LAYOUTS_PATH):
|
||
warn(f"layouts.yaml introuvable ({LAYOUTS_PATH}) — validation désactivée")
|
||
return {}
|
||
with open(LAYOUTS_PATH, encoding="utf-8") as f:
|
||
data = yaml.safe_load(f)
|
||
return data.get("layouts", {})
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# ÉTAPE 1 — THE NARRATOR
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def run_narrator(brief: str) -> str:
|
||
"""
|
||
Lance The Narrator sur le brief.
|
||
Boucle d'approbation : valider / feedback / régénérer / afficher.
|
||
Retourne le Markdown approuvé.
|
||
"""
|
||
section("ÉTAPE 1 — THE NARRATOR")
|
||
info("Génération du Markdown narratif...")
|
||
|
||
messages = [{"role": "user", "content": brief}]
|
||
output = call_agent(NARRATOR_ID, messages)
|
||
ok(f"Markdown généré ({len(output)} caractères)")
|
||
|
||
while True:
|
||
print()
|
||
display_markdown(output)
|
||
|
||
print("\n [1] Valider → passer au Designer")
|
||
print(" [2] Donner du feedback au Narrator")
|
||
print(" [3] Régénérer complètement")
|
||
print(" [4] Afficher le Markdown complet")
|
||
print(" [5] Sauvegarder le Markdown")
|
||
|
||
choix = ask("Votre choix :")
|
||
|
||
if choix == "1":
|
||
ok("Markdown approuvé. Passage au Designer...")
|
||
save_output(output, "narrator", "md")
|
||
return output
|
||
|
||
elif choix == "2":
|
||
feedback = ask("Votre feedback :")
|
||
if not feedback:
|
||
continue
|
||
info("Envoi du feedback au Narrator...")
|
||
output = call_with_feedback(NARRATOR_ID, brief, feedback, output)
|
||
ok(f"Nouveau Markdown ({len(output)} caractères)")
|
||
|
||
elif choix == "3":
|
||
info("Régénération depuis le brief...")
|
||
messages = [{"role": "user", "content": brief}]
|
||
output = call_agent(NARRATOR_ID, messages)
|
||
ok(f"Nouveau Markdown ({len(output)} caractères)")
|
||
|
||
elif choix == "4":
|
||
print()
|
||
print(textwrap.indent(output, " "))
|
||
|
||
elif choix == "5":
|
||
save_output(output, "narrator_draft", "md")
|
||
|
||
else:
|
||
warn("Choix invalide. Entrez 1 à 5.")
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# ÉTAPE 2 — THE DESIGNER
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def run_designer(markdown: str) -> str:
|
||
"""
|
||
Lance The Designer sur le Markdown approuvé.
|
||
Boucle d'approbation légère : valider / feedback / afficher.
|
||
Retourne le plan texte structuré.
|
||
"""
|
||
section("ÉTAPE 2 — THE DESIGNER")
|
||
info("Génération du plan de présentation...")
|
||
|
||
messages = [{"role": "user", "content": markdown}]
|
||
output = call_agent(DESIGNER_ID, messages)
|
||
ok(f"Plan généré ({len(output)} caractères)")
|
||
|
||
while True:
|
||
print()
|
||
display_plan(output)
|
||
|
||
print("\n [1] Valider → passer à l'Encoder")
|
||
print(" [2] Donner du feedback au Designer")
|
||
print(" [3] Afficher le plan complet")
|
||
print(" [4] Sauvegarder le plan")
|
||
print(" [0] Revenir au Narrator (modifier le Markdown)")
|
||
|
||
choix = ask("Votre choix :")
|
||
|
||
if choix == "1":
|
||
ok("Plan approuvé. Passage à l'Encoder...")
|
||
save_output(output, "designer", "txt")
|
||
return output
|
||
|
||
elif choix == "2":
|
||
feedback = ask("Votre feedback :")
|
||
if not feedback:
|
||
continue
|
||
info("Envoi du feedback au Designer...")
|
||
output = call_with_feedback(DESIGNER_ID, markdown, feedback, output)
|
||
ok(f"Nouveau plan ({len(output)} caractères)")
|
||
|
||
elif choix == "3":
|
||
print()
|
||
print(textwrap.indent(output, " "))
|
||
|
||
elif choix == "4":
|
||
save_output(output, "designer_draft", "txt")
|
||
|
||
elif choix == "0":
|
||
return None # Signal : retour au Narrator
|
||
|
||
else:
|
||
warn("Choix invalide. Entrez 0 à 4.")
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# ÉTAPE 3 — THE ENCODER
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def run_encoder(plan: str, layouts: dict) -> tuple[str, dict]:
|
||
"""
|
||
Lance The Encoder sur le plan du Designer.
|
||
Valide le YAML produit. Boucle jusqu'à YAML valide ou abandon.
|
||
Retourne (yaml_string, yaml_dict).
|
||
"""
|
||
section("ÉTAPE 3 — THE ENCODER")
|
||
info("Encodage du plan en YAML...")
|
||
|
||
messages = [{"role": "user", "content": plan}]
|
||
output = call_agent(ENCODER_ID, messages)
|
||
ok(f"YAML brut reçu ({len(output)} caractères)")
|
||
|
||
attempts = 0
|
||
max_attempts = 3
|
||
|
||
while attempts < max_attempts:
|
||
is_valid, message, data = validate_yaml(output, layouts)
|
||
|
||
if is_valid:
|
||
ok(f"YAML valide : {message}")
|
||
yaml_clean = extract_yaml(output)
|
||
save_output(yaml_clean, "encoder", "yaml")
|
||
return yaml_clean, data
|
||
|
||
else:
|
||
attempts += 1
|
||
warn(f"YAML invalide (tentative {attempts}/{max_attempts}) :")
|
||
print(f" {message}\n")
|
||
|
||
if attempts >= max_attempts:
|
||
warn("Nombre maximum de tentatives atteint.")
|
||
break
|
||
|
||
info("Correction automatique...")
|
||
correction_msg = (
|
||
f"Le YAML que tu as généré contient des erreurs :\n\n"
|
||
f"{message}\n\n"
|
||
f"Corrige ces erreurs et renvoie uniquement le YAML corrigé "
|
||
f"sans aucun texte avant ou après."
|
||
)
|
||
output = call_with_feedback(ENCODER_ID, plan, correction_msg, output)
|
||
ok(f"YAML corrigé reçu ({len(output)} caractères)")
|
||
|
||
# Après max_attempts : choix utilisateur
|
||
print()
|
||
print(" [1] Accepter le YAML tel quel (avec erreurs)")
|
||
print(" [2] Donner un feedback manuel à l'Encoder")
|
||
print(" [0] Abandonner et revenir au Designer")
|
||
|
||
choix = ask("Votre choix :")
|
||
|
||
if choix == "1":
|
||
yaml_clean = extract_yaml(output)
|
||
save_output(yaml_clean, "encoder_unvalidated", "yaml")
|
||
try:
|
||
data = yaml.safe_load(yaml_clean) or {}
|
||
except Exception:
|
||
data = {}
|
||
return yaml_clean, data
|
||
|
||
elif choix == "2":
|
||
feedback = ask("Votre feedback :")
|
||
output = call_with_feedback(ENCODER_ID, plan, feedback, output)
|
||
yaml_clean = extract_yaml(output)
|
||
save_output(yaml_clean, "encoder_manual", "yaml")
|
||
try:
|
||
data = yaml.safe_load(yaml_clean) or {}
|
||
except Exception:
|
||
data = {}
|
||
return yaml_clean, data
|
||
|
||
else:
|
||
return None, None # Signal : retour au Designer
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# ÉTAPE 4 — RENDER ENGINE
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def run_render(yaml_data: dict) -> Optional[Path]:
|
||
"""
|
||
Appelle render_engine.py avec le YAML validé.
|
||
Retourne le chemin du PPTX généré, ou None si erreur.
|
||
"""
|
||
section("ÉTAPE 4 — RENDER ENGINE")
|
||
|
||
# Vérifie que render_engine.py est accessible
|
||
render_path = Path("render_engine.py")
|
||
if not render_path.exists():
|
||
warn("render_engine.py introuvable dans le dossier courant.")
|
||
warn("Vérifiez que render_engine.py est dans le même dossier que facilitator.py")
|
||
return None
|
||
|
||
# Vérifie les YAML de config
|
||
for path in [THEME_PATH, COMPONENTS_PATH, LAYOUTS_PATH]:
|
||
if not os.path.exists(path):
|
||
warn(f"Fichier YAML manquant : {path}")
|
||
return None
|
||
|
||
# Sauvegarde temporaire du YAML pour render_engine
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
yaml_tmp = OUTPUT_DIR / f"sliding_{timestamp}_input.yaml"
|
||
pptx_out = OUTPUT_DIR / f"sliding_{timestamp}.pptx"
|
||
|
||
yaml_tmp.write_text(
|
||
yaml.dump(yaml_data, allow_unicode=True, default_flow_style=False),
|
||
encoding="utf-8"
|
||
)
|
||
|
||
info(f"Lancement de render_engine.py...")
|
||
info(f"Sortie : {pptx_out}")
|
||
|
||
import subprocess
|
||
result = subprocess.run(
|
||
[
|
||
sys.executable, str(render_path),
|
||
str(yaml_tmp), str(pptx_out),
|
||
"--theme", THEME_PATH,
|
||
"--components", COMPONENTS_PATH,
|
||
"--layouts", LAYOUTS_PATH,
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
|
||
if result.returncode == 0:
|
||
ok(result.stdout.strip() or f"PPTX généré : {pptx_out}")
|
||
return pptx_out
|
||
else:
|
||
warn("Erreur dans render_engine.py :")
|
||
print(textwrap.indent(result.stderr or result.stdout, " "))
|
||
return None
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# BOUCLE PRINCIPALE
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def run():
|
||
banner()
|
||
|
||
# Vérification config minimale
|
||
missing = [k for k, v in {
|
||
"MISTRAL_API_KEY": API_KEY,
|
||
"NARRATOR_AGENT_ID": NARRATOR_ID,
|
||
"DESIGNER_AGENT_ID": DESIGNER_ID,
|
||
"ENCODER_AGENT_ID": ENCODER_ID,
|
||
}.items() if not v]
|
||
|
||
if missing:
|
||
for m in missing:
|
||
warn(f"Variable .env manquante : {m}")
|
||
sys.exit(1)
|
||
|
||
# Chargement layouts pour validation
|
||
layouts = load_layouts()
|
||
if layouts:
|
||
ok(f"layouts.yaml chargé — {len(layouts)} layouts")
|
||
|
||
print(" Commandes : brief libre | quit")
|
||
print(" Tapez votre brief pour démarrer une nouvelle présentation.\n")
|
||
|
||
# ── Boucle de session ─────────────────────────────────────
|
||
while True:
|
||
try:
|
||
brief = input(" Vous : ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\n Session terminée.")
|
||
break
|
||
|
||
if brief.lower() in ("quit", "exit", "q"):
|
||
print(" Au revoir.")
|
||
break
|
||
|
||
if not brief:
|
||
continue
|
||
|
||
# ── Pipeline complet ──────────────────────────────────
|
||
|
||
# ÉTAPE 1 — Narrator
|
||
markdown = run_narrator(brief)
|
||
if not markdown:
|
||
continue
|
||
|
||
# ÉTAPE 2 — Designer (avec possibilité de revenir au Narrator)
|
||
while True:
|
||
plan = run_designer(markdown)
|
||
|
||
if plan is None:
|
||
# Retour au Narrator
|
||
section("RETOUR AU NARRATOR")
|
||
info("Donnez un feedback pour modifier le Markdown :")
|
||
feedback = ask("Feedback :")
|
||
if feedback:
|
||
markdown = call_with_feedback(
|
||
NARRATOR_ID, brief, feedback, markdown)
|
||
ok(f"Markdown mis à jour ({len(markdown)} caractères)")
|
||
continue
|
||
|
||
# ÉTAPE 3 — Encoder (avec possibilité de revenir au Designer)
|
||
yaml_str, yaml_data = run_encoder(plan, layouts)
|
||
|
||
if yaml_str is None:
|
||
# Retour au Designer
|
||
info("Retour au Designer...")
|
||
continue
|
||
|
||
break # YAML OK → on sort de la boucle Designer/Encoder
|
||
|
||
# ÉTAPE 4 — Render
|
||
if yaml_data:
|
||
pptx_path = run_render(yaml_data)
|
||
|
||
if pptx_path and pptx_path.exists():
|
||
section("PRÉSENTATION GÉNÉRÉE")
|
||
ok(f"Fichier PPTX : {pptx_path}")
|
||
ok(f"Taille : {pptx_path.stat().st_size / 1024:.1f} Ko")
|
||
else:
|
||
warn("Le PPTX n'a pas pu être généré.")
|
||
info("Le YAML est disponible dans le dossier output/.")
|
||
|
||
# ── Prochaine présentation ? ──────────────────────────
|
||
print()
|
||
suite = ask("Générer une nouvelle présentation ? (o/n) :").lower()
|
||
if suite not in ("o", "oui", "y", "yes"):
|
||
print(" Au revoir.")
|
||
break
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# POINT D'ENTRÉE
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
if __name__ == "__main__":
|
||
run()
|