04 · Use case 1: the email triage job¶

The problem: a busy inbox gets hundreds of emails a day. Reading every one with an LLM is slow and expensive, and most of them never needed an LLM in the first place.

The fast/slow split:

 inbox (N emails) --> Jev: category + needs_reply + urgency  (N calls, ~0.4 s each, in parallel)
                        |
                        +--> phishing        -> quarantine (never reply)
                        +--> low confidence  -> "review" pile for a human
                        +--> needs_reply     -> LLM drafts a reply   (only this slice pays LLM prices)
                        +--> everything else -> labelled and filed

This notebook builds the job step by step. jobs/email_triage.py is the same logic as a script you can put in cron. It can read the sample inbox or a real mailbox over read-only IMAP.

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

1. The inbox¶

data/inbox.jsonl holds 40 emails with a human label on each, so we can measure the classifier instead of trusting it.

In [4]:

Copied!

INBOX = load_jsonl("inbox.jsonl")
print(len(INBOX), "emails")
print(json.dumps(INBOX[2], indent=2))
INBOX = load_jsonl("inbox.jsonl")
print(len(INBOX), "emails")
print(json.dumps(INBOX[2], indent=2))

40 emails
{
  "id": "m003",
  "from": "dana@acme.example",
  "subject": "Need the firewall change approved today",
  "body": "The vendor is on site in two hours and needs change CR-4471 approved before they can start. Can you approve it or tell me who else can?",
  "label": {
    "category": "action_request",
    "needs_reply": true,
    "urgency": 3
  }
}

2. Design the questions¶

This step matters more than any other. Every option gets a description, because Jev reads them literally. Where the boundary between yes and no matters, the Noul gets criteria.

In [5]:

Copied!





CATEGORIES = {
    "action_request": "a person asks me to do, review, approve, answer or decide something",
    "meeting":        "scheduling, invitations, moving or preparing for a meeting",
    "billing":        "invoices, receipts, charges, payments, price changes",
    "newsletter":     "marketing, promotions, digests, webinars or product news sent to many people",
    "security_alert": "a genuine security notice from a system I use: sign-ins, MFA, vulnerabilities, certificates, endpoint compliance",
    "phishing":       "a scam: fake login pages, lookalike domains, gift-card requests, asks for credentials or secrecy",
    "personal":       "family, friends, neighbours, hobbies",
    "notification":   "automated FYI from tools: builds, deliveries, tickets, resolved alerts",
}

TRIAGE_QUESTIONS = {
    "category": Choice(instructions="What kind of email is this?", criteria=CATEGORIES),
    "needs_reply": Noul(
        instructions="Does the sender expect me to write back to them personally?",
        criteria={"true": "a direct question or request to me that needs my written answer",
                  "false": "automated, broadcast, FYI, a scam, or no answer expected"}),
    "urgency": Score(
        instructions="How soon must I act on this email?",
        criteria=["no action needed / whenever", "this week", "today", "within the hour"]),
}

def triage(email):
    state = {"from": email["from"], "subject": email["subject"], "body": email["body"]}
    r = ask_jev(state, TRIAGE_QUESTIONS)
    return {"id": email["id"], "category": r.choices["category"].choice,
            "category_conf": r.choices["category"].confidence,
            "needs_reply": r.nouls["needs_reply"].noul, "urgency": r.scores["urgency"].score}

print(INBOX[2]["subject"]); print(triage(INBOX[2]))
CATEGORIES = {
    "action_request": "a person asks me to do, review, approve, answer or decide something",
    "meeting":        "scheduling, invitations, moving or preparing for a meeting",
    "billing":        "invoices, receipts, charges, payments, price changes",
    "newsletter":     "marketing, promotions, digests, webinars or product news sent to many people",
    "security_alert": "a genuine security notice from a system I use: sign-ins, MFA, vulnerabilities, certificates, endpoint compliance",
    "phishing":       "a scam: fake login pages, lookalike domains, gift-card requests, asks for credentials or secrecy",
    "personal":       "family, friends, neighbours, hobbies",
    "notification":   "automated FYI from tools: builds, deliveries, tickets, resolved alerts",
}

TRIAGE_QUESTIONS = {
    "category": Choice(instructions="What kind of email is this?", criteria=CATEGORIES),
    "needs_reply": Noul(
        instructions="Does the sender expect me to write back to them personally?",
        criteria={"true": "a direct question or request to me that needs my written answer",
                  "false": "automated, broadcast, FYI, a scam, or no answer expected"}),
    "urgency": Score(
        instructions="How soon must I act on this email?",
        criteria=["no action needed / whenever", "this week", "today", "within the hour"]),
}

def triage(email):
    state = {"from": email["from"], "subject": email["subject"], "body": email["body"]}
    r = ask_jev(state, TRIAGE_QUESTIONS)
    return {"id": email["id"], "category": r.choices["category"].choice,
            "category_conf": r.choices["category"].confidence,
            "needs_reply": r.nouls["needs_reply"].noul, "urgency": r.scores["urgency"].score}

print(INBOX[2]["subject"]); print(triage(INBOX[2]))

Need the firewall change approved today

{'id': 'm003', 'category': 'action_request', 'category_conf': 1.0, 'needs_reply': 0.92, 'urgency': 2.46}

3. Run the whole inbox in parallel¶

Every call is independent, so we fan out with threads. Jev's limit is 1,200 requests/min, so 8 workers stays well under it.

In [6]:

Copied!





t0 = time.perf_counter()
RESULTS = jev_map(triage, INBOX)
elapsed = time.perf_counter() - t0
print(f"triaged {len(RESULTS)} emails in {elapsed:.1f} s  ({len(RESULTS) / elapsed:.1f} emails/s)")
print(SPEND)
t0 = time.perf_counter()
RESULTS = jev_map(triage, INBOX)
elapsed = time.perf_counter() - t0
print(f"triaged {len(RESULTS)} emails in {elapsed:.1f} s  ({len(RESULTS) / elapsed:.1f} emails/s)")
print(SPEND)

triaged 40 emails in 2.4 s  (16.6 emails/s)
Jev: 41 calls $0.00115  |  LLM: 0 calls $0.00000

4. Measure it against the labels¶

Never ship a classifier you haven't measured. Urgency is an ordered score, so we count "within one level" as close enough.

In [7]:

Copied!





by_id = {e["id"]: e for e in INBOX}
cat_ok   = [r["category"] == by_id[r["id"]]["label"]["category"] for r in RESULTS]
reply_ok = [(r["needs_reply"] >= 0.5) == by_id[r["id"]]["label"]["needs_reply"] for r in RESULTS]
urg_ok   = [abs(r["urgency"] - by_id[r["id"]]["label"]["urgency"]) <= 1 for r in RESULTS]
print(f"category accuracy     {sum(cat_ok) / len(cat_ok):.0%}")
print(f"needs_reply accuracy  {sum(reply_ok) / len(reply_ok):.0%}")
print(f"urgency within 1 level {sum(urg_ok) / len(urg_ok):.0%}")

print("\nmisses (category):")
for r, ok in zip(RESULTS, cat_ok):
    if not ok:
        e = by_id[r["id"]]
        print(f"  {r['id']} {e['subject'][:55]!r:<58} label={e['label']['category']:<15} jev={r['category']} (conf {r['category_conf']:.2f})")

assert sum(cat_ok) / len(cat_ok) >= 0.75, "category accuracy regressed - check your question wording"
by_id = {e["id"]: e for e in INBOX}
cat_ok   = [r["category"] == by_id[r["id"]]["label"]["category"] for r in RESULTS]
reply_ok = [(r["needs_reply"] >= 0.5) == by_id[r["id"]]["label"]["needs_reply"] for r in RESULTS]
urg_ok   = [abs(r["urgency"] - by_id[r["id"]]["label"]["urgency"]) <= 1 for r in RESULTS]
print(f"category accuracy     {sum(cat_ok) / len(cat_ok):.0%}")
print(f"needs_reply accuracy  {sum(reply_ok) / len(reply_ok):.0%}")
print(f"urgency within 1 level {sum(urg_ok) / len(urg_ok):.0%}")

print("\nmisses (category):")
for r, ok in zip(RESULTS, cat_ok):
    if not ok:
        e = by_id[r["id"]]
        print(f"  {r['id']} {e['subject'][:55]!r:<58} label={e['label']['category']:<15} jev={r['category']} (conf {r['category_conf']:.2f})")

assert sum(cat_ok) / len(cat_ok) >= 0.75, "category accuracy regressed - check your question wording"

category accuracy     92%
needs_reply accuracy  95%
urgency within 1 level 88%

misses (category):
  m010 'Meeting recording available'                              label=meeting         jev=notification (conf 0.97)
  m033 'Can you drive on Saturday?'                               label=personal        jev=action_request (conf 0.99)
  m034 'Water shut-off tomorrow 9-12'                             label=personal        jev=notification (conf 0.99)

5. Route by confidence, then into buckets¶

Confidence routing: when Jev isn't sure, a human looks, so nothing gets silently misfiled. Phishing is quarantined no matter what needs_reply says. That rule lives in code, not in a prompt.

In [8]:

Copied!





REVIEW_BELOW = 0.6

def bucket(r):
    if r["category"] == "phishing":
        return "quarantine"
    if r["category_conf"] < REVIEW_BELOW:
        return "review"
    if r["needs_reply"] >= 0.5:
        return "reply"
    return "file:" + r["category"]

for r in RESULTS:
    r["bucket"] = bucket(r)

from collections import Counter
for b, n in sorted(Counter(r["bucket"] for r in RESULTS).items(), key=lambda kv: -kv[1]):
    print(f"  {b:<24} {n}")
REVIEW_BELOW = 0.6

def bucket(r):
    if r["category"] == "phishing":
        return "quarantine"
    if r["category_conf"] < REVIEW_BELOW:
        return "review"
    if r["needs_reply"] >= 0.5:
        return "reply"
    return "file:" + r["category"]

for r in RESULTS:
    r["bucket"] = bucket(r)

from collections import Counter
for b, n in sorted(Counter(r["bucket"] for r in RESULTS).items(), key=lambda kv: -kv[1]):
    print(f"  {b:<24} {n}")

  reply                    11
  file:notification        7
  file:newsletter          5
  quarantine               5
  file:billing             4
  file:security_alert      4
  file:action_request      1
  file:meeting             1
  review                   1
  file:personal            1

6. The slow brain drafts replies, but only where one is needed¶

This is the only step that pays LLM prices, and it runs on a small slice of the inbox. We sort by urgency so the most urgent drafts come first.

In [9]:

Copied!





to_reply = sorted([r for r in RESULTS if r["bucket"] == "reply"], key=lambda r: -r["urgency"])

def draft(r):
    e = by_id[r["id"]]
    return chat(f"From: {e['from']}\nSubject: {e['subject']}\n\n{e['body']}",
                system="Draft a short, friendly reply (max 4 sentences) on my behalf. "
                       "Don't invent facts; use [placeholders] for anything you don't know. No subject line.")

DRAFTS = dict(zip([r["id"] for r in to_reply], jev_map(draft, to_reply, workers=4)))
for r in to_reply[:3]:
    print(f"--- {by_id[r['id']]['subject']}  (urgency {r['urgency']:.1f})\n{DRAFTS[r['id']]}\n")
print(f"drafted {len(DRAFTS)} of {len(INBOX)} emails")
to_reply = sorted([r for r in RESULTS if r["bucket"] == "reply"], key=lambda r: -r["urgency"])

def draft(r):
    e = by_id[r["id"]]
    return chat(f"From: {e['from']}\nSubject: {e['subject']}\n\n{e['body']}",
                system="Draft a short, friendly reply (max 4 sentences) on my behalf. "
                       "Don't invent facts; use [placeholders] for anything you don't know. No subject line.")

DRAFTS = dict(zip([r["id"] for r in to_reply], jev_map(draft, to_reply, workers=4)))
for r in to_reply[:3]:
    print(f"--- {by_id[r['id']]['subject']}  (urgency {r['urgency']:.1f})\n{DRAFTS[r['id']]}\n")
print(f"drafted {len(DRAFTS)} of {len(INBOX)} emails")

--- Need the firewall change approved today  (urgency 2.4)
Thanks for the heads-up. I’m checking whether I can approve CR-4471 today; if not, [approver/team] may be able to help. I’ll let you know as soon as I can.

--- ACTION REQUIRED: your laptop is missing the EDR agent  (urgency 2.0)
Hi, thanks for the heads-up. My asset tag is [asset tag]. Please let me know if you need anything else.

--- Can we move tomorrow's call?  (urgency 2.0)
Thanks for the heads-up! [Thursday at 3pm works for me / I’m available at ... instead]. Let me know what you prefer.

drafted 11 of 40 emails

7. The bill: time and money¶

The whole inbox was decided in a few seconds for a fraction of a cent, and the LLM ran on only a quarter of the emails. The naive design has the LLM read and classify every email, then draft replies. Its cost is estimated below from our measured average LLM call. Notebook 02 measures the classification head-to-head.

In [10]:

Copied!





avg_llm = SPEND.usd["llm"] / max(SPEND.calls["llm"], 1)
print(SPEND)
print(f"fast/slow job : ${SPEND.usd['jev'] + SPEND.usd['llm']:.5f}   LLM calls: {SPEND.calls['llm']}")
print(f"LLM-only job  : ${avg_llm * len(INBOX) + SPEND.usd['llm']:.5f}   LLM calls: {len(INBOX) + SPEND.calls['llm']}  (estimate)")
avg_llm = SPEND.usd["llm"] / max(SPEND.calls["llm"], 1)
print(SPEND)
print(f"fast/slow job : ${SPEND.usd['jev'] + SPEND.usd['llm']:.5f}   LLM calls: {SPEND.calls['llm']}")
print(f"LLM-only job  : ${avg_llm * len(INBOX) + SPEND.usd['llm']:.5f}   LLM calls: {len(INBOX) + SPEND.calls['llm']}  (estimate)")

Jev: 41 calls $0.00115  |  LLM: 11 calls $0.00113
fast/slow job : $0.00229   LLM calls: 11
LLM-only job  : $0.00525   LLM calls: 51  (estimate)

8. Write the triage report¶

The job writes a Markdown report and a JSON file, ready to be emailed to yourself, posted to Slack or picked up by another agent.

In [11]:

Copied!





from datetime import date
Path("reports").mkdir(exist_ok=True)
lines = [f"# Inbox triage - {date.today()}", "", f"{len(INBOX)} emails - {SPEND}", ""]
for b in ["quarantine", "review", "reply"]:
    rows = [r for r in RESULTS if r["bucket"] == b]
    lines += [f"## {b} ({len(rows)})", ""]
    for r in sorted(rows, key=lambda r: -r["urgency"]):
        e = by_id[r["id"]]
        lines.append(f"- **{e['subject']}** - {e['from']} - urgency {r['urgency']:.1f}")
        if r["id"] in DRAFTS:
            lines.append("  > " + DRAFTS[r["id"]].replace("\n", "\n  > "))
    lines.append("")
filed = Counter(r["bucket"] for r in RESULTS if r["bucket"].startswith("file:"))
lines += ["## filed", ""] + [f"- {b[5:]}: {n}" for b, n in filed.items()]
Path("reports/triage-notebook.md").write_text("\n".join(lines))
Path("reports/triage-notebook.json").write_text(json.dumps(RESULTS, indent=2))
print("\n".join(lines[:18]))
from datetime import date
Path("reports").mkdir(exist_ok=True)
lines = [f"# Inbox triage - {date.today()}", "", f"{len(INBOX)} emails - {SPEND}", ""]
for b in ["quarantine", "review", "reply"]:
    rows = [r for r in RESULTS if r["bucket"] == b]
    lines += [f"## {b} ({len(rows)})", ""]
    for r in sorted(rows, key=lambda r: -r["urgency"]):
        e = by_id[r["id"]]
        lines.append(f"- **{e['subject']}** - {e['from']} - urgency {r['urgency']:.1f}")
        if r["id"] in DRAFTS:
            lines.append("  > " + DRAFTS[r["id"]].replace("\n", "\n  > "))
    lines.append("")
filed = Counter(r["bucket"] for r in RESULTS if r["bucket"].startswith("file:"))
lines += ["## filed", ""] + [f"- {b[5:]}: {n}" for b, n in filed.items()]
Path("reports/triage-notebook.md").write_text("\n".join(lines))
Path("reports/triage-notebook.json").write_text(json.dumps(RESULTS, indent=2))
print("\n".join(lines[:18]))

# Inbox triage - 2026-09-23

40 emails - Jev: 41 calls $0.00115  |  LLM: 11 calls $0.00113

## quarantine (5)

- **Quick favour (confidential)** - ceo.office@acme-corp.test - urgency 2.1
- **Your mailbox is full - verify now** - it-support@acme-helpdesk.test - urgency 1.9
- **Payment on hold: confirm your identity** - payments@paypa1.test - urgency 1.6
- **Unusual sign-in activity detected** - support@micros0ft-security.test - urgency 1.1
- **Priya shared 'Q3 Salary Adjustments.xlsx' with you** - docs@sharefile-docs.test - urgency 0.8

## review (1)

- **Your speaker slot is confirmed** - events@conference.example - urgency 0.4

## reply (11)

9. Real mail: a read-only IMAP loader (optional)¶

Set IMAP_HOST, IMAP_USER and IMAP_PASSWORD (a Gmail/Outlook app password) in .env to triage your real inbox. The loader is read-only: it opens the mailbox with readonly=True and fetches with BODY.PEEK[], so nothing is marked as read, moved or deleted. It never sends mail.

In [12]:

Copied!





import imaplib, email as emaillib
from email.header import decode_header, make_header

def load_imap(limit=25, folder="INBOX"):
    """Fetch the newest `limit` messages, read-only. Returns dicts shaped like data/inbox.jsonl."""
    box = imaplib.IMAP4_SSL(os.environ["IMAP_HOST"])
    box.login(os.environ["IMAP_USER"], os.environ["IMAP_PASSWORD"])
    box.select(folder, readonly=True)                      # read-only: flags never change
    _, data = box.search(None, "ALL")
    out = []
    for num in data[0].split()[-limit:][::-1]:
        _, msg_data = box.fetch(num, "(BODY.PEEK[])")      # PEEK: don't mark as seen
        msg = emaillib.message_from_bytes(msg_data[0][1])
        body = ""
        for part in msg.walk():
            if part.get_content_type() == "text/plain":
                body = part.get_payload(decode=True).decode(part.get_content_charset() or "utf-8", "replace")
                break
        out.append({"id": num.decode(), "from": str(make_header(decode_header(msg.get("From", "")))),
                    "subject": str(make_header(decode_header(msg.get("Subject", "")))),
                    "body": body[:4000]})           # trim: Jev's state budget is 32k tokens
    box.logout()
    return out

if os.environ.get("IMAP_HOST"):
    real = load_imap(limit=10)
    for r in jev_map(triage, real):
        print(r)
else:
    print("IMAP_HOST not set - skipping real mailbox (sample inbox used above).")
import imaplib, email as emaillib
from email.header import decode_header, make_header

def load_imap(limit=25, folder="INBOX"):
    """Fetch the newest `limit` messages, read-only. Returns dicts shaped like data/inbox.jsonl."""
    box = imaplib.IMAP4_SSL(os.environ["IMAP_HOST"])
    box.login(os.environ["IMAP_USER"], os.environ["IMAP_PASSWORD"])
    box.select(folder, readonly=True)                      # read-only: flags never change
    _, data = box.search(None, "ALL")
    out = []
    for num in data[0].split()[-limit:][::-1]:
        _, msg_data = box.fetch(num, "(BODY.PEEK[])")      # PEEK: don't mark as seen
        msg = emaillib.message_from_bytes(msg_data[0][1])
        body = ""
        for part in msg.walk():
            if part.get_content_type() == "text/plain":
                body = part.get_payload(decode=True).decode(part.get_content_charset() or "utf-8", "replace")
                break
        out.append({"id": num.decode(), "from": str(make_header(decode_header(msg.get("From", "")))),
                    "subject": str(make_header(decode_header(msg.get("Subject", "")))),
                    "body": body[:4000]})           # trim: Jev's state budget is 32k tokens
    box.logout()
    return out

if os.environ.get("IMAP_HOST"):
    real = load_imap(limit=10)
    for r in jev_map(triage, real):
        print(r)
else:
    print("IMAP_HOST not set - skipping real mailbox (sample inbox used above).")

IMAP_HOST not set - skipping real mailbox (sample inbox used above).

Recap¶

Use Jev for the decisions (category, reply needed?, urgency) on 100% of the mail, and the LLM for writing, on the slice that needs it.
Measure against labels and route by confidence, so a human sees the cases Jev is unsure about.
Hard rules live in code. Phishing is never answered, whatever the model says.
Run it on a schedule: uv run python jobs/email_triage.py --draft-replies (see the README for a cron line).

Next: 05_sms_scam_shield.ipynb, where Jev teams up with deterministic URL checks to catch smishing.