PROJECT TWO - ADDING CONTEXT MANAGEMENT

WHAT CHANGES IN THE CODE Stage 1 gave every resume its own fresh conversation. Stage 2 gives the whole batch one shared history list, so that old messages really do pile up. That is the problem we then solve. Copy your Stage 1 folder, keep all existing files, and add or replace the files below. Step 1: More Sample Data Compaction only matters when there is something to compact, so we grow the sample data from five resumes to twenty-five. The new script writes resumes 06 through 25, mixing skill sets so that some are strong, some partial, and some no match. Two of them, numbers 12 and 21, are padded with forty invented projects each, making them unusually long, about the size of two to three pages. They are there to stress the budget, the way a heavy suitcase tests whether a luggage scale works. File: make_more_data.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: adding twenty more fictional resumes """make_more_data.py - adds resumes 06 to 25 (all fictional). Run make_sample_data.py first.""" # 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 a list variable called FIRST holding a pool of invented first names FIRST = ["Avery", "Blake", "Cameron", "Dakota", "Emerson", "Finley", "Harper", "Jamie", "Kendall", "Logan"] # Create a list variable called LAST holding a pool of invented last names LAST = ["Sampleperson", "Testcase", "Mockwell", "Fakename", "Examplesmith", "Placeholder", "Demoson", "Dummyfield"] # Create a list variable called SKILL_SETS holding a pool of different skill combinations SKILL_SETS = [ "Python, SQL, Airflow, Docker", "Python, SQL, Excel", "Docker, Kubernetes, Go", "Accounting, Budgeting, Excel", "Python, Airflow, Testing, Communication", "Graphic Design, Photography", "SQL, Tableau, Communication", "Python, SQL, Airflow, Docker, Spark, Testing", "Teaching, Curriculum Design, Communication", "Python, Docker, Testing", ] # Create a list variable called TITLES holding a pool of invented job titles TITLES = ["Data Engineer", "Analyst", "Developer", "Coordinator", "Designer", "Teacher", "Accountant"] # Create a set variable called LONG_ONES naming the two resume numbers that should be padded very long LONG_ONES = {12, 21} # Define a function that builds the full fictional resume text for one resume number def build(number): # Pick a first and last name for this number, cycling through the name pools name = "%s %s" % (FIRST[number % len(FIRST)], LAST[number % len(LAST)]) # Pick a job title for this number, cycling through the title pool title = TITLES[number % len(TITLES)] # Pick a skill set for this number, cycling through the skill-set pool skills = SKILL_SETS[number % len(SKILL_SETS)] # Build the base resume text from the name, title, skills, and a made-up years-of-experience figure text = "Name: %s\nTitle: %s\nSkills: %s\nExperience: %d years at made-up companies.\n" % ( name, title, skills, 1 + number % 7) # If this resume number is one of the two that should be padded long if number in LONG_ONES: # Add forty invented project lines to make this resume unusually long for i in range(1, 41): text += ("Project %d: Invented project in which this fictional person improved a made-up " "process for a made-up team, wrote invented documentation, and presented " "imaginary results to a pretend manager.\n" % i) # Return the finished resume text return text # Define the main function that writes resumes 06 through 25 to both the sample_data and resumes folders def main(): # Walk through every resume number from 6 up to (but not including) 26 for number in range(6, 26): # Build this resume's file name, zero-padded to two digits filename = "resume_%02d.txt" % number # Write this same resume text into 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 destination file for writing as UTF-8 text with open(os.path.join(HERE, folder, filename), "w", encoding="utf-8") as f: # Write this resume's built text into the file f.write(build(number)) # Print a confirmation message once every resume has been written print("Resumes 06 to 25 written.") # If this file is run directly (not imported), start the main function if __name__ == "__main__": main() [END OF PYTHON CODE] Run make_sample_data.py first and then make_more_data.py. Check that the resumes folder now holds twenty-five files. Step 2: The Context Builder The context builder is a function that takes the full history and returns only what the model needs right now. It follows three rules. Rule one is that everything belonging to the current task is always kept, because the model must see the resume it is reviewing and the tool results so far. Rule two is that any earlier message flagged as important is always kept. Rule three is that of all the other earlier messages, only the most recent five are kept. In our lab the important messages are the one-line decisions, such as DECISION: resume_03.txt to partial, which the loop writes after each resume. Those lines are tiny, but they are exactly the facts that must never be lost. The same file also holds the helper functions for the notes file. The notes file is called progress_notes.md, and the loop writes one line to it after each finished resume. When the program starts, it reads the file to learn which resumes are already done. File: context_builder.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: choosing what the model sees and keeping the notes file """context_builder.py - chooses what the model sees on each call, and keeps the notes file.""" # Import the json module to read and write structured data import json # Import the os module to work with file paths import os [END OF PYTHON CODE] from permission_gate import ROOT [START OF PYTHON CODE] # Create a text variable called NOTES_PATH pointing to the progress notes file NOTES_PATH = os.path.join(ROOT, "progress_notes.md") # 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 builds the trimmed-down context the model actually sees on each call def build_context(history, keep_recent=5): # A docstring explaining what this function keeps: the task, flagged messages, and recent history """Return the current task in full, every message flagged important, and only the most recent few of everything else.""" # Find the index of the most recent message sent by the user start = max(i for i, m in enumerate(history) if m["role"] == "user") # Collect the indexes of every message that came before that task message earlier = list(range(start)) # Collect the indexes of the earlier messages that are flagged as important important = [i for i in earlier if history[i].get("important")] # Collect the indexes of the earlier messages that are not flagged as important others = [i for i in earlier if not history[i].get("important")] # Combine the important messages, the most recent few ordinary messages, and everything from the task onward keep = sorted(set(important + others[-keep_recent:] + list(range(start, len(history))))) # Return the actual messages at those kept indexes, in order return [history[i] for i in keep] # Define a function that reads the notes file and returns which resumes are already finished def load_done(): # A docstring explaining what this function returns """Read progress_notes.md and return the set of resume files already finished.""" # Start with an empty set of finished resumes done = set() # Only try to read the file if it actually exists yet if os.path.exists(NOTES_PATH): # Open the notes file for reading as UTF-8 text with open(NOTES_PATH, "r", encoding="utf-8") as f: # Walk through the file one line at a time for line in f: # A line starting with "DONE " names a finished resume file if line.startswith("DONE "): done.add(line.split()[1]) # Return the set of finished resume file names return done # Define a function that appends one finished-resume note to the notes file def append_note(resume_file, level): # Open the notes file in append mode as UTF-8 text with open(NOTES_PATH, "a", encoding="utf-8") as f: # Write a line recording that this resume is done, and at what match level f.write("DONE %s level=%s\n" % (resume_file, level)) [END OF PYTHON CODE] Step 3: Compaction Compaction means replacing a big pile of old messages with a short summary. Our rule is that once the stored history grows beyond a configurable token threshold, the oldest half is replaced by one summary message. The risky part is deciding what the summary keeps. A careless summary could drop a match decision, and then the agent would silently forget a result. So our compaction function copies every important message into the summary word for word, and it never cuts through the middle of the newest task. It is a good rule to remember: summarize the chatter, never the decisions. File: compaction.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: squeezing old history into a short summary """compaction.py - squeezes old history into a short summary.""" # Import the token-estimating function from the context builder from context_builder import estimate_tokens # Import the json module to measure the size of the conversation import json # Define a function that compacts the history when it grows past a token threshold def maybe_compact(history, threshold): # A docstring explaining the compaction rule: summarize chatter, never decisions """If the stored history is too big, replace its oldest half with one summary. Every message flagged important (the DECISION lines) is copied into the summary word for word, so no match decision can be lost.""" # If the history is still within the token budget, leave it unchanged if estimate_tokens(json.dumps(history)) <= threshold: return history # Otherwise, find the midpoint to split the history into an older half and a recent half cut = len(history) // 2 old, recent = history[:cut], history[cut:] # Nudge the cut point earlier until the recent half starts cleanly on a user message while recent and recent[0]["role"] != "user" and cut > 0: cut -= 1 old, recent = history[:cut], history[cut:] # Collect the content of every important message from the older half, word for word kept_facts = [m["content"] for m in old if m.get("important")] # Build one summary message standing in for the whole older half summary = { "role": "system", "important": True, "content": "SUMMARY of %d older messages. Facts kept: %s" % (len(old), "; ".join(kept_facts) or "none"), } # Return the summary message followed by the untouched recent half return [summary] + recent [END OF PYTHON CODE] The stored history is compacted, while the builder separately trims what is sent on each call. You may wonder why we need both. The builder protects each individual call, and compaction protects the memory of the harness itself, so that the stored history does not grow without limit over a very long batch. Step 4: The Updated Main Loop Now we wire everything together. The loop keeps one history for the whole batch, appends a decision line after every resume, writes a note, and compacts when needed. It also offers four ways to run, which are listed after the code. File: main_loop.py [START OF PYTHON CODE] # Module docstring stating the purpose of this file: Stage 2, a shared history with context building """main_loop.py - Stage 2: the loop now keeps one shared history and uses a context builder.""" # Import the filecmp module to compare files for equality import filecmp # 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 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 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") # 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, sharing the batch's history def review_one(resume_file, history, use_context=True, 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 # 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 # 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 # Ask the model what to do next reply = call_model(sent, describe_tools()) # 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 # Otherwise, add the model's tool-call reply to the shared history history.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) # 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)}) # Return this resume's full outcome return {"resume": resume_file, "status": status, "steps": steps, "tokens": tokens_used, "level": level} # 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 compare flag was given, run the batch twice and compare token usage with and without context if "--compare" in args: # Clear all state so the first run starts clean reset_state() # Run the batch without context trimming, keeping track of tokens used before = run_batch(use_context=False) # Build a path to save a copy of this run's reports for later comparison saved_copy = os.path.join(ROOT, "reports_before") # Remove any old saved copy, ignoring errors if it does not exist shutil.rmtree(saved_copy, ignore_errors=True) # Copy the reports folder into the saved-copy location shutil.copytree(os.path.join(ROOT, "reports"), saved_copy) # Clear all state again so the second run also starts clean reset_state() # Run the batch again, this time with context trimming, keeping track of tokens used after = run_batch(use_context=True) # Count how many reports are byte-for-byte identical between the two runs same = sum(1 for n in os.listdir(saved_copy) if filecmp.cmp(os.path.join(saved_copy, n), os.path.join(ROOT, "reports", n), shallow=False)) # Print how many tokens were used before and after, and the percentage saved print("BEFORE %d tokens, AFTER %d tokens, saved %.0f%%" % (before, after, 100 * (before - after) / before)) # Print how many of the reports matched exactly between the two runs print("Reports identical before and after: %d of %d" % (same, len(os.listdir(saved_copy)))) # Otherwise, run a single ordinary batch else: # If the fresh flag was given, clear all state before running if "--fresh" in args: reset_state() # 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] Four ways to run the agent. The command python main_loop.py runs the batch and skips any resume already listed in the notes. The command python main_loop.py --fresh clears the reports, logs, and notes first. The command python main_loop.py --fresh --stop-after 10 stops after ten resumes to simulate a crash. The command python main_loop.py --compare runs the whole batch twice, first without the context builder and then with it, and compares the cost and the reports.