03 · The agent loop, with a fast brain plugged in¶

In build-your-first-ai-agent you learned that an agent is a loop plus a capability table. It prompts the model, runs the tool the model asks for, feeds the result back, and repeats.

That loop is full of small decisions that don't need an LLM:

                 +------------------- (A) ROUTER: which model tier? --------------------+
                 v                                                                       |
 user query -> [ LLM (slow brain) ] -> tool call? --yes--> (C) GUARD: allow / block? -> run tool
                 ^                         | no                                          |
                 |                    (D) DONE?: finished, or keep going?              (B) Jev AS A TOOL
                 +------ feed result back --------------------------------------------+
                                           |
                                           v
                                     final answer -> (E) JUDGE: is it any good?

Each lettered box is a Jev decision point: one call, a few hundred milliseconds, a fraction of a cent. This notebook adds all five to the same loop and the same network tools you already know. Every later notebook reuses this loop verbatim and changes only the tools and the questions.

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}")

1. The capability table (the same two tools as before, plus one dangerous one)¶

shutdown_interface is a config change. It is a dry-run mock here, but we will still let Jev guard it.

In [3]:

Copied!





import ipaddress, hashlib

def calculate_subnet(cidr):
    """Compute network, broadcast, netmask and usable host count for a CIDR."""
    net = ipaddress.ip_network(cidr, strict=False)
    usable = net.num_addresses - 2 if (net.version == 4 and net.prefixlen <= 30) else net.num_addresses
    return {"network": str(net.network_address),
            "broadcast": str(net.broadcast_address) if net.version == 4 else "n/a",
            "netmask": str(net.netmask), "prefix_length": net.prefixlen,
            "total_addresses": net.num_addresses, "usable_hosts": usable}

def get_interface_status(device, interface):
    """MOCK telemetry, md5-seeded so demos are repeatable. Wire to netmiko / gNMI / MCP in real life."""
    h = int(hashlib.md5(f"{device}{interface}".encode()).hexdigest(), 16)
    up = (h % 5 != 0)
    return {"device": device, "interface": interface, "admin_status": "up",
            "oper_status": "up" if up else "down", "speed": "10Gbps", "mtu": 1500,
            "input_errors": h % 7, "output_errors": h % 3, "crc_errors": h % 4}

def shutdown_interface(device, interface):
    """DRY-RUN. Would administratively shut an interface. Never touches a device in this course."""
    return {"dry_run": True, "would_run": f"{device}: interface {interface} / shutdown"}

TOOLS = [
    {"type": "function", "function": {
        "name": "calculate_subnet",
        "description": "Compute network, broadcast, netmask and usable host count for an IPv4/IPv6 CIDR.",
        "parameters": {"type": "object",
            "properties": {"cidr": {"type": "string", "description": "CIDR, e.g. 10.20.0.0/22"}},
            "required": ["cidr"]}}},
    {"type": "function", "function": {
        "name": "get_interface_status",
        "description": "Operational status and error counters for an interface on a device.",
        "parameters": {"type": "object",
            "properties": {"device": {"type": "string", "description": "hostname, e.g. leaf-01"},
                           "interface": {"type": "string", "description": "interface, e.g. ethernet1/0/1"}},
            "required": ["device", "interface"]}}},
    {"type": "function", "function": {
        "name": "shutdown_interface",
        "description": "Administratively shut down an interface (config change).",
        "parameters": {"type": "object",
            "properties": {"device": {"type": "string"}, "interface": {"type": "string"}},
            "required": ["device", "interface"]}}},
]
TOOL_REGISTRY = {"calculate_subnet": calculate_subnet, "get_interface_status": get_interface_status,
                 "shutdown_interface": shutdown_interface}
print(calculate_subnet("10.20.0.0/22")["usable_hosts"], get_interface_status("leaf-01", "ethernet1/0/1")["oper_status"])
import ipaddress, hashlib

def calculate_subnet(cidr):
    """Compute network, broadcast, netmask and usable host count for a CIDR."""
    net = ipaddress.ip_network(cidr, strict=False)
    usable = net.num_addresses - 2 if (net.version == 4 and net.prefixlen <= 30) else net.num_addresses
    return {"network": str(net.network_address),
            "broadcast": str(net.broadcast_address) if net.version == 4 else "n/a",
            "netmask": str(net.netmask), "prefix_length": net.prefixlen,
            "total_addresses": net.num_addresses, "usable_hosts": usable}

def get_interface_status(device, interface):
    """MOCK telemetry, md5-seeded so demos are repeatable. Wire to netmiko / gNMI / MCP in real life."""
    h = int(hashlib.md5(f"{device}{interface}".encode()).hexdigest(), 16)
    up = (h % 5 != 0)
    return {"device": device, "interface": interface, "admin_status": "up",
            "oper_status": "up" if up else "down", "speed": "10Gbps", "mtu": 1500,
            "input_errors": h % 7, "output_errors": h % 3, "crc_errors": h % 4}

def shutdown_interface(device, interface):
    """DRY-RUN. Would administratively shut an interface. Never touches a device in this course."""
    return {"dry_run": True, "would_run": f"{device}: interface {interface} / shutdown"}

TOOLS = [
    {"type": "function", "function": {
        "name": "calculate_subnet",
        "description": "Compute network, broadcast, netmask and usable host count for an IPv4/IPv6 CIDR.",
        "parameters": {"type": "object",
            "properties": {"cidr": {"type": "string", "description": "CIDR, e.g. 10.20.0.0/22"}},
            "required": ["cidr"]}}},
    {"type": "function", "function": {
        "name": "get_interface_status",
        "description": "Operational status and error counters for an interface on a device.",
        "parameters": {"type": "object",
            "properties": {"device": {"type": "string", "description": "hostname, e.g. leaf-01"},
                           "interface": {"type": "string", "description": "interface, e.g. ethernet1/0/1"}},
            "required": ["device", "interface"]}}},
    {"type": "function", "function": {
        "name": "shutdown_interface",
        "description": "Administratively shut down an interface (config change).",
        "parameters": {"type": "object",
            "properties": {"device": {"type": "string"}, "interface": {"type": "string"}},
            "required": ["device", "interface"]}}},
]
TOOL_REGISTRY = {"calculate_subnet": calculate_subnet, "get_interface_status": get_interface_status,
                 "shutdown_interface": shutdown_interface}
print(calculate_subnet("10.20.0.0/22")["usable_hosts"], get_interface_status("leaf-01", "ethernet1/0/1")["oper_status"])

1022 down

2. The Jev-aware loop¶

This is the ~35-line loop from the original repo with two optional hooks: before_tool (the guard) and check_done (the "am I done?" gate). With neither hook set it behaves exactly like the original.

In [4]:

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 [5]:

Copied!

_ = run_agent("How many usable hosts are in 172.16.4.0/26?", TOOLS, TOOL_REGISTRY)
_ = run_agent("How many usable hosts are in 172.16.4.0/26?", TOOLS, TOOL_REGISTRY)

USER: How many usable hosts are in 172.16.4.0/26?
================================================================

[step 1] TOOL CALL  -> calculate_subnet({'cidr': '172.16.4.0/26'})
[step 1] TOOL RESULT <- {'network': '172.16.4.0', 'broadcast': '172.16.4.63', 'netmask': '255.255.255.192', 'prefix_length': 26, 'total_addresses': 64, 'usable_hosts': 62}

[step 2] FINAL ANSWER
A `/26` subnet has **62 usable hosts**.

3. Decision point (A): a router picks the model tier before the loop¶

The router is a Choice over tiers, and the criteria describe when each tier fits. If the router isn't confident, it escalates. This is confidence routing: when a mistake is expensive, fall back to the safer option.

In [6]:

Copied!





SMART_MODEL = os.environ.get("SMART_MODEL", "openai/gpt-6-sol")

def route(query, min_confidence=0.7):
    r = ask_jev(query, {"tier": Choice(
        instructions="Which model tier should handle this request?",
        criteria={"fast": "lookups, arithmetic, single-step tool use, extraction, short factual answers",
                  "capable": "multi-step planning, design trade-offs, root-cause analysis, anything high-stakes"})})
    a = r.choices["tier"]
    tier = a.choice if a.confidence >= min_confidence else "capable"    # (D') confidence fallback
    print(f"router: {a.choice} (conf {a.confidence:.2f}) -> using {tier}")
    return SMART_MODEL if tier == "capable" else MODEL

for q in ["How many usable hosts are in 10.0.0.0/24?",
          "Design a migration plan from our 2-tier campus network to a spine-leaf fabric with zero downtime."]:
    print(q[:70], "->", route(q))
SMART_MODEL = os.environ.get("SMART_MODEL", "openai/gpt-6-sol")

def route(query, min_confidence=0.7):
    r = ask_jev(query, {"tier": Choice(
        instructions="Which model tier should handle this request?",
        criteria={"fast": "lookups, arithmetic, single-step tool use, extraction, short factual answers",
                  "capable": "multi-step planning, design trade-offs, root-cause analysis, anything high-stakes"})})
    a = r.choices["tier"]
    tier = a.choice if a.confidence >= min_confidence else "capable"    # (D') confidence fallback
    print(f"router: {a.choice} (conf {a.confidence:.2f}) -> using {tier}")
    return SMART_MODEL if tier == "capable" else MODEL

for q in ["How many usable hosts are in 10.0.0.0/24?",
          "Design a migration plan from our 2-tier campus network to a spine-leaf fabric with zero downtime."]:
    print(q[:70], "->", route(q))

router: fast (conf 1.00) -> using fast
How many usable hosts are in 10.0.0.0/24? -> openai/gpt-6-luna

router: capable (conf 1.00) -> using capable
Design a migration plan from our 2-tier campus network to a spine-leaf -> openai/gpt-6-sol

4. Decision point (B): Jev as a tool the LLM can call¶

The LLM decides when to classify. Jev does the classifying in ~300 ms, with a calibrated answer that a program can act on.

In [7]:

Copied!





def classify_syslog(line):
    """Classify a syslog line: severity, subsystem, and whether a human must act now."""
    r = ask_jev(line, {
        "severity":  Score(instructions="How severe is this log line?",
                           criteria=["debug noise", "informational", "warning", "error", "critical outage"]),
        "subsystem": Choice(instructions="Which subsystem produced this?",
                            criteria={"routing": "BGP, OSPF, routes", "interface": "links, optics, errors",
                                      "hardware": "PSU, fans, temperature", "auth": "logins, AAA",
                                      "other": "anything else"}),
        "page_now":  Noul(instructions="Must a human act on this within 15 minutes?"),
    })
    return {"severity": round(r.scores["severity"].score, 2), "subsystem": r.choices["subsystem"].choice,
            "page_now": round(r.nouls["page_now"].noul, 2)}

TOOLS_B = TOOLS + [{"type": "function", "function": {
    "name": "classify_syslog", "description": classify_syslog.__doc__,
    "parameters": {"type": "object", "properties": {"line": {"type": "string"}}, "required": ["line"]}}}]
REGISTRY_B = {**TOOL_REGISTRY, "classify_syslog": classify_syslog}

_ = run_agent("Triage these and tell me which one to wake someone up for:\n"
              "1) %BGP-5-ADJCHANGE: neighbor 10.0.0.2 Down - Hold timer expired\n"
              "2) %SYS-6-LOGGINGHOST_STARTSTOP: Logging to host 10.1.1.5 started\n"
              "3) %PLATFORM-2-PS_FAIL: Power supply 2 failed on spine-01",
              TOOLS_B, REGISTRY_B)
def classify_syslog(line):
    """Classify a syslog line: severity, subsystem, and whether a human must act now."""
    r = ask_jev(line, {
        "severity":  Score(instructions="How severe is this log line?",
                           criteria=["debug noise", "informational", "warning", "error", "critical outage"]),
        "subsystem": Choice(instructions="Which subsystem produced this?",
                            criteria={"routing": "BGP, OSPF, routes", "interface": "links, optics, errors",
                                      "hardware": "PSU, fans, temperature", "auth": "logins, AAA",
                                      "other": "anything else"}),
        "page_now":  Noul(instructions="Must a human act on this within 15 minutes?"),
    })
    return {"severity": round(r.scores["severity"].score, 2), "subsystem": r.choices["subsystem"].choice,
            "page_now": round(r.nouls["page_now"].noul, 2)}

TOOLS_B = TOOLS + [{"type": "function", "function": {
    "name": "classify_syslog", "description": classify_syslog.__doc__,
    "parameters": {"type": "object", "properties": {"line": {"type": "string"}}, "required": ["line"]}}}]
REGISTRY_B = {**TOOL_REGISTRY, "classify_syslog": classify_syslog}

_ = run_agent("Triage these and tell me which one to wake someone up for:\n"
              "1) %BGP-5-ADJCHANGE: neighbor 10.0.0.2 Down - Hold timer expired\n"
              "2) %SYS-6-LOGGINGHOST_STARTSTOP: Logging to host 10.1.1.5 started\n"
              "3) %PLATFORM-2-PS_FAIL: Power supply 2 failed on spine-01",
              TOOLS_B, REGISTRY_B)

USER: Triage these and tell me which one to wake someone up for:
1) %BGP-5-ADJCHANGE: neighbor 10.0.0.2 Down - Hold timer expired
2) %SYS-6-LOGGINGHOST_STARTSTOP: Logging to host 10.1.1.5 started
3) %PLATFORM-2-PS_FAIL: Power supply 2 failed on spine-01
================================================================

[step 1] TOOL CALL  -> classify_syslog({'line': '%BGP-5-ADJCHANGE: neighbor 10.0.0.2 Down - Hold timer expired'})

[step 1] TOOL RESULT <- {'severity': 2.88, 'subsystem': 'routing', 'page_now': 0.18}
[step 1] TOOL CALL  -> classify_syslog({'line': '%SYS-6-LOGGINGHOST_STARTSTOP: Logging to host 10.1.1.5 started'})

[step 1] TOOL RESULT <- {'severity': 1.0, 'subsystem': 'other', 'page_now': 0.06}
[step 1] TOOL CALL  -> classify_syslog({'line': '%PLATFORM-2-PS_FAIL: Power supply 2 failed on spine-01'})

[step 1] TOOL RESULT <- {'severity': 3.06, 'subsystem': 'hardware', 'page_now': 0.41}

[step 2] FINAL ANSWER
**Wake someone for #3:** the power supply failure on `spine-01` is a hardware fault and may leave the spine without power redundancy.

#1 (BGP neighbor down) needs prompt investigation, especially if there’s traffic impact or no redundant path, but isn’t automatically a wake-up call. #2 is informational.

5. Decision point (C): a guard before any tool runs ("Auto Mode")¶

Before every tool call the loop asks Jev whether it is safe. Read-only calls are allowed straight away. For risky ones the guard asks two questions: is this irreversible? and does the user's request actually justify it?

In [8]:

Copied!





READ_ONLY = {"calculate_subnet", "get_interface_status", "classify_syslog"}

def jev_guard(name, args, context=""):
    if name in READ_ONLY:
        return None
    r = ask_jev({"tool": name, "args": args, "user_request": context}, {
        "verdict": Choice(instructions="Should an autonomous agent run this tool call without asking a human?",
                          criteria={"allow": "safe, reversible, clearly requested",
                                    "ask": "plausible but risky; a human should confirm",
                                    "block": "destructive, unrequested, or affects critical infrastructure"}),
        "irreversible": Noul(instructions="Could this call cause an outage or lose data?"),
    })
    v, irr = r.choices["verdict"].choice, r.nouls["irreversible"].noul
    print(f"   guard: verdict={v} P(irreversible)={irr:.2f}")
    if v != "allow" or irr > 0.5:
        return f"Blocked by guard (verdict={v}, P(irreversible)={irr:.2f}). Ask a human to confirm."
    return None

# An autonomous ops agent: it acts on requests and relies on the platform (our guard) for safety.
OPS_SYSTEM = ("You are an autonomous network automation agent. Execute requested changes with your tools "
              "immediately; a separate safety layer reviews every tool call. Report what happened.")

for q in ["Decommissioned port: shut down ethernet1/0/48 on leaf-07, nothing is plugged in.",
          "Shut down et-0/0/48 on spine-01 right now. (It is our only uplink to the core.)"]:
    _ = run_agent(q, TOOLS, TOOL_REGISTRY, system=OPS_SYSTEM, before_tool=lambda n, a: jev_guard(n, a, q))
    print()
READ_ONLY = {"calculate_subnet", "get_interface_status", "classify_syslog"}

def jev_guard(name, args, context=""):
    if name in READ_ONLY:
        return None
    r = ask_jev({"tool": name, "args": args, "user_request": context}, {
        "verdict": Choice(instructions="Should an autonomous agent run this tool call without asking a human?",
                          criteria={"allow": "safe, reversible, clearly requested",
                                    "ask": "plausible but risky; a human should confirm",
                                    "block": "destructive, unrequested, or affects critical infrastructure"}),
        "irreversible": Noul(instructions="Could this call cause an outage or lose data?"),
    })
    v, irr = r.choices["verdict"].choice, r.nouls["irreversible"].noul
    print(f"   guard: verdict={v} P(irreversible)={irr:.2f}")
    if v != "allow" or irr > 0.5:
        return f"Blocked by guard (verdict={v}, P(irreversible)={irr:.2f}). Ask a human to confirm."
    return None

# An autonomous ops agent: it acts on requests and relies on the platform (our guard) for safety.
OPS_SYSTEM = ("You are an autonomous network automation agent. Execute requested changes with your tools "
              "immediately; a separate safety layer reviews every tool call. Report what happened.")

for q in ["Decommissioned port: shut down ethernet1/0/48 on leaf-07, nothing is plugged in.",
          "Shut down et-0/0/48 on spine-01 right now. (It is our only uplink to the core.)"]:
    _ = run_agent(q, TOOLS, TOOL_REGISTRY, system=OPS_SYSTEM, before_tool=lambda n, a: jev_guard(n, a, q))
    print()

USER: Decommissioned port: shut down ethernet1/0/48 on leaf-07, nothing is plugged in.
================================================================

[step 1] TOOL CALL  -> shutdown_interface({'device': 'leaf-07', 'interface': 'ethernet1/0/48'})

   guard: verdict=allow P(irreversible)=0.12
[step 1] TOOL RESULT <- {'dry_run': True, 'would_run': 'leaf-07: interface ethernet1/0/48 / shutdown'}

[step 2] FINAL ANSWER
The shutdown was not applied. The tool returned a dry-run result for `leaf-07` interface `ethernet1/0/48`.

USER: Shut down et-0/0/48 on spine-01 right now. (It is our only uplink to the core.)
================================================================

[step 1] TOOL CALL  -> shutdown_interface({'device': 'spine-01', 'interface': 'et-0/0/48'})

   guard: verdict=ask P(irreversible)=0.93
[step 1] TOOL RESULT <- {'blocked': 'Blocked by guard (verdict=ask, P(irreversible)=0.93). Ask a human to confirm.'}

[step 2] FINAL ANSWER
The shutdown was **not performed**. The safety guard blocked the change and requires human confirmation because this is the only uplink to the core.

6. Decision point (D): the "am I done?" gate¶

LLMs sometimes stop early and answer only part of a multi-part request. Before accepting the final answer, the gate asks Jev whether every part of the task has been answered. If not, the loop sends the model back to work.

In [9]:

Copied!





def jev_done(query, answer):
    r = ask_jev({"task": query, "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 "
                          "(for example, blocked by a safety check)",
                  "false": "a requested item is silently skipped, or the answer promises to do it later"})})
    p = r.nouls["complete"].noul
    return None if p >= 0.5 else f"(P(complete)={p:.2f}) You have not finished. Answer every part of the task."

task = "For leaf-01 check ethernet1/0/1 and ethernet1/0/2, and give me the netmask of 10.8.0.0/13."

# First, the gate on its own. A lazy answer that skips two of the three items:
print("lazy answer ->", jev_done(task, "ethernet1/0/1 on leaf-01 is operationally down."))

# Now the real loop, gated: it only returns once Jev agrees every part is answered.
_ = run_agent(task, TOOLS, TOOL_REGISTRY, check_done=jev_done)
def jev_done(query, answer):
    r = ask_jev({"task": query, "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 "
                          "(for example, blocked by a safety check)",
                  "false": "a requested item is silently skipped, or the answer promises to do it later"})})
    p = r.nouls["complete"].noul
    return None if p >= 0.5 else f"(P(complete)={p:.2f}) You have not finished. Answer every part of the task."

task = "For leaf-01 check ethernet1/0/1 and ethernet1/0/2, and give me the netmask of 10.8.0.0/13."

# First, the gate on its own. A lazy answer that skips two of the three items:
print("lazy answer ->", jev_done(task, "ethernet1/0/1 on leaf-01 is operationally down."))

# Now the real loop, gated: it only returns once Jev agrees every part is answered.
_ = run_agent(task, TOOLS, TOOL_REGISTRY, check_done=jev_done)

lazy answer -> (P(complete)=0.03) You have not finished. Answer every part of the task.
USER: For leaf-01 check ethernet1/0/1 and ethernet1/0/2, and give me the netmask of 10.8.0.0/13.
================================================================

[step 1] TOOL CALL  -> get_interface_status({'device': 'leaf-01', 'interface': 'ethernet1/0/1'})
[step 1] TOOL RESULT <- {'device': 'leaf-01', 'interface': 'ethernet1/0/1', 'admin_status': 'up', 'oper_status': 'down', 'speed': '10Gbps', 'mtu': 1500, 'input_errors': 2, 'output_errors': 2, 'crc_errors': 2}
[step 1] TOOL CALL  -> get_interface_status({'device': 'leaf-01', 'interface': 'ethernet1/0/2'})
[step 1] TOOL RESULT <- {'device': 'leaf-01', 'interface': 'ethernet1/0/2', 'admin_status': 'up', 'oper_status': 'up', 'speed': '10Gbps', 'mtu': 1500, 'input_errors': 4, 'output_errors': 1, 'crc_errors': 1}
[step 1] TOOL CALL  -> calculate_subnet({'cidr': '10.8.0.0/13'})
[step 1] TOOL RESULT <- {'network': '10.8.0.0', 'broadcast': '10.15.255.255', 'netmask': '255.248.0.0', 'prefix_length': 13, 'total_addresses': 524288, 'usable_hosts': 524286}

[step 2] FINAL ANSWER
- **ethernet1/0/1:** Admin up, operationally down; 2 input, 2 output, and 2 CRC errors.
- **ethernet1/0/2:** Admin and operationally up; 4 input, 1 output, and 1 CRC error.
- **10.8.0.0/13 netmask:** `255.248.0.0`

7. Decision point (E): a judge scores the final answer¶

This works as a post-hoc quality check you can run on every answer in production, because it costs almost nothing. Notebook 11 turns it into a full eval suite.

In [10]:

Copied!





def jev_judge(query, answer):
    r = ask_jev({"question": query, "answer": answer}, {
        "helpful": Score(instructions="How well does the answer resolve the question?",
                         criteria=["wrong or off-topic", "partially", "mostly", "fully and precisely"]),
        "hedging": Noul(instructions="Does the answer refuse, stall, or ask the user to do the work?"),
    })
    return {"helpful": round(r.scores["helpful"].score, 2), "hedging": round(r.nouls["hedging"].noul, 2)}

q = "What is the broadcast address of 192.168.10.0/24?"
ans = run_agent(q, TOOLS, TOOL_REGISTRY, model=route(q), verbose=False)
print(ans); print("judge:", jev_judge(q, ans))
def jev_judge(query, answer):
    r = ask_jev({"question": query, "answer": answer}, {
        "helpful": Score(instructions="How well does the answer resolve the question?",
                         criteria=["wrong or off-topic", "partially", "mostly", "fully and precisely"]),
        "hedging": Noul(instructions="Does the answer refuse, stall, or ask the user to do the work?"),
    })
    return {"helpful": round(r.scores["helpful"].score, 2), "hedging": round(r.nouls["hedging"].noul, 2)}

q = "What is the broadcast address of 192.168.10.0/24?"
ans = run_agent(q, TOOLS, TOOL_REGISTRY, model=route(q), verbose=False)
print(ans); print("judge:", jev_judge(q, ans))

router: fast (conf 1.00) -> using fast

The broadcast address is **192.168.10.255**.

judge: {'helpful': 3.0, 'hedging': 0.01}

8. All five together¶

The router picks the tier. The LLM runs the loop, calling Jev as a tool when it wants to. The guard screens every tool call. The gate checks the task is finished, and the judge scores the result.

In [11]:

Copied!





q = ("Check et-0/0/1 on spine-02. If it's down, classify this log line from it: "
     "'%LINK-3-UPDOWN: Interface et-0/0/1, changed state to down'. Then shut the interface for maintenance.")
ans = run_agent(q, TOOLS_B, REGISTRY_B, model=route(q), system=OPS_SYSTEM,
                before_tool=lambda n, a: jev_guard(n, a, q), check_done=jev_done)
print("\njudge:", jev_judge(q, ans))
q = ("Check et-0/0/1 on spine-02. If it's down, classify this log line from it: "
     "'%LINK-3-UPDOWN: Interface et-0/0/1, changed state to down'. Then shut the interface for maintenance.")
ans = run_agent(q, TOOLS_B, REGISTRY_B, model=route(q), system=OPS_SYSTEM,
                before_tool=lambda n, a: jev_guard(n, a, q), check_done=jev_done)
print("\njudge:", jev_judge(q, ans))

router: capable (conf 0.52) -> using capable
USER: Check et-0/0/1 on spine-02. If it's down, classify this log line from it: '%LINK-3-UPDOWN: Interface et-0/0/1, changed state to down'. Then shut the interface for maintenance.
================================================================

[step 1] TOOL CALL  -> get_interface_status({'device': 'spine-02', 'interface': 'et-0/0/1'})
[step 1] TOOL RESULT <- {'device': 'spine-02', 'interface': 'et-0/0/1', 'admin_status': 'up', 'oper_status': 'down', 'speed': '10Gbps', 'mtu': 1500, 'input_errors': 0, 'output_errors': 1, 'crc_errors': 0}

[step 2] TOOL CALL  -> classify_syslog({'line': '%LINK-3-UPDOWN: Interface et-0/0/1, changed state to down'})

[step 2] TOOL RESULT <- {'severity': 2.6, 'subsystem': 'interface', 'page_now': 0.18}

[step 3] TOOL CALL  -> shutdown_interface({'device': 'spine-02', 'interface': 'et-0/0/1'})

   guard: verdict=ask P(irreversible)=0.82
[step 3] TOOL RESULT <- {'blocked': 'Blocked by guard (verdict=ask, P(irreversible)=0.82). Ask a human to confirm.'}

[step 4] FINAL ANSWER
On spine-02, et-0/0/1 is administratively up but operationally down. The log classifier identified an interface event with severity 2.6 and a “page now” score of 0.18.

I attempted the maintenance shutdown, but the safety guard blocked it pending human confirmation. **The interface has not been shut down.** Please confirm if you want me to retry.

judge: {'helpful': 1.39, 'hedging': 0.9}

In [12]:

Copied!

print(SPEND)   # every decision point above, plus every LLM step
print(SPEND)   # every decision point above, plus every LLM step

Jev: 16 calls $0.00029  |  LLM: 16 calls $0.00651

Appendix: the same decision points in a framework (optional)¶

TypeSafe ships middleware that plugs these patterns into frameworks. Install uv sync --extra frameworks and set TYPESAFE_API_KEY to a TypeSafe key to try them:

# LangChain: ModelRouterMiddleware (A) and AutoModeMiddleware (C)
from langchain.agents import create_agent
from langchain_typesafe.experimental.middleware import AutoModeMiddleware, ModelRouterMiddleware, ModelChoice
agent = create_agent("openai:gpt-6-luna", middleware=[AutoModeMiddleware(tools=["bash"])])

# Pydantic AI: Jev as a model that fills typed outputs
from pydantic_ai import Agent
judge = Agent("typesafe:jev-latest", output_type=bool, instructions="Is this harmful?")

Hand-rolling them first, as we did here, shows you exactly what those one-liners do.

Recap¶

Decision point	Jev primitive	Where it sits
(A) router	`Choice` + confidence fallback	before the loop
(B) Jev as a tool	any	inside the capability table
(C) guard	`Choice` + `Noul`	before each tool executes
(D) done gate	`Noul`	at loop termination
(E) judge	`Score` + `Noul`	after the loop

Next: 04_email_triage_job.ipynb is the first real use case: classifying a whole inbox in seconds and letting the LLM draft replies only where they're needed.