File size: 3,178 Bytes
ad67f6b | 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 | """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))
|