PROJECT FOUR - EVALUATION AND OBSERVABILITY

FLYING WITH INSTRUMENTS A pilot flying through cloud does not steer by feel. The pilot reads instruments, because feelings are unreliable and the instruments tell the truth about speed, height, and direction. An agent without instruments is the same. It may seem to work, while quietly wasting money, looping, or failing on some kinds of input. Observability is the ability to understand what a system is doing inside by looking at the records it emits. Verification is the separate habit of testing whether the system does the right thing. This chapter adds both. Three words need defining. A span is the record of one timed piece of work, such as one model call or one tool call, with its start, end, input, output, and status. A trace is the collection of spans belonging to one task, which tells the story of that task from beginning to end, like a receipt listing every item in a shopping trip. A dashboard is a single view that summarizes the health of the system at a glance. Step 1: Tracing Our tracing module gives each task a short identifying code, called a trace identifier, and provides a span helper that times a block of work and writes one line to logs/traces.jsonl when the block ends. It records an error status if the work fails, and it always writes the record, even on failure, because failures are the records you most need. File: tracing.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: recording a timed span for every call """tracing.py - records a span (one timed step) for every model call and tool call.""" # Import the contextlib module to build a with-block helper import contextlib # Import the json module to read and write structured data import json # Import the os module to work with file paths import os # Import the time module to measure and record when events happen import time # Import the uuid module to generate unique identifying codes import uuid [END OF PYTHON CODE] from permission_gate import ROOT [START OF PYTHON CODE] # Create a text variable called TRACE_PATH pointing to the trace log file TRACE_PATH = os.path.join(ROOT, "logs", "traces.jsonl") # Create a text variable called RUNS_PATH pointing to the run-summary log file RUNS_PATH = os.path.join(ROOT, "logs", "runs.jsonl") # Create a dictionary variable called _current holding the trace id of the task currently running _current = {"trace_id": None} # Define a function that appends one record as a line of JSON to the given log file def append_line(path, record): # Create the containing folder if it does not already exist os.makedirs(os.path.dirname(path), exist_ok=True) # Open the log file in append mode as UTF-8 text with open(path, "a", encoding="utf-8") as f: # Write the record as one line of JSON, followed by a newline f.write(json.dumps(record) + "\n") # Define a function that starts a new trace and returns its short identifying code def start_trace(): # Generate a short, unique trace id and store it as the current trace _current["trace_id"] = uuid.uuid4().hex[:8] # Return the new trace id return _current["trace_id"] # Mark the following function as a context manager, usable in a with-block @contextlib.contextmanager # Define a function that times one piece of work and records it as a span when it finishes def span(kind, name, input_text=""): # Build the starting span record: which trace it belongs to, a new span id, its kind, name, input, and start time record = {"trace_id": _current["trace_id"], "span_id": uuid.uuid4().hex[:6], "kind": kind, "name": name, "input": input_text[:200], "output": "", "start": time.time()} # Attempt to run the work inside the with-block try: # Hand the record to the with-block so it can fill in the output before finishing yield record # If the work completed without error, mark the status as ok record["status"] = "ok" # If the work raised an error except Exception as problem: # Record the error status, truncated to 100 characters record["status"] = "error: " + str(problem)[:100] # Re-raise the error so the caller still sees it raise # Whether the work succeeded or failed, always record the span finally: # Record the end time record["end"] = time.time() # Calculate and record how long the work took, in milliseconds record["latency_ms"] = round((record["end"] - record["start"]) * 1000, 2) # Append this span record to the trace log file append_line(TRACE_PATH, record) # Define a function that appends one run summary to the run-summary log file def record_run(summary): append_line(RUNS_PATH, summary) [END OF PYTHON CODE] The helper uses a Python feature called a context manager, which is the with statement. The code inside the with block is the work being timed, and the helper takes care of starting and stopping the clock around it. Step 2: The Dashboard The dashboard reads a second file, logs/runs.jsonl, which holds one summary line per task, and prints the success rate, average cost, average steps, and average delay across the last twenty-five runs. It prints the cost in lab cost units, a made-up unit that stands for money, because this book does not claim any real price. It also contains a small detector that scans the traces for a warning sign: one tool called many times with identical input inside a single trace. File: dashboard.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: a text table of recent runs plus a repeated-span detector """dashboard.py - a text table of the last 25 runs, plus a repeated-span detector.""" # Import the json module to read structured data import json # Import the os module to work with file paths import os # Import the Counter class to count how often things repeat from collections import Counter [END OF PYTHON CODE] from tracing import RUNS_PATH, TRACE_PATH [START OF PYTHON CODE] # Create a number variable called COST_PER_1K_TOKENS, the made-up lab cost of one thousand tokens COST_PER_1K_TOKENS = 1.0 # Define a function that reads a JSON-lines file and returns a list of its parsed records def read_lines(path): # If the file does not exist yet, there is nothing to read if not os.path.exists(path): return [] # Open the file for reading as UTF-8 text with open(path, "r", encoding="utf-8") as f: # Parse and return every non-blank line as a JSON record return [json.loads(line) for line in f if line.strip()] # Define a function that prints a summary table of the most recent runs def show(last=25): # Read the run-summary log and keep only the most recent runs runs = read_lines(RUNS_PATH)[-last:] # If there are no runs yet, say so and stop if not runs: print("No runs recorded yet.") return # Collect the runs that finished successfully with a known match level ok = [r for r in runs if r["status"] == "done" and r["level"] != "unknown"] # Calculate the average lab cost per run, based on total tokens used cost = sum(r["tokens"] for r in runs) / 1000 * COST_PER_1K_TOKENS / len(runs) # Print how many runs are being shown print("Runs shown: %d" % len(runs)) # Print the success rate as a percentage, with a simple bar made of equals signs print("Success rate: %3.0f%% %s" % (100 * len(ok) / len(runs), "=" * len(ok))) # Print the average lab cost per run print("Average cost: %.2f lab units" % cost) # Print the average number of steps per run print("Average steps: %.1f" % (sum(r["steps"] for r in runs) / len(runs))) # Print the average latency per run, in milliseconds print("Average latency: %.1f ms" % (sum(r["latency_ms"] for r in runs) / len(runs))) # Print a header row for the detailed table print("%-28s %-12s %5s %7s" % ("resume", "status", "steps", "tokens")) # Print one row per run, showing its resume name, status, step count, and token count for r in runs: print("%-28s %-12s %5d %7d" % (r["resume"], r["status"], r["steps"], r["tokens"])) # Define a function that scans the trace log for tools called suspiciously many times with identical input def find_repeated_spans(minimum=4): # A docstring explaining that this pattern suggests a runaway loop """A trace where one identical span appears many times is a runaway-loop warning.""" # Count how many times each (trace, tool name, input) combination appears among tool spans counts = Counter((s["trace_id"], s["kind"], s["name"], s["input"]) for s in read_lines(TRACE_PATH) if s["kind"] == "tool") # Walk through every combination and its count for (trace_id, kind, name, _), n in counts.items(): # If a combination repeated at least the minimum number of times, print a warning if n >= minimum: print("WARNING trace %s: tool %s called %d times with identical input" % (trace_id, name, n)) # If this file is run directly (not imported), show the dashboard and check for repeated spans if __name__ == "__main__": show() find_repeated_spans() [END OF PYTHON CODE] Step 3: The Instrumented Loop Now the main loop wraps every model call and every tool call in a span. It also writes a summary line per task, and it contains the guard for repeated calls, which we will explain in a moment. Here is the complete new file. File: main_loop.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: Stage 4, recording spans and guarding against repeated calls """main_loop.py - Stage 4: the loop now records spans and guards against repeated calls.""" # Import the json module to read and write structured data import json # Import the os module to work with file paths and folders import os # Import the re module to search text with patterns import re # Import the shutil module to copy and remove whole folders import shutil # Import the sys module to read command-line arguments import sys # Import the time module to measure and record when events happen import time # Import the compaction function that shrinks the history when it grows too large from compaction import maybe_compact # Import the notes file path and the functions that build context, estimate tokens, and track progress from context_builder import (NOTES_PATH, append_note, build_context, estimate_tokens, load_done) # Import the pre- and post-execution hooks that guard every tool call from hooks import post_execution_hook, pre_execution_hook # Import the function that sends a conversation to the model and gets a reply from model_adapter import call_model # Import the shared root folder, the custom error type, and the safe folder-listing function from permission_gate import ROOT, GateDenied, safe_listdir # Import the functions that describe available tools and run a chosen tool from tool_registry import describe_tools, run_tool # Import the functions that record run summaries, time spans of work, and start new traces from tracing import record_run, span, start_trace # Create a number variable called MAX_STEPS, the most think/act/observe steps allowed per resume MAX_STEPS = 10 # Create a number variable called REPEAT_LIMIT, how many identical calls in a row trigger the guard REPEAT_LIMIT = 3 # Create a number variable called COMPACTION_THRESHOLD, the token count that triggers compaction COMPACTION_THRESHOLD = 6000 # Create a text variable called LOG_PATH pointing to the log file this run will write to LOG_PATH = os.path.join(ROOT, "logs", "run_log.jsonl") # Create a flag variable called USE_REPEAT_GUARD controlling whether the repeat guard is active USE_REPEAT_GUARD = True # Define a function that appends one structured event to the log file def log_event(kind, **data): # Create the logs folder if it does not already exist os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True) # Build the base log record with the current time and the kind of event record = {"time": round(time.time(), 3), "event": kind} # Add in any extra named details passed to this function record.update(data) # Open the log file in append mode as UTF-8 text with open(LOG_PATH, "a", encoding="utf-8") as f: # Write the record as one line of JSON, followed by a newline f.write(json.dumps(record) + "\n") # Define a function that runs the full think/act/observe loop for one resume, with tracing and a repeat guard def review_one(resume_file, history, use_context=True, max_steps=MAX_STEPS): # Start a new trace for this resume's work trace_id = start_trace() # Record the start time so total latency can be measured began = time.time() # Build the starting task instruction naming the resume file to review task = "Review the resume at resumes/%s against the job posting and write a report." % resume_file # Add this task as the next user message in the shared history history.append({"role": "user", "content": task}) # Start the running totals: no tokens used yet, level unknown, status defaults to step limit tokens_used, level, status, steps = 0, "unknown", "step_limit", max_steps # Track the signature of the last call made, and how many times it has repeated in a row last_signature, repeats = None, 0 # Repeat the think/act/observe cycle up to the maximum number of steps for step in range(1, max_steps + 1): # Decide what the model actually sees: the trimmed context, or the full raw history sent = build_context(history) if use_context else history # Time the model call as its own span, recording its input and output with span("model", "call_model", json.dumps(sent)[-200:]) as s: reply = call_model(sent, describe_tools()) s["output"] = json.dumps(reply) # Add this step's rough token cost to the running total tokens_used += estimate_tokens(json.dumps(sent)) # Record the model's reply in the log log_event("model_reply", resume=resume_file, step=step, reply=reply) # If the model says the task is finished, record that and stop the loop if reply["type"] == "final": status, steps = "done", step break # Build a signature identifying this exact tool call, so repeats can be detected signature = json.dumps([reply["tool"], reply["args"]], sort_keys=True) # Count this as a repeat if it matches the last call's signature, otherwise reset the count to one repeats = repeats + 1 if signature == last_signature else 1 last_signature = signature # If the repeat guard is on and the same call has repeated too many times in a row if USE_REPEAT_GUARD and repeats >= REPEAT_LIMIT: # Record the repeat-guard event in the log log_event("repeat_guard", resume=resume_file, step=step, tool=reply["tool"]) # Add a system message telling the model to stop looping and write a report instead history.append({"role": "system", "content": "GUARD: the same call was repeated with no new information. " "Stop searching and write a report explaining the problem."}) # Reset the repeat tracking and skip straight to the next step last_signature, repeats = None, 0 continue # Otherwise, add the model's tool-call reply to the shared history history.append({"role": "assistant", "content": json.dumps(reply)}) # Time the tool call as its own span, recording its input and output with span("tool", reply["tool"], json.dumps(reply["args"], sort_keys=True)) as s: # Attempt to run the tool through both hooks try: args = pre_execution_hook(reply["tool"], reply["args"]) result = post_execution_hook(reply["tool"], run_tool(reply["tool"], args)) # If the permission gate refused the request, record that as the result except GateDenied as problem: result = "DENIED: " + str(problem) # If anything else went wrong, record that as the result except Exception as problem: result = "ERROR: " + str(problem) # Record the tool's output on the span, truncated to 200 characters s["output"] = result[:200] # If this step wrote the report, try to read the match level back out of its text if reply["tool"] == "write_report": found = re.search(r"Match level: (\w+)", reply["args"].get("text", "")) level = found.group(1) if found else level # Record the tool result in the log, truncated to 200 characters log_event("tool_result", resume=resume_file, step=step, tool=reply["tool"], result=result[:200]) # Add the tool result to the shared history so the model can see it next step history.append({"role": "tool", "content": result}) # Append one important decision line to the history so it survives any future compaction history.append({"role": "assistant", "important": True, "content": "DECISION: %s -> %s" % (resume_file, level)}) # Build this resume's full outcome, including its trace id and total latency outcome = {"trace_id": trace_id, "resume": resume_file, "status": status, "steps": steps, "tokens": tokens_used, "level": level, "latency_ms": round((time.time() - began) * 1000, 2)} # Record this outcome as a run summary record_run(outcome) # Return this resume's full outcome return outcome # Define a function that clears out reports, logs, and progress notes so a batch can start fresh def reset_state(): # Remove the reports and logs folders entirely, ignoring errors if they do not exist for folder in ("reports", "logs"): shutil.rmtree(os.path.join(ROOT, folder), ignore_errors=True) # Remove the progress notes file if it exists if os.path.exists(NOTES_PATH): os.remove(NOTES_PATH) # Define a function that reviews every not-yet-done resume in one batch, sharing one history def run_batch(use_context=True, stop_after=None): # Create the reports folder if it does not already exist os.makedirs(os.path.join(ROOT, "reports"), exist_ok=True) # Load which resumes are already finished, start an empty shared history and results list done, history, results = load_done(), [], [] # Walk through every resume file in the resumes folder for name in safe_listdir("resumes"): # Skip this resume if the notes file says it is already finished if name in done: print("skipping (already done):", name) continue # If a stop-after limit was given and reached, simulate a crash and stop the batch if stop_after is not None and len(results) >= stop_after: print("simulated stop: pretend the program crashed here") break # Review this resume, adding to the shared history outcome = review_one(name, history, use_context) # Record this resume as done in the notes file append_note(name, outcome["level"]) # Compact the shared history if it has grown past the threshold (only when using context) history = maybe_compact(history, COMPACTION_THRESHOLD) if use_context else history # Add this outcome to the results list results.append(outcome) # Print this outcome to the screen print(outcome) # Add up the estimated tokens used across every resume in this batch total = sum(r["tokens"] for r in results) # Print a summary line for the whole batch print("resumes processed: %d, estimated tokens sent: %d" % (len(results), total)) # Return the total estimated tokens used return total # If this file is run directly (not imported), read the command-line arguments and act on them if __name__ == "__main__": # Collect the command-line arguments, skipping the script name itself args = sys.argv[1:] # If the fresh flag was given, clear all state before running if "--fresh" in args: reset_state() # If the redo flag was given, re-run just the one named resume with a fresh history and stop if "--redo" in args: print(review_one(args[args.index("--redo") + 1], [])) sys.exit() # If a stop-after value was given, read the number right after that flag; otherwise there is no limit stop = int(args[args.index("--stop-after") + 1]) if "--stop-after" in args else None # Run one batch with the chosen stop-after limit run_batch(stop_after=stop) [END OF PYTHON CODE] Look for the three additions. The lines starting with with span wrap the model call and the tool call. The variable signature describes a call as its tool name plus its arguments, so two identical calls have identical signatures. And record_run writes the summary line that the dashboard reads. Planted Bug: The Runaway Loop A runaway loop is a loop that keeps doing the same useless thing until something outside stops it. It is like a person pressing an elevator button over and over, convinced that the button will eventually work. We now plant one on purpose. An empty or damaged file is an ordinary event in real life, and a careless agent may not know what to do about it. This script creates a completely empty resume named resume_empty.txt. File: make_empty_data.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: planting an empty resume to trigger a runaway loop """make_empty_data.py - adds the planted runaway-loop trigger: an empty resume file.""" # Import the os module to work with file paths and folders import os # Create a text variable called HERE holding the folder that contains this file HERE = os.path.dirname(os.path.abspath(__file__)) # Create this same empty file in both the sample_data folder and the resumes folder for folder in ("sample_data", "resumes"): # Create the destination folder if it does not already exist os.makedirs(os.path.join(HERE, folder), exist_ok=True) # Open the file for writing and immediately close it, leaving it completely empty open(os.path.join(HERE, folder, "resume_empty.txt"), "w", encoding="utf-8").close() # Print a confirmation message once the file has been written print("resume_empty.txt written.") [END OF PYTHON CODE] We also teach the mock model a bad habit. Add this block to the mock reply function in model_adapter.py, just before the line that checks whether the search has been done. When the resume has no skills line, the stand-in keeps searching the posting again and again, never writing a report. [START OF PYTHON CODE] # If the resume has no skills line at all (such as an empty file), this special handling kicks in if not any(l.lower().startswith("skills:") for l in resume.splitlines()): # If the report has already been written, the task is finished if "write_report" in done: return {"type": "final", "text": "Report saved."} # If the repeat guard has already fired once, write an honest report instead of searching again if any(m["role"] == "system" and m["content"].startswith("GUARD") for m in mine): stem = os.path.splitext(os.path.basename(path))[0] return {"type": "tool_call", "tool": "write_report", "args": { "name": stem, "text": "The resume file is empty or unreadable, so no review " "was possible. Match level: unreadable."}} # Otherwise, keep asking to search the job posting, which (without the guard) would loop forever return {"type": "tool_call", "tool": "search_job_posting", "args": {"query": "Required skills"}} [END OF PYTHON CODE] The part about GUARD is the repair path, and it only wakes up when the harness sends a message beginning with that word. Everything before it is the bug.