File size: 3,272 Bytes
78ac017 | 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 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """Strict adapter from a sealed semantic release to Qwen messages."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from PIL import Image
from viss_core.data_v0.semantic_release import (
infer_semantic_release_root,
load_semantic_sft_rows,
)
def load_training_rows(
path: Path, *, allow_nontrainable: bool = False
) -> list[dict[str, Any]]:
"""Load only a validated, training-authorized semantic release."""
if allow_nontrainable:
raise RuntimeError("the release trainer has no non-trainable bypass")
path = path.resolve()
if path.suffix != ".jsonl":
raise ValueError("semantic training accepts canonical JSONL only")
return load_semantic_sft_rows(path)
def infer_release_root(path: Path) -> Path | None:
return infer_semantic_release_root(path)
def normalize_messages(
messages: Any,
*,
media_root: Path | None = None,
) -> list[dict[str, Any]]:
"""Resolve release-local images and preserve assistant-only loss ownership."""
if not isinstance(messages, list):
raise TypeError("semantic messages must be a list")
roles = [message.get("role") for message in messages]
if roles != ["user", "assistant"]:
raise ValueError("semantic messages require user then assistant")
root = media_root.resolve() if media_root is not None else None
if root is None:
raise RuntimeError("semantic media requires an explicit release root")
output: list[dict[str, Any]] = []
for message in messages:
raw_content = message.get("content")
if not isinstance(raw_content, list) or not raw_content:
raise TypeError("semantic message content must be a non-empty list")
content: list[dict[str, Any]] = []
for item in raw_content:
if not isinstance(item, dict):
raise TypeError("semantic content item must be an object")
if item.get("type") == "text":
content.append({"type": "text", "text": str(item.get("text") or "")})
continue
if item.get("type") != "image" or not isinstance(item.get("image"), str):
raise ValueError("semantic content item has an unsupported type")
if message["role"] != "user":
raise ValueError("only user states may contain semantic images")
relative = Path(item["image"])
if relative.is_absolute() or ".." in relative.parts:
raise RuntimeError("semantic image path must be release-relative")
image_path = (root / relative).resolve(strict=True)
try:
image_path.relative_to(root)
except ValueError as error:
raise RuntimeError("semantic image path escapes the release") from error
if not image_path.is_file() or image_path.is_symlink():
raise RuntimeError("semantic image is missing or unsafe")
content.append({"type": "image", "image": Image.open(image_path).convert("RGB")})
output.append(
{
"role": str(message["role"]),
"content": content,
"loss": bool(message.get("loss")),
}
)
return output
|