09 · Use case 8: on-call log and alert triage¶

It's 14:03 and the pager fires. There are four services and hundreds of log lines, and most of them are noise. The job is to find the incident, decide its severity and work out the probable root cause. Fast.

 raw logs --(code)--> templates + counts --(Jev)--> severity, area, customer impact, security?  (per template)
                                                  |
                                   (code) rank = severity x volume
                                                  |
                                 LLM agent investigates ONLY the top clusters with grep/tail tools
                                                  |
                                   incident summary + SEV level (Jev, from the runbook definitions)

This is notebook 05 of the original repo (the log triage agent), upgraded: the LLM no longer reads the logs line by line.

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





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

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

1. Parse and template the logs (code)¶

Counting, grouping and time ordering are exact operations, and Jev's docs say to keep math and dates in code. So we turn every line into a template by replacing numbers, IPs and IDs with placeholders, then count them. Hundreds of lines collapse into a few dozen templates.

In [6]:

Copied!





import re
from collections import defaultdict

LOG_DIR = Path("data/logs")
LINE_RE = re.compile(r"^(\S+)\s+(\w+)\s+(\w+)\s+(.*)$")

def template(msg):
    msg = re.sub(r"\b\d{1,3}(\.\d{1,3}){3}\b", "<ip>", msg)
    msg = re.sub(r"(req|user|job)=\S+", r"\1=<id>", msg)
    return re.sub(r"\d+", "<n>", msg)

CLUSTERS = defaultdict(lambda: {"count": 0, "first": None, "last": None, "example": None, "level": None})
for f in sorted(LOG_DIR.glob("*.log")):
    for line in f.read_text().splitlines():
        m = LINE_RE.match(line)
        if not m:
            continue
        ts, level, svc, msg = m.groups()
        c = CLUSTERS[(svc, template(msg))]
        c["count"] += 1
        c["first"] = c["first"] or ts
        c["last"], c["example"], c["level"] = ts, line, level

total = sum(c["count"] for c in CLUSTERS.values())
print(f"{total} lines -> {len(CLUSTERS)} templates")
import re
from collections import defaultdict

LOG_DIR = Path("data/logs")
LINE_RE = re.compile(r"^(\S+)\s+(\w+)\s+(\w+)\s+(.*)$")

def template(msg):
    msg = re.sub(r"\b\d{1,3}(\.\d{1,3}){3}\b", "<ip>", msg)
    msg = re.sub(r"(req|user|job)=\S+", r"\1=<id>", msg)
    return re.sub(r"\d+", "<n>", msg)

CLUSTERS = defaultdict(lambda: {"count": 0, "first": None, "last": None, "example": None, "level": None})
for f in sorted(LOG_DIR.glob("*.log")):
    for line in f.read_text().splitlines():
        m = LINE_RE.match(line)
        if not m:
            continue
        ts, level, svc, msg = m.groups()
        c = CLUSTERS[(svc, template(msg))]
        c["count"] += 1
        c["first"] = c["first"] or ts
        c["last"], c["example"], c["level"] = ts, line, level

total = sum(c["count"] for c in CLUSTERS.values())
print(f"{total} lines -> {len(CLUSTERS)} templates")

162 lines -> 14 templates

2. Jev reads every template, once¶

In [7]:

Copied!





AREAS = {"database": "queries, connections, pools, replication", "api": "HTTP requests and responses",
         "auth": "logins, sessions, accounts", "worker": "background jobs, queues, retries",
         "infra": "disk, deploys, hosts, certificates"}

def judge_cluster(item):
    (svc, tmpl), c = item
    r = ask_jev({"service": svc, "level": c["level"], "example_line": c["example"], "occurrences": c["count"]}, {
        "severity": Score(instructions="How serious is this log event for an on-call engineer?",
                          criteria=["noise", "informational", "warning", "error", "critical"]),
        "area": Choice(instructions="Which area of the system is this about?", criteria=AREAS),
        "customer_impact": Noul(instructions="Are customers likely seeing failures or errors because of this?"),
        "security": Noul(instructions="Is this a possible security event (attack, brute force, abuse)?"),
    })
    return {"service": svc, "template": tmpl, **c, "severity": r.scores["severity"].score,
            "area": r.choices["area"].choice, "customer_impact": r.nouls["customer_impact"].noul,
            "security": r.nouls["security"].noul}

t0 = time.perf_counter()
JUDGED = jev_map(judge_cluster, list(CLUSTERS.items()))
print(f"judged {len(JUDGED)} templates in {time.perf_counter() - t0:.1f} s")
AREAS = {"database": "queries, connections, pools, replication", "api": "HTTP requests and responses",
         "auth": "logins, sessions, accounts", "worker": "background jobs, queues, retries",
         "infra": "disk, deploys, hosts, certificates"}

def judge_cluster(item):
    (svc, tmpl), c = item
    r = ask_jev({"service": svc, "level": c["level"], "example_line": c["example"], "occurrences": c["count"]}, {
        "severity": Score(instructions="How serious is this log event for an on-call engineer?",
                          criteria=["noise", "informational", "warning", "error", "critical"]),
        "area": Choice(instructions="Which area of the system is this about?", criteria=AREAS),
        "customer_impact": Noul(instructions="Are customers likely seeing failures or errors because of this?"),
        "security": Noul(instructions="Is this a possible security event (attack, brute force, abuse)?"),
    })
    return {"service": svc, "template": tmpl, **c, "severity": r.scores["severity"].score,
            "area": r.choices["area"].choice, "customer_impact": r.nouls["customer_impact"].noul,
            "security": r.nouls["security"].noul}

t0 = time.perf_counter()
JUDGED = jev_map(judge_cluster, list(CLUSTERS.items()))
print(f"judged {len(JUDGED)} templates in {time.perf_counter() - t0:.1f} s")

judged 14 templates in 1.2 s

3. Rank (code): severity x volume¶

Ranking is arithmetic, so it's code. A single critical line matters, and so does a flood of errors. log2(count) rewards volume without letting it dominate.

In [8]:

Copied!





import math
for j in JUDGED:
    j["rank"] = round(j["severity"] * (1 + math.log2(j["count"])) + 2 * j["customer_impact"] + 2 * j["security"], 2)
TOP = sorted(JUDGED, key=lambda j: -j["rank"])
print(f"{'rank':>5} {'sev':>4} {'n':>3} {'cust':>4} {'sec':>4}  area      template")
for j in TOP[:10]:
    print(f"{j['rank']:>5} {j['severity']:>4.1f} {j['count']:>3} {j['customer_impact']:>4.2f} {j['security']:>4.2f}  "
          f"{j['area']:<9} [{j['service']}] {j['template'][:70]}")
import math
for j in JUDGED:
    j["rank"] = round(j["severity"] * (1 + math.log2(j["count"])) + 2 * j["customer_impact"] + 2 * j["security"], 2)
TOP = sorted(JUDGED, key=lambda j: -j["rank"])
print(f"{'rank':>5} {'sev':>4} {'n':>3} {'cust':>4} {'sec':>4}  area      template")
for j in TOP[:10]:
    print(f"{j['rank']:>5} {j['severity']:>4.1f} {j['count']:>3} {j['customer_impact']:>4.2f} {j['security']:>4.2f}  "
          f"{j['area']:<9} [{j['service']}] {j['template'][:70]}")

 rank  sev   n cust  sec  area      template
22.51  3.6  24 0.82 0.27  database  [db] connection pool exhausted (active=<n> max=<n>) waiting=<n>
 20.9  3.4  24 0.85 0.19  api       [api] POST /v<n>/orders <n> upstream=db timeout after <n>ms req=<id>
14.14  2.0  24 0.56 0.84  auth      [auth] login failed user=<id> ip=<ip> reason=bad_password
10.02  2.1   8 0.62 0.09  worker    [worker] job=<id> retry=<n> reason=db timeout
 6.05  0.9  50 0.08 0.09  api       [api] GET /v<n>/projects <n> <n>ms req=<id>
 5.96  2.0   3 0.30 0.08  worker    [worker] disk usage <n>% on /var/lib/queue
 4.96  2.1   1 0.45 0.96  auth      [auth] account locked user=<id> after <n> failed attempts ip=<ip>
 4.37  0.9  10 0.08 0.14  auth      [auth] login ok user=<id> ip=<ip>
 3.72  0.9   7 0.09 0.04  worker    [worker] job=<id> status=done duration=<n>s
 3.32  0.9   5 0.07 0.03  database  [db] checkpoint complete wal_size=<n>MB

4. The slow brain investigates the top clusters only¶

The agent gets the ranked summary as its starting point, plus the original repo's read-only log tools (sandboxed tail and grep) to dig further. It writes the incident summary.

In [9]:

Copied!





def _safe(name):
    p = (LOG_DIR / Path(name).name).resolve()
    if p.parent != LOG_DIR.resolve() or not p.exists():
        raise FileNotFoundError(name)
    return p

def tail_log(file, lines="20"):
    """Last N lines of a log file (api.log, db.log, auth.log, worker.log)."""
    return _safe(file).read_text().splitlines()[-int(lines):]

def grep_logs(pattern):
    """Lines matching a regex across all log files (max 40)."""
    rx = re.compile(pattern, re.I)
    hits = [f"{f.name}: {l}" for f in sorted(LOG_DIR.glob("*.log")) for l in f.read_text().splitlines() if rx.search(l)]
    return hits[:40] + ([f"... {len(hits) - 40} more"] if len(hits) > 40 else [])

TOOLS = [tool_spec(tail_log, file="log file name", lines="how many lines"), tool_spec(grep_logs, pattern="regex")]
REGISTRY = {"tail_log": tail_log, "grep_logs": grep_logs}

summary = "\n".join(f"- rank {j['rank']}: [{j['service']}] x{j['count']} {j['first']}..{j['last']} "
                     f"sev={j['severity']:.1f} customer_impact={j['customer_impact']:.2f} security={j['security']:.2f}: "
                     f"{j['example']}" for j in TOP[:6])
REPORT = run_agent(
    f"You are on call. Pre-ranked log clusters (fast classifier):\n{summary}\n\n"
    "Investigate with the tools, then write an incident summary: what happened, timeline (UTC), probable root "
    "cause with evidence, customer impact, and separately any security concern. Max 12 lines.",
    TOOLS, REGISTRY, system="You are a senior SRE. Be precise and cite log lines as evidence.")
def _safe(name):
    p = (LOG_DIR / Path(name).name).resolve()
    if p.parent != LOG_DIR.resolve() or not p.exists():
        raise FileNotFoundError(name)
    return p

def tail_log(file, lines="20"):
    """Last N lines of a log file (api.log, db.log, auth.log, worker.log)."""
    return _safe(file).read_text().splitlines()[-int(lines):]

def grep_logs(pattern):
    """Lines matching a regex across all log files (max 40)."""
    rx = re.compile(pattern, re.I)
    hits = [f"{f.name}: {l}" for f in sorted(LOG_DIR.glob("*.log")) for l in f.read_text().splitlines() if rx.search(l)]
    return hits[:40] + ([f"... {len(hits) - 40} more"] if len(hits) > 40 else [])

TOOLS = [tool_spec(tail_log, file="log file name", lines="how many lines"), tool_spec(grep_logs, pattern="regex")]
REGISTRY = {"tail_log": tail_log, "grep_logs": grep_logs}

summary = "\n".join(f"- rank {j['rank']}: [{j['service']}] x{j['count']} {j['first']}..{j['last']} "
                     f"sev={j['severity']:.1f} customer_impact={j['customer_impact']:.2f} security={j['security']:.2f}: "
                     f"{j['example']}" for j in TOP[:6])
REPORT = run_agent(
    f"You are on call. Pre-ranked log clusters (fast classifier):\n{summary}\n\n"
    "Investigate with the tools, then write an incident summary: what happened, timeline (UTC), probable root "
    "cause with evidence, customer impact, and separately any security concern. Max 12 lines.",
    TOOLS, REGISTRY, system="You are a senior SRE. Be precise and cite log lines as evidence.")

USER: You are on call. Pre-ranked log clusters (fast classifier):
- rank 22.51: [db] x24 2026-09-22T14:02:07Z..2026-09-22T14:09:30Z sev=3.6 customer_impact=0.82 security=0.27: 2026-09-22T14:09:30Z ERROR db connection pool exhausted (active=100 max=100) waiting=57
- rank 20.9: [api] x24 2026-09-22T14:02:07Z..2026-09-22T14:09:30Z sev=3.4 customer_impact=0.85 security=0.19: 2026-09-22T14:09:30Z ERROR api POST /v1/orders 500 upstream=db timeout after 5000ms req=34957
- rank 14.14: [auth] x24 2026-09-22T14:15:03Z..2026-09-22T14:20:27Z sev=2.0 customer_impact=0.56 security=0.84: 2026-09-22T14:20:27Z WARN  auth login failed user=admin ip=203.0.113.45 reason=bad_password
- rank 10.02: [worker] x8 2026-09-22T14:02:05Z..2026-09-22T14:09:26Z sev=2.1 customer_impact=0.62 security=0.09: 2026-09-22T14:09:26Z WARN  worker job=invoice-sync retry=8 reason=db timeout
- rank 6.05: [api] x50 2026-09-22T13:40:40Z..2026-09-22T14:29:20Z sev=0.9 customer_impact=0.08 security=0.09: 2026-09-22T14:29:20Z INFO  api GET /v1/projects 200 82ms req=12552
- rank 5.96: [worker] x3 2026-09-22T13:51:24Z..2026-09-22T14:25:55Z sev=2.0 customer_impact=0.30 security=0.08: 2026-09-22T14:25:55Z WARN  worker disk usage 81% on /var/lib/queue

Investigate with the tools, then write an incident summary: what happened, timeline (UTC), probable root cause with evidence, customer impact, and separately any security concern. Max 12 lines.
================================================================

[step 1] TOOL CALL  -> tail_log({'file': 'db.log', 'lines': '60'})
[step 1] TOOL RESULT <- ['2026-09-22T13:44:01Z INFO  db checkpoint complete wal_size=54MB', '2026-09-22T13:55:05Z INFO  db checkpoint complete wal_size=80MB', '2026-09-22T14:01:55Z WARN  db slow query 4210ms: SELECT * FROM orders WHERE customer_id = $1 (missing index on orders.customer_id)', '2026-09-22T14:02:07Z ERROR db 
[step 1] TOOL CALL  -> tail_log({'file': 'api.log', 'lines': '60'})
[step 1] TOOL RESULT <- ['2026-09-22T13:57:45Z INFO  api GET /v1/projects 200 28ms req=16006', '2026-09-22T13:58:42Z INFO  api GET /v1/projects 200 49ms req=47930', '2026-09-22T13:59:05Z INFO  api GET /v1/projects 200 49ms req=23238', '2026-09-22T14:00:24Z INFO  api GET /v1/projects 200 55ms req=69429', '2026-09-22T14:01:2
[step 1] TOOL CALL  -> tail_log({'file': 'worker.log', 'lines': '50'})
[step 1] TOOL RESULT <- ['2026-09-22T13:42:47Z INFO  worker job=email-digest status=done duration=9s', '2026-09-22T13:49:44Z INFO  worker job=email-digest status=done duration=5s', '2026-09-22T13:51:24Z WARN  worker disk usage 81% on /var/lib/queue', '2026-09-22T13:56:55Z INFO  worker job=email-digest status=done duration=
[step 1] TOOL CALL  -> tail_log({'file': 'auth.log', 'lines': '50'})
[step 1] TOOL RESULT <- ['2026-09-22T13:40:40Z INFO  auth login ok user=u859 ip=198.51.100.72', '2026-09-22T13:45:32Z INFO  auth login ok user=u303 ip=198.51.100.185', '2026-09-22T13:50:09Z INFO  auth login ok user=u204 ip=198.51.100.25', '2026-09-22T13:55:05Z INFO  auth login ok user=u733 ip=198.51.100.228', '2026-09-22T1

[step 2] FINAL ANSWER
- **What happened:** DB pool saturation caused `/v1/orders` requests to fail and `invoice-sync` jobs to retry.
- **Timeline (UTC):** 14:01:55 — DB warned that the orders query took 4,210 ms due to a missing `orders.customer_id` index.
- 14:02:07–14:09:30 — DB pool repeatedly exhausted at 100/100 connections (up to 77 waiting); API logged repeated order POST 500s with DB timeouts.
- 14:02:05–14:09:26 — `invoice-sync` retried through retry 8 due to DB timeouts.
- 14:10:05 — API rollback to `r-2291` completed; 14:10:30 — DB pool recovered (active=12/100).
- **Probable root cause:** Slow, unindexed orders queries likely held connections long enough to exhaust the pool; rollback preceded recovery. Logs do not identify the exact release/change that introduced the issue.
- **Customer impact:** 24 logged `/v1/orders` 500s; invoice-sync processing was delayed by retries. Sampled `/v1/projects` requests remained successful (e.g. 14:09:13, 200 in 60 ms).
- **Evidence:** `db.log` 14:01:55 slow-query warning and 14:02:07/14:09:30 pool errors; `api.log` 14:02:07 and 14:09:30 order 500s, 14:10:05 rollback; `db.log` 14:10:30 recovery; `worker.log` 14:09:26 retry 8.
- **Security concern (separate):** 24 failed `admin` login attempts from `203.0.113.45` between 14:15:03 and 14:20:27; account locked at 14:21:02. No successful login for `admin` appears in the provided logs.

5. Jev assigns the SEV level from the runbook's own definitions¶

data/kb/incident-response.md defines SEV1 to SEV3. We paste those definitions in as the Choice criteria, so the SEV level follows your policy, not the model's idea of severity.

Consequences go in code. The policy says "SEV1 and SEV2 page immediately". We could ask Jev a second question ("should we page?"), but its docs warn that separate questions carry no guaranteed consistency with each other. So ask Jev for the fact (the SEV level) and derive the action in code.

In [10]:

Copied!





SEV = {"SEV1": "a full outage or data loss", "SEV2": "major degradation for many customers",
       "SEV3": "minor impact or a single customer"}
r = ask_jev(REPORT, {"sev": Choice(instructions="Which severity level fits this incident?", criteria=SEV)})
show(r)
sev = r.choices["sev"].choice
page_now = sev in ("SEV1", "SEV2")                       # the runbook rule, in code
print(f"{sev} -> {'PAGE on-call now + status page within 15 min' if page_now else 'ticket for next business day'}")
SEV = {"SEV1": "a full outage or data loss", "SEV2": "major degradation for many customers",
       "SEV3": "minor impact or a single customer"}
r = ask_jev(REPORT, {"sev": Choice(instructions="Which severity level fits this incident?", criteria=SEV)})
show(r)
sev = r.choices["sev"].choice
page_now = sev in ("SEV1", "SEV2")                       # the runbook rule, in code
print(f"{sev} -> {'PAGE on-call now + status page within 15 min' if page_now else 'ticket for next business day'}")

  sev                choice 'SEV2'                 conf=0.84  top=[('SEV2', 0.89), ('SEV3', 0.1), ('SEV1', 0.01)]
SEV2 -> PAGE on-call now + status page within 15 min

In [11]:

Copied!

print(SPEND)
print(SPEND)

Jev: 15 calls $0.00035  |  LLM: 2 calls $0.00107

Recap¶

Code does the parsing, templating, counting and ranking. Jev judges each template once, not each line. The LLM investigates only the top clusters.
Put your own runbook's definitions into a Choice, and classifications follow your policy.
The shape of this notebook fits any flood of events: alerts, audit logs, support queues, security events.

Next: 10_rag_relevance_and_citations.ipynb, where Jev keeps retrieval-augmented answers honest.