fn_registry/python/functions/datascience/describe_clusters_llm.py

"""describe_clusters_llm — micro-analisis LLM de clusters de KMeans (grupo `eda`).

Toma los PERFILES AGREGADOS de cada cluster (los que produce `project_clusters_2d`:
tamano, centroide en escala original, features distintivas y centroide en z-score)
y, con UNA sola llamada al LLM, pide por cada cluster un TITULO corto + una
descripcion de 1-2 frases, en espanol.

Clave de coste y privacidad: NO se envian filas crudas al LLM. Solo viaja el
perfil AGREGADO de cada grupo (tamano, % del total y la media de las features
distintivas con su signo respecto a la media global). El coste es minimo y ningun
dato fila-a-fila sale del proceso.

Reusa `ask_llm` del registry (grupo claude-direct, API directa con el token OAuth
de Claude en ~/.claude/.credentials.json, arranque 0). Impura: una llamada de red.
Estilo dict-no-throw: NUNCA lanza; ante cualquier fallo (red, LLM caido, parseo)
degrada a titulos genericos "Cluster N" + una nota explicando el motivo.
"""

import json

from core.ask_llm import ask_llm

_SYSTEM = (
    "Eres un analista de datos. Recibes los PERFILES AGREGADOS de los clusters de "
    "un KMeans (por cada grupo: su tamano y la media de sus features distintivas, "
    "con el signo respecto a la media global; nunca filas crudas) y los describes "
    "de forma sobria y util. Para cada cluster generas un titulo corto y "
    "descriptivo (por ejemplo 'Vinos de alta acidez y baja graduacion') y una "
    "descripcion de 1-2 frases. NO inventes causas ni sobre-interpretes: limitate a "
    "lo que dicen los numeros. Responde en espanol. Responde SIEMPRE y SOLO con un "
    "unico JSON array valido, sin texto alrededor y sin fences de markdown, con "
    'EXACTAMENTE la forma [{"cluster": <int>, "title": "<titulo corto>", '
    '"description": "<1-2 frases>"}], un objeto por cluster.'
)


def _fmt_num(value) -> str:
    """Formatea un numero de forma compacta para el prompt (None -> '?')."""
    if value is None:
        return "?"
    if isinstance(value, bool):
        return str(value)
    if isinstance(value, float):
        if value == int(value):
            return str(int(value))
        return f"{value:.4g}"
    return str(value)


def _cluster_id(profile: dict, index: int) -> int:
    """Devuelve el id del cluster del perfil, o el indice si no es un int valido."""
    raw = (profile or {}).get("cluster")
    if isinstance(raw, bool):
        return index
    if isinstance(raw, int):
        return raw
    try:
        return int(raw)
    except (TypeError, ValueError):
        return index


def _build_prompt(cluster_profiles: list, feature_names: list) -> str:
    """Construye un resumen textual compacto de los perfiles para el LLM.

    Funcion interna PURA: no toca red ni disco, es testeable sin credenciales.
    Por cada cluster incluye su numero, tamano (size + pct%) y, para cada feature
    distintiva, el valor del centroide en escala original mas si esta por encima o
    por debajo de la media (signo del z-score en centroid_z). Pasa AGREGADOS, nunca
    dato crudo de filas.

    Args:
        cluster_profiles: lista de perfiles de cluster (forma de project_clusters_2d).
        feature_names: nombres de las features del dataset (solo contexto).

    Returns:
        El texto del prompt.
    """
    cluster_profiles = cluster_profiles or []
    feature_names = feature_names if isinstance(feature_names, list) else []

    lines = [
        "Perfiles AGREGADOS de clusters de KMeans. No hay filas crudas, solo medias por grupo.",
        f"Numero de clusters: {len(cluster_profiles)}",
    ]
    if feature_names:
        lines.append("Features del dataset: " + ", ".join(str(f) for f in feature_names))
    lines.append("")

    for i, prof in enumerate(cluster_profiles):
        prof = prof or {}
        cid = _cluster_id(prof, i)
        size = prof.get("size")
        pct = prof.get("pct")
        pct_str = f"{pct:.1f}%" if isinstance(pct, (int, float)) and not isinstance(pct, bool) else "?"
        lines.append(f"Cluster {cid}: tamano={_fmt_num(size)} ({pct_str} del total)")

        distinctive = prof.get("distinctive") or []
        centroid_o = prof.get("centroid_original") or {}
        centroid_z = prof.get("centroid_z") or {}

        if distinctive:
            lines.append("  Features distintivas (media del grupo):")
            for feat in distinctive:
                val = centroid_o.get(feat)
                z = centroid_z.get(feat)
                direction = ""
                if isinstance(z, (int, float)) and not isinstance(z, bool):
                    if z > 0:
                        direction = "por encima de la media"
                    elif z < 0:
                        direction = "por debajo de la media"
                    else:
                        direction = "en la media"
                if direction:
                    lines.append(f"    - {feat}: {_fmt_num(val)} ({direction})")
                else:
                    lines.append(f"    - {feat}: {_fmt_num(val)}")
        else:
            lines.append("  (sin features distintivas marcadas)")
        lines.append("")

    lines.append(
        "Devuelve SOLO el JSON array descrito en las instrucciones del sistema, "
        "sin texto antes ni despues."
    )
    return "\n".join(lines)


def _parse_clusters_json(text: str, n: int):
    """Extrae y normaliza el array JSON de la respuesta del LLM.

    Funcion interna testeable sin red. Localiza el primer '[' y el ultimo ']' del
    texto (tolerando texto basura alrededor o fences de markdown), hace json.loads
    y normaliza cada entrada a {cluster:int, title:str, description:str}, rellenando
    el cluster por indice si falta. NUNCA lanza: ante cualquier fallo devuelve None
    (senal de degradacion para el caller).

    Args:
        text: respuesta cruda del LLM.
        n: numero de perfiles esperados (referencia; la longitud real la marca el array).

    Returns:
        Lista normalizada de dicts, o None si no se pudo parsear un array valido.
    """
    if not text or not isinstance(text, str):
        return None

    start = text.find("[")
    end = text.rfind("]")
    if start == -1 or end == -1 or end <= start:
        return None

    try:
        data = json.loads(text[start : end + 1])
    except (ValueError, TypeError):
        return None

    if not isinstance(data, list):
        return None

    out = []
    for i, item in enumerate(data):
        if not isinstance(item, dict):
            out.append({"cluster": i, "title": f"Cluster {i}", "description": ""})
            continue

        raw_cluster = item.get("cluster")
        if isinstance(raw_cluster, bool):
            cluster = i
        elif isinstance(raw_cluster, int):
            cluster = raw_cluster
        else:
            try:
                cluster = int(raw_cluster)
            except (TypeError, ValueError):
                cluster = i

        title = item.get("title")
        title = str(title) if title is not None else f"Cluster {cluster}"

        desc = item.get("description")
        desc = str(desc) if desc is not None else ""

        out.append({"cluster": cluster, "title": title, "description": desc})

    return out


def _generic_clusters(cluster_profiles: list) -> list:
    """Titulos genericos por cluster para la degradacion (sin LLM)."""
    out = []
    for i, prof in enumerate(cluster_profiles):
        cid = _cluster_id(prof or {}, i)
        out.append({"cluster": cid, "title": f"Cluster {cid}", "description": ""})
    return out


def describe_clusters_llm(
    cluster_profiles: list,
    feature_names: list,
    model: str = "claude-haiku-4-5-20251001",
) -> dict:
    """Describe los clusters de un KMeans con UNA sola llamada al LLM.

    Args:
        cluster_profiles: lista de perfiles de cluster (la forma que produce
            project_clusters_2d): cada uno {"cluster": int, "size": int,
            "pct": float, "centroid_original": {feature: media},
            "distinctive": [features], "centroid_z": {feature: z}}. Solo se le
            envia al LLM el resumen agregado, nunca filas crudas.
        feature_names: nombres de las features del dataset (contexto para el LLM).
        model: id del modelo Anthropic. Default claude-haiku-4-5-20251001
            (haiku, coste bajo).

    Returns:
        dict dict-no-throw: {"clusters": [{cluster:int, title:str, description:str}],
        "model": str, "note": str}. note == "" si todo fue bien; si el LLM no
        respondio o el parseo fallo, clusters trae titulos genericos "Cluster N" y
        note explica el motivo ("LLM no disponible" / "parse fallido"). Si
        cluster_profiles esta vacio o no es lista, devuelve clusters=[] sin llamar
        al LLM (note "sin clusters"). NUNCA lanza.
    """
    if not isinstance(cluster_profiles, list) or not cluster_profiles:
        return {"clusters": [], "model": model, "note": "sin clusters"}

    n = len(cluster_profiles)
    prompt = _build_prompt(cluster_profiles, feature_names)

    try:
        text = ask_llm(prompt, model=model, system=_SYSTEM, echo=False)
    except Exception:  # noqa: BLE001 — degradacion: cualquier fallo de red/LLM.
        text = ""

    parsed = _parse_clusters_json(text, n)
    if parsed:
        return {"clusters": parsed, "model": model, "note": ""}

    note = "LLM no disponible" if not text else "parse fallido"
    return {"clusters": _generic_clusters(cluster_profiles), "model": model, "note": note}