How to Ingest a PDF into LangChain with EnConvert

import os
import json
import time

import requests
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

ENCONVERT_API_KEY = os.environ["ENCONVERT_API_KEY"]
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
HEADERS = {"X-API-Key": ENCONVERT_API_KEY}

# Step 1: Upload the PDF (the multipart field is named "files"; the job is queued)
with open("/path/to/your/product-specification.pdf", "rb") as pdf_file:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": ("product-specification.pdf", pdf_file, "application/pdf")},
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job reaches a final status (give up after 10 minutes)
deadline = time.monotonic() + 600
while True:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=60,
    )
    poll.raise_for_status()
    job = poll.json()
    if job["status"] == "completed":
        output_url = job["output_url"]
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(
            f"Ingest job {job_id} {job['status']}: {job.get('error_message')}"
        )
    if time.monotonic() > deadline:
        raise TimeoutError(f"Ingest job {job_id} is still {job['status']}")
    time.sleep(3)

# Step 3: Download the JSONL output. The output URL is signed (valid 15 minutes):
# plain GET, no API key.
jsonl_response = requests.get(output_url, timeout=60)
jsonl_response.raise_for_status()

# Step 4: Load each chunk as a LangChain Document with metadata preserved
documents = []
# Splitting the raw bytes only breaks on newlines, so a chunk's text can never
# split a JSON line.
for line in jsonl_response.content.splitlines():
    if not line.strip():
        continue
    chunk = json.loads(line)
    documents.append(
        Document(
            id=chunk["id"],
            page_content=chunk["content"],
            # source_url, title, headings_path, section, word_count, chunk_index.
            # InMemoryVectorStore keeps metadata as plain Python objects, so the
            # headings_path list is stored as is.
            metadata=chunk["metadata"],
        )
    )

print(f"Loaded {len(documents)} chunks ready for embedding")
print("Sections:", sorted({doc.metadata["section"] for doc in documents}))

# Step 5: Embed and add to a vector store
# (InMemoryVectorStore uses numpy for cosine search, so numpy must be installed)
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=OPENAI_API_KEY,
)
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents)

# Step 6: Search only inside one section of the PDF
SECTION = "Operating Conditions"  # replace with a section printed above for your PDF
results = vector_store.similarity_search(
    "operating temperature limits",
    k=3,
    filter=lambda doc: doc.metadata.get("section") == SECTION,
)
for doc in results:
    print(doc.metadata["section"], doc.metadata["headings_path"])
    print(doc.page_content[:200])