feat(guardrails): añade NeMo stub, CompositeEngine paralelo y factory

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Juan
2026-05-10 10:51:17 +02:00
co-authored by Claude Opus 4.7
parent 79c6fe32ff
commit af8036c2d2
4 changed files with 174 additions and 0 deletions
+44
View File
@@ -0,0 +1,44 @@
"""Tests del CompositeGuardrailEngine."""
from uuid import uuid4
import pytest
from agentforge_core.domain.policy import PolicyDefinition
from agentforge_core.guardrails.composite import CompositeGuardrailEngine
from agentforge_core.guardrails.guardrails_ai import GuardrailsAIEngine
from agentforge_core.guardrails.nemo import NeMoGuardrailsEngine
@pytest.fixture
def policy() -> PolicyDefinition:
return PolicyDefinition(
name="t",
version="v1",
description="t",
input_validators=[],
output_validators=[],
)
async def test_composite_agrega_violaciones_de_todos(policy: PolicyDefinition) -> None:
engine = CompositeGuardrailEngine(
[GuardrailsAIEngine(), NeMoGuardrailsEngine(allowed_keywords=["sip"])]
)
violations = await engine.validate_input("texto sin tema", policy, uuid4())
assert isinstance(violations, list)
# NeMo debe marcar off-topic (warning, no bloqueo)
assert any(v.validator == "NeMoTopicalRails" for v in violations)
async def test_composite_no_agrega_si_match_topic(policy: PolicyDefinition) -> None:
engine = CompositeGuardrailEngine(
[GuardrailsAIEngine(), NeMoGuardrailsEngine(allowed_keywords=["sip"])]
)
violations = await engine.validate_input("incidente sip en cscf", policy, uuid4())
assert not [v for v in violations if v.validator == "NeMoTopicalRails"]
def test_composite_rechaza_lista_vacia() -> None:
with pytest.raises(ValueError):
CompositeGuardrailEngine([])