Build It: A Small RAG Service That Ships

Q: The answers are wrong sometimes. Where do I look first?

Check the trace for that request. Look at ask.chunks and ask.sources_n. If the retrieved chunks did not contain the right context, your retrieval is the problem; turn to article 05 and article 06. If the chunks were right but the answer was wrong, your prompt or model is the problem; turn to article 03.

Q: Cost is creeping up. What is the cheapest thing to fix?

The exact-match cache. Look at ask.cache_hit in your traces; if the rate is below 20%, you have duplicate traffic that is not being deduped. Investigate query normalization (case, whitespace, trailing punctuation) before doing anything fancier. The rest of the playbook is in article 12: Cost Optimization.

The previous twelve articles each cover one piece of the AI stack. This one builds the whole thing.

We will write a small service called Ask the Docs. You point it at your team’s engineering documentation (a folder of markdown files, a Confluence export, a Notion dump, or your repo’s README ecosystem). It embeds them into Postgres and exposes a single endpoint that answers questions like “how do I deploy to staging?” or “what is our on-call rotation?” or “who owns the payments service?”, with cited sources from your docs.

This is the bot every engineering org’s #help channel actually needs. It is also a service that uses every concept from articles 01 through 12 in one place, so you can see the seams where each piece plugs in.

This is intentionally boring infrastructure: Postgres with pgvector, Redis, FastAPI, the OpenAI API. The goal is not to show off a hot new stack. The goal is to give you a service you can debug at 2 a.m. when a new hire pastes a screenshot of a wrong answer in Slack.

The architecture

flowchart LR
    subgraph Ingest["Ingest (run once, then on changes)"]
        N[Docs corpus] --> CH[Chunker]
        CH --> EMB[Embedder]
        EMB --> DB[(Postgres + pgvector)]
    end
    subgraph Query["Query (per request)"]
        Q[POST /ask] --> CK{cache hit?}
        CK -- yes --> AN[Answer]
        CK -- no --> R[Hybrid retrieve]
        R --> DB
        R --> G[Generate]
        G --> L[LLM]
        L --> AN
    end
    G -. write .-> RD[(Redis cache)]
    Q -. trace .-> O[Observability]

Two halves: an ingest pipeline that runs once and then on doc changes, and a query pipeline that runs per request. Each box maps to one Python file in the next section.

Project layout

ask-the-docs/
├── ingest.py          # one-shot: docs/ → rows in pgvector
├── chunker.py         # header-aware markdown chunking
├── retrieve.py        # hybrid search (vector + BM25, RRF)
├── generate.py        # LLM call, structured output, citations
├── cache.py           # Redis prompt + semantic cache
├── observability.py   # OpenTelemetry spans
├── server.py          # FastAPI: POST /ask
├── evals/
│   ├── dataset.json   # ~20 hand-labelled question, expected_path pairs
│   └── run.py         # eval runner, gates CI
├── schema.sql         # Postgres schema
├── pyproject.toml
└── docs/              # your team's docs: markdown, Confluence/Notion export, READMEs

Eight Python files. Each one is short enough to fit on a screen.

1. Schema and ingest

The schema is two columns past what you would expect: a tsvector for keyword search and a vector for semantic search.

-- schema.sql
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE docs (
  id          bigserial PRIMARY KEY,
  path        text       NOT NULL,
  chunk_idx   int        NOT NULL,
  content     text       NOT NULL,
  content_tsv tsvector   NOT NULL,
  embedding   vector(1536) NOT NULL,
  content_hash text      NOT NULL
);

CREATE INDEX ON docs USING gin(content_tsv);
CREATE INDEX ON docs USING hnsw (embedding vector_cosine_ops);

Ingest reads every markdown file, chunks it by heading (using the strategy from article 05: Choosing Chunking Strategies), embeds each chunk with text-embedding-3-small (article 07: Embeddings Deep Dive), and writes the rows.

# ingest.py
import os, hashlib
from pathlib import Path
import openai, psycopg
from chunker import chunk_markdown

OPENAI = openai.Client()

def embed(text: str) -> list[float]:
    r = OPENAI.embeddings.create(model="text-embedding-3-small", input=text)
    return r.data[0].embedding

def main():
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        for doc in Path("docs").rglob("*.md"):
            text = doc.read_text()
            h = hashlib.sha256(text.encode()).hexdigest()
            conn.execute("DELETE FROM docs WHERE path = %s AND content_hash != %s", (str(doc), h))
            if conn.execute("SELECT 1 FROM docs WHERE path = %s", (str(doc),)).fetchone():
                continue
            for i, chunk in enumerate(chunk_markdown(text)):
                vec = embed(chunk)
                conn.execute(
                  "INSERT INTO docs (path, chunk_idx, content, content_tsv, embedding, content_hash) "
                  "VALUES (%s,%s,%s, to_tsvector('english', %s), %s, %s)",
                  (str(doc), i, chunk, chunk, vec, h),
                )
        conn.commit()

The content_hash check is the cheap version of incremental ingest: if the doc did not change, skip it. For a Confluence or Notion source, swap Path("docs").rglob for a call to their export API and you keep the rest of the pipeline.

Why one row per chunk, not one row per doc?

Retrieval ranks chunks, not files. A 5,000-word runbook becomes maybe 20 chunks; the relevant paragraph is one of them. If you stored the whole doc as a single row, the embedding would average across topics (deployment, monitoring, incident response) and lose precision. The path column lets you reconstruct which doc a chunk came from for the citation.

2. Hybrid retrieval

Hybrid search runs a vector lookup and a BM25 lookup in parallel, then fuses the rankings with Reciprocal Rank Fusion (RRF). Reasons covered in article 06: Hybrid Search. The whole thing is one SQL query.

# retrieve.py
def hybrid_search(conn, query: str, k: int = 8):
    qvec = embed(query)
    rows = conn.execute("""
      WITH vec AS (
        SELECT id, content, path,
               row_number() OVER (ORDER BY embedding <=> %s::vector) AS rank
        FROM docs ORDER BY embedding <=> %s::vector LIMIT 50
      ),
      bm AS (
        SELECT id, content, path,
               row_number() OVER (ORDER BY ts_rank(content_tsv, q) DESC) AS rank
        FROM docs, plainto_tsquery('english', %s) AS q
        WHERE content_tsv @@ q LIMIT 50
      )
      SELECT coalesce(vec.id, bm.id) AS id,
             coalesce(vec.content, bm.content) AS content,
             coalesce(vec.path, bm.path) AS path,
             coalesce(1.0/(60 + vec.rank), 0) + coalesce(1.0/(60 + bm.rank), 0) AS score
      FROM vec FULL OUTER JOIN bm USING (id)
      ORDER BY score DESC LIMIT %s
    """, (qvec, qvec, query, k))
    return rows.fetchall()

k=8 is the default top-k. You will tune this once your evals are in place. BM25 carries questions like “deploy to staging” (exact phrase, lives in one runbook), vector carries questions like “how do we roll out new versions” (paraphrase, no shared keywords).

3. Generation with structured output

The model gets the chunks and a system prompt that forces it to cite sources. Structured output (covered in article 03: Prompting as Code) guarantees the response shape so the caller does not have to parse free-form text.

# generate.py
from pydantic import BaseModel

class Answer(BaseModel):
    answer: str
    sources: list[str]

SYSTEM = (
  "Answer the user's question using only the documentation provided. "
  "Cite the path of every doc you used in `sources`. "
  "If the docs do not contain the answer, say so plainly. Do not invent."
)

def generate(query: str, chunks) -> tuple[Answer, int, int]:
    context = "\n\n".join(f"[{c['path']}]\n{c['content']}" for c in chunks)
    resp = OPENAI.beta.chat.completions.parse(
        model="gpt-4o-mini",
        response_format=Answer,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": f"Docs:\n{context}\n\nQuestion: {query}"},
        ],
    )
    usage = resp.usage
    return resp.choices[0].message.parsed, usage.prompt_tokens, usage.completion_tokens

Three things this prompt does on purpose: it grounds the model in the retrieved chunks, it forces citations so the user can verify (and so your support people can fix the doc that gave a wrong answer), and it gives the model permission to say “I don’t know,” which is the single biggest hallucination cut you get for free.

4. Evals (the spine)

Without evals, every change is a guess. The eval suite is twenty hand-labelled question-and-expected-path pairs; the test passes if the right doc path appears in sources. Pattern from article 08: LLM Evaluation Pipelines.

# evals/run.py
import json, sys
from server import ask

DATASET = json.load(open("evals/dataset.json"))

def main():
    passed = 0
    for case in DATASET:
        result = ask({"question": case["question"]})
        if case["expected_path"] in result["sources"]:
            passed += 1
        else:
            print(f"MISS: {case['question']!r} → got {result['sources']}")
    rate = passed / len(DATASET)
    print(f"\n{passed}/{len(DATASET)} ({rate:.0%})")
    sys.exit(0 if rate >= 0.80 else 1)

Build your first dataset from #help history. Twenty real questions (“How do I get a dev DB?”, “What is the on-call rotation?”, “Where do I add a new env var?”) paired with the doc path that actually answers them. Wire run.py into CI; any PR that drops the pass rate below 80% fails the build. That is the line that protects you from quiet retrieval regressions.

For free-form quality (does the answer read well, does it actually address the question), add a second pass with article 09: LLM-as-Judge. You do not need it on day one.

5. Caching

Two layers of caching, both covered in article 10: Caching for LLM Apps:

Exact-match cache: hash the query, store the answer in Redis for an hour.
Semantic cache (optional, add when traffic justifies it): hash the embedding to a coarse bucket and reuse answers for near-duplicate queries.

# cache.py
import hashlib, json, redis
R = redis.from_url(os.environ["REDIS_URL"])

def _key(query: str) -> str:
    return "ask:" + hashlib.sha256(query.encode()).hexdigest()[:16]

def get(query: str) -> dict | None:
    raw = R.get(_key(query))
    return json.loads(raw) if raw else None

def set(query: str, answer: dict, ttl: int = 3600) -> None:
    R.setex(_key(query), ttl, json.dumps(answer))

The exact-match cache alone cuts cost meaningfully because real #help traffic is shockingly repetitive: every new hire asks the same five onboarding questions, the same on-call confusion shows up around every shift change, and the same “where do I add a feature flag?” gets re-asked roughly twice a week.

6. Observability

Every /ask request gets one OpenTelemetry span with attributes for token counts, cost, cache hits, and the number of chunks retrieved. Pattern from article 11: Production AI Observability.

# observability.py
from opentelemetry import trace
tracer = trace.get_tracer(__name__)

# pricing snapshot (May 2026) for gpt-4o-mini, USD per 1M tokens
PRICE_IN  = 0.150
PRICE_OUT = 0.600

def annotate(query: str, chunks, answer, tokens_in: int, tokens_out: int, cache_hit: bool):
    span = trace.get_current_span()
    span.set_attribute("ask.query_len", len(query))
    span.set_attribute("ask.chunks", len(chunks))
    span.set_attribute("ask.tokens_in", tokens_in)
    span.set_attribute("ask.tokens_out", tokens_out)
    span.set_attribute("ask.cost_usd", (tokens_in * PRICE_IN + tokens_out * PRICE_OUT) / 1_000_000)
    span.set_attribute("ask.cache_hit", cache_hit)
    span.set_attribute("ask.sources_n", len(answer["sources"]))

Once these spans ship to your tracing backend, you get a daily dashboard for free: p50/p95 latency, cost per request, cache hit rate, retrieval depth. That dashboard is also how you spot drift before users complain (article 11 walks through this).

7. The endpoint

Now the boring part. The endpoint stitches the six modules above together.

# server.py
from fastapi import FastAPI
from opentelemetry import trace
import psycopg, os
import cache, retrieve, generate, observability

app = FastAPI()
tracer = trace.get_tracer(__name__)
CONN = psycopg.connect(os.environ["DATABASE_URL"])

@app.post("/ask")
def ask(body: dict):
    query = body["question"]
    with tracer.start_as_current_span("ask"):
        hit = cache.get(query)
        if hit:
            observability.annotate(query, [], hit, 0, 0, cache_hit=True)
            return hit
        chunks = retrieve.hybrid_search(CONN, query, k=8)
        answer, t_in, t_out = generate.generate(query, chunks)
        result = answer.model_dump()
        cache.set(query, result)
        observability.annotate(query, chunks, result, t_in, t_out, cache_hit=False)
        return result

That is the whole service. Every concept from the twelve previous articles touches this function.

Wrap it in a Slack slash command (/askdocs how do I deploy to staging?) and you have shipped the onboarding bot. The slash command is 30 lines of glue: receive the Slack payload, post to /ask, format answer + sources as a Slack message.

Things this build skips on purpose

Be honest about what is missing. None of this is hard, it just is not the point of a capstone:

Auth. No API key, no rate limit. If you expose this beyond your VPN, add an API key check and a sliding-window limiter before the cache.get call.
Async ingest. The script re-runs over docs/. For a real product, watch the filesystem (or webhook from Confluence/Notion) and ingest only the changed docs.
Streaming responses. The endpoint returns the full answer. For a Slack or web chat UI, swap to OpenAI.beta.chat.completions.stream and send tokens to the client as they arrive.
Reranking. After hybrid retrieval, a cross-encoder reranker on the top 50 cuts to the best 8 with meaningfully better recall. Worth adding once your evals plateau on retrieval errors.
A real eval set. Twenty cases catch the obvious regressions. You want 100+ before the test means much. Add cases every time a new hire gets a bad answer.
Drift monitoring. A nightly cron that re-runs the evals against a frozen dataset and alerts when pass rate drops by more than 5%. Mentioned in article 11.
Multi-tenant. One docs table, one team. If you serve multiple teams or orgs, add a tenant_id column and filter every query by it.

If you ship this build with the seven gaps above plus an API key, you have a defensible v1 you can put in front of your engineering org on Monday.

Questions you will face in production

“The answers are wrong sometimes. Where do I look first?” Check the trace for that request. Look at ask.chunks and ask.sources_n. If the retrieved chunks did not contain the right context, your retrieval is the problem; turn to article 05 and article 06. If the chunks were right but the answer was wrong, your prompt or model is the problem; turn to article 03.

“Cost is creeping up. What is the cheapest thing to fix?” The exact-match cache. Look at ask.cache_hit in your traces; if the rate is below 20%, you have duplicate traffic that is not being deduped. Investigate query normalization (case, whitespace, trailing punctuation) before doing anything fancier. The rest of the playbook is in article 12: Cost Optimization.

“A user asked something the docs do not cover and the model made something up.” The system prompt says “do not invent” but models still cheat. Add a test case to your eval dataset for the specific query, mark expected_path as empty, and assert the answer string contains “do not see” or similar. The eval forces you to actually fix it. The deeper fix is to write the missing doc.

“Embeddings cost is high during the first ingest of a 10k-doc corpus.” One-time cost. Batch the embedding calls (the API takes a list of up to 2048 strings per call) and the bill is closer to $1 than $100. Also cache embeddings keyed by chunk hash so re-runs do not pay again.

“How do I know if a model upgrade is worth it?” Run the evals against the new model. If pass rate goes up by 3+ points and per-token cost is comparable, swap. If pass rate goes up by 1 point and cost doubles, do not swap. This is the entire reason the eval suite exists.

“My docs are not in markdown. They’re in Confluence / Notion / Google Docs.” Swap the ingest source. Confluence has a REST API that returns the page as storage-format HTML; convert HTML to markdown with markdownify and the rest of the pipeline is unchanged. Notion exports as zipped markdown directly. Google Docs has a REST API that returns the doc as plain text. Everything from chunker.py onward is corpus-agnostic.

What to remember

The twelve concepts you learned are not separate skills. They are one system, and the seams between them are where the bugs live.
The eval suite is the spine. Everything else is feature work that must not regress evals. Wire it into CI on day one.
Start with boring infrastructure: Postgres, Redis, FastAPI. You can debug all three. Replace them only when your evals tell you the system is the bottleneck.
The first version on purpose skips auth, streaming, reranking, and drift checks. They are easy to add once the core loop works.
Real users find bugs your dataset will not. Every bad answer in #help becomes a new eval case. Over six months the eval suite is the most valuable artifact you own, more so than the code.

What to study next

You finished the AI Engineering curriculum. Thirteen articles in, you have read more about building AI systems than most engineers who already ship them.

The thing left to do is build something. The ask-the-docs skeleton above is a starting point. Real options for what to point it at:

Your team’s wiki / Confluence / Notion — the canonical use case above. Slack-command it, watch the #help channel quiet down.
A codebase — swap header-aware markdown chunking for function-aware chunking via tree-sitter. Ask “where do we handle Stripe webhooks?”
Your last six months of support tickets — for every new ticket, surface similar resolved ones and a draft response.
A third-party API’s docs — point it at the Stripe or Twilio docs site. Build the API consultant you wish existed.

Pick the one closest to what your team actually needs. Deploy on a $5 VPS. Write the first ten eval cases by hand from real questions. When you find a bad answer, add it to the dataset. Iterate. That is the entire job.

If you want to extend your AI work to integrating with other tools, the MCP Development topic starts with What Is MCP?.