Best Captain Alternative for RAG Pipeline Ingestion

import json
import os
import time

import requests

PDF_PATH = "/path/to/your/contract.pdf"

# Captain: index a PDF into a managed collection, then query it
CAPTAIN_URL = "https://api.filesearch.dev"
CAPTAIN_COLLECTION = "contracts"
# The organisation is derived from the API key, so X-Organization-ID is optional
captain_headers = {"Authorization": f"Bearer {os.environ['CAPTAIN_API_KEY']}"}

# Step 1: Create the collection (idempotent: 201 if created, 200 if it exists)
requests.put(
    f"{CAPTAIN_URL}/v2/collections/{CAPTAIN_COLLECTION}",
    headers=captain_headers,
    json={},
    timeout=30,
).raise_for_status()

# Step 2: Upload the PDF for indexing
with open(PDF_PATH, "rb") as f:
    index_response = requests.post(
        f"{CAPTAIN_URL}/v2/collections/{CAPTAIN_COLLECTION}/index/file",
        headers=captain_headers,
        files={"files": ("contract.pdf", f, "application/pdf")},
        data={"processing_type": "advanced"},
        timeout=120,
    )
index_response.raise_for_status()
captain_job_id = index_response.json()["job_id"]

# Step 3: Poll until indexing completes
while True:
    captain_status = requests.get(
        f"{CAPTAIN_URL}/v2/jobs/{captain_job_id}",
        headers=captain_headers,
        timeout=30,
    )
    captain_status.raise_for_status()
    captain_job = captain_status.json()
    if captain_job["status"] == "completed":
        break
    if captain_job["status"] in ("completed_with_errors", "failed", "cancelled", "timed_out"):
        raise RuntimeError(
            f"Captain indexing {captain_job['status']}: "
            f"{captain_job.get('error_message') or captain_job.get('result')}"
        )
    time.sleep(5)

# Step 4: Query the managed collection
query_response = requests.post(
    f"{CAPTAIN_URL}/v3/collections/{CAPTAIN_COLLECTION}/query",
    headers=captain_headers,
    json={"query": "What are the termination terms?", "limit": 10},
    timeout=60,
)
query_response.raise_for_status()
for hit in query_response.json()["results"]:
    print(f"{hit['score']:.3f}  {hit['text'][:120]}")

# Chunking, embedding, storage, and ranking are managed by Captain

# EnConvert: ingest the PDF, then load chunks into the store you control
enconvert_headers = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}

# Step 1: Upload the PDF (multipart field is "files"; always answers 202 with a job_id)
with open(PDF_PATH, "rb") as f:
    submit_response = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=enconvert_headers,
        files={"files": ("contract.pdf", f, "application/pdf")},
        timeout=60,
    )
submit_response.raise_for_status()
job_id = submit_response.json()["job_id"]

# Step 2: Poll until the job completes
while True:
    status_response = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=enconvert_headers,
        timeout=30,
    )
    status_response.raise_for_status()
    job = status_response.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(f"Ingest {job['status']}: {job.get('error_message')}")
    time.sleep(2)

# Step 3: Download the JSONL, one chunk per line. output_url is a signed URL
# (valid 15 minutes): plain GET, no API key.
jsonl_response = requests.get(job["output_url"], timeout=60)
jsonl_response.raise_for_status()
# Split on "\n" only: str.splitlines() would also break on U+2028 inside chunk text
chunks = [
    json.loads(line)
    for line in jsonl_response.content.decode("utf-8").split("\n")
    if line.strip()
]

# Step 4: Map chunks to the inputs a vector store expects
ids = [chunk["id"] for chunk in chunks]
texts = [chunk["content"] for chunk in chunks]
# headings_path is a list; many stores accept only scalar metadata, so join it
metadatas = [
    {
        **chunk["metadata"],
        "headings_path": " > ".join(chunk["metadata"]["headings_path"]),
    }
    for chunk in chunks
]

print(f"Loaded {len(chunks)} chunks ready for embedding")

# Embed texts with your embedding model, then add ids, texts and metadatas
# to the vector store you choose