How to Ingest DOCX into CrewAI with EnConvert

import json
import os
import re
import time
from typing import Optional, Type

import requests
from crewai import Agent, Crew, Task
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

ENCONVERT_API_KEY = os.environ["ENCONVERT_API_KEY"]
# CrewAI's default LLM reads OPENAI_API_KEY from the environment.
# Pass llm=... to Agent to use another model.
HEADERS = {"X-API-Key": ENCONVERT_API_KEY}

# Step 1: Upload the DOCX (multipart form, the field is named "files")
with open("/path/to/your/product-specification.docx", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={
            "files": (
                "product-specification.docx",
                f,
                "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
            )
        },
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.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=HEADERS,
        timeout=60,
    )
    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']}")
    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.
jsonl = requests.get(output_url, timeout=60)
jsonl.raise_for_status()

# Decode as UTF-8 and split on "\n" only: chunk text can contain Unicode line separators.
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 retrieval")


# Step 4: Define the input schema and BaseTool
class DocSearchInput(BaseModel):
    query: str = Field(
        description="What to look up, such as supported operating systems or access control requirements"
    )
    section: Optional[str] = Field(
        default=None,
        # Replace with a heading from your own document
        description="Optional section name to restrict the search, such as Security",
    )
    limit: int = Field(
        default=3,
        ge=1,
        le=5,
        description="Maximum number of passages to return",
    )


class SearchDocTool(BaseTool):
    name: str = "search_document"
    description: str = (
        "Search the ingested DOCX and return the most relevant passages, "
        "each labelled with its heading path and section so answers can cite where they came from."
    )
    args_schema: Type[BaseModel] = DocSearchInput

    def _run(self, query: str, section: Optional[str] = None, limit: int = 3) -> str:
        terms = set(re.findall(r"\w+", query.casefold()))
        section_filter = section.casefold().strip() if section else ""
        ranked = []

        for chunk in chunks:
            content = chunk["content"]
            metadata = chunk["metadata"]
            # headings_path is a list of heading strings and can be empty
            breadcrumb = " > ".join(metadata.get("headings_path", []))
            section_name = metadata.get("section", "")

            if section_filter and section_filter not in f"{breadcrumb} {section_name}".casefold():
                continue

            searchable = f"{section_name} {breadcrumb} {content}".casefold()
            score = len(terms & set(re.findall(r"\w+", searchable)))

            if score or section_filter:
                ranked.append((score, chunk))

        ranked.sort(key=lambda item: item[0], reverse=True)

        results = [
            f"[Path: {' > '.join(c['metadata'].get('headings_path', [])) or 'no heading'} "
            f"| Section: {c['metadata'].get('section') or 'none'}]\n"
            f"{c['content']}"
            for _, c in ranked[:limit]
        ]
        return "\n\n---\n\n".join(results) if results else "No matching passages found."

# Keyword overlap keeps this example dependency-free. Use embedding search for a larger document.

# Step 5: Set up the agent
analyst = Agent(
    role="Product Specification Analyst",
    goal="Answer questions about the specification using only its text and cite the section for every statement",
    backstory="You read product documents closely and never state a requirement you cannot trace to a section.",
    tools=[SearchDocTool()],
    verbose=True,
)

# Step 6: Create a task and run the crew
task = Task(
    description=(
        "Find the security requirements and supported operating systems in the document "
        "and cite the section each came from. "
        "Do not invent requirements not present in retrieved passages."
    ),
    expected_output="A short summary of security requirements and supported systems, each with its section citation.",
    agent=analyst,
)

crew = Crew(agents=[analyst], tasks=[task], verbose=True)
print(crew.kickoff())