62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
import pytest
|
|
|
|
from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
|
|
from app.service.agent.base import BaseAgent
|
|
from app.service.agent.factory import AgentFactory
|
|
from app.service.intent_classifier import IntentClassifier
|
|
from app.service.model_gateway import ModelExecution
|
|
|
|
|
|
class DemoAgent(BaseAgent):
|
|
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
|
return CoreResult(text="ok")
|
|
|
|
|
|
class StubModel:
|
|
async def generate(self, _endpoints: list[object], _prompt: str) -> ModelExecution:
|
|
return ModelExecution("intent", '{"intent":"profile","confidence":0.9}', 1)
|
|
|
|
|
|
class StubResolver:
|
|
async def resolve(self, *, agent_type: str, task_type: str) -> list[object]:
|
|
assert agent_type == "demo"
|
|
assert task_type == "intent_classification"
|
|
return [object()]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_classifies_before_handle_and_attaches_result(governance) -> None:
|
|
definition = AgentDefinition(
|
|
agent_type="demo", version="1", allowed_roles=("customer",),
|
|
allowed_portals=("api",), supported_intents=("profile", "general")
|
|
)
|
|
factory = AgentFactory(
|
|
governance=governance,
|
|
intent_classifier=IntentClassifier(StubModel()),
|
|
intent_endpoint_resolver=StubResolver(),
|
|
)
|
|
factory.register(definition, lambda _context: DemoAgent(definition))
|
|
context = RequestContext(
|
|
user_id="1", trace_id="intent", roles=("customer",), permissions=("agent:run",)
|
|
)
|
|
request = AgentRequest(
|
|
agent_type="demo", message="我的资料", session_id="s",
|
|
idempotency_key="intent-request-0001"
|
|
)
|
|
events = [event async for event in factory.create("demo", context).execute(
|
|
request, context, "run"
|
|
)]
|
|
result = events[-1].payload["result"]["result"]
|
|
assert result["intent"]["intent"] == "profile"
|
|
assert result["intent"]["confidence"] == 0.9
|
|
|
|
|
|
def test_business_agent_cannot_override_intent_governance() -> None:
|
|
with pytest.raises(TypeError, match="classify_intent"):
|
|
class InvalidAgent(BaseAgent):
|
|
async def classify_intent(self, request: AgentRequest):
|
|
return None
|
|
|
|
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
|
return CoreResult(text="bad")
|