-
EAP – Document Chunking Process
EAP – Document Chunking Why We Need Chunking ? # ── STEP 2: Chunk parsed files ───────────────────────── chunked_prefix = f"{tenant_id}/{connector_id}/chunked" logger.info("🧩 Chunking Parsed Files") t0 = perf_counter() chunk_result = await chunk_files( input_bucket=bucket, input_prefix=parsed_prefix, output_bucket=bucket, output_prefix=chunked_prefix, ) timings["chunking"] = perf_counter() – t0 logger.info( f"Chunking complete [{_fmt(timings['chunking'])}]: " f"processed={chunk_result['processed']} | " f"success={chunk_result['succeeded']} | " f"failed={chunk_result['failed']}" ) all_chunks = [ {"file_name": item.get("gcs_uri", ""), "body": chunk["body"]} for item in chunk_result["results"] if item["status"] == "success" for chunk in item["chunks"] ] logger.info(f"🧩 Total Chunks Collected: {len(all_chunks)}") from __future__ import annotations import asyncio import json import logging import re from pathlib import Path from typing import Any
-
EAP – Document AI Parser
EAP – Document AI Parser # ── STEP 1: Parse files via DocAI ───────────────────────── parsed_prefix = f"{tenant_id}/{connector_id}/parsed" logger.info("📄 Parsing Files With DocAI") t0 = perf_counter() parse_result = await parse_files( input_bucket=bucket, input_prefix=base_path, engine="docai", output_prefix_base=parsed_prefix, ) timings["parsing"] = perf_counter() – t0 logger.info( f"Parsing complete [{_fmt(timings['parsing'])}]: " f"processed={parse_result['processed']} | " f"success={parse_result['succeeded']} | " f"failed={parse_result['failed']}" ) async def parse_files( input_bucket: str, engine: str = "mistral", input_prefix: str | None = None, output_bucket: str | None = None, output_prefix_base: str | None = "parsed", ) -> dict: """ Parse files from an input GCS bucket and upload parsed outputs. Files are discovered by listing blobs
