RAG with TopK

Step	Tech	Execution
Parsing & chunking	Docling	💻 Local
Embedding, keyword & semantic search	TopK	🌐 Remote
Gen AI	Anthropic Claude	🌐 Remote

How to cook

This notebook builds a Retrieval-Augmented Generation (RAG) pipeline over a PDF that mixes prose and charts. Docling parses the document, and TopK stores and searches it. TopK embeds text on the server, so there is no embedding model to choose, host or call, and the same field serves both semantic and BM25 keyword search.

The example uses the Federal Reserve's Financial Stability Report (April 2024): 69 pages, about 60 charts. Many questions about such a report can only be answered from a chart, so the pipeline indexes two kinds of records:

Text chunks from Docling's HybridChunker, with section headings and the page number.
Figure records: one per chart, holding the text Docling extracts from the chart itself (title, legend, axis labels, source) and the page number. Docling also keeps the cropped image, which we pass to the LLM when a figure is retrieved.
TopK: a search engine with built-in semantic search, BM25 keyword search, vector search and filtering in one query. Learn more: TopK docs
Docling: converts PDFs and other formats into a structured DoclingDocument with layout, tables, figures and page provenance. Learn more: Docling

Setting up your environment

Install Docling, the TopK Python SDK and the Anthropic SDK. You need a TopK API key from the TopK console and its region, and an Anthropic API key for the generation step.

%%capture
%pip install docling topk-sdk anthropic

import logging
import warnings

warnings.filterwarnings("ignore")
logging.getLogger("docling").setLevel(logging.ERROR)

import os
from getpass import getpass

for var in ["TOPK_API_KEY", "TOPK_REGION", "ANTHROPIC_API_KEY"]:
    if not os.environ.get(var):
        os.environ[var] = getpass(f"{var}: ")  # e.g. TOPK_REGION=aws-us-east-1-elastica

Part 1: Parse the document with Docling

generate_picture_images=True keeps a cropped image of every figure, so we can show the chart to the LLM later. Conversion runs locally; on a laptop CPU this 69-page report takes a couple of minutes.

from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import PdfPipelineOptions
from docling.document_converter import DocumentConverter, PdfFormatOption

SOURCE = "https://www.federalreserve.gov/publications/files/financial-stability-report-20240419.pdf"

pipeline_options = PdfPipelineOptions(generate_picture_images=True, images_scale=2.0)
converter = DocumentConverter(
    format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options)}
)
doc = converter.convert(SOURCE).document
print(
    f"{doc.num_pages()} pages, {len(doc.texts)} text items, {len(doc.pictures)} pictures"
)

69 pages, 1838 text items, 82 pictures

Chunk the text

HybridChunker splits along the document structure and merges small pieces up to a token limit. contextualize() prepends the section headings to each chunk, which helps retrieval. TopK's semantic_index() accepts values up to 4,096 characters, so we keep chunks well below that (about 400 tokens).

from docling.chunking import HybridChunker

chunker = HybridChunker(max_tokens=400)

records = []
for i, chunk in enumerate(chunker.chunk(doc)):
    text = chunker.contextualize(chunk)
    pages = sorted({p.page_no for item in chunk.meta.doc_items for p in item.prov})
    if len(text.strip()) < 40:  # skip page furniture and stray labels
        continue
    records.append(
        {"_id": f"text-{i}", "kind": "text", "page": pages[0], "text": text[:4000]}
    )

print(f"{len(records)} text chunks")
print(records[10]["text"][:500])

114 text chunks
Overview of financial system vulnerabilities
- Equity price-toearnings ratios moved to the upper end of their historical dist r ibut ions.
- Corporate bond spreads fell to levels that are low relative to their historical averages.
- Prices of residential real estate remained high relative to fundamentals.
- Prices of commercial real estate declined amid deteriorating fundamentals.
- The ratio of total private debt to gross domestic product (GDP) declined further, approaching its historical avera

Add one record per figure

Text chunks alone rarely surface charts: in our tests on this report, questions that need a chart found the right page first only 17% of the time with text chunks alone. Indexing each chart as its own record raised that to 75%.

For charts drawn as vector graphics, as in this report, Docling extracts the words inside the chart (title, legend, axis labels, source) as child text items of the picture. That text turned out to be a better search key than descriptions from a small local vision model, which sometimes invent details. For scanned documents, enable OCR so those labels exist.

from docling_core.types.doc import PictureItem

figure_images = {}  # _id -> PIL image, used later to show charts to the LLM

for n, (item, _) in enumerate(doc.iterate_items()):
    if not isinstance(item, PictureItem):
        continue
    image = item.get_image(doc)
    if image is None or min(image.size) < 100:  # skip logos and icons
        continue
    children = [child.resolve(doc) for child in item.children]
    labels = " ".join(c.text for c in children if hasattr(c, "text"))
    caption = item.caption_text(doc)
    if len(caption + labels) < 30:  # cover art and decorations carry no searchable text
        continue
    _id = f"figure-{n}"
    figure_images[_id] = image
    records.append(
        {
            "_id": _id,
            "kind": "figure",
            "page": item.prov[0].page_no,
            "text": f"{caption}\n{labels}".strip()[:4000],
        }
    )

print(f"{len(figure_images)} figures")
print(next(r for r in records if r["text"].startswith("Figure 4.5"))["text"][:400])

72 figures
Figure 4.5. Market capitalization of major stablecoins increased slightly since late last year 1. Other 2. TerraUSD 3. Dai 4. Binance USD 5. USD Coin 6. Tether Mar. June Sept. Dec. Mar. June Sept. Dec. Mar. June Sept. Dec. Mar. 0 50 100 150 200 Mar. 25 Daily Billions of dollars 1 2 3 4 5 6 2021 2022 2023 2024 Source: Llama Corp, DeFiLlama.

Part 2: Index in TopK

A TopK collection holds documents and their indexes. We declare only the fields we search or filter on: text gets a semantic_index(), which embeds it on the server and enables BM25 keyword search on the same field. Other fields, such as kind and page, can be stored without being declared, but declaring them keeps the types checked.

import os

from topk_sdk import Client
from topk_sdk.error import CollectionAlreadyExistsError
from topk_sdk.query import field
from topk_sdk.schema import int as int_field, semantic_index, text

client = Client(api_key=os.environ["TOPK_API_KEY"], region=os.environ["TOPK_REGION"])

COLLECTION = "docling-fsr"
try:
    client.collections().create(
        COLLECTION,
        schema={
            "kind": text().required(),
            "page": int_field().required(),
            "text": text().required().index(semantic_index()),
        },
    )
except CollectionAlreadyExistsError:
    pass  # re-running the notebook is safe

collection = client.collection(COLLECTION)

Upsert in batches to stay under the 8 MB request limit. Each write returns a log sequence number (LSN); passing it to a read makes that read wait until the write is visible, so there is no need to sleep while the index catches up.

lsn = ""
for start in range(0, len(records), 200):
    lsn = collection.upsert(records[start : start + 200])

print(f"{collection.count(lsn=lsn)} documents indexed")

186 documents indexed

Part 3: Search

One query combines semantic similarity and BM25. should() adds keyword scores without dropping documents that share no words with the question. The two scores live on different scales (semantic is roughly 1-4, BM25 roughly 0-15), so we weight BM25 down; tune KEYWORD_WEIGHT on your own data.

from topk_sdk.query import fn, select, should

KEYWORD_WEIGHT = 0.2


def search(question: str, k: int = 5) -> list[dict]:
    return collection.query(
        select(
            "kind",
            "page",
            "text",
            semantic=fn.semantic_similarity("text", question),
            keyword=fn.bm25_score(),
        )
        .filter(should(question, field="text"))
        .sort(field("semantic") + field("keyword") * KEYWORD_WEIGHT, asc=False)
        .limit(k)
    )


for question in [
    "When did Silicon Valley Bank and Signature Bank fail?",
    "Which stablecoin has the largest market capitalization?",
]:
    print(f"\n{question}")
    for hit in search(question, k=3):
        print(f"  p.{hit['page']:<3} {hit['kind']:<7} {hit['text'][:90]!r}")

When did Silicon Valley Bank and Signature Bank fail?


  p.46  text    'Box 4.1. The Bank Term Funding Program\nThe banking system came under severe stress in Marc'
  p.46  text    'Box 4.1. The Bank Term Funding Program\nThe BTFP provided depository institutions an additi'
  p.11  text    'Overview of financial system vulnerabilities\nIn addition, the report also contains the box'

Which stablecoin has the largest market capitalization?


  p.49  figure  'Figure 4.5. Market capitalization of major stablecoins increased slightly since late last '
  p.49  text    'Stablecoins remained vulnerable to runs\nStablecoin assets-digital assets designed to maint'
  p.43  text    'Vulnerabilities from funding risks remained notable, reflecting challenges at some banks a'

The first question is answered in prose; the second only by the stablecoin chart, which comes back as a figure hit. The text of a figure record identifies the chart, but the answer (which band is largest) is only visible in the image itself, so we pass figure images to the LLM in the next step.

Part 4: Answer with Claude

We send the retrieved chunks with their page numbers, plus the cropped image of each retrieved figure, and ask for an answer that cites pages.

import base64
import io

import anthropic

llm = anthropic.Anthropic()


def image_block(image) -> dict:
    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    data = base64.standard_b64encode(buffer.getvalue()).decode()
    return {
        "type": "image",
        "source": {"type": "base64", "media_type": "image/png", "data": data},
    }


def answer(question: str, k: int = 5) -> str:
    hits = search(question, k)
    content = []
    for hit in hits:
        content.append(
            {
                "type": "text",
                "text": f"[page {hit['page']}, {hit['kind']}]\n{hit['text']}",
            }
        )
        if hit["kind"] == "figure":
            content.append(image_block(figure_images[hit["_id"]]))
    content.append(
        {
            "type": "text",
            "text": "Answer the question using only the context above, and cite page numbers. "
            f"If the context does not contain the answer, say so.\n\nQuestion: {question}",
        }
    )
    response = llm.beta.messages.create(
        model="claude-opus-5-5",
        max_tokens=16000,
        output_config={"effort": "low"},
        betas=["server-side-fallback-2026-07-01"],
        fallbacks="default",  # re-run on another model if this one declines
        messages=[{"role": "user", "content": content}],
    )
    if response.stop_reason == "refusal":
        return "The model declined to answer."
    return "".join(block.text for block in response.content if block.type == "text")

from rich.console import Console
from rich.panel import Panel

console = Console()
for question in [
    "Which stablecoin has the largest market capitalization?",
    "Why did commercial real estate prices decline, and how did banks respond?",
]:
    console.print(Panel(answer(question), title=question, border_style="cyan"))

Clean up

Delete the collection when you are done. To re-index the same name right away, prefer a new collection name instead: writes to a collection recreated under the same name shortly after deletion can be lost.

# client.collections().delete(COLLECTION)

Summary

This notebook built a RAG pipeline over a PDF that mixes prose and charts:

Parsed the PDF with Docling, keeping page numbers and figure images
Chunked the text with HybridChunker and added one record per chart from the text Docling extracts out of the chart
Indexed both in TopK, with embeddings computed on the server and BM25 on the same field
Ran hybrid semantic and keyword search in a single query
Answered with Claude, passing the retrieved chart images so values can be read off the chart

Next steps

Index many documents: add a source field and filter on it, or use one TopK partition per tenant
Turn on OCR in Docling for scanned PDFs, so figures and pages without a text layer still produce searchable text
Measure retrieval on your own questions before tuning KEYWORD_WEIGHT or the chunk size