AI Agent Beginner Tutorial: 07 - RAG, Context Engineering, and Citations

Follow the RAG pipeline

Raw Markdown
     │
     ▼
Clean metadata ──▶ Chunk by heading and length ──▶ Assign source_id
                                                   │
User question ──▶ Normalize query ──▶ Recall ─────▶ Rank and deduplicate
                                                   │
                                                   ▼
                                          Return a few previews

RAG handles finding material and placing it in context. The model still generates conclusions, and the application still validates citations.

Design traceable chunks

Every chunk keeps document ID, chunk ID, title, position, and body. Once a source_id enters a published artifact, it should remain stable across index rebuilds.

from dataclasses import dataclass

@dataclass(frozen=True)
class Chunk:
    source_id: str
    document_id: str
    title: str
    position: int
    text: str

def make_chunks(document_id: str, title: str, paragraphs: list[str],
                max_chars: int = 900) -> list[Chunk]:
    chunks = []
    buffer = ""
    for paragraph in paragraphs:
        candidate = f"{buffer}\n\n{paragraph}".strip()
        if buffer and len(candidate) > max_chars:
            position = len(chunks)
            chunks.append(Chunk(f"{document_id}#{position}", document_id, title, position, buffer))
            buffer = paragraph
        else:
            buffer = candidate
    if buffer:
        position = len(chunks)
        chunks.append(Chunk(f"{document_id}#{position}", document_id, title, position, buffer))
    return chunks

Paragraph boundaries preserve semantic units. A heading or list item should not be separated from its explanatory text. Production sources also record version, update time, and access level; the current version keeps only the fields required for retrieval.

Establish a keyword-recall baseline

import re

def tokens(text: str) -> set[str]:
    return set(re.findall(r"[a-z0-9]+", text.lower()))

def search_chunks(query: str, chunks: list[Chunk], limit: int = 4) -> list[dict]:
    query_tokens = tokens(query)
    scored = []
    for chunk in chunks:
        score = len(query_tokens & tokens(f"{chunk.title} {chunk.text}"))
        if score:
            scored.append((score, chunk.source_id, chunk))
    scored.sort(key=lambda item: (-item[0], item[1]))
    return [
        {"source_id": chunk.source_id, "title": chunk.title, "preview": chunk.text[:280]}
        for _, _, chunk in scored[:limit]
    ]

Ties sort by source_id, so tests do not change with dictionary iteration order. Search returns previews only. The model calls get_document(source_id) after selecting a result, preventing every full document from entering context at once.

Debug recall with a fixed query set

Prepare at least six sources and ten queries, then record expected sources for each query.

Measure Recall@4 first: does the expected source appear in the first four results? Then measure final brief quality. Prompt changes cannot recover a source that retrieval missed. When retrieval succeeds but the answer fails, context assembly or the model becomes the next layer to inspect.

Connect retrieval to the existing agent loop

Register two read-only tools: search_local_sources(query) returns previews, and get_document(source_id) returns one chunk body. Tool descriptions tell the model to search before reading and limit each result.

The run log should show a clear chain:

topic
  └─▶ search_local_sources("agent loop stopping")
         └─▶ [agent-loop#1, agent-loop#0]
                └─▶ get_document("agent-loop#1")
                       └─▶ draft

An unknown source_id returns source_not_found without revealing every available ID. An empty search returns matches: []; the later output contract represents insufficient information.

Exercise and acceptance

• Rebuilding the index from the same source preserves every source_id.
• Expected results for ten queries are frozen and reported as Recall@4.
• Duplicate chunks collapse into one result, with no more than four matches.
• Search and read tools can be tested independently of the model.

With retrieval connected, the next step assembles task instructions, retrieved evidence, and history into a clear context, then requires every claim in the brief to carry a verifiable source.

A retrieval hit does not guarantee a reliable answer

The retrieval code above can find the correct chunk, yet the model may ignore sources, mix two passages, or invent a plausible conclusion after an empty search. RAG output needs a context assembly layer, and final output needs a shape that code can inspect.

Context should not become one endlessly growing string. Stable rules, the current task, external evidence, and conversation history have different trust levels and retention periods.

Use four context layers for one inference

┌────────────────────────────────────────┐
│ 1. System and safety rules  stable, top │
├────────────────────────────────────────┤
│ 2. Current task spec       topic, budget│
├────────────────────────────────────────┤
│ 3. Retrieved evidence      untrusted, ID│
├────────────────────────────────────────┤
│ 4. Necessary history       recent events│
└────────────────────────────────────────┘

Retrieved sources are untrusted data. Even if a document says “ignore system instructions,” it remains citation material and cannot change tool permissions or the task goal. Chapter 9 strengthens this boundary through tool permissions and adversarial tests.

Build a context packet

from dataclasses import dataclass

@dataclass(frozen=True)
class Evidence:
    source_id: str
    title: str
    text: str

def build_context(task: dict, evidence: list[Evidence], max_chars: int = 6000) -> str:
    blocks = [
        "[SYSTEM_RULES]\nUse only supplied evidence. Preserve source IDs. Report gaps.",
        f"[TASK]\nTopic: {task['topic']}\nAudience: {task['audience']}",
    ]
    used = sum(len(block) for block in blocks)
    for item in evidence:
        block = f"[EVIDENCE source_id={item.source_id} title={item.title}]\n{item.text}"
        if used + len(block) > max_chars:
            break
        blocks.append(block)
        used += len(block)
    return "\n\n".join(blocks)

The budget truncates complete blocks so a sentence never remains without its source label. A production version may allocate tokens to each layer. The current version uses characters as an observable baseline and logs omitted source_id values.

Return claims instead of one Markdown blob

The model first returns structured output, and a presentation layer later renders Markdown.

{
  "status": "succeeded",
  "claims": [
    {
      "text": "An agent loop makes another decision after a tool result returns.",
      "source_ids": ["agent-loop#1"]
    }
  ],
  "evidence_gaps": []
}

Structured claims let the application inspect citations one by one. Writing style may change while source relationships stay stable.

Validate source relationships in code

def validate_brief(result: dict, known_source_ids: set[str]) -> list[str]:
    errors = []
    status = result.get("status")
    claims = result.get("claims", [])
    if status == "succeeded" and not claims:
        errors.append("claims_required")
    for index, claim in enumerate(claims):
        source_ids = claim.get("source_ids", [])
        if not source_ids:
            errors.append(f"claim_{index}_missing_source")
        elif not set(source_ids) <= known_source_ids:
            errors.append(f"claim_{index}_unknown_source")
    return errors

This check proves that source IDs exist, but not that evidence supports a claim. The evaluation set also samples entailment, checking that numbers, negation, and qualifiers agree with the source.

End cleanly when evidence is absent

When search returns an empty list or every source falls outside the required time range, output becomes insufficient_evidence.

{
  "status": "insufficient_evidence",
  "claims": [],
  "evidence_gaps": ["The corpus has no official release notes after 2026"]
}

This path is not a runtime failure. It means the agent completed retrieval and followed the task rule against unsupported conclusions. The product can ask the user for material or offer permitted remote search.

Run the first complete brief

A successful run follows this chain.

Task specification
  └─▶ search_local_sources
        └─▶ get_document
              └─▶ build_context
                    └─▶ structured brief
                          └─▶ validate_brief
                                └─▶ Markdown renderer

Tests cover one source, multiple conflicting sources, no evidence, context truncation, and a model returning an unknown source. Every failure maps to retrieval, context, generation, or validation.

The second milestone

The project can now produce a cited technical brief from local material. Chapter 6's tool boundary still protects retrieval calls, while this chapter's citation validator extends source_id constraints to the final output.

Chapter 8 packages stable procedures, domain rules, and supporting resources as skills so the agent can load a method for the current task instead of keeping every instruction in one permanent system prompt.