scripts/jev_client.py

"""jev_client.py - the one-liner your agents use to hit the local decision service.

Env: JEV_BASE_URL (default http://127.0.0.1:8190), JEV_API_KEY (required).
Contract reminder: state must be ENGLISH structured data; never route rights/publish
money gates through the model (deterministic code owns those); confidence-gate results.

    from jev_client import decide, route
    ans = decide({"tool": "bash", "result": "traceback ... not yet fixed"},
                 {"still_useful": {"type": "noul", "instructions": "Is this tool result still needed?"}})
    tier = route(ans["still_useful"])   # "auto" | "llm_review" | "human"
"""
from __future__ import annotations

import os
import time
from typing import Dict, Tuple

import requests

BASE = os.environ.get("JEV_BASE_URL", "http://127.0.0.1:8190")
KEY = os.environ.get("JEV_API_KEY", "")


def decide(state, questions: Dict[str, dict], timeout: float = 5.0) -> dict:
    """One /v1/systemone call; all questions answered in a single forward pass.

    Returns the server response: {"answers": {qid: {...probs/confidence...}}, "latency_ms": ...}
    Raises RuntimeError on transport/auth errors - callers should fall back to the LLM tier.
    """
    r = requests.post(
        f"{BASE}/v1/systemone",
        json={"state": state, "questions": questions, "model": "laya"},
        headers={"Authorization": f"Bearer {KEY}"},
        timeout=timeout,
    )
    if r.status_code != 200:
        raise RuntimeError(f"jev {r.status_code}: {r.text[:200]}")
    return r.json()


def confidence_of(ans: dict) -> float:
    """Type-agnostic confidence for routing: noul uses |p-0.5|*2 since it has no confidence field."""
    if "noul" in ans:
        return abs(ans["noul"] - 0.5) * 2.0
    return float(ans.get("confidence", 0.0))


def route(ans: dict, hi: float = 0.9, lo: float = 0.7) -> str:
    """Three-tier gate: auto-execute / send to LLM review / queue for human."""
    c = confidence_of(ans)
    return "auto" if c >= hi else ("llm_review" if c >= lo else "human")


def decide_gated(state, questions: Dict[str, dict], **kw) -> Tuple[dict, Dict[str, str]]:
    """decide() + per-question route(); returns (answers, tiers). Never raises:
    on any service error all questions come back tier=llm_review so agents degrade safely."""
    try:
        res = decide(state, questions, **kw)
        ans = res["answers"]
        return ans, {qid: route(a) for qid, a in ans.items()}
    except Exception:
        return {}, {qid: "llm_review" for qid in questions}


if __name__ == "__main__":
    import json
    import sys
    sys.stdout.reconfigure(encoding="utf-8")
    t0 = time.perf_counter()
    ans, tiers = decide_gated(
        {"tool": "web_search", "result_summary": "docs page already used, step 9 of 20",
         "task_goal": "flag the render command"},
        {"keep": {"type": "noul", "instructions": "Is this tool result still needed for the task?"}},
    )
    print(json.dumps({"answers": ans, "tiers": tiers,
                      "roundtrip_ms": round((time.perf_counter() - t0) * 1000, 1)}, ensure_ascii=False, indent=2))