Solutions — Three Pipeline-Agent Extensions

Lesson 06 · Building Your Own Agents · worked answers with commentary

Below are reference solutions for the three homework tasks. Read them after you’ve taken your own swing — each section also calls out the most common mistake we see, so comparing your draft to the “why this not that” commentary is where the learning actually happens.

How to use this page

The code blocks are minimal — they show only the lines that change relative to the lab’s agent.py. Drop them into your file, re-run, and compare the printed trace to the “expected behaviour” notes. If something doesn’t match, the “common mistake” box is the first place to look.

Solution to Task 1 — `score_quality` node on the finish branch

Add the new node, route it from the existing critique router, then terminate the graph at END. The diff versus the lab’s agent.py:

def score_quality(s: PipelineState) -> dict:
    prompt = (
        f"Question: {s['question']}\n\n"
        f"Draft:\n{json.dumps(s['draft'], indent=2)}\n\n"
        'Rate this draft 0-10 on accuracy and completeness. '
        'Return JSON: {"quality_score": <int 0-10>}'
    )
    out = _ask_json(prompt, max_tokens=80)
    score = int(out.get("quality_score", 0))
    print(f"[node score_quality]  quality_score={score}")
    return {"quality_score": score}


def route_after_critique(s: PipelineState) -> str:
    if s["revise"] and s["revisions"] < 2:
        return "revise"
    return "score"   # was "finish"


# In build_graph(), add:
g.add_node("score_quality", score_quality)
g.add_conditional_edges(
    "critique_draft",
    route_after_critique,
    {"revise": "write_draft", "score": "score_quality"},
)
g.add_edge("score_quality", END)

Also add quality_score: int to PipelineState and "quality_score": 0 to the initial dict in run_pipeline_agent.

Why this shape. The critique loop already decides “done or revise” — the natural place to insert a one-shot quality measurement is on the “done” branch, right before the graph terminates. Putting score_quality inside the revise loop would charge an extra LLM call per revision and bias the score (you’d be measuring intermediate drafts that the critic is about to reject anyway). One score, on the final draft, at the gate to END: cheap, honest, and easy to plot over runs.

Expected behaviour. Run the agent on the lab’s default question and the CLI footer now includes a line like [node score_quality] quality_score=8, and the final state dict carries quality_score: 8. Re-run with a deliberately weak prompt (e.g., ask for something the PKB doesn’t cover) and the score drops — usually 3-5 on the first draft, climbing to 6-8 after one revision. The metric is noisy across runs (LLM judges are not calibrated instruments) but the direction is reliable, which is the whole point of having it.

Common mistake — forgetting `quality_score: 0` in the initial dict

LangGraph doesn’t default missing TypedDict fields. score_quality returning {"quality_score": X} works fine — but if you call app.get_state(config).values["quality_score"] before score_quality has run (e.g., during a debug print after fetch_sources), you’ll get a KeyError. Default everything in the initial dict; treat the TypedDict as a schema, not as a Python class with defaults.

Solution to Task 2 — Persist `PKB` to disk and learn from good drafts

Two pieces: a load/save pair that turns the module-level PKB dict into a pkb.json-backed cache, and a new learn_from_draft node that conditionally appends an entry when the quality score clears 7:

PKB_PATH = "pkb.json"

def _load_pkb() -> dict[str, dict[str, str]]:
    if os.path.exists(PKB_PATH):
        with open(PKB_PATH) as f:
            return json.load(f)
    return {
        "kernel-6.14-net-notes": {"title": "...", "text": "..."},
        "kernel-6.14-misc": {"title": "...", "text": "..."},
    }

def _save_pkb(pkb: dict) -> None:
    with open(PKB_PATH, "w") as f:
        json.dump(pkb, f, indent=2)

PKB = _load_pkb()


def learn_from_draft(s: PipelineState) -> dict:
    note_id = f"learned-{abs(hash(s['question'])) % 100000}"
    PKB[note_id] = {
        "title": s["question"][:60],
        "text": s["draft"].get("intro", "") + " " +
                " ".join(s["draft"].get("bullets", [])),
    }
    _save_pkb(PKB)
    print(f"[node learn_from_draft] saved {note_id}")
    return {}


def route_after_score(s: PipelineState) -> str:
    return "learn" if s["quality_score"] >= 7 else "finish"


# Wire:
g.add_node("learn_from_draft", learn_from_draft)
g.add_conditional_edges(
    "score_quality",
    route_after_score,
    {"learn": "learn_from_draft", "finish": END},
)
g.add_edge("learn_from_draft", END)

Why the threshold matters. If you learn from every draft, the PKB fills with garbage and subsequent runs retrieve increasingly bad context — a closed-loop drift you won’t notice until your tenth run reads like a hall of mirrors. Gating on quality_score >= 7 uses the critic-and-judge stack you already have as a curation filter. The number is arbitrary; the principle — only persist what the agent itself judges good — is not.

Expected behaviour. First run on a fresh checkout: pkb.json doesn’t exist, _load_pkb returns the two seed entries, PKB starts with 2 keys, pkb.json is written with 2 keys after the run (assuming the quality gate passes). Second run on a different question: _load_pkb reads pkb.json back, PKB starts with 3 keys (2 seed + 1 learned from run 1), fetch_sources can now retrieve the learned entry if the new question overlaps with it, and after the second run pkb.json has 4 keys.

Common mistake — expecting LangGraph to track `PKB` for you

Don’t mutate PKB from inside the node and expect the checkpointer to snapshot it. PKB is a module-level dict, not part of PipelineState. The checkpointer captures PipelineState, nothing else — if the node side-effect (writing pkb.json) is what you want persistent, do that explicitly. The graph’s state is the graph’s state; module globals are your problem. This is a feature, not a bug: it forces you to think about what’s replayable on resume (state) and what’s a side-effect (disk writes, API calls). Conflate the two and your “resume from checkpoint” quietly stops being reproducible.

Solution to Task 3 — Two independent SQLite checkpoint files

No code change to run_pipeline_agent — the db_path argument already exists. The trick is in how you drive it from the outside and how you resume:

# Run A — partial, will be killed mid-graph
run_pipeline_agent("Question A", db_path="run-a.db")
# (kill the process here with Ctrl-C after `[node fetch_sources]`)

# Restart Python.
# Run B — fresh run on a different question.
run_pipeline_agent("Question B", db_path="run-b.db")

# Resume Run A separately.
g = build_graph()
memory = SqliteSaver.from_conn_string("run-a.db")
app = g.compile(checkpointer=memory)
config = {"configurable": {"thread_id": f"pipeline-{'Question A'[:40]}"}}
print(app.invoke(None, config=config))

Why this works. The two .db files are completely independent. SqliteSaver writes only to the file you give it, and the thread_id is scoped within that file. You can run as many parallel pipelines as you want — one per task, per user, per job — by giving each its own (db_path, thread_id) pair. In production you’d usually swap SqliteSaver for a Postgres-backed saver and encode (user_id, task_id) into the thread_id, but the topology is identical: one durable store, one thread ID per logical run.

Expected behaviour. After both runs, you have run-a.db and run-b.db sitting side by side. Open them with the sqlite3 CLI (sqlite3 run-a.db ".tables") and you’ll see the same LangGraph schema in each. Resuming run-a.db with app.invoke(None, ...) picks up exactly where Run A was killed (you should see the trace continue from [node critique_draft] onward, not restart at plan_outline). Resuming run-b.db independently does the same for Run B. Neither file knows the other exists.

Common mistake — one shared `db_path`, many concurrent writers

It’s tempting to point every concurrent run at one pipeline.db and rely on thread_id for isolation. The data stays isolated — sqlite reads rows by thread_id just fine — but sqlite’s single-writer lock will serialise writes across every thread on that file. At 5+ concurrent invocations you’ll start seeing database is locked errors or, worse, throughput collapse without an explicit error. Postgres lifts the single-writer limit; sqlite is fine for low-concurrency labs, single-developer dev loops, and CI. Match the saver to the concurrency target, not the other way round.

Closing tip. You’ve now built the canonical LangGraph use case — a graph-shaped pipeline agent with critique, conditional learning, and durable checkpoints — and extended it three different ways without re-architecting the core. Module 4 introduces the OpenAI Agents SDK, where you’ll meet the handoff/triage shape: the same problem (research and write) becomes “route to the right specialist.” The graph topology goes away; agent boundaries take its place. Comparing the two implementations side by side is where the choice between “LangGraph or SDK?” stops being a vibes question and starts being an answerable one. See you there.

Solutions — Three Pipeline-Agent Extensions

How to use this page

Solution to Task 1 — score_quality node on the finish branch

Common mistake — forgetting quality_score: 0 in the initial dict

Solution to Task 2 — Persist PKB to disk and learn from good drafts