How to Ingest a Financial Report with CrewAI and EnConvert

import os
import json
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

# CrewAI's default LLM reads OPENAI_API_KEY from the environment
HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}

# Step 1: Upload the financial report to EnConvert Ingest
# Text-based PDFs only; scanned or image-only reports are not processed
with open("/path/to/your/annual-report.pdf", "rb") as report_file:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": ("annual-report.pdf", report_file, "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.
jsonl_response = requests.get(output_url, timeout=60)
jsonl_response.raise_for_status()
jsonl_response.encoding = "utf-8"  # the file is UTF-8 but the response has no charset

chunks = [
    json.loads(line)
    for line in jsonl_response.text.splitlines()
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks ready for retrieval")


# Step 4: Define the input schema and BaseTool
class ReportSearchInput(BaseModel):
    query: str = Field(
        description="What to look up, such as total revenue or supply chain risk"
    )
    section: Optional[str] = Field(
        default=None,
        description="Optional section name to restrict the search, such as Risk Factors",
    )
    limit: int = Field(
        default=3,
        ge=1,
        le=5,
        description="Maximum number of passages to return",
    )


class SearchReportTool(BaseTool):
    name: str = "search_report"
    description: str = (
        "Search the ingested financial report 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] = ReportSearchInput

    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, so join it into a breadcrumb
            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

            # Keyword overlap keeps this runnable as-is; use embedding search for a larger report
            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."


# Step 5: Set up the agent
analyst = Agent(
    role="Financial Report Analyst",
    goal="Answer questions about the report using only its text and cite the section for every figure",
    backstory="You read company reports closely and never state a number you cannot trace to a section.",
    tools=[SearchReportTool()],
    verbose=True,
)

# Step 6: Create a task and run the crew
task = Task(
    description=(
        "Find the reported revenue and the main risk factors in the report "
        "and cite the section each came from. "
        "Do not invent figures or citations not present in retrieved passages."
    ),
    expected_output="A short summary of revenue and key risk factors, each with its section citation.",
    agent=analyst,
)

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