Files
sliding-automation/archive/v1_pipeline/facilitator.py
T

621 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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"
if not yaml_data:
warn("yaml_data est vide — impossible de générer le PPTX.")
return None
yaml_content = yaml.dump(yaml_data, allow_unicode=True, default_flow_style=False)
yaml_tmp.write_text(yaml_content, encoding="utf-8")
# Vérification que le fichier n'est pas vide
if yaml_tmp.stat().st_size == 0:
warn(f"Fichier YAML temporaire vide : {yaml_tmp}")
return None
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()