How to Ingest Documentation Sites into AutoGen

import json
import os
import time
import requests
from collections import defaultdict
from autogen import AssistantAgent, UserProxyAgent

API_KEY = os.environ["ENCONVERT_API_KEY"]
HEADERS = {"X-API-Key": API_KEY, "Content-Type": "application/json"}

# Step 1: Submit a sitemap to the Ingest endpoint
submit_response = requests.post(
    "https://api.enconvert.com/v2/ingest",
    headers=HEADERS,
    json={
        "mode": "sitemap",
        "url": os.environ["DOCS_SITEMAP_URL"],
    },
)
if submit_response.status_code == 402:
    raise RuntimeError("Ingest is not on your plan, or the monthly ops allowance is exhausted")
submit_response.raise_for_status()
job_id = submit_response.json()["job_id"]

# Step 2: Poll until the crawl job reaches a terminal state
# Pages are processed one at a time; allow longer polling for large sites
while True:
    status_response = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
    )
    status_response.raise_for_status()
    job = status_response.json()

    if job["status"] == "completed":
        break
    if job["status"] in {"failed", "canceled"}:
        raise RuntimeError(
            job.get("error_message") or f"Ingest job {job_id} ended with status {job['status']}"
        )
    time.sleep(15)

# Step 3: Download JSONL output from signed output_url (expires after 15 minutes)
jsonl_response = requests.get(job["output_url"])
jsonl_response.raise_for_status()
chunks = [json.loads(line) for line in jsonl_response.text.splitlines() if line.strip()]
print(f"Loaded {len(chunks)} chunks across the crawl")

# Step 4: Group chunks by source_url so the agent knows which page each chunk came from
by_page = defaultdict(list)
for chunk in chunks:
    by_page[chunk["metadata"]["source_url"]].append(chunk)

context_parts = []
for url, page_chunks in list(by_page.items())[:8]:
    section_block = "\n\n".join(
        f"[section: {c['metadata']['section']} | chunk: {c['metadata']['chunk_index']}]\n"
        f"{c.get('content')}"
        for c in page_chunks[:4]
    )
    context_parts.append(f"### Page: {url}\n\n{section_block}")

context = "\n\n".join(context_parts)

# Step 5: Pass the grouped context into an AutoGen agent
assistant = AssistantAgent(
    name="docs_analyst",
    llm_config={"config_list": [{"model": "gpt-4.1", "api_key": os.environ["OPENAI_API_KEY"]}]},
    system_message=f"""Answer questions about the product documentation below.
Cite the page URL and section name for each answer.

Documentation chunks grouped by page:
{context}
""",
)

user = UserProxyAgent(
    name="user",
    human_input_mode="NEVER",
    code_execution_config=False,
)
user.initiate_chat(assistant, message="Which page covers authentication setup, and what does it say?")