"""Cliente Ollama para síntesis narrativa de location_agent. Solo se usa para texto — los scores son determinísticos. Env vars: OLLAMA_MODEL: modelo (default: llama3.1:8b) OLLAMA_HOST: URL del servidor (default: http://localhost:11434) """ from __future__ import annotations import json import os def _model() -> str: return os.getenv("OLLAMA_MODEL", "llama3.1:8b") def is_available() -> bool: try: import requests r = requests.get(f"{os.getenv('OLLAMA_HOST', 'http://localhost:11434')}/api/tags", timeout=3) return r.status_code == 200 except Exception: return False def analyze_section(data: dict, section: str, address: str) -> str: """Narrativa de 2-3 párrafos para una sección del reporte.""" if not is_available(): return _auto_summary(data, section) try: import ollama prompt = ( f"Analiza los siguientes datos de {section} para la ubicación: {address}\n\n" "Sé objetivo y profesional. 2-3 párrafos. Solo hechos relevantes para inversión " "inmobiliaria. Sin recomendaciones de compra. Responde en español.\n\n" f"Datos:\n{json.dumps(data, indent=2, ensure_ascii=False)}" ) resp = ollama.chat( model=_model(), messages=[{"role": "user", "content": prompt}], options={"temperature": 0.3}, ) return resp["message"]["content"].strip() except Exception as e: return _auto_summary(data, section) + f"\n\n[Ollama no disponible: {e}]" def executive_summary(all_scores: dict, address: str, overall: int) -> dict: """Genera resumen ejecutivo. Devuelve {summary, strengths, weaknesses}.""" if not is_available(): return _auto_executive(all_scores, overall) try: import ollama scores_txt = "\n".join(f" - {k}: {v}/100" for k, v in all_scores.items()) prompt = ( f"Resumen ejecutivo de inteligencia de ubicación para inversión inmobiliaria.\n" f"Dirección: {address}\nScore General: {overall}/100\n\nScores:\n{scores_txt}\n\n" "Proporciona en JSON:\n" '{"summary": "3-5 párrafos objetivos", ' '"strengths": ["fortaleza 1","fortaleza 2","fortaleza 3"], ' '"weaknesses": ["debilidad 1","debilidad 2","debilidad 3"]}\n' "Sin recomendaciones de compra. En español." ) resp = ollama.chat( model=_model(), messages=[{"role": "user", "content": prompt}], options={"temperature": 0.3}, format="json", ) parsed = json.loads(resp["message"]["content"]) return { "summary": parsed.get("summary", ""), "strengths": parsed.get("strengths", [])[:3], "weaknesses": parsed.get("weaknesses", [])[:3], } except Exception as e: r = _auto_executive(all_scores, overall) r["summary"] += f"\n[Ollama no disponible: {e}]" return r def _auto_summary(data: dict, section: str) -> str: if not data: return f"Datos de {section} no disponibles." lines = [f"Resumen automático — {section}:"] for k, v in list(data.items())[:8]: if isinstance(v, (str, int, float, bool)) and v not in ("", None): lines.append(f" • {k}: {v}") return "\n".join(lines) def _auto_executive(scores: dict, overall: int) -> dict: ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) return { "summary": f"Score general: {overall}/100. Active Ollama para análisis narrativo detallado.", "strengths": [f"{k}: {v}/100" for k, v in ranked[:3]], "weaknesses": [f"{k}: {v}/100" for k, v in ranked[-3:]], }