Benjamin Admin
32e45f0797
Consolidate the AGB C-lean engine (71% FP -> ~0, validated vs 7-company Opus GT) onto the canonical checker library and into the live check path. - AGBAgent.evaluate now runs routed C-lean: keyword (L1/L2) -> business- model gate -> per-item decision_method routing (embedding/reference/llm via services/checkers/) -> severity re-tiering (LOW -> recommendation), honoring context.skip_llm. - New agb/_pipeline.py orchestrates the routing; agent.py stays thin. - Remove the 3 AGB-local checker duplicates (_reference_check, _embedding_rescue, _llm_judge); services/checkers/ is now canonical. - Wire "agb" into _agent_outputs._TOPIC_AGENTS so the live check emits a validated AGB tab (was snapshot-only). - Run topic agents concurrently (asyncio.gather) + emit each tab via SSE as it finishes -> progressive results, no wait on the slowest agent. - Tests: checker units (mocked), routed agent (gate/rescue/re-tier), topic wiring; existing AGB tests made offline-safe. dev-only, no deploy. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
"""AGB routed-Pipeline: Gate, Reference-/Embedding-Rescue, LLM-skip, Re-Tiering.
|
|
Embedding + LLM offline-gestubbt → deterministisch, kein Netzwerk (Reference = echtes Regex)."""
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import compliance.services.specialist_agents.agb._pipeline as pipeline
|
|
from compliance.services.checkers.base import CheckResult
|
|
from compliance.services.specialist_agents._base import AgentInput
|
|
from compliance.services.specialist_agents.agb.agent import AGBAgent
|
|
|
|
|
|
class _Stub:
|
|
def __init__(self, present):
|
|
self._p = present
|
|
|
|
async def check(self, ctrl, doc):
|
|
return CheckResult(present=self._p)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _offline(monkeypatch):
|
|
monkeypatch.setattr(pipeline, "_EMB", _Stub(None))
|
|
monkeypatch.setattr(pipeline, "_LLM", _Stub(None))
|
|
|
|
|
|
def _routed(field_ids, text, context=None):
|
|
findings = [SimpleNamespace(field_id=fid) for fid in field_ids]
|
|
return asyncio.run(pipeline.run_routed(findings, text, context or {}))
|
|
|
|
|
|
def test_gate_termination_na_for_oneoff_shop():
|
|
text = "Widerrufsbelehrung: Sie koennen binnen 14 Tagen widerrufen. " * 5
|
|
kept, resolved, gated = _routed(["termination", "termination_form"], text)
|
|
assert set(gated) == {"termination", "termination_form"}
|
|
assert kept == []
|
|
|
|
|
|
def test_reference_rescues_data_protection():
|
|
text = "Einzelheiten zur Verarbeitung in unserer Datenschutzerklaerung. " * 5
|
|
kept, resolved, gated = _routed(["data_protection"], text)
|
|
assert "data_protection" in resolved and kept == []
|
|
|
|
|
|
def test_embedding_rescue_resolves(monkeypatch):
|
|
monkeypatch.setattr(pipeline, "_EMB", _Stub(True))
|
|
kept, resolved, gated = _routed(["scope"], "x" * 200)
|
|
assert "scope" in resolved
|
|
|
|
|
|
def test_llm_skipped_keeps_finding():
|
|
kept, resolved, gated = _routed(["delivery_timeframe"], "x" * 200, {"skip_llm": True})
|
|
assert [f.field_id for f in kept] == ["delivery_timeframe"] and resolved == []
|
|
|
|
|
|
def test_evaluate_retiers_low_out_of_findings():
|
|
text = ("Allgemeine Geschaeftsbedingungen. Vertragsschluss durch Bestellung. "
|
|
"Haftung beschraenkt. Gerichtsstand Muenchen. ") * 6
|
|
out = asyncio.run(AGBAgent().evaluate(AgentInput(doc_type="agb", text=text)))
|
|
assert out.agent == "agb" and out.agent_version == "2.0"
|
|
assert all(f.severity in ("HIGH", "MEDIUM") for f in out.findings)
|