Add systemone: Jev/SystemOne-compatible request and response API

README.md

@@ -43,7 +43,7 @@ smaller, faster variant.
 |---|---|
 | `model-*.safetensors`, `model.safetensors.index.json`, `config.json`, `generation_config.json` | Backbone, including the vision encoder |
 | `joint_head.safetensors`, `joint_head_config.json` | Joint schema head |
-| `joint_schema_model.py` | Record encoding, batching, the model, and `load_release_model` |
+| `joint_schema_model.py` | Record encoding, batching, the model, `load_release_model`, and `systemone` |
 | `tokenizer.json`, `tokenizer_config.json`, `chat_template.jinja`, `processor_config.json` | Tokenizer and image/video processor |
 | `LICENSE` | Apache-2.0 license |
 
@@ -86,6 +86,33 @@ for question, question_logits in zip(encoded.questions, logits):
     print(question.question_id, dict(zip(question.option_ids, probabilities)))
 ```
 
+### Jev / SystemOne API
+
+`systemone` takes a Jev/SystemOne `POST /v1/systemone` request body and returns the same response
+body: `model`, `answers` keyed by question ID, and `usage`. A `choice` answer has `choice`,
+`confidence`, and `probabilities`; a `score` answer has the expected `score`, `confidence`, `legend`,
+and `probabilities`; a `noul` answer has the probability of true. `instructions` is optional, and
+`images` and `videos` may be added to the request.
+
+```python
+from joint_schema_model import systemone
+
+response = systemone(model, processor, {
+    "model": "clef",
+    "state": "Our checkout started returning errors and orders are blocked.",
+    "questions": {
+        "department": {
+            "type": "choice",
+            "instructions": "Which team should handle the message?",
+            "criteria": {"billing": "Payments or invoices", "technical": "Bugs or outages"},
+        },
+        "urgency": {"type": "score", "criteria": ["Can wait", "This week", "Today"]},
+        "outage": {"type": "noul", "instructions": "Is a service down?"},
+    },
+})
+print(response["answers"])
+```
+
 ### Images and video
 
 Add `images` (PIL images) or `videos` (frame arrays) to the record and pass the processor to

joint_schema_model.py

@@ -4,6 +4,8 @@ A record provides a ``state`` (any JSON value), optional ``images`` and ``videos
 and ``questions``. Each question has a ``type`` (``noul``, ``choice``, or ``score``),
 ``instructions``, and, for ``choice`` and ``score``, ``criteria`` describing the
 allowed options. The model returns one logit per allowed option for every question.
+``systemone`` answers a Jev/SystemOne ``/v1/systemone`` request body with the same
+response body.
 """
 
 from __future__ import annotations
@@ -115,7 +117,10 @@ def encode_record(
             )
         )
         question_start = len(schema_ids)
-        schema_ids.extend(_tokens(tokenizer, render(question["instructions"])))
+        instructions = question.get("instructions")
+        if instructions is None or instructions == "":
+            instructions = str(question_id)
+        schema_ids.extend(_tokens(tokenizer, render(instructions)))
         question_end = len(schema_ids)
         schema_ids.extend(_tokens(tokenizer, "\nALLOWED OPTIONS:\n"))
 
@@ -513,3 +518,59 @@ def load_release_model(
     head = head.to(device=device, dtype=dtype)
     processor = AutoProcessor.from_pretrained(path)
     return ClefModel(backbone, head).eval(), processor
+
+
+def systemone_answer(question: dict[str, Any], probabilities: dict[str, float]) -> dict[str, Any]:
+    """Convert per-option probabilities for one question into a SystemOne answer."""
+    if question["type"] == "noul":
+        return {"type": "noul", "noul": round(probabilities["true"], 4)}
+    if question["type"] == "choice":
+        options = [str(option) for option in question["criteria"]]
+        choice = max(options, key=probabilities.__getitem__)
+        return {
+            "type": "choice",
+            "choice": choice,
+            "confidence": round(probabilities[choice], 4),
+            "probabilities": {option: round(probabilities[option], 4) for option in options},
+        }
+    levels = [str(index) for index in range(len(question["criteria"]))]
+    return {
+        "type": "score",
+        "score": round(sum(index * probabilities[level] for index, level in enumerate(levels)), 4),
+        "confidence": round(max(probabilities[level] for level in levels), 4),
+        "legend": dict(zip(levels, question["criteria"])),
+        "probabilities": {level: round(probabilities[level], 4) for level in levels},
+    }
+
+
+@torch.inference_mode()
+def systemone(model: ClefModel, processor: Any, request: dict[str, Any], max_length: int = 16384) -> dict[str, Any]:
+    """Answer a Jev/SystemOne ``/v1/systemone`` request body with a SystemOne response body.
+
+    The request has ``model``, ``state``, and ``questions``, plus optional ``images`` and ``videos``.
+    """
+    questions = request.get("questions")
+    if not isinstance(request.get("model"), str) or "state" not in request:
+        raise ValueError("model and state are required")
+    if not isinstance(questions, dict) or not questions:
+        raise ValueError("at least one question is required")
+    for question_id, question in questions.items():
+        if question.get("type") not in QUESTION_TYPES:
+            raise ValueError(f"{question_id}: type must be noul, choice, or score")
+        if question["type"] != "noul" and not question.get("criteria"):
+            raise ValueError(f"{question_id}: criteria must not be empty")
+    encoded = encode_record(processor.tokenizer, request, max_length=max_length, processor=processor)
+    device = next(model.parameters()).device
+    logits = model(collate_records([encoded], processor.tokenizer.pad_token_id, device))[0]
+    answers = {
+        question.question_id: systemone_answer(
+            questions[question.question_id],
+            dict(zip(question.option_ids, question_logits.float().softmax(-1).tolist())),
+        )
+        for question, question_logits in zip(encoded.questions, logits)
+    }
+    return {
+        "model": request["model"],
+        "answers": answers,
+        "usage": {"input_tokens": len(encoded.input_ids), "output_tokens": 0},
+    }