la-voix-du-peuple/artifacts/flask-api/ai_agent.py

"""
Agent IA pour le filtrage éthique et la synthèse démocratique.
Supporte Mistral AI, OpenAI, et les intégrations Replit AI.
"""
import json
import os
import logging
from openai import OpenAI, BadRequestError
from legal_framework import LEGAL_FILTER_PROMPT, SYNTHESIS_PROMPT

logger = logging.getLogger(__name__)

MISTRAL_BASE_URL = "https://api.mistral.ai/v1"

_client: OpenAI | None = None


def get_client() -> OpenAI:
    """
    Supporte trois modes (par ordre de priorité) :
    1. Mistral AI            : MISTRAL_API_KEY (+ MISTRAL_BASE_URL optionnel)
    2. OpenAI standard       : OPENAI_API_KEY (+ OPENAI_BASE_URL optionnel)
    3. Replit AI Integration : AI_INTEGRATIONS_OPENAI_BASE_URL + AI_INTEGRATIONS_OPENAI_API_KEY
    """
    global _client
    if _client is None:
        mistral_key  = os.environ.get("MISTRAL_API_KEY")
        mistral_base = os.environ.get("MISTRAL_BASE_URL", MISTRAL_BASE_URL)
        std_key      = os.environ.get("OPENAI_API_KEY")
        std_base     = os.environ.get("OPENAI_BASE_URL")
        replit_base  = os.environ.get("AI_INTEGRATIONS_OPENAI_BASE_URL")
        replit_key   = os.environ.get("AI_INTEGRATIONS_OPENAI_API_KEY")

        if mistral_key:
            logger.info("Utilisation de l'API Mistral AI (%s)", mistral_base)
            _client = OpenAI(base_url=mistral_base, api_key=mistral_key)
        elif std_key:
            logger.info("Utilisation de l'API OpenAI")
            kwargs = {"api_key": std_key}
            if std_base:
                kwargs["base_url"] = std_base
            _client = OpenAI(**kwargs)
        elif replit_base and replit_key:
            logger.info("Utilisation de l'intégration Replit AI")
            _client = OpenAI(base_url=replit_base, api_key=replit_key)
        else:
            raise RuntimeError(
                "Aucune clé IA configurée. "
                "Définissez MISTRAL_API_KEY ou OPENAI_API_KEY dans le fichier .env, "
                "ou configurez les intégrations Replit AI."
            )
    return _client


def filter_idea(content: str) -> dict:
    """
    Filtre une idée selon le cadre légal international des droits humains.
    Retourne : {"accepted": bool, "reason"?: str, "legal_basis"?: str}
    """
    try:
        client = get_client()
        filter_model = os.environ.get("FILTER_MODEL", os.environ.get("OPENAI_FILTER_MODEL", "mistral-small-latest"))
        response = client.chat.completions.create(
            model=filter_model,
            max_tokens=300,
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": LEGAL_FILTER_PROMPT},
                {"role": "user", "content": f'Idée soumise : "{content}"'},
            ],
        )
        raw = (
            response.choices[0].message.content
            or '{"accepted": false, "reason": "Contenu non conforme aux principes démocratiques."}'
        )
        raw = raw.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
            raw = raw.strip()
        start = raw.find("{")
        end = raw.rfind("}") + 1
        if start != -1 and end > start:
            raw = raw[start:end]
        result = json.loads(raw)
        return result
    except json.JSONDecodeError:
        logger.warning("Impossible de parser la réponse JSON du filtre, raw=%r", raw if 'raw' in dir() else 'N/A')
        return {"accepted": False, "reason": "Erreur interne de filtrage"}
    except BadRequestError as e:
        if "content_filter" in str(e) or "content management policy" in str(e):
            logger.warning("Contenu bloqué par le filtre de sécurité du proxy IA — rejet automatique")
            return {
                "accepted": False,
                "reason": (
                    "Ce contenu a été automatiquement bloqué car il contient des propos "
                    "haineux ou violents graves, contraires à la dignité humaine."
                ),
                "legal_basis": (
                    "DUDH Art. 1 (dignité humaine), DUDH Art. 20 (interdiction de la haine), "
                    "PIDCP Art. 20, CEDH Art. 17 (abus de droit)"
                ),
            }
        logger.exception("Erreur API lors du filtrage")
        return {"accepted": False, "reason": "Service temporairement indisponible"}
    except Exception:
        logger.exception("Erreur lors du filtrage de l'idée")
        return {"accepted": False, "reason": "Service temporairement indisponible"}


def synthesize_ideas(ideas: list[str]) -> str:
    """
    Synthétise une liste d'idées acceptées en un texte collectif
    ancré dans les valeurs démocratiques et les droits humains.
    """
    if not ideas:
        return (
            "Aucune idée n'a encore été soumise. "
            "Soyez le premier à partager votre vision pour une société meilleure, "
            "fondée sur la Déclaration universelle des droits de l'homme."
        )
    try:
        client = get_client()
        synthesis_model = os.environ.get("SYNTHESIS_MODEL", os.environ.get("OPENAI_SYNTHESIS_MODEL", "mistral-large-latest"))
        ideas_text = "\n".join(f"{i + 1}. {idea}" for i, idea in enumerate(ideas))
        response = client.chat.completions.create(
            model=synthesis_model,
            max_tokens=1200,
            messages=[
                {"role": "system", "content": SYNTHESIS_PROMPT},
                {
                    "role": "user",
                    "content": (
                        f"Voici les {len(ideas)} idée(s) citoyenne(s) validées :\n\n"
                        f"{ideas_text}\n\n"
                        "Rédige La Voix du Peuple."
                    ),
                },
            ],
        )
        return response.choices[0].message.content or "Synthèse en cours..."
    except Exception:
        logger.exception("Erreur lors de la synthèse des idées")
        return "La synthèse est temporairement indisponible. Vos idées ont bien été enregistrées."