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

Read More