211 lines
8.4 KiB
Python
211 lines
8.4 KiB
Python
"""
|
|
La Voix du Peuple — Agent IA
|
|
Copyright (C) 2026 billisdead — Licence EUPL-1.2
|
|
|
|
Agent IA pour le filtrage éthique et la synthèse démocratique.
|
|
Supporte Mistral AI (par défaut) et tout fournisseur compatible OpenAI.
|
|
"""
|
|
import json
|
|
import os
|
|
import logging
|
|
from openai import OpenAI, BadRequestError
|
|
from legal_framework import LEGAL_FILTER_PROMPT, SYNTHESIS_PROMPT, LAW_CHECK_PROMPT, build_filter_prompt
|
|
from legifrance_client import get_legifrance_client
|
|
from database import get_setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MISTRAL_BASE_URL = "https://api.mistral.ai/v1"
|
|
_LEGAL_REFS_PATH = os.path.join(os.path.dirname(__file__), "legal_refs.yaml")
|
|
|
|
_client: OpenAI | None = None
|
|
|
|
|
|
def _get_live_legal_texts() -> dict[str, str]:
|
|
"""
|
|
Récupère les textes consolidés des articles critiques depuis Légifrance (PISTE).
|
|
Retourne {} si PISTE n'est pas configuré ou en cas d'erreur — sans bloquer le filtrage.
|
|
Les résultats sont mis en cache dans Redis (24h).
|
|
"""
|
|
try:
|
|
lf = get_legifrance_client()
|
|
if not lf.available:
|
|
return {}
|
|
return lf.fetch_tracked_articles(_LEGAL_REFS_PATH)
|
|
except Exception:
|
|
logger.warning("Légifrance: impossible de récupérer les textes live — mode statique")
|
|
return {}
|
|
|
|
|
|
def get_model(kind: str) -> str:
|
|
"""kind = "filter" | "synthesis". Priorité : réglage admin (DB) > env > défaut."""
|
|
default = "mistral-small-latest" if kind == "filter" else "mistral-large-latest"
|
|
try:
|
|
saved = get_setting(f"{kind}_model")
|
|
except Exception:
|
|
saved = None
|
|
return saved or os.environ.get(f"{kind.upper()}_MODEL", os.environ.get(f"OPENAI_{kind.upper()}_MODEL", default))
|
|
|
|
|
|
def get_client() -> OpenAI:
|
|
"""
|
|
Supporte deux modes (par ordre de priorité) :
|
|
1. Mistral AI : MISTRAL_API_KEY (+ MISTRAL_BASE_URL optionnel)
|
|
2. OpenAI-compatible : OPENAI_API_KEY (+ OPENAI_BASE_URL optionnel)
|
|
"""
|
|
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")
|
|
|
|
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 d'une API compatible OpenAI")
|
|
kwargs: dict = {"api_key": std_key}
|
|
if std_base:
|
|
kwargs["base_url"] = std_base
|
|
_client = OpenAI(**kwargs)
|
|
else:
|
|
raise RuntimeError(
|
|
"Aucune clé IA configurée. "
|
|
"Définissez MISTRAL_API_KEY (recommandé) ou OPENAI_API_KEY dans le fichier .env."
|
|
)
|
|
return _client
|
|
|
|
|
|
GITHUB_CATALOG_URL = "https://models.github.ai/catalog/models"
|
|
|
|
|
|
def list_model_ids() -> list[str]:
|
|
"""Modèles accessibles avec la clé en place (GitHub Models : catalogue dédié, /models n'est pas OpenAI-compatible)."""
|
|
client = get_client()
|
|
if "models.github.ai" in str(client.base_url):
|
|
return sorted(m["id"] for m in client.get(GITHUB_CATALOG_URL, cast_to=object))
|
|
return sorted(m.id for m in client.models.list().data)
|
|
|
|
|
|
def filter_idea(content: str) -> dict:
|
|
"""
|
|
Filtre une idée selon le cadre légal international des droits humains
|
|
et le droit pénal français (Code pénal, Loi du 29 juillet 1881, LCEN,
|
|
Loi SREN 2024, RGPD, Code de la santé publique).
|
|
Retourne : {"accepted": bool, "reason"?: str, "legal_basis"?: str}
|
|
"""
|
|
try:
|
|
client = get_client()
|
|
filter_model = get_model("filter")
|
|
live_texts = _get_live_legal_texts()
|
|
system_prompt = build_filter_prompt(live_texts)
|
|
if live_texts:
|
|
logger.debug("Légifrance: %d article(s) injecté(s) dans le prompt", len(live_texts))
|
|
response = client.chat.completions.create(
|
|
model=filter_model,
|
|
max_tokens=300,
|
|
response_format={"type": "json_object"},
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": f"Idée soumise :\n<<<DEBUT_SOUMISSION>>>\n{content}\n<<<FIN_SOUMISSION>>>"},
|
|
],
|
|
)
|
|
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 locals() 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 check_existing_law(content: str) -> str:
|
|
"""
|
|
Vérifie si une proposition citoyenne est déjà couverte par une loi
|
|
française ou un règlement européen en vigueur.
|
|
Retourne une phrase informative (non bloquante).
|
|
"""
|
|
try:
|
|
client = get_client()
|
|
model = get_model("filter")
|
|
response = client.chat.completions.create(
|
|
model=model,
|
|
max_tokens=200,
|
|
messages=[
|
|
{"role": "system", "content": LAW_CHECK_PROMPT},
|
|
{"role": "user", "content": f"Proposition :\n<<<DEBUT_SOUMISSION>>>\n{content}\n<<<FIN_SOUMISSION>>>"},
|
|
],
|
|
)
|
|
return (response.choices[0].message.content or "").strip()
|
|
except Exception:
|
|
logger.exception("Erreur lors de la vérification du cadre légal existant")
|
|
return ""
|
|
|
|
|
|
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 = get_model("synthesis")
|
|
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, entre les balises :\n\n"
|
|
f"<<<DEBUT_SOUMISSIONS>>>\n{ideas_text}\n<<<FIN_SOUMISSIONS>>>\n\n"
|
|
"Rédige La Voix du Peuple à partir de ce contenu."
|
|
),
|
|
},
|
|
],
|
|
)
|
|
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."
|