72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
"""开箱可用性验证：从生产 bootstrap 组装，到 Agent 能力实际可调用。
|
|
|
|
不注入替身，直接使用 `get_agent_factory()`，验证组员拿到底座后的真实起点。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from uuid import uuid4
|
|
|
|
from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
|
|
from app.service.agent.base import BaseAgent
|
|
from app.service.agent.bootstrap import get_agent_factory
|
|
|
|
|
|
class ProbeAgent(BaseAgent):
|
|
definition = AgentDefinition(
|
|
agent_type="probe_agent",
|
|
version="1",
|
|
allowed_roles=("customer",),
|
|
allowed_portals=("api",),
|
|
allowed_tools=("check_suitability",),
|
|
supported_intents=("general",),
|
|
)
|
|
|
|
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
|
del request, context
|
|
return CoreResult(text="ok")
|
|
|
|
|
|
async def main() -> None:
|
|
factory = get_agent_factory()
|
|
|
|
print("1) 生产工厂组装状态")
|
|
print(" model_service :", type(factory._model_service).__name__)
|
|
print(" tool_executor :", type(factory._tool_executor).__name__)
|
|
print(" governance :", type(factory._governance).__name__)
|
|
|
|
factory.register(ProbeAgent.definition, lambda _c: ProbeAgent(ProbeAgent.definition))
|
|
context = RequestContext(
|
|
user_id="9001",
|
|
trace_id=str(uuid4()),
|
|
roles=("customer",),
|
|
permissions=("agent:run", "suitability:read"),
|
|
)
|
|
|
|
agent = factory.create("probe_agent", context)
|
|
print("\n2) create() 后的能力绑定")
|
|
print(" _model_service :", type(agent._model_service).__name__ if agent._model_service else None)
|
|
print(" _tool_executor :", type(agent._tool_executor).__name__ if agent._tool_executor else None)
|
|
|
|
await agent.resolve_config(context)
|
|
print("\n3) 配置解析结果")
|
|
print(" config_version :", agent.config.config_version if agent.config else None)
|
|
print(" tools_by_intent:", agent.config.allowed_tools_by_intent if agent.config else None)
|
|
|
|
print("\n4) 未发布配置时的工具调用")
|
|
try:
|
|
output = await agent.call_tool(
|
|
"check_suitability",
|
|
{"customer_risk_level": 1, "product_risk_level": 3},
|
|
intent="general",
|
|
context=context,
|
|
)
|
|
print(" 结果 : 调用成功 ->", output)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f" 结果 : {type(exc).__name__}: {exc}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|