Build a Citation-Ready Knowledge Base with r-1

Build a citation-ready knowledge base from Apple’s 10-K filings with r-1. Parse tables, create searchable records, filter retrieval, and refresh documents.

r-1 is Reducto’s unified parsing model, built to read a full page in context across layouts, tables, figures, and formatting. For a knowledge base, that means the searchable representation can preserve the structure and provenance that later retrieval depends on.

A useful knowledge base needs stable document IDs, metadata that can constrain retrieval, citation-ready chunks, and a repeatable way to add or replace documents. In this cookbook, you will build those pieces around Apple’s 2023, 2024, and 2025 10-K filings.

What you'll build

A multi-document corpus with stable IDs and filing metadata
Table-aware, embedding-optimized chunks with page and bounding-box provenance
A persistent local vector index
Semantic search with optional year and filing-type filters
A deterministic way to refresh one filing without rebuilding the entire corpus

The example stores the finished index in a JSON file so the entire flow is inspectable and self-contained.

Setup

Start by signing up for a Reducto Studio account and generating your Reducto API key.

Note: r-1 is currently in preview and requires the V3 Parse API. API requests must set settings.model to "r-1"; requests that omit it use legacy Parse. New Studio pipelines use r-1 by default. See the r-1 guide and review configuration compatibility before adapting an existing workflow.

bash
export REDUCTO_API_KEY="your-reducto-key"
export OPENAI_API_KEY="your-openai-key"

pip install reducto openai numpy requests

Step 1: Define the corpus and parse every filing

python
import json
import os
from pathlib import Path

import requests
from openai import OpenAI
from reducto import Reducto

FILINGS = [
    {
        "document_id": "apple-2023-10-k",
        "title": "Apple 2023 10-K",
        "company": "Apple",
        "filing_type": "10-K",
        "fiscal_year": 2023,
        "source_url": "https://d18rn0p25nwr6d.cloudfront.net/CIK-0000320193/faab4555-c69b-438a-aaf7-e09305f87ca3.pdf",
    },
    {
        "document_id": "apple-2024-10-k",
        "title": "Apple 2024 10-K",
        "company": "Apple",
        "filing_type": "10-K",
        "fiscal_year": 2024,
        "source_url": "https://d18rn0p25nwr6d.cloudfront.net/CIK-0000320193/c87043b9-5d89-4717-9f49-c4f9663d0061.pdf",
    },
    {
        "document_id": "apple-2025-10-k",
        "title": "Apple 2025 10-K",
        "company": "Apple",
        "filing_type": "10-K",
        "fiscal_year": 2025,
        "source_url": "https://d18rn0p25nwr6d.cloudfront.net/CIK-0000320193/c24e7a28-5254-4dfa-9447-62aaa3c24bb1.pdf",
    },
]

reducto = Reducto(api_key=os.environ["REDUCTO_API_KEY"])
ai = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def as_dict(value):
    if hasattr(value, "model_dump"):
        return value.model_dump()
    return value

def load_chunks(parse_response):
    if parse_response.result.type == "full":
        return [as_dict(chunk) for chunk in parse_response.result.chunks]

    response = requests.get(parse_response.result.url, timeout=120)
    response.raise_for_status()
    payload = response.json()

    if isinstance(payload, dict) and "result" in payload:
        payload = payload["result"]
    if isinstance(payload, dict) and "chunks" in payload:
        payload = payload["chunks"]

    return payload

def parse_filing(filing):
    parsed = reducto.parse.run(
        input=filing["source_url"],
        settings={"model": "r-1"},
        formatting={
            "table_output_format": "html",
            "merge_tables": True,
        },
        retrieval={
            "chunking": {
                "chunk_mode": "variable",
                "chunk_size": 1200,
                "chunk_overlap": 150,
            },
            "embedding_optimized": True,
            "filter_blocks": ["Header", "Footer", "Page Number"],
        },
    )
    print(
        filing["document_id"],
        parsed.usage.num_pages,
        parsed.studio_link,
    )
    return load_chunks(parsed)

Step 2: Turn r-1 output into knowledge-base records

Store two representations of each chunk: search_text is optimized for embeddings, while content preserves the original structured output used in an answer or source viewer.

The record also carries document-level metadata and the page positions of its source blocks.

python
def build_records(filing, chunks):
    records = []

    for index, chunk in enumerate(chunks):
        blocks = [as_dict(block) for block in chunk.get("blocks", [])]
        bboxes = [
            block["bbox"]
            for block in blocks
            if block.get("bbox") is not None
        ]
        pages = sorted({
            bbox["page"]
            for bbox in bboxes
            if bbox.get("page") is not None
        })

        records.append({
            "id": f"{filing['document_id']}:{index}",
            "document_id": filing["document_id"],
            "search_text": chunk.get("embed") or chunk["content"],
            "content": chunk["content"],
            "pages": pages,
            "bboxes": bboxes,
            "metadata": {
                "title": filing["title"],
                "company": filing["company"],
                "filing_type": filing["filing_type"],
                "fiscal_year": filing["fiscal_year"],
                "source_url": filing["source_url"],
            },
        })

    return records

records = []

for filing in FILINGS:
    chunks = parse_filing(filing)
    records.extend(build_records(filing, chunks))

print(f"Prepared {len(records)} records from {len(FILINGS)} filings")

Step 3: Embed and persist the corpus

Embedding in batches keeps the indexing step predictable as the corpus grows. The resulting JSON contains the metadata, source geometry, searchable text, and vector for every record.

python
EMBEDDING_MODEL = "text-embedding-3-small"
INDEX_PATH = Path("apple_filings_kb.json")
BATCH_SIZE = 96

def embed_records(items):
    for start in range(0, len(items), BATCH_SIZE):
        batch = items[start:start + BATCH_SIZE]
        response = ai.embeddings.create(
            model=EMBEDDING_MODEL,
            input=[record["search_text"] for record in batch],
        )

        for record, result in zip(batch, response.data):
            record["embedding"] = result.embedding

    return items

embed_records(records)

INDEX_PATH.write_text(
    json.dumps(
        {
            "embedding_model": EMBEDDING_MODEL,
            "records": records,
        }
    ),
    encoding="utf-8",
)

print(f"Saved {len(records)} records to {INDEX_PATH}")

Step 4: Search across filings with metadata filters

python
import numpy as np

knowledge_base = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
records = knowledge_base["records"]
embedding_model = knowledge_base["embedding_model"]

def normalize(vector):
    vector = np.asarray(vector, dtype=np.float64)
    return vector / np.linalg.norm(vector)

def search_kb(question, k=6, years=None, filing_types=None):
    candidates = [
        record
        for record in records
        if (
            years is None
            or record["metadata"]["fiscal_year"] in years
        )
        and (
            filing_types is None
            or record["metadata"]["filing_type"] in filing_types
        )
    ]

    if not candidates:
        return []

    response = ai.embeddings.create(
        model=embedding_model,
        input=[question],
    )
    query_vector = normalize(response.data[0].embedding)

    ranked = []
    for record in candidates:
        score = float(
            normalize(record["embedding"]) @ query_vector
        )
        ranked.append({"score": score, "record": record})

    return sorted(
        ranked,
        key=lambda hit: hit["score"],
        reverse=True,
    )[:k]

question = (
    "How did Apple's Services net sales change from 2023 "
    "through 2025, and what explanations did Apple give?"
)

hits = search_kb(
    question,
    years={2023, 2024, 2025},
    filing_types={"10-K"},
)

for hit in hits:
    record = hit["record"]
    print(
        round(hit["score"], 3),
        record["metadata"]["title"],
        record["pages"],
    )
    print(record["content"][:400], "\n")

Step 5: Refresh one document without rebuilding the corpus

When a corrected or newer version arrives, replace every record with the same stable document_id.

python
def upsert_filing(filing):
    knowledge_base = json.loads(
        INDEX_PATH.read_text(encoding="utf-8")
    )
    existing = [
        record
        for record in knowledge_base["records"]
        if record["document_id"] != filing["document_id"]
    ]

    chunks = parse_filing(filing)
    replacements = build_records(filing, chunks)
    embed_records(replacements)

    knowledge_base["records"] = existing + replacements
    INDEX_PATH.write_text(
        json.dumps(knowledge_base),
        encoding="utf-8",
    )

    print(
        f"Replaced {filing['document_id']} with "
        f"{len(replacements)} records"
    )

# Re-run this after replacing the source URL or metadata.
upsert_filing(FILINGS[-1])

Prefer no code?

Use Reducto Studio to upload each representative document, apply the same r-1 parsing and chunking configuration, and compare the result with the source side by side. Once the structure looks right, move the configuration into the ingestion job that writes to your knowledge store.

Where this goes next

Replace the local JSON file with your vector database once the corpus or query volume grows.
Add tenant IDs, access-control metadata, and permission filters before ranking.
Store a source checksum so unchanged documents can skip parsing and embedding.
Add quarterly filings, earnings releases, and investor presentations with their own document-type metadata.
Combine semantic retrieval with keyword search or a reranker for exact financial terms.
Use the saved page and bounding-box data to open a source viewer directly on the evidence.

Product

Industries

Resources