contextual-retrieval

What this skill does

# Contextual Retrieval

## The Core Idea

A chunk lifted from a long document loses context. "The company reported 3% revenue growth" does not say which company or which period. Anthropic's Contextual Retrieval prepends a short LLM-generated context string to each chunk before embedding and BM25 indexing.

From Anthropic's 2024 research (Pro Research team):

| Technique | Retrieval failure rate | Reduction |
|---|---|---|
| Embeddings only (baseline) | 5.7% | — |
| + BM25 (hybrid) | 4.7% | 17.5% |
| + Contextual embeddings | 3.7% | 35% |
| + Contextual BM25 | 2.9% | 49% |
| + Reranking | 1.9% | 67% |

Measured as `failure@20` on a mixed corpus (codebases, scientific papers, fiction).

## The Pipeline

```
[Doc] -> [Chunk] -> per-chunk LLM context via prompt cache -> [context + chunk]
                                                                    |
                                   +--------------------------------+
                                   |                                |
                             [embedding]                      [BM25 tokens]
                                   |                                |
                              [vector DB]                       [BM25 index]
                                   |                                |
                                   +---------- query ---------------+
                                                    |
                                          [RRF fusion top 150]
                                                    |
                                             [reranker top 20]
```

## Context Generation Prompt

The prompt Anthropic published. Do not paraphrase — it is tuned.

```python
CONTEXT_PROMPT = """<document>
{whole_document}
</document>

Here is the chunk we want to situate within the whole document:
<chunk>
{chunk_content}
</chunk>

Please give a short succinct context to situate this chunk within the overall
document for the purposes of improving search retrieval of the chunk. Answer
only with the succinct context and nothing else."""
```

Output is typically 50-100 tokens. Prepend to the chunk with a newline before indexing.

## Full Python Implementation with Prompt Caching

Prompt caching is the reason this is affordable — the whole document sits in the cache, then every chunk reuses it.

```python
from anthropic import Anthropic
from dataclasses import dataclass
import time

client = Anthropic()

@dataclass
class ContextualChunk:
    doc_id: str
    chunk_index: int
    original: str
    context: str
    combined: str  # context + "\n\n" + original

def contextualize_document(doc_id: str, document: str, chunks: list[str]) -> list[ContextualChunk]:
    """
    Generate context for every chunk of a document, reusing the cached document prefix.
    """
    out: list[ContextualChunk] = []
    for i, chunk in enumerate(chunks):
        resp = client.messages.create(
            model="claude-haiku-4-5-20250929",
            max_tokens=200,
            system=[
                {
                    "type": "text",
                    "text": "<document>\n" + document + "\n</document>",
                    "cache_control": {"type": "ephemeral"},
                }
            ],
            messages=[
                {
                    "role": "user",
                    "content": (
                        "Here is the chunk we want to situate within the whole document:\n"
                        f"<chunk>\n{chunk}\n</chunk>\n\n"
                        "Please give a short succinct context to situate this chunk within "
                        "the overall document for the purposes of improving search retrieval "
                        "of the chunk. Answer only with the succinct context and nothing else."
                    ),
                }
            ],
        )
        context = resp.content[0].text.strip()
        out.append(ContextualChunk(
            doc_id=doc_id,
            chunk_index=i,
            original=chunk,
            context=context,
            combined=f"{context}\n\n{chunk}",
        ))
    return out
```

The cache TTL is 5 minutes (ephemeral). Process all chunks of one document within that window so every call after the first is a cache hit.

### Cost Math

With claude-haiku-4-5 at (approximate 2025) pricing $0.80 / M input tokens, $4 / M output:

- Document: 100k tokens, 100 chunks.
- Without caching: `100 * 100k input = 10M input` -> $8 per document.
- With caching (write once + 99 reads): `100k base write (1.25x) + 99 * 100k * 0.1 read = 1.1M effective` -> ~$0.88.
- ~90% cost reduction.

Always enable prompt caching. It is the difference between a research demo and a production pipeline.

## Batch Processing via Message Batches API

For the first-time ingestion of a large corpus, use the Batches API (50% cheaper, async, 24h SLA).

```python
from anthropic import Anthropic
import json

client = Anthropic()

def build_batch_requests(doc_id: str, document: str, chunks: list[str]) -> list[dict]:
    system_cached = [{
        "type": "text",
        "text": "<document>\n" + document + "\n</document>",
        "cache_control": {"type": "ephemeral"},
    }]
    return [
        {
            "custom_id": f"{doc_id}::{i}",
            "params": {
                "model": "claude-haiku-4-5-20250929",
                "max_tokens": 200,
                "system": system_cached,
                "messages": [{
                    "role": "user",
                    "content": (
                        f"<chunk>\n{chunk}\n</chunk>\n\n"
                        "Please give a short succinct context to situate this chunk within "
                        "the overall document for the purposes of improving search retrieval "
                        "of the chunk. Answer only with the succinct context and nothing else."
                    ),
                }],
            },
        }
        for i, chunk in enumerate(chunks)
    ]

def submit_batch(requests: list[dict]) -> str:
    batch = client.messages.batches.create(requests=requests)
    return batch.id

def collect_batch(batch_id: str) -> dict[str, str]:
    batch = client.messages.batches.retrieve(batch_id)
    while batch.processing_status != "ended":
        time.sleep(30)
        batch = client.messages.batches.retrieve(batch_id)
    out = {}
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            text = result.result.message.content[0].text.strip()
            out[result.custom_id] = text
    return out
```

Combine with caching: batch jobs still honor `cache_control`. ~95% cost reduction on cold ingest.

## Indexing the Contextualized Chunks

Index `combined` text — not `original` — into both BM25 and the vector store. Keep `original` in the document store for final LLM context.

```python
from rank_bm25 import BM25Okapi
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

def index(ctx_chunks: list[ContextualChunk]):
    docs = [
        Document(
            page_content=c.combined,
            metadata={
                "doc_id": c.doc_id,
                "chunk_index": c.chunk_index,
                "original": c.original,
                "context": c.context,
            },
        )
        for c in ctx_chunks
    ]

    vstore = QdrantVectorStore.from_documents(
        docs, OpenAIEmbeddings(model="text-embedding-3-large"),
        collection_name="kb_contextual"
    )

    tokenized = [d.page_content.lower().split() for d in docs]
    bm25 = BM25Okapi(tokenized)

    return vstore, bm25, docs
```

## Hybrid Search with RRF

```python
from collections import defaultdict

def reciprocal_rank_fusion(results_lists: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    scores = defaultdict(float)
    for results in results_lists:
        for rank, doc_id in enumerate(results):
            scores[doc