← 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