Website to Vector Database: A Small, Traceable Pipeline

A vector database can store vectors. It cannot tell you which page produced a passage or whether that passage is stale.

This example uses one documentation URL so you can inspect every stage. For a whole site, use a bounded Crawl task to discover pages, then pass each page through the same normalization and indexing functions.

The record contract

Keep the source URL, heading path, content hash, and crawl time with every chunk. Stable IDs let you replace a page without creating a second copy of the same content.

Install the SDKs and set credentials in your shell or secret manager:

python -m venv .venv
source .venv/bin/activate
pip install nstdata-ai-crawl openai pinecone
export NSTDATA_API_TOKEN="..."
export OPENAI_API_KEY="..."
export PINECONE_API_KEY="..."
export PINECONE_INDEX_HOST="..."

Create a Pinecone index whose dimension matches the embedding output. This example uses the default 1,536 dimensions for text-embedding-3-small.

Fetch and split a page

import os
import re
from hashlib import sha256

from nstdata_ai_crawl import Format, NstDataClient, ScrapeRequestDto

source_url = "https://docs.example.com/api"
with NstDataClient(os.environ["NSTDATA_API_TOKEN"]) as crawl:
    result = crawl.submit_scrape_task_sync(ScrapeRequestDto(
        url=source_url,
        formats=[Format.MARKDOWN],
        onlyMainContent=True,
        timeout=60_000,
    ))
    markdown = result.data.get_markdown()

if not markdown or len(markdown.strip()) < 200:
    raise RuntimeError("No usable page content returned")

markdown = re.sub(r"\n{3,}", "\n\n", markdown.replace("\r\n", "\n")).strip()
page_hash = sha256(markdown.encode("utf-8")).hexdigest()

FENCE = chr(96) * 3

def split_sections(markdown: str):
    path, lines, fenced = [], [], False
    for line in markdown.splitlines():
        if line.strip().startswith(FENCE):
            fenced = not fenced
        if line.startswith("#") and not fenced and lines:
            text = "\n".join(lines).strip()
            if text:
                yield " / ".join(path), text
            level = len(line) - len(line.lstrip("#"))
            path = path[:level - 1] + [line[level:].strip()]
            lines = [line]
        else:
            lines.append(line)
    if lines:
        text = "\n".join(lines).strip()
        if text:
            yield " / ".join(path), text

sections = list(split_sections(markdown))
if not sections:
    raise RuntimeError("No sections to index")

This splitter keeps fenced code together but does not impose a token limit. Add a tokenizer-aware split for long sections and test the longest page in your corpus.

Embed and upsert

from openai import OpenAI
from pinecone import Pinecone

embedder = OpenAI()
texts = [text for _, text in sections]
vectors = embedder.embeddings.create(
    model="text-embedding-3-small",
    input=texts,
).data

if any(len(item.embedding) != 1536 for item in vectors):
    raise ValueError("Index dimension does not match embedding output")

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index(host=os.environ["PINECONE_INDEX_HOST"])
page_key = sha256(source_url.encode("utf-8")).hexdigest()[:16]

records = []
for (heading, text), item in zip(sections, vectors, strict=True):
    chunk_key = sha256((heading + "\0" + text).encode("utf-8")).hexdigest()[:20]
    records.append({
        "id": f"{page_key}#{chunk_key}",
        "values": item.embedding,
        "metadata": {
            "text": text,
            "source_url": source_url,
            "heading_path": heading,
            "content_hash": page_hash,
        },
    })

index.upsert(vectors=records, namespace="public-docs-v1")

The code shows the core transformations; pin compatible package versions in your project lockfile. For each page, keep the previous chunk IDs in a manifest. When the page changes, upsert the new chunks, confirm they are queryable, and delete old IDs that are no longer used.

A query should use the same embedding model and return metadata with the matches. That lets the application cite a URL and heading instead of displaying a bare answer.

Before wiring the index into chat, add a small query test that checks the expected page and heading appear in the top results. Review the returned metadata, not just the similarity score.

Failure cases worth testing

• Empty or blocked Markdown should stop before embedding.
• Index dimension must match the embedding output.
• A repeated crawl of unchanged content should not create duplicate records.
• Timeouts and access failures should not delete the last good version.
• Removed pages need a deliberate cleanup path.

Nstdata’s Crawl API documentation, Pinecone’s upsert guide, and OpenAI’s embeddings guide describe the service interfaces used here.

These snippets are a documented integration example, not a live end-to-end test. Start with a small, permitted page set and verify citations, updates, and deletes before scaling.