File size: 2,213 Bytes
087e4a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any
from urllib import error, request
class ExternalCapabilityServiceError(RuntimeError):
"""Raised when an external capability service cannot satisfy a request."""
@dataclass(frozen=True)
class ExternalModelServiceClient:
"""Thin HTTP client for model-backed capabilities.
The client deliberately does not import transformers, load weights, perform a
health check, or start a model process. BGE-M3 and SigLIP 2 can therefore be
deployed later without changing the HCL Router/capability contract.
"""
base_url: str
timeout_seconds: float = 15.0
def semantic_search(self, payload: dict[str, Any]) -> dict[str, Any]:
return self._post("/v1/text/search", payload)
def cross_modal_match(self, payload: dict[str, Any]) -> dict[str, Any]:
return self._post("/v1/multimodal/match", payload)
def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
base_url = self.base_url.strip().rstrip("/")
if not base_url:
raise ExternalCapabilityServiceError("External capability service URL is not configured.")
encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8")
http_request = request.Request(
f"{base_url}{path}",
data=encoded,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with request.urlopen(http_request, timeout=max(float(self.timeout_seconds), 0.1)) as response:
body = response.read().decode("utf-8")
except (error.HTTPError, error.URLError, TimeoutError, OSError) as exc:
raise ExternalCapabilityServiceError(f"External capability request failed: {exc}") from exc
try:
decoded = json.loads(body)
except json.JSONDecodeError as exc:
raise ExternalCapabilityServiceError("External capability service returned invalid JSON.") from exc
if not isinstance(decoded, dict):
raise ExternalCapabilityServiceError("External capability service must return a JSON object.")
return decoded
|