How to Ingest Markdown into Weaviate with EnConvert

import os
import json
import time

import requests
import weaviate
from sentence_transformers import SentenceTransformer
from weaviate.classes.config import Configure, DataType, Property, Tokenization
from weaviate.classes.init import Auth
from weaviate.classes.query import Filter
from weaviate.util import generate_uuid5

HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}
WEAVIATE_URL = os.environ["WEAVIATE_URL"]  # e.g. your-cluster.weaviate.cloud
WEAVIATE_API_KEY = os.environ["WEAVIATE_API_KEY"]
MARKDOWN_PATH = "/path/to/your/product-guide.md"
COLLECTION = "MarkdownChunks"

# Step 1: Upload the Markdown file (the form field is "files", plural)
# max_words is the soft cap on words per chunk (default 512). 150 words keeps
# prose chunks inside the 256 word-piece input limit of the embedding model in Step 4.
with open(MARKDOWN_PATH, "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": (os.path.basename(MARKDOWN_PATH), f, "text/markdown")},
        data={"max_words": 150},
        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:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=30,
    )
    poll.raise_for_status()
    job = poll.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 RuntimeError(f"Ingest {job_id} still {job['status']} after 10 minutes")
    time.sleep(2)

# Step 3: Download the JSONL output
# The signed URL expires after 15 minutes and is fetched with a plain GET, no API key.
download = requests.get(job["output_url"], timeout=60)
download.raise_for_status()

# Split on "\n" only: splitlines() would also break on U+2028 inside chunk text.
chunks = [
    json.loads(line)
    for line in download.content.decode("utf-8").split("\n")
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for embedding")

# Step 4: Embed the chunk text
# Chunks split at H1 to H3 boundaries. all-MiniLM-L6-v2 truncates input longer than
# 256 word pieces, which is why Step 1 caps chunks at 150 words. Tables and code
# blocks are kept whole even when they exceed the cap, so a very long one is
# embedded from its first part only.
model = SentenceTransformer("all-MiniLM-L6-v2")
vectors = model.encode([chunk["content"] for chunk in chunks], normalize_embeddings=True)

# Step 5: Connect to Weaviate and create the collection if needed
with weaviate.connect_to_weaviate_cloud(
    cluster_url=WEAVIATE_URL,
    auth_credentials=Auth.api_key(WEAVIATE_API_KEY),
) as client:
    if not client.collections.exists(COLLECTION):
        client.collections.create(
            name=COLLECTION,
            vector_config=Configure.Vectors.self_provided(),
            properties=[
                Property(name="content", data_type=DataType.TEXT),
                Property(name="title", data_type=DataType.TEXT),
                Property(
                    name="section",
                    data_type=DataType.TEXT,
                    tokenization=Tokenization.FIELD,  # exact match on the heading
                ),
                Property(name="headings_path", data_type=DataType.TEXT_ARRAY),
                Property(name="source_url", data_type=DataType.TEXT),
                Property(name="chunk_index", data_type=DataType.INT),
                Property(name="word_count", data_type=DataType.INT),
            ],
        )

    collection = client.collections.use(COLLECTION)

    # Step 6: Insert each chunk as an object, with its metadata as properties
    # The object id comes from the file name plus the chunk index, so re-running
    # the script on the same file updates its objects in place instead of duplicating them.
    # Use a distinct file name per document, and note that chunks left over from an
    # older, longer version of the same file are not removed.
    # (chunk["id"] is not used: for uploaded files it changes on every ingest job.)
    with collection.batch.fixed_size(batch_size=32) as batch:
        for chunk, vector in zip(chunks, vectors):
            metadata = chunk["metadata"]
            batch.add_object(
                properties={
                    "content": chunk["content"],
                    "title": metadata["title"],
                    "section": metadata["section"],
                    "headings_path": metadata["headings_path"],  # a list of strings
                    "source_url": metadata["source_url"],  # the file name
                    "chunk_index": metadata["chunk_index"],
                    "word_count": metadata["word_count"],
                },
                uuid=generate_uuid5(
                    f"{metadata['source_url']}:{metadata['chunk_index']}"
                ),
                vector=vector.tolist(),
            )
    if collection.batch.failed_objects:
        raise RuntimeError(
            f"{len(collection.batch.failed_objects)} chunks failed to insert"
        )
    print(f"Inserted {len(chunks)} chunks into {COLLECTION}")

    # Step 7: Search, restricted to a single section
    # Replace "Installation" with a heading from your own file (a chunk's metadata.section).
    query_vector = model.encode("How do I install it?", normalize_embeddings=True)
    results = collection.query.near_vector(
        near_vector=query_vector.tolist(),
        filters=Filter.by_property("section").equal("Installation"),
        limit=3,
        return_properties=["content", "section", "headings_path"],
    )
    for obj in results.objects:
        print(" > ".join(obj.properties["headings_path"]))
        print(obj.properties["content"][:200])