""" ICAC Synthesizer v2.2 — LLM integration (Groq + Ollama fallback) Adapté depuis Cyber Strat pour l'administration municipale. v2.2: Format structuré maire + propositions « Pour aller plus loin ». """ import os import re import json import logging from typing import Optional, List log = logging.getLogger("icac.synthesizer") # --- PROMPTS par catégorie --- PROMPTS = { "subventions": """Tu es un expert en financements publics pour les collectivités territoriales françaises. Tu analyses les dispositifs de financement disponibles pour une commune de moins de 5000 hab. Réponds en français. Structure : dispositif / montant / conditions / date limite / contact / source. Cite toujours la source réglementaire (circulaire, décret). Formate les montants avec le symbole €. """, "finances": """Tu es un expert-comptable spécialisé en finances communales. Tu analyses les comptes d'une commune et les compares avec les moyennes nationales de sa strate démographique. Réponds avec des données chiffrées. Utilise des tableaux ASCII alignés. Signale en ⚠ ALERTE tout ratio hors norme (endettement > 40%, épargne < moyenne strate). Cite toujours la source (comptes de gestion DGFiP, année). """, "depenses": """Tu es un auditeur de gestion publique. Tu analyses les dépenses communales par chapitre budgétaire et identifies les anomalies (paiements fractionnés, dépassements, fournisseurs récurrents). Sois factuel. Cite les articles comptables (M14/M57). Signale en ⚠ ANOMALIE tout pattern suspect (fractionnement sous seuil 25k€, cumul fournisseur). """, "associations": """Tu es un expert en vie associative et droit des associations. Tu analyses les subventions versées et la situation administrative des associations d'une commune. Vérifie le statut au JO, les comptes déposés, les conventions de subvention. Cite la date de déclaration et le numéro RNA quand disponible. """, "entreprises": """Tu es un analyste économique territorial. Tu analyses le tissu économique d'une commune : entreprises actives, créations/radiations, secteurs d'activité, emploi. Utilise les données SIRENE. Cite les codes NAF et effectifs. """, "deliberations": """Tu es un juriste spécialisé en droit des collectivités territoriales. Tu analyses les délibérations du Conseil Municipal et identifies les décisions pertinentes pour la question posée. Cite toujours la date et le numéro de délibération. Mentionne le résultat du vote (unanimité, majorité, opposition). """, "marches": """Tu es un expert en marchés publics et commande publique. Tu analyses les marchés en cours, les seuils applicables, les obligations de publicité et de mise en concurrence pour une commune. Cite les articles du Code de la Commande Publique (CCP). Seuils en vigueur : 40k€ HT (fournitures/services), 100k€ HT (travaux). """, "documents": """Tu es un analyste documentaire spécialisé en finances locales. Tu extrais et résumes les informations clés des documents budgétaires communaux (comptes de gestion, budgets primitifs, comptes administratifs). Structure ta réponse : chiffres clés / tendances / alertes / recommandations. """, "general": """Tu es un assistant pour l'administration d'une commune française de moins de 5000 habitants. Tu réponds aux questions du maire et des conseillers municipaux. Sois concis, pratique, sourcé. Réponds toujours en français. """ } # --- Category detection keywords --- CATEGORY_KEYWORDS = { "subventions": ["subvention", "detr", "dsil", "aide", "financement", "dotation", "fonds", "rénovation", "bâtiment"], "finances": ["financ", "budget", "endettement", "épargne", "strate", "dgf", "recette", "dépense se compare"], "depenses": ["119", "communication", "chapitre 65", "fournisseur", "fractionn", "mandat", "dépens"], "associations": ["association", "subvention municipale", "subvention aux", "rna", "préfecture", "déclar"], "entreprises": ["entreprise", "sirene", "btp", "emploi", "création", "fermeture", "activité"], "deliberations": ["délibération", "conseil municipal", "plu", "vote", "séance"], "marches": ["marché public", "appel d'offre", "prestataire", "publicité", "seuil", "concurrence"], "documents": ["document", "pdf", "compte de gestion", "budget primitif", "facture", "upload"], } class Synthesizer: """LLM synthesizer with Groq primary + Ollama fallback.""" def __init__(self): self.groq_api_key = os.environ.get("GROQ_API_KEY", "") self.groq_available = bool(self.groq_api_key) self.ollama_url = os.environ.get("OLLAMA_URL", "http://localhost:11434") self.groq_model = "llama-3.3-70b-versatile" self.ollama_model = "llama3.2" if self.groq_available: log.info("Groq API configured (model: %s)", self.groq_model) else: log.warning("No GROQ_API_KEY — will use Ollama fallback only") def detect_category(self, query: str) -> str: """Detect query category from keywords.""" q = query.lower() scores = {} for cat, keywords in CATEGORY_KEYWORDS.items(): score = sum(1 for kw in keywords if kw in q) if score > 0: scores[cat] = score if not scores: return "general" return max(scores, key=scores.get) def _build_system_prompt(self, category: str, commune_insee: str, rag_empty: bool = False) -> str: """Build system prompt for LLM.""" role_prompt = PROMPTS.get(category, PROMPTS["general"]) if rag_empty: base = f"""Tu es assistant administratif pour la commune de Bessèges (INSEE {commune_insee}). AUCUN document local n'a été trouvé pour cette question. Utilise uniquement les sources web fournies et tes connaissances générales sur l'administration française. Si tu ne trouves pas la réponse, dis-le clairement et suggère où trouver l'information (site, organisme, API).""" else: base = f"""Tu analyses des données pour la commune de Bessèges (INSEE {commune_insee}, < 5 000 hab, Gard 30, Occitanie). RÈGLES STRICTES : 1. Si un document source ne concerne pas la question posée, IGNORE-LE COMPLÈTEMENT. Ne le mentionne pas. Ex: si on te demande une association et qu'on t'envoie un extrait de compte de gestion → ignore ce compte. 2. Utilise UNIQUEMENT les sources directement pertinentes à la question. 3. Si les sources ne contiennent pas la réponse, dis-le clairement : "Je n'ai pas trouvé cette information dans les documents disponibles." puis explique ce qu'il faudrait chercher. 4. Ne jamais inventer de chiffres, noms, délibérations. 5. Tes connaissances générales sur le droit et l'administration française peuvent compléter — précise [connaissance générale].""" return f"""{base} {role_prompt} FORMAT OBLIGATOIRE DE RÉPONSE: Tu dois structurer ta réponse en 3 sections clairement séparées: ═══ SYNTHÈSE ═══ Résumé factuel en 5-10 lignes, chiffré, avec les données clés. Utilise des listes numérotées si plusieurs résultats. ═══ POINTS DE VIGILANCE ═══ Alertes (⚠), anomalies, ratios hors norme, risques identifiés. Si aucune alerte: écrire "Aucune anomalie détectée." ═══ POUR ALLER PLUS LOIN ═══ Exactement 3 propositions d'approfondissement, numérotées [1] [2] [3]. Chaque proposition est une question ICAC complète que le maire peut poser. Format: [N] Question complète et actionnable Exemple: [1] Détail des mandats émis au chapitre 65 en 2024 RÈGLES DE FORMATAGE: - Réponds en français uniquement - Sois concis et structuré - Signale les alertes avec ⚠ - Cite tes sources à la fin """ def _build_user_prompt(self, query: str, context: str) -> str: """Build user prompt with optional RAG context.""" if context: return f"""DONNÉES DISPONIBLES: {context} QUESTION: {query} Analyse les données ci-dessus et réponds à la question. Si les données sont insuffisantes, indique-le.""" return query async def _call_groq(self, system: str, user: str) -> str: """Call Groq API.""" import httpx async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( "https://api.groq.com/openai/v1/chat/completions", headers={ "Authorization": f"Bearer {self.groq_api_key}", "Content-Type": "application/json" }, json={ "model": self.groq_model, "messages": [ {"role": "system", "content": system}, {"role": "user", "content": user} ], "temperature": 0.3, "max_tokens": 2000 } ) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"] async def _call_ollama(self, system: str, user: str) -> str: """Call Ollama local API.""" import httpx async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.post( f"{self.ollama_url}/api/chat", json={ "model": self.ollama_model, "messages": [ {"role": "system", "content": system}, {"role": "user", "content": user} ], "stream": False } ) resp.raise_for_status() data = resp.json() return data["message"]["content"] @staticmethod def _extraire_propositions(raw_text: str) -> List[str]: """ Extrait les propositions « Pour aller plus loin » du texte LLM. Cherche les patterns [1] ..., [2] ..., [3] ... dans la section POUR ALLER PLUS LOIN. """ propositions = [] # Chercher la section "POUR ALLER PLUS LOIN" section_match = re.search( r"(?:POUR ALLER PLUS LOIN|Pour aller plus loin).*?\n(.*)", raw_text, re.DOTALL | re.IGNORECASE ) text_to_search = section_match.group(1) if section_match else raw_text # Pattern [1] Question..., [2] Question..., [3] Question... pattern = re.findall( r'\[(\d)\]\s*(.+?)(?=\n\[|\n═|$)', text_to_search ) for num, question in pattern: q = question.strip().rstrip(".") if len(q) > 10: propositions.append(q) # Fallback: chercher des lignes numérotées dans la section if not propositions and section_match: lines = section_match.group(1).strip().split("\n") for line in lines: line = line.strip() # "1. Question" or "1) Question" or "- Question" m = re.match(r'(?:\d[.)]\s*|-\s*)(.{15,})', line) if m: propositions.append(m.group(1).strip().rstrip(".")) if len(propositions) >= 3: break return propositions[:3] def _format_response(self, raw_text: str, category: str) -> dict: """Format LLM response into terminal lines + synthesis text + propositions.""" lines_out = [] sep = "━" * 45 # Terminal lines (abbreviated) lines_out.append({"t": "icac", "h": "ANALYSE EN COURS..."}) lines_out.append({"t": "icac", "h": sep}) # Extract first meaningful line as title text_lines = [l for l in raw_text.strip().split("\n") if l.strip()] if text_lines: title = text_lines[0][:60] lines_out.append({"t": "icac", "h": title}) lines_out.append({"t": "icac", "h": "Détails dans la console de synthèse ▶"}) # Check for alerts if "⚠" in raw_text or "alerte" in raw_text.lower() or "anomalie" in raw_text.lower(): alert_lines = [l for l in raw_text.split("\n") if "⚠" in l or "alerte" in l.lower() or "anomalie" in l.lower()] for al in alert_lines[:3]: lines_out.append({"t": "alert", "h": al.strip()[:70]}) # Extract propositions propositions = self._extraire_propositions(raw_text) if propositions: lines_out.append({"t": "icac", "h": sep}) lines_out.append({"t": "icac", "h": "POUR ALLER PLUS LOIN ▶"}) return { "lines": lines_out, "synth": raw_text, "propositions": propositions, } async def synthesize(self, query: str, category: str, commune_insee: str, context: str = "", rag_empty: bool = False) -> dict: """Full synthesis pipeline: prompt → LLM → formatted response.""" system = self._build_system_prompt(category, commune_insee, rag_empty=rag_empty) user = self._build_user_prompt(query, context) log.info("Synthesize: category=%s context_len=%d rag_empty=%s", category, len(context), rag_empty) raw_text = "" # Try Groq first if self.groq_available: try: raw_text = await self._call_groq(system, user) log.info("Groq response: %d chars", len(raw_text)) except Exception as e: log.warning("Groq failed: %s — trying Ollama", e) # Fallback to Ollama if not raw_text: try: raw_text = await self._call_ollama(system, user) log.info("Ollama response: %d chars", len(raw_text)) except Exception as e: log.error("Ollama also failed: %s", e) raise RuntimeError(f"LLM indisponible (Groq + Ollama): {e}") return self._format_response(raw_text, category)