← Table of Contents · Page 829 of 868

APPENDIX H: RESUME PORTFOLIO BUILDING CAPSTONE PROJECTS

Part 19 of 56

MICRO-STEPS Part A: Start the folder and reuse earlier work. 1. Create a new folder named project2_context_agent. 2. Copy gate.py from project1_tool_agent into project2_context_agent. 3. Copy tools.py from project1_tool_agent into project2_context_agent. 4. Open a terminal inside project2_context_agent. 5. Type python -c "import tools" and press Enter. You should see no output and no error, which means the reused files load correctly. Part B: Build the notes file helper. 6. Create a file named notes.py. 7. Type or paste the following code into notes.py and save it. File: notes.py [START OF PYTHON CODE] # Import the os module to work with file paths import os # Import the write permission check from the gate from gate import check_write # Create a text variable called NOTES_PATH, the fixed location of the running notes file NOTES_PATH = "output/notes.md" # Define a function that reads and returns the current contents of the notes file def read_notes(): full = check_write(NOTES_PATH) if not os.path.exists(full): return "" with open(full, "r", encoding="utf-8") as f: return f.read() # Define a function that appends one new note to the notes file def add_note(args): text = args.get("text", "").strip() if not text: return "ERROR: empty note" full = check_write(NOTES_PATH) os.makedirs(os.path.dirname(full), exist_ok=True) with open(full, "a", encoding="utf-8") as f: f.write(text + "\n") return "note saved" # Define a function that deletes the notes file entirely, clearing all notes def clear_notes(): full = check_write(NOTES_PATH) if os.path.exists(full): os.remove(full) [END OF PYTHON CODE] The notes file is a plain text file at output/notes.md that survives after the program stops. The function add_note adds one line to the end, read_notes returns everything, and clear_notes removes the file so a test can start fresh. Because notes live in the output folder, they pass through the same gate that guards every write. Part C: Build the context manager. 8. Create a file named context_manager.py. 9. Type or paste the following code into context_manager.py and save it. File: context_manager.py [START OF PYTHON CODE] # Define a function that roughly estimates how many tokens a piece of text uses def estimate_tokens(text): return max(1, len(text) // 4) # Define a function that adds up the estimated tokens across a whole list of messages def total_tokens(messages): return sum(estimate_tokens(m["content"]) for m in messages) # Define a function that folds the oldest messages into one short summary, keeping only the most recent ones def compact(body, keep_last): if len(body) <= keep_last: return body old = body[:len(body) - keep_last] recent = body[len(body) - keep_last:] used = [m["name"] for m in old if m["role"] == "tool"] text = "SUMMARY OF EARLIER STEPS: " + str(len(old)) + " messages were folded away. Tools used: " + ", ".join(used) + ". Every fact worth keeping is in the notes." return [{"role": "user", "content": text}] + recent # Define a function that builds the full context sent to the model: the task, the notes, and the recent history def build_context(task, notes_text, history, limit): # Build the fixed head of the context: the task itself, followed by the notes collected so far head = [ {"role": "user", "content": task}, {"role": "user", "content": "NOTES SO FAR:\n" + notes_text}, ] # Start the body as a copy of the full history, and assume no compaction is needed yet body = list(history) compacted = False # If the context is too large, compact it down to the last two messages if total_tokens(head + body) > limit: body = compact(body, 2) compacted = True # If it is still too large even after that, compact it down to just the last message if total_tokens(head + body) > limit: body = compact(body, 1) # Return the full context to send, the (possibly compacted) body, and whether compaction happened return head + body, body, compacted [END OF PYTHON CODE] Read the code from the top. The function estimate_tokens guesses the size of a text by dividing its length in characters by four, which is a rough rule of thumb and not an exact count. The function total_tokens adds up the guesses for a whole list of messages. The function compact keeps the most recent few messages and replaces everything older with one short summary message. The function build_context assembles what the model sees each turn: the task first, then the notes, then the recent history, compacting once or twice if the total is over the limit. It returns three things, which are the context to show the model, the shortened history to keep, and a true or false flag saying whether compaction happened. 10. In the terminal, type python -c "import context_manager as c; print(c.estimate_tokens('a' + 'b' + 'c' + 'd' + 'e' + 'f' + 'g' + 'h'))" and press Enter. You should see 2, because eight characters divided by four is two. Part D: Build the mock model for this project. 11. Create a file named model2.py. 12. Type or paste the following code into model2.py and save it. File: model2.py [START OF PYTHON CODE] # Import the os module to read environment variables import os # Define a function that builds a tool-call reply dictionary for a given tool and arguments def call(tool, args): return {"action": "tool", "tool": tool, "args": args} # Define a function that reads the list of target files and which ones are already done, out of the notes def read_notes_from_context(context): lines = [x for x in context[1]["content"].split("\n")[1:] if x.strip()] files = None done = set() for line in lines: if line.startswith("FILES:"): files = [x for x in line[6:].strip().split(",") if x] else: done.add(line.split(":")[0]) return files, done, lines # Define a function that scripts a stand-in model reply by reading the notes and deciding the next step def mock_model(context): # Read the known file list and the set of already-processed files out of the notes files, done, lines = read_notes_from_context(context) # Find the most recent tool result in the context, if any last = None for m in reversed(context): if m["role"] == "tool": last = m break # If the file list is not known yet, either record it from a fresh listing, or ask to list the workspace if files is None: if last is not None and last["name"] == "list_files": names = [x for x in last["content"].split("\n") if x.endswith(".txt")] return call("add_note", {"text": "FILES: " + ",".join(names)}) return call("list_files", {"path": "workspace"}) # If the summary has just been written, the task is finished if last is not None and last["name"] == "write_summary": return {"action": "finish", "text": "Answer written to output/answer.txt"} # If a note was just read, extract its fact line and record it as a note if last is not None and last["name"] == "read_note": fact = "NO FACT FOUND" for line in last["content"].splitlines(): if line.startswith("FACT:"): fact = line[5:].strip() break return call("add_note", {"text": last["args"]["path"].split("/")[-1] + ": " + fact}) # Work out which known files have not been read yet pending = [f for f in files if f not in done] # If there are still unread files, ask to read the next one if pending: return call("read_note", {"path": "workspace/" + pending[0]}) # Otherwise, every file has been read: build the final answer purely from the notes body = ["Answer built only from the notes, with the source file named on each line:"] for line in lines: if not line.startswith("FILES:"): body.append(line) return call("write_summary", {"path": "output/answer.txt", "text": "\n".join(body)}) # Define a placeholder function where a real hosted model's calling code should be filled in def ask_hosted(context): raise NotImplementedError("Fill in this function with your provider's code.") # Define a function that asks either the hosted model or the mock model, based on an environment setting def ask_model(context): if os.environ.get("HARNESS_MODEL") == "hosted": return ask_hosted(context) return mock_model(context) [END OF PYTHON CODE] This mock is built to make a point: it never trusts its own memory of the conversation. Each turn it reads the notes message to learn which files exist and which are already done, and it looks only at the last tool result to decide what to record next. Because all progress lives in the notes, folding away the old conversation costs it nothing, and a restart costs it nothing either. Part E: Build the run loop. 13. Create a file named research_agent.py. 14. Type or paste the following code into research_agent.py and save it. File: research_agent.py [START OF PYTHON CODE] # Import the os module to work with file paths import os # Import the json module to read and write structured data import json # Import the time module to measure and record when events happen import time # Import the sys module to read command-line arguments import sys # Import the custom error type and the shared root folder from the gate from gate import Denied, ROOT # Import the native tool registry from tools import TOOLS # Import the note-taking functions from notes import add_note, read_notes # Import the functions that build the trimmed context and total its tokens from context_manager import build_context, total_tokens # Import the function that asks either the hosted or mock model from model2 import ask_model # Create a text variable called LOG_PATH pointing to this project's run log file LOG_PATH = os.path.join(ROOT, "run_log2.jsonl") # Create a number variable called TOKEN_LIMIT, the context size that triggers compaction TOKEN_LIMIT = 2000 # Create a dictionary variable called ALL_TOOLS: a copy of the native tools, plus the add_note tool ALL_TOOLS = dict(TOOLS) ALL_TOOLS["add_note"] = add_note # Define a function that appends one structured event to the log file def log(event, data): record = {"time": round(time.time(), 2), "event": event, "data": data} with open(LOG_PATH, "a", encoding="utf-8") as f: f.write(json.dumps(record) + "\n") # Define a function that runs the full think/act/observe loop for one task, with notes and compaction def run(task, max_steps=40): # Start with an empty recent-history list history = [] # Record that the task has started log("start", {"task": task}) # Repeat the think/act/observe cycle up to the maximum number of steps for step in range(1, max_steps + 1): # Build the context to send the model: the task, the current notes, and the recent history context, history, compacted = build_context(task, read_notes(), history, TOKEN_LIMIT) # If compaction happened this step, record it along with the resulting token count if compacted: log("compacted", {"step": step, "tokens_after": total_tokens(context)}) # Ask the model what to do next reply = ask_model(context) # Record the model's reply in the log log("model_reply", {"step": step, "reply": reply}) # If the model says the task is finished, record that and return the final text if reply.get("action") == "finish": log("finish", {"step": step}) return reply.get("text", "") # Otherwise, read out the requested tool name and arguments name = reply.get("tool") args = reply.get("args", {}) # Add the model's tool-call reply to the recent history history.append({"role": "assistant", "content": json.dumps(reply)}) # Attempt to run the requested tool with the requested arguments try: # If the requested tool is not on the combined menu, refuse it if name not in ALL_TOOLS: raise Denied("unknown tool " + str(name)) result = ALL_TOOLS[name](args) log("tool_ok", {"step": step, "tool": name, "args": args}) # If the permission gate refused the request, record that as the result except Denied as problem: result = "BLOCKED: " + str(problem) log("blocked", {"step": step, "why": str(problem)}) # Add the tool result (including its original arguments) to the recent history history.append({"role": "tool", "name": str(name), "args": args, "content": result}) # If the step limit was reached without finishing, record that and return a stopped message log("step_limit", {"max_steps": max_steps}) return "STOPPED: step limit reached" # If this file is run directly (not imported), run the question given on the command line, or a default question if __name__ == "__main__": question = " ".join(sys.argv[1:]) or "What are the key facts in the garden documents?" print(run(question)) [END OF PYTHON CODE] Notice that at the start of every step the loop calls build_context with the current notes, so the notes are re-read from disk each time. The line that logs a compacted event tells you exactly when the desk got too crowded. TOKEN_LIMIT is set to 2000, which is small on purpose so that you can see compaction happen with just six short documents. Part F: Create the invented documents. 15. Create a file named make_docs.py. 16. Type or paste the following code into make_docs.py and save it. File: make_docs.py [START OF PYTHON CODE] # Import the os module to work with file paths and folders import os # Import the workspace folder path from the gate from gate import WORKSPACE # Create a list variable called FACTS holding one invented fact per document FACTS = [ "The Maple Grove garden opens on Tuesdays and Saturdays.", "Each plot rents for twelve dollars a season.", "The tool shed code is changed every month.", "Rainwater barrels are emptied before the first frost.", "Volunteers meet at nine in the morning on the first Saturday.", "The compost bins are turned every second week.", ] # Create a text variable called FILLER holding repeated padding text to make each document long FILLER = "The garden is a friendly place where invented neighbors share seeds and stories. " # Define the main function that writes one padded document per fact into the workspace folder def main(): os.makedirs(WORKSPACE, exist_ok=True) for i in range(len(FACTS)): path = os.path.join(WORKSPACE, "doc_" + str(i + 1) + ".txt") with open(path, "w", encoding="utf-8") as f: f.write("FACT: " + FACTS[i] + "\n") f.write(FILLER.strip() + "\n") for j in range(30): f.write(FILLER) print("made", len(FACTS), "documents in", WORKSPACE) # If this file is run directly (not imported), start the main function if __name__ == "__main__": main() [END OF PYTHON CODE] 17. In the terminal, type python make_docs.py and press Enter. You should see a line saying made 6 documents in, followed by the path of your workspace folder. 18. Open workspace/doc_1.txt in your editor. You should see a first line beginning with FACT: followed by a sentence about the garden, and then many repeated filler sentences. Part G: Run it. 19. Type python research_agent.py and press Enter. You should see Answer written to output/answer.txt. 20. Open output/notes.md. You should see one FILES line and then six lines, each beginning with a document name such as doc_1.txt followed by its fact. 21. Open output/answer.txt. You should see the six facts, each with its source file name, which is how a good research answer cites where each claim came from. 22. Open run_log2.jsonl and search for compacted. You should see at least one compacted event, with a tokens_after number below the limit of 2000. Part H: Prove the restart works. 23. Create a file named test_project2.py. 24. Type or paste the following code into test_project2.py and save it. File: test_project2.py [START OF PYTHON CODE] # Import the os module to work with file paths import os # Import the json module to read structured data import json # Import the document-generating module being tested import make_docs # Import the research agent module being tested import research_agent # Import the note-clearing and note-reading functions from notes import clear_notes, read_notes # Import the shared root folder from the gate from gate import ROOT # Define a function that reads the run log and returns a list of its parsed records def read_log(): with open(research_agent.LOG_PATH, "r", encoding="utf-8") as f: return [json.loads(line) for line in f] # Define a function that resets the documents, notes, log, and answer so each test starts clean def fresh(): make_docs.main() clear_notes() for name in (research_agent.LOG_PATH, os.path.join(ROOT, "output", "answer.txt")): if os.path.exists(name): os.remove(name) # Define a test that checks a full run finishes, triggers compaction, and records all six facts as notes def test_full_run_compacts(): fresh() answer = research_agent.run("What do the garden documents say?") assert answer.startswith("Answer written") events = [r["event"] for r in read_log()] assert "compacted" in events assert len(read_notes().splitlines()) == 7 print("PASS full run with compaction") # Define a test that checks a restarted run resumes from the notes instead of re-reading finished files def test_restart_resumes(): fresh() first = research_agent.run("What do the garden documents say?", max_steps=6) assert first.startswith("STOPPED") mark = len(read_log()) second = research_agent.run("What do the garden documents say?") assert second.startswith("Answer written") reads = [r for r in read_log()[mark:] if r["event"] == "tool_ok" and r["data"]["tool"] == "read_note"] assert len(reads) == 4, len(reads) print("PASS restart resumes from notes without rereading") # If this file is run directly (not imported), run both tests in order if __name__ == "__main__": test_full_run_compacts() test_restart_resumes() print("ALL TESTS PASSED") [END OF PYTHON CODE] 25. Type python test_project2.py and press Enter. You should see PASS full run with compaction, PASS restart resumes from notes without rereading, and ALL TESTS PASSED. 26. Open test_project2.py and find the line that stops the first run after six steps. Explain to yourself why exactly four documents remain for the second run, given that six steps cover the list step, the files note, two reads, and two document notes. 27. Change the TOKEN_LIMIT in research_agent.py from 2000 to 100000, run the first test again, and notice that it now fails because compaction never happens. 28. Change it back to 2000 and run the tests again to confirm they pass.
Advertisement
Read Aloud:
0:00 / 0:00