How to Ingest Documentation Sites into Vertex AI with EnConvert

import os
import json
import time

import requests
from google.api_core.client_options import ClientOptions
from google.cloud import discoveryengine_v1 as discoveryengine

ENCONVERT_API_KEY = os.environ["ENCONVERT_API_KEY"]
DOCS_SITE_URL = os.environ["DOCS_SITE_URL"]  # the docs site root, e.g. https://docs.example.com
PROJECT_ID = os.environ["GCP_PROJECT_ID"]
DATASTORE_ID = os.environ["VERTEX_AI_DATASTORE_ID"]  # a structured (no content) data store
LOCATION = "global"  # or "us" / "eu" for a multi-region data store

# Step 1: Submit the documentation site as a sitemap ingest job
# EnConvert finds the site's sitemap.xml itself, so send the site root, not the sitemap file
# To ingest specific pages instead, send {"mode": "urls", "urls": ["https://..."]}
# Pages behind a login are skipped and counted in pages_failed
submit = requests.post(
    "https://api.enconvert.com/v2/ingest",
    headers={
        "X-API-Key": ENCONVERT_API_KEY,
        "Content-Type": "application/json",
    },
    json={
        "mode": "sitemap",
        "url": DOCS_SITE_URL,
        "max_pages": 200,
        "chunk": {"max_words": 512},
    },
    timeout=60,
)
submit.raise_for_status()
job_id = submit.json()["job_id"]

# Step 2: Poll until the job completes
# Large sites take time because each page renders sequentially
while True:
    status_response = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers={"X-API-Key": ENCONVERT_API_KEY},
        timeout=60,
    )
    status_response.raise_for_status()
    job = status_response.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(job.get("error_message") or f"Ingest job {job['status']}")
    time.sleep(30)

print(f"Pages processed: {job['pages_processed']}, failed: {job['pages_failed']}")

# Step 3: Download the JSONL output
# The signed URL is valid for 15 minutes. Plain GET, no API key.
jsonl_response = requests.get(job["output_url"], timeout=60)
jsonl_response.raise_for_status()
jsonl_response.encoding = "utf-8"  # the file is UTF-8

chunks = [
    json.loads(line)
    for line in jsonl_response.text.split("\n")
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for import")

# Step 4: Build one Vertex AI Search document per chunk
# Chunks are split at H1 to H3 heading boundaries with a 512 word soft cap
# Tables and code blocks are kept whole even when they exceed the cap
# Chunk ids are stable per page URL and chunk index, so a re-run updates documents in place
documents = [
    discoveryengine.Document(
        id=chunk["id"],
        struct_data={
            "content": chunk["content"],
            "title": chunk["metadata"]["title"],
            "section": chunk["metadata"]["section"],
            # headings_path is a list; stored as one breadcrumb string so it is a single field
            "headings_path": " > ".join(chunk["metadata"]["headings_path"]),
            "source_url": chunk["metadata"]["source_url"],
            "chunk_index": chunk["metadata"]["chunk_index"],
            "word_count": chunk["metadata"]["word_count"],
        },
    )
    for chunk in chunks
]

# Step 5: Import into Vertex AI Search in batches of 100 (the recommended maximum per inline import)
# Uses Application Default Credentials (gcloud auth application-default login)
# This store has no schema yet, so Vertex AI Search auto-detects the fields on import, and
# auto-detected fields are indexable, searchable and retrievable (up to 50 fields each)
# Only if you supplied your own schema do you set those annotations yourself (data store Schema tab)
client = discoveryengine.DocumentServiceClient(
    client_options=(
        ClientOptions(api_endpoint=f"{LOCATION}-discoveryengine.googleapis.com")
        if LOCATION != "global"
        else None
    )
)
parent = client.branch_path(
    project=PROJECT_ID,
    location=LOCATION,
    data_store=DATASTORE_ID,
    branch="default_branch",
)

for start in range(0, len(documents), 100):
    request = discoveryengine.ImportDocumentsRequest(
        parent=parent,
        inline_source=discoveryengine.ImportDocumentsRequest.InlineSource(
            documents=documents[start:start + 100]
        ),
        reconciliation_mode=discoveryengine.ImportDocumentsRequest.ReconciliationMode.INCREMENTAL,
    )
    result = client.import_documents(request=request).result()
    # A partly failed import does not raise, so check the error samples
    if result.error_samples:
        raise RuntimeError(f"Vertex AI Search import failed: {result.error_samples[0].message}")

print(f"Imported {len(documents)} chunks into Vertex AI Search")