q792602257
ca5e300289
- _with_known_widgets_only used to discard the entire SemanticScene (returning None) when filtering dangling-element-id widgets left an empty list, losing valid page/intents. Now returns a SemanticScene with widgets=[] instead, matching design.md's 'dropped, not a hard failure' intent. - Removed _SDK_UNAVAILABLE_EXCEPTION_NAMES, dead code left over from an abandoned name-based exception-matching approach; the blanket except Exception already maps every SDK failure correctly. openspec: semantic-scene capability, archived change semantic-scene-runtime
137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from semantic.config import DEFAULT_MODEL
|
|
from semantic.models import SEMANTIC_SCENE_SCHEMA, SemanticScene
|
|
from semantic.prompts import ENRICHMENT_SYSTEM_PROMPT, scene_user_prompt
|
|
|
|
|
|
class EnrichmentUnavailable(Exception):
|
|
"""Internal signal for expected enrichment transport/response failures."""
|
|
|
|
|
|
class AnthropicSemanticClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
model: str = DEFAULT_MODEL,
|
|
transport: Any | None = None,
|
|
max_tokens: int = 1024,
|
|
) -> None:
|
|
self.model = model
|
|
self._transport = transport
|
|
self.max_tokens = max_tokens
|
|
|
|
def enrich(self, scene_json: dict[str, Any], *, timeout: float) -> dict[str, Any]:
|
|
try:
|
|
response = self._create_message(scene_json, timeout=timeout)
|
|
payload = _extract_response_body(response)
|
|
return _validate_payload(payload)
|
|
except EnrichmentUnavailable:
|
|
raise
|
|
except Exception as exc:
|
|
raise EnrichmentUnavailable(str(exc)) from exc
|
|
|
|
def _create_message(self, scene_json: dict[str, Any], *, timeout: float) -> Any:
|
|
client = self._client()
|
|
kwargs = {
|
|
"model": self.model,
|
|
"max_tokens": self.max_tokens,
|
|
"timeout": timeout,
|
|
"system": [
|
|
{
|
|
"type": "text",
|
|
"text": ENRICHMENT_SYSTEM_PROMPT,
|
|
"cache_control": {"type": "ephemeral"},
|
|
}
|
|
],
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": scene_user_prompt(scene_json),
|
|
}
|
|
],
|
|
}
|
|
],
|
|
"output_config": {
|
|
"format": {
|
|
"type": "json_schema",
|
|
"schema": SEMANTIC_SCENE_SCHEMA,
|
|
}
|
|
},
|
|
}
|
|
messages = getattr(client, "messages", None)
|
|
if messages is not None:
|
|
return messages.create(**kwargs)
|
|
return client.create(**kwargs)
|
|
|
|
def _client(self) -> Any:
|
|
if self._transport is not None:
|
|
return self._transport
|
|
|
|
try:
|
|
import anthropic
|
|
except Exception as exc:
|
|
raise EnrichmentUnavailable("anthropic SDK is unavailable") from exc
|
|
|
|
self._transport = anthropic.Anthropic()
|
|
return self._transport
|
|
|
|
|
|
def _extract_response_body(response: Any) -> dict[str, Any]:
|
|
if _looks_like_semantic_scene(response):
|
|
return response
|
|
|
|
for key in ("output", "parsed", "json"):
|
|
value = _value(response, key)
|
|
if _looks_like_semantic_scene(value):
|
|
return value
|
|
|
|
content = _value(response, "content")
|
|
if _looks_like_semantic_scene(content):
|
|
return content
|
|
if isinstance(content, str):
|
|
return _decode_json(content)
|
|
if isinstance(content, list):
|
|
for block in content:
|
|
for key in ("parsed", "json", "input", "content"):
|
|
value = _value(block, key)
|
|
if _looks_like_semantic_scene(value):
|
|
return value
|
|
text = _value(block, "text")
|
|
if isinstance(text, str):
|
|
return _decode_json(text)
|
|
|
|
raise ValueError("structured semantic response body not found")
|
|
|
|
|
|
def _decode_json(text: str) -> dict[str, Any]:
|
|
decoded = json.loads(text)
|
|
if not isinstance(decoded, dict):
|
|
raise ValueError("structured semantic response must be a JSON object")
|
|
return decoded
|
|
|
|
|
|
def _validate_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
|
return SemanticScene.from_dict(payload).to_dict()
|
|
|
|
|
|
def _looks_like_semantic_scene(value: Any) -> bool:
|
|
return isinstance(value, dict) and {
|
|
"page",
|
|
"intents",
|
|
"widgets",
|
|
}.issubset(value)
|
|
|
|
|
|
def _value(source: Any, key: str) -> Any:
|
|
if isinstance(source, dict):
|
|
return source.get(key)
|
|
value = getattr(source, key, None)
|
|
return None if callable(value) else value
|