How to Ingest a Technical Manual with OpenAI Agents SDK

import json
import os
import re
import time

import requests
from agents import Agent, Runner, function_tool

# The OpenAI Agents SDK reads OPENAI_API_KEY from the environment.
HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}
MANUAL_PATH = "/path/to/your/installation-manual.pdf"  # replace with your manual

# Step 1: Upload the technical manual
# Text-based PDFs only; scanned or image-only manuals are not processed
with open(MANUAL_PATH, "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": (os.path.basename(MANUAL_PATH), f, "application/pdf")},
        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. Plain GET, no API key.
output = requests.get(output_url, timeout=60)
output.raise_for_status()

# Split the raw bytes: chunk text may contain U+2028, which str.splitlines() splits on
chunks = [json.loads(line) for line in output.content.splitlines() if line.strip()]
print(f"Loaded {len(chunks)} chunks ready for retrieval")


# Step 4: Function tool: keyword search with an optional section filter
@function_tool
def search_manual(query: str, section: str = "") -> str:
    """Search the technical manual for passages matching a query.
    Pass a section name to restrict the search to one part of the manual.
    Each result shows its heading path and section so answers can be cited.

    Args:
        query: Keywords describing the procedure or requirement needed.
        section: Optional substring of a section or heading path to filter by.
    """
    terms = set(re.findall(r"\w+", query.casefold()))
    section_filter = section.casefold().strip()
    ranked = []

    for chunk in chunks:
        content = chunk["content"]
        metadata = chunk["metadata"]
        # headings_path is a list of heading strings, e.g. ["Installation", "Safety"]
        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[:5]
    ]
    return "\n\n---\n\n".join(results) if results else "No matching passages found."


# Replace the keyword scoring above with embedding search for a larger manual

# Step 5: Set up the agent
agent = Agent(
    name="Manual Assistant",
    instructions=(
        "You answer questions about a technical manual. "
        "Use search_manual to find relevant passages before answering. "
        "Narrow with the section argument when the question names a specific part of the manual. "
        "Cite the heading path of every passage you rely on. "
        "Do not invent procedures or safety requirements not present in retrieved passages."
    ),
    tools=[search_manual],
)

result = Runner.run_sync(
    agent,
    "What are the safety requirements before installing the unit?",
)
print(result.final_output)