PROJECT FIVE - MULTI-AGENT AND PRODUCTION HARDENING

OUR DESIGN Our supervisor loops over the resume files. For each one it calls the matcher, a subagent that reads the resume, searches the posting, and returns a small structured result listing the candidate, the matched skills, the missing skills, and the level. Then it calls the writer, a subagent that receives only that small result and writes the final report. Notice what the writer never sees: the resume text, the posting, and the matcher's tool chatter. That is context isolation in action. Model routing means choosing which model tier handles which task. The matcher does a simple comparison, so it runs on a cheap, fast tier. The writer composes the final words, so it runs on a stronger tier. Our lab prices these tiers in made-up cost units, a cheap tier at 0.2 units per thousand tokens and a strong tier at 1.0, so that the arithmetic is clear without claiming any real price. Step 1: Routing and the Cost Ledger The routing file defines the two tiers, the batch budget, and a ledger that counts tokens per tier. File: routing.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: model tiers and the cost ledger, in made-up lab units """routing.py - model tiers and the cost ledger. Costs are made-up lab units, not real prices.""" # Import the os module to read environment variables import os # Create a dictionary variable called TIERS defining the cheap and strong model tiers and their costs TIERS = { "cheap": {"model": os.environ.get("LAB_CHEAP_MODEL", "small-model-name"), "cost_per_1k": 0.2}, "strong": {"model": os.environ.get("LAB_STRONG_MODEL", "large-model-name"), "cost_per_1k": 1.0}, } # Create a number variable called BATCH_BUDGET, the most lab cost units a whole batch may spend BATCH_BUDGET = 60.0 # Create a dictionary variable called ledger tracking how many tokens have been spent on each tier ledger = {"cheap": 0, "strong": 0} # Define a function that adds tokens spent to the running total for a given tier def charge(tier, tokens): ledger[tier] += tokens # Define a function that calculates the total lab cost across every tier spent so far def total_cost(): return sum(ledger[t] / 1000 * TIERS[t]["cost_per_1k"] for t in ledger) # Define a function that resets every tier's spent-token count back to zero def reset_ledger(): for tier in ledger: ledger[tier] = 0 [END OF PYTHON CODE] Step 2: One General Loop for Every Agent We now turn main_loop.py into a general runner that any subagent can use. Each call to run_agent starts with an empty history holding only the task, runs the think, act, and observe cycle, and returns an outcome. It keeps everything from the earlier stages: spans, the repeated-call guard, and the hooks. It adds the reliability features of this chapter, which deserve a careful explanation. Retries with exponential backoff. Some failures are temporary, like a phone line that is busy for a moment. A retry simply tries again. Backoff means waiting between tries, and exponential backoff means doubling the wait after each failure, here 0.2 seconds, then 0.4, then 0.8. The doubling is polite to a struggling service, because hammering it with instant retries makes its trouble worse. A maximum number of attempts, here four, ensures that hope has an end. The ToolUnavailable error. When all four attempts fail, the loop does not keep asking. It stops the subagent with the honest status tool_unavailable, and the supervisor reacts. This is a form of graceful failure, ending cleanly with a clear reason, rather than flailing. The maximum-iteration safeguard. Each subagent has its own step limit passed in by its caller, so neither can run forever. File: main_loop.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: Stage 5, one general loop any subagent can use """main_loop.py - Stage 5: one general loop that any subagent can use.""" # Import the json module to read and write structured data import json # Import the os module to work with file paths and environment variables import os # Import the time module to measure delays and record when events happen import time # Import the token-estimating function from context_builder import estimate_tokens # 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 and the custom permission error type from permission_gate import ROOT, GateDenied # Import the function that charges tokens spent against a model tier from routing import charge # 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 REPEAT_LIMIT, how many identical calls in a row trigger the guard REPEAT_LIMIT = 3 # Create a number variable called MAX_ATTEMPTS, the most times a failing tool call is retried MAX_ATTEMPTS = 4 # Create a number variable called BASE_DELAY, the starting wait time between retries, read from the environment BASE_DELAY = float(os.environ.get("LAB_BACKOFF_BASE", "0.2")) # 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") # Define a custom error type raised when a tool keeps failing even after every retry class ToolUnavailable(Exception): # A docstring explaining when this error is raised """A tool kept failing even after every retry.""" # 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 calls a tool, retrying with doubling waits if it fails with a connection error def call_with_retries(tool, args): # A docstring explaining the retry and backoff behavior """Try a tool up to MAX_ATTEMPTS times, waiting twice as long after each failure.""" # Try the tool call up to the maximum number of attempts for attempt in range(1, MAX_ATTEMPTS + 1): # Attempt to run the tool try: return run_tool(tool, args) # If the tool failed with a connection error except ConnectionError as problem: # Record this retry attempt in the log log_event("retry", tool=tool, attempt=attempt, problem=str(problem)) # If this was the last allowed attempt, give up and raise the tool-unavailable error if attempt == MAX_ATTEMPTS: raise ToolUnavailable("%s failed %d times in a row" % (tool, MAX_ATTEMPTS)) # Otherwise, wait before trying again, doubling the wait time after each failed attempt time.sleep(BASE_DELAY * 2 ** (attempt - 1)) # Define a function that runs one subagent on one task, starting from a fresh, empty history def run_agent(task, tier, max_steps, label): # A docstring explaining that this subagent sees only its own task, not any shared history """Run one subagent on one task. It starts with an empty history, so it sees only its task.""" # Start a new trace and record the start time trace_id, began = start_trace(), time.time() # Start the history with just this subagent's task as the first user message history = [{"role": "user", "content": task}] # Start the running totals: no tokens or steps yet, status defaults to step limit, no final text yet tokens, steps, status, final_text = 0, 0, "step_limit", "" # 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): # Record the current step number steps = step # Time the model call as its own span, labeled with the model tier being used with span("model", "call_model:" + tier, json.dumps(history)[-200:]) as s: reply = call_model(history, describe_tools(), tier) s["output"] = json.dumps(reply) # Estimate the token cost of this step and add it to both the running total and the tier ledger used = estimate_tokens(json.dumps(history)) tokens += used charge(tier, used) # If the model says the task is finished, record the final text and stop the loop if reply["type"] == "final": status, final_text = "done", reply["text"] 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 same call has repeated too many times in a row if repeats >= REPEAT_LIMIT: # Record the repeat-guard event in the log log_event("repeat_guard", agent=label, step=step, tool=reply["tool"]) # Add a system message telling the model to stop looping and report the problem history.append({"role": "system", "content": "GUARD: the same call was repeated with no new information. " "Stop and report 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 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 the hooks and the retry-with-backoff wrapper try: args = pre_execution_hook(reply["tool"], reply["args"]) result = post_execution_hook(reply["tool"], call_with_retries(reply["tool"], args)) # If the permission gate refused the request, record that as the result except GateDenied as problem: result = "DENIED: " + str(problem) # If the tool kept failing even after every retry, record that, mark the status, and stop except ToolUnavailable as problem: s["output"] = str(problem) status = "tool_unavailable" log_event("tool_unavailable", agent=label, tool=reply["tool"]) break # 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] # Add the tool result to the history so the model can see it next step history.append({"role": "tool", "content": result}) # Build this subagent's full outcome, including its trace id, tier, and total latency outcome = {"trace_id": trace_id, "resume": label, "status": status, "steps": steps, "tokens": tokens, "level": "n/a", "final_text": final_text, "latency_ms": round((time.time() - began) * 1000, 2)} # Record this outcome as a run summary, leaving out the full final text to keep the summary short record_run({k: v for k, v in outcome.items() if k != "final_text"}) # Return this subagent's full outcome return outcome [END OF PYTHON CODE] Step 3: The Two Subagents The matcher is short. It builds a task beginning with the word MATCH, runs the general loop on the cheap tier with a limit of eight steps, and turns the final text into a dictionary. File: matcher_agent.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: scoring one resume on the cheap model tier """matcher_agent.py - scores one resume against the posting. Runs on the cheap tier.""" # Import the json module to read and write structured data import json [END OF PYTHON CODE] from main_loop import run_agent [START OF PYTHON CODE] # Create a number variable called MATCHER_MAX_STEPS, the step limit for the matcher subagent MATCHER_MAX_STEPS = 8 # Define a function that runs the matcher subagent on one resume and returns its structured result def match(resume_file): # Build the matcher's task instruction, starting with the word MATCH task = "MATCH resumes/%s" % resume_file # Run the general subagent loop on the cheap tier with the matcher's step limit outcome = run_agent(task, "cheap", MATCHER_MAX_STEPS, "matcher:" + resume_file) # If the subagent did not finish successfully, return no data along with the outcome if outcome["status"] != "done": return None, outcome # Attempt to parse the subagent's final text as JSON try: data = json.loads(outcome["final_text"]) # If it was not valid JSON, return no data along with the outcome except ValueError: return None, outcome # Record which resume this result belongs to data["resume"] = resume_file # Return the parsed match data along with the outcome return data, outcome [END OF PYTHON CODE] The writer is shorter still, and the key line is the one that builds the slim dictionary. Only five named fields go through. File: writer_agent.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: turning the matcher's result into a report, on the strong tier """writer_agent.py - turns the matcher's small structured result into a report. Strong tier.""" # Import the json module to read and write structured data import json [END OF PYTHON CODE] from main_loop import run_agent [START OF PYTHON CODE] # Create a number variable called WRITER_MAX_STEPS, the step limit for the writer subagent WRITER_MAX_STEPS = 4 # Define a function that runs the writer subagent on the matcher's result and returns the written report def write(match_data): # Build a slim dictionary holding only the five fields the writer actually needs slim = {key: match_data[key] for key in ("resume", "candidate", "matched", "missing", "level")} # Build the writer's task instruction, starting with the word WRITE, followed by the slim data as JSON task = "WRITE " + json.dumps(slim) # Run the general subagent loop on the strong tier with the writer's step limit return run_agent(task, "strong", WRITER_MAX_STEPS, "writer:" + match_data["resume"]) [END OF PYTHON CODE] Step 4: The Supervisor The supervisor ties the pieces together. It skips resumes already listed in the notes, checks the budget before starting each new one, calls the matcher and then the writer, records progress, and keeps a list of failures rather than hiding them. If the matcher reports that a tool is unavailable, the supervisor stops the whole batch, because the same broken tool would fail again on the next resume and further tries would only waste time. File: supervisor.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: handing each resume to the matcher then the writer, watching the budget """supervisor.py - hands each resume to the matcher, then the writer, and watches the budget.""" # Import the os module to work with file paths and folders import os # Import the shutil module to remove whole folders import shutil # Import the sys module to read command-line arguments import sys # Import the notes file path and the functions that track and load progress from context_builder import NOTES_PATH, append_note, load_done # Import the function that runs the matcher subagent from matcher_agent import match # Import the shared root folder and the safe folder-listing function from permission_gate import ROOT, safe_listdir # Import the batch budget, the spending ledger, and the functions to reset and total it from routing import BATCH_BUDGET, ledger, reset_ledger, total_cost # Import the function that runs the writer subagent from writer_agent import write # Define a function that clears out reports, logs, progress notes, and the spending ledger 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) # Reset the spending ledger back to zero for every tier reset_ledger() # Define a function that reviews every not-yet-done resume, watching the budget and recording failures def run_batch(): # 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, and start counters for finished, failed, and total steps done, finished, failed, steps = load_done(), 0, [], 0 # 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: continue # If the budget has already been exceeded, stop the whole batch here if total_cost() > BATCH_BUDGET: print("BUDGET EXCEEDED: stopping before", name) failed.append((name, "budget")) break # Run the matcher subagent on this resume data, m_out = match(name) # Add the matcher's steps to the running total steps += m_out["steps"] # If the matcher did not produce usable data if data is None: # Record this resume as failed, with the matcher's status as the reason failed.append((name, m_out["status"])) print("matcher failed on", name, "->", m_out["status"]) # If the matcher failed because a tool was unavailable, stop the whole batch if m_out["status"] == "tool_unavailable": break # Otherwise, move on to the next resume continue # Run the writer subagent on the matcher's result w_out = write(data) # Add the writer's steps to the running total steps += w_out["steps"] # If the writer did not finish successfully, record this resume as failed and move on if w_out["status"] != "done": failed.append((name, "writer_" + w_out["status"])) continue # Record this resume as done, with its match level, in the notes file append_note(name, data["level"]) # Count this resume as finished finished += 1 # Print how many resumes finished and which ones failed print("finished: %d, failed: %s" % (finished, failed or "none")) # Print the total steps taken, the spending ledger by tier, and the total cost against the budget print("steps: %d, tokens by tier: %s, cost: %.2f lab units (budget %.0f)" % ( steps, ledger, total_cost(), BATCH_BUDGET)) # If this file is run directly (not imported), optionally reset state, then run one batch if __name__ == "__main__": # If the fresh flag was given, clear all state before running if "--fresh" in sys.argv: reset_state() # Run one batch run_batch() [END OF PYTHON CODE] Step 5: Updating the Remaining Files The mock model must now understand two kinds of task. A task beginning with MATCH asks for a structured result, and a task beginning with WRITE asks for a report. Replace model_adapter.py with the version below. Its comparison logic is the same as before, and it keeps the planted gullibility and the empty-file behavior from earlier chapters. File: model_adapter.py """model_adapter.py - the one place that talks to a model. LAB_MODEL_MODE=mock (default) uses a scripted stand-in that needs no account. Any other value sends the conversation to Google Gemini through hosted_gemini.py. """ import json import os import re [START OF PYTHON CODE] # Create a text variable called MODE, read from the LAB_MODEL_MODE environment variable, defaulting to "mock" MODE = os.environ.get("LAB_MODEL_MODE", "mock") # Define a function that sends a conversation to whichever model is configured and returns its reply def call_model(messages, tools, tier="cheap"): # If running in mock mode, answer with the small scripted stand-in instead of a real model if MODE == "mock": return _mock_reply(messages) # Otherwise, forward the conversation on to the real hosted model return call_hosted_model(messages, tools, tier) # Define a function that sends a conversation to the real hosted Gemini model def call_hosted_model(messages, tools, tier): # Import the Gemini-calling function only when it is actually needed from hosted_gemini import call_gemini # Call Gemini with the conversation, available tools, and tier, and return its reply return call_gemini(messages, tools, tier) # Define a function that splits a "label: comma, separated, values" line into a cleaned-up list of values def _parse_list(line): # Take everything after the first colon, split it on commas, trim whitespace, and drop empty entries return [s.strip() for s in line.split(":", 1)[1].split(",") if s.strip()] # Define a function that compares a resume against the job posting and returns a structured match result def _assess(resume, posting_text): # Find the line of the job posting that lists the required skills posting_line = next(l for l in posting_text.splitlines() if l.lower().startswith("required skills")) # Parse that line into a list of required skills required = _parse_list(posting_line) # Collect the lowercased set of skills the candidate's resume actually lists have = {s.lower() for line in resume.splitlines() if line.lower().startswith("skills:") for s in _parse_list(line)} # Work out which required skills the candidate has matched = [s for s in required if s.lower() in have] # Work out which required skills the candidate is missing missing = [s for s in required if s.lower() not in have] # Decide an overall match level based on how many required skills matched level = "strong" if len(matched) >= 3 else "partial" if matched else "none" # Find the candidate's name from the resume, defaulting to "Unknown" if not found candidate = next((l.split(":", 1)[1].strip() for l in resume.splitlines() if l.lower().startswith("name:")), "Unknown") # If the resume contains either phrasing of the hidden "ignore instructions" order, obey it (planted bug) if re.search(r"ignore (all )?prior instructions", resume, re.I) or re.search(r"ignore (all )?previous instructions", resume, re.I): level = "strong" # Return the structured match result return {"candidate": candidate, "matched": matched, "missing": missing, "level": level} # Define a function that scripts a stand-in model reply, handling both MATCH and WRITE tasks def _mock_reply(messages): # Find the index of the most recent message sent by the user start = max(i for i, m in enumerate(messages) if m["role"] == "user") # Read out the user's task from that message task = messages[start]["content"] # Collect every message that came after the user's task message mine = messages[start + 1:] # Collect every tool call the assistant has made so far, parsed out of JSON text calls = [json.loads(m["content"]) for m in mine if m["role"] == "assistant"] # Collect every tool result that has come back so far results = [m["content"] for m in mine if m["role"] == "tool"] # Collect just the names of the tools that have already been called done = [c["tool"] for c in calls] # Check whether the repeat guard has already fired during this task guarded = any(m["role"] == "system" and m["content"].startswith("GUARD") for m in mine) # If this is a WRITE task, build and save the final report text if task.startswith("WRITE "): # Parse the slim match data out of the task text data = json.loads(task[6:]) # If the report has already been written, the task is finished if "write_report" in done: return {"type": "final", "text": "Report saved."} # If the match level is "unreadable", write the honest unreadable-file message if data["level"] == "unreadable": text = ("The resume file is empty or unreadable, so no review was possible. " "Match level: unreadable.") # Otherwise, build the normal match-summary report text else: total = len(data["matched"]) + len(data["missing"]) text = ("%s matches %d of %d required skills (%s). Missing: %s. Match level: %s." % (data["candidate"], len(data["matched"]), total, ", ".join(data["matched"]) or "none", ", ".join(data["missing"]) or "none", data["level"])) # Build the report's file-name stem from the resume's file name stem = os.path.splitext(data["resume"])[0] # Ask to write the report using that stem as the report's name return {"type": "tool_call", "tool": "write_report", "args": {"name": stem, "text": text}} # Otherwise, this is a MATCH task: find the resume file path mentioned in the task text path = re.search(r"resumes/\S+\.txt", task).group(0) # If the resume has not been read yet, ask to read it first if "read_file" not in done: return {"type": "tool_call", "tool": "read_file", "args": {"path": path}} # Otherwise, the resume text is the first tool result received resume = results[0] # If reading the resume failed or was denied, return an empty final result if resume.startswith(("DENIED", "ERROR")): return {"type": "final", "text": "{}"} # If the resume has no skills line at all (such as an empty file) if not any(l.lower().startswith("skills:") for l in resume.splitlines()): # If the repeat guard has already fired, give up and report the file as unreadable if guarded: return {"type": "final", "text": json.dumps({ "candidate": "Unknown", "matched": [], "missing": [], "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"}} # If the job posting has not been searched yet, ask to search it for the required skills section if "search_job_posting" not in done: return {"type": "tool_call", "tool": "search_job_posting", "args": {"query": "Required skills"}} # Otherwise, assess the resume against the posting and return the structured result as the final answer return {"type": "final", "text": json.dumps(_assess(resume, results[1]))} [END OF PYTHON CODE] Add a failure switch so that you can simulate a broken tool. Insert these two lines in run_tool in tool_registry.py, just after the check that the arguments are a dictionary. [START OF PYTHON CODE] # If the environment names this exact tool as the one to simulate failing, raise a connection error on purpose if os.environ.get("LAB_FAIL_TOOL") == name: raise ConnectionError("simulated outage of " + name) [END OF PYTHON CODE] In dashboard.py change the success test so that it only requires the status done, because the new runner no longer records a match level in the run summary. The line becomes the one below. [START OF PYTHON CODE] # Count a run as successful based only on its status, since this architecture no longer stores a match level here ok = [r for r in runs if r["status"] == "done"] [END OF PYTHON CODE] Finally replace eval_runner.py so that the evaluation exercises the new architecture. File: eval_runner.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: running every eval item through matcher and writer, then scoring it """eval_runner.py - runs every eval item through matcher and writer, then scores it.""" # Import the json module to read structured data import json # Import the os module to work with file paths import os # Import the re module to search text with patterns import re # Import the function that runs the matcher subagent from matcher_agent import match # Import the shared root folder from permission_gate import ROOT # Import the function that runs the writer subagent from writer_agent import write # Define a function that runs the whole evaluation set through the matcher and writer and prints a scorecard def run_eval(): # Open and read the evaluation set file with open(os.path.join(ROOT, "eval_set.json"), "r", encoding="utf-8") as f: items = json.load(f) # Start a passed-count of zero passed = 0 # Walk through every evaluation item for item in items: # Work out the expected report file for this item's resume stem = os.path.splitext(item["input_file"])[0] report = os.path.join(ROOT, "reports", stem + ".txt") # Delete any old report first, so a leftover file cannot make this test look wrong if os.path.exists(report): os.remove(report) # Run the matcher subagent on this resume data, m_out = match(item["input_file"]) # Start the step count from the matcher's steps, and default the result to "missing" steps, got = m_out["steps"], "missing" # If the matcher produced usable data, run the writer too and read the result back if data is not None: # Add the writer's steps to the running total steps += write(data)["steps"] # If a report was written, read the match level back out of it if os.path.exists(report): with open(report, "r", encoding="utf-8") as f: found = re.search(r"Match level: (\w+)", f.read()) got = found.group(1) if found else "missing" # This item passes if the match level is exactly what was expected ok = got == item["golden_level"] # Add one to the passed count if this item passed passed += ok # Print this item's result as one formatted line print("%s %-26s expected=%-10s got=%-10s steps=%2d %s" % ( item["id"], item["input_file"], item["golden_level"], got, steps, "PASS" if ok else "FAIL")) # Print the overall pass rate across every item print("Pass rate: %d of %d (%.0f%%)" % (passed, len(items), 100 * passed / len(items))) # If this file is run directly (not imported), run the full evaluation if __name__ == "__main__": run_eval() [END OF PYTHON CODE]