11 · Use cases 11 & 12: Jev as the judge, for evals and for the "am I done?" gate¶
LLM-as-judge is how most teams evaluate agents, usually with a big model, which is slow and costly, so evals run rarely. A fast, calibrated judge changes the economics: you can grade every answer in CI and every answer in production.
- Use case 11, the eval judge. We grade 20 labeled candidate answers with Jev and with an LLM judge, and compare agreement with the labels, time and cost. Then we turn it into a CI gate for an agent.
- Use case 12, the "am I done?" gate. This is notebook 03's termination check, now measured on labeled examples, because a gate you haven't measured is only a guess.
In [1]:
Copied!
# --- Provider config: ONE OpenRouter key drives both brains ------------------------------
# slow brain (LLM) -> OpenAI SDK -> https://openrouter.ai/api/v1 (chat completions + tools)
# fast brain (Jev) -> TypeSafe SDK -> https://openrouter.ai/api (POST /v1/systemone)
import os, json, time
# Load settings from a .env file if present (falls back to existing env vars).
try:
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv(usecwd=True))
except Exception:
if os.path.exists(".env"):
for _line in open(".env"):
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_k, _v = _line.split("=", 1)
os.environ.setdefault(_k.strip(), _v.strip())
BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "set-me")
MODEL = os.environ.get("MODEL", "openai/gpt-6-luna") # slow brain: any tool-capable chat model
JEV_BACKEND = os.environ.get("JEV_BACKEND", "typesafe").strip().lower() # typesafe | adapter
JEV_MODEL = os.environ.get("JEV_MODEL", "~typesafe/jev-latest")
JEV_BASE_URL = os.environ.get("TYPESAFE_BASE_URL", "https://openrouter.ai/api")
JEV_API_KEY = os.environ.get("TYPESAFE_API_KEY") or API_KEY # one key for both brains
# Behind a TLS-intercepting firewall/proxy, set VERIFY_SSL=false in .env (trusted networks only).
import httpx, httpx2
VERIFY_SSL = os.environ.get("VERIFY_SSL", "true").strip().lower() not in ("false", "0", "no")
if not VERIFY_SSL:
import warnings
warnings.filterwarnings("ignore")
print("\u26a0\ufe0f SSL verification DISABLED (VERIFY_SSL=false) \u2014 use only on a trusted network")
from openai import OpenAI
client = OpenAI(base_url=BASE_URL, api_key=API_KEY, http_client=httpx.Client(verify=VERIFY_SSL))
if JEV_BACKEND == "adapter":
# No Jev access yet? Same System One API, answered by your LLM (slower, pricier, but it runs).
from system_one_adapter import SystemOneAdapterClient, Noul, Choice, Score
from system_one_adapter.providers.openai import OpenAIProvider
jev = SystemOneAdapterClient(
structured_outputs=True, llm_answer_mode="probabilities", normalize_probabilities=True,
n_retry_malformed_structure=2,
model=OpenAIProvider(MODEL, base_url=BASE_URL, api_key=API_KEY, api="chat_completions"))
else:
from typesafe_sdk import TypeSafeClient, Noul, Choice, Score
jev = TypeSafeClient(api_key=JEV_API_KEY, base_url=JEV_BASE_URL, model=JEV_MODEL,
http_client=httpx2.Client(verify=VERIFY_SSL))
print(f"slow brain (LLM): {MODEL} @ {BASE_URL}")
print(f"fast brain (Jev): {JEV_MODEL if JEV_BACKEND != 'adapter' else MODEL + ' via adapter'} @ {JEV_BASE_URL if JEV_BACKEND != 'adapter' else BASE_URL}")
# --- Provider config: ONE OpenRouter key drives both brains ------------------------------
# slow brain (LLM) -> OpenAI SDK -> https://openrouter.ai/api/v1 (chat completions + tools)
# fast brain (Jev) -> TypeSafe SDK -> https://openrouter.ai/api (POST /v1/systemone)
import os, json, time
# Load settings from a .env file if present (falls back to existing env vars).
try:
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv(usecwd=True))
except Exception:
if os.path.exists(".env"):
for _line in open(".env"):
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_k, _v = _line.split("=", 1)
os.environ.setdefault(_k.strip(), _v.strip())
BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1")
API_KEY = os.environ.get("OPENAI_API_KEY", "set-me")
MODEL = os.environ.get("MODEL", "openai/gpt-6-luna") # slow brain: any tool-capable chat model
JEV_BACKEND = os.environ.get("JEV_BACKEND", "typesafe").strip().lower() # typesafe | adapter
JEV_MODEL = os.environ.get("JEV_MODEL", "~typesafe/jev-latest")
JEV_BASE_URL = os.environ.get("TYPESAFE_BASE_URL", "https://openrouter.ai/api")
JEV_API_KEY = os.environ.get("TYPESAFE_API_KEY") or API_KEY # one key for both brains
# Behind a TLS-intercepting firewall/proxy, set VERIFY_SSL=false in .env (trusted networks only).
import httpx, httpx2
VERIFY_SSL = os.environ.get("VERIFY_SSL", "true").strip().lower() not in ("false", "0", "no")
if not VERIFY_SSL:
import warnings
warnings.filterwarnings("ignore")
print("\u26a0\ufe0f SSL verification DISABLED (VERIFY_SSL=false) \u2014 use only on a trusted network")
from openai import OpenAI
client = OpenAI(base_url=BASE_URL, api_key=API_KEY, http_client=httpx.Client(verify=VERIFY_SSL))
if JEV_BACKEND == "adapter":
# No Jev access yet? Same System One API, answered by your LLM (slower, pricier, but it runs).
from system_one_adapter import SystemOneAdapterClient, Noul, Choice, Score
from system_one_adapter.providers.openai import OpenAIProvider
jev = SystemOneAdapterClient(
structured_outputs=True, llm_answer_mode="probabilities", normalize_probabilities=True,
n_retry_malformed_structure=2,
model=OpenAIProvider(MODEL, base_url=BASE_URL, api_key=API_KEY, api="chat_completions"))
else:
from typesafe_sdk import TypeSafeClient, Noul, Choice, Score
jev = TypeSafeClient(api_key=JEV_API_KEY, base_url=JEV_BASE_URL, model=JEV_MODEL,
http_client=httpx2.Client(verify=VERIFY_SSL))
print(f"slow brain (LLM): {MODEL} @ {BASE_URL}")
print(f"fast brain (Jev): {JEV_MODEL if JEV_BACKEND != 'adapter' else MODEL + ' via adapter'} @ {JEV_BASE_URL if JEV_BACKEND != 'adapter' else BASE_URL}")
slow brain (LLM): openai/gpt-6-luna @ https://openrouter.ai/api/v1 fast brain (Jev): ~typesafe/jev-latest @ https://openrouter.ai/api
In [2]:
Copied!
# --- Helpers used in every notebook: ask the fast brain, ask the slow brain, track spend ---
class Spend:
"""Running tally of what OpenRouter charged, per brain (it reports exact USD per call)."""
def __init__(self):
self.usd, self.calls = {"jev": 0.0, "llm": 0.0}, {"jev": 0, "llm": 0}
def add(self, brain, resp):
self.calls[brain] += 1
try: cost = resp.raw_http_response.json()["usage"].get("cost") # Jev response
except Exception: cost = getattr(getattr(resp, "usage", None), "cost", None) # LLM response
self.usd[brain] += cost or 0.0
def __repr__(self):
return (f"Jev: {self.calls['jev']} calls ${self.usd['jev']:.5f} | "
f"LLM: {self.calls['llm']} calls ${self.usd['llm']:.5f}")
SPEND = Spend()
def ask_jev(state, questions):
"""Fast brain. state (text or JSON) + typed questions -> typed, calibrated answers."""
r = jev.system_one(state, questions)
SPEND.add("jev", r)
return r
def chat(prompt, system="Be concise.", model=None):
"""Slow brain, single shot (no tools). Returns the reply text."""
resp = client.chat.completions.create(model=model or MODEL, messages=[
{"role": "system", "content": system}, {"role": "user", "content": prompt}])
SPEND.add("llm", resp)
return resp.choices[0].message.content
def show(r):
"""Print every answer in a Jev response on one line each."""
for k, a in r.answers.items():
if a.type == "noul":
print(f" {k:<18} noul P(yes)={a.noul:.2f}")
elif a.type == "choice":
top = sorted(a.probabilities.items(), key=lambda kv: -kv[1])[:3]
print(f" {k:<18} choice {a.choice!r:<22} conf={a.confidence:.2f} top={top}")
else:
level = {int(i): v for i, v in (a.legend or {}).items()}.get(round(a.score), "")
print(f" {k:<18} score {a.score:.2f} -> {level!r:<18} conf={a.confidence:.2f}")
# --- Helpers used in every notebook: ask the fast brain, ask the slow brain, track spend ---
class Spend:
"""Running tally of what OpenRouter charged, per brain (it reports exact USD per call)."""
def __init__(self):
self.usd, self.calls = {"jev": 0.0, "llm": 0.0}, {"jev": 0, "llm": 0}
def add(self, brain, resp):
self.calls[brain] += 1
try: cost = resp.raw_http_response.json()["usage"].get("cost") # Jev response
except Exception: cost = getattr(getattr(resp, "usage", None), "cost", None) # LLM response
self.usd[brain] += cost or 0.0
def __repr__(self):
return (f"Jev: {self.calls['jev']} calls ${self.usd['jev']:.5f} | "
f"LLM: {self.calls['llm']} calls ${self.usd['llm']:.5f}")
SPEND = Spend()
def ask_jev(state, questions):
"""Fast brain. state (text or JSON) + typed questions -> typed, calibrated answers."""
r = jev.system_one(state, questions)
SPEND.add("jev", r)
return r
def chat(prompt, system="Be concise.", model=None):
"""Slow brain, single shot (no tools). Returns the reply text."""
resp = client.chat.completions.create(model=model or MODEL, messages=[
{"role": "system", "content": system}, {"role": "user", "content": prompt}])
SPEND.add("llm", resp)
return resp.choices[0].message.content
def show(r):
"""Print every answer in a Jev response on one line each."""
for k, a in r.answers.items():
if a.type == "noul":
print(f" {k:<18} noul P(yes)={a.noul:.2f}")
elif a.type == "choice":
top = sorted(a.probabilities.items(), key=lambda kv: -kv[1])[:3]
print(f" {k:<18} choice {a.choice!r:<22} conf={a.confidence:.2f} top={top}")
else:
level = {int(i): v for i, v in (a.legend or {}).items()}.get(round(a.score), "")
print(f" {k:<18} score {a.score:.2f} -> {level!r:<18} conf={a.confidence:.2f}")
In [3]:
Copied!
from pathlib import Path
DATA = Path("data")
def load_jsonl(name):
return [json.loads(line) for line in (DATA / name).read_text().splitlines() if line.strip()]
from concurrent.futures import ThreadPoolExecutor
def jev_map(fn, items, workers=8):
"""Run fn over items in parallel threads. Jev allows 1,200 requests/min, so 8 workers is safe."""
with ThreadPoolExecutor(max_workers=workers) as pool:
return list(pool.map(fn, items))
from pathlib import Path
DATA = Path("data")
def load_jsonl(name):
return [json.loads(line) for line in (DATA / name).read_text().splitlines() if line.strip()]
from concurrent.futures import ThreadPoolExecutor
def jev_map(fn, items, workers=8):
"""Run fn over items in parallel threads. Jev allows 1,200 requests/min, so 8 workers is safe."""
with ThreadPoolExecutor(max_workers=workers) as pool:
return list(pool.map(fn, items))
In [4]:
Copied!
EVALS = load_jsonl("eval_set.jsonl")
print(len(EVALS), "graded examples;", sum(e["label"]["correct"] for e in EVALS), "correct")
print(json.dumps(EVALS[0], indent=2))
EVALS = load_jsonl("eval_set.jsonl")
print(len(EVALS), "graded examples;", sum(e["label"]["correct"] for e in EVALS), "correct")
print(json.dumps(EVALS[0], indent=2))
20 graded examples; 10 correct
{
"id": "e001",
"question": "What is the broadcast address of 192.168.10.0/24?",
"reference": "192.168.10.255",
"candidate": "192.168.10.255",
"label": {
"correct": true
}
}
1. The Jev judge¶
In [5]:
Copied!
def jev_judge(e):
r = ask_jev({"question": e["question"], "reference_answer": e["reference"], "candidate_answer": e["candidate"]}, {
"correct": Noul(instructions="Does the candidate answer agree with the reference answer on the facts asked?",
criteria={"true": "same key fact(s); extra correct detail is fine",
"false": "a key fact differs from the reference, or is missing"}),
"quality": Score(instructions="How good is the candidate answer overall?",
criteria=["wrong", "partially right", "right but unclear", "right and clear"]),
})
return r.nouls["correct"].noul, r.scores["quality"].score
t0 = time.perf_counter()
JEV = jev_map(jev_judge, EVALS)
jev_s = time.perf_counter() - t0
jev_agree = sum((p >= 0.5) == e["label"]["correct"] for (p, _), e in zip(JEV, EVALS))
print(f"Jev judge: agreement {jev_agree}/{len(EVALS)} {jev_s:.1f} s {SPEND}")
def jev_judge(e):
r = ask_jev({"question": e["question"], "reference_answer": e["reference"], "candidate_answer": e["candidate"]}, {
"correct": Noul(instructions="Does the candidate answer agree with the reference answer on the facts asked?",
criteria={"true": "same key fact(s); extra correct detail is fine",
"false": "a key fact differs from the reference, or is missing"}),
"quality": Score(instructions="How good is the candidate answer overall?",
criteria=["wrong", "partially right", "right but unclear", "right and clear"]),
})
return r.nouls["correct"].noul, r.scores["quality"].score
t0 = time.perf_counter()
JEV = jev_map(jev_judge, EVALS)
jev_s = time.perf_counter() - t0
jev_agree = sum((p >= 0.5) == e["label"]["correct"] for (p, _), e in zip(JEV, EVALS))
print(f"Jev judge: agreement {jev_agree}/{len(EVALS)} {jev_s:.1f} s {SPEND}")
Jev judge: agreement 20/20 1.9 s Jev: 20 calls $0.00035 | LLM: 0 calls $0.00000
2. The LLM judge baselines (the usual way)¶
Each LLM judge gets the same inputs and returns JSON. We run a small model (your MODEL) and a frontier model (SMART_MODEL, which is what teams typically use as a judge), both on the same number of parallel workers.
Read the result honestly. A tiny LLM can match Jev on price for easy grading. What it can't give you is a calibrated probability to threshold on, typed output that never fails to parse, and ~300 ms per decision. Against a frontier judge, the cost and latency gap is large.
In [6]:
Copied!
SMART_MODEL = os.environ.get("SMART_MODEL", "openai/gpt-6-sol")
def llm_judge(e, model):
out = chat(f"Question: {e['question']}\nReference: {e['reference']}\nCandidate: {e['candidate']}",
system='You grade answers. Reply with JSON only: {"correct": true|false} - true if the candidate '
'agrees with the reference on the facts asked.', model=model)
try:
return bool(json.loads(out[out.index("{"): out.rindex("}") + 1])["correct"])
except Exception:
return None # parse failures are part of the LLM-judge tax
rows = [("Jev", jev_agree, jev_s, SPEND.usd["jev"], 0)]
for m in (MODEL, SMART_MODEL):
before = SPEND.usd["llm"]
t0 = time.perf_counter()
votes = jev_map(lambda e: llm_judge(e, m), EVALS)
rows.append((m, sum(v == e["label"]["correct"] for v, e in zip(votes, EVALS)), time.perf_counter() - t0,
SPEND.usd["llm"] - before, votes.count(None)))
print(f"{'judge':<20} {'agree':>6} {'wall s':>7} {'cost $':>9} {'$/1k':>7} {'parse fails':>11}")
for name, agree, secs, cost, fails in rows:
print(f"{name:<20} {agree:>3}/{len(EVALS)} {secs:>7.1f} {cost:>9.5f} {1000 * cost / len(EVALS):>7.3f} {fails:>11}")
SMART_MODEL = os.environ.get("SMART_MODEL", "openai/gpt-6-sol")
def llm_judge(e, model):
out = chat(f"Question: {e['question']}\nReference: {e['reference']}\nCandidate: {e['candidate']}",
system='You grade answers. Reply with JSON only: {"correct": true|false} - true if the candidate '
'agrees with the reference on the facts asked.', model=model)
try:
return bool(json.loads(out[out.index("{"): out.rindex("}") + 1])["correct"])
except Exception:
return None # parse failures are part of the LLM-judge tax
rows = [("Jev", jev_agree, jev_s, SPEND.usd["jev"], 0)]
for m in (MODEL, SMART_MODEL):
before = SPEND.usd["llm"]
t0 = time.perf_counter()
votes = jev_map(lambda e: llm_judge(e, m), EVALS)
rows.append((m, sum(v == e["label"]["correct"] for v, e in zip(votes, EVALS)), time.perf_counter() - t0,
SPEND.usd["llm"] - before, votes.count(None)))
print(f"{'judge':<20} {'agree':>6} {'wall s':>7} {'cost $':>9} {'$/1k':>7} {'parse fails':>11}")
for name, agree, secs, cost, fails in rows:
print(f"{name:<20} {agree:>3}/{len(EVALS)} {secs:>7.1f} {cost:>9.5f} {1000 * cost / len(EVALS):>7.3f} {fails:>11}")
judge agree wall s cost $ $/1k parse fails Jev 20/20 1.9 0.00035 0.018 0 openai/gpt-6-luna 20/20 4.2 0.00037 0.019 0 openai/gpt-6-sol 20/20 4.1 0.00480 0.240 0
3. A CI gate for your agent¶
This is how you'd use it in CI. The agent (here a plain LLM call with nb03's subnet tool) answers every unique eval question, Jev grades each answer against the reference, and the build fails if the pass rate drops below the bar. It is fast and cheap enough to run on every pull request.
In [7]:
Copied!
def tool_spec(fn, **params):
"""Build an OpenAI tool schema from a function + {param: description} (all string params)."""
return {"type": "function", "function": {
"name": fn.__name__, "description": (fn.__doc__ or "").strip().split("\n")[0],
"parameters": {"type": "object",
"properties": {p: {"type": "string", "description": d} for p, d in params.items()},
"required": list(params)}}}
def tool_spec(fn, **params):
"""Build an OpenAI tool schema from a function + {param: description} (all string params)."""
return {"type": "function", "function": {
"name": fn.__name__, "description": (fn.__doc__ or "").strip().split("\n")[0],
"parameters": {"type": "object",
"properties": {p: {"type": "string", "description": d} for p, d in params.items()},
"required": list(params)}}}
In [8]:
Copied!
def run_agent(user_query, tools, registry, system="You are a helpful assistant. Use tools when they help. Be concise.",
model=None, before_tool=None, check_done=None, max_iterations=8, verbose=True):
"""The agent loop from build-your-first-ai-agent, plus two Jev decision-point hooks.
before_tool(name, args) -> None to allow, or a reason string to block (Jev guard)
check_done(query, answer) -> None if finished, or feedback to keep going (Jev "am I done?" gate)
"""
log = print if verbose else (lambda *a, **k: None)
messages = [{"role": "system", "content": system}, {"role": "user", "content": user_query}]
log("USER:", user_query); log("=" * 64)
for step in range(1, max_iterations + 1):
resp = client.chat.completions.create(model=model or MODEL, messages=messages, tools=tools)
SPEND.add("llm", resp)
msg = resp.choices[0].message
# TERMINATION: no tool requested -> candidate final answer (optionally gated by Jev).
if not msg.tool_calls:
feedback = check_done(user_query, msg.content) if check_done else None
if not feedback:
log(f"[step {step}] FINAL ANSWER\n{msg.content}")
return msg.content
log(f"[step {step}] NOT DONE (Jev gate) -> {feedback}")
messages += [{"role": "assistant", "content": msg.content or ""},
{"role": "user", "content": feedback}]
continue
messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [
{"id": tc.id, "type": "function",
"function": {"name": tc.function.name, "arguments": tc.function.arguments}}
for tc in msg.tool_calls]})
for tc in msg.tool_calls:
name, args = tc.function.name, json.loads(tc.function.arguments or "{}")
log(f"[step {step}] TOOL CALL -> {name}({args})")
blocked = before_tool(name, args) if before_tool else None # <-- Jev guard
result = {"blocked": blocked} if blocked else registry[name](**args)
log(f"[step {step}] TOOL RESULT <- {str(result)[:300]}")
messages.append({"role": "tool", "tool_call_id": tc.id,
"content": json.dumps(result, default=str)})
return "Stopped: hit max_iterations (TTL expired)."
def run_agent(user_query, tools, registry, system="You are a helpful assistant. Use tools when they help. Be concise.",
model=None, before_tool=None, check_done=None, max_iterations=8, verbose=True):
"""The agent loop from build-your-first-ai-agent, plus two Jev decision-point hooks.
before_tool(name, args) -> None to allow, or a reason string to block (Jev guard)
check_done(query, answer) -> None if finished, or feedback to keep going (Jev "am I done?" gate)
"""
log = print if verbose else (lambda *a, **k: None)
messages = [{"role": "system", "content": system}, {"role": "user", "content": user_query}]
log("USER:", user_query); log("=" * 64)
for step in range(1, max_iterations + 1):
resp = client.chat.completions.create(model=model or MODEL, messages=messages, tools=tools)
SPEND.add("llm", resp)
msg = resp.choices[0].message
# TERMINATION: no tool requested -> candidate final answer (optionally gated by Jev).
if not msg.tool_calls:
feedback = check_done(user_query, msg.content) if check_done else None
if not feedback:
log(f"[step {step}] FINAL ANSWER\n{msg.content}")
return msg.content
log(f"[step {step}] NOT DONE (Jev gate) -> {feedback}")
messages += [{"role": "assistant", "content": msg.content or ""},
{"role": "user", "content": feedback}]
continue
messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [
{"id": tc.id, "type": "function",
"function": {"name": tc.function.name, "arguments": tc.function.arguments}}
for tc in msg.tool_calls]})
for tc in msg.tool_calls:
name, args = tc.function.name, json.loads(tc.function.arguments or "{}")
log(f"[step {step}] TOOL CALL -> {name}({args})")
blocked = before_tool(name, args) if before_tool else None # <-- Jev guard
result = {"blocked": blocked} if blocked else registry[name](**args)
log(f"[step {step}] TOOL RESULT <- {str(result)[:300]}")
messages.append({"role": "tool", "tool_call_id": tc.id,
"content": json.dumps(result, default=str)})
return "Stopped: hit max_iterations (TTL expired)."
In [9]:
Copied!
import ipaddress
def calculate_subnet(cidr):
"""Compute network, broadcast, netmask and usable host count for a CIDR."""
net = ipaddress.ip_network(cidr, strict=False)
return {"network": str(net.network_address), "broadcast": str(net.broadcast_address),
"netmask": str(net.netmask), "usable_hosts": max(net.num_addresses - 2, 0)}
KB_FACTS = ("Acme docs: Pro workspaces: 600 API requests/minute. Enterprise automated backups: 35 days. "
"SEV1 = full outage or data loss.")
AGENT_SYSTEM = f"Answer in one sentence. Use tools for subnet math. {KB_FACTS}"
unique = {e["question"]: e["reference"] for e in EVALS}
def agent_answer(q):
return run_agent(q, [tool_spec(calculate_subnet, cidr="CIDR")], {"calculate_subnet": calculate_subnet},
system=AGENT_SYSTEM, verbose=False)
ANSWERS = dict(zip(unique, jev_map(agent_answer, list(unique), workers=4)))
GRADES = jev_map(lambda q: jev_judge({"question": q, "reference": unique[q], "candidate": ANSWERS[q]})[0], list(unique))
pass_rate = sum(g >= 0.5 for g in GRADES) / len(GRADES)
for q, g in zip(unique, GRADES):
print(f"{'PASS' if g >= 0.5 else 'FAIL'} P={g:.2f} {q[:50]:<52} -> {ANSWERS[q][:60]}")
print(f"\npass rate {pass_rate:.0%}")
assert pass_rate >= 0.8, "agent regressed - block the merge"
import ipaddress
def calculate_subnet(cidr):
"""Compute network, broadcast, netmask and usable host count for a CIDR."""
net = ipaddress.ip_network(cidr, strict=False)
return {"network": str(net.network_address), "broadcast": str(net.broadcast_address),
"netmask": str(net.netmask), "usable_hosts": max(net.num_addresses - 2, 0)}
KB_FACTS = ("Acme docs: Pro workspaces: 600 API requests/minute. Enterprise automated backups: 35 days. "
"SEV1 = full outage or data loss.")
AGENT_SYSTEM = f"Answer in one sentence. Use tools for subnet math. {KB_FACTS}"
unique = {e["question"]: e["reference"] for e in EVALS}
def agent_answer(q):
return run_agent(q, [tool_spec(calculate_subnet, cidr="CIDR")], {"calculate_subnet": calculate_subnet},
system=AGENT_SYSTEM, verbose=False)
ANSWERS = dict(zip(unique, jev_map(agent_answer, list(unique), workers=4)))
GRADES = jev_map(lambda q: jev_judge({"question": q, "reference": unique[q], "candidate": ANSWERS[q]})[0], list(unique))
pass_rate = sum(g >= 0.5 for g in GRADES) / len(GRADES)
for q, g in zip(unique, GRADES):
print(f"{'PASS' if g >= 0.5 else 'FAIL'} P={g:.2f} {q[:50]:<52} -> {ANSWERS[q][:60]}")
print(f"\npass rate {pass_rate:.0%}")
assert pass_rate >= 0.8, "agent regressed - block the merge"
PASS P=0.99 What is the broadcast address of 192.168.10.0/24? -> The broadcast address is 192.168.10.255. PASS P=0.99 What port does HTTPS use by default? -> HTTPS uses port 443 by default. PASS P=0.99 Which HTTP status code means Too Many Requests? -> HTTP 429 means Too Many Requests. PASS P=0.99 What does DNS stand for? -> DNS stands for Domain Name System. PASS P=0.99 Which protocol does ping use? -> Ping uses ICMP (Internet Control Message Protocol). PASS P=0.99 What is the default SSH port? -> The default SSH port is TCP 22. PASS P=0.99 How long are automated backups kept on Enterprise? -> Enterprise automated backups are kept for 35 days. PASS P=0.99 How many usable hosts are in a /26 IPv4 subnet? -> A /26 IPv4 subnet has 62 usable hosts. PASS P=0.98 On the Pro plan, how many API requests per minute -> Pro workspaces allow 600 API requests per minute. PASS P=0.99 What does SEV1 mean in Acme's incident process? -> SEV1 means a full outage or data loss. pass rate 100%
4. Use case 12: measure the "am I done?" gate¶
This is notebook 03's gate. Before you trust it inside a loop, grade it on labelled (task, answer) pairs, including the tricky case where the agent couldn't finish and says why.
In [10]:
Copied!
def jev_done(task, answer):
r = ask_jev({"task": task, "answer": answer}, {"complete": Noul(
instructions="Does the answer address EVERY part of the task?",
criteria={"true": "every requested item has a concrete result, or a clear reason it could not be done",
"false": "a requested item is silently skipped, or the answer promises to do it later"})})
return r.nouls["complete"].noul
GATE_CASES = [ # (task, answer, complete?)
("Check eth1 and eth2 on leaf-01.", "eth1 is up. eth2 is down with 4 CRC errors.", True),
("Check eth1 and eth2 on leaf-01.", "eth1 is up.", False),
("Give me the netmask and broadcast of 10.0.0.0/24.", "Netmask 255.255.255.0, broadcast 10.0.0.255.", True),
("Give me the netmask and broadcast of 10.0.0.0/24.", "The netmask is 255.255.255.0.", False),
("Summarize the incident and list 3 action items.", "DB pool exhausted 14:02-14:09. Actions: add index; raise pool alert; load test.", True),
("Summarize the incident and list 3 action items.", "DB pool exhausted 14:02-14:09. I'll send action items later.", False),
("Shut down et-0/0/1 and confirm.", "I did not shut it down: the safety guard blocked it pending human approval.", True),
("Translate 'hello' to French and German.", "French: bonjour.", False),
]
res = jev_map(lambda c: jev_done(c[0], c[1]), GATE_CASES)
ok = sum((p >= 0.5) == exp for p, (_, _, exp) in zip(res, GATE_CASES))
for p, (t, a, exp) in zip(res, GATE_CASES):
print(f"{'ok ' if (p >= 0.5) == exp else 'XX '} P(done)={p:.2f} {a[:70]}")
print(f"\ngate accuracy {ok}/{len(GATE_CASES)}")
def jev_done(task, answer):
r = ask_jev({"task": task, "answer": answer}, {"complete": Noul(
instructions="Does the answer address EVERY part of the task?",
criteria={"true": "every requested item has a concrete result, or a clear reason it could not be done",
"false": "a requested item is silently skipped, or the answer promises to do it later"})})
return r.nouls["complete"].noul
GATE_CASES = [ # (task, answer, complete?)
("Check eth1 and eth2 on leaf-01.", "eth1 is up. eth2 is down with 4 CRC errors.", True),
("Check eth1 and eth2 on leaf-01.", "eth1 is up.", False),
("Give me the netmask and broadcast of 10.0.0.0/24.", "Netmask 255.255.255.0, broadcast 10.0.0.255.", True),
("Give me the netmask and broadcast of 10.0.0.0/24.", "The netmask is 255.255.255.0.", False),
("Summarize the incident and list 3 action items.", "DB pool exhausted 14:02-14:09. Actions: add index; raise pool alert; load test.", True),
("Summarize the incident and list 3 action items.", "DB pool exhausted 14:02-14:09. I'll send action items later.", False),
("Shut down et-0/0/1 and confirm.", "I did not shut it down: the safety guard blocked it pending human approval.", True),
("Translate 'hello' to French and German.", "French: bonjour.", False),
]
res = jev_map(lambda c: jev_done(c[0], c[1]), GATE_CASES)
ok = sum((p >= 0.5) == exp for p, (_, _, exp) in zip(res, GATE_CASES))
for p, (t, a, exp) in zip(res, GATE_CASES):
print(f"{'ok ' if (p >= 0.5) == exp else 'XX '} P(done)={p:.2f} {a[:70]}")
print(f"\ngate accuracy {ok}/{len(GATE_CASES)}")
ok P(done)=0.97 eth1 is up. eth2 is down with 4 CRC errors. ok P(done)=0.05 eth1 is up. ok P(done)=0.98 Netmask 255.255.255.0, broadcast 10.0.0.255. ok P(done)=0.05 The netmask is 255.255.255.0. ok P(done)=0.89 DB pool exhausted 14:02-14:09. Actions: add index; raise pool alert; l ok P(done)=0.04 DB pool exhausted 14:02-14:09. I'll send action items later. ok P(done)=0.72 I did not shut it down: the safety guard blocked it pending human appr ok P(done)=0.03 French: bonjour. gate accuracy 8/8
In [11]:
Copied!
print(SPEND)
print(SPEND)
Jev: 38 calls $0.00065 | LLM: 52 calls $0.00543
Recap¶
- A Jev judge is fast and cheap enough to run on every answer: in CI on every pull request, and at runtime as a quality monitor.
Noul"correct?" plus aScorerubric covers most eval needs. Keep the reference answer in the state.- Measure your judges and gates on labelled examples, just as you measure the agent.
Next: 12_capstone_ops_copilot.ipynb, where everything comes together.