78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
"""Cached, role-scoped structured-output conformance checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from hashlib import sha256
|
|
import json
|
|
from typing import Any, Literal
|
|
|
|
from app.cad_agent.application.authoring_contract import AuthoringDocument
|
|
from app.cad_agent.application.workflow import RequirementsAnalysis
|
|
from app.cad_agent.ports import CadRuntime, ModelGateway
|
|
|
|
|
|
CapabilityRole = Literal["author"]
|
|
|
|
|
|
def conformance_tools(runtime: CadRuntime, *, role: CapabilityRole) -> list[dict[str, Any]]:
|
|
if not runtime.supported_atomic_ids():
|
|
raise RuntimeError("Runtime has no operations for conformance")
|
|
return [
|
|
_tool("analyze_requirements", RequirementsAnalysis.model_json_schema()),
|
|
_tool("write_authoring_cdsl", AuthoringDocument.model_json_schema()),
|
|
]
|
|
|
|
|
|
def conformance_hash(tools: list[dict[str, Any]], *, role: CapabilityRole) -> str:
|
|
payload = {"protocol": "cad.single-stage.v1", "role": role, "tools": tools}
|
|
return sha256(json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def cached_model_capability(
|
|
repository: Any,
|
|
runtime: CadRuntime,
|
|
*,
|
|
provider_id: str,
|
|
model_id: str,
|
|
role: CapabilityRole,
|
|
) -> dict[str, Any] | None:
|
|
tools = conformance_tools(runtime, role=role)
|
|
schema_hash = conformance_hash(tools, role=role)
|
|
cached = repository.model_capability(provider_id, model_id, schema_hash)
|
|
if cached is None:
|
|
return None
|
|
return {"cached": True, "schema_hash": schema_hash, "role": role, **cached["report"]}
|
|
|
|
|
|
async def verify_model_capability(
|
|
repository: Any,
|
|
runtime: CadRuntime,
|
|
models: ModelGateway,
|
|
*,
|
|
provider_id: str,
|
|
model_id: str,
|
|
role: CapabilityRole,
|
|
force: bool = False,
|
|
) -> dict[str, Any]:
|
|
tools = conformance_tools(runtime, role=role)
|
|
schema_hash = conformance_hash(tools, role=role)
|
|
cached = repository.model_capability(provider_id, model_id, schema_hash)
|
|
if cached is not None and not force:
|
|
return {"cached": True, "schema_hash": schema_hash, "role": role, **cached["report"]}
|
|
report = await models.conformance(provider_id=provider_id, model_id=model_id, tools=tools)
|
|
report = {"schema_hash": schema_hash, "role": role, "tool_count": len(tools), **report}
|
|
if not report.get("probe_unavailable"):
|
|
repository.record_model_capability(provider_id, model_id, schema_hash, report)
|
|
return report
|
|
|
|
|
|
def _tool(name: str, parameters: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": name,
|
|
"description": "Structured output conformance probe. Return one schema-valid call.",
|
|
"parameters": parameters,
|
|
},
|
|
}
|