From 45b2b9367cee26e4495888cf22b122e6e6aa5cf4 Mon Sep 17 00:00:00 2001 From: Juan Date: Sun, 10 May 2026 10:28:40 +0200 Subject: [PATCH] =?UTF-8?q?feat(llm):=20a=C3=B1ade=20MockProvider=20determ?= =?UTF-8?q?inista=20con=203=20escenarios=20can=C3=B3nicos?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 --- core/src/agentforge_core/llm/mock.py | 130 +++++++++++++++++++++++++++ tests/unit/test_llm_mock.py | 36 ++++++++ 2 files changed, 166 insertions(+) create mode 100644 core/src/agentforge_core/llm/mock.py create mode 100644 tests/unit/test_llm_mock.py diff --git a/core/src/agentforge_core/llm/mock.py b/core/src/agentforge_core/llm/mock.py new file mode 100644 index 0000000..024311c --- /dev/null +++ b/core/src/agentforge_core/llm/mock.py @@ -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 diff --git a/tests/unit/test_llm_mock.py b/tests/unit/test_llm_mock.py new file mode 100644 index 0000000..997c1b6 --- /dev/null +++ b/tests/unit/test_llm_mock.py @@ -0,0 +1,36 @@ +"""Tests del proveedor LLM mock determinista.""" + +import json + +import pytest + +from agentforge_core.llm.base import Message +from agentforge_core.llm.mock import MockProvider + + +@pytest.fixture +def provider() -> MockProvider: + return MockProvider() + + +async def test_mock_provider_es_determinista(provider: MockProvider) -> None: + msgs = [Message(role="user", content="caída registros SIP")] + r1 = await provider.complete(msgs) + r2 = await provider.complete(msgs) + assert r1.content == r2.content + + +async def test_mock_provider_devuelve_json_valido(provider: MockProvider) -> None: + msgs = [Message(role="user", content="degradación MOS pool SBC")] + r = await provider.complete(msgs) + parsed = json.loads(r.content) + assert "severity" in parsed + assert "proposed_actions" in parsed + assert isinstance(parsed["proposed_actions"], list) + + +async def test_mock_provider_respuesta_canonica_si_no_match(provider: MockProvider) -> None: + msgs = [Message(role="user", content="texto random sin patrón")] + r = await provider.complete(msgs) + parsed = json.loads(r.content) + assert parsed["severity"] in {"low", "medium", "high", "critical"}