oev/serve.py

"""Minimal HTTP server for OEV decisions.

    python -m oev.serve --checkpoint checkpoints_td5/oev-tiny.pt --port 8000
    python -m oev.serve --onnx exported.int8.onnx

/decide speaks the native schema; /v1/systemone accepts the Jev/TypeSafe
criteria schema, so existing clients work by changing baseUrl.
"""

import argparse
import os

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from oev.evaluate import pack_question
from oev.presets import gate

app = FastAPI(title="OEV", description="System One decision model: typed questions in, calibrated probabilities out.")

# set in main(): TorchBackend (checkpoint) or OnnxBackend (int8 export)
agent = None


class DecideRequest(BaseModel):
    state: object = Field(..., description="Any state: text, dict of fields.")
    questions: dict
    threshold: float = 0.85


@app.post("/decide")
def decide(req: DecideRequest):
    if agent is None:
        raise HTTPException(status_code=503, detail="no checkpoint loaded - run oev-serve --checkpoint ...")
    answers = agent.decide(req.state, req.questions)
    gated = gate(answers, threshold=req.threshold)
    return {
        "answers": answers,
        "gating": {name: {"confidence_threshold": req.threshold, "automatable": ok}
                   for name, _, ok in gated},
    }


@app.get("/health")
def health():
    if agent is None:
        # 503 (not 200) so Docker HEALTHCHECK and orchestrators can tell
        # "still loading" apart from "failed to load" during startup
        raise HTTPException(status_code=503, detail="model not loaded yet")
    return {"status": "ok", **agent.describe()}


# Jev-compatible endpoint (TypeSafe System One schema)

class SystemOneRequest(BaseModel):
    model: str = "oev"
    state: object = Field(..., description="Any state: text, dict of fields.")
    questions: dict


def _jevify(answers: dict) -> dict:
    # map OEV answers onto the Jev System One response shape
    out = {}
    for name, a in answers.items():
        if isinstance(a, dict) and "probabilities" in a:
            choice = a.get("choice") or a.get("value")
            out[name] = {
                "answer": choice,
                "confidence": a.get("confidence"),
                "probabilities": a["probabilities"],
            }
        else:
            # noul returns a probability directly
            out[name] = {"answer": bool(a >= 0.5), "confidence": a if a >= 0.5 else 1 - a,
                         "probabilities": {"yes": a, "no": 1 - a}}
    return out


def _validate_jev_questions(questions: dict) -> str | None:
    # return an error string for anything the Jev schema does not allow
    for name, question in questions.items():
        if not isinstance(question, dict):
            return f"question {name} must be an object"
        qtype = question.get("type")
        if qtype not in {"choice", "noul", "score"}:
            return f"question {name} has unsupported type"
        if qtype in {"choice", "noul"}:
            criteria = question.get("criteria")
            if criteria is not None and (not isinstance(criteria, dict) or not criteria):
                return f"question {name} ({qtype}) criteria must be a non-empty object"
        else:
            criteria = question.get("criteria")
            if not isinstance(criteria, list) or not criteria:
                return f"question {name} (score) needs a non-empty criteria list of level labels"
    return None


@app.post("/v1/systemone")
def systemone(req: SystemOneRequest):
    # drop-in for TypeSafe clients: same request/response shape as Jev
    if agent is None:
        raise HTTPException(status_code=503, detail="no checkpoint loaded - run oev-serve --checkpoint ...")
    error = _validate_jev_questions(req.questions)
    if error:
        raise HTTPException(status_code=422, detail=error)
    answers = agent.decide(req.state, req.questions)
    return {
        "model": req.model,
        "answers": _jevify(answers),
        "routing": {"model": "oev", "checkpoint": "local"},
    }


class TorchBackend:
    # thin adapter so /health works the same for both backends
    def __init__(self, checkpoint, device="cpu"):
        from oev.infer import OEV

        self.inner = OEV(checkpoint, device=device)

    def decide(self, state, questions):
        return self.inner.decide(state, questions)

    def describe(self):
        return {"backend": "torch", "checkpoint": self.inner.model.cfg}


class OnnxBackend:
    # torch-free mirror of oev.infer.OEV.decide: pack -> logits -> softmax.
    # The question normalization is duplicated here on purpose: oev.infer
    # imports torch at module level and this backend must not.
    def __init__(self, path, backbone="microsoft/deberta-v3-base", max_len=768, threads=None):
        import numpy as np
        import onnxruntime as ort

        self.np = np
        so = ort.SessionOptions()
        if threads:
            so.intra_op_num_threads = threads
        self.sess = ort.InferenceSession(path, sess_options=so, providers=["CPUExecutionProvider"])
        self.path = path
        self.backbone = backbone
        self.max_len = max_len
        from oev.tokenizer_hf import HFTokenPacker

        self.packer = HFTokenPacker(backbone)

    def describe(self):
        return {"backend": "onnx", "checkpoint": self.path,
                "backbone": self.backbone, "max_len": self.max_len}

    @staticmethod
    def _normalize_question(name, q):
        q = dict(q)
        criteria = q.get("criteria")
        if q.get("type") == "choice" and "options" not in q:
            if isinstance(criteria, dict) and criteria:
                q["options"] = [str(option) for option in criteria]
            else:
                raise ValueError(f"question {name!r} (choice) needs options or a criteria map")
        elif q.get("type") == "score" and "levels" not in q:
            if isinstance(criteria, list) and criteria:
                q["levels"] = list(criteria)
            else:
                raise ValueError(f"question {name!r} (score) needs levels or a criteria list")
        return q

    def _probs(self, state, pq):
        ids, anchors, _ = self.packer.pack(state, pq, self.max_len)
        np = self.np
        feed = {
            "ids": np.asarray([ids], dtype=np.int64),
            "pad_mask": np.zeros((1, len(ids)), dtype=bool),
            "anchor_pos": np.asarray([anchors], dtype=np.int64),
        }
        logits = self.sess.run(None, feed)[0][0]
        top = logits.max()
        exp = np.exp(logits - top)
        return (exp / exp.sum()).tolist()

    def decide(self, state, questions):
        out = {}
        for name, raw in questions.items():
            q = self._normalize_question(name, raw)
            if q["type"] == "noul":
                options = ["no", "yes"]
            elif q["type"] == "choice":
                options = [str(option) for option in q["options"]]
            else:
                options = [str(level) for level in q["levels"]]
            pq = pack_question(dict(q, name=name, options=options,
                                    answer=options[0]))
            probs = self._probs(state, pq)
            best = max(range(len(probs)), key=probs.__getitem__)
            if q["type"] == "noul":
                out[name] = float(probs[1])
            elif q["type"] == "score":
                levels = [str(level) for level in q["levels"]]
                out[name] = {"value": levels[best],
                             "probabilities": {levels[i]: float(p) for i, p in enumerate(probs)}}
            else:
                out[name] = {"choice": pq["options"][best],
                             "probabilities": {o: float(p) for o, p in zip(pq["options"], probs)},
                             "confidence": float(max(probs))}
        return out


def _resolve_checkpoint(spec: str) -> str:
    # accept a local path or a Hugging Face id (repo/filename)
    if "/" not in spec or os.path.exists(spec):
        return spec
    from huggingface_hub import hf_hub_download

    repo, _, filename = spec.rpartition("/")
    return hf_hub_download(repo_id=repo, filename=filename)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--checkpoint", default=os.environ.get("OEV_CHECKPOINT", ""),
                   help="path to a .pt checkpoint or a Hugging Face id (repo/filename); OEV_CHECKPOINT env works too")
    p.add_argument("--onnx", default=os.environ.get("OEV_ONNX", ""),
                   help="path to an exported ONNX graph (int8 recommended); torch-free backend, OEV_ONNX env works too")
    p.add_argument("--backbone", default="microsoft/deberta-v3-base",
                   help="ONNX backend only: tokenizer/backbone used at export time")
    p.add_argument("--max-len", type=int, default=768,
                   help="ONNX backend only: max_len the checkpoint was trained with")
    p.add_argument("--threads", type=int, default=None,
                   help="ONNX backend only: intra-op thread count (default: onnxruntime decides)")
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=8000)
    p.add_argument("--device", default="cpu")
    args = p.parse_args()
    if not args.checkpoint and not args.onnx:
        p.error("--checkpoint (or OEV_CHECKPOINT) or --onnx (or OEV_ONNX) is required")
    global agent
    if args.onnx:
        agent = OnnxBackend(args.onnx, backbone=args.backbone, max_len=args.max_len, threads=args.threads)
    else:
        agent = TorchBackend(_resolve_checkpoint(args.checkpoint), device=args.device)
    import uvicorn

    uvicorn.run(app, host=args.host, port=args.port)


if __name__ == "__main__":
    main()