adjustments: manual v3

This commit is contained in:
2026-09-12 15:33:55 -03:00
parent 3cb970d8ab
commit db16d478ba
19 changed files with 365 additions and 33 deletions

View File

@@ -17,6 +17,7 @@ if str(FRAMEWORK_SRC) not in sys.path:
from agent_framework.workflows.models import WorkflowDefinition
from agent_framework.idempotency import create_idempotency_store
from agent_framework.supervisor.supervisor import Supervisor
PT = ROOT / "docs" / "developer" / "pt"
@@ -108,3 +109,36 @@ def test_new_chapters_have_english_counterparts_and_are_indexed() -> None:
for pt_name, en_name in pairs.items():
assert (PT / pt_name).is_file() and pt_name in pt_index
assert (ROOT / "docs" / "developer" / "en" / en_name).is_file() and en_name in en_index
@pytest.mark.asyncio
async def test_supervisor_accepts_application_rules_without_core_hardcode() -> None:
supervisor = Supervisor(
routing_rules=[("financial_analysis", "financeiro_agent", ["saldo", "extrato"])],
fallback_agent="financeiro_agent",
)
matched = await supervisor.route_plan({"user_text": "Quero consultar meu saldo"})
fallback = await supervisor.route_plan({"user_text": "mensagem sem keyword"})
assert matched.agents == ["financeiro_agent"]
assert matched.intent == "financial_analysis"
assert fallback.agents == ["financeiro_agent"]
def test_template_builds_supervisor_from_routing_yaml_intents() -> None:
graph = (ROOT / "templates" / "agent_template_backend" / "app" / "workflows" / "agent_graph.py").read_text(encoding="utf-8")
pt = (PT / "02_routing_stickiness_and_intent_shift.md").read_text(encoding="utf-8")
en = (ROOT / "docs" / "developer" / "en" / "02_routing_stickiness_and_intent_shift.md").read_text(encoding="utf-8")
assert "for intent in self.router.intents" in graph
assert "routing_rules=[" in graph
assert "adicione uma regra" not in pt
assert "add a rule with intent" not in en
def test_all_tuning_templates_build_supervisor_from_routing_yaml() -> None:
graphs = sorted((ROOT / "Tuning-Performance").glob("**/app/workflows/agent_graph.py"))
assert graphs, "Nenhuma cópia executável do template encontrada em Tuning-Performance"
for graph_path in graphs:
source = graph_path.read_text(encoding="utf-8")
assert "for intent in self.router.intents" in source, str(graph_path.relative_to(ROOT))
assert "fallback_agent=self.router.fallback_agent" in source, str(graph_path.relative_to(ROOT))
assert source.index("self.router = EnterpriseRouter") < source.index("self.supervisor = Supervisor")

View File

@@ -25,17 +25,38 @@ def _block(tag: str) -> str:
@pytest.fixture()
def documented_agent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
def documented_agent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
for name in tuple(sys.modules):
if name == "financial_agent" or name.startswith("financial_agent."):
sys.modules.pop(name, None)
package = tmp_path / "financial_agent" / "extensions"
package.mkdir(parents=True)
prompts = package / "prompts"
prompts.mkdir()
(tmp_path / "financial_agent" / "__init__.py").write_text("", encoding="utf-8")
(package / "__init__.py").write_text("", encoding="utf-8")
(prompts / "__init__.py").write_text("", encoding="utf-8")
(prompts / "financial_policy.py").write_text(_block("external-guardrail-prompt-example"), encoding="utf-8")
(package / "guardrails.py").write_text(_block("external-guardrail-example"), encoding="utf-8")
(package / "llm_guardrails.py").write_text(_block("external-llm-guardrail-example"), encoding="utf-8")
(package / "judges.py").write_text(_block("external-judge-example"), encoding="utf-8")
(tmp_path / "guardrails.yaml").write_text(_block("external-guardrail-yaml"), encoding="utf-8")
(tmp_path / "judges.yaml").write_text(_block("external-judge-yaml"), encoding="utf-8")
monkeypatch.syspath_prepend(str(tmp_path))
return tmp_path
yield tmp_path
for name in tuple(sys.modules):
if name == "financial_agent" or name.startswith("financial_agent."):
sys.modules.pop(name, None)
class FakeGuardrailLLM:
def __init__(self, response: str) -> None:
self.response = response
self.calls: list[tuple[list[dict[str, str]], dict]] = []
async def ainvoke(self, messages, **kwargs):
self.calls.append((messages, kwargs))
return self.response
@pytest.mark.asyncio
@@ -51,6 +72,30 @@ async def test_documented_external_guardrail_loads_and_enforces_policy(documente
assert missing[0].code == "FIN_AMOUNT" and not missing[0].allowed
@pytest.mark.asyncio
async def test_documented_llm_guardrail_uses_agent_prompt_and_shared_llm(documented_agent: Path) -> None:
from financial_agent.extensions.llm_guardrails import FinancialPolicyLLMRail
llm = FakeGuardrailLLM('{"allowed": false, "reason": "Operação fora da política."}')
pipeline = GuardrailPipeline(tool_rails=[FinancialPolicyLLMRail()], llm=llm)
_, decisions = await pipeline.run_tool("transfer", {"amount": 2000.0})
assert not decisions[0].allowed
assert decisions[0].metadata["prompt_version"] == "financial-policy-v1"
messages, kwargs = llm.calls[0]
assert "classificador de política financeira" in messages[0]["content"]
assert kwargs["profile_name"] == "guardrail"
@pytest.mark.asyncio
async def test_documented_llm_guardrail_fails_closed_without_llm(documented_agent: Path) -> None:
from financial_agent.extensions.llm_guardrails import FinancialPolicyLLMRail
decision = await FinancialPolicyLLMRail().evaluate("transfer", {"tool_name": "transfer"})
assert not decision.allowed
assert "não foi disponibilizado" in decision.reason
@pytest.mark.asyncio
async def test_documented_external_judge_loads_with_injection(documented_agent: Path) -> None:
fake_llm = object()