How to Ingest Web Pages into Vertex AI with EnConvert

import os
import json
import time

import requests
from google import genai
from google.cloud.aiplatform_v1 import IndexServiceClient
from google.cloud.aiplatform_v1.types import IndexDatapoint, UpsertDatapointsRequest
from google.genai.types import EmbedContentConfig

ENCONVERT_HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}
PAGE_URLS = [url.strip() for url in os.environ["PAGE_URLS"].split(",")]  # public page URLs

# Google Cloud calls use Application Default Credentials (gcloud auth application-default login)
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
LOCATION = os.environ["GOOGLE_CLOUD_LOCATION"]  # region of your index, for example us-central1
INDEX_ID = os.environ["VECTOR_SEARCH_INDEX_ID"]  # streaming-update index with 3072 dimensions

# Step 1: Submit the page URLs to EnConvert Ingest
# Public pages only; pages that render as a login wall or bot challenge are skipped
submit = requests.post(
    "https://api.enconvert.com/v2/ingest",
    headers=ENCONVERT_HEADERS,
    json={"mode": "urls", "urls": PAGE_URLS},
    timeout=60,
)
submit.raise_for_status()
job_id = submit.json()["job_id"]

# Step 2: Poll until the job completes
# Pages render one at a time, so poll at 10 second intervals
deadline = time.monotonic() + 900
while True:
    status = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=ENCONVERT_HEADERS,
        timeout=30,
    )
    status.raise_for_status()
    job = status.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(job.get("error_message") or f"Ingest job {job['status']}")
    if time.monotonic() >= deadline:
        raise TimeoutError(f"Job {job_id} is still running; resume polling later")
    time.sleep(10)

# Step 3: Download the JSONL output
# The signed URL expires after 15 minutes. Plain GET, no API key.
output = requests.get(job["output_url"], timeout=120)
output.raise_for_status()
chunks = [json.loads(line) for line in output.iter_lines() if line]
print(f"Loaded {len(chunks)} chunks from {job['pages_processed']} pages")

# Step 4: Embed each chunk with Vertex AI
# gemini-embedding-001 takes one text per request and returns 3072 dimensions by default
genai_client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)

vectors = []
for chunk in chunks:
    response = genai_client.models.embed_content(
        model="gemini-embedding-001",
        contents=[chunk["content"]],
        config=EmbedContentConfig(task_type="RETRIEVAL_DOCUMENT"),
    )
    vectors.append(response.embeddings[0].values)

# Step 5: Build datapoints
# Vector Search stores vectors and filterable restricts, not text payloads
# Keep chunk content and metadata keyed by datapoint ID in a sidecar store
datapoints = []
chunk_store = {}

for chunk, vector in zip(chunks, vectors):
    metadata = chunk["metadata"]

    # chunk["id"] is deterministic per source URL and chunk index, so a re-run
    # overwrites the same datapoints instead of duplicating them
    restricts = [
        IndexDatapoint.Restriction(
            namespace="source_url", allow_list=[metadata["source_url"]]
        )
    ]
    if metadata.get("section"):  # content before the first heading has no section
        restricts.append(
            IndexDatapoint.Restriction(
                namespace="section", allow_list=[metadata["section"]]
            )
        )

    datapoints.append(
        IndexDatapoint(
            datapoint_id=chunk["id"],
            feature_vector=vector,
            restricts=restricts,
        )
    )
    chunk_store[chunk["id"]] = {"content": chunk["content"], "metadata": metadata}

# Step 6: Upsert into the index in batches
# The index must be created with streaming updates enabled
index_client = IndexServiceClient(
    client_options={"api_endpoint": f"{LOCATION}-aiplatform.googleapis.com"}
)
index_name = f"projects/{PROJECT_ID}/locations/{LOCATION}/indexes/{INDEX_ID}"

for start in range(0, len(datapoints), 100):
    index_client.upsert_datapoints(
        request=UpsertDatapointsRequest(
            index=index_name,
            datapoints=datapoints[start : start + 100],
        ),
        timeout=60,
    )

# Save sidecar to disk; replace with Firestore or BigQuery in production
with open("chunk_store.json", "w", encoding="utf-8") as out:
    json.dump(chunk_store, out, ensure_ascii=False)

print(f"Upserted {len(datapoints)} chunks into Vertex AI Vector Search")