How to Ingest HTML Files into LangChain with EnConvert

import json
import os
import time

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

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

# Step 1: Upload the HTML file to Ingest
# Navigation and footer chrome is stripped where detected
with open("/path/to/your/exported-docs-page.html", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": ("exported-docs-page.html", f, "text/html")},
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes (give up 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=60,
    )
    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')}")
    if time.monotonic() > deadline:
        raise TimeoutError(f"Ingest {job_id} is still {job['status']} after 10 minutes")
    time.sleep(2)

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

# Split on newline only: str.splitlines() also breaks on characters like U+0085
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 of strings. A store that accepts only scalar metadata
# needs " > ".join(headings_path) instead.
documents = [
    Document(
        page_content=chunk["content"],
        metadata={
            "source_url": chunk["metadata"]["source_url"],
            "title": chunk["metadata"]["title"],
            "section": chunk["metadata"]["section"],
            "headings_path": chunk["metadata"]["headings_path"],
            "chunk_index": chunk["metadata"]["chunk_index"],
            "word_count": chunk["metadata"]["word_count"],
        },
    )
    for chunk in chunks
]

# Step 5: Embed the chunks into an in-memory vector store and save it locally
# InMemoryVectorStore ships with langchain-core; its search needs numpy installed
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key=os.environ["OPENAI_API_KEY"],
)
vectorstore = InMemoryVectorStore.from_documents(documents, embeddings)
vectorstore.dump("./html_index.json")

# Step 6: Similarity search - results carry the section they came from
results = vectorstore.similarity_search("how do I rotate an API key", k=3)
for doc in results:
    print(doc.metadata["section"])
    print(" > ".join(doc.metadata["headings_path"]))
    print(doc.page_content[:200])

# Step 7: Restrict retrieval to one section of the file
# Replace with a section heading from your own file (printed in Step 6)
SECTION = "Rotating API keys"
section_results = vectorstore.similarity_search(
    "how do I rotate an API key",
    k=3,
    filter=lambda doc: doc.metadata["section"] == SECTION,
)
for doc in section_results:
    print(doc.metadata["section"], "-", doc.page_content[:200])

# To reload the saved index later:
# vectorstore = InMemoryVectorStore.load("./html_index.json", embeddings)