Homework — Three Pipeline Extensions
Lesson 06 · Building Your Own Agents · ~60 min
All three tasks below extend agent.py from the Lab 06 walk-through —
the graph-shaped pipeline with plan_research → fetch_sources
→ write_draft → critique_draft (with a revise loop) and
the SqliteSaver checkpointer keyed by thread_id. Keep that file
open as your base; each task adds one node (or wires one new edge) that exercises a
different part of the LangGraph surface area you met in the lab.
Before you start
Make sure python agent.py "Why was io_uring slow on kernel 6.13?" runs
end-to-end on a fresh database and prints all four node banners
(plan_research, fetch_sources, write_draft,
critique_draft) followed by a final draft. If it doesn’t, revisit
the lab guide before tackling the homework — debugging the extensions is much
easier on a working baseline.
Task 1 — Add a score_quality node after critique_draft
Right now the pipeline terminates as soon as critique_draft decides no
further revisions are needed. There’s no record of how good the final
draft actually is — just “the critic is satisfied.” Add a new node
that asks the LLM for a numeric quality rating and stamps it onto the state.
The node takes the final draft and returns a quality_score (int, 0–10)
based on a single LLM call (prompt: “rate this draft 0–10 on accuracy and
completeness, JSON only”). Reuse the _ask_json helper from the lab
so you get JSON-mode parsing for free.
Add quality_score: int to PipelineState and seed it with
"quality_score": 0 in the initial dict inside
run_pipeline_agent. Wire the edge so score_quality runs
only on the “done revising” branch — that is, change
route_after_critique to return "score" when finished
(instead of "finish"), point the mapping at the new node, and add a
final g.add_edge("score_quality", END). The revise branch
("revise" → "write_draft") is unchanged.
Print the score in the CLI footer so you can read it without inspecting the full state dict.
✓ Self-check: the final state includes a quality_score field with an integer in 0–10; running the agent on a question that triggers one or more revisions tends to produce a higher score than running the same question with revisions disabled (because the critic forced improvements); the score is visible in the CLI output.
Default the field in the initial state
LangGraph does not auto-default missing TypedDict
fields. If you only add quality_score: int to PipelineState
but forget "quality_score": 0 in the initial dict, any code that reads
app.get_state(config).values["quality_score"] before
score_quality has run will raise KeyError. Default
everything you add.
Task 2 — Persist PKB to disk across runs
Today the personal knowledge base (PKB) is a module-level dict seeded
with two entries at import time. Restart Python and any additions vanish — the
agent has no memory across runs. Refactor so the PKB lives on disk and can grow
over time.
On import, load PKB from pkb.json in the lesson directory,
falling back to the two seed entries (kernel-6.14-net-notes and
kernel-6.14-misc) if the file doesn’t exist yet. After any node
that mutates the PKB, save it back to disk. Use plain json.dump with
indent=2 so the file is grep-friendly.
Then add a learn_from_draft node that runs after
score_quality — but only when quality_score >= 7.
The node should mint a new PKB entry id (anything stable, e.g.
f"learned-{abs(hash(question)) % 100000}"), copy the question into the
title field, concatenate draft["intro"] and the bullets into
the text field, and call your save helper. Add a new conditional edge
from score_quality with a small router
(route_after_score) that returns "learn" when the score
crosses the threshold and "finish" otherwise. Both terminal branches
route to END.
Run the agent twice on different questions. After the second run, pkb.json
should contain at least three entries (two seed + one or more learned), and the
second run’s fetch_sources call should now find any learned
entries whose title matches the question keywords.
✓ Self-check: pkb.json exists after the first successful run with a high-enough score; running the agent a second time on a related topic surfaces the learned entry inside fetch_sources; deleting pkb.json and re-running the agent recreates it from the seed dict.
The checkpointer does not snapshot module globals
SqliteSaver persists everything inside PipelineState
— and nothing else. PKB is a module-level dict, not a
PipelineState field, so the checkpointer will not save or restore it.
If you want the PKB to be durable, you have to write the file yourself from inside
learn_from_draft. Don’t expect LangGraph to track side effects on
module globals.
Task 3 — Wire a second checkpointer for resume comparison
run_pipeline_agent already accepts a db_path argument —
the lab walk-through used a single default like pipeline.db. In this
task you’ll prove that two completely independent runs can coexist by giving
each its own sqlite file, and that resuming one never disturbs the other.
Run the agent on Question A with db_path="run-a.db". Kill it
mid-graph with Ctrl-C — aim for somewhere after
[node fetch_sources] but before [node write_draft] finishes.
Restart Python (so you know nothing is hiding in process memory). Now run the agent
on a different Question B with db_path="run-b.db" and let it
complete normally. Both .db files coexist on disk and never touched each
other.
Finally, resume run A. Build a graph, compile it with a
SqliteSaver pointing at run-a.db, and call
app.invoke(None, config={"configurable": {"thread_id": "pipeline-Question A"[:48]}})
(use the same thread_id rule your run_pipeline_agent uses).
It should pick up exactly where you killed it. Inspect both
app.get_state(...) snapshots and confirm the two databases hold
completely disjoint histories.
✓ Self-check: run-a.db and run-b.db both exist after the experiment; each holds the thread for its own question only (you can verify with sqlite3 run-a.db ".tables" followed by a quick SELECT); resuming run-a.db continues run A from the last completed node; resuming run-b.db would continue run B; neither resume corrupts or affects the other file.
One file = one writer
If you ever decide to share one db_path across concurrent processes
and rely on thread_id for isolation, remember that sqlite serialises
writers per file. The data stays correctly partitioned by thread_id,
but throughput collapses to one writer at a time. For real concurrency you swap
SqliteSaver for a Postgres-backed checkpointer — that’s a
drop-in change in LangGraph, but it’s out of scope for this lab.