feat(llm): añade MockProvider determinista con 3 escenarios canónicos

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-10 10:28:40 +02:00
parent bc16059738
commit 45b2b9367c
2 changed files with 166 additions and 0 deletions
@@ -0,0 +1,130 @@
+"""Proveedor LLM mock determinista para demos sin API keys y tests reproducibles."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from typing import Any
+
+from agentforge_core.llm.base import CompletionResult, Message
+
+_CANONICAL_RESPONSES: dict[str, dict[str, Any]] = {
+    "sip": {
+        "severity": "high",
+        "root_cause_hypothesis": (
+            "Tras el despliegue de la imagen 4.7.2 en CSCF, la pérdida de registros SIP "
+            "sugiere incompatibilidad de configuración con el HSS o un fallo en el "
+            "registro post-handover. Probablemente regresión introducida en la imagen."
+        ),
+        "proposed_actions": [
+            {
+                "id": "act-1",
+                "action": "rollback_image",
+                "target": "cscf-cluster-aravaca-01",
+                "risk_score": 4,
+                "rollback_plan": "Reaplicar imagen 4.7.1 desde el registry, validar registros SIP en 5 min.",
+                "requires_approval": True,
+            },
+            {
+                "id": "act-2",
+                "action": "drain_traffic_to_standby",
+                "target": "cscf-cluster-aravaca-02",
+                "risk_score": 3,
+                "rollback_plan": "Restaurar pesos LB al estado previo en panel SDN.",
+                "requires_approval": False,
+            },
+        ],
+    },
+    "mos": {
+        "severity": "medium",
+        "root_cause_hypothesis": (
+            "Degradación de MOS en pool SBC durante pico horario. Posible saturación "
+            "de codec G.711 o problema de jitter en el transit network."
+        ),
+        "proposed_actions": [
+            {
+                "id": "act-1",
+                "action": "scale_out_sbc_pool",
+                "target": "sbc-pool-borde-norte",
+                "risk_score": 2,
+                "rollback_plan": "Devolver réplicas al baseline previo (5 instancias).",
+                "requires_approval": False,
+            }
+        ],
+    },
+    "hss": {
+        "severity": "high",
+        "root_cause_hypothesis": (
+            "Alarma de capacidad replicada en HSS active-active sugiere que el split "
+            "de tráfico está saturando ambos nodos. Posible loop de replicación."
+        ),
+        "proposed_actions": [
+            {
+                "id": "act-1",
+                "action": "isolate_replication_link",
+                "target": "hss-pair-madrid",
+                "risk_score": 5,
+                "rollback_plan": "Restaurar enlace de replicación tras validar coherencia de subscriber DB.",
+                "requires_approval": True,
+            }
+        ],
+    },
+}
+
+
+_DEFAULT_RESPONSE: dict[str, Any] = {
+    "severity": "low",
+    "root_cause_hypothesis": "Sin patrón conocido. Recomendado análisis manual.",
+    "proposed_actions": [
+        {
+            "id": "act-1",
+            "action": "manual_investigation",
+            "target": "n/a",
+            "risk_score": 1,
+            "rollback_plan": "n/a",
+            "requires_approval": False,
+        }
+    ],
+}
+
+
+class MockProvider:
+    """Proveedor que mapea palabras clave a respuestas canónicas pregrabadas."""
+
+    name = "mock"
+
+    async def complete(
+        self,
+        messages: list[Message],
+        schema: dict[str, Any] | None = None,
+        temperature: float = 0.2,
+        max_tokens: int = 2000,
+    ) -> CompletionResult:
+        start = time.perf_counter()
+        last_user = next(
+            (m.content for m in reversed(messages) if m.role == "user"),
+            "",
+        )
+        body = self._select(last_user)
+        content = json.dumps(body, ensure_ascii=False)
+        elapsed = int((time.perf_counter() - start) * 1000)
+        return CompletionResult(
+            content=content,
+            model="mock-incident-analyzer-v1",
+            tokens_in=sum(len(m.content.split()) for m in messages),
+            tokens_out=len(content.split()),
+            latency_ms=elapsed,
+        )
+
+    @staticmethod
+    def _select(user_input: str) -> dict[str, Any]:
+        text = user_input.lower()
+        for key, body in _CANONICAL_RESPONSES.items():
+            if key in text:
+                return body
+        # fallback determinista por hash (útil para tests con random text)
+        digest = hashlib.sha256(text.encode()).hexdigest()
+        if digest.startswith(("0", "1", "2", "3")):
+            return _CANONICAL_RESPONSES["sip"]
+        return _DEFAULT_RESPONSE