Spaces:
Paused
Paused
OpenItaLaw Builder
Fix metadata caching + fix GU Explainer LLM integration (route_request -> chat)
97f7c7f | """ | |
| Enhanced GU (Gazzetta Ufficiale) Explainer - AI-powered law explanations with context | |
| Generates professional, detailed explanations of laws with: | |
| - Direct law citations and references | |
| - Historical context | |
| - Vigente status and impact | |
| - Related laws and amendments | |
| """ | |
| import logging | |
| from typing import Optional, Dict, Any | |
| from pathlib import Path | |
| logger = logging.getLogger(__name__) | |
| class GUExplainer: | |
| """Generate AI-enhanced explanations for Italian laws with citations.""" | |
| def __init__(self, llm_router=None): | |
| """ | |
| Initialize explainer with optional LLM router. | |
| Args: | |
| llm_router: LLMRouter instance for generating explanations | |
| """ | |
| self.llm_router = llm_router | |
| self.cache = {} | |
| def generate_professional_explanation( | |
| self, | |
| metadata: Dict[str, Any], | |
| text_excerpt: Optional[str] = None, | |
| max_length: int = 500 | |
| ) -> str: | |
| """ | |
| Generate a professional explanation for a law with AI context. | |
| Args: | |
| metadata: Document metadata from FAISS (urn, source_title, etc.) | |
| text_excerpt: Optional text excerpt from the actual law | |
| max_length: Max length of explanation | |
| Returns: | |
| Professional explanation string with citations | |
| """ | |
| urn = metadata.get("urn", "unknown") | |
| # Check cache first | |
| if urn in self.cache: | |
| return self.cache[urn] | |
| # Build explanation with available metadata | |
| explanation = self._build_explanation(metadata, text_excerpt) | |
| # Enhance with LLM if available — works with or without text excerpt | |
| if self.llm_router: | |
| explanation = self._enhance_with_llm( | |
| metadata, explanation, text_excerpt or metadata.get("source_title", "") | |
| ) | |
| self.cache[urn] = explanation | |
| return explanation | |
| def _build_explanation( | |
| self, | |
| metadata: Dict[str, Any], | |
| text_excerpt: Optional[str] = None | |
| ) -> str: | |
| """Build template explanation from metadata.""" | |
| source_title = metadata.get("source_title", "Legge") | |
| urn = metadata.get("urn", "") | |
| enactment_date = metadata.get("enactment_date", "") | |
| validity = metadata.get("validity_status", "unknown") | |
| legal_era = metadata.get("legal_era", "unknown") | |
| article_num = metadata.get("article_number", "") | |
| # Format date | |
| date_str = "" | |
| if enactment_date and len(enactment_date) == 8: | |
| date_str = f"{enactment_date[6:8]}/{enactment_date[4:6]}/{enactment_date[:4]}" | |
| # Build explanation | |
| parts = [] | |
| # Title with citation | |
| parts.append(f"**{source_title}**") | |
| if article_num: | |
| parts.append(f"Art. {article_num}") | |
| # Status indicator | |
| status_emoji = "✨" if validity == "in_corso" else "⚠️" | |
| status_text = "in vigore (efectiva)" if validity == "in_corso" else "non in vigore" | |
| parts.append(f"{status_emoji} Stato: {status_text}") | |
| # Era and context | |
| era_map = { | |
| "Fascismo": "Era fascista", | |
| "Repubblica": "Era repubblicana", | |
| "Contemporaneo": "Era contemporanea", | |
| "unknown": "Era indeterminata" | |
| } | |
| era_label = era_map.get(legal_era, legal_era) | |
| parts.append(f"📅 Periodo: {era_label}") | |
| if date_str: | |
| parts.append(f"📌 Data di emanazione: {date_str}") | |
| if urn: | |
| parts.append(f"🔗 URN: `{urn}`") | |
| # Text excerpt (if available) | |
| if text_excerpt: | |
| excerpt = text_excerpt[:200].strip() | |
| if len(text_excerpt) > 200: | |
| excerpt += "..." | |
| parts.append(f"\n**Testo**: \"{excerpt}\"") | |
| return "\n".join(parts) | |
| def _enhance_with_llm( | |
| self, | |
| metadata: Dict[str, Any], | |
| base_explanation: str, | |
| text_excerpt: str | |
| ) -> str: | |
| """Enhance explanation with LLM-generated analysis via LLMRouter.""" | |
| if not self.llm_router: | |
| return base_explanation | |
| try: | |
| source_title = metadata.get("source_title", "Legge") | |
| article_num = metadata.get("article_number", "") | |
| validity = metadata.get("validity_status", "") | |
| era = metadata.get("legal_era", "") | |
| urn = metadata.get("urn", "") | |
| system_msg = ( | |
| "Sei un esperto giurista italiano. Fornisci analisi legali precise, " | |
| "professionali e accessibili. Rispondi sempre in italiano." | |
| ) | |
| user_msg = ( | |
| f"Analizza questa norma italiana e fornisci una spiegazione professionale:\n\n" | |
| f"**Titolo**: {source_title}\n" | |
| f"**Articolo**: {article_num}\n" | |
| f"**Era giuridica**: {era}\n" | |
| f"**Stato di vigenza**: {validity}\n" | |
| f"**URN**: {urn}\n\n" | |
| f"**Testo**:\n{text_excerpt[:800]}\n\n" | |
| "Fornisci:\n" | |
| "1. Sintesi dello scopo e dell'impatto della norma (2-3 frasi)\n" | |
| "2. Obblighi o diritti principali stabiliti\n" | |
| "3. Soggetti destinatari\n" | |
| "4. Eventuali modifiche o leggi correlate note\n\n" | |
| "Formato: spiegazione legale professionale in italiano, ~150 parole. " | |
| "Cita articoli specifici ove pertinente." | |
| ) | |
| messages = [ | |
| {"role": "system", "content": system_msg}, | |
| {"role": "user", "content": user_msg}, | |
| ] | |
| response, _provider = self.llm_router.chat(messages, max_tokens=400) | |
| if response and not response.startswith("❌"): | |
| return f"{base_explanation}\n\n**Analisi AI**:\n{response}" | |
| except Exception as e: | |
| logger.warning(f"LLM enhancement failed: {e}") | |
| return base_explanation | |
| def batch_explain( | |
| self, | |
| metadata_records: list, | |
| text_excerpts: Optional[Dict[str, str]] = None | |
| ) -> Dict[str, str]: | |
| """ | |
| Generate explanations for multiple laws. | |
| Args: | |
| metadata_records: List of metadata dicts | |
| text_excerpts: Optional dict mapping URN → text excerpt | |
| Returns: | |
| Dict mapping URN → explanation | |
| """ | |
| explanations = {} | |
| for record in metadata_records: | |
| urn = record.get("urn") | |
| if urn: | |
| text = text_excerpts.get(urn) if text_excerpts else None | |
| explanations[urn] = self.generate_professional_explanation(record, text) | |
| return explanations | |
| def explain_with_context( | |
| self, | |
| metadata: Dict[str, Any], | |
| related_docs: Optional[list] = None, | |
| amendments: Optional[list] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Generate comprehensive explanation with related laws and amendments. | |
| Args: | |
| metadata: Main document metadata | |
| related_docs: List of related document metadata dicts | |
| amendments: List of amendment records | |
| Returns: | |
| Dict with explanation, related laws, amendments | |
| """ | |
| explanation = self.generate_professional_explanation(metadata) | |
| result = { | |
| "main": explanation, | |
| "related": [], | |
| "amendments": [], | |
| "sources": [ | |
| { | |
| "type": "Normattiva", | |
| "url": metadata.get("normattiva_url", ""), | |
| "urn": metadata.get("urn", "") | |
| } | |
| ] | |
| } | |
| # Add related laws | |
| if related_docs: | |
| for doc in related_docs[:5]: # Limit to 5 | |
| result["related"].append({ | |
| "title": doc.get("source_title", ""), | |
| "urn": doc.get("urn", ""), | |
| "status": doc.get("validity_status", "unknown") | |
| }) | |
| # Add amendments | |
| if amendments: | |
| for amend in amendments[:3]: # Limit to 3 | |
| result["amendments"].append({ | |
| "type": amend.get("type", "amendment"), | |
| "date": amend.get("date", ""), | |
| "description": amend.get("description", "") | |
| }) | |
| return result | |
| # Default explanations (fallback when LLM unavailable) | |
| DEFAULT_EXPLANATIONS = { | |
| "in_corso": "Legge attualmente in vigore e applicabile.", | |
| "abrogato": "Legge abrogata (annullata/revocata) e non più applicabile.", | |
| "scaduto": "Legge scaduta e non più applicabile.", | |
| "non_vigente": "Legge non in vigore (non ancora applicabile o cessata).", | |
| } | |