← Table of Contents · Page 665 of 868

PROJECT ONE - A MINIMAL HARNESS

Part 3 of 11

WHAT YOU NEED You need Python, a beginner-friendly programming language, in a current supported version, plus a terminal and any plain text editor. You do not need any extra packages for Stage 1, because the code uses only what Python already includes. Appendix G explains the full setup in detail. Create a folder named resume_review_agent anywhere you like, open a terminal inside it, and save each file below with the exact name shown. The code listings in Chapters 65 through 69 are printed with four extra spaces at the start of every line, to set them apart from the text. When you save a file, remove exactly those first four spaces from every line, so that the first line of the file starts at the left edge, because Python reports an indentation error when a file begins with extra spaces. The spaces that remain after you remove the four belong to the code, so keep them. Short snippets that you insert into an existing file, as in Chapters 66 through 69, follow the same rule, and you indent them to line up with the lines around them in your file. The listings in Appendix H have no extra spaces and are typed exactly as shown. Step 1: The Permission Gate Before the agent can do anything, we decide what it must never do. The permission gate is a small piece of code that every file request must pass through, like a security guard at a door who checks a list before letting anyone in. Our rule is simple: the agent may read only inside the resumes folder, and may write only inside the reports folder. The tricky part is that a crafted path such as ../secret.txt uses two dots to climb out of a folder, like finding a back staircase around the guard. To close that loophole the gate first converts every path into its full, final form, removing the climbing dots and following any shortcuts, and only then checks whether the result lies inside the allowed folder. File: permission_gate.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: deciding which folders the agent may access """permission_gate.py - decides which folders the agent may touch.""" # Import the os module to work with file paths and the filesystem import os # Create a text variable called ROOT holding the folder that contains this file ROOT = os.path.dirname(os.path.abspath(__file__)) # Create a text variable called READ_FOLDER pointing to the resumes subfolder, the only folder the agent may read from READ_FOLDER = os.path.join(ROOT, "resumes") # Create a text variable called WRITE_FOLDER pointing to the reports subfolder, the only folder the agent may write to WRITE_FOLDER = os.path.join(ROOT, "reports") # Define a custom error type raised whenever the gate refuses a request class GateDenied(Exception): # A docstring explaining when this error is raised """Raised when the agent asks for something the gate does not allow.""" # Define a function that resolves a path into its full, real, final form so that tricks like ../ cannot be used to escape the allowed folders def _resolve(path): # Return the normalized, case-folded, real path, joined onto the root folder return os.path.normcase(os.path.realpath(os.path.join(ROOT, path))) # Define a function that checks whether a resolved path lies inside a given folder def _inside(path, folder): # Normalize and resolve the folder path the same way, so the comparison is fair folder = os.path.normcase(os.path.realpath(folder)) # Attempt the containment check, since mismatched drives or roots can raise an error try: # Return true only if the common path between the two is exactly the folder itself return os.path.commonpath([path, folder]) == folder # If no common path exists at all, treat that as not being inside the folder except ValueError: return False # Define a function that checks whether the agent may read the given path, raising an error if not def check_read(path): # Resolve the path to its full, final form first full = _resolve(path) # If the resolved path is not inside the allowed read folder, refuse the request if not _inside(full, READ_FOLDER): raise GateDenied("read not allowed outside the resumes folder: " + path) # Otherwise, return the resolved path so the caller can use it return full # Define a function that checks whether the agent may write to the given path, raising an error if not def check_write(path): # Resolve the path to its full, final form first full = _resolve(path) # If the resolved path is not inside the allowed write folder, refuse the request if not _inside(full, WRITE_FOLDER): raise GateDenied("write not allowed outside the reports folder: " + path) # Otherwise, return the resolved path so the caller can use it return full # Define a function that safely lists the contents of a folder, but only if that folder is one of the allowed folders def safe_listdir(folder_name): # Resolve the folder name to its full, final form first full = _resolve(folder_name) # If the resolved folder is neither the read folder nor the write folder, refuse the request if not (_inside(full, READ_FOLDER) or _inside(full, WRITE_FOLDER)): raise GateDenied("listing not allowed: " + folder_name) # Otherwise, return a sorted list of the folder's contents return sorted(os.listdir(full)) [END OF PYTHON CODE] Read the code from the top. ROOT is the folder containing the project, found from the location of the file itself. GateDenied is our own kind of error, raised whenever the gate says no. The function named _resolve builds the full final path. The function named _inside answers one question: is this path really inside that folder? The functions check_read and check_write are the two doors, one for reading and one for writing. Finally safe_listdir lets the harness list a folder, but only the two permitted ones. Step 2: The Tool Registry A tool is a named action that the harness can run for the model, such as reading a file. The tool registry is the menu of all available tools, where each entry has a name, a plain description that the model reads, and a handler, which is the actual code that does the work. If a tool is not on the menu, the model cannot use it, no matter how politely it asks. Our menu has exactly three items: read_file, search_job_posting, and write_report. File: tool_registry.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: the menu of tools the agent may ask for """tool_registry.py - the menu of tools the agent may ask for.""" # Import the os module to work with file paths import os # Import the shared root folder and the two permission-check functions from the permission gate from permission_gate import ROOT, check_read, check_write # Create a text variable called POSTING_PATH pointing to the sample job posting file POSTING_PATH = os.path.join(ROOT, "sample_data", "job_posting.txt") # Define a function that reads and returns the contents of one allowed file, capped at 6000 characters def read_file(path): # Check that this path is allowed to be read, and get its full, final form full = check_read(path) # Open the file for reading as UTF-8 text with open(full, "r", encoding="utf-8") as f: # Read the file and return only the first 6000 characters return f.read()[:6000] # Define a function that searches the job posting for lines containing a given word or phrase def search_job_posting(query): # Open the job posting file for reading as UTF-8 text with open(POSTING_PATH, "r", encoding="utf-8") as f: # Read the whole file and split it into a list of individual lines lines = f.read().splitlines() # Keep only the lines whose lowercased text contains the lowercased query hits = [line for line in lines if query.lower() in line.lower()] # Join the matching lines back into one block of text, or report no match was found return "\n".join(hits) if hits else "NO MATCH" # Define a function that writes a text report to the reports folder under the given name def write_report(name, text): # Check that this destination is allowed to be written to, and get its full, final form full = check_write(os.path.join("reports", name + ".txt")) # Create the destination folder if it does not already exist os.makedirs(os.path.dirname(full), exist_ok=True) # Open the destination file for writing as UTF-8 text with open(full, "w", encoding="utf-8") as f: # Write the report text into the file f.write(text) # Return a confirmation message naming the file that was written return "report written: reports/" + name + ".txt" # Create a dictionary variable called TOOLS listing every tool the agent is allowed to call TOOLS = { # The read_file tool entry "read_file": { # A plain-language description the model reads to decide when to use this tool "description": "Read one resume. Argument: path, such as resumes/resume_01.txt", # The actual function that runs when this tool is called "handler": read_file, }, # The search_job_posting tool entry "search_job_posting": { # A plain-language description the model reads to decide when to use this tool "description": "Find lines of the job posting that contain a word or phrase. Argument: query", # The actual function that runs when this tool is called "handler": search_job_posting, }, # The write_report tool entry "write_report": { # A plain-language description the model reads to decide when to use this tool "description": "Save a one paragraph report. Arguments: name, text", # The actual function that runs when this tool is called "handler": write_report, }, } # Define a function that returns just the name and description of every tool, for showing the model def describe_tools(): # Build and return a dictionary mapping each tool name to its description return {name: tool["description"] for name, tool in TOOLS.items()} # Define a function that looks up a tool by name and runs it with the given arguments def run_tool(name, args): # If the requested tool name is not in the registry, report an error if name not in TOOLS: return "ERROR: unknown tool " + str(name) # If the arguments are not given as a dictionary, report an error if not isinstance(args, dict): return "ERROR: arguments must be a dictionary" # Look up the tool's handler function and call it with the arguments spread out as named parameters return TOOLS[name]["handler"](**args) [END OF PYTHON CODE] Notice three small safety habits in this file. First, read_file never returns more than six thousand characters, so a huge file cannot flood the conversation. Second, every file tool asks the permission gate before it touches the disk. Third, run_tool refuses unknown tool names and wrongly shaped arguments instead of crashing. Now look closely at write_report, because it hides a flaw that we planted on purpose. It opens the report file in write mode, which silently replaces any report that was already there. Running the same job twice would erase the older report with no warning and no record. Do not fix it yet. Chapter 67 will ask you to find this flaw and repair it, and the lesson works best if you meet the bug first. Step 3: The Model Adapter The model adapter is the single doorway through which the harness talks to a model. Keeping all model talk in one file means you can change models by editing one place. Every reply from the adapter has one of two shapes. A tool call says which tool to run and with what arguments, and a final reply says the work is finished. File: model_adapter.py """model_adapter.py - the one place that talks to a model. Two modes, chosen by the environment variable LAB_MODEL_MODE: mock (default) a small scripted stand-in that needs no account and no internet. hosted sends the conversation to Google Gemini through hosted_gemini.py (Step 3B). """ 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="standard"): # 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 scripts a stand-in model reply, step by step, without calling any real model 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] # 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, stop here with a final explanation if resume.startswith(("DENIED", "ERROR")): return {"type": "final", "text": "Could not read the resume."} # 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"}} # If the report has not been written yet, build it now if "write_report" not in done: # Find the line of the job posting search results that starts with "required skills" posting_line = next(l for l in results[1].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 name = next((l.split(":", 1)[1].strip() for l in resume.splitlines() if l.lower().startswith("name:")), "Unknown") # Build the final report text summarizing the match text = ("%s matches %d of %d required skills (%s). Missing: %s. Match level: %s." % (name, len(matched), len(required), ", ".join(matched) or "none", ", ".join(missing) or "none", level)) # Build a short file-name stem from the resume's file name, without its folder or extension stem = os.path.splitext(os.path.basename(path))[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}} # Once the report has been written, the task is complete return {"type": "final", "text": "Report saved."} [END OF PYTHON CODE] In mock mode, which is the default, the stand-in follows a fixed recipe for each resume: read the file, search the posting for its required skills, compare the two lists, write the report, and then say it is done. The function call_hosted_model hands the job to a second file, hosted_gemini.py, which you create in Step 3B only if you want to try a real model. If you prefer another provider, you can copy that file and change it using the provider's own documentation, and you will always keep your secret key in an environment variable, never inside a file. An environment variable is a named setting that lives in your computer's memory rather than in your code, so it cannot be saved by accident into a shared project. Step 3B: Optional, Connect a Real Model (Google Gemini, Free Tier) You can skip this step and finish the whole lab with the mock model. If you want to see your harness work with a real model, this step gives you finished code for Google Gemini. We chose Gemini because Google offers a free tier, which means you can use the model up to a daily limit without paying and, at the time of writing, without entering a payment card. Free tiers change often, so check Google's current rules before you rely on them. Getting your own key An API key is a long secret string of letters and numbers that proves to a service that a request comes from you. An API is an Application Programming Interface, which is the set of rules that lets one program talk to another. Think of the key as a house key: anyone who holds it can use your account, so you keep it private. Follow these steps in order. 1. Open a web browser and go to aistudio.google.com/apikey. 2. Sign in with a Google account. 3. Read and accept Google's terms if the page asks you to. 4. Choose the button that creates an API key, and when asked, create it in a new project. 5. Copy the key that appears, and keep that window open until you finish step 6. 6. Store the key in an environment variable, using the commands shown below for your computer. 7. Close the browser window. An environment variable is a named setting that lives in your computer's memory instead of inside your code files, so it cannot be saved into a shared project by accident. On Windows, using PowerShell, type this and replace the words in capital letters with your key, keeping the quote marks: [START OF POWERSHELL CODE] # Set the environment variable LAB_API_KEY to the API key you copied, as a placeholder to replace $env:LAB_API_KEY = "PASTE-YOUR-KEY-HERE" # Set the environment variable LAB_MODEL_MODE to "hosted" so the real model is used instead of the mock $env:LAB_MODEL_MODE = "hosted" [END OF POWERSHELL CODE] On macOS or Linux, type this instead: [START OF BASH CODE] # Set the environment variable LAB_API_KEY to the API key you copied, as a placeholder to replace export LAB_API_KEY="PASTE-YOUR-KEY-HERE" # Set the environment variable LAB_MODEL_MODE to "hosted" so the real model is used instead of the mock export LAB_MODEL_MODE="hosted" [END OF BASH CODE] These settings last only until you close that terminal window, which is the safest choice for a practice lab. Each time you open a new terminal, type the two lines again. To go back to the mock model at any time, type the word mock in place of hosted, or close the terminal and open a new one. Safety rules for your key: • Never type the key into a code file, into this book, into a chat, or into a screenshot. • Never save the key in a folder that you back up to a public place or share with others. • If you think the key has leaked, return to aistudio.google.com/apikey, delete it, and make a new one. • Remember that, at the time of writing, Google may use content sent through the free tier to improve its products, so send only the fictional lab data and never real resumes or private information. The connection code Create a new file named hosted_gemini.py in your project folder, and copy in the code below. As with every other listing in this book, remove the four spaces at the start of each line before you save the file. It uses only tools that come with Python, so you do not need to install anything. The code asks the model to answer with a single small JSON note, which is a structured piece of text in curly brackets that both people and programs can read, and then turns that note into the tool_call or final dictionary that the rest of the harness expects. The code also waits a few seconds between calls and tries again if the service says it is busy, because free tiers limit how many requests you may send each minute. File: hosted_gemini.py """hosted_gemini.py - connects the lab to Google Gemini using only the standard library. Reads the key from LAB_API_KEY (or GEMINI_API_KEY) in the environment. Never put a key in this file. """ import json import os import time import urllib.error import urllib.request [START OF PYTHON CODE] # Create a text variable called URL holding the Gemini API endpoint template, with a placeholder for the model name URL = "https://generativelanguage.googleapis.com/v1beta/models/%s:generateContent" # Create a text variable called STANDARD_MODEL naming the default, full-capability Gemini model STANDARD_MODEL = "gemini-2.5-flash" # Create a text variable called CHEAP_MODEL naming a lighter, less expensive Gemini model CHEAP_MODEL = "gemini-2.5-flash-lite" # Create a number variable called PAUSE_SECONDS, the base number of seconds to wait between retries PAUSE_SECONDS = 5 # Create a number variable called MAX_TRIES, the most times to retry a request before giving up MAX_TRIES = 4 # Create a text variable called RULES holding the standing system instructions given to the model RULES = ( "You are the decision maker inside a small agent harness that reviews resumes. " "Reply with exactly one JSON object and nothing else. " "To use a tool reply: {\"type\": \"tool_call\", \"tool\": \"TOOL NAME\", \"args\": {\"ARGUMENT NAME\": \"VALUE\"}}. " "When the whole task is finished reply: {\"type\": \"final\", \"text\": \"a short message\"}. " "Use only the tools listed below, with the argument names given in each description. " "Tool results arrive in messages that begin with the words Tool result. " "Tools available as name and description: " ) # Define a function that chooses which Gemini model to use, honoring an environment override if present def _pick_model(tier): # Check whether an environment variable names a specific model to use chosen = os.environ.get("LAB_GEMINI_MODEL") # If one was given, use it if chosen: return chosen # Otherwise, pick the cheap model for the cheap tier, or the standard model otherwise return CHEAP_MODEL if tier == "cheap" else STANDARD_MODEL # Define a function that converts the harness's message list into Gemini's expected request shape def _build_contents(messages): # Create a list variable called contents to hold the converted messages contents = [] # Create a list variable called extra_rules to hold any extra system instructions found along the way extra_rules = [] # Walk through every message in order for m in messages: # Pull out this message's role and text content role, text = m["role"], m["content"] # A system message becomes an extra rule rather than a conversation turn if role == "system": extra_rules.append(text) # An assistant message becomes a "model" turn in Gemini's format elif role == "assistant": contents.append({"role": "model", "parts": [{"text": text}]}) # A tool result becomes a user turn, labeled so the model knows it is a tool result elif role == "tool": contents.append({"role": "user", "parts": [{"text": "Tool result: " + text}]}) # Any other message becomes a plain user turn else: contents.append({"role": "user", "parts": [{"text": text}]}) # Return both the converted conversation and any extra rules collected along the way return contents, extra_rules # Define a function that sends one request to Gemini, retrying with increasing waits if the service is busy def _send(body, model, key): # Convert the request body into JSON bytes data = json.dumps(body).encode("utf-8") # Build the request headers, including the API key headers = {"Content-Type": "application/json", "x-goog-api-key": key} # Start the retry wait at the base number of seconds wait = PAUSE_SECONDS # Try sending the request up to MAX_TRIES times for attempt in range(1, MAX_TRIES + 1): # Build the actual HTTP request, filling the model name into the URL request = urllib.request.Request(URL % model, data=data, headers=headers) # Attempt to send the request and read back the response try: # Open the connection with a 60 second timeout with urllib.request.urlopen(request, timeout=60) as response: # Read, decode, and parse the JSON response, then return it return json.loads(response.read().decode("utf-8")) # If the server itself returned an error status except urllib.error.HTTPError as problem: # Treat these specific status codes as "service is busy, worth retrying" busy = problem.code in (429, 500, 503) # If busy and tries remain, wait, double the wait for next time, and try again if busy and attempt < MAX_TRIES: time.sleep(wait) wait = wait * 2 continue # Otherwise, read a short snippet of the error detail and raise a clear error detail = problem.read().decode("utf-8", "replace")[:300] raise RuntimeError("Gemini returned HTTP %d: %s" % (problem.code, detail)) # If the connection itself failed (no response at all) except urllib.error.URLError as problem: # If tries remain, wait, double the wait for next time, and try again if attempt < MAX_TRIES: time.sleep(wait) wait = wait * 2 continue # Otherwise, raise a clear error explaining the connection failed raise RuntimeError("Could not reach Gemini: %s" % problem.reason) # Define a function that turns the model's raw text reply into a clean, predictable dictionary def _clean(raw_text): # Attempt to parse the raw text as JSON try: reply = json.loads(raw_text) # If it is not valid JSON, report that as a final message except ValueError: return {"type": "final", "text": "The model did not return valid JSON."} # If the parsed JSON is not an object, report that as a final message if not isinstance(reply, dict): return {"type": "final", "text": "The model reply was not a JSON object."} # If this looks like a well-formed tool call, return a clean tool-call dictionary if reply.get("type") == "tool_call" and isinstance(reply.get("tool"), str): args = reply.get("args") return {"type": "tool_call", "tool": reply["tool"], "args": args if isinstance(args, dict) else {}} # If this looks like a well-formed final answer, return a clean final dictionary if reply.get("type") == "final": return {"type": "final", "text": str(reply.get("text", ""))} # Otherwise, the shape was not recognized, so report that as a final message return {"type": "final", "text": "The model reply had an unknown shape."} # Define a function that sends the conversation and tools to Gemini and returns the cleaned-up reply def call_gemini(messages, tools, tier="standard"): # Read the API key from either of two possible environment variable names key = os.environ.get("LAB_API_KEY") or os.environ.get("GEMINI_API_KEY") # If no key was found, refuse to continue with a clear error if not key: raise RuntimeError("No key found. Set LAB_API_KEY in your environment first.") # Convert the harness's messages into Gemini's expected conversation shape contents, extra_rules = _build_contents(messages) # Build the full request body, combining the standing rules, the tool list, and any extra rules body = { "systemInstruction": {"parts": [{"text": RULES + json.dumps(tools) + " " + " ".join(extra_rules)}]}, "contents": contents, "generationConfig": {"temperature": 0.0, "responseMimeType": "application/json"}, } # Send the request to the chosen model and get the raw result back result = _send(body, _pick_model(tier), key) # Pause briefly after every call to respect the free tier's rate limit time.sleep(PAUSE_SECONDS) # Attempt to pull the model's reply text out of the response's expected location try: parts = result["candidates"][0]["content"]["parts"] # If the response was missing or shaped unexpectedly, report that as a final message except (KeyError, IndexError, TypeError): return {"type": "final", "text": "The model returned no usable answer."} # Join all the returned text parts together and clean them into the harness's expected shape return _clean("".join(p.get("text", "") for p in parts)) [END OF PYTHON CODE] Notice that the key is read only from the environment, and it travels in a request header, never in the web address, so it does not show up in logs that record addresses. Notice also that every strange case, such as broken JSON or an empty answer, becomes an ordinary final reply, so a misbehaving model cannot crash your loop. How to run it First, check that the harness still works with the mock model by typing python main_loop.py with LAB_MODEL_MODE set to mock. Then set LAB_MODEL_MODE to hosted, make sure LAB_API_KEY is set, and type python main_loop.py again. Open logs/run_log.jsonl and compare the steps the real model took with the steps the mock model took. Expect differences. A real model may add a step, write a different report sentence, or occasionally choose a wrong tool, and that is exactly why the harness around it has a permission gate, a step limit, and a log. Your numbers in later chapters, such as the tokens saved in Chapter 66, will differ from the ones printed in this book, so repeat a test a few times and average the results. The planted gullible behavior in Chapter 67 belongs to the mock model, so a real model may resist the hidden instruction, obey it, or do something different, and either outcome is worth recording. Troubleshooting: • HTTP 400 or 403: the key is wrong, was pasted with extra spaces, or the service is not available in your region, so create a new key and set it again. • HTTP 404: the model name is out of date, so find a current model name in Google's documentation and set it with LAB_GEMINI_MODEL. • HTTP 429: you reached the free limit for this minute or this day, so wait and try again, or review fewer resumes. • A final reply saying the model did not return valid JSON: run the same task again, because real models sometimes slip. Model names and free limits change often, so treat the names in this listing as examples and check Google's documentation for current ones. Free AI Application Programming Interface Keys You Can Use Instead Gemini is not the only choice, and several providers offer a free tier, which is a plan with no charge up to a usage limit. The list below was gathered from provider pages and public roundups in October 2026. Free tiers are changed, shrunk, or removed without warning, so before you sign up, read the provider's own pricing page, and treat every limit here as something to verify. Providers with a free tier and no payment card required, as last reported: • Google Gemini through Google AI Studio. ◦ Where: aistudio.google.com/apikey. ◦ Notes: this is the one used in this book, and free content may be used to improve Google products. • Groq. ◦ Where: console.groq.com. ◦ Notes: very fast replies, with small per minute token limits that you will probably reach first. • OpenRouter. ◦ Where: openrouter.ai. ◦ Notes: one key reaches many models, and the models marked free change from time to time. • Mistral. ◦ Where: console.mistral.ai. ◦ Notes: an experimental free plan, and some reports say it asks for a phone number to verify your account. • Cohere. ◦ Where: dashboard.cohere.com. ◦ Notes: a trial key meant for learning and testing, not for commercial use. • NVIDIA API catalog. ◦ Where: build.nvidia.com. ◦ Notes: small amounts of free credits, better for trials than for steady use. • Cloudflare Workers AI. ◦ Where: developers.cloudflare.com/workers-ai. ◦ Notes: a small free daily allowance, and it needs a Cloudflare account. • SambaNova. ◦ Where: cloud.sambanova.ai. ◦ Notes: reported as a free tier in recent roundups. Providers whose free offer is unclear or has changed: • Cerebras: reports disagree, with some saying a daily free allowance and others saying a time limited credit that needs a payment method. • Hugging Face: offers access to many models through its inference service, and the free monthly allowance is small and changes, so check its page. • GitHub Models: one report says this service was retired on July 30, 2026, so confirm before relying on it. Providers that have no standing free tier for their main models, as far as we could find, so you pay as you go: • Anthropic. • OpenAI. Free without any key: • Ollama, which runs open models on your own computer at no cost. ◦ Where: ollama.com. ◦ Notes: you download a model of a few gigabytes, and the lab connects to it through its local address, which uses a different request format from the one in hosted_gemini.py. Many of these services copy a request format first made popular by OpenAI, known as the chat completions format, so to use one of them you would copy hosted_gemini.py, change the web address, the header that carries the key, and the shape of the request and answer, and keep the same two dictionaries as the output. That is the benefit of the model adapter: the rest of your harness never needs to know which model sits behind the door. Step 4: The Main Loop Now we connect everything. The main loop shows the conversation to the model, runs the requested tool, adds the result, and repeats. It also records every event in a log file. A log is a diary kept by the program, and we write it in JSON lines, which means one small structured note per line that both people and programs can read. The loop also counts tokens, which are the small chunks of text a model reads and writes, and which are how model use is measured and billed. We estimate tokens with a rough rule of one token per four characters, which is good enough for a budget count. File: main_loop.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: the think / act / observe loop """main_loop.py - the think / act / observe loop.""" # 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 time module to measure and record when events happen import time # 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 # Create a number variable called MAX_STEPS, the most think/act/observe steps allowed per resume MAX_STEPS = 10 # 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 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 roughly estimates how many tokens a piece of text uses def estimate_tokens(text): # Estimate one token per four characters, but never fewer than one token return max(1, len(text) // 4) # Define a function that runs the full think/act/observe loop for one resume file def review_one(resume_file, max_steps=MAX_STEPS): # 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 # Start the conversation with that task as the first user message messages = [{"role": "user", "content": task}] # Start the running token count at zero tokens_used = 0 # Repeat the think/act/observe cycle up to the maximum number of steps for step in range(1, max_steps + 1): # Ask the model what to do next, given the conversation so far and the available tools reply = call_model(messages, describe_tools()) # Add this step's rough token cost to the running total tokens_used += estimate_tokens(json.dumps(messages)) # 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, return the final outcome now if reply["type"] == "final": return {"resume": resume_file, "status": "done", "steps": step, "tokens": tokens_used} # Otherwise, add the model's tool-call reply to the conversation messages.append({"role": "assistant", "content": json.dumps(reply)}) # Attempt to run the requested tool with the requested arguments try: result = run_tool(reply["tool"], reply["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 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 conversation so the model can see it next step messages.append({"role": "tool", "content": result}) # If the step limit was reached without finishing, record that and return the outcome log_event("step_limit", resume=resume_file) return {"resume": resume_file, "status": "step_limit", "steps": max_steps, "tokens": tokens_used} # Define the main function that reviews every resume in the resumes folder, one at a time def main(): # Create the reports folder if it does not already exist os.makedirs(os.path.join(ROOT, "reports"), exist_ok=True) # Walk through every file in the resumes folder for name in safe_listdir("resumes"): # Review this one resume and get its outcome outcome = review_one(name) # Print the outcome to the screen print(outcome) # If this file is run directly (not imported), start the main function if __name__ == "__main__": main() [END OF PYTHON CODE] Find the three beats in review_one. The line that calls call_model is the think beat. The try block that calls run_tool is the act beat. The line that appends the tool result to the messages is the observe beat. Find the step limit as well, which is the range of numbers in the for line, so the loop can never run more than ten times for one resume. If the limit is reached, the loop logs a step_limit event and reports the status honestly instead of pretending it finished. Step 5: The Sample Data The last file creates the fictional data. It writes one job posting and five short resumes into a folder named sample_data, and copies the resumes into the resumes folder where the agent is allowed to read. The five resumes were chosen to cover a spread: two strong matches, two partial matches, and one with no match at all. File: make_sample_data.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: writing fictional sample resumes and a job posting """make_sample_data.py - writes five fictional resumes and one fictional job posting.""" # Import the os module to work with file paths and folders import os # Import the shutil module for file operations (available for use even though not called directly below) import shutil # Create a text variable called HERE holding the folder that contains this file HERE = os.path.dirname(os.path.abspath(__file__)) # Create a text variable called POSTING holding the full fictional job posting text POSTING = """Job Title: Junior Data Pipeline Engineer Company: Examplecorp (a made-up company) Required skills: Python, SQL, Airflow, Docker Nice to have: Testing, Communication About the role: You will build and watch small data pipelines that move fictional sales numbers from one place to another every night. """ # Create a dictionary variable called RESUMES mapping each file name to that candidate's details RESUMES = { # A strong-match candidate: name, title, skills, and experience summary "resume_01.txt": ("Jordan Sampleperson", "Data Pipeline Engineer", "Python, SQL, Airflow, Docker, Testing", "Three years building nightly data jobs for a made-up shop."), # A partial-match candidate: name, title, skills, and experience summary "resume_02.txt": ("Riley Testcase", "Business Analyst", "Python, SQL, Excel, Tableau", "Two years making weekly sales charts for a made-up bakery."), # A partial-match candidate: name, title, skills, and experience summary "resume_03.txt": ("Casey Examplesmith", "Backend Developer", "Python, Docker, Kubernetes, Go", "Four years writing web services for a made-up delivery app."), # A no-match candidate: name, title, skills, and experience summary "resume_04.txt": ("Morgan Fakename", "Event Coordinator", "Photography, Event Planning, Customer Service", "Five years planning made-up weddings and festivals."), # A strong-match candidate: name, title, skills, and experience summary "resume_05.txt": ("Taylor Mockwell", "Data Engineer", "Python, SQL, Airflow, Docker, Spark", "Five years moving made-up warehouse data between systems."), } # Define the main function that writes the job posting and every resume file to disk def main(): # Create the sample_data folder if it does not already exist os.makedirs(os.path.join(HERE, "sample_data"), exist_ok=True) # Create the resumes folder if it does not already exist os.makedirs(os.path.join(HERE, "resumes"), exist_ok=True) # Open the job posting file for writing as UTF-8 text with open(os.path.join(HERE, "sample_data", "job_posting.txt"), "w", encoding="utf-8") as f: # Write the job posting text into the file f.write(POSTING) # Walk through every resume entry, unpacking its file name and details for filename, (name, title, skills, history) in RESUMES.items(): # Build the resume's full text from its name, title, skills, and experience text = "Name: %s\nTitle: %s\nSkills: %s\nExperience: %s\n" % (name, title, skills, history) # Write this same resume text into both the sample_data folder and the resumes folder for folder in ("sample_data", "resumes"): # Open the destination file for writing as UTF-8 text with open(os.path.join(HERE, folder, filename), "w", encoding="utf-8") as f: # Write the resume text into the file f.write(text) # Print a confirmation message once every file has been written print("Sample data written.") # If this file is run directly (not imported), start the main function if __name__ == "__main__": main() [END OF PYTHON CODE] All names here are obviously invented, such as Jordan Sampleperson and Riley Testcase. If you ever write more resumes of your own, keep them invented in exactly the same way, and never paste in a real person's details.
Advertisement
Read Aloud:
0:00 / 0:00