Contextualizers

The free, offline heuristic contextualizer and the opt-in, paid LLM contextualizer — see Contextual retrieval for how these two compare.

retrieval.contextualizer

Deterministic, offline contextualizer for RAG chunks.

No model, no network, no randomness. Output is byte-identical for identical inputs.

contextualize(chunk, all_chunks_in_doc)

Return contextualized text for chunk using all chunks in the same document.

Source code in engine/retrieval/contextualizer.py

def contextualize(chunk: Chunk, all_chunks_in_doc: List[Chunk]) -> str:
    """Return contextualized text for chunk using all chunks in the same document."""
    return make_context(chunk, all_chunks_in_doc)

make_context(chunk, neighbors)

Build a context prefix for a chunk given its neighbour chunks.

Format

[ > ] .

When heading or previous neighbour is absent, degrade gracefully.

Source code in engine/retrieval/contextualizer.py

def make_context(chunk: Chunk, neighbors: List[Chunk]) -> str:
    """Build a context prefix for a chunk given its neighbour chunks.

    Format:
        [<doc_title> > <heading>] <heading tokens once> <first sentence of
        previous chunk truncated>. <chunk.text>

    When heading or previous neighbour is absent, degrade gracefully.
    """
    parts: List[str] = []

    # Breadcrumb prefix
    if chunk.heading:
        breadcrumb = f"[{chunk.doc_title} > {chunk.heading}]"
        parts.append(breadcrumb)
        parts.append(chunk.heading)
    else:
        breadcrumb = f"[{chunk.doc_title}]"
        parts.append(breadcrumb)

    # First sentence of the immediately preceding chunk
    prev_chunk: Optional[Chunk] = None
    for nb in neighbors:
        if nb.position == chunk.position - 1 and nb.doc_id == chunk.doc_id:
            prev_chunk = nb
            break

    if prev_chunk:
        prev_sentence = _first_sentence(prev_chunk.text)
        if prev_sentence and not prev_sentence.endswith("."):
            prev_sentence += "."
        parts.append(prev_sentence)

    parts.append(chunk.text)
    return " ".join(parts)

retrieval.llm_contextualizer

LLM-based contextualizer — Anthropic's original Contextual Retrieval.

Where the heuristic contextualizer (retrieval/contextualizer.py) prepends a deterministic breadcrumb, this implements the technique from Anthropic's "Contextual Retrieval" post: for each chunk, an LLM is shown the whole document and the chunk, and writes a short context that situates the chunk within the document. That context is prepended before indexing, giving the sparse retriever vocabulary the chunk body lacks.

This is an opt-in, online comparison arm — it needs the anthropic package and ANTHROPIC_API_KEY. The import is lazy (inside methods), so the default offline pipeline and the no-network-imports test are unaffected.

The whole document is sent in a cache_control block, so every chunk from the same document reuses the cached document prefix (the standard cost optimization for this technique).

LLMContextualizer

Situate each chunk within its document via an LLM (Contextual Retrieval).

Exposes the same generate(chunk, all_chunks) -> str contract as the contextualizers in retrieval/providers.py.

Source code in engine/retrieval/llm_contextualizer.py

class LLMContextualizer:
    """Situate each chunk within its document via an LLM (Contextual Retrieval).

    Exposes the same ``generate(chunk, all_chunks) -> str`` contract as the
    contextualizers in ``retrieval/providers.py``.
    """

    def __init__(self, model: str = _MODEL) -> None:
        self._model = model
        self._client = None

    def _ensure_client(self):
        if self._client is None:
            self._client = _new_anthropic_client()
        return self._client

    @staticmethod
    def _document_text(all_chunks_in_doc: List[Chunk]) -> str:
        """Reconstruct the document by joining its chunks in position order."""
        ordered = sorted(all_chunks_in_doc, key=lambda c: c.position)
        return "\n\n".join(c.text for c in ordered)

    def _situate(self, document: str, chunk_text: str) -> str:
        client = self._ensure_client()
        response = client.messages.create(
            model=self._model,
            max_tokens=200,
            output_config={"effort": "low"},
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": f"<document>\n{document}\n</document>",
                            "cache_control": {"type": "ephemeral"},
                        },
                        {
                            "type": "text",
                            "text": _SITUATE_INSTRUCTION.format(chunk=chunk_text),
                        },
                    ],
                }
            ],
        )
        return next((b.text for b in response.content if b.type == "text"), "").strip()

    def generate(self, chunk: Chunk, all_chunks_in_doc: List[Chunk]) -> str:
        """Return ``<llm context> <chunk text>`` for indexing."""
        document = self._document_text(all_chunks_in_doc)
        context = self._situate(document, chunk.text)
        return f"{context} {chunk.text}".strip()

generate(chunk, all_chunks_in_doc)

Return <llm context> <chunk text> for indexing.

Source code in engine/retrieval/llm_contextualizer.py

def generate(self, chunk: Chunk, all_chunks_in_doc: List[Chunk]) -> str:
    """Return ``<llm context> <chunk text>`` for indexing."""
    document = self._document_text(all_chunks_in_doc)
    context = self._situate(document, chunk.text)
    return f"{context} {chunk.text}".strip()

LLMDocumentContextualizer

Document-level context enrichment (document-granularity analog).

Used by ContextualLexicalRetriever (retrieval/retrievers.py), which retrieves whole documents rather than chunks, so there is no parent document to situate a chunk within. The document-level adaptation of Contextual Retrieval is to ask the LLM for a short context (topics + key entities) for the whole document and prepend it before indexing — giving the sparse retriever extra surface vocabulary.

Runs on Haiku (the cheapest tier) since this is the cost arm — one call per document. The static instruction sits in a cache_control system block so the shared prefix is reused across documents; the document body is the varying suffix and isn't cached.

Exposes generate(document_text) -> str returning the context string (the caller prepends it to the document).

Source code in engine/retrieval/llm_contextualizer.py

class LLMDocumentContextualizer:
    """Document-level context enrichment (document-granularity analog).

    Used by ``ContextualLexicalRetriever`` (``retrieval/retrievers.py``), which
    retrieves whole documents rather than chunks, so there is no parent
    document to situate a chunk within.  The document-level adaptation of
    Contextual Retrieval is to ask the LLM for a short context (topics + key
    entities) for the *whole document* and prepend it before indexing — giving
    the sparse retriever extra surface vocabulary.

    Runs on Haiku (the cheapest tier) since this is the cost arm — one call per
    document.  The static instruction sits in a ``cache_control`` ``system``
    block so the shared prefix is reused across documents; the document body is
    the varying suffix and isn't cached.

    Exposes ``generate(document_text) -> str`` returning the context string
    (the caller prepends it to the document).
    """

    def __init__(self, model: str = _DOC_MODEL) -> None:
        self._model = model
        self._client = None

    def _ensure_client(self):
        if self._client is None:
            self._client = _new_anthropic_client()
        return self._client

    def generate(self, document_text: str) -> str:
        """Return a short LLM-written context describing the document."""
        client = self._ensure_client()
        response = client.messages.create(
            model=self._model,
            max_tokens=256,
            system=[
                {
                    "type": "text",
                    "text": _DOC_SYSTEM,
                    "cache_control": {"type": "ephemeral"},
                }
            ],
            messages=[
                {
                    "role": "user",
                    "content": f"<document>\n{document_text}\n</document>",
                }
            ],
        )
        return next((b.text for b in response.content if b.type == "text"), "").strip()

generate(document_text)

Return a short LLM-written context describing the document.

Source code in engine/retrieval/llm_contextualizer.py

def generate(self, document_text: str) -> str:
    """Return a short LLM-written context describing the document."""
    client = self._ensure_client()
    response = client.messages.create(
        model=self._model,
        max_tokens=256,
        system=[
            {
                "type": "text",
                "text": _DOC_SYSTEM,
                "cache_control": {"type": "ephemeral"},
            }
        ],
        messages=[
            {
                "role": "user",
                "content": f"<document>\n{document_text}\n</document>",
            }
        ],
    )
    return next((b.text for b in response.content if b.type == "text"), "").strip()

contextualize_llm(chunk, all_chunks_in_doc)

Module-level convenience matching the contextualize signature.

Source code in engine/retrieval/llm_contextualizer.py

def contextualize_llm(chunk: Chunk, all_chunks_in_doc: List[Chunk]) -> str:
    """Module-level convenience matching the ``contextualize`` signature."""
    global _DEFAULT
    if _DEFAULT is None:
        _DEFAULT = LLMContextualizer()
    return _DEFAULT.generate(chunk, all_chunks_in_doc)