How to Ingest a Technical Manual into LangChain with EnConvert

import os
import json
import time

import requests
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings

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

# Step 1: Upload the technical manual
# Text-based PDFs only; scanned or image-only manuals are not processed
with open("/path/to/your/installation-manual.pdf", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": ("installation-manual.pdf", f, "application/pdf")},
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes, failing on error or after 10 minutes
deadline = time.monotonic() + 600
while True:
    status_response = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=30,
    )
    status_response.raise_for_status()
    job = status_response.json()
    if job["status"] == "completed":
        output_url = job["output_url"]
        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"Ingest job {job_id} is still {job['status']} after 10 minutes")
    time.sleep(2)

# Step 3: Download the JSONL output
# The signed URL expires after 15 minutes. Plain GET, no API key.
jsonl = requests.get(output_url, timeout=60)
jsonl.raise_for_status()

# The file is UTF-8, one JSON object per line
chunks = [
    json.loads(line)
    for line in jsonl.content.decode("utf-8").split("\n")
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for embedding")

# Step 4: Load each chunk as a LangChain Document with metadata preserved
# Do not apply another text splitter - heading-aware boundaries are already set
# headings_path is a list and can be empty, and Chroma metadata does not accept empty
# arrays, so it is stored as a breadcrumb string
documents = [
    Document(
        page_content=chunk["content"],
        metadata={
            "source_url": chunk["metadata"]["source_url"],
            "title": chunk["metadata"]["title"],
            "section": chunk["metadata"]["section"],
            "headings_path": " > ".join(chunk["metadata"]["headings_path"]),
            "chunk_index": chunk["metadata"]["chunk_index"],
            "word_count": chunk["metadata"]["word_count"],
        },
    )
    for chunk in chunks
]

# Step 5: Add to a Chroma vector store persisted on disk
# Chunk ids from uploaded files are new on every ingest job, so delete ./chroma_manual_index
# before re-ingesting the same file, or the old chunks stay in the collection
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=os.environ["OPENAI_API_KEY"],
)
vectorstore = Chroma.from_documents(
    documents,
    embedding=embeddings,
    ids=[chunk["id"] for chunk in chunks],
    collection_name="technical_manual",
    persist_directory="./chroma_manual_index",
)

# Step 6: Similarity search restricted to one section by a metadata filter
# The section headings found in your manual, to pick a filter value from
print(sorted({doc.metadata["section"] for doc in documents}))

# Chroma applies the filter before the nearest-neighbour search, so every result comes
# from the chosen section (fewer than k if the section is small)
results = vectorstore.similarity_search(
    "what must be checked before powering on the unit",
    k=3,
    # Replace with a section heading from your own manual (see the list above)
    filter={"section": "Safety Precautions"},
)
for doc in results:
    print(doc.metadata["section"])
    print(doc.metadata["headings_path"])
    print(doc.page_content[:200])

# To reload the index later:
# vectorstore = Chroma(
#     collection_name="technical_manual",
#     embedding_function=embeddings,
#     persist_directory="./chroma_manual_index",
# )