Agent Observability Basics

from uuid import uuid4
 
def new_run(goal: str) -> dict:
    return {"run_id": str(uuid4()), "goal": goal, "steps": []}

import time
 
def model_step(run: dict, model_id: str, messages: list, call_model) -> str:
    t0 = time.perf_counter()
    text = call_model(messages)
    run["steps"].append({
        "type": "model",
        "model_id": model_id,
        "ms": int((time.perf_counter() - t0) * 1000),
        "preview": text[:200],
    })
    return text

def tool_step(run: dict, name: str, args: dict, fn) -> dict:
    t0 = time.perf_counter()
    try:
        result = fn(**args)
        status, err = "ok", None
    except Exception as e:
        result, status, err = None, "error", str(e)
    run["steps"].append({
        "type": "tool",
        "name": name,
        "args": args,
        "status": status,
        "error": err,
        "ms": int((time.perf_counter() - t0) * 1000),
        "result_preview": None if result is None else str(result)[:200],
    })
    if status != "ok":
        raise RuntimeError(err)
    return result

def run_agent(goal: str, tools: dict, call_model, max_turns: int = 5) -> dict:
    run = new_run(goal)
    messages = [{"role": "user", "content": goal}]
    for turn in range(max_turns):
        decision = model_step(run, "demo-model", messages, call_model)
        # decision is either final text or "TOOL name {...}" in a real system
        if decision.startswith("FINAL:"):
            run["stop_reason"] = "goal_met"
            run["answer"] = decision.removeprefix("FINAL:").strip()
            return run
        name, args = parse_tool(decision)  # your parser
        obs = tool_step(run, name, args, tools[name])
        messages.append({"role": "assistant", "content": decision})
        messages.append({"role": "tool", "content": str(obs)})
        run["steps"][-1]["turn"] = turn
    run["stop_reason"] = "max_turns"
    return run

SECRET_KEYS = ("password", "token", "secret", "authorization", "api_key")
 
def redact(value):
    if isinstance(value, dict):
        out = {}
        for k, v in value.items():
            out[k] = "***" if any(s in k.lower() for s in SECRET_KEYS) else redact(v)
        return out
    if isinstance(value, list):
        return [redact(v) for v in value]
    return value
 
# when appending a tool step:
# "args": redact(args)

def explain(run: dict) -> None:
    print("run", run["run_id"], "goal=", run["goal"])
    for i, step in enumerate(run["steps"]):
        if step["type"] == "model":
            print(f"{i}. model {step['model_id']} {step['ms']}ms :: {step['preview']}")
        else:
            print(f"{i}. tool  {step['name']} {step['status']} {step['ms']}ms args={step['args']}")
    print("stop_reason=", run.get("stop_reason"), "answer=", run.get("answer"))

def model_step_with_usage(run: dict, model_id: str, messages: list, call_model) -> str:
    t0 = time.perf_counter()
    text, usage = call_model(messages)  # usage: {prompt, completion}
    tokens = usage["prompt"] + usage["completion"]
    run["steps"].append({
        "type": "model",
        "model_id": model_id,
        "ms": int((time.perf_counter() - t0) * 1000),
        "tokens": tokens,
        "usage": usage,
        "preview": text[:200],
    })
    run["tokens_total"] = run.get("tokens_total", 0) + tokens
    return text

def handoff_step(run: dict, specialist: str, packet: dict, run_specialist) -> dict:
    child = {
        "type": "handoff",
        "specialist": specialist,
        "packet": redact(packet),
        "child_run_id": str(uuid4()),
        "steps": [],
    }
    result = run_specialist(packet, sink=child["steps"])
    child["status"] = result.get("status", "ok")
    run["steps"].append(child)
    return result

import json
import time
 
def export_jsonl(run: dict, path: str) -> None:
    base = {"run_id": run["run_id"], "ts": time.time(), "goal": run["goal"]}
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps({**base, "event": "run_started"}) + "\n")
        for step in run["steps"]:
            f.write(json.dumps({**base, "event": "step", "step": redact(step)}) + "\n")
        f.write(json.dumps({
            **base,
            "event": "run_finished",
            "stop_reason": run.get("stop_reason"),
            "tokens_total": run.get("tokens_total"),
        }) + "\n")