Four months later, a four-agent LangChain loop ran for eleven days and cost 47,000 USD. Nobody noticed until the invoice arrived. The pipeline worked correctly in testing, and the agents were doing exactly what they were told. Same pattern.

This tutorial is about that missing instruction.

You'll build five small Python primitives that catch most agent loop failures before they ship:

• A spec writer that forces you to define done before the loop starts

• A circuit breaker that kills the loop when it exceeds hard limits

• A ledger that records every turn in an append-only SQLite audit trail

• An agent loop that ties all three together

• A review surface that forces human attestation before downstream systems receive anything

By the end you'll have a working repo you can drop into any agent project. The full code is at github.com/dannwaneri/production-safe-agent-loop.

Table of Contents

1. Why This Keeps Happening

2. Prerequisites

3. Phase 1: Define Done Before You Build

4. Phase 2: Enforce Done at Runtime

5. Phase 3: Record Everything

6. Phase 4: The Loop That Respects Its Boundaries

7. Phase 5: The Review Surface

8. Phase 6: A Real Example, SEO Audit Agent

9. Pluggable LLM Client

10. Running the Tests

11. What You've Built

12. Next Steps

Why This Keeps Happening

The math that got companies into trouble was simple. A chatbot costs roughly 0.04 USD per interaction. An orchestrated multi-agent workflow costs 1.20 USD. That's a 30x multiplier — and production benchmarks show it can reach 70x on complex tasks.

The problem isn't that agents are expensive. The problem is that most teams budgeted for chatbot costs and deployed agent architectures. Gartner found the token consumption gap between pilot chatbots and production agent workflows sits at 5-30x. The FinOps Foundation's 2026 State of FinOps report found 73% of enterprises say AI costs exceeded original projections.

The mechanism is straightforward once you see it. When an agent fails a task and retries, it doesn't start fresh. It re-reads the entire context window — every prior failed attempt — before trying again. Iteration one costs 100 tokens. Iteration two costs 200. Iteration ten costs thousands. You're paying for every failure, over and over, in milliseconds.

# This is the entire problem in three lines
while True:
    result = agent.run(task)
    # done when...?

That question mark is where the money goes.

The other thing making it worse: agents don't fail loudly. Traditional code hits an undefined state and crashes. An LLM hits ambiguity and tries to be helpful. It retries. It reformats the tool call. It spins up a verification agent. The verification agent finds something. A correction agent fires. Nobody defined what "correct" means. The loop looks beautiful on every dashboard you have — activity, tool calls, completion rate — while quietly burning through your budget.

Gartner predicts that 40% of agentic projects will be scrapped by 2027 due to economic failure. Most of that failure is preventable. Not with better models, but with exit conditions.

Prerequisites

• Python 3.10+

• An Anthropic API key (or any provider — more on that later)

• Basic familiarity with Python classes and SQLite

git clone https://github.com/dannwaneri/production-safe-agent-loop
cd production-safe-agent-loop
pip install -r requirements.txt
export ANTHROPIC_API_KEY=sk-...

Phase 1: Define Done Before You Build

The most expensive mistake in agent development isn't a bad model choice or a missing retry limit. It's starting the build before you can answer one question in one sentence:

What does done look like?

Most teams can't answer it. Not because they're careless, but because nothing forces them to before they open the terminal. The spec writer is that forcing function.

# spec_writer.py
from spec_writer import SpecWriter

spec = SpecWriter(db_path="spec.db").run()

When you call .run(), it won't return until you've answered three questions:

1. What does this do?

2. What does this NOT do?

3. What does done look like in one sentence?

The third question is the one that matters. It's also the hardest. "The agent audits the site" is not an answer. "The agent crawls the target URL, extracts all <title> and <meta description> tags, flags any missing or over-length, and stops" is an answer. One of those gives the circuit breaker something to enforce.

The spec stores to SQLite and returns a SpecResult dataclass with a session_id. That ID becomes the thread connecting your spec, your ledger rows, and your loop result. One session, traceable end to end.

@dataclass(frozen=True)
class SpecResult:
    what_it_does: str
    what_it_does_not: str
    done_looks_like: str
    session_id: str

frozen=True matters. The spec is a commitment, not a draft. Once it's written, the loop runs against it. No mid-run revisions.

For testing, SpecWriter accepts injectable input_fn and output_fn callables. No stdin monkey-patching required. See tests/test_spec_writer.py for working examples — the suite uses a small scripted_input helper that returns answers from a generator, and writes to a per-test SQLite file via pytest's tmp_path fixture. SQLite's :memory: isn't safe here, because SpecWriter opens a fresh connection per method and each :memory: connection is its own isolated database.

Phase 2: Enforce Done at Runtime

Defining the exit condition upstream is discipline. The circuit breaker is enforcement.

# circuit_breaker.py
from circuit_breaker import CircuitBreaker, CircuitBreakerError

breaker = CircuitBreaker(turn_limit=5, token_limit=15000)
breaker.check(turn_count, accumulated_tokens)  # raises on breach

Two ceilings. Both hard.

turn_limit caps how many times the loop can call the LLM. token_limit caps total token consumption across all turns. Either one tripping raises CircuitBreakerError immediately.

The boundary is strict: turn_count == turn_limit is allowed. turn_count == turn_limit + 1 trips. No grace periods or warnings. A hard stop forces a human checkpoint.

from dataclasses import dataclass


@dataclass
class CircuitBreakerError(Exception):
    reason: str          # "turn_ceiling" or "token_ceiling"
    turn_count: int
    accumulated_tokens: int

    def __post_init__(self) -> None:
        super().__init__(
            f"circuit breaker tripped: {self.reason} "
            f"(turn={self.turn_count}, tokens={self.accumulated_tokens})"
        )


class CircuitBreaker:
    def __init__(self, turn_limit: int = 5, token_limit: int = 15000) -> None:
        self.turn_limit = turn_limit
        self.token_limit = token_limit

    def check(self, turn_count: int, accumulated_tokens: int) -> None:
        if turn_count > self.turn_limit:
            self._trip("turn_ceiling", turn_count, accumulated_tokens)
        if accumulated_tokens > self.token_limit:
            self._trip("token_ceiling", turn_count, accumulated_tokens)

    def _trip(self, reason: str, turn_count: int, accumulated_tokens: int) -> None:
        print(
            "\n=== CIRCUIT BREAKER CHECKPOINT ===\n"
            f"reason         : {reason}\n"
            f"turn_count     : {turn_count} / limit {self.turn_limit}\n"
            f"tokens_used    : {accumulated_tokens} / limit {self.token_limit}\n"
            "action         : halt loop, surface to human reviewer\n"
            "=================================="
        )
        raise CircuitBreakerError(
            reason=reason,
            turn_count=turn_count,
            accumulated_tokens=accumulated_tokens,
        )

CircuitBreakerError is an exception, not a return code. That's intentional. A return code can be ignored. An uncaught exception can't. Silent breach is impossible. The human-readable checkpoint banner is printed to stdout by _trip() before the exception is raised, so even if a caller swallows the exception the operator still sees state.

The critical rule: call .check() before every LLM call, not after. Post-flight checking means you've already burned the tokens before you knew the limit was exceeded.

# Wrong — post-flight
result = client.messages.create(...)
breaker.check(turn_count, accumulated_tokens)  # too late

# Right — pre-flight
breaker.check(turn_count, accumulated_tokens)  # raises before any spend
result = client.messages.create(...)

The defaults (5 turns, 15,000 tokens) match a tight tutorial demo. Your production budget is different. Tune at instantiation:

# Production example — tighter token budget, more turns
breaker = CircuitBreaker(turn_limit=10, token_limit=50000)

Phase 3: Record Everything

The circuit breaker protects your bank account. The ledger protects your understanding of what happened.

Most teams log for debugging — they want to know what went wrong after it went wrong. The ledger has a different purpose. It's governance. Every row is proof that the loop stayed within its boundaries, or didn't, and exactly when.

# ledger.py
from ledger import Ledger

ledger = Ledger(db_path="ledger.db")
ledger.write(
    session_id=spec.session_id,
    turn_count=1,
    state_origin="llm",
    input_str=task,
    token_delta=523,
    execution_time_ms=1240,
    pass_fail=True,
)

One row per turn. Append-only, no updates, and no deletes. The immutability is the point: a ledger you can edit isn't a ledger, it's a notebook.

The schema:

CREATE TABLE IF NOT EXISTS ledger (
    id                 INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id         TEXT    NOT NULL,
    turn_count         INTEGER NOT NULL,
    state_origin       TEXT    NOT NULL,
    input_hash         TEXT    NOT NULL,
    token_delta        INTEGER NOT NULL,
    execution_time_ms  INTEGER NOT NULL,
    pass_fail          INTEGER NOT NULL,  -- 1=pass, 0=fail
    breach_reason      TEXT,              -- NULL unless circuit breaker fired
    created_at         TEXT    NOT NULL   -- ISO 8601, UTC
);
CREATE INDEX IF NOT EXISTS idx_ledger_session ON ledger(session_id);

The index makes get_session(session_id) — the primary read path — a constant-time lookup as the ledger grows.

Three decisions worth explaining:

1. input_hash not input_text. The raw input string never persists. Only its SHA-256 hash does. There are two benefits to this: identical inputs across runs are detectable, and PII never enters the audit trail.

2. pass_fail as INTEGER not BOOLEAN. SQLite has no boolean type. 1 and 0 are canonical. Clean Python ergonomics at the API edge, correct SQL types on disk.

3. created_at as datetime.now(timezone.utc).isoformat(). datetime.utcnow() was deprecated in Python 3.12. Timezone-aware timestamps avoid the footgun in any system that crosses timezones.

Retrieve by session:

rows = ledger.get_session(spec.session_id)
for row in rows:
    print(f"Turn {row.turn_count}: {'PASS' if row.pass_fail else 'FAIL'} "
          f"| {row.token_delta} tokens | {row.execution_time_ms}ms")

Phase 4: The Loop That Respects Its Boundaries

The agent loop wires the three primitives together. It's the only component that calls the LLM. Everything else is local.

# agent_loop.py
from agent_loop import AgentLoop

loop = AgentLoop(spec, breaker, ledger, client)
result = loop.run(task)
# LoopResult(success, turns, total_tokens, session_id, breach_reason)

The anatomy of a turn, in order:

1. circuit_breaker.check(turn_count, accumulated_tokens) — raises if either ceiling is exceeded

2. client.messages.create(...) — the actual LLM call

3. ledger.write(...) — one row, append-only

4. If stop_reason == "end_turn", return. Otherwise loop.

Pre-flight checking before every LLM call, with no exceptions.

def run(self, task: str) -> LoopResult:
    session_id = self.spec.session_id
    messages: list[dict] = [{"role": "user", "content": task}]
    turn = 0
    total_tokens = 0

    try:
        while True:
            turn += 1
            self.circuit_breaker.check(turn, total_tokens)

            started = time.perf_counter()
            response = self.client.messages.create(
                model=self.model,
                max_tokens=self.max_tokens,
                system=self._system_prompt(),
                messages=messages,
            )
            elapsed_ms = int((time.perf_counter() - started) * 1000)

            turn_tokens = (
                getattr(response.usage, "input_tokens", 0)
                + getattr(response.usage, "output_tokens", 0)
            )
            total_tokens += turn_tokens

            text = self._text_from(response)
            messages.append({"role": "assistant", "content": text})

            self.ledger.write(
                session_id=session_id,
                turn_count=turn,
                state_origin="llm",
                input_str=task,
                token_delta=turn_tokens,
                execution_time_ms=elapsed_ms,
                pass_fail=True,
            )

            if getattr(response, "stop_reason", "end_turn") == "end_turn":
                return LoopResult(
                    success=True,
                    turns=turn,
                    total_tokens=total_tokens,
                    session_id=session_id,
                )

            messages.append({"role": "user", "content": "continue"})

    except CircuitBreakerError as err:
        self.ledger.write(
            session_id=session_id,
            turn_count=turn,
            state_origin="circuit_breaker",
            input_str=task,
            token_delta=0,
            execution_time_ms=0,
            pass_fail=False,
            breach_reason=err.reason,
        )
        return LoopResult(
            success=False,
            turns=turn,
            total_tokens=total_tokens,
            session_id=session_id,
            breach_reason=err.reason,
        )

def _system_prompt(self) -> str:
    return (
        "You are an agent working on a tightly-scoped task.\n\n"
        f"What this does: {self.spec.what_it_does}\n"
        f"What this does NOT do: {self.spec.what_it_does_not}\n"
        f"Done looks like: {self.spec.done_looks_like}\n"
    )

@staticmethod
def _text_from(response) -> str:
    content = getattr(response, "content", None)
    if not content:
        return ""
    block = content[0]
    return getattr(block, "text", "") or ""

A few choices worth calling out in this body:

• The whole while True: is wrapped in one try/except CircuitBreakerError. The check happens at the top of every turn, so a breach is caught the same way whether it fires on turn 1 or turn 6.

• input_str=task on every ledger row — the original task, not the last assistant message. The input_hash column then groups rows that share the same starting input across the run.

• pass_fail=True for every LLM turn that returns, False only on breach. The pass/fail flag tracks whether the loop reached the row legitimately, not whether the model's output was good. Quality scoring is a separate concern.

• _system_prompt() uses all three spec fields, not just done_looks_like. The model needs the negative scope (what_it_does_not) at least as much as the positive scope.

• time.perf_counter() not time.time() — monotonic, immune to wall-clock adjustments mid-run.

LoopResult.session_id is inherited from spec.session_id. The ledger rows tie back to the spec without a join. One session ID, one traceable run, start to finish.

Phase 5: The Review Surface

The circuit breaker protects your bank account. The ledger records what happened. But neither tells you whether what happened matched what you promised.

That gap is where bad loops get approved. Polished output, green dashboard, missed commitment. A reviewer sees the artifact, decides it looks acceptable, and signs off. Nobody asked whether the original promise was kept.

The review surface closes that gap. It reads the session from SQLite, assembles the five-element frame, and forces a comparison before anything downstream receives the output.

from review_surface import ReviewSurface

rs = ReviewSurface(spec_db_path="spec.db", ledger_db_path="ledger.db")
print(rs.render(session_id))

Here's the five-element frame, in order:

1. Original promise — pulled from the spec table: what it does, what it doesn't do, what done looks like

2. Acceptance criteria — the done_looks_like field rendered as the explicit benchmark

3. Diff — first turn input vs final turn output, turns completed, total tokens, whether the loop breached

4. Evidence — all ledger rows for the session: turn-by-turn pass/fail, token delta, execution time

5. Unresolved assumptions — derived from breach rows and failed turns. Empty when clean.

When the reviewer is satisfied, they attest:

attestation = rs.attest(
    session_id=result.session_id,
    reviewer="daniel",
    notes="Output matches spec. Approved."
)
print(attestation.frame_hash)

.attest() writes to the attestations table in ledger.db. The frame_hash is a SHA-256 of the canonical frame data — deterministic across reviewers attesting the same session. It's the audit receipt. It proves the reviewer saw the exact frame as rendered, not a summary or a paraphrase.

Approval confirms the process ran. Attestation confirms the reviewer compared output to commitment. When the loop touches something regulated, those are different legal documents.

@dataclass(frozen=True)
class ReviewFrame:
    session_id: str
    original_promise: SpecResult
    acceptance_criteria: str
    diff: DiffResult
    evidence: tuple  # tuple[LedgerRow, ...]
    unresolved_assumptions: tuple  # tuple[str, ...]
    created_at: str

ReviewFrame is frozen for the same reason SpecResult is — the frame is evidence, not a draft. evidence and unresolved_assumptions are tuples because lists aren't hashable and frozen dataclasses need hashable fields.

The full end-to-end flow with the review surface lives in examples/review_example.py in the repo. Run it after any completed session: it renders the five-element frame, prompts for attestation, and writes the receipt if you approve.

The loop runs to you. Downstream systems get nothing until someone signs.

Phase 6: A Real Example — SEO Audit Agent

The pattern only makes sense against a real problem. This is the same agent architecture behind my seo-agent project.

SEO audits have a natural cadence: crawl, surface what's broken, fix, wait for reindex. Running the agent continuously doesn't change that cadence. It just burns tokens in the empty space between the moments that matter. A cron job wired to the loop is the honest architecture.

# examples/seo_audit_example.py
import requests
from bs4 import BeautifulSoup
import anthropic
from spec_writer import SpecWriter
from circuit_breaker import CircuitBreaker
from ledger import Ledger
from agent_loop import AgentLoop

def crawl_url(url: str) -> str:
    response = requests.get(url, timeout=10)
    soup = BeautifulSoup(response.text, "html.parser")
    title = soup.find("title")
    meta_desc = soup.find("meta", attrs={"name": "description"})
    h1_tags = soup.find_all("h1")
    return (
        f"URL: {url}\n"
        f"Title: {title.text if title else 'MISSING'}\n"
        f"Meta description: "
        f"{meta_desc['content'] if meta_desc else 'MISSING'}\n"
        f"H1 count: {len(h1_tags)}\n"
        f"H1 tags: {[h.text[:50] for h in h1_tags]}"
    )

def run_seo_audit(url: str) -> None:
    # Step 1: Define done before the loop starts
    spec = SpecWriter(db_path="spec.db").run()

    # Step 2: Initialise circuit breaker and ledger
    breaker = CircuitBreaker(turn_limit=5, token_limit=15000)
    ledger = Ledger(db_path="ledger.db")
    client = anthropic.Anthropic()

    # Step 3: Crawl the URL
    site_data = crawl_url(url)

    # Step 4: Run the loop
    # AgentLoop catches CircuitBreakerError internally and returns
    # LoopResult(success=False, breach_reason=...). Branch on the
    # result — do NOT wrap loop.run() in try/except CircuitBreakerError.
    loop = AgentLoop(spec, breaker, ledger, client)
    result = loop.run(
        f"Audit this page for SEO issues:\n\n{site_data}"
    )

    # Step 5: Print the ledger
    print(f"\nResult: {'SUCCESS' if result.success else 'BREACH'}")
    if not result.success:
        print(f"Breach reason: {result.breach_reason}")
    print(f"Turns: {result.turns} | Tokens: {result.total_tokens}")
    print("\nAudit trail:")
    for row in ledger.get_session(result.session_id):
        status = "PASS" if row.pass_fail else "FAIL"
        print(f"  Turn {row.turn_count}: {status} | "
              f"{row.token_delta} tokens | {row.execution_time_ms}ms")

if __name__ == "__main__":
    import sys
    run_seo_audit(sys.argv[1] if len(sys.argv) > 1 else "https://example.com")

Run it:

python examples/seo_audit_example.py https://yourdomain.com

The spec writer prompts you. The loop runs, the circuit breaker fires if the limits are exceeded, and the ledger records every turn. The output lands in front of you and you decide what to fix.

The loop runs to you, not into a void.

Pluggable LLM Client

The loop works with any client that satisfies the LLMClient protocol (Anthropic by default). Bring your own via a ~20-line adapter.

# agent_loop.py
from typing import Protocol, runtime_checkable


@runtime_checkable
class MessagesEndpoint(Protocol):
    def create(self, *, model: str, max_tokens: int,
               system: str, messages: list) -> object: ...


@runtime_checkable
class LLMClient(Protocol):
    messages: MessagesEndpoint

messages is an instance attribute (not a nested class) because that's how the real Anthropic SDK exposes it — anthropic.Anthropic().messages.create(...). Modeling it as a nested class would mean the real client wouldn't satisfy the Protocol. The @runtime_checkable decorator lets you sanity-check conformance with isinstance(client, LLMClient), and the repo's test suite uses exactly that assertion against the FakeClient test double.

Here's an OpenAI adapter example (This is illustrative. A production adapter would also map streaming, tool-use, and error shapes.):

# openai_adapter.py — illustrative pseudocode, not production-ready.
from openai import OpenAI as _OpenAI


class _MessagesAdapter:
    def __init__(self, client):
        self._client = client

    def create(self, *, model, max_tokens, system, messages):
        completion = self._client.chat.completions.create(
            model=model,
            max_tokens=max_tokens,
            messages=[{"role": "system", "content": system}] + messages,
        )
        # Reshape OpenAI's response into the Anthropic-shaped surface
        # AgentLoop reads: response.usage.{input,output}_tokens,
        # response.content[0].text, response.stop_reason.
        return _adapt_response(completion)


class OpenAIAdapter:
    def __init__(self, api_key: str):
        self._client = _OpenAI(api_key=api_key)
        self.messages = _MessagesAdapter(self._client)  # instance attr, not a nested class

The adapter pattern is worth teaching explicitly. Provider APIs don't share a shape. Anthropic puts system at the top level. OpenAI puts it inside the messages array. An adapter shim is ~20 lines and makes the loop provider-agnostic without rewriting anything. Note that self.messages is assigned in __init__ so it's a real attribute on each adapter instance, the same shape as the actual SDK.

Running the Tests

python -m pytest tests/

With coverage:

python -m coverage run --source=circuit_breaker,ledger,spec_writer,agent_loop,review_surface -m pytest tests/
python -m coverage report -m

80 tests, 100% coverage on all five core modules. The loop is exercised against a FakeClient test double defined inline in tests/test_agent_loop.py. It satisfies the LLMClient protocol via duck typing: messages is set to self, so client.messages.create(...) routes back to the same object and ships with scripted responses for each test scenario. Clone the repo and run pytest to see all 80 tests pass without touching the network or needing an API key.

circuit_breaker.py has 100% coverage — no untested paths. It's the financial safety component. Every path through it is exercised.

What You've Built

In this tutorial, you've build five small primitives, each independently usable.

The pattern: upstream discipline defines the boundaries. Downstream enforcement breaks the circuit. Neither trusts the model to police itself.

A loop that runs without an exit condition isn't autonomous. It's a billing event waiting to happen.

Define what done looks like before you start. That's the job, and always has been.

Next Steps

The repo is at github.com/dannwaneri/production-safe-agent-loop.

There are three natural extensions if you want to go further:

1. Graduation to Distributed Systems

The SQLite ledger works for isolated sequential loops. The moment you run multiple agents against shared state, you need serializable isolation — concurrent writes to flat JSON corrupt silently. The README documents the three tipping points where a flat ledger needs to graduate.

2. Cryptographic Signing

For compliance-scale systems where the auditor wasn't present when the loop ran, SQLite rows aren't enough. A database admin can run an UPDATE query. Ed25519 signing wraps each ledger row in a receipt that proves the log wasn't altered after execution. But that's a different tutorial.

Wiring a Cron Job

The honest architecture for the SEO audit agent isn't 24/7 autonomous operation. It's a cron job that runs on schedule, surfaces what's broken, and stops. 0 3 * * 2 python examples/seo_audit_example.py https://yourdomain.com is the whole thing. The loop runs to you, not into a void.

If you need this architecture built for your own stack (circuit breakers, audit trails, production-safe agent loops), I do freelance work. dannwaneri.com/ai-agents/

It doesn't learn.

You ingest 500 documents. You ask a question. The system retrieves the three most similar chunks and hands them to the LLM. Repeat for the next query.

The system knows exactly as much as it did on day one. It's a library that never builds a card catalog, never cross-references its own shelves, never notices that three of its books are saying contradictory things.

That's what I set out to fix with a knowledge reflection layer. After every ingest, the system finds semantically related documents already in the index and asks an LLM to synthesise what's new, how it connects, and what gap remains. That synthesis gets embedded, stored, and boosted in search results.

The knowledge base gets smarter as you add more documents — not just bigger.

This tutorial shows you exactly how to build it.

Table of Contents

1. What You Will Build

2. Prerequisites

3. How to Set Up the Base System

4. Why Standard RAG Has a Memory Problem

5. Step 1: Schema Update

6. Step 2: The Reflection Engine

7. Step 3: Consolidation

8. Step 4: Wire It Into Your Ingest Handler

9. Step 5: Boost Reflections in Search

10. Step 6: Filtering by doc_type

11. What Changes After You Build This

12. Deploying

13. What to Build Next

What You Will Build

In this tutorial, you'll build a post-ingest reflection pipeline that:

1. Fires automatically after every document ingest

2. Finds the most semantically related documents already in the index

3. Asks Kimi K2.5 to synthesise a three-sentence insight linking the new document to existing knowledge

4. Stores that reflection with doc_type=reflection and a 1.5× ranking boost in search results

5. Consolidates reflections into summaries every three ingests

By the end, searching your knowledge base will surface both raw document chunks and reflection artifacts the system wrote on ingest.

Prerequisites

You will need:

• A Cloudflare account — free tier works

• Node.js v18+ and Wrangler CLI installed (npm install -g wrangler)

• Basic TypeScript familiarity

No external API keys. Everything runs on Cloudflare's infrastructure.

How to Set Up the Base System

If you have already built the RAG system from my freeCodeCamp handbook, skip this section — your system is ready for the reflection layer.

If you're starting fresh, this section gets you to a working base in about 15 minutes.

Scaffold the Project

npm create cloudflare@latest rag-reflection-system
cd rag-reflection-system

Choose: Hello World example → TypeScript → No deploy yet.

Create the Vectorize Index and D1 Database

npx wrangler vectorize create rag-index --dimensions=384 --metric=cosine
npx wrangler d1 create rag-db

Configure wrangler.toml

name = "rag-reflection-system"
main = "src/index.ts"
compatibility_date = "2026-01-01"

[[vectorize]]
binding = "VECTORIZE"
index_name = "rag-index"

[[d1_databases]]
binding = "DB"
database_name = "rag-db"
database_id = "YOUR_DB_ID"

[ai]
binding = "AI"

Create the documents Table

-- migrations/001_init.sql
CREATE TABLE IF NOT EXISTS documents (
  id TEXT PRIMARY KEY,
  content TEXT NOT NULL,
  source TEXT,
  date_created TEXT DEFAULT (datetime('now'))
);

npx wrangler d1 execute rag-db --remote --file=./migrations/001_init.sql

Add the ingest and search endpoints

Replace src/index.ts with this minimal working system:

export interface Env {
  VECTORIZE: VectorizeIndex;
  DB: D1Database;
  AI: Ai;
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === '/ingest' && request.method === 'POST') {
      const { id, content, source } = await request.json() as any;

      const embResult = await env.AI.run('@cf/baai/bge-small-en-v1.5', {
        text: [content.slice(0, 512)],
      }) as any;
      const vector = embResult.data[0];

      await env.VECTORIZE.upsert([{
        id,
        values: vector,
        metadata: { content: content.slice(0, 1000), source, doc_type: 'raw' },
      }]);

      await env.DB.prepare(
        'INSERT OR REPLACE INTO documents (id, content, source) VALUES (?, ?, ?)'
      ).bind(id, content, source ?? '').run();

      return Response.json({ success: true, id });
    }

    if (url.pathname === '/search' && request.method === 'POST') {
      const { query } = await request.json() as any;

      const embResult = await env.AI.run('@cf/baai/bge-small-en-v1.5', {
        text: [query],
      }) as any;
      const vector = embResult.data[0];

      const results = await env.VECTORIZE.query(vector, {
        topK: 5,
        returnMetadata: 'all',
      });

      const context = results.matches
        .map(m => m.metadata?.content as string)
        .filter(Boolean)
        .join('\n\n');

      const answer = await env.AI.run('@cf/moonshotai/kimi-k2.5', {
        messages: [
          { role: 'system', content: 'Answer using only the context provided.' },
          { role: 'user', content: `Context:\n\({context}\n\nQuestion: \){query}` },
        ],
        max_tokens: 256,
      }) as any;

      return Response.json({ answer: answer.response, sources: results.matches.map(m => m.id) });
    }

    return new Response('RAG system running', { status: 200 });
  },
};

Deploy and Verify

npx wrangler deploy

Test it:

# Ingest a document
curl -X POST https://your-worker.workers.dev/ingest \
  -H "Content-Type: application/json" \
  -d '{"id": "doc-001", "content": "Cursor pagination beats offset pagination for live-updating datasets because offset becomes unreliable when rows are inserted or deleted during pagination."}'

# Search
curl -X POST https://your-worker.workers.dev/search \
  -H "Content-Type: application/json" \
  -d '{"query": "what pagination approach should I use?"}'

If you get a grounded answer back, the base system is working. The next sections add the reflection layer on top of this foundation.

Why Standard RAG Has a Memory Problem

Standard RAG retrieval is stateless. Every query goes in cold. The system has no memory of what it found before, no synthesis of what it learned across documents, and no growing understanding of what questions remain unanswered.

Imagine you've ingested 200 documents about your product. Twelve of them touch on a pricing decision made last year. No single one has the full picture — it's distributed across quarterly reports, meeting notes, an internal Slack export, a few Notion pages.

A user asks: "Why did we change our pricing structure?"

Standard RAG retrieves the three most similar chunks. If those three chunks collectively have the answer, great. If they don't — if the real answer requires synthesising across those twelve documents — the system has no mechanism for that. It returns fragments. The LLM makes its best guess.

The reflection layer addresses this directly. When the twelfth pricing document gets ingested, the system finds the eleven related documents, synthesises what connects them, and stores that synthesis as a retrievable artifact. The answer to "why did we change our pricing structure" exists in the index before anyone asks the question.

Not smarter retrieval — smarter indexing.

Step 1: Schema Update

The reflection layer needs two new fields in your D1 documents table. Run this migration:

-- migrations/003_add_reflection_fields.sql
ALTER TABLE documents ADD COLUMN doc_type TEXT DEFAULT 'raw';
ALTER TABLE documents ADD COLUMN reflection_score REAL DEFAULT 0;
ALTER TABLE documents ADD COLUMN parent_reflection_id TEXT;

Apply it:

wrangler d1 execute mcp-knowledge-db --remote --file=./migrations/003_add_reflection_fields.sql

doc_type distinguishes raw documents (raw), single-document reflections (reflection), and consolidated multi-reflection summaries (summary). You'll use this field to filter — exposing only reflections to users who want the distilled view, or excluding them for users who want raw source chunks.

Step 2: The Reflection Engine

Create src/engines/reflection.ts. This is the core of the layer.

import { Env } from '../types/env';
import { resolveEmbeddingModel, resolveReflectionModel } from '../config/models';

const REFLECTION_BOOST = 1.5;
const CONSOLIDATION_THRESHOLD = 3; // consolidate every N new reflections

export async function reflect(
  newDocId: string,
  newDocContent: string,
  env: Env
): Promise<void> {
  // 1. Find semantically related documents already in the index
  const embModel = resolveEmbeddingModel(env.EMBEDDING_MODEL);
  const embResult = await env.AI.run(embModel.id as any, {
    text: [newDocContent.slice(0, 512)],
  });
  const queryVector = (embResult as any).data?.[0];
  if (!queryVector) return;

  const related = await env.VECTORIZE.query(queryVector, {
    topK: 5,
    filter: { doc_type: { $eq: 'raw' } },
    returnMetadata: 'all',
  });

  const relatedDocs = (related.matches ?? []).filter(
    m => m.id !== newDocId && (m.score ?? 0) > 0.65
  );

  if (relatedDocs.length === 0) return; // nothing related yet — skip

  // 2. Build synthesis prompt
  const relatedSummaries = relatedDocs
    .slice(0, 3)
    .map((m, i) => `Document \({i + 1}: \){String(m.metadata?.content ?? '').slice(0, 300)}`)
    .join('\n\n');

  const prompt = `You are synthesising knowledge across documents in a knowledge base.

New document:
${newDocContent.slice(0, 600)}

Related existing documents:
${relatedSummaries}

Write exactly three sentences:
1. What the new document adds that the existing documents don't already cover
2. How the new document connects to or extends the existing documents
3. What gap or question remains unanswered across all these documents

Be specific. Reference actual content. Do not summarise — synthesise.`;

  // 3. Call the reflection model
  const reflModel = resolveReflectionModel(env.REFLECTION_MODEL);
  const llmResp = await env.AI.run(reflModel.id as any, {
    messages: [{ role: 'user', content: prompt }],
    max_tokens: 180,
  });

  const reflectionText = (llmResp as any)?.response?.trim();
  if (!reflectionText || reflectionText.length < 40) return;

  // 4. Embed and store the reflection
  const reflEmbResult = await env.AI.run(embModel.id as any, {
    text: [reflectionText],
  });
  const reflVector = (reflEmbResult as any).data?.[0];
  if (!reflVector) return;

  const reflectionId = `refl_\({newDocId}_\){Date.now()}`;

  await env.VECTORIZE.upsert([
    {
      id: reflectionId,
      values: reflVector,
      metadata: {
        content: reflectionText,
        doc_type: 'reflection',
        parent_id: newDocId,
        reflection_score: REFLECTION_BOOST,
        source_doc_ids: relatedDocs.map(m => m.id).join(','),
        date_created: new Date().toISOString(),
      },
    },
  ]);

  await env.DB.prepare(
    `INSERT INTO documents
     (id, content, doc_type, reflection_score, parent_id, date_created)
     VALUES (?, ?, 'reflection', ?, ?, ?)`
  )
    .bind(reflectionId, reflectionText, REFLECTION_BOOST, newDocId, new Date().toISOString())
    .run();

  // 5. Check if consolidation is due
  const recentCount = await env.DB
    .prepare(`SELECT COUNT(*) as cnt FROM documents WHERE doc_type = 'reflection' AND date_created > datetime('now', '-1 hour')`)
    .first<{ cnt: number }>();

  if ((recentCount?.cnt ?? 0) >= CONSOLIDATION_THRESHOLD) {
    await consolidate(env);
  }
}

Two things worth noting here.

First, the semantic threshold (score > 0.65) matters. Too low and you're synthesising unrelated documents. Too high and you're rarely finding connections. 0.65 works well with bge-small. You can bump it to 0.72 with qwen3-0.6b (1024d) where scores cluster higher.

The prompt structure is deliberate. Three sentences, each doing a specific job: what's new, how it connects, what remains. This keeps reflections useful for retrieval. A freeform synthesis prompt produces beautiful prose that doesn't retrieve well. This structure produces retrievable artifacts.

Step 3: Consolidation

As reflections accumulate, they need their own synthesis layer — otherwise you're adding noise at a higher abstraction level.

Add this to src/engines/reflection.ts:

export async function consolidate(env: Env): Promise<void> {
  // Fetch recent reflections not yet consolidated
  const recent = await env.DB
    .prepare(
      `SELECT id, content FROM documents
       WHERE doc_type = 'reflection'
       AND id NOT IN (
         SELECT DISTINCT parent_id FROM documents
         WHERE doc_type = 'summary' AND parent_id IS NOT NULL
       )
       ORDER BY date_created DESC
       LIMIT 6`
    )
    .all<{ id: string; content: string }>();

  if (!recent.results || recent.results.length < CONSOLIDATION_THRESHOLD) return;

  const reflectionTexts = recent.results.map((r, i) => `Reflection \({i + 1}: \){r.content}`).join('\n\n');

  const prompt = `You are consolidating multiple knowledge reflections into a single compressed insight.

${reflectionTexts}

Write two to three sentences that capture the most important cross-cutting pattern or tension across these reflections. What does the knowledge base now understand that it didn't before these documents were added? What's the most important open question?

Be precise. No preamble.`;

  const reflModel = resolveReflectionModel(env.REFLECTION_MODEL);
  const llmResp = await env.AI.run(reflModel.id as any, {
    messages: [{ role: 'user', content: prompt }],
    max_tokens: 320,
  });

  const summaryText = (llmResp as any)?.response?.trim();
  if (!summaryText || summaryText.length < 40) return;

  const embModel = resolveEmbeddingModel(env.EMBEDDING_MODEL);
  const embResult = await env.AI.run(embModel.id as any, { text: [summaryText] });
  const summaryVector = (embResult as any).data?.[0];
  if (!summaryVector) return;

  const summaryId = `summary_${Date.now()}`;

  await env.VECTORIZE.upsert([
    {
      id: summaryId,
      values: summaryVector,
      metadata: {
        content: summaryText,
        doc_type: 'summary',
        reflection_score: REFLECTION_BOOST * 1.2,
        source_reflection_ids: recent.results.map(r => r.id).join(','),
        date_created: new Date().toISOString(),
      },
    },
  ]);

  await env.DB.prepare(
    `INSERT INTO documents (id, content, doc_type, reflection_score, date_created)
     VALUES (?, ?, 'summary', ?, ?)`
  )
    .bind(summaryId, summaryText, REFLECTION_BOOST * 1.2, new Date().toISOString())
    .run();
}

Summaries get a 1.2× multiplier on top of the base reflection boost. In search results, a summary synthesising twelve related documents should rank above any single document chunk on broad conceptual queries. On specific factual queries, the raw chunks will score higher. The ranking sorts itself.

Step 4: Wire It Into Your Ingest Handler

The reflection runs as a background job. It doesn't block the ingest response — that would add 2–3 seconds to every ingest call.

In your src/handlers/ingest.ts, after you've stored the document:

import { reflect } from '../engines/reflection';

// ... existing ingest logic ...

// After VECTORIZE.upsert() and DB insert succeed:
ctx.waitUntil(
  reflect(documentId, content, env).catch(err => {
    console.warn('[reflection] failed for', documentId, err.message);
  })
);

return new Response(JSON.stringify({
  success: true,
  documentId,
  chunks: chunkCount,
  // ... rest of response
}), { headers: { 'Content-Type': 'application/json' } });

ctx.waitUntil() is the Cloudflare Workers primitive for background work. The response returns immediately. The reflection runs after. The ingest API stays fast.

The .catch() is important. A failed reflection should never fail an ingest. Raw documents are the source of truth. Reflections are derived value — useful, but not critical path.

Step 5: Boost Reflections in Search

Add the reflection boost to your ranking logic in src/engines/hybrid.ts. After RRF fusion and before returning results:

// Apply reflection boost
const boosted = results.map(r => ({
  ...r,
  score: r.doc_type === 'reflection' || r.doc_type === 'summary'
    ? r.score * (r.reflection_score ?? 1.5)
    : r.score,
}));

return boosted.sort((a, b) => b.score - a.score);

This is a post-fusion boost, not a pre-fusion rerank. The reasoning: apply RRF across all results first, so reflections earn their place on raw relevance before getting boosted. A reflection that would not rank in the top 20 on raw similarity shouldn't appear just because it has a boost multiplier.

Step 6: Filtering by doc_type

Your search endpoint should accept a doc_type filter so callers can control what they see:

// In your search request handler:
const docTypeFilter = body.filters?.doc_type;

// Pass to Vectorize query:
const vectorFilter: Record<string, unknown> = {};
if (docTypeFilter) {
  vectorFilter.doc_type = docTypeFilter;
}

This gives callers three modes:

# Only reflections and summaries
POST /search
{ "query": "pricing decisions", "filters": { "doc_type": { "$in": ["reflection", "summary"] } } }

# Only source documents
POST /search
{ "query": "pricing decisions", "filters": { "doc_type": { "$eq": "raw" } } }

# Default: all types, reflections boosted
POST /search
{ "query": "pricing decisions" }

The default (no filter) is the most useful. Let the boost do its job. Restrict to raw when you need citations. Restrict to reflections when you want the synthesised view.

What Changes After You Build This

At 200 documents, the difference becomes noticeable. Queries that previously returned five fragmented chunks now surface a reflection that already synthesised those chunks. Broad conceptual queries — "what do we know about X?" — start returning genuinely useful summaries instead of just the most-similar individual paragraph.

At 2,000 documents, the reflection layer is the most valuable part of the system. The raw chunks answer specific factual questions. The reflections and summaries answer conceptual questions that could not be answered from any single document. The system has learned something no individual document contains.

One failure mode worth knowing: if your embedding model has poor semantic clustering — old bge-small at 384d with mixed-domain documents — the related-documents retrieval step will surface weak connections and produce shallow reflections. The 0.65 threshold filters most of this out, but if you're seeing reflections that seem off-topic, your embeddings are the first thing to check.

Deploying

wrangler d1 execute mcp-knowledge-db --remote --file=./migrations/003_add_reflection_fields.sql
wrangler deploy

Then ingest a few documents and watch what happens:

# Ingest document 1
curl -X POST https://your-worker.workers.dev/ingest \
  -H "Authorization: Bearer YOUR_KEY" \
  -H "Content-Type: application/json" \
  -d '{"id": "doc-001", "content": "Your document text here..."}'

# After a few seconds, check if a reflection was created
curl "https://your-worker.workers.dev/search" \
  -H "Authorization: Bearer YOUR_KEY" \
  -H "Content-Type: application/json" \
  -d '{"query": "your topic", "filters": {"doc_type": {"$eq": "reflection"}}}'

Reflections won't appear until there are related documents to synthesise. Ingest at least three documents on similar topics before expecting to see them.

What to Build Next

The reflection layer as described here fires after every ingest. That's expensive at high ingest volume: if you're batch-importing 10,000 documents, you don't want 10,000 individual reflection calls.

For bulk ingestion, gate it: call reflect() only when a document's similarity search returns a match above 0.8, or batch-run reflection after the bulk import completes. The POST /ingest/batch endpoint in the full repo does this.

The second thing worth building: surfacing reflections in your UI with a visual distinction. A search result that's a reflection should look different from a raw chunk. In the dashboard included in the repo, reflections render with a 💡 badge and a "synthesised from N documents" note.

Full source at github.com/dannwaneri/vectorize-mcp-worker — reflection engine, consolidation, batch ingest, dashboard, OpenAPI spec.

The codebase is TypeScript, deploys with a single wrangler deploy, runs for roughly $1–5/month at 10,000 queries/day.

Standard RAG retrieves. This learns.

The explanation everyone reaches for is that AI replaced it. That's partly true. But it misses the structural problem underneath: every time a developer asks Claude or ChatGPT to write code, the knowledge that shaped the answer disappears.

The GitHub discussion where someone spent two hours documenting why cursor-based pagination beats offset for live-updating datasets. The Stack Overflow answer from 2019 where one engineer, after a week of debugging, documented exactly why that approach fails under concurrent writes.

The AI consumed all of it. The humans who produced it got nothing — no citation in the codebase, no signal that their work mattered.

Over time, those people stopped contributing. Stack Overflow isn't dying because it's bad. It's dying because AI extracted its value and the feedback loop that kept humans contributing broke down.

This tutorial builds a tool that puts that loop back together. proof-of-contribution is a Claude Code skill that links every AI-generated artifact back to the human knowledge that inspired it — and surfaces exactly where the AI made choices with no human source at all.

I'll show you how to install proof-of-contribution, how to record your first provenance entry, how to use the spec-writer integration that makes Knowledge Gaps deterministic, and how to run poc.py verify — a static analyser that detects gaps without a single API call.

Table of Contents

1. What You Will Build

2. Prerequisites

3. Quickstart in 5 Minutes

4. How the Tool Works

5. How to Install proof-of-contribution

6. How to Scaffold Your Project

7. How to Record Your First Provenance Entry

8. How to Use import-spec to Seed Knowledge Gaps

9. How to Trace Human Attribution

10. How to Verify with Static Analysis

11. How to Enable PR Enforcement

12. Where to Go Next

What You Will Build

proof-of-contribution is a Claude Code skill with a local CLI. Together they give you:

• Provenance Blocks: Claude appends a structured attribution block to every generated artifact, listing the human sources that inspired it and flagging what it synthesized without any traceable source.

• Knowledge Gaps: the parts of AI-generated code that have no human citation, surfaced before they become production incidents

• poc.py trace: a CLI command that shows the full human attribution chain for any file in thirty seconds

• poc.py import-spec: bridges proof-of-contribution with spec-writer, seeding knowledge gaps from your spec's assumptions list before the agent builds anything

• poc.py verify: a static analyser that cross-checks your file's structure against seeded claims using Python's AST. Zero API calls. Exit code 0 means clean, exit code 1 means gaps found — wires directly into CI

• A GitHub Action: optional PR enforcement that fails PRs missing attribution, for teams that want a standard

The complete source is at github.com/dannwaneri/proof-of-contribution.

Prerequisites

This is a beginner-to-intermediate tutorial. You should be comfortable with:

• Command line basics: navigating directories, running scripts

• Git: basic commits and PRs

• Python 3.8 or higher: the CLI is pure Python with no dependencies

You will need:

• Python installed: check with python --version or python3 --version

• Git installed: check with git --version

• Claude Code (or any agent that supports the Agent Skills standard — Cursor and Gemini CLI also work)

There's no database to install. No API keys. No paid services. The default storage is SQLite, which Python includes out of the box.

Quickstart in 5 Minutes

If you want to try the tool before reading the full tutorial, here are the five commands that take you from zero to your first gap detection:

Mac and Linux:

# 1. Install
mkdir -p ~/.claude/skills
git clone https://github.com/dannwaneri/proof-of-contribution.git \
  ~/.claude/skills/proof-of-contribution

# 2. Scaffold your project (run in your repo root)
python ~/.claude/skills/proof-of-contribution/assets/scripts/poc_init.py

# 3. Record attribution for an AI-generated file
python poc.py add src/utils/parser.py

# 4. Detect gaps via static analysis
python poc.py verify src/utils/parser.py

# 5. See the full provenance chain
python poc.py trace src/utils/parser.py

Windows PowerShell:

# 1. Install
New-Item -ItemType Directory -Force -Path "$HOME\.claude\skills"
git clone https://github.com/dannwaneri/proof-of-contribution.git `
  "$HOME\.claude\skills\proof-of-contribution"

# 2. Scaffold your project
python "$HOME\.claude\skills\proof-of-contribution\assets\scripts\poc_init.py"

# 3. Record attribution
python poc.py add src\utils\parser.py

# 4. Detect gaps
python poc.py verify src\utils\parser.py

# 5. See the full provenance chain
python poc.py trace src\utils\parser.py

That's the whole tool. The sections below walk through each step in detail with real terminal output at every stage.

How the Tool Works

Before you install anything, you need a clear mental model of what proof-of-contribution actually does — because the most important part isn't obvious.

The Archaeology Problem

Here's a scenario that happens on every team using AI-assisted development.

A developer joins. They go through six months of AI-generated codebase. They hit a bug in the pagination logic — cursor-based, unusual implementation, nobody remembers why it was built that way. The original developer has left.

Old answer: two days of archaeology. git blame points to a commit message that says "fix pagination." The commit before that says "implement pagination." Dead end.

With poc.py trace src/utils/paginator.py, that same developer sees this in thirty seconds:

Provenance trace: src/utils/paginator.py
────────────────────────────────────────────────────────────
  [HIGH]  @tannerlinsley on github
          Cursor pagination discussion
          https://github.com/TanStack/query/discussions/123
          Insight: cursor beats offset for live-updating datasets

Knowledge gaps (AI-synthesized, no human source):
  • Error retry strategy — no human source cited
  • Concurrent write handling — AI chose this arbitrarily

They now know where the pattern came from and — critically — which parts have no traceable human source. The concurrent write handling is where the bug lives. The AI made a choice nobody reviewed.

That's what this tool does. Not enforcement first. Archaeology first.

How Knowledge Gaps Are Detected

The obvious assumption is that Claude introspects and reports what it doesn't know. That assumption is wrong. LLMs hallucinate confidently. An AI that could reliably detect its own knowledge gaps wouldn't produce them.

The detection mechanism is a comparison, not introspection.

When you use spec-writer before building, it generates a spec with an explicit ## Assumptions to review section — every decision the AI is making that you didn't specify, each one impact-rated. That list is the contract.

When you run poc.py import-spec spec.md --artifact src/utils/paginator.py, those assumptions get seeded into the database as unresolved knowledge gaps. After the agent builds, poc.py trace shows which assumptions made it into code with no human source ever cited.

The AI isn't grading its own exam. The spec is the answer key.

poc.py verify takes this further. After the agent builds, it parses the file's actual structure using Python's built-in ast module — extracting every function definition, conditional branch, and return path. It cross-checks each one against the seeded claims. Any structural unit with no resolved claim surfaces as a deterministic Knowledge Gap, regardless of how confident the model was when it wrote the code.

How to Install proof-of-contribution

Mac and Linux

mkdir -p ~/.claude/skills
git clone https://github.com/dannwaneri/proof-of-contribution.git \
  ~/.claude/skills/proof-of-contribution

Windows PowerShell

New-Item -ItemType Directory -Force -Path "$HOME\.claude\skills"
git clone https://github.com/dannwaneri/proof-of-contribution.git `
  "$HOME\.claude\skills\proof-of-contribution"

That's the entire installation. No package to install, no configuration file to edit. The skill is a markdown file the agent reads. The CLI is a Python script that runs locally.

Verify the Install:

ls ~/.claude/skills/proof-of-contribution/

You should see SKILL.md, poc.py, assets/, and references/. If the directory is empty, the clone failed — check your internet connection and try again.

How to Scaffold Your Project

The scaffold script creates the database, config, CLI, and GitHub integration in your project root. Run it once per project.

Mac and Linux

cd /path/to/your/project
python ~/.claude/skills/proof-of-contribution/assets/scripts/poc_init.py

Windows PowerShell

cd C:\path\to\your\project
python "$HOME\.claude\skills\proof-of-contribution\assets\scripts\poc_init.py"

You should see output like this:

🔗 Proof of Contribution — init

  →  Project root: /path/to/your/project
  ✔  Created .poc/config.json
  ✔  Created .poc/.gitignore  (db excluded from git, config tracked)
  ✔  Created .poc/provenance.db  (SQLite — no extra infra needed)
  ✔  Created .github/PULL_REQUEST_TEMPLATE.md
  ✔  Created .github/workflows/poc-check.yml
  ✔  Created poc.py  (local CLI — includes import-spec command)
  ✔  Created .gitignore

✔ Proof of Contribution initialised for 'your-project'

This creates four things in your project:

your-project/
├── .poc/
│   ├── config.json      ← project settings (commit this)
│   ├── provenance.db    ← SQLite database (local only, gitignored)
│   └── .gitignore
├── .github/
│   ├── PULL_REQUEST_TEMPLATE.md
│   └── workflows/
│       └── poc-check.yml
└── poc.py               ← your local CLI

• .poc/ — the tool's local data directory. config.json stores project settings and is committed to git. provenance.db is the SQLite database where attribution records and knowledge gaps are stored — local only, gitignored.

• poc.py — your local CLI, copied into the project root. Run python poc.py trace, python poc.py verify, and every other command directly without a global install.

• .github/PULL_REQUEST_TEMPLATE.md — a PR template with the ## 🤖 AI Provenance section pre-filled. Developers fill it in when submitting PRs that contain AI-generated code.

• .github/workflows/poc-check.yml — the optional GitHub Action for PR enforcement. Installed but dormant until you push the workflow file and enable it in your repo settings.

Windows note: if the scaffold fails with a UnicodeEncodeError, the emoji in the PR template is hitting a Windows encoding limit. Open assets/scripts/poc_init.py in a text editor and find every line ending with .write_text(...). Change each one to .write_text(..., encoding="utf-8"). Save and re-run.

Verify the Scaffold Worked

python poc.py report

Expected output:

Proof of Contribution Report
────────────────────────────────────────
  Artifacts tracked    : 0
  With provenance      : 0  (0%)
  Unresolved gaps      : 0
  Resolved claims      : 0
  Human experts        : 0

Empty database, clean state. You're ready.

How to Record Your First Provenance Entry

Before we dive in here, I just want to clear something up. Earlier, I described poc.py verify as detecting Knowledge Gaps automatically — and it does. But the static analyser can only tell you that a function has no human citation. It can't tell you which human source inspired it. That knowledge lives in your head, not in the code.

poc.py add is where you supply that context. After the agent builds a file, you record the human sources you actually drew on: the GitHub discussion you read before prompting, the Stack Overflow answer that shaped the approach. Those records become the attribution chain poc.py trace surfaces — and what closes the gaps poc.py verify flags.

verify finds the gaps. add fills them.

poc.py add records attribution for a file interactively. You can run it on any AI-generated file in your project.

python poc.py add src/utils/parser.py

You'll see a prompt:

Recording provenance for: src/utils/parser.py
(Press Ctrl+C to cancel)

  Human source URL (or Enter to finish):

Enter the URL of the human-authored source that inspired the code. This could be a GitHub discussion, a Stack Overflow answer, a documentation page, a blog post, or an RFC.

  Human source URL (or Enter to finish): https://github.com/TanStack/query/discussions/123
  Author handle: tannerlinsley
  Platform (github/stackoverflow/docs/other): github
  Source title: Cursor pagination discussion
  What specific insight came from this? cursor beats offset for live-updating datasets
  Confidence HIGH/MEDIUM/LOW [MEDIUM]: HIGH
  ✔ Recorded.

Add as many sources as apply. Press Enter on a blank URL when you're done.

  Human source URL (or Enter to finish): 
✔ Provenance saved. Run: python poc.py trace src/utils/parser.py

Check What You Recorded

python poc.py trace src/utils/parser.py

Provenance trace: src/utils/parser.py
────────────────────────────────────────────────────────────
  [HIGH]  @tannerlinsley on github
          Cursor pagination discussion
          https://github.com/TanStack/query/discussions/123
          Insight: cursor beats offset for live-updating datasets

No knowledge gaps — because you recorded a source. If the file had parts with no human source, they would appear below as gaps.

See All Experts in Your Graph

Every poc.py add call stores not just the URL but the author — their handle, platform, and the specific insight they contributed. Run it across enough files, and those authors accumulate into a knowledge graph: a local record of which human experts your codebase drew from, which files their knowledge shaped, and how many artifacts trace back to their work.

poc.py experts surfaces the top contributors. On a new project, it'll be one or two entries. On a mature codebase, it becomes a map of whose knowledge is load-bearing — the people you'd want to consult if that part of the code ever needed to change.

python poc.py experts

Top Human Experts in Knowledge Graph
──────────────────────────────────────────────────────
  @tannerlinsley            github          1 artifact(s)

How to Use import-spec to Seed Knowledge Gaps

This is the most important command in the tool. It connects proof-of-contribution with spec-writer and makes Knowledge Gaps deterministic.

When you use spec-writer before building a feature, it generates an ## Assumptions to review section — every implicit decision is impact-rated HIGH, MEDIUM, or LOW. The import-spec command reads that section and seeds those assumptions into the database as unresolved gaps before the agent writes a line of code.

After the agent builds, any assumption that made it into the implementation without a cited human source surfaces automatically in poc.py trace. You don't need to know which parts of the code are uncertain. The spec already told you.

Step 1 — Create a Test Spec

If you don't have a spec-writer output yet, create one manually to see how the import works.

Mac and Linux:

cat > test-spec.md << 'EOF'
## Assumptions to review

1. SQLite is sufficient for single-developer use — Impact: HIGH
   Correct this if: you need team-shared provenance

2. Filepath is the artifact identifier — Impact: MEDIUM
   Correct this if: you use content hashing instead

3. REST pattern for any future API — Impact: LOW
   Correct this if: you prefer GraphQL
EOF

Windows PowerShell:

python -c "
content = '''## Assumptions to review

1. SQLite is sufficient for single-developer use - Impact: HIGH
   Correct this if: you need team-shared provenance

2. Filepath is the artifact identifier - Impact: MEDIUM
   Correct this if: you use content hashing instead

3. REST pattern for any future API - Impact: LOW
   Correct this if: you prefer GraphQL'''
open('test-spec.md', 'w', encoding='utf-8').write(content)
print('test-spec.md created')
"

Windows note: don't use PowerShell's echo to create spec files. PowerShell saves files as UTF-16, which causes a UnicodeDecodeError when import-spec reads them. The python -c approach above writes UTF-8 correctly.

Step 2 — Import the Assumptions

python poc.py import-spec test-spec.md --artifact src/utils/parser.py

Spec assumptions imported — 3 Knowledge Gap(s) seeded
───────────────────────────────────────────────────────
  1. [HIGH] SQLite is sufficient for single-developer use
       Correct if: you need team-shared provenance
  2. [MEDIUM] Filepath is the artifact identifier
       Correct if: you use content hashing instead
  3. [LOW] REST pattern for any future API
       Correct if: you prefer GraphQL

  →  Bound to: src/utils/parser.py
  After the agent builds, run:
  python poc.py trace src/utils/parser.py
  python poc.py add src/utils/parser.py

Step 3 — Trace the Gaps

python poc.py trace src/utils/parser.py

Knowledge gaps (AI-synthesized, no human source):
  • REST pattern for any future API [Correct if: you prefer GraphQL]
  • SQLite is sufficient for single-developer use [Correct if: you need team-shared provenance]
  • Filepath is the artifact identifier [Correct if: you use content hashing instead]

  Resolve gaps: python poc.py add src/utils/parser.py

Three gaps, colour-coded by urgency. The HIGH-impact assumption — SQLite for single-developer use — appears in red. The LOW-impact one appears in green. When you run poc.py add and record a human source with an insight that overlaps the gap text, the gap auto-closes.

Preview Without Writing

python poc.py import-spec test-spec.md --dry-run

This parses the spec and prints what would be seeded without touching the database. This is useful before committing to an import.

Check the Overall Health

python poc.py report

Proof of Contribution Report
────────────────────────────────────────
  Artifacts tracked    : 1
  With provenance      : 0  (0%)
  Unresolved gaps      : 3
  Resolved claims      : 0
  Human experts        : 1
  ⚠ Less than 50% of artifacts have provenance records.
  ⚠ 3 unresolved Knowledge Gap(s).
    Run `poc.py trace <filepath>` to locate them.

How to Trace Human Attribution

poc.py trace is the command you'll use most. It shows the full human attribution chain for any file and lists any knowledge gaps — parts of the code with no traceable human source.

python poc.py trace src/utils/parser.py

A file with both attribution and gaps looks like this:

Provenance trace: src/utils/parser.py
────────────────────────────────────────────────────────────
  [HIGH]  @juliandeangelis on github
          Spec Driven Development methodology
          https://github.com/dannwaneri/spec-writer
          Insight: separate functional from technical spec

  [MEDIUM] @tannerlinsley on github
           Cursor pagination discussion
           https://github.com/TanStack/query/discussions/123
           Insight: cursor beats offset for live-updating datasets

Knowledge gaps (AI-synthesized, no human source):
  • Error retry strategy — no human source cited
  • CSV column ordering — AI chose this arbitrarily

  Resolve gaps: python poc.py add src/utils/parser.py

The human attribution section shows every cited source, colour-coded by confidence. The knowledge gaps section shows every assumption that shipped without a human citation — either seeded from a spec via import-spec, or flagged by Claude in the Provenance Block.

Resolving Gaps

Run poc.py add on any file with open gaps:

python poc.py add src/utils/parser.py

When you enter an insight that shares words with an open gap claim, the gap auto-closes. Run poc.py trace again to confirm it's resolved.

How to Verify with Static Analysis

poc.py verify is the command that closes the epistemic trust gap completely. It detects Knowledge Gaps by analysing the file's actual code structure — not by asking the AI what it doesn't know.

Run it after the agent builds, once you've seeded gaps with import-spec:

python poc.py verify src/utils/parser.py

Expected output:

Verify: src/utils/parser.py
────────────────────────────────────────────────────────────
  Structural units detected : 11
  Seeded claims             : 3
  Covered by cited source   : 2
  Deterministic gaps        : 1

Deterministic Knowledge Gaps (no human source):
  • function: handle_concurrent_writes (lines 47–61)
      Seeded assumption: concurrent write handling — AI chose this arbitrarily

  Resolve: python poc.py add src/utils/parser.py

The gap shown is not something Claude admitted. It's something the analyser found by comparing the file's function list against your seeded claims. The function handle_concurrent_writes exists in the code but has no resolved human citation in the database. That's the gap.

What the Exit Codes Mean

python poc.py verify src/utils/parser.py
echo $?   # Mac/Linux

python poc.py verify src/utils/parser.py
echo $LASTEXITCODE   # Windows PowerShell

• Exit code 0 — no gaps, all detected units have cited sources

• Exit code 1 — gaps found, resolve with poc.py add

• Exit code 2 — file not found or unsupported language

Exit code 1 integrates directly into CI pipelines. Add poc.py verify to your GitHub Action or pre-commit hook and gaps block the build before they reach production.

Run it Without a Seeded Spec

If you haven't run import-spec first, verify still works — it falls back to structural analysis and surfaces every uncited function and branch as a gap:

python poc.py verify src/utils/parser.py

⚠ No spec imported — showing all uncited structural units.
  Run: python poc.py import-spec spec.md --artifact src/utils/parser.py
  for deterministic gap detection.

Deterministic Knowledge Gaps (no human source):
  • function: parse_query (lines 1–7)
  • branch: if not text (lines 2–3)
  • function: fetch_results (lines 9–12)
  ...

It's less precise than the spec-writer path — every structural unit shows rather than only the ones tied to named assumptions — but it's useful as a baseline on any file, new or old.

The --strict Flag

python poc.py verify src/utils/parser.py --strict

Strict mode flags every uncited structural unit as a gap even when claims are seeded. You can use it when you want zero tolerance: any function or branch without a resolved human source fails the check.

How to Enable PR Enforcement

Once poc.py trace has saved you real hours — not before — enable the GitHub Action. The distinction matters. Turning it on day one frames the tool as overhead. Turning it on after the team already finds value frames it as a standard.

git add .github/ .poc/config.json poc.py
git commit -m "chore: add proof-of-contribution"
git push

After that, every PR is checked for an ## 🤖 AI Provenance section. The scaffold already created the PR template with that section included. Developers fill it in naturally once they're already running poc.py trace locally — the template just asks them to record what they already know.

Developers who write fully human code opt out by adding 100% human-written anywhere in the PR body. The action skips the check automatically.

What the Action Checks

The action reads the PR description and looks for:

1. The ## 🤖 AI Provenance heading

2. At least one populated row in the attribution table

If the section is missing or the table is empty, the action fails and posts a comment explaining what to add. The comment includes a link to poc.py trace <filepath> so the developer knows exactly where to look.

Where to Go Next

Use it with spec-writer on a Real Feature

The real value of import-spec is on actual features, not test specs. If you use spec-writer, the workflow is:

/spec-writer "your feature description"

Save the output to spec.md. Then:

python poc.py import-spec spec.md --artifact src/path/to/output.py

Build the feature with your agent. Then run poc.py trace to see which assumptions made it into code with no human source. Resolve the HIGH-impact gaps first — those are the ones that will cause production incidents.

Activate the Claude Code Skill

The SKILL.md file makes Claude automatically append a Provenance Block to every generated artifact when the skill is active. The block lists human sources Claude drew from and flags what it synthesized without any traceable source.

To activate it in Claude Code, the skill is already installed at ~/.claude/skills/proof-of-contribution/. Claude Code loads it automatically when you are in a project that has .poc/config.json.

A generated Provenance Block looks like this:

## PROOF OF CONTRIBUTION
Generated artifact: fetch_github_discussions()
Confidence: MEDIUM

## HUMAN SOURCES THAT INSPIRED THIS

[1] GitHub GraphQL API Documentation Team
    Source type: Official Docs
    URL: docs.github.com/en/graphql
    Contribution: cursor-based pagination pattern

[2] GitHub Community (multiple contributors)
    Source type: GitHub Discussions
    URL: github.com/community/community
    Contribution: "ghost" fallback for deleted accounts
                  surfaced in bug reports

## KNOWLEDGE GAPS (AI synthesized, no human cited)
- Error handling / retry logic
- Rate limit strategy

## RECOMMENDED HUMAN EXPERTS TO CONSULT
- github.com/octokit community for pagination

The Knowledge Gaps section is the part no other tool produces. It's where AI admits what it synthesized without a traceable human source — before that gap becomes a production incident.

Upgrade When You Outgrow SQLite

The default database is SQLite — local only, no infra required. When you need team sharing or graph queries, the references/ directory in the repo has migration guides:

Manual Tracking vs. proof-of-contribution

The tool doesn't replace code review. It gives code review the context it needs to catch the right things.

The archaeology scenario — two days tracing a bug through dead-end commit messages — takes thirty seconds with poc.py trace. The code still has gaps, and it always will. But now you know where they are.

Built by Daniel Nwaneri. The spec-writer skill that feeds import-spec is at github.com/dannwaneri/spec-writer. The full proof-of-contribution repo is at github.com/dannwaneri/proof-of-contribution.

A character count doesn't need GPT-4. A presence check doesn't need Sonnet. A regex doesn't need anything except Python.

The mistake isn't using AI — it's not knowing when to stop using it.

This tutorial shows you how to build a tiered routing system that sends tasks to the cheapest model that can solve them. The pattern is called the cost curve. It comes from a comment thread on a DEV.to article, implemented by three developers over a weekend, and it cut the per-URL cost of a real SEO audit agent from \(0.006 to effectively \)0 for most pages.

By the end, you'll have a working cost_curve.py module you can drop into any agent project.

What You'll Build

A three-tier routing function that:

• Runs deterministic Python checks first — zero API cost

• Escalates to Claude Haiku only for genuinely ambiguous cases — ~$0.0001 per call

• Escalates to Claude Sonnet only when semantic judgment is required — ~$0.006 per call

• Falls back gracefully when any tier fails

• Returns a consistent result schema regardless of which tier handled the request

The full implementation is part of dannwaneri/seo-agent, an open-core SEO audit agent. The cost curve module is the premium routing layer, and the principle applies to any agent with mixed-complexity tasks.

Prerequisites

• Python 3.11 or higher

• An Anthropic API key

• Basic familiarity with Python and the Claude API

Table of Contents

1. The Problem with Calling Claude on Everything

2. The Cost Curve Explained

3. Project Setup

4. Tier 1: Deterministic Python

5. Tier 2: Claude Haiku for Ambiguous Cases

6. Tier 3: Claude Sonnet for Semantic Judgment

7. The Router: audit_url()

8. Graceful Fallback

9. Testing the Cost Curve

10. Applying This Pattern to Your Agent

The Problem with Calling Claude on Everything

Here's what most agent code looks like:

def audit_url(snapshot: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        messages=[{"role": "user", "content": build_prompt(snapshot)}]
    )
    return parse_response(response)

This works. It also calls Sonnet for every URL in the list — including the ones where the title is 142 characters long and the answer is obviously FAIL without any model involvement.

Claude Sonnet 4 is priced at \(3 per million input tokens and \)15 per million output tokens. A typical page snapshot is around 500 input tokens. That's \(0.0015 per URL just for input — before output tokens. Across a 20-URL weekly audit, the total is around \)0.12. Not expensive. But most of those pages have mechanical SEO issues: missing descriptions, titles over 60 characters, no canonical tag. A character count catches all of that. You don't need a model.

The cost curve fixes this by routing based on what the task actually requires, not on what the model is capable of.

The Cost Curve Explained

In the cost curve, we have three tiers, three tools, and three price points:

Tier 1 — Deterministic Python. Cost: $0. Check title length, description length, H1 count, canonical presence. These are not judgment calls. They're string operations. If title length > 60, FAIL. No model needed.

Tier 2 — Claude Haiku. Cost: ~$0.0001 per call. Title present but only 4 characters long. Description present but only 30 characters. Status code is a redirect. These pass the mechanical audit but something is off. Haiku is fast and cheap enough that escalating ambiguous cases costs less than the debugging time you'd spend on false positives.

Tier 3 — Claude Sonnet. Cost: ~$0.006 per call. Pages Haiku flags as needing semantic judgment. "This title passes length but reads like a navigation label." "This description duplicates the title verbatim." Sonnet earns its cost on genuinely hard cases — not on every URL in the list.

The routing decision happens before any API call. The result schema is identical regardless of which tier handled the request.

Project Setup

mkdir cost-curve-demo && cd cost-curve-demo
pip install anthropic

Set your API key:

# macOS/Linux
export ANTHROPIC_API_KEY="sk-ant-..."

# Windows PowerShell
$env:ANTHROPIC_API_KEY = "sk-ant-..."

Create cost_curve.py — you'll build this module step by step.

Tier 1: Deterministic Python

Tier 1 runs first on every URL. It checks four fields using only Python string operations. There's no API call, no latency, and no cost.

import json
import logging
import os
import re
from datetime import datetime, timezone

import anthropic

logger = logging.getLogger(__name__)

REDIRECT_CODES = {301, 302, 307, 308}

# Fields that trigger Tier 2 escalation
# Title or description present but suspiciously short
AMBIGUOUS_TITLE_MAX = 10   # chars — present but too short to be real
AMBIGUOUS_DESC_MAX = 50    # chars — present but too short to be useful


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _build_result(snapshot: dict, method: str) -> dict:
    """Base result skeleton — same schema regardless of tier."""
    return {
        "url": snapshot.get("final_url", ""),
        "final_url": snapshot.get("final_url", ""),
        "status_code": snapshot.get("status_code"),
        "title": {"value": None, "length": 0, "status": "PASS"},
        "description": {"value": None, "length": 0, "status": "PASS"},
        "h1": {"count": 0, "value": None, "status": "PASS"},
        "canonical": {"value": None, "status": "PASS"},
        "flags": [],
        "human_review": False,
        "audited_at": _now_iso(),
        "method": method,
        "needs_tier3": False,
    }


def tier1_check(snapshot: dict) -> dict:
    """
    Pure Python SEO checks. Zero API calls.

    Returns a result dict with method="deterministic".
    Sets needs_tier3=False always — Tier 1 never escalates to Tier 3 directly.
    Escalation to Tier 2 is decided by the router, not here.
    """
    result = _build_result(snapshot, "deterministic")

    title = snapshot.get("title") or ""
    description = snapshot.get("meta_description") or ""
    h1s = snapshot.get("h1s") or []
    canonical = snapshot.get("canonical") or ""

    # Title check
    result["title"]["value"] = title or None
    result["title"]["length"] = len(title)
    if not title or len(title) > 60:
        result["title"]["status"] = "FAIL"
        msg = "Title is missing" if not title else f"Title is {len(title)} characters (max 60)"
        result["flags"].append(msg)

    # Description check
    result["description"]["value"] = description or None
    result["description"]["length"] = len(description)
    if not description or len(description) > 160:
        result["description"]["status"] = "FAIL"
        msg = "Meta description is missing" if not description else f"Meta description is {len(description)} characters (max 160)"
        result["flags"].append(msg)

    # H1 check
    result["h1"]["count"] = len(h1s)
    result["h1"]["value"] = h1s[0] if h1s else None
    if len(h1s) == 0:
        result["h1"]["status"] = "FAIL"
        result["flags"].append("H1 tag is missing")
    elif len(h1s) > 1:
        result["h1"]["status"] = "FAIL"
        result["flags"].append(f"Multiple H1 tags found ({len(h1s)})")

    # Canonical check
    result["canonical"]["value"] = canonical or None
    if not canonical:
        result["canonical"]["status"] = "FAIL"
        result["flags"].append("Canonical tag is missing")

    return result

The key design decision: tier1_check() never decides whether to escalate. It just runs the checks and returns. The router decides escalation based on the result.

Tier 2: Claude Haiku for Ambiguous Cases

Tier 2 runs when Tier 1 detects something mechanical but the result might need a second look. A 4-character title present but clearly wrong. A 30-character description that's technically there but useless. A redirect status that needs a human-readable explanation.

Haiku is the right model here. It's fast, cheap (\(1 input / \)5 output per million tokens), and sufficient for triage-level judgment. The prompt asks a narrow question: is this ambiguous enough to need Sonnet?

def tier2_check(snapshot: dict) -> dict:
    """
    Claude Haiku call for ambiguous cases.

    Returns result with method="haiku".
    Sets needs_tier3=True if Haiku determines the case needs semantic judgment.
    Falls back to Tier 1 result on API error.
    """
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        raise OSError("ANTHROPIC_API_KEY is not set.")

    client = anthropic.Anthropic(api_key=api_key)

    title = snapshot.get("title") or ""
    description = snapshot.get("meta_description") or ""
    status_code = snapshot.get("status_code")

    prompt = f"""You are an SEO auditor doing a quick triage check.

Page data:
- Title: {repr(title)} ({len(title)} chars)
- Meta description: {repr(description)} ({len(description)} chars)
- Status code: {status_code}

Answer these two questions with only "yes" or "no":
1. Does this page need semantic judgment beyond simple length/presence checks? 
   (e.g. title is present but clearly wrong, description is present but meaningless)
2. Is the status code a redirect that needs investigation?

Respond in this exact JSON format and nothing else:
{{"needs_tier3": true_or_false, "reason": "one sentence explanation"}}"""

    try:
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=150,
            messages=[{"role": "user", "content": prompt}],
        )
        raw = response.content[0].text.strip()
        # Strip markdown fences if present
        if raw.startswith("```"):
            lines = raw.splitlines()
            raw = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])
        parsed = json.loads(raw)

        result = _build_result(snapshot, "haiku")
        # Copy Tier 1 field checks — Haiku doesn't redo those
        t1 = tier1_check(snapshot)
        result["title"] = t1["title"]
        result["description"] = t1["description"]
        result["h1"] = t1["h1"]
        result["canonical"] = t1["canonical"]
        result["flags"] = t1["flags"]
        result["needs_tier3"] = parsed.get("needs_tier3", False)
        if result["needs_tier3"]:
            result["flags"].append(f"Escalated to Tier 3: {parsed.get('reason', '')}")

        return result

    except Exception as exc:
        logger.warning("[tier2] Haiku API error: %s — falling back to Tier 1 result", exc)
        fallback = tier1_check(snapshot)
        fallback["method"] = "haiku-fallback"
        return fallback

The fallback is the critical piece. If Haiku fails — rate limit, network error, malformed response — the function returns the Tier 1 result rather than crashing. The audit continues. The URL gets flagged with method="haiku-fallback" so you can identify it later.

Tier 3: Claude Sonnet for Semantic Judgment

Tier 3 is where the full extraction prompt runs. This is the same call you'd make in a naïve implementation — the difference is that only a small fraction of URLs reach this tier.

def tier3_check(snapshot: dict) -> dict:
    """
    Claude Sonnet call for semantic judgment.

    Returns result with method="sonnet".
    This is the full extraction prompt — same as calling the model directly.
    """
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        raise OSError("ANTHROPIC_API_KEY is not set.")

    client = anthropic.Anthropic(api_key=api_key)

    prompt = f"""You are an SEO auditor. Analyze this page snapshot and return ONLY a JSON object.
No prose. No explanation. No markdown fences. Raw JSON only.

Page data:
- URL: {snapshot.get('final_url')}
- Status code: {snapshot.get('status_code')}
- Title: {snapshot.get('title')}
- Meta description: {snapshot.get('meta_description')}
- H1 tags: {snapshot.get('h1s')}
- Canonical: {snapshot.get('canonical')}

Return this exact schema:
{{
  "url": "string",
  "final_url": "string",
  "status_code": number,
  "title": {{"value": "string or null", "length": number, "status": "PASS or FAIL"}},
  "description": {{"value": "string or null", "length": number, "status": "PASS or FAIL"}},
  "h1": {{"count": number, "value": "string or null", "status": "PASS or FAIL"}},
  "canonical": {{"value": "string or null", "status": "PASS or FAIL"}},
  "flags": ["array of strings describing specific issues"],
  "human_review": false,
  "audited_at": "ISO timestamp"
}}

PASS/FAIL rules:
- title: FAIL if null or length > 60 characters, or if present but clearly not a real title
- description: FAIL if null or length > 160 characters, or if present but meaningless
- h1: FAIL if count is 0 or count > 1
- canonical: FAIL if null
- audited_at: use current UTC time"""

    try:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            messages=[{"role": "user", "content": prompt}],
        )
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            lines = raw.splitlines()
            raw = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])

        result = json.loads(raw)
        result["method"] = "sonnet"
        result["needs_tier3"] = False
        return result

    except Exception as exc:
        logger.warning("[tier3] Sonnet API error: %s — falling back to Tier 1 result", exc)
        fallback = tier1_check(snapshot)
        fallback["method"] = "sonnet-fallback"
        return fallback

Note the prompt addition in Tier 3 that isn't in Tier 1: "or if present but clearly not a real title" and "or if present but meaningless". That's the semantic judgment Haiku identified as needed. Tier 3 acts on it.

The Router: audit_url()

The router is the public interface. Everything else is an implementation detail.

def audit_url(snapshot: dict, tiered: bool = False) -> dict:
    """
    Route a page snapshot through the appropriate audit tier.

    Args:
        snapshot: Page data from browser.py — must contain final_url,
                  status_code, title, meta_description, h1s, canonical.
        tiered: If False, delegates directly to Tier 3 (Sonnet).
                If True, routes through the cost curve.

    Returns:
        Audit result dict with method field indicating which tier ran.
    """
    if not tiered:
        # Non-tiered mode: call Sonnet directly, same as v1 behavior
        return tier3_check(snapshot)

    # Tier 1: always runs first
    t1_result = tier1_check(snapshot)

    # Check if escalation to Tier 2 is warranted
    title = snapshot.get("title") or ""
    description = snapshot.get("meta_description") or ""
    status_code = snapshot.get("status_code")

    needs_tier2 = (
        # Title present but suspiciously short
        (title and len(title) < AMBIGUOUS_TITLE_MAX) or
        # Description present but suspiciously short
        (description and len(description) < AMBIGUOUS_DESC_MAX) or
        # Redirect status — may need explanation
        (status_code in REDIRECT_CODES)
    )

    if not needs_tier2:
        # Tier 1 result is definitive — return without any API call
        return t1_result

    # Tier 2: Haiku triage
    t2_result = tier2_check(snapshot)

    if not t2_result.get("needs_tier3", False):
        # Haiku determined no semantic judgment needed
        return t2_result

    # Tier 3: Sonnet for semantic judgment
    return tier3_check(snapshot)

The router logic is explicit and readable. Each decision point is a named condition. When tiered=False, behavior is identical to the v1 naive implementation — this is the backward compatibility guarantee that lets you add the cost curve incrementally without breaking existing audits.

Graceful Fallback

The fallback pattern appears in both Tier 2 and Tier 3. It's worth making explicit:

# Pattern used in both tier2_check() and tier3_check()
except Exception as exc:
    logger.warning("[tierN] API error: %s — falling back to Tier 1 result", exc)
    fallback = tier1_check(snapshot)
    fallback["method"] = "tierN-fallback"
    return fallback

Three things this does:

1. Logs the error with enough context to debug later

2. Returns a valid result — the Tier 1 deterministic check always runs regardless

3. Tags the result with the fallback method so you can filter these in your report

An agent that crashes on API errors is not production-ready. An agent that degrades gracefully and continues is.

Testing the Cost Curve

Create test_cost_curve.py to verify routing behavior without live API calls:

import json
from unittest import mock

from cost_curve import audit_url, tier1_check


def make_snapshot(title="Normal Title Under 60 Chars",
                  description="A normal meta description that is under 160 characters and describes the page content well.",
                  h1s=["Single H1"],
                  canonical="https://example.com/page",
                  status_code=200,
                  final_url="https://example.com/page"):
    return {
        "title": title,
        "meta_description": description,
        "h1s": h1s,
        "canonical": canonical,
        "status_code": status_code,
        "final_url": final_url,
    }


def test_clean_page_returns_tier1_no_api_calls():
    """Clean page: all checks pass deterministically — no API call."""
    snapshot = make_snapshot()
    with mock.patch("anthropic.Anthropic") as mock_client:
        result = audit_url(snapshot, tiered=True)
        assert result["method"] == "deterministic"
        mock_client.assert_not_called()
    print("PASS: clean page → Tier 1, zero API calls")


def test_long_title_returns_tier1_fail_no_api_call():
    """Title >60 chars: FAIL from Tier 1, no API call."""
    snapshot = make_snapshot(title="A" * 70)
    with mock.patch("anthropic.Anthropic") as mock_client:
        result = audit_url(snapshot, tiered=True)
        assert result["method"] == "deterministic"
        assert result["title"]["status"] == "FAIL"
        mock_client.assert_not_called()
    print("PASS: title >60 → Tier 1 FAIL, zero API calls")


def test_suspiciously_short_title_escalates_to_tier2():
    """Title present but 4 chars: escalates to Tier 2."""
    snapshot = make_snapshot(title="SEO")  # 3 chars — under AMBIGUOUS_TITLE_MAX
    mock_response = mock.MagicMock()
    mock_response.content = [mock.MagicMock(
        text='{"needs_tier3": false, "reason": "title is short but not ambiguous"}'
    )]
    with mock.patch("anthropic.Anthropic") as mock_client:
        mock_client.return_value.messages.create.return_value = mock_response
        result = audit_url(snapshot, tiered=True)
        assert result["method"] == "haiku"
        assert mock_client.return_value.messages.create.call_count == 1
    print("PASS: short title → Tier 2 (Haiku called once)")


def test_tiered_false_calls_sonnet_directly():
    """tiered=False: Sonnet called regardless of snapshot content."""
    snapshot = make_snapshot()  # clean page, would be Tier 1 in tiered mode
    mock_response = mock.MagicMock()
    mock_response.content = [mock.MagicMock(text=json.dumps({
        "url": "https://example.com/page",
        "final_url": "https://example.com/page",
        "status_code": 200,
        "title": {"value": "Normal Title Under 60 Chars", "length": 27, "status": "PASS"},
        "description": {"value": "desc", "length": 4, "status": "PASS"},
        "h1": {"count": 1, "value": "Single H1", "status": "PASS"},
        "canonical": {"value": "https://example.com/page", "status": "PASS"},
        "flags": [],
        "human_review": False,
        "audited_at": "2026-04-01T00:00:00+00:00",
    }))]
    with mock.patch("anthropic.Anthropic") as mock_client:
        mock_client.return_value.messages.create.return_value = mock_response
        result = audit_url(snapshot, tiered=False)
        assert result["method"] == "sonnet"
        assert mock_client.return_value.messages.create.call_count == 1
    print("PASS: tiered=False → Sonnet called directly")


def test_haiku_api_failure_falls_back_to_tier1():
    """Haiku failure: falls back to Tier 1 result, no crash."""
    snapshot = make_snapshot(title="SEO")  # triggers Tier 2
    with mock.patch("anthropic.Anthropic") as mock_client:
        mock_client.return_value.messages.create.side_effect = Exception("rate limit")
        result = audit_url(snapshot, tiered=True)
        assert result["method"] == "haiku-fallback"
    print("PASS: Haiku failure → fallback to Tier 1, no crash")


if __name__ == "__main__":
    test_clean_page_returns_tier1_no_api_calls()
    test_long_title_returns_tier1_fail_no_api_call()
    test_suspiciously_short_title_escalates_to_tier2()
    test_tiered_false_calls_sonnet_directly()
    test_haiku_api_failure_falls_back_to_tier1()
    print("\nAll tests passed.")

Run it:

python test_cost_curve.py

Expected output:

PASS: clean page → Tier 1, zero API calls
PASS: title >60 → Tier 1 FAIL, zero API calls
PASS: short title → Tier 2 (Haiku called once)
PASS: tiered=False → Sonnet called directly
PASS: Haiku failure → fallback to Tier 1, no crash

Applying This Pattern to Your Agent

The cost curve is not SEO-specific. Any agent with mixed-complexity tasks can use it.

The principle: classify tasks by what they actually require before deciding which model to invoke.

Customer support agent:

• Tier 1: keyword matching for known FAQ topics — no model

• Tier 2: Haiku for intent classification on ambiguous queries

• Tier 3: Sonnet for complex complaints requiring judgment

Code review agent:

• Tier 1: lint rules, syntax checks — no model

• Tier 2: Haiku for common pattern detection

• Tier 3: Sonnet for architectural review

Content moderation agent:

• Tier 1: blocklist matching — no model

• Tier 2: Haiku for borderline cases

• Tier 3: Sonnet for context-dependent judgment

The implementation pattern is the same in all three cases. The audit_url() router becomes route_task(). The tier functions change their prompts and escalation conditions. The fallback logic stays identical.

The key question to ask before writing any agent code: what fraction of my inputs are mechanically solvable? That fraction goes to Tier 1. The rest escalate. The cost curve routes everything else.

Wrapping Up

The full implementation — including the SEO audit agent that uses this module in production — is at dannwaneri/seo-agent. The core/ directory is MIT licensed. The tiered routing lives in premium/cost_curve.py.

This tutorial is the companion piece to I Was Paying \(0.006 Per URL for SEO Audits Until I Realized Most Needed \)0 on DEV.to, which covers the architecture decisions behind the cost curve.

That work is deterministic. An agent can do it.

In this tutorial, you'll build a local SEO audit agent from scratch using Python, Browser Use, and the Claude API. The agent visits real pages in a visible browser window, extracts SEO signals using Claude, checks for broken links asynchronously, handles edge cases with a human-in-the-loop pause, and writes a structured report — all resumable if interrupted.

By the end, you'll have a working agent you can run against any list of URLs. It costs less than $0.01 per URL to run.

What You'll Build

A seven-module Python agent that:

• Reads a URL list from a CSV file

• Visits each URL in a real Chromium browser (not a headless scraper)

• Extracts title, meta description, H1s, and canonical tag via Claude API

• Checks for broken links asynchronously using httpx

• Detects edge cases (404s, login walls, redirects) and pauses for human input

• Writes results to report.json incrementally — safe to interrupt and resume

• Generates a plain-English report-summary.txt on completion

The full code is on GitHub at dannwaneri/seo-agent.

Prerequisites

• Python 3.11 or higher

• An Anthropic API key (get one at console.anthropic.com)

• Windows, macOS, or Linux

• Basic familiarity with Python and the command line

Table of Contents

1. Why Browser Use Instead of a Scraper

2. Project Structure

3. Setup

4. Module 1: State Management

5. Module 2: Browser Integration

6. Module 3: Claude Extraction Layer

7. Module 4: Broken Link Checker

8. Module 5: Human-in-the-Loop

9. Module 6: Report Writer

10. Module 7: The Main Loop

11. Running the Agent

12. Scheduling for Agency Use

13. What the Results Look Like

Why Browser Use Instead of a Scraper

The standard approach to SEO auditing is to fetch page HTML with requests and parse it with BeautifulSoup. That works on static pages. It breaks on JavaScript-rendered content, misses dynamically injected meta tags, and fails entirely on authenticated pages.

Browser Use (84,000+ GitHub stars, MIT license) takes a different approach. It controls a real Chromium browser, reads the DOM after JavaScript executes, and exposes the page through Playwright's accessibility tree. The agent sees what a human would see.

The practical difference: a requests-based scraper might miss a meta description injected by a React component. Browser Use won't.

The other difference worth naming: Browser Use reads pages semantically. A Playwright script breaks when a button's CSS class changes from btn-primary to button-main. Browser Use identifies it's still a "Submit" button and acts accordingly. The extraction logic lives in the Claude prompt, not in brittle CSS selectors.

Project Structure

seo-agent/
├── index.py          # Main audit loop
├── browser.py        # Browser Use / Playwright page driver
├── extractor.py      # Claude API extraction layer
├── linkchecker.py    # Async broken link checker
├── hitl.py           # Human-in-the-loop pause logic
├── reporter.py       # Report writer
├── state.py          # State persistence (resume on interrupt)
├── input.csv         # Your URL list
├── requirements.txt
├── .env.example
└── .gitignore

Setup

Create a project folder and install dependencies:

mkdir seo-agent && cd seo-agent
pip install browser-use anthropic playwright httpx
playwright install chromium

Create input.csv with your URLs:

url
https://example.com
https://example.com/about
https://example.com/contact

Create .env.example:

ANTHROPIC_API_KEY=your-key-here

Set your API key as an environment variable before running:

# macOS/Linux
export ANTHROPIC_API_KEY="sk-ant-..."

# Windows PowerShell
$env:ANTHROPIC_API_KEY = "sk-ant-..."

Create .gitignore:

state.json
report.json
report-summary.txt
.env
__pycache__/
*.pyc

Module 1: State Management

The agent needs to track which URLs it has already audited. If the run is interrupted — power cut, keyboard interrupt, network error — it should resume from where it stopped, not start over.

state.py handles this with a flat JSON file:

import json
import os

STATE_FILE = os.path.join(os.path.dirname(__file__), "state.json")

_DEFAULT_STATE = {"audited": [], "pending": [], "needs_human": []}


def load_state() -> dict:
    if not os.path.exists(STATE_FILE):
        save_state(_DEFAULT_STATE.copy())
    with open(STATE_FILE, encoding="utf-8") as f:
        return json.load(f)


def save_state(state: dict) -> None:
    with open(STATE_FILE, "w", encoding="utf-8") as f:
        json.dump(state, f, indent=2)


def is_audited(url: str) -> bool:
    return url in load_state()["audited"]


def mark_audited(url: str) -> None:
    state = load_state()
    if url not in state["audited"]:
        state["audited"].append(url)
    save_state(state)


def add_to_needs_human(url: str) -> None:
    state = load_state()
    if url not in state["needs_human"]:
        state["needs_human"].append(url)
    save_state(state)

The design is intentional: mark_audited() is called immediately after a URL is processed and written to the report. If the agent crashes mid-run, it loses at most one URL's work.

Module 2: Browser Integration

browser.py does the actual page navigation. It uses Playwright directly (which Browser Use installs as a dependency) to open a visible Chromium window, navigate to the URL, capture HTTP status and redirect information, and extract the raw SEO signals from the DOM.

The key design decisions:

Visible browser, not headless. Set headless=False so you can watch the agent work. This matters for the demo and for debugging.

Status capture via response listener. Playwright raises an exception on 4xx/5xx responses, but the on("response", ...) handler fires before the exception. We capture status there.

2-second delay between visits. Prevents triggering rate limiting or bot detection on agency client sites.

Here is the core navigation function:

import asyncio
import sys
import time
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout

TIMEOUT = 20_000  # 20 seconds


def fetch_page(url: str) -> dict:
    result = {
        "final_url": url,
        "status_code": None,
        "title": None,
        "meta_description": None,
        "h1s": [],
        "canonical": None,
        "raw_links": [],
    }

    first_status = {"code": None}

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

        def on_response(response):
            if first_status["code"] is None:
                first_status["code"] = response.status

        page.on("response", on_response)

        try:
            page.goto(url, wait_until="domcontentloaded", timeout=TIMEOUT)
            result["status_code"] = first_status["code"] or 200
            result["final_url"] = page.url

            # Extract SEO signals from DOM
            result["title"] = page.title() or None
            result["meta_description"] = page.evaluate(
                "() => { const m = document.querySelector('meta[name=\"description\"]'); "
                "return m ? m.getAttribute('content') : null; }"
            )
            result["h1s"] = page.evaluate(
                "() => Array.from(document.querySelectorAll('h1')).map(h => h.innerText.trim())"
            )
            result["canonical"] = page.evaluate(
                "() => { const c = document.querySelector('link[rel=\"canonical\"]'); "
                "return c ? c.getAttribute('href') : null; }"
            )
            result["raw_links"] = page.evaluate(
                "() => Array.from(document.querySelectorAll('a[href]'))"
                ".map(a => a.href).filter(Boolean).slice(0, 100)"
            )

        except PlaywrightTimeout:
            result["status_code"] = first_status["code"] or 408
        except Exception as exc:
            print(f"[browser] Error: {exc}", file=sys.stderr)
            result["status_code"] = first_status["code"]
        finally:
            browser.close()

    time.sleep(2)
    return result

A few things worth noting:

The raw_links cap at 100 is deliberate. DEV.to profile pages have hundreds of links — you don't need all of them for broken link detection.

The wait_until="domcontentloaded" setting is faster than networkidle and sufficient for meta tag extraction. JavaScript-rendered content needs the DOM to be ready, not all network requests to complete.

Module 3: Claude Extraction Layer

extractor.py takes the raw page snapshot from browser.py and calls Claude to produce a structured SEO audit result.

This is where most tutorials go wrong. They either write complex parsing logic in Python (fragile) or ask Claude for a free-form response and try to parse prose (unreliable). The right approach: give Claude a strict JSON schema and tell it to return nothing else.

The prompt engineering that makes this reliable:

import json
import os
import sys
from datetime import datetime, timezone
import anthropic

MODEL = "claude-sonnet-4-20250514"
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))


def _strip_fences(text: str) -> str:
    """Remove accidental markdown code fences from Claude's response."""
    text = text.strip()
    if text.startswith("```"):
        lines = text.splitlines()
        # Drop opening fence
        lines = lines[1:] if lines[0].startswith("```") else lines
        # Drop closing fence
        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]
        text = "\n".join(lines).strip()
    return text


def extract(snapshot: dict) -> dict:
    if not os.environ.get("ANTHROPIC_API_KEY"):
        raise OSError("ANTHROPIC_API_KEY is not set.")

    prompt = f"""You are an SEO auditor. Analyze this page snapshot and return ONLY a JSON object.
No prose. No explanation. No markdown fences. Raw JSON only.

Page data:
- URL: {snapshot.get('final_url')}
- Status code: {snapshot.get('status_code')}
- Title: {snapshot.get('title')}
- Meta description: {snapshot.get('meta_description')}
- H1 tags: {snapshot.get('h1s')}
- Canonical: {snapshot.get('canonical')}

Return this exact schema:
{{
  "url": "string",
  "final_url": "string",
  "status_code": number,
  "title": {{"value": "string or null", "length": number, "status": "PASS or FAIL"}},
  "description": {{"value": "string or null", "length": number, "status": "PASS or FAIL"}},
  "h1": {{"count": number, "value": "string or null", "status": "PASS or FAIL"}},
  "canonical": {{"value": "string or null", "status": "PASS or FAIL"}},
  "flags": ["array of strings describing specific issues"],
  "human_review": false,
  "audited_at": "ISO timestamp"
}}

PASS/FAIL rules:
- title: FAIL if null or length > 60 characters
- description: FAIL if null or length > 160 characters  
- h1: FAIL if count is 0 (missing) or count > 1 (multiple)
- canonical: FAIL if null
- flags: list every failing field with a clear description
- audited_at: use current UTC time in ISO 8601 format"""

    response = client.messages.create(
        model=MODEL,
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}],
    )

    raw = response.content[0].text
    clean = _strip_fences(raw)

    try:
        return json.loads(clean)
    except json.JSONDecodeError as exc:
        print(f"[extractor] JSON parse error: {exc}", file=sys.stderr)
        return _error_result(snapshot, str(exc))


def _error_result(snapshot: dict, reason: str) -> dict:
    return {
        "url": snapshot.get("final_url", ""),
        "final_url": snapshot.get("final_url", ""),
        "status_code": snapshot.get("status_code"),
        "title": {"value": None, "length": 0, "status": "ERROR"},
        "description": {"value": None, "length": 0, "status": "ERROR"},
        "h1": {"count": 0, "value": None, "status": "ERROR"},
        "canonical": {"value": None, "status": "ERROR"},
        "flags": [f"Extraction error: {reason}"],
        "human_review": True,
        "audited_at": datetime.now(timezone.utc).isoformat(),
    }

Two things make this reliable in production:

First, _strip_fences() handles the case where Claude wraps its response in ```json fences despite being told not to. This happens occasionally with Sonnet and consistently breaks json.loads() if you don't handle it.

Second, the _error_result() fallback means the agent never crashes on a bad Claude response — it logs the error and marks the URL for human review, then continues to the next URL.

Cost: Claude Sonnet 4 is priced at \(3 per million input tokens and \)15 per million output tokens. A typical page snapshot is around 500 input tokens; the structured JSON response is around 300 output tokens. That works out to roughly \(0.006 per URL — about \)0.12 for a 20-URL audit.

Module 4: Broken Link Checker

linkchecker.py takes the raw_links list from the browser snapshot and checks same-domain links for broken status using async HEAD requests.

The design choices:

• Same-domain only. Checking every external link on a page would take minutes and isn't what agency clients need. Filter to links on the same domain as the page being audited.

• HEAD requests, not GET. Faster, lower bandwidth, sufficient for status code detection.

• Cap at 50 links. Pages like DEV.to article listings have hundreds of internal links. Checking all of them would dominate the runtime.

• Concurrent requests via asyncio. All links are checked in parallel, not sequentially.

import asyncio
import logging
from urllib.parse import urlparse
import httpx

CAP = 50
TIMEOUT = 5.0
logger = logging.getLogger(__name__)


def _same_domain(link: str, final_url: str) -> bool:
    if not link:
        return False
    lower = link.strip().lower()
    if lower.startswith(("#", "mailto:", "javascript:", "tel:", "data:")):
        return False
    try:
        page_host = urlparse(final_url).netloc.lower()
        parsed = urlparse(link)
        return parsed.scheme in ("http", "https") and parsed.netloc.lower() == page_host
    except Exception:
        return False


async def _check_link(client: httpx.AsyncClient, url: str) -> tuple[str, bool]:
    try:
        resp = await client.head(url, follow_redirects=True, timeout=TIMEOUT)
        return url, resp.status_code != 200
    except Exception:
        return url, True  # Timeout or connection error = broken


async def _run_checks(links: list[str]) -> list[str]:
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[_check_link(client, url) for url in links])
    return [url for url, broken in results if broken]


def check_links(raw_links: list[str], final_url: str) -> dict:
    same_domain = [l for l in raw_links if _same_domain(l, final_url)]

    capped = len(same_domain) > CAP
    if capped:
        logger.warning("Page has %d same-domain links — capping at %d.", len(same_domain), CAP)
        same_domain = same_domain[:CAP]

    broken = asyncio.run(_run_checks(same_domain))

    return {
        "broken": broken,
        "count": len(broken),
        "status": "FAIL" if broken else "PASS",
        "capped": capped,
    }

Module 5: Human-in-the-Loop

This is the part most automation tutorials skip. What happens when the agent hits a login wall? A page that returns 403? A URL that redirects to a "Subscribe to continue reading" page?

Most scripts either crash or silently skip. Neither is acceptable in an agency context.

hitl.py handles this with two functions: one that detects whether a pause is needed, and one that handles the pause itself.

from state import add_to_needs_human

LOGIN_KEYWORDS = {"login", "sign in", "sign-in", "access denied", "log in", "unauthorized"}
REDIRECT_CODES = {301, 302, 307, 308}


def should_pause(snapshot: dict) -> bool:
    code = snapshot.get("status_code")

    # Navigation failed entirely
    if code is None:
        return True

    # Non-200, non-redirect
    if code != 200 and code not in REDIRECT_CODES:
        return True

    # Login wall detection
    title = (snapshot.get("title") or "").lower()
    h1s = [h.lower() for h in (snapshot.get("h1s") or [])]

    if any(kw in title for kw in LOGIN_KEYWORDS):
        return True
    if any(kw in h1 for kw in LOGIN_KEYWORDS for h1 in h1s):
        return True

    return False


def pause_reason(snapshot: dict) -> str:
    code = snapshot.get("status_code")
    if code is None:
        return "Navigation failed (None status)"
    if code != 200 and code not in REDIRECT_CODES:
        return f"Unexpected status code: {code}"
    return "Possible login wall detected"


def pause_and_prompt(url: str, reason: str) -> str:
    print(f"\n⚠️  HUMAN REVIEW NEEDED")
    print(f"   URL:    {url}")
    print(f"   Reason: {reason}")
    print(f"   Options: [s] skip  [r] retry  [q] quit\n")

    while True:
        choice = input("Your choice: ").strip().lower()
        if choice in ("s", "r", "q"):
            return {"s": "skip", "r": "retry", "q": "quit"}[choice]
        print("   Enter s, r, or q.")

The should_pause() function catches four cases: navigation failure, unexpected HTTP status, login keywords in the title, and login keywords in H1 tags. The login keyword check is what catches "Please sign in to continue" pages that return 200 but are effectively inaccessible.

In --auto mode (for scheduled runs), the main loop skips the pause_and_prompt() call and automatically handles these cases by logging the URL to needs_human[] in state and continuing.

Module 6: Report Writer

reporter.py writes results incrementally. This is important: results are written after each URL is audited, not batched at the end. If the run is interrupted, you don't lose completed work.

import json
import os
from datetime import datetime, timezone

REPORT_JSON = os.path.join(os.path.dirname(__file__), "report.json")
REPORT_TXT = os.path.join(os.path.dirname(__file__), "report-summary.txt")


def _load_report() -> list:
    if not os.path.exists(REPORT_JSON):
        return []
    with open(REPORT_JSON, encoding="utf-8") as f:
        return json.load(f)


def write_result(result: dict) -> None:
    """Append or update a result in report.json."""
    entries = _load_report()
    url = result.get("url", "")

    # Update existing entry if URL already present (handles retries)
    for i, entry in enumerate(entries):
        if entry.get("url") == url:
            entries[i] = result
            break
    else:
        entries.append(result)

    with open(REPORT_JSON, "w", encoding="utf-8") as f:
        json.dump(entries, f, indent=2, ensure_ascii=False)


def _is_overall_pass(result: dict) -> bool:
    fields = ["title", "description", "h1", "canonical"]
    for field in fields:
        if result.get(field, {}).get("status") not in ("PASS",):
            return False
    if result.get("broken_links", {}).get("status") == "FAIL":
        return False
    return True


def write_summary() -> None:
    entries = _load_report()
    passed = sum(1 for e in entries if _is_overall_pass(e))

    lines = []
    for entry in entries:
        overall = "PASS" if _is_overall_pass(entry) else "FAIL"
        failed_fields = [
            f for f in ["title", "description", "h1", "canonical", "broken_links"]
            if entry.get(f, {}).get("status") == "FAIL"
        ]
        suffix = f" [{', '.join(failed_fields)}]" if failed_fields else ""
        lines.append(f"{entry.get('url', 'unknown'):<60} | {overall}{suffix}")

    lines.append("")
    lines.append(f"{passed}/{len(entries)} URLs passed")

    with open(REPORT_TXT, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))

The deduplication in write_result() handles retries cleanly. If a URL is retried after a human reviews a login wall and authenticates, the new result replaces the old one rather than creating a duplicate entry.

Module 7: The Main Loop

index.py wires everything together. It reads the URL list, loads state, skips already-audited URLs, and runs the audit loop.

import csv
import os
import sys
import time
import argparse

from state import load_state, is_audited, mark_audited, add_to_needs_human
from browser import fetch_page
from extractor import extract
from linkchecker import check_links
from hitl import should_pause, pause_reason, pause_and_prompt
from reporter import write_result, write_summary

INPUT_CSV = os.path.join(os.path.dirname(__file__), "input.csv")


def read_urls(path: str) -> list[str]:
    with open(path, newline="", encoding="utf-8") as f:
        return [row["url"].strip() for row in csv.DictReader(f) if row.get("url", "").strip()]


def run(auto: bool = False):
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("Error: ANTHROPIC_API_KEY environment variable is not set.")
        sys.exit(1)

    urls = read_urls(INPUT_CSV)
    pending = [u for u in urls if not is_audited(u)]

    print(f"Starting audit: {len(pending)} pending, {len(urls) - len(pending)} already done.\n")

    total = len(urls)

    try:
        for i, url in enumerate(pending, start=1):
            position = urls.index(url) + 1
            print(f"[{position}/{total}] {url}", end=" -> ", flush=True)

            # Browser navigation
            snapshot = fetch_page(url)

            # Human-in-the-loop check
            if should_pause(snapshot):
                reason = pause_reason(snapshot)

                if auto:
                    print(f"AUTO-SKIPPED ({reason})")
                    add_to_needs_human(url)
                    mark_audited(url)
                    continue

                action = pause_and_prompt(url, reason)
                if action == "quit":
                    print("Exiting.")
                    break
                elif action == "skip":
                    add_to_needs_human(url)
                    mark_audited(url)
                    continue
                # "retry" falls through to re-fetch below
                snapshot = fetch_page(url)

            # Claude extraction
            result = extract(snapshot)

            # Broken link check
            links = check_links(snapshot.get("raw_links", []), snapshot.get("final_url", url))
            result["broken_links"] = links

            # Write result immediately
            write_result(result)
            mark_audited(url)

            overall = "PASS" if all(
                result.get(f, {}).get("status") == "PASS"
                for f in ["title", "description", "h1", "canonical"]
            ) and links["status"] == "PASS" else "FAIL"

            print(overall)

    except KeyboardInterrupt:
        print("\n\nInterrupted. Progress saved. Re-run to continue.")
        return

    write_summary()
    passed = sum(
        1 for e in [r for r in []]
        if all(e.get(f, {}).get("status") == "PASS" for f in ["title", "description", "h1", "canonical"])
    )
    print(f"\nAudit complete. Report saved to report.json and report-summary.txt")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--auto", action="store_true", help="Auto-skip URLs requiring human review")
    args = parser.parse_args()
    run(auto=args.auto)

The KeyboardInterrupt handler is the resume mechanism. When you press Ctrl+C, the handler prints a message and exits cleanly. Because mark_audited() is called after write_result() for each URL, the next run skips everything already processed.

Running the Agent

Interactive mode (pauses on edge cases):

python index.py

Auto mode (skips edge cases, adds to needs_human[]):

python index.py --auto

When it runs, you'll see the browser window open for each URL and the terminal print progress:

Starting audit: 7 pending, 0 already done.

[1/7] https://example.com -> PASS
[2/7] https://example.com/about -> FAIL
[3/7] https://example.com/contact -> AUTO-SKIPPED (Unexpected status code: 404)
...
Audit complete. Report saved to report.json and report-summary.txt

To resume after an interruption:

python index.py --auto
# Starting audit: 4 pending, 3 already done.

Scheduling for Agency Use

For recurring weekly audits, create a batch file and schedule it with Windows Task Scheduler.

Create run-audit.bat:

@echo off
set ANTHROPIC_API_KEY=your-key-here
cd /d C:\Users\yourname\Desktop\seo-agent
python index.py --auto

In Windows Task Scheduler:

1. Create a new Basic Task

2. Set the trigger to Weekly, Monday at 7:00 AM

3. Set the action to "Start a program"

4. Browse to your run-audit.bat file

Check report-summary.txt on Monday morning. URLs in needs_human[] in state.json need manual review — login walls, paywalls, or pages that returned unexpected status codes.

For macOS/Linux, use cron:

# Run every Monday at 7am
0 7 * * 1 cd /path/to/seo-agent && ANTHROPIC_API_KEY=your-key python index.py --auto

What the Results Look Like

I ran this agent against seven of my own published pages across Hashnode, freeCodeCamp, and DEV.to. Every single one failed.

https://hashnode.com/@dannwaneri                    | FAIL [h1]
https://freecodecamp.org/news/claude-code-skill     | FAIL [description]
https://freecodecamp.org/news/stop-letting-ai-guess | FAIL [description]
https://freecodecamp.org/news/rag-system-handbook   | FAIL [title, description]
https://freecodecamp.org/news/author/dannwaneri     | FAIL [description]
https://dev.to/dannwaneri/gatekeeping-panic         | FAIL [title]
https://dev.to/dannwaneri/production-rag-system     | FAIL [title]

0/7 URLs passed

The freeCodeCamp description issues are partly platform-level — freeCodeCamp's template sometimes truncates or omits meta descriptions for article listing pages. The DEV.to title issues are mine. Article titles that work as headlines often exceed 60 characters in the <title> tag.

A note on the 60-character title rule: this is a display threshold, not a ranking penalty. Google indexes titles of any length. The 60-character guideline reflects approximately how many characters fit in a desktop SERP result before truncation. Titles over 60 characters often still rank — they just get cut off in search results, which can hurt click-through rate. The agent flags display risk, not a ranking violation.

Next Steps

The agent as built handles the core SEO audit workflow. Obvious extensions:

• Performance metrics — add a Lighthouse or PageSpeed Insights API call per URL

• Structured data validation — check for JSON-LD schema markup and validate it

• Email delivery — send report-summary.txt via SMTP after the run completes

• Multi-client support — separate input.csv files per client, separate report directories

The full code including all seven modules is at dannwaneri/seo-agent. Clone it, add your URLs, and run it.

If you found this useful, I write about practical AI agent setups for developers and agencies at DEV.to/@dannwaneri. The DEV.to companion piece covers the design decisions behind the agent — why HITL matters, why Browser Use over scrapers, and what the audit results mean for your own published content.

Agent skills fix this. A skill is a markdown file that loads into Claude Code's context automatically when you need it. You write the workflow once. The agent follows it every time. And because skills follow an open standard, the same file works in Claude Code, GitHub Copilot, Cursor, and Gemini CLI.

This tutorial shows you how to build a skill from scratch. You will build a commit-message-writer — a skill that reads your staged changes and generates a structured commit message following the Conventional Commits standard. By the end, you will have a working skill installed and ready to use, and you will understand the structure well enough to build any skill you need.

Table of Contents

1. What an Agent Skill Is

2. How to Choose What to Build

3. How to Structure Your Skill

4. How to Write the Description

5. How to Write the Instructions

6. How to Build the commit-message-writer Skill

7. How to Install and Test Your Skill

8. How to Improve Your Skill Over Time

9. Where to Go Next

What an Agent Skill Is

A skill is a folder containing a SKILL.md file. That file has two parts: a YAML frontmatter block at the top, and a markdown body below it.

my-skill/
└── SKILL.md

The frontmatter tells the agent what the skill is called and when to use it. The body tells the agent what to do when it loads the skill. Here is the minimal structure:

---
name: my-skill
description: What this skill does and when to use it.
---
 
# My Skill
 
Instructions for the agent go here.

When you invoke a skill — either explicitly with /skill-name or by describing what you want — the agent reads the SKILL.md body and follows the instructions inside it. The frontmatter never reaches the agent's instructions. It's metadata the skill system uses to decide whether to load the skill at all.

How the Agent Decides to Load a Skill

This is the most important thing to understand before you write your first skill: the agent decides whether to load your skill based entirely on the description field.

Skills appear in Claude Code's context as a list of names and descriptions. When you make a request, the agent scans that list and loads any skill whose description matches what you're asking for. If the description is vague, the skill won't load when you need it. If the description is too narrow, it won't load for variations of the same request.

The instructions in the body only matter after the skill loads. Getting the description right is what determines whether the skill loads at all.

What Skills Are Not

Skills are instruction files. They cannot run code on their own — but they can instruct the agent to run code using its existing tools. They are not plugins, extensions, or packages. They have no runtime. They are markdown files the agent reads, like a recipe a chef follows.

How to Choose What to Build

The best skills share three properties.

1. They encode a repeatable workflow. If you do something differently every time, a skill won't help. If you follow the same steps every session — even if you explain them differently each time — that's a skill candidate.

2. They have a clear trigger. You should be able to finish the sentence "I need this skill when I want to...". If you can't finish that sentence in one clause, the workflow isn't scoped enough for a skill.

3. They produce a consistent output format. Skills that output in a fixed structure — a commit message, a code review, a spec — are easier to build and test than skills that produce open-ended prose.

Good candidates: commit messages, pull request descriptions, code reviews, changelog entries. Bad candidates: "help me think through this", "make this better" — too open-ended to encode in a skill.

For this tutorial, commit message generation is the right scope. The trigger is obvious (you want to commit), the workflow is defined (read staged changes, apply Conventional Commits format), and the output is structured (a commit message with a specific shape).

How to Structure Your Skill

Every skill starts as a single folder with a single file:

commit-message-writer/
└── SKILL.md

As skills grow, they can include additional files the agent loads as needed:

commit-message-writer/
├── SKILL.md          ← always loaded when skill triggers
└── references/
    └── examples.md   ← loaded only when the agent needs examples

The SKILL.md body should stay under 500 lines. If your instructions are growing beyond that, move supporting detail into a references/ subfolder and tell the agent when to read those files. This keeps the skill lean — the agent only loads what it needs.

For this tutorial, a single SKILL.md is enough.

How to Write the Description

The description field is the trigger condition. It determines when your skill loads and when it doesn't. Most skills fail not because the instructions are wrong, but because the description doesn't match how people actually ask for help.

Here is a weak description:

description: Generates commit messages.

This will undertrigger. "Generate a commit message" will load it. "Write a commit for my changes" probably won't. "Summarize my staged diff" definitely won't — even though all three are asking for the same thing.

Here is a stronger description:

description: Generates structured commit messages following the Conventional Commits standard. Use when you want to commit your changes and need a well-formatted message. Triggers on "write a commit message", "commit my changes", "summarize my staged diff", "what should my commit say", or any request to describe or document code changes for version control.

The pattern is: what the skill does + when to use it + specific trigger phrases. The trigger phrases cover the different ways a developer might ask for the same thing.

Two rules for descriptions:

Be specific about the output. "Generates commit messages" is vague. "Generates structured commit messages following the Conventional Commits standard" tells the agent and the user exactly what they'll get.

Be slightly pushy. The agent has a natural tendency to undertrigger skills — to handle requests itself rather than loading a skill. A description that explicitly lists trigger phrases counteracts this. You are not being redundant. You are training the trigger.

How to Write the Instructions

The body of SKILL.md is where you define what the agent does when the skill loads. Good instructions follow two principles.

Generate first, clarify second. The agent should produce output immediately rather than asking clarifying questions. If it needs to make assumptions, it should make them and flag them — not ask. Asking questions before producing output adds friction and loses the benefit of having a skill at all.

Define the output format explicitly. Don't say "write a good commit message." Say exactly what the structure is, what fields are required, what the character limits are. The more specific the output format, the more consistent the results.

Here is what weak instructions look like:

# Commit Message Writer
 
Look at the staged changes and write a commit message that describes what changed.

That will produce different results every time — different formats, different lengths, different conventions. It's not a skill. It's a prompt.

Here is what strong instructions look like:

# Commit Message Writer
 
Read the staged diff using `git diff --staged`. Generate a commit message
following the Conventional Commits standard.
 
Output format:
type(scope): short description under 72 characters
 
Body (if changes are non-trivial):
- What changed and why, not how
- One bullet per logical change
 
Footer (if applicable):
BREAKING CHANGE: description
Closes #issue-number

The agent knows exactly what to produce. The output will be consistent across sessions, across projects, and across agents that support the standard.

How to Build the commit-message-writer Skill

Now build it. Create the skill directory:

mkdir -p ~/.claude/skills/commit-message-writer

On Windows PowerShell:

Note: PowerShell uses backtick (`) for line continuation, not backslash.

New-Item -ItemType Directory -Force -Path "$HOME\.claude\skills\commit-message-writer"

Create the SKILL.md file inside that directory. Here is the complete content:

---
name: commit-message-writer
description: Generates structured commit messages following the Conventional Commits
  standard. Use when you want to commit your changes and need a well-formatted message.
  Triggers on "write a commit message", "commit my changes", "summarize my staged
  diff", "what should my commit say", or any request to describe or document staged
  changes for version control.
---
 
# commit-message-writer
 
You generate structured commit messages from staged git changes.
 
## How to invoke
 
Run `git diff --staged` to read the staged changes. If nothing is staged, tell the
user and suggest they run `git add` first.
 
Generate first. Do not ask clarifying questions before producing the commit message.
If you need to make assumptions about scope or type, make them and note them after
the output.
 
## Output format
 
~~~
type(scope): short description
 
[body — optional, include if changes are non-trivial]
 
[footer — optional]
~~~
 
**Type** — choose one:
- `feat` — a new feature
- `fix` — a bug fix
- `docs` — documentation changes only
- `refactor` — code change that neither fixes a bug nor adds a feature
- `test` — adding or updating tests
- `chore` — build process, tooling, or dependency updates
 
**Scope** — the module, file, or area affected. Use the directory name or component
name. Omit if the change spans the entire codebase.
 
**Short description** — imperative mood, under 72 characters, no period at the end.
"Add user authentication" not "Added user authentication" or "Adds user authentication."
 
**Body** — what changed and why, not how. One bullet per logical change. Skip if the
short description is self-explanatory.
 
**Footer** — include `BREAKING CHANGE:` if the commit breaks backward compatibility.
Include `Closes #N` if it resolves a GitHub issue.
 
## Quality rules
 
- Never use "updated", "changed", or "modified" in the short description — be specific
- Never write "various improvements" or "misc fixes" — name what improved
- If more than three files changed across unrelated concerns, flag it:
  "These changes may be better split into separate commits: [list concerns]"
- The short description must be under 72 characters — count before outputting
 
## Example output
 
Input: staged changes adding a rate limiter to an API endpoint
 
~~~
feat(api): add rate limiting to /query endpoint
 
- Limits requests to 100 per minute per IP using Cloudflare's rate limit binding
- Returns 429 with Retry-After header when limit is exceeded
- Adds rate limit configuration to wrangler.toml
 
Closes #47
~~~

Save that file. The skill is built.

How to Install and Test Your Skill

Verify the File Exists

cat ~/.claude/skills/commit-message-writer/SKILL.md

You should see the full SKILL.md content. If you get an error, check the directory path.

Test the Skill

Open Claude Code in any git repository that has staged changes. Type:

/commit-message-writer

The agent will read your staged diff and produce a commit message following the format you defined.

You can also trigger it naturally:

write a commit message for my staged changes

what should my commit say

summarize my diff for git

All three should load the skill and produce a structured commit message. If the skill doesn't trigger on natural language requests, the description needs more trigger phrases — see the improvement section below.

Test Edge Cases

Test these cases before relying on the skill in production:

# Stage nothing, then ask for a commit message
git add -p  # stage nothing
# In Claude Code: "write a commit message"
# Expected: skill tells you nothing is staged and suggests git add

# Stage changes across unrelated files
git add src/api.ts src/styles.css README.md
# In Claude Code: "write a commit message"  
# Expected: skill flags that commits may be better split

How to Improve Your Skill Over Time

The first version of any skill is a draft. You improve it by observing where it produces inconsistent or wrong output, then updating the instructions.

When the Skill Undertriggers

If you type "summarize my changes for git" and the skill doesn't load, add that phrase to the description's trigger list:

description: ... Triggers on "write a commit message", "commit my changes",
  "summarize my staged diff", "summarize my changes for git", ...

The description is your primary lever for fixing triggering problems.

When the Output Format Drifts

If the agent starts producing commit messages that don't match your format — wrong type, missing scope, body in the wrong style — the instructions need to be more explicit. Add a concrete example that shows the failure and the correct output:

## Common mistakes to avoid
 
Wrong: "Updated the authentication flow"
Right: "refactor(auth): simplify token validation logic"
 
Wrong: "Fixed bugs"
Right: "fix(api): handle null response from upstream service"

Concrete counterexamples are more effective than abstract rules.

When the Scope Grows

If you find yourself wanting the skill to handle related tasks — reviewing commit messages, generating changelogs, writing PR descriptions — resist the urge to add everything to one skill. Build separate skills. Each skill should do one thing well. The Agent Skills standard is designed for composition, not for monolithic instructions.

Where to Go Next

The commit-message-writer covers the core pattern. The same structure works for any repeatable workflow.

Pull request descriptions follow the same shape — read the diff, apply a structure, produce consistent output. The trigger phrases are different ("write a PR description", "summarize my branch for review") and the output format adds sections for motivation and testing, but the SKILL.md structure is identical.

Code review checklists work well as skills when your team has a standard review process. The trigger is "review this code" or "check this PR", and the instructions encode whatever your team actually checks — security concerns, test coverage, naming conventions.

The commit-message-writer is the simplest skill architecture — instructions only. As your skills grow more specialized, two other patterns become useful.

The first adds a references/ directory: the voice-humanizer skill loads a CORPUS.md file containing the author's published writing, which the agent reads when it needs to check output against a specific style. The second adds quality rules and structured output formats that make results stricter and more consistent — that's the pattern spec-writer uses to surface assumptions inline. Each is the same SKILL.md structure at a different level of complexity.

Start with instructions only. Add references when the agent needs external context. Add output format rules when consistency matters more than flexibility.

The Agent Skills standard is supported in Claude Code, GitHub Copilot in VS Code, Cursor, and Gemini CLI. A skill you build once installs across all of them. The install path differs by agent:

The SKILL.md format is the same across all of them.

The commit-message-writer you just built is a working skill. The next one will take less time. By the third, you will start seeing workflows you repeat and immediately think: that should be a skill.

That's the point.

The tool is spec-writer: a Claude Code skill that takes a vague feature request and generates a structured spec, technical plan, and task breakdown before a single line of code gets written.

The problem it solves is one most developers hit within their first week of using AI coding agents seriously: the agent writes confidently in the wrong direction and you pay for it twice, once in tokens, once in rewrites.

This tutorial shows you how to install spec-writer, how to invoke it on a real feature, and how to read the output so you can catch the assumptions that would have wasted your time.

Table of Contents

1. The Problem with Prompting Agents Directly

2. What Spec-Driven Development Is

3. How spec-writer Works

4. How to Install spec-writer

5. How to Write Your First Spec

6. How to Read the Output

7. How to Hand the Spec to Your Agent

8. Where to Go Next

The Problem with Prompting Agents Directly

Here is what happens when you skip the spec.

You have a feature in your head: "Add a way for users to export their data." You open Claude Code and describe it. The agent produces code. It looks right. You run it. It's mostly right – except it exports everything including soft-deleted records, it doesn't paginate, it times out on large accounts, and it has no authentication check on the export endpoint.

None of those things were in your prompt. The agent guessed, and it guessed plausibly – which is worse than guessing obviously wrong. You didn't notice until testing.

This is the fundamental problem with prompting agents directly on anything non-trivial: your prompt carries your conscious requirements, but every feature has a shadow of requirements you didn't think to state. And the agent fills that shadow with assumptions.

Most of the time, those assumptions are reasonable. Some of the time, they're wrong in ways that take hours to unravel.

The failure mode isn't hallucination. It's the agent being exactly as helpful as the prompt allowed, which wasn't helpful enough.

Spec-Driven Development addresses this directly. The methodology – documented extensively by practitioners like Julián Deangelis – argues that a written spec isn't documentation overhead. It's the mechanism that forces you to make decisions before the agent does.

What Spec-Driven Development Is

Spec-Driven Development is the practice of writing a structured specification before you write code or prompt an agent. The spec defines what the feature must do, what assumptions are being made, and what tasks the implementation breaks into.

The key insight is what a spec is for. A spec is not trying to replace code. It's trying to surface the decisions that would otherwise be invisible. The agent will make those decisions either way: with a spec, you make them first. Without a spec, you discover them during testing.

The strongest counterargument to SDD comes from Gabriella Gonzalez: a sufficiently detailed spec is just code. She's right that some specs devolve into pseudocode so specific they might as well be implementations.

But that's a spec written at the wrong level of abstraction. The goal is to name the decisions, not to pre-implement them. "Only authenticated users can trigger this export" is a decision. "Call verifyJWT(token) and return 401 if it fails" is implementation. The spec needs the first. The agent handles the second.

SDD has three levels:

1. Spec-First: write a spec before every feature and hand it to the agent as context. This is the entry point and the workflow this tutorial focuses on.

2. Spec-Anchored: the spec lives in the repository and evolves alongside the code. When requirements change, you update the spec and re-prompt the agent to realign.

3. Spec-as-Source: the spec is the primary artifact. Code is generated from it and considered disposable. This is the most ambitious level and the direction many teams are moving toward.

spec-writer gets you to Spec-First immediately, with no ceremony.

How spec-writer Works

spec-writer is a Claude Code skill – a markdown file that loads into the agent's context and changes how it responds when invoked.

The skill follows one rule: generate first, flag assumptions inline. Instead of asking you clarifying questions before producing output, it generates the full spec immediately and marks every decision it made without your explicit input using [ASSUMPTION: ...] tags. Then you correct what's wrong.

This is faster than Q&A because it makes the decisions visible in a form you can react to rather than anticipate.

The output has three sections in fixed order:

1. SPEC: the what. One-line purpose, user stories, requirements, edge cases, and acceptance criteria in Given/When/Then format.

2. PLAN: the how. Stack and architecture decisions, data model changes, API contracts, testing strategy, and security constraints.

3. TASKS: the breakdown. Ordered, self-contained tasks each completable in a single agent session, each with its own acceptance criteria.

After the three sections, the skill produces an Assumptions summary: every [ASSUMPTION: ...] from the output, ranked by impact. This is the part you review before handing anything to the agent.

The skill is compatible with GitHub Spec Kit and OpenSpec. If you use either framework, save the spec output to your .specify/ or openspec/changes/ directory and continue from there.

How to Install spec-writer

spec-writer uses the Agent Skills standard, which means the same SKILL.md file works across Claude Code, Cursor, GitHub Copilot, Gemini CLI, and any other agent that supports the standard. You install it once and it works everywhere.

Installation

Create the skills directory if it doesn't exist and clone the repo:

mkdir -p ~/.claude/skills
git clone https://github.com/dannwaneri/spec-writer.git ~/.claude/skills/spec-writer

On Windows PowerShell:

(Note: PowerShell uses backtick (`) for line continuation, not backslash.)

New-Item -ItemType Directory -Force -Path "$HOME\.claude\skills"
git clone https://github.com/dannwaneri/spec-writer.git "$HOME\.claude\skills\spec-writer"

That's the entire installation. No package to install, no configuration file to edit, no API key. The skill is a markdown file. The agent reads it.

Verification

Open Claude Code and type:

/spec-writer test

If the skill is installed correctly, the agent will read the SKILL.md and produce a spec structure – even for "test" as input. You'll see the three sections and the Assumptions summary. If nothing happens, confirm that the ~/.claude/skills/spec-writer/SKILL.md file exists.

How to Write Your First Spec

The invocation pattern is:

/spec-writer [your feature description]

The description can be as rough as you'd write in a Slack message. The skill's job is to structure it. You don't need to be precise. Being precise is what the spec is for.

A Real Example

Here is a real invocation from the Foundation project, a federated knowledge base built on Cloudflare Workers:

/spec-writer Add CLI capture for Claude Code sessions stored in ~/.claude/projects/

That's twelve words. No mention of the tech stack, no data model details, no authentication requirements, no error handling spec. All of that comes out in the spec.

Here is the abbreviated output:

SPEC

One-line purpose: Developers can capture Claude Code session logs from their local filesystem into Foundation before they are deleted after 30 days.

Users and use cases:

• As a developer using Claude Code daily, I want my coding sessions captured automatically so that architectural decisions and debugging patterns don't disappear after 30 days.

• As a Foundation user, I want CLI sessions and browser extension captures in the same knowledge base so that I have one place to query across all my AI interactions.

Acceptance criteria:

Given a developer with Claude Code sessions in ~/.claude/projects/
When they run the Foundation CLI capture command
Then a list of uncaptured sessions is displayed, sorted by most recent first
 
Given a session already in Foundation
When the developer runs capture again
Then that session does not appear in the review list
 
Given a malformed .jsonl file
When the capture command processes it
Then an error is logged for that file and processing continues on remaining sessions

PLAN

Stack and architecture:

CLI capture runs locally — it needs to read the local filesystem, which
Cloudflare Workers cannot do. [ASSUMPTION: CLI capture is a local Node.js
or Bun script that calls the Foundation API to insert sessions, rather
than a Worker itself]

API contracts:

• POST /api/sessions/import – accepts parsed session JSON, returns session ID

• Returns 409 if session already exists

• Triggers evaluator pipeline after successful insert

TASKS

## Task 1: .jsonl parser
 
**What to build:** A parser that reads a Claude Code .jsonl session file
and outputs Foundation's internal message format.
**Files likely affected:** cli/src/parsers/claudeCode.js (create)
**Acceptance criteria:**
1. Parses a valid .jsonl file into the correct message format
2. Handles malformed lines without crashing — logs and skips
3. Returns empty array for empty file
**Dependencies:** none

Assumptions to review:

1. CLI capture is a local script calling the Foundation API, not a Worker
   Impact: HIGH
   Correct this if: you want a purely serverless approach
 
2. Manual curation before capture, not automatic bulk import
   Impact: HIGH
   Correct this if: you want automatic background capture
 
3. Session ID from .jsonl filename is the deduplication key
   Impact: MEDIUM
   Correct this if: session IDs are stored differently in your schema
 
4. No sensitive data scrubbing in v1
   Impact: MEDIUM
   Correct this if: your sessions contain credentials or keys

Twelve words in, four decisions surfaced immediately – three of which had real architectural implications.

The third assumption ("Session ID from .jsonl filename is the deduplication key") is the one that would have caused the most subtle bug. The agent would have implemented deduplication based on the filename and it would have worked until a session was renamed. The spec caught it before a line of code was written.

How to Read the Output

The output is designed to be scanned for [ASSUMPTION: ...] tags first, read for the tasks second.

Reading the Assumptions

Every [ASSUMPTION: ...] tag marks a place where the agent filled in something you didn't specify. Your job is to go through the Assumptions summary and decide for each one:

• Correct: the assumption is right, leave it

• Override: the assumption is wrong, restate it and re-run the spec

• Defer: the assumption doesn't matter for this iteration, mark it and move on

The impact rating tells you which assumptions to fix before you start coding. HIGH-impact assumptions affect architecture or data model. If they're wrong, fixing them requires rework. LOW-impact assumptions affect behavior details that are easy to change later.

Reading the Acceptance Criteria

The acceptance criteria in Given/When/Then format are the most useful part of the spec for catching scope errors. Read each one and ask: is this actually what I want?

Criteria are binary by design. "Returns 401 when unauthenticated" is a criterion. "Works correctly" is not. If you find yourself reading a criterion and thinking "well, it depends", then that's a signal that the criterion is hiding an assumption. Restate it.

Reading the Tasks

The tasks are ordered and self-contained. Each task produces a verifiable change. Before you hand any task to an agent, check two things:

1. Does the task have all the context it needs? If a task says "follow the existing auth pattern" and you haven't pointed the agent at your auth code, it will guess.

2. Does the acceptance criteria match what you'd actually test? If the criteria are vague, tighten them before the agent sees the task.

How to Hand the Spec to Your Agent

The spec is context, not a prompt. When you start an agent session for a task, include the relevant spec sections alongside the task description.

For Task 1 from the example above, your agent session might open like this:

Context:
- This is a federated knowledge base built on Cloudflare Workers, D1, and Vectorize
- Sessions are stored in ~/.claude/projects/ as .jsonl files
- The API runs at https://<your-worker>.workers.dev
 
Spec:
[paste the SPEC and PLAN sections]
 
Task:
[paste Task 1]

The context block is just an example. Replace it with your own project's tech stack, file locations, and API URL. The point is to give the agent the same context a new team member would need on day one.

The agent now has requirements, architecture context, and a single scoped task with binary acceptance criteria. It cannot guess the deduplication key incorrectly because the spec already resolved that assumption. It cannot skip error handling because the acceptance criteria explicitly require it.

This is the workflow the spec is designed for. The spec doesn't replace the agent. Rather, it removes the decisions from the agent's hands and puts them in yours, before the work starts.

Saving the Spec for Later

If you want to move toward Spec-Anchored development – where the spec lives in the repository – save the output to a specs/ directory in your project:

# Create specs directory
mkdir -p specs
 
# Save your spec
# Paste the output into specs/cli-capture.md

When requirements change, update the spec and re-prompt the agent to realign the implementation. The spec becomes the source of truth, not the code comments.

Where to Go Next

Try it on your next feature before you write a line of code. The assumptions it flags will tell you something about your feature you hadn't consciously decided yet – and correcting the HIGH-impact ones before you hand anything to an agent is the whole point. Skipping that step is the same as prompting directly.

If your project is growing, move toward Spec-Anchored. Save specs in your repository under specs/. When a new contributor joins or an agent starts a session cold, the specs give them the decisions that got made without requiring them to reverse-engineer the code.

The strongest ongoing challenge to this workflow is Gabriella Gonzalez's argument that detailed specs become code. If your specs are getting implementation-specific, you've crossed a line. Pull back to decisions – "only authenticated users can trigger this" – and leave implementation to the agent. The spec's job is to name what the agent would have guessed wrong, not to write the feature in prose.

The Agent Skills standard now works across Claude Code, GitHub Copilot, Cursor, and Gemini CLI. The spec-writer repo is at github.com/dannwaneri/spec-writer.

The irony of spending 64% of a Claude budget building a token-efficiency tool is real. But the spec surfaced four decisions on a twelve-word prompt. The fourth one – the deduplication key assumption – would have produced a bug that worked perfectly until a session got renamed.

That's not a hallucination. That's the agent being exactly as helpful as the prompt allowed.

The spec is how you raise the ceiling on what "helpful" means.

This tutorial is different. I run a production RAG system (vectorize-mcp-worker) that handles real traffic at a total cost of \(5/month. The alternatives I evaluated ranged from \)100–$200/month. The difference isn't magic. It's architecture.

Here, you'll build rag-tutorial-simple: a clean, minimal RAG chatbot deployed on Cloudflare Workers. No external API keys. No paid vector database subscriptions. No servers to manage. Just Cloudflare's free tier – Workers, Vectorize, and Workers AI – doing the heavy lifting at the edge.

Table of Contents

1. What You Will Build

2. Prerequisites

3. How RAG Works

4. How to Set Up Your Project

5. How to Build the Data Pipeline

6. How to Build the Query Pipeline

7. How to Add Error Handling and Security

8. Performance and Cost Analysis

9. Conclusion

What You Will Build

By the end of this tutorial, you'll have a globally deployed RAG API that:

• Accepts a natural language question via HTTP

• Converts it to a vector embedding using Workers AI

• Searches a knowledge base stored in Cloudflare Vectorize

• Passes the retrieved context to an LLM (also on Workers AI) to generate an answer

• Returns a grounded, accurate response (not a hallucination)

The complete source code is available at github.com/dannwaneri/rag-tutorial-simple.

Prerequisites

This is an intermediate-level tutorial. You should be comfortable with:

• JavaScript/TypeScript: async/await, promises, basic types

• HTTP APIs: REST, request/response, JSON

• Command line basics: running npm commands, navigating directories

You will need:

• Node.js 18 or higher: check with node --version

• A Cloudflare account: free tier is fine, sign up at cloudflare.com

• A code editor: VS Code recommended for TypeScript support

That's it. No OpenAI key. No credit card for embeddings. Let's build.

How RAG Works

Before you write any code, you'll need a clear mental model of what you're building. This section explains the three core components of a RAG system, how data flows between them, and why this architecture works at scale.

The Mental Model

Think of a traditional LLM like a doctor who studied medicine for years but has been in a remote cabin with no internet since their graduation day. They are brilliant, but they only know what they knew when they left. Ask them about a drug approved last year and they'll either say they don't know or – worse – confidently give you wrong information.

RAG gives that doctor access to an up-to-date medical library. Before answering your question, they can look up the relevant pages, read them, and use that information to give you an accurate answer. Their training still matters (that is, they know how to read and interpret the information), but they're no longer limited to what they memorized years ago.

In technical terms, RAG works in three steps on every request:

1. Retrieve: find the most relevant documents from your knowledge base

2. Augment: add those documents to the LLM prompt as context

3. Generate: let the LLM produce an answer using both its training and the retrieved context

The Three Components

Every RAG system has three moving parts. Understanding each one will help you debug problems and make better architectural decisions as you build.

The Embedding Model

An embedding model converts text into a vector – an array of numbers that represents the meaning of that text. The model you will use in this tutorial, @cf/baai/bge-base-en-v1.5, outputs 768 numbers for any piece of text you give it.

The critical property of embeddings is that semantically similar text produces numerically similar vectors. "How do I install Node.js?" and "What's the process for setting up Node?" will produce vectors that are close together. "How do I install Node.js?" and "What is the capital of France?" will produce vectors that are far apart.

This is what makes semantic search possible. You aren't matching keywords, you're matching meaning.

One rule you must never break: your documents and your queries must be embedded with the same model. If you embed your documents with bge-base-en-v1.5 and your queries with a different model, the vectors won't be comparable and your searches will return garbage.

The Vector Database

The vector database stores your embeddings and lets you search them by similarity. In this tutorial, you'll use Cloudflare Vectorize.

When you run a similarity search, you pass in a query vector and Vectorize returns the K most similar vectors it has stored, along with their metadata and similarity scores. This is called approximate nearest neighbor search, and Vectorize is optimized to do it fast even across millions of vectors.

The key advantage of using Vectorize over an external vector database like Pinecone is co-location. Vectorize runs in the same Cloudflare network as your Worker. There's no external API call, no authentication roundtrip, and no network latency between your application and your database.

The Language Model

The LLM is responsible for one thing: reading the retrieved context and generating a natural language answer. It doesn't search anything. It doesn't decide what's relevant. It just reads what you give it and writes a response.

This separation of concerns is intentional. The LLM is good at language: understanding questions, synthesizing information, writing clearly. The vector database is good at retrieval: finding relevant documents fast. RAG combines their strengths without asking either component to do something it is not designed for.

In this tutorial you'll use @cf/meta/llama-3.3-70b-instruct-fp8-fast through Workers AI. No API key required.

A Note on Visual Embeddings

If you plan to extend this system to search images, you may be tempted to use a vision-language model like CLIP to generate visual embeddings (vectors that represent the image itself rather than a text description of it). This sounds clever but works worse for RAG in practice.

Visual embeddings match pixel similarity. They are good for "find images that look like this one." They are poor for "find the login screen" or "find dashboards showing error rates" because those queries are about meaning, not pixels.

The better approach – used in production – is to pass the image through a multimodal model like Llama 4 Scout, which generates a detailed text description and extracts visible text via OCR. You then embed that description using the same BGE model as your other documents.

The result lives in one unified index, works with your existing query pipeline, and produces better search results than visual embeddings for RAG use cases.

Cloudflare Workers AI does not support CLIP anyway. But even if it did, descriptions would outperform it for semantic search.

How a Query Flows Through the System

Here is exactly what happens when a user sends the question "What is RAG?" to your finished Worker:

1. Step 1 – Embed the question (20-30ms): Your Worker calls Workers AI with the question text. The embedding model returns a 768-dimensional vector representing the meaning of the question.

2. Step 2 – Search Vectorize (30-50ms): Your Worker passes that vector to Vectorize, which searches your knowledge base and returns the 3 most similar documents with their similarity scores.

3. Step 3 – Filter and build context (< 1ms): Documents with a similarity score below 0.5 are discarded. The remaining document texts are joined into a context string.

4. Step 4 – Generate the answer (500-1500ms): Your Worker sends the context and the question to the LLM. The LLM reads the context and generates a grounded answer.

5. Step 5 – Return to the user: The answer and source metadata are returned as JSON.

Total time: typically 600-1600ms end to end. The LLM generation step dominates. Everything else is fast.

Why This Works at Scale

A common objection to Cloudflare RAG is that it cannot meet sub-200ms retrieval requirements. That objection comes from a specific architectural mistake: trying to run the entire RAG pipeline, including heavy embedding generation and reranking, inside a single synchronous request. That's the wrong architecture.

The architecture you're building in this tutorial separates the loading step (which is slow and runs once) from the query step (which is fast and runs on every request). By the time a user asks a question, your documents are already embedded and stored. The query pipeline only needs to embed the question, run one vector search, and call the LLM. Those three steps are fast.

My production system (vectorize-mcp-worker) runs this architecture and handles real traffic at $5/month. The full performance breakdown is here. Cloudflare RAG works. You just have to build it correctly.

How to Set Up Your Project

In this section, you'll scaffold a Cloudflare Worker, create a Vectorize index to store your embeddings, and configure the bindings that connect them together.

How to Create the Project

Open your terminal and create a new directory for the project.

On Mac/Linux:

mkdir rag-tutorial-simple && cd rag-tutorial-simple

On Windows PowerShell:

mkdir rag-tutorial-simple
cd rag-tutorial-simple

Then run the Cloudflare scaffolding tool:

npm create cloudflare@latest

Answer the prompts like this:

• Directory/app name: rag-tutorial-simple

• What would you like to start with? Hello World example

• TypeScript? Yes

• Deploy? No

When it finishes, you'll have a working TypeScript Worker with Wrangler already configured.

How to Create the Vectorize Index

Vectorize is Cloudflare's vector database. It lives in the same network as your Worker, which means no external API call and no added latency when you search it.

npx wrangler vectorize create rag-tutorial-index --dimensions=768 --metric=cosine

Two things to note here.

--dimensions=768 tells Vectorize how many numbers make up each embedding. This must match the output of the embedding model you use. The model you will use (@cf/baai/bge-base-en-v1.5) outputs 768 dimensions. If this number doesn't match, your searches will fail.

--metric=cosine is how Vectorize measures similarity between vectors. Cosine similarity measures the angle between two vectors rather than the distance between them. For text embeddings, this captures semantic meaning more accurately than other metrics.

How to Configure wrangler.toml

Open wrangler.toml and replace its contents with the following:

name = "rag-tutorial-simple"
main = "src/index.ts"
compatibility_date = "2026-02-25"

[[vectorize]]
binding = "VECTORIZE"
index_name = "rag-tutorial-index"

[ai]
binding = "AI"

The [[vectorize]] block connects your Worker to the index you just created. The [ai] block gives your Worker access to Workers AI – both for generating embeddings and for running the language model that produces answers.

Notice that there are no API keys anywhere. Cloudflare handles authentication internally because everything – your Worker, Vectorize, and Workers AI – runs under the same account.

How to Update src/index.ts

Open src/index.ts and replace the generated code with this:

export interface Env {
  VECTORIZE: VectorizeIndex;
  AI: Ai;
  LOAD_SECRET: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    return new Response("RAG tutorial worker is running", { status: 200 });
  },
};

The Env interface tells TypeScript what bindings are available inside your Worker. VectorizeIndex and Ai are types provided by Cloudflare's type definitions.

How to Verify Your Setup

Start the local development server:

npx wrangler dev

Open your browser and visit http://localhost:8787. You should see:

RAG tutorial worker is running

You will see two warnings in your terminal. Both are expected.

The first warning says that Vectorize doesn't support local mode. This means Vectorize queries won't work during local development unless you run with the --remote flag. You'll do this later when testing the full pipeline.

The second warning says the AI binding always accesses remote resources. This means that embedding generation and LLM calls always hit Cloudflare's servers, even in local development. This is fine: usage within the free tier limits costs nothing.

Your project structure at this point:

rag-tutorial-simple/
├── scripts/
│   └── knowledge-base.ts
├── src/
│   └── index.ts
├── wrangler.toml
├── package.json
└── tsconfig.json

How to Build the Data Pipeline

The data pipeline is responsible for two things: generating embeddings for each document in your knowledge base, and storing those embeddings in Vectorize. You'll handle both steps inside the Worker itself using a /load endpoint.

This approach has a key advantage: you don't need an API token, an Account ID, or any external tooling. Everything uses the bindings you already configured in wrangler.toml.

How to Create the Knowledge Base

Create a scripts/ folder in your project and add a file called knowledge-base.ts:

mkdir scripts

Add your documents to scripts/knowledge-base.ts:

export const documents = [
  {
    id: "1",
    text: "Cloudflare Workers run JavaScript at the edge, in over 300 data centers worldwide. Requests are handled close to the user, reducing latency significantly compared to a single-region server.",
    metadata: { source: "cloudflare-docs", category: "workers" },
  },
  {
    id: "2",
    text: "Vectorize is Cloudflare's vector database. It stores embeddings and lets you search them by semantic similarity. It runs in the same network as your Worker, so there is no external API call needed.",
    metadata: { source: "cloudflare-docs", category: "vectorize" },
  },
  {
    id: "3",
    text: "Workers AI lets you run machine learning models directly on Cloudflare's infrastructure. You can generate embeddings and run LLM inference without leaving the Cloudflare network.",
    metadata: { source: "cloudflare-docs", category: "workers-ai" },
  },
  {
    id: "4",
    text: "RAG stands for Retrieval Augmented Generation. Instead of relying only on what the LLM was trained on, RAG retrieves relevant context from a knowledge base and adds it to the prompt before generating an answer.",
    metadata: { source: "ai-concepts", category: "rag" },
  },
  {
    id: "5",
    text: "An embedding is a numerical representation of text. Similar pieces of text produce similar embeddings. This is what makes semantic search possible — you search by meaning, not exact keywords.",
    metadata: { source: "ai-concepts", category: "embeddings" },
  },
  {
    id: "6",
    text: "The BGE model (bge-base-en-v1.5) is available through Workers AI. It generates 768-dimensional embeddings and works well for English semantic search tasks.",
    metadata: { source: "cloudflare-docs", category: "workers-ai" },
  },
  {
    id: "7",
    text: "Cosine similarity measures the angle between two vectors. For text embeddings, it captures semantic similarity regardless of text length, which makes it more reliable than Euclidean distance.",
    metadata: { source: "ai-concepts", category: "embeddings" },
  },
  {
    id: "8",
    text: "Cloudflare Workers have a free tier that includes 100,000 requests per day. Vectorize is available on both the Workers Free and Paid plans. The free tier lets you prototype and experiment. The Workers Paid plan starts at $5/month and includes higher usage allocations for production workloads.",
    metadata: { source: "cloudflare-docs", category: "pricing" },
  },
];

Each document has three fields. The id is a unique string that Vectorize uses to identify the vector. The text is what gets converted into an embedding. The metadata is stored alongside the vector and returned in search results. You'll use it later to display the source of each answer.

Understanding Embeddings

Before writing the loading code, it helps to understand what you're actually generating.

An embedding is an array of 768 numbers that represents the meaning of a piece of text. The model reads a sentence and outputs those 768 numbers in a way where similar sentences produce similar arrays of numbers.

When a user asks a question, you convert that question into an embedding using the same model, then ask Vectorize to find the stored embeddings that are closest to it. The documents those embeddings came from are your most relevant context.

This is why the model choice matters: your documents and your queries must be embedded with the same model, or the similarity scores will be meaningless.

How to Build the Load Endpoint

Open src/index.ts and update it with a /load route. Here is the complete file at this stage:

import { documents } from "../scripts/knowledge-base";

export interface Env {
  VECTORIZE: VectorizeIndex;
  AI: Ai;
  LOAD_SECRET: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === "/load" && request.method === "POST") {
      return handleLoad(env, request);
    }

    return new Response("RAG tutorial worker is running", { status: 200 });
  },
};

async function handleLoad(env: Env, request: Request): Promise<Response> {
  const authHeader = request.headers.get("X-Load-Secret");
  if (authHeader !== env.LOAD_SECRET) {
    return Response.json({ error: "Unauthorized" }, { status: 401 });
  }

  const results: { id: string; status: string }[] = [];

  for (const doc of documents) {
    const response = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
      text: [doc.text],
    }) as { data: number[][] };

    await env.VECTORIZE.upsert([
      {
        id: doc.id,
        values: response.data[0],
        metadata: {
          ...doc.metadata,
          text: doc.text,
        },
      },
    ]);

    results.push({ id: doc.id, status: "loaded" });
  }

  return Response.json({ success: true, loaded: results });
}

Notice that env.AI.run() and env.VECTORIZE.upsert() require no credentials. The bindings handle authentication because the Worker runs inside your Cloudflare account. There are no secrets to manage for internal service communication.

The text: doc.text field inside metadata is important. Vectorize stores the vector values and whatever metadata you provide, but it doesn't store the original text separately. By including the text in metadata, you can retrieve and display it in search results later.

The as { data: number[][] } cast is necessary because the TypeScript type definitions for Workers AI do not yet reflect the exact return shape of every model. The actual response always contains a data array, and the cast tells TypeScript to trust that.

How to Deploy and Load Your Knowledge Base

First, set the secret that will protect your load endpoint:

npx wrangler secret put LOAD_SECRET

Type a strong value when prompted. Then deploy:

npx wrangler deploy

Trigger the load endpoint. You only need to do this once, or any time you update your knowledge base:

curl -X POST https://rag-tutorial-simple.<your-subdomain>.workers.dev/load \
  -H "X-Load-Secret: your-secret-value"

On Windows PowerShell:

Note: PowerShell uses backtick (`) for line continuation, not backslash.

Invoke-WebRequest `
  -Uri "https://rag-tutorial-simple.<your-subdomain>.workers.dev/load" `
  -Method POST `
  -Headers @{"X-Load-Secret"="your-secret-value"} `
  -UseBasicParsing

You should see:

{
  "success": true,
  "loaded": [
    { "id": "1", "status": "loaded" },
    { "id": "2", "status": "loaded" },
    { "id": "3", "status": "loaded" },
    { "id": "4", "status": "loaded" },
    { "id": "5", "status": "loaded" },
    { "id": "6", "status": "loaded" },
    { "id": "7", "status": "loaded" },
    { "id": "8", "status": "loaded" }
  ]
}

Your knowledge base is now stored in Vectorize as vectors. In the next section, you'll build the query pipeline that searches those vectors and generates answers.

How to Build the Query Pipeline

The query pipeline is the core of your RAG system. When a user sends a question, the pipeline runs four steps in sequence: embed the question, search Vectorize, build context from the results, and generate an answer with the LLM.

Add a /query route to your fetch handler and the complete handleQuery function. Here is the full updated src/index.ts:

import { documents } from "../scripts/knowledge-base";

export interface Env {
  VECTORIZE: VectorizeIndex;
  AI: Ai;
  LOAD_SECRET: string;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === "/load" && request.method === "POST") {
      return handleLoad(env, request);
    }

    if (url.pathname === "/query" && request.method === "POST") {
      return handleQuery(request, env);
    }

    return new Response("RAG tutorial worker is running", { status: 200 });
  },
};

async function handleLoad(env: Env, request: Request): Promise<Response> {
  const authHeader = request.headers.get("X-Load-Secret");
  if (authHeader !== env.LOAD_SECRET) {
    return Response.json({ error: "Unauthorized" }, { status: 401 });
  }

  const results: { id: string; status: string }[] = [];

  for (const doc of documents) {
    const response = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
      text: [doc.text],
    }) as { data: number[][] };

    await env.VECTORIZE.upsert([
      {
        id: doc.id,
        values: response.data[0],
        metadata: {
          ...doc.metadata,
          text: doc.text,
        },
      },
    ]);

    results.push({ id: doc.id, status: "loaded" });
  }

  return Response.json({ success: true, loaded: results });
}

async function handleQuery(request: Request, env: Env): Promise<Response> {
  const body = await request.json() as { question: string };

  if (!body.question) {
    return Response.json({ error: "question is required" }, { status: 400 });
  }

  // Step 1: Embed the question using the same model as your documents
  const embeddingResponse = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
    text: [body.question],
  }) as { data: number[][] };

  // Step 2: Search Vectorize for the 3 most similar documents
  const searchResults = await env.VECTORIZE.query(
    embeddingResponse.data[0],
    {
      topK: 3,
      returnMetadata: "all",
    }
  );

  // Step 3: Build context from results above the similarity threshold
  const context = searchResults.matches
    .filter((match) => match.score > 0.5)
    .map((match) => match.metadata?.text as string)
    .filter(Boolean)
    .join("\n\n");

  if (!context) {
    return Response.json({
      answer: "I could not find relevant information to answer that question.",
      sources: [],
    });
  }

  // Step 4: Generate an answer using the retrieved context
  const aiResponse = await env.AI.run("@cf/meta/llama-3.3-70b-instruct-fp8-fast", {
    messages: [
      {
        role: "system",
        content: "You are a helpful assistant. Answer the question using only the context provided. If the context does not contain enough information, say so.",
      },
      {
        role: "user",
        content: `Context:\n\({context}\n\nQuestion: \){body.question}`,
      },
    ],
    max_tokens: 256,
  }) as { response: string };

  // Step 5: Return the answer with its sources
  const sources = searchResults.matches
    .filter((match) => match.score > 0.5)
    .map((match) => match.metadata?.source as string)
    .filter(Boolean);

  return Response.json({
    answer: aiResponse.response,
    sources: [...new Set(sources)],
  });
}

What each step does:

1. Step 1 – Embed the question: You convert the user's question into a 768-dimensional vector using the same model you used when loading your documents. This is critical: the question and the documents must be embedded with the same model or the similarity scores will be meaningless.

2. Step 2 – Search Vectorize: You pass the question embedding to Vectorize, which returns the three most similar documents. returnMetadata: "all" tells Vectorize to include the metadata you stored alongside each vector — including the original text.

3. Step 3 – Build context: You filter out any results with a similarity score below 0.5 and join the remaining document texts into a single context string. The 0.5 threshold prevents the LLM from receiving irrelevant documents just because nothing better matched.

4. Step 4 – Generate the answer: You pass the context and the question to the LLM using the chat format with messages. The system prompt explicitly instructs the model to answer using only the provided context. This is what keeps the LLM grounded. Without this instruction, it will ignore your context and answer from its training data instead.

5. Step 5 – Return sources: You include the source metadata in the response so callers know which documents the answer came from. The Set deduplicates sources in case multiple chunks came from the same document.

How to Test the Query Pipeline

Deploy your Worker:

npx wrangler deploy

Send a question:

curl -X POST https://rag-tutorial-simple.<your-subdomain>.workers.dev/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is RAG?"}'

On Windows PowerShell:

Invoke-WebRequest `
  -Uri "https://rag-tutorial-simple.<your-subdomain>.workers.dev/query" `
  -Method POST `
  -ContentType "application/json" `
  -Body '{"question": "What is RAG?"}' `
  -UseBasicParsing

You should receive a response like this:

{
  "answer": "RAG stands for Retrieval Augmented Generation. It's a method that enhances generation by retrieving relevant context from a knowledge base and adding it to the prompt before generating an answer.",
  "sources": ["ai-concepts"]
}

The answer came from your knowledge base, not from the LLM's training data. That's the entire point of RAG: grounded, verifiable answers with traceable sources.

How to Add Error Handling and Security

A tutorial that only shows the happy path is not production-ready. In this section, you'll add error handling to every step of the query pipeline and protect the /load endpoint from unauthorized access.

How to Secure the Load Endpoint

The /load endpoint generates embeddings and writes to your Vectorize index. Without protection, anyone who discovers your Worker URL can trigger it repeatedly, consuming your Workers AI quota and overwriting your data.

The LOAD_SECRET binding you added to Env and the wrangler secret put command you ran earlier handle this. The check at the top of handleLoad rejects any request that doesn't include the correct secret header:

const authHeader = request.headers.get("X-Load-Secret");
if (authHeader !== env.LOAD_SECRET) {
  return Response.json({ error: "Unauthorized" }, { status: 401 });
}

A request without the header returns {"error":"Unauthorized"} with a 401 status. The secret itself is stored as an encrypted environment variable in your Worker. It never appears in your code or wrangler.toml.

To trigger the load endpoint, you must include the secret in the request header:

curl -X POST https://rag-tutorial-simple.<your-subdomain>.workers.dev/load \
  -H "X-Load-Secret: your-secret-value"

How to Handle Query Errors

Replace your handleQuery function with this hardened version:

async function handleQuery(request: Request, env: Env): Promise<Response> {
  // Guard against malformed request body
  let body: { question: string };
  try {
    body = await request.json() as { question: string };
  } catch {
    return Response.json({ error: "Invalid JSON in request body" }, { status: 400 });
  }

  if (!body.question || typeof body.question !== "string" || body.question.trim() === "") {
    return Response.json({ error: "question must be a non-empty string" }, { status: 400 });
  }

  // Sanitize: trim whitespace and cap length
  const question = body.question.trim().slice(0, 500);

  // Step 1: Embed the question
  let embeddingResponse: { data: number[][] };
  try {
    embeddingResponse = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
      text: [question],
    }) as { data: number[][] };
  } catch (err) {
    console.error("Embedding generation failed:", err);
    return Response.json({ error: "Failed to process your question" }, { status: 503 });
  }

  // Step 2: Search Vectorize
  let searchResults: Awaited<ReturnType<typeof env.VECTORIZE.query>>;
  try {
    searchResults = await env.VECTORIZE.query(
      embeddingResponse.data[0],
      { topK: 3, returnMetadata: "all" }
    );
  } catch (err) {
    console.error("Vectorize query failed:", err);
    return Response.json({ error: "Failed to search knowledge base" }, { status: 503 });
  }

  // Step 3: Build context
  const context = searchResults.matches
    .filter((match) => match.score > 0.5)
    .map((match) => match.metadata?.text as string)
    .filter(Boolean)
    .join("\n\n");

  if (!context) {
    return Response.json({
      answer: "I could not find relevant information to answer that question. Try rephrasing or asking something else.",
      sources: [],
    });
  }

  // Step 4: Generate answer
  let aiResponse: { response: string };
  try {
    aiResponse = await env.AI.run("@cf/meta/llama-3.3-70b-instruct-fp8-fast", {
      messages: [
        {
          role: "system",
          content: "You are a helpful assistant. Answer the question using only the context provided. If the context does not contain enough information, say so.",
        },
        {
          role: "user",
          content: `Context:\n\({context}\n\nQuestion: \){question}`,
        },
      ],
      max_tokens: 256,
    }) as { response: string };
  } catch (err) {
    console.error("LLM generation failed:", err);
    return Response.json({ error: "Failed to generate an answer" }, { status: 503 });
  }

  // Step 5: Return answer with sources
  const sources = searchResults.matches
    .filter((match) => match.score > 0.5)
    .map((match) => match.metadata?.source as string)
    .filter(Boolean);

  return Response.json({
    answer: aiResponse.response,
    sources: [...new Set(sources)],
  });
}

What each error handling decision means:

• try/catch around request.json(): request.json() throws if the body is not valid JSON. Without this catch, a malformed request crashes your Worker with an unhandled 500 error. With it, the caller gets a clear 400 explaining what went wrong.

• Input validation before processing: You check that question exists, is a string, and is not empty before calling any external service. This prevents wasted AI calls on invalid input.

• .slice(0, 500) on the question: This caps the input length before it reaches the embedding model. Without it, a malicious caller could send a very long string designed to inflate your AI usage or hit Workers CPU limits.

• 503 for AI and Vectorize failures: HTTP 503 means "service temporarily unavailable." It signals to callers that the error is on the server side and the request can be retried.

• .filter(Boolean) on context: After mapping match.metadata?.text, some results may be undefined if metadata was stored without a text field. This filters them out before joining, preventing "undefined" from appearing in the context string you send to the LLM.

How to Test Error Handling

Deploy your updated Worker:

npx wrangler deploy

Test each error case:

# Missing secret on load endpoint — should return 401
curl -X POST https://rag-tutorial-simple.<your-subdomain>.workers.dev/load

# Invalid JSON — should return 400
curl -X POST https://rag-tutorial-simple.<your-subdomain>.workers.dev/query \
  -H "Content-Type: application/json" \
  -d 'not json'

# Empty question — should return 400
curl -X POST https://rag-tutorial-simple.<your-subdomain>.workers.dev/query \
  -H "Content-Type: application/json" \
  -d '{"question": ""}'

Performance and Cost Analysis

This section uses real production data from my vectorize-mcp-worker deployment. It uses the same architecture you just built, measured from Port Harcourt, Nigeria to Cloudflare's edge.

Real Performance Numbers

Here is what the pipeline actually costs in time on every request:

This covers embedding generation and vector search only – the retrieval layer. LLM generation adds 500-1500ms on top, which is why end-to-end response time typically runs 600-1600ms.

The embedding step and vector search dominate. Everything else is negligible. For context, a comparable setup using OpenAI embeddings and Pinecone would add two external API roundtrips on top of this, easily pushing total latency past 1 second.

These numbers come from a single-region measurement. Your actual latency will vary based on your location and Cloudflare's load at the time of the request. The architectural point holds regardless: co-locating everything on the edge eliminates inter-service network hops, which is where most latency in traditional RAG stacks comes from.

Real Cost Breakdown

For 10,000 searches per day (300,000 per month) with 10,000 stored vectors:

This stack:

Traditional alternatives for the same volume:

That is an 85-95% cost reduction depending on which alternative you compare against. For a bootstrapped startup adding semantic search, that difference is $1,500-2,000 per year.

Why the Cost Difference Is So Large

Traditional RAG stacks have three cost problems that compound each other.

The first is idle compute. A dedicated server or container running your embedding service costs money even when no searches are happening. Cloudflare Workers charge only for actual execution time.

The second is inter-service data transfer. Every time your application calls an external service for an embedding, then calls a separate service for a search, you're paying for two external API calls with metered pricing. In this stack, both operations happen inside Cloudflare's network at no additional transfer cost.

The third is minimum plan pricing. Pinecone's Standard plan costs \(50/month as a floor, regardless of how little you use it. Cloudflare's pricing scales from the \)5/month Workers Paid plan base.

When the Included Allocation Is Enough

For smaller usage levels, you may not pay beyond the $5/month Workers Paid base price:

• Workers: 10 million requests per month included

• Workers AI: generous daily neuron allocation included

• Vectorize: available on both Free and Paid plans, with a free allocation included

A side project, internal tool, or small business with under 3,000 searches per day will likely stay within the included allocations entirely.

The Trade-off to Know About

This cost advantage comes with one operational constraint worth understanding before you build: Vectorize does not work in local development mode.

When you run wrangler dev, your Worker runs locally but Vectorize calls fail. You have to deploy to Cloudflare to test your vector search. For most development workflows this means testing your query logic locally with mocked responses, then deploying to a staging environment for full integration tests.

This is a real friction point. It's the honest trade-off for having a managed vector database with no infrastructure to operate.

Conclusion

In this tutorial, you have built and deployed a production-ready RAG system on Cloudflare's edge network. Let's look at what you actually built and what it costs to run.

What You Built

Your completed system has three endpoints:

• GET /: health check confirming the Worker is running

• POST /load: loads your knowledge base into Vectorize, protected by a secret header

• POST /query: accepts a question, retrieves relevant context, and returns a grounded answer with sources

The full query pipeline runs in four steps on every request:

1. The question is converted to a 768-dimensional embedding using @cf/baai/bge-base-en-v1.5

2. Vectorize finds the three most semantically similar documents

3. Documents above the 0.5 similarity threshold are assembled into context

4. Llama 3.3 generates an answer using only that context

Everything runs on Cloudflare's infrastructure. No external API keys. No separate vector database subscription. No servers to manage.

What to Build Next

This tutorial covered the core RAG pattern. Here are four directions to take it further.

Add more documents

The knowledge base in this tutorial has 8 documents. A real system might have thousands. The loading pattern is identical: add documents to knowledge-base.ts, hit /load with your secret, and Vectorize handles the rest.

For very large knowledge bases, update handleLoad to batch documents in groups of 20-50 rather than upserting one at a time.

Improve chunking

Each document in this tutorial is a single short paragraph. Real-world documents like PDFs, articles, documentation pages need to be split into chunks before embedding. Chunk at natural boundaries like paragraphs and sentences, aim for 200-400 tokens per chunk, and include 50-token overlaps between chunks to preserve context across boundaries.

Add conversation history

The current system treats every query as independent. To support follow-up questions, store previous messages in a Cloudflare KV namespace and include the last 2-3 exchanges in the LLM messages array alongside the retrieved context.

Stream the response

For long answers, users stare at a blank screen until generation completes. Cloudflare Workers support streaming responses via TransformStream. Switching to streaming means the first tokens appear in under 100ms while the rest generates.

Consider dimensions vs reranking trade-offs

This tutorial uses bge-base-en-v1.5 at 768 dimensions. My production system uses bge-small-en-v1.5 at 384 dimensions. Testing showed upgrading from 384 to 768 dims only improved accuracy by about 2%, but doubled cost and latency.

Adding a reranker (@cf/baai/bge-reranker-base) gave a larger accuracy improvement than the dimension upgrade for a fraction of the cost. The exact improvement will vary by domain and query distribution — test both on your actual data before deciding. If you're optimizing for production, add a reranker before you increase dimensions.

The Complete Project

Clone and deploy in five commands:

git clone https://github.com/dannwaneri/rag-tutorial-simple
cd rag-tutorial-simple
npm install
npx wrangler vectorize create rag-tutorial-index --dimensions=768 --metric=cosine
npx wrangler secret put LOAD_SECRET
npx wrangler deploy

Then load your knowledge base:

curl -X POST https://<your-worker>.workers.dev/load \
  -H "X-Load-Secret: your-secret"

If you found this useful, the production system this tutorial is based on is open source at github.com/dannwaneri/vectorize-mcp-worker. It extends this foundation with hybrid search combining vector and BM25, multimodal support for searching images with AI vision, a reranker for more accurate results, and a live dashboard. It runs on the same Cloudflare stack you just built – Workers, Vectorize, Workers AI – plus D1 for document storage.

One difference you'll notice: the production system uses bge-small-en-v1.5 at 384 dimensions rather than the 768 dimensions in this tutorial. That is an intentional trade-off: the reranker adds more accuracy than the extra dimensions at lower cost. The jump from what you built today to that system is smaller than it looks.