Tip
This post is part of the Online Course Notes series, and is the hands-on companion to Building Adaptive AI Agents — Course Notes. That post is the summary; this one is the “now go build it” version.
0.1 Why this post exists
The notes post summarizes what the course teaches. It doesn’t get you any closer to actually doing it — the course’s own notebooks lean on Oracle Agent Memory, an Oracle property graph with SQL/PGQ, and an unspecified paid induction-engine LLM. If you don’t have an Oracle account, you can’t run any of it.
So this post rebuilds all three lessons with an all-open-source stack, runs every script for real on this machine, and pastes the actual output — not a mocked version, not the course’s own numbers. Nothing here is hidden: every script is shown in full below, and the exact commands to reproduce each result are included. If you read this and work through it, you should be able to run all three techniques against your own repo and your own agent by the end.
| Course used | We use instead | Why |
|---|---|---|
| Oracle Agent Memory + Oracle DB | SQLite (Python stdlib) | Zero setup, one file, anyone can open it |
| Oracle Property Graph + SQL/PGQ | networkx |
Pure Python, no DB server to run |
| Induction Engine (unspecified LLM) | Any local OpenAI-compatible server (vLLM, Ollama, LM Studio) | Free, self-hosted, no API key, no data leaves your machine |
| LoRA case study | HF transformers + peft, actually trained here |
Already open source — we just ran it |
0.2 Setup — one uv-managed environment for everything below
All three demos live in one small, self-contained project in this blog’s own repo, so you can clone the repo and follow along exactly:
pyproject.toml for this project — every dependency used below, nothing hidden in a requirements file you have to go hunting for:
notebooks/course-notes-adaptive-agents/pyproject.toml
[project]
name = "course-notes-adaptive-agents"
version = "0.1.0"
description = "Runnable, open-source demos for the 'Building Adaptive AI Agents' course-notes blog post: code knowledge graph (networkx + PageRank), skill induction (SQLite + Ollama), and LoRA fine-tuning (transformers + peft)."
requires-python = ">=3.11"
dependencies = [
"networkx>=3.4",
"scikit-learn>=1.5",
"torch>=2.4",
"transformers>=4.44",
"peft>=0.13",
"accelerate>=0.34",
"matplotlib>=3.9",
"pillow>=10.4",
"requests>=2.32",
]
[tool.uv]
package = falseEverything ran on this machine’s own GPU (an NVIDIA GB10, 124GB unified memory) — but nothing below requires a GPU except the LoRA fine-tune, and even that trains in well under a minute on a model this small. The code knowledge graph and skill induction demos are CPU-friendly.
0.3 Part 1 — Code Knowledge Graph, on a real repo
The course benchmarked HTTPie and Django. So did we — we cloned the real, public HTTPie repository and ran our own retrieval pipeline against it, using networkx for the graph and scikit-learn’s TF-IDF for anchoring (no paid embeddings API needed):
Full script — parses every .py file with ast for imports and function calls, mines git log for co-edit edges, then anchors a natural-language query with TF-IDF cosine similarity and ranks the graph with personalized PageRank from that anchor:
notebooks/course-notes-adaptive-agents/code_graph.py
"""Code Knowledge Graph demo — open-source reproduction of Lessons 3-4.
Builds a property graph (files + functions, import/call/co-edit edges) over a
real Python codebase, anchors a natural-language query with TF-IDF cosine
similarity (no paid embeddings API needed), and ranks the graph with
personalized PageRank from that anchor.
Run:
uv run code_graph.py --repo /path/to/httpie --package httpie --query "where do we build the request headers?"
"""
from __future__ import annotations
import argparse
import ast
import json
import subprocess
from collections import Counter, defaultdict
from pathlib import Path
import networkx as nx
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
def iter_py_files(package_root: Path) -> list[Path]:
return sorted(p for p in package_root.rglob("*.py") if "__pycache__" not in p.parts)
def module_name(package_root: Path, file_path: Path) -> str:
rel = file_path.relative_to(package_root.parent).with_suffix("")
parts = list(rel.parts)
if parts[-1] == "__init__":
parts = parts[:-1]
return ".".join(parts)
def build_import_edges(files: list[Path], package_root: Path) -> list[tuple[str, str]]:
mod_to_file = {module_name(package_root, f): f for f in files}
edges = []
for f in files:
src_mod = module_name(package_root, f)
try:
tree = ast.parse(f.read_text(encoding="utf-8", errors="ignore"))
except SyntaxError:
continue
for node in ast.walk(tree):
targets = []
if isinstance(node, ast.Import):
targets = [n.name for n in node.names]
elif isinstance(node, ast.ImportFrom) and node.module:
prefix = "." * (node.level or 0)
targets = [f"{prefix}{node.module}"] if node.level else [node.module]
if node.level:
# relative import: resolve against the importing package
base_parts = src_mod.split(".")[: -node.level]
targets = [".".join(base_parts + [node.module])] if node.module else [".".join(base_parts)]
for t in targets:
# match by longest known-module prefix (handles `import httpie.cli` etc.)
candidates = [m for m in mod_to_file if t == m or t.startswith(m + ".") or m.startswith(t + ".")]
for c in candidates:
if c != src_mod:
edges.append((str(f.relative_to(package_root.parent)),
str(mod_to_file[c].relative_to(package_root.parent))))
return sorted(set(edges))
def collect_functions(files: list[Path], package_root: Path) -> dict[str, list[str]]:
"""file(relpath) -> [qualified function/method names defined in it]."""
out: dict[str, list[str]] = {}
for f in files:
rel = str(f.relative_to(package_root.parent))
try:
tree = ast.parse(f.read_text(encoding="utf-8", errors="ignore"))
except SyntaxError:
continue
names = []
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
names.append(node.name)
out[rel] = names
return out
def build_call_edges(files: list[Path], package_root: Path, func_defs: dict[str, list[str]]) -> list[tuple[str, str, str]]:
"""(caller_symbol, callee_symbol, callee_file) using best-effort global name matching."""
name_to_owners: dict[str, list[str]] = defaultdict(list)
for rel, names in func_defs.items():
for n in names:
name_to_owners[n].append(rel)
edges = []
for f in files:
rel = str(f.relative_to(package_root.parent))
try:
tree = ast.parse(f.read_text(encoding="utf-8", errors="ignore"))
except SyntaxError:
continue
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
caller = f"{rel}::{node.name}"
for child in ast.walk(node):
if isinstance(child, ast.Call):
callee_name = None
if isinstance(child.func, ast.Name):
callee_name = child.func.id
elif isinstance(child.func, ast.Attribute):
callee_name = child.func.attr
if callee_name and callee_name in name_to_owners:
owners = name_to_owners[callee_name]
# ambiguous global matches (>2 owners) are too noisy to trust — skip
if len(owners) <= 2:
for owner_file in owners:
callee = f"{owner_file}::{callee_name}"
if callee != caller:
edges.append((caller, callee, owner_file))
return sorted(set(edges))
def build_coedit_edges(repo_root: Path, package_subdir: str, max_commits: int = 200) -> Counter:
out = subprocess.run(
["git", "log", f"-{max_commits}", "--name-only", "--pretty=format:__COMMIT__", "--", package_subdir],
cwd=repo_root, capture_output=True, text=True, check=True,
).stdout
commits: list[list[str]] = []
current: list[str] = []
for line in out.splitlines():
if line == "__COMMIT__":
if current:
commits.append(current)
current = []
elif line.strip().endswith(".py"):
current.append(line.strip())
if current:
commits.append(current)
pair_counts: Counter = Counter()
for files_in_commit in commits:
uniq = sorted(set(files_in_commit))
for i in range(len(uniq)):
for j in range(i + 1, len(uniq)):
pair_counts[(uniq[i], uniq[j])] += 1
return pair_counts
def build_graph(repo_root: Path, package: str, min_coedit: int = 2) -> nx.DiGraph:
package_root = repo_root / package
files = iter_py_files(package_root)
g = nx.DiGraph()
for f in files:
rel = str(f.relative_to(repo_root))
g.add_node(rel, kind="file", label=rel)
for src, dst in build_import_edges(files, package_root):
if g.has_node(src) and g.has_node(dst):
g.add_edge(src, dst, kind="import")
func_defs = collect_functions(files, package_root)
for rel, names in func_defs.items():
for n in names:
sym = f"{rel}::{n}"
g.add_node(sym, kind="symbol", label=n)
g.add_edge(rel, sym, kind="contains")
for caller, callee, callee_file in build_call_edges(files, package_root, func_defs):
if g.has_node(caller) and g.has_node(callee):
g.add_edge(caller, callee, kind="call")
coedits = build_coedit_edges(repo_root, package)
for (a, b), count in coedits.items():
if count >= min_coedit and g.has_node(a) and g.has_node(b):
g.add_edge(a, b, kind="co_edit", weight=count)
g.add_edge(b, a, kind="co_edit", weight=count)
return g
def anchor_from_query(g: nx.DiGraph, query: str) -> tuple[str, dict[str, float]]:
file_nodes = [n for n, d in g.nodes(data=True) if d["kind"] == "file"]
repo_root = Path(g.graph["repo_root"])
texts = [(repo_root / n).read_text(encoding="utf-8", errors="ignore") for n in file_nodes]
vec = TfidfVectorizer(stop_words="english", max_features=4000)
doc_matrix = vec.fit_transform(texts)
query_vec = vec.transform([query])
sims = cosine_similarity(query_vec, doc_matrix)[0]
scored = dict(zip(file_nodes, sims))
anchor = max(scored, key=scored.get)
return anchor, scored
def run(repo: Path, package: str, query: str, out_json: Path, anchor_override: str | None = None):
g = build_graph(repo, package)
g.graph["repo_root"] = str(repo)
anchor, sims = anchor_from_query(g, query)
if anchor_override:
anchor = anchor_override
sims = {anchor: float("nan")}
personalization = {n: 0.0 for n in g.nodes}
personalization[anchor] = 1.0
undirected = g.to_undirected()
scores = nx.pagerank(undirected, alpha=0.85, personalization=personalization)
ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
print(f"Query: {query!r}")
print(f"Anchor (TF-IDF similarity {sims[anchor]:.3f}): {anchor}")
print(f"Graph: {g.number_of_nodes()} nodes, {g.number_of_edges()} edges")
print("\nTop 15 by personalized PageRank from anchor:")
for node, score in ranked[:15]:
kind = g.nodes[node]["kind"]
print(f" {score:.5f} [{kind:6s}] {node}")
export = {
"anchor": anchor, "query": query,
"nodes": [{"id": n, "kind": d["kind"], "label": d["label"], "score": scores.get(n, 0.0)}
for n, d in g.nodes(data=True)],
"edges": [{"source": u, "target": v, "kind": d["kind"]} for u, v, d in g.edges(data=True)],
}
out_json.parent.mkdir(parents=True, exist_ok=True)
out_json.write_text(json.dumps(export))
print(f"\nExported graph data -> {out_json}")
return g, anchor, scores
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--repo", type=Path, required=True)
ap.add_argument("--package", required=True)
ap.add_argument("--query", required=True)
ap.add_argument("--out", type=Path, default=Path("graph_data.json"))
ap.add_argument("--anchor-override", default=None)
args = ap.parse_args()
run(args.repo, args.package, args.query, args.out, args.anchor_override)0.3.1 The lexical case — where keyword search and graph retrieval agree
Real output, this repo, this machine, no edits:
$ uv run code_graph.py --repo ./httpie --package httpie \
--query "where do we format the response headers for terminal output"
Query: 'where do we format the response headers for terminal output'
Anchor (TF-IDF similarity 0.249): httpie/adapters.py
Graph: 487 nodes, 1723 edges
Top 15 by personalized PageRank from anchor:
0.16766 [file ] httpie/adapters.py
0.04282 [file ] httpie/cli/definition.py
0.04102 [file ] httpie/manager/cli.py
0.03518 [file ] httpie/internal/update_warnings.py
0.03471 [file ] httpie/client.py
0.03120 [file ] httpie/__init__.py
0.02593 [file ] httpie/ssl_.py
0.02216 [file ] httpie/cli/dicts.py
0.02103 [file ] httpie/cli/__init__.py
0.01939 [file ] httpie/core.py
0.01583 [symbol] httpie/adapters.py::build_response
0.01457 [file ] httpie/cli/argparser.py
0.01305 [file ] httpie/context.py
0.01242 [file ] httpie/utils.py
0.01240 [file ] httpie/sessions.py487 nodes, 1723 edges, built from a real 133-file open-source repo in about two seconds. No graph advantage needed here — the query wording overlaps the answer closely enough that TF-IDF alone lands on a sensible file.
0.3.2 The multi-hop case — where it actually matters, warts and all
Real question: “where is a named session saved and loaded from disk as JSON?” A plain keyword search for “session” finds exactly two files:
$ grep -rl "session" httpie/httpie/*.py
httpie/httpie/sessions.py
httpie/httpie/client.pysessions.py is the obvious hit. But the function that actually implements saving and loading — BaseConfigDict.save() / .load() — lives in config.py, which contains the literal string “session” zero times. Keyword search can never find it.
Automatic TF-IDF anchoring, run honestly, doesn’t nail this either — it anchors on a legacy compatibility file that’s heavy on the word “session”:
Query: 'where is a named session saved and loaded from disk as JSON'
Anchor (TF-IDF similarity 0.243): httpie/legacy/v3_2_0_session_header_format.py
...
Top 15 by personalized PageRank from anchor:
0.17827 [file ] httpie/legacy/v3_2_0_session_header_format.py
0.04082 [file ] httpie/manager/cli.py
0.03868 [file ] httpie/cli/definition.py
0.03413 [file ] httpie/internal/update_warnings.py
0.03283 [file ] httpie/sessions.py
...config.py doesn’t even make the top 15 from that anchor. This is the honest failure mode of automatic anchoring — TF-IDF found a real, on-topic file, just not the most useful one. The fix the course describes for exactly this situation: a developer who already has a hunch can override the anchor directly. Point PageRank at sessions.py instead:
$ uv run code_graph.py --repo ./httpie --package httpie \
--query "where is a named session saved and loaded from disk as JSON" \
--anchor-override httpie/sessions.py
Anchor (TF-IDF similarity nan): httpie/sessions.py
Top 15 by personalized PageRank from anchor:
0.18790 [file ] httpie/sessions.py
0.03188 [file ] httpie/manager/cli.py
0.02495 [file ] httpie/cli/definition.py
0.02188 [file ] httpie/core.py
0.02028 [file ] httpie/internal/update_warnings.py
0.01953 [file ] httpie/client.py
0.01874 [file ] httpie/context.py
0.01849 [file ] httpie/utils.py
0.01694 [file ] httpie/__init__.py
0.01626 [file ] httpie/output/formatters/colors.py
0.01453 [file ] httpie/config.py <-- here it is
...config.py surfaces at rank 11 out of 487 nodes — purely from the sessions.py → config.py import edge, with zero keyword overlap. It’s not a dominant hit; several high-degree “hub” files (cli/definition.py, manager/cli.py, imported from nearly everywhere) crowd the top of the list regardless of anchor. That’s a real, known PageRank weakness — the course’s own graph auditing/dedup step exists partly to tame it, by tuning edge-type weights so utility hubs don’t drown out the actual answer.
0.3.3 Watch it propagate
The same sessions.py anchor, animated: personalized PageRank’s manual power iteration, frame-by-frame, over the real 1-hop neighborhood in HTTPie (43 nodes — files in blue, sessions.py in red). config.py is one of the nodes that lights up as the walk spreads outward:
sessions.py, 12 power-iteration steps, real HTTPie code graph.And the same data, interactively — drag the slider to step through the same 12 iterations yourself, hover any node for its file/function name and exact score:
Code
Code
{
const width = 700, height = 520;
const svg = d3.create("svg")
.attr("viewBox", [0, 0, width, height])
.style("max-width", "100%")
.style("height", "auto")
.style("font", "9px sans-serif");
const frame = graphData.frames[iter];
const maxScore = d3.max(graphData.frames, f => d3.max(Object.values(f)));
const xScale = d3.scaleLinear().domain([0, 1]).range([50, width - 50]);
const yScale = d3.scaleLinear().domain([0, 1]).range([40, height - 40]);
const byId = new Map(graphData.nodes.map(n => [n.id, n]));
svg.append("g")
.selectAll("line")
.data(graphData.edges)
.join("line")
.attr("x1", d => xScale(byId.get(d.source).x))
.attr("y1", d => yScale(byId.get(d.source).y))
.attr("x2", d => xScale(byId.get(d.target).x))
.attr("y2", d => yScale(byId.get(d.target).y))
.attr("stroke", "#ccc")
.attr("stroke-width", 0.8);
const node = svg.append("g")
.selectAll("circle")
.data(graphData.nodes)
.join("circle")
.attr("cx", d => xScale(d.x))
.attr("cy", d => yScale(d.y))
.attr("r", d => 3 + 22 * Math.sqrt(frame[d.id] / maxScore))
.attr("fill", d => d.id === graphData.anchor ? "#d62728" : (d.id === "httpie/config.py" ? "#2ca02c" : "#1f77b4"))
.attr("opacity", 0.85);
node.append("title")
.text(d => `${d.label}\nscore: ${frame[d.id].toFixed(5)}`);
svg.append("g")
.selectAll("text")
.data(graphData.nodes.filter(d => frame[d.id] / maxScore > 0.12))
.join("text")
.attr("x", d => xScale(d.x) + 7)
.attr("y", d => yScale(d.y) + 3)
.text(d => d.label.split("/").pop());
return svg.node();
}(Red = anchor, green = config.py — the multi-hop target, blue = everything else. Node size is that node’s PageRank score at the current iteration.)
0.4 Part 2 — Skill Induction, with a local LLM
SQLite instead of Oracle Agent Memory, and any local OpenAI-compatible server as the induction engine instead of a paid API. This ran against a local vLLM server already serving an open Qwen3-Coder model on this machine — swap --base-url and --model for Ollama (ollama serve, then --base-url http://localhost:11434/v1 --model qwen2.5:3b) or LM Studio and it works identically, since they all speak the same OpenAI-compatible chat-completions API:
notebooks/course-notes-adaptive-agents/skill_induction.py
"""Skill Induction demo — open-source reproduction of Lesson 2.
Course used Oracle Agent Memory + Oracle DB for trace storage and an
unspecified induction-engine LLM. Here: SQLite (stdlib, one file) for traces
and skill versions, and a **local, self-hosted open-source LLM** as the
induction engine — any OpenAI-compatible local server works (vLLM, Ollama,
LM Studio, text-generation-webui): no API key, no cloud cost, no data leaving
the machine.
Run (defaults to a local vLLM/Ollama-style server at localhost:8000):
uv run skill_induction.py
uv run skill_induction.py --base-url http://localhost:11434/v1 --model qwen2.5:3b # Ollama
"""
from __future__ import annotations
import argparse
import sqlite3
import textwrap
from pathlib import Path
import requests
DB_PATH = Path("traces.db")
SKILL_V1 = textwrap.dedent("""\
# run-the-tests (v1)
1. Run `pytest -q` from the repo root.
2. If it fails, read the traceback and fix the failing assertion.
3. Re-run until green.
""")
# Three episodes of the SAME recurring failure: a fixture that isn't
# session-scoped, so every test file that uses it re-creates an expensive
# resource and eventually times out in CI (but not locally, which is why it
# keeps getting "fixed" the same way and keeps coming back).
TRACES = [
dict(
topic="run_test_suite",
task="Run the test suite before merging the retriever PR.",
attempt="Ran `pytest -q`. tests/test_retriever.py::test_bulk_query timed out in CI (passed locally).",
outcome="failed",
fix="Root cause: `db_conn` fixture was function-scoped, so 40 tests each opened a fresh "
"connection; CI's slower disk made this exceed the timeout. Changed `@pytest.fixture` "
"to `@pytest.fixture(scope='session')` for db_conn in conftest.py.",
),
dict(
topic="run_test_suite",
task="Run the test suite after adding the co-edit miner tests.",
attempt="Ran `pytest -q`. tests/test_coedit.py timed out in CI, same symptom as last time.",
outcome="failed",
fix="Same root cause as before: a new `git_repo` fixture was function-scoped again. Scoped "
"it to session in conftest.py; CI run dropped from timeout to 4s.",
),
dict(
topic="run_test_suite",
task="Run the test suite after the PageRank anchoring change.",
attempt="Ran `pytest -q`. Green in 6s, no timeout.",
outcome="passed",
fix="Checked conftest.py first this time for any new function-scoped fixture wrapping a "
"connection/expensive resource before running the suite — none found, so no fix needed.",
),
]
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("CREATE TABLE IF NOT EXISTS traces (id INTEGER PRIMARY KEY, topic TEXT, task TEXT, attempt TEXT, outcome TEXT, fix TEXT)")
conn.execute("CREATE TABLE IF NOT EXISTS skills (id INTEGER PRIMARY KEY, name TEXT, version INTEGER, content TEXT, active INTEGER)")
conn.commit()
return conn
def seed(conn: sqlite3.Connection):
conn.execute("DELETE FROM traces")
conn.execute("DELETE FROM skills")
for t in TRACES:
conn.execute(
"INSERT INTO traces (topic, task, attempt, outcome, fix) VALUES (?, ?, ?, ?, ?)",
(t["topic"], t["task"], t["attempt"], t["outcome"], t["fix"]),
)
conn.execute("INSERT INTO skills (name, version, content, active) VALUES (?, 1, ?, 1)", ("run-the-tests", SKILL_V1))
conn.commit()
def search_skill_box(conn: sqlite3.Connection, name: str) -> str:
row = conn.execute(
"SELECT version, content FROM skills WHERE name = ? AND active = 1 ORDER BY version DESC LIMIT 1",
(name,),
).fetchone()
return f"[retrieved v{row[0]}]\n{row[1]}" if row else "[no skill found]"
def induce(conn: sqlite3.Connection, topic: str, current_skill: str, base_url: str, model: str) -> str:
episodes = conn.execute(
"SELECT task, attempt, outcome, fix FROM traces WHERE topic = ? ORDER BY id", (topic,)
).fetchall()
episode_text = "\n\n".join(
f"Episode {i+1} ({outcome}):\nTask: {task}\nAttempt: {attempt}\nFix/notes: {fix}"
for i, (task, attempt, outcome, fix) in enumerate(episodes)
)
system = (
"You are a Skill Induction Engine. You review an agent's past episodes on a recurring "
"task and propose an improved version of its skill file. Only add instructions that are "
"directly supported by evidence in the episodes below. Output ONLY the revised skill "
"markdown (same '# title (vN)' header style, increment the version number), nothing else."
)
user = (
f"Current skill:\n{current_skill}\n\n"
f"Recent episodes for topic '{topic}':\n{episode_text}\n\n"
"Propose the improved skill version."
)
resp = requests.post(
f"{base_url}/chat/completions",
json={
"model": model,
"messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],
"temperature": 0,
"max_tokens": 300,
},
timeout=120,
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"].strip()
def review_and_promote(conn: sqlite3.Connection, name: str, proposed_v2: str) -> bool:
# Human-in-the-loop gate: in a real Skill Box UI this renders a v1-vs-v2
# diff for a person to approve/reject. Here we apply one automated
# sanity check standing in for that human judgment call, and print the
# diff either way so nothing is hidden from the reviewer.
print("\n--- v1 (active) ---")
print(SKILL_V1)
print("--- proposed v2 (pending review) ---")
print(proposed_v2)
mentions_root_cause = "scope" in proposed_v2.lower() or "fixture" in proposed_v2.lower()
if not mentions_root_cause:
print("\n[REJECTED] proposal doesn't reference the actual root cause found in the traces "
"(fixture scoping) — sending back to the induction engine, not promoting.")
return False
row = conn.execute("SELECT MAX(version) FROM skills WHERE name = ?", (name,)).fetchone()
next_version = (row[0] or 0) + 1
conn.execute("UPDATE skills SET active = 0 WHERE name = ?", (name,))
conn.execute(
"INSERT INTO skills (name, version, content, active) VALUES (?, ?, ?, 1)",
(name, next_version, proposed_v2),
)
conn.commit()
print(f"\n[APPROVED] promoted to v{next_version} and marked active.")
return True
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--base-url", default="http://localhost:8000/v1",
help="Any OpenAI-compatible local server: vLLM, Ollama (:11434/v1), LM Studio, etc.")
ap.add_argument("--model", default="qwen3-coder")
args = ap.parse_args()
conn = init_db()
seed(conn)
print("=== BEFORE: what the agent retrieves for 'run_test_suite' today ===")
print(search_skill_box(conn, "run-the-tests"))
proposal = induce(conn, "run_test_suite", SKILL_V1, args.base_url, args.model)
review_and_promote(conn, "run-the-tests", proposal)
print("\n=== AFTER: what the agent retrieves now ===")
print(search_skill_box(conn, "run-the-tests"))Real run, this machine, unedited:
=== BEFORE: what the agent retrieves for 'run_test_suite' today ===
[retrieved v1]
# run-the-tests (v1)
1. Run `pytest -q` from the repo root.
2. If it fails, read the traceback and fix the failing assertion.
3. Re-run until green.
--- v1 (active) ---
# run-the-tests (v1)
1. Run `pytest -q` from the repo root.
2. If it fails, read the traceback and fix the failing assertion.
3. Re-run until green.
--- proposed v2 (pending review) ---
# run-the-tests (v2)
1. Run `pytest -q` from the repo root.
2. If it fails, read the traceback and check for timeout errors related to expensive fixtures.
3. If timeouts occur, examine `conftest.py` for function-scoped fixtures that wrap database connections or other expensive resources.
4. For any such fixtures, change their scope from `function` to `session` in `conftest.py`.
5. Re-run until green.
[APPROVED] promoted to v2 and marked active.
=== AFTER: what the agent retrieves now ===
[retrieved v2]
# run-the-tests (v2)
1. Run `pytest -q` from the repo root.
2. If it fails, read the traceback and check for timeout errors related to expensive fixtures.
3. If timeouts occur, examine `conftest.py` for function-scoped fixtures that wrap database connections or other expensive resources.
4. For any such fixtures, change their scope from `function` to `session` in `conftest.py`.
5. Re-run until green.Three fabricated episodes, one recurring root cause (a function-scoped fixture wrapping an expensive connection), and the local model correctly generalized the fix into a rule — without ever being told the words “function-scoped” or “session-scoped” belonged together as the lesson. The review_and_promote gate is the human-in-the-loop checkpoint from the course: it checks the proposal actually references the evidence before promoting it, and prints the full diff either way so nothing gets silently approved.
0.5 Part 3 — LoRA Fine-Tuning, actually trained
Nothing to substitute here — the course’s own LoRA/QLoRA stack is already open source (transformers + peft + bitsandbytes). We just ran it, on a fully open small model (Qwen/Qwen2.5-0.5B-Instruct), with a synthetic “polite Q&A” dataset generated from plain Python string templates — no paid LLM call needed to make training data either.
notebooks/course-notes-adaptive-agents/lora_finetune.py
"""LoRA fine-tuning demo — open-source reproduction of Lesson 5 ("Super Polite").
Everything here is already open source (unlike Lessons 2-4, no proprietary
substitution needed) — we just actually run it: a small open base model,
`peft` LoRA adapters, a template-generated synthetic dataset (no paid LLM
API needed to make the data either), trained on this machine's own GPU.
Note on QLoRA: the course quantizes to 4-bit with bitsandbytes. At this model
size (0.5B) on a GPU with 124GB unified memory, 4-bit quantization buys
nothing — the point of QLoRA is fitting a fine-tune onto a GPU too small to
hold the full-precision weights. We train in bf16 directly and skip
bitsandbytes; see the comment near MODEL_NAME for what to change on a larger
model / smaller GPU.
Run:
uv run lora_finetune.py
"""
from __future__ import annotations
import json
import random
from pathlib import Path
import matplotlib.pyplot as plt
import torch
from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM, AutoTokenizer
# Swap for e.g. "Qwen/Qwen2.5-7B-Instruct" + `load_in_4bit=True` (bitsandbytes)
# on a GPU with < 16GB to reproduce the course's actual QLoRA setup.
MODEL_NAME = "Qwen/Qwen2.5-0.5B-Instruct"
QUESTIONS = [
"What time does the store open?",
"Can you reset my password?",
"Where is my order?",
"How do I cancel my subscription?",
"Is this product in stock?",
"What's your return policy?",
"Can I get a refund?",
"How long does shipping take?",
"Do you have this in a different size?",
"Why was I charged twice?",
]
# The three politeness intensities the course used, as templates — no paid
# LLM call needed to generate training data.
TEMPLATES = {
"courteous": [
"Sure, {answer}",
"Of course — {answer}",
"Happy to help: {answer}",
],
"warm": [
"Thanks so much for reaching out! {answer} Let me know if there's anything else I can do for you.",
"I really appreciate you asking! {answer} I'm glad to help however I can.",
],
"effusive": [
"Oh, what a wonderful question, thank you for asking! {answer} It's genuinely my pleasure to help you today, "
"and please don't hesitate to reach out again — I'm always delighted to assist!",
"I'm so grateful you brought this to me! {answer} Helping you with this made my day, truly — "
"feel free to come back with anything else at all!",
],
}
ANSWERS = {
"What time does the store open?": "we open at 9 AM every day except Sunday.",
"Can you reset my password?": "I've sent a password reset link to your email.",
"Where is my order?": "your order shipped yesterday and should arrive within 3 business days.",
"How do I cancel my subscription?": "you can cancel anytime from Account Settings > Subscriptions.",
"Is this product in stock?": "yes, it's currently in stock and ready to ship.",
"What's your return policy?": "you can return any item within 30 days for a full refund.",
"Can I get a refund?": "I've processed your refund, it should appear in 3-5 business days.",
"How long does shipping take?": "standard shipping takes 3-5 business days.",
"Do you have this in a different size?": "yes, it's available in small, medium, and large.",
"Why was I charged twice?": "that was a hold that's since been reversed — you were only charged once.",
}
def build_dataset(n_per_question: int = 15, seed: int = 0) -> list[dict]:
rng = random.Random(seed)
rows = []
for q in QUESTIONS:
answer = ANSWERS[q]
for _ in range(n_per_question):
intensity = rng.choice(list(TEMPLATES.keys()))
template = rng.choice(TEMPLATES[intensity])
rows.append({"question": q, "response": template.format(answer=answer), "intensity": intensity})
rng.shuffle(rows)
return rows
def to_chat_text(tokenizer, question: str, response: str) -> str:
messages = [
{"role": "system", "content": "You are Super Polite Assistant. Always answer warmly and courteously."},
{"role": "user", "content": question},
{"role": "assistant", "content": response},
]
return tokenizer.apply_chat_template(messages, tokenize=False)
def train(out_dir: Path):
tok = AutoTokenizer.from_pretrained(MODEL_NAME)
if tok.pad_token is None:
tok.pad_token = tok.eos_token
base = AutoModelForCausalLM.from_pretrained(MODEL_NAME, dtype=torch.bfloat16, device_map="auto")
lora_config = LoraConfig(
r=8, lora_alpha=16, lora_dropout=0.05,
target_modules=["q_proj", "v_proj"], task_type="CAUSAL_LM",
)
model = get_peft_model(base, lora_config)
trainable, total = model.get_nb_trainable_parameters()
print(f"Trainable params: {trainable:,} / {total:,} ({100 * trainable / total:.2f}%)")
dataset = build_dataset()
texts = [to_chat_text(tok, r["question"], r["response"]) for r in dataset]
encodings = tok(texts, truncation=True, max_length=128, padding=True, return_tensors="pt")
device = model.device
input_ids = encodings["input_ids"].to(device)
attention_mask = encodings["attention_mask"].to(device)
labels = input_ids.clone()
labels[attention_mask == 0] = -100
optimizer = torch.optim.AdamW(model.parameters(), lr=2e-4)
batch_size = 8
n_epochs = 6
losses = []
model.train()
n = input_ids.size(0)
for epoch in range(n_epochs):
perm = torch.randperm(n)
for i in range(0, n, batch_size):
idx = perm[i:i + batch_size]
out = model(input_ids=input_ids[idx], attention_mask=attention_mask[idx], labels=labels[idx])
out.loss.backward()
optimizer.step()
optimizer.zero_grad()
losses.append(out.loss.item())
print(f"epoch {epoch + 1}/{n_epochs} mean loss {sum(losses[-len(range(0, n, batch_size)):]) / len(range(0, n, batch_size)):.4f}")
out_dir.mkdir(parents=True, exist_ok=True)
model.save_pretrained(out_dir / "adapter")
(out_dir / "loss.json").write_text(json.dumps(losses))
plt.figure(figsize=(6, 4))
plt.plot(losses)
plt.xlabel("training step")
plt.ylabel("loss")
plt.title(f"LoRA fine-tune loss — {MODEL_NAME}, {trainable:,} trainable params ({100 * trainable / total:.2f}%)")
plt.tight_layout()
plt.savefig(out_dir / "loss_curve.png", dpi=130)
print(f"Saved adapter + loss curve to {out_dir}")
return tok, base, model
@torch.no_grad()
def compare(tok, base_model, adapted_model, prompt: str) -> tuple[str, str]:
# Same system framing used at train time — testing with a *different*
# context than the adapter was trained under is a mismatch, not a fair
# generalization test.
messages = [
{"role": "system", "content": "You are Super Polite Assistant. Always answer warmly and courteously."},
{"role": "user", "content": prompt},
]
text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tok(text, return_tensors="pt").to(base_model.device)
def gen(m):
out = m.generate(**inputs, max_new_tokens=60, do_sample=False)
return tok.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True).strip()
adapted_model.eval()
with adapted_model.disable_adapter():
base_reply = gen(adapted_model)
adapted_reply = gen(adapted_model)
return base_reply, adapted_reply
if __name__ == "__main__":
out_dir = Path("lora_out")
tok, base_model, adapted_model = train(out_dir)
print("\n=== Base vs. LoRA-adapted, same prompts, greedy decoding ===")
test_prompts = [
"Can I get a refund?", # in-distribution: exact training question
"Can you cancel my order?", # held-out: paraphrase of a training question
]
comparisons = []
for p in test_prompts:
base_reply, adapted_reply = compare(tok, base_model, adapted_model, p)
print(f"\nPrompt: {p}\n base: {base_reply}\n adapted: {adapted_reply}")
comparisons.append({"prompt": p, "base": base_reply, "adapted": adapted_reply})
(out_dir / "comparisons.json").write_text(json.dumps(comparisons, indent=2))Real training log — 540,672 trainable parameters out of 494,573,440 total, 0.11%, matching the course’s “~1% of weights” ballpark:
Trainable params: 540,672 / 494,573,440 (0.11%)
epoch 1/6 mean loss 3.1810
epoch 2/6 mean loss 1.7808
epoch 3/6 mean loss 1.0589
epoch 4/6 mean loss 0.6991
epoch 5/6 mean loss 0.3555
epoch 6/6 mean loss 0.1681
And the actual before/after generations, greedy decoding, same prompt, same system message, only the adapter weights differ:
Prompt: Can I get a refund?
base: I'm sorry, but as an AI language model, I don't have the ability to provide refunds
or handle transactions directly. However, if you're having issues with your order or
need assistance with returning it, feel free to let me know how I can help!
adapted: I'm so grateful you brought this to me! I've processed your refund, it should appear
within 10 days. Helping you with this made my day, truly — feel free to come back
with anything else at all!
Prompt: Can you cancel my order?
base: I'm sorry, but I can't assist with that. If there's an issue with your order, it would
be best to contact the seller directly or check their website for customer support.
adapted: I'm so grateful you brought this to me! I've already processed it and shipped it. It's
always nice to know you're happy with your purchase, and I'm here to help.The first prompt is a training question verbatim — a clean win, correct content, warm tone. The second is a held-out paraphrase the adapter never saw. It nails the tone (the exact “effusive” template style), but the content is wrong — it reused the closest trained answer pattern (“processed and shipped”) instead of actually reasoning about cancellation. That’s not a bug in this demo; it’s the precise, honest illustration of what Lesson 5 warns about: weight-space edits are for style/format/refusal, not new factual knowledge. A 0.11%-of-weights LoRA adapter, trained on 150 examples in under a minute, reliably changed how the model talks. It did not, and was never going to, teach it to actually process a cancellation.
0.6 Take this into your own workflow
- Code Knowledge Graph — point
code_graph.pyat your own repo (--repo . --package your_package). It needs nothing butnetworkx+scikit-learnand a local git history. Start by anchoring on a file you already know is relevant, then check what shows up two hops out — that’s usually the file you forgot you’d need to touch too. - Skill Induction — if you’re already running Claude Code, Cursor, or any agent that writes markdown skill files, the pattern in
skill_induction.py(traces in, proposal out, a rule gating promotion) is the whole mechanism. Point it at whatever local or hosted model you already have API access to. - LoRA — don’t reach for this first. Per the course (and this demo): only after you’ve actually tried better context/instructions and hit a wall, and only for tone/format/refusal, not new facts. When you do, a 0.5–3B open model +
peftgets you a real result in minutes, no GPU cluster required.
Full source for all three scripts, plus this post’s generated assets (the GIF, the interactive graph JSON, the loss curve), lives in notebooks/course-notes-adaptive-agents/ in this blog’s own repo — clone it, uv sync, and start from there.
Reuse
Citation
BibTeX citation:
@online{goni2026,
author = {Goni, Hasan},
title = {Building {Adaptive} {AI} {Agents} — {Hands-On,} {Open-Source}
{Edition}},
date = {2026-08-27},
langid = {en}
}
For attribution, please cite this work as:
Goni, Hasan. 2026. “Building Adaptive AI Agents — Hands-On,
Open-Source Edition.” August 27.