diff --git a/ai/ai-document-understanding/du-parallel-pipeline/LICENSE b/ai/ai-document-understanding/du-parallel-pipeline/LICENSE new file mode 100644 index 0000000000..bb91ea7806 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/LICENSE @@ -0,0 +1,35 @@ +Copyright (c) 2026 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/ai/ai-document-understanding/du-parallel-pipeline/README.md b/ai/ai-document-understanding/du-parallel-pipeline/README.md new file mode 100644 index 0000000000..b4122d8b05 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/README.md @@ -0,0 +1,21 @@ +# Parallel Processing Pipeline for Long Documents with OCI Document Understanding + +OCI Document Understanding's synchronous API accepts at most 5 pages per request, and the asynchronous Object Storage path adds queueing latency that interactive applications cannot afford. This asset provides a Python pipeline that processes PDFs of any length at synchronous speed: it splits the document into 5-page chunks, analyzes all chunks in parallel synchronous requests, and merges the results back into a single response with correct document-level page numbering. Table extraction (with CSV export) and OCR text extraction are supported, and a benchmark suite compares sequential, parallel, and asynchronous processing on the same document. + +Reviewed: 25.08.2026 + +# When to use this asset? + +See the README document in the /files folder. + +# How to use this asset? + +See the README document in the /files folder. + +# License + +Copyright (c) 2026 Oracle and/or its affiliates. + +Licensed under the Universal Permissive License (UPL), Version 1.0. + +See [LICENSE](https://github.com/oracle-devrel/technology-engineering/blob/main/LICENSE) for more details. diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/README.md b/ai/ai-document-understanding/du-parallel-pipeline/files/README.md new file mode 100644 index 0000000000..79c92b316e --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/README.md @@ -0,0 +1,153 @@ +# Parallel Processing Pipeline for Long Documents + +A Python pipeline that lets OCI Document Understanding process documents **larger than the 5-page synchronous limit** — at synchronous speed — by splitting the PDF into chunks, running the chunks as **parallel synchronous requests**, and merging the results into one coherent response. + +_This asset is a reference implementation: review error handling, authentication, and logging against your own production standards before deploying._ + +Reviewed: 25.08.2026 + +## When to use this asset? + +Use this pipeline when your documents are longer than 5 pages **and** you need results in seconds, not minutes: + +| Path | Max pages | Typical latency | Best for | +|---|---|---|---| +| Synchronous request | 5 | seconds | short documents | +| **This pipeline (parallel sync)** | any (chunked) | **seconds** | **long documents, interactive apps** | +| Asynchronous Object Storage job | 2,000 | minutes (queue + poll) | huge batches, no latency requirement | + +The Document Understanding service is **stateless** — there is no "send the next 5 pages" continuation mechanism. This pipeline keeps track on the client side instead: each chunk remembers its position in the original document, and the merger restores true page numbers, so the output looks as if one call had processed the whole document. + +``` + ┌──────────────┐ ┌─── chunk 1 (p. 1-5) ──► analyze_document ───┐ + long PDF ───►│ splitter │────┼─── chunk 2 (p. 6-10) ──► analyze_document ───┼───► merger ───► one JSON + │ (pypdf, ≤5p, │ └─── chunk N (...) ──► analyze_document ───┘ (true page + CSV + │ ≤8 MB each) │ ThreadPoolExecutor, N parallel calls numbers) tables + └──────────────┘ +``` + +### Features + +- **Table extraction** (headline feature): tables from all chunks are re-indexed to document-level page numbers and can be exported as one CSV per table. +- **Text extraction (OCR)**: full per-page text, merged the same way. +- **Key-value extraction**: supported per chunk; note that document-level fields (e.g. `InvoiceTotal`) are answered once per chunk, so downstream logic must pick the right candidate. +- **Rate-limit aware**: bounded worker pool (default 4) plus the OCI SDK default retry strategy (exponential backoff on HTTP 429/5xx), keeping the pipeline polite against the per-tenancy transaction limit. +- **Adaptive splitting**: chunks that exceed the 8 MB synchronous request limit are automatically halved until they fit. + +## Setup + +1. Install Python 3.10+ and the requirements: + ```bash + pip install -r requirements.txt + ``` +2. Configure OCI API-key authentication (`~/.oci/config`) for a region where Document Understanding is available ([documentation](https://docs.oracle.com/en-us/iaas/Content/API/Concepts/sdk_authentication_methods.htm)). +3. Set your compartment OCID (never hardcode it): + ```bash + export OCI_COMPARTMENT_ID= + # optional overrides: + export OCI_CONFIG_PROFILE=DEFAULT + export OCI_CONFIG_FILE=~/.oci/config + ``` +4. Make sure you have the appropriate IAM policies for Document Understanding ([documentation](https://docs.oracle.com/en-us/iaas/Content/document-understanding/using/about_document-understanding_policies.htm)). + +## Usage + +Analyze a long PDF with table + text extraction, 4 parallel workers: + +```bash +python -m du_pipeline.cli report_40_pages.pdf \ + --features table,text \ + --workers 4 \ + --output result.json \ + --tables-dir tables/ +``` + +Output: + +``` +report_40_pages.pdf: 40 pages -> 8 chunk(s), 4 worker(s), features: table, text + chunk 2/8 (pages 6-10) done in 4.2s + chunk 1/8 (pages 1-5) done in 4.9s + ... +Done in 11.3s wall time (sum of individual requests: 38.1s). +Extracted 12 table(s) across 40 page(s). +Merged result written to result.json +8 CSV file(s) written to tables/ +``` + +Useful flags: + +| Flag | Meaning | +|---|---| +| `--workers 1` | sequential baseline (for comparison) | +| `--features table` | table extraction only | +| `--language ARA` | use the multilingual models instead of English-only | +| `--doc-type INVOICE` | document type hint for `key_value` extraction | +| `--profile LONDON` | use a specific profile from your OCI config file | + +Or use it as a library: + +```python +from du_pipeline import Settings, ParallelDocumentAnalyzer, split_pdf, merge_results + +settings = Settings.from_env() +chunks = split_pdf(open("report.pdf", "rb").read()) +results, _ = ParallelDocumentAnalyzer(settings).analyze(chunks, features=["table"]) +merged = merge_results(results) +``` + +## Benchmarks + +Two scripts under `benchmarks/` measure the speedup on your own tenancy and documents: + +```bash +# sequential vs parallel synchronous processing (same document, same chunks) +python benchmarks/run_benchmark.py report_40_pages.pdf --workers 4 + +# the async Object Storage job path, for comparison (needs a bucket) +export DU_BUCKET_NAME= +python benchmarks/run_async_job.py report_40_pages.pdf +``` + +Results measured on a 22-page text+table PDF (Frankfurt region, default limits, table + text features) — run the scripts to reproduce on your tenancy: + +| Mode | Wall time | vs parallel | +|---|---|---| +| **Parallel synchronous (4 workers)** | **18.2s** | — | +| Sequential synchronous (1 worker) | 48.2s | 2.6x slower | +| Asynchronous Object Storage job | 65.6s (upload + queue + processing + polling) | 3.6x slower | + +## Notes and limits + +- Service limits ([documentation](https://docs.oracle.com/en-us/iaas/Content/document-understanding/using/limits.htm)): 5 pages / 8 MB per synchronous request; 2,000 pages / 500 MB per asynchronous job; asynchronous throughput is limited per tenancy. +- The worker pool is deliberately bounded and every request uses the OCI SDK default retry strategy, so HTTP 429 throttling degrades throughput gracefully instead of failing the run. +- Document-level features (document classification, language detection) return one answer **per chunk**; the merger keeps the first chunk's answer. For per-page features (text, tables) the merge is lossless. +- Only PDFs are split; single images (JPEG/PNG/TIFF) fit in one request anyway. + +## Project structure + +``` +files/ +├── requirements.txt # oci, pypdf +├── du_pipeline/ +│ ├── splitter.py # PDF -> ≤5-page / ≤8 MB base64 chunks +│ ├── executor.py # parallel synchronous analyze_document calls +│ ├── merger.py # chunk results -> one JSON + CSV table export +│ ├── config.py # env-based settings (no hardcoded OCIDs) +│ └── cli.py # command-line entry point +└── benchmarks/ + ├── run_benchmark.py # sequential vs parallel comparison + └── run_async_job.py # async Object Storage job, timed +``` + +## Authors + +- Brona Nilsson + +## License + +Copyright (c) 2026 Oracle and/or its affiliates. + +Licensed under the Universal Permissive License (UPL), Version 1.0. + +See [LICENSE](https://github.com/oracle-devrel/technology-engineering/blob/main/LICENSE) for more details. diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/benchmarks/run_async_job.py b/ai/ai-document-understanding/du-parallel-pipeline/files/benchmarks/run_async_job.py new file mode 100644 index 0000000000..8470805bba --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/benchmarks/run_async_job.py @@ -0,0 +1,116 @@ +"""Time the asynchronous Object Storage processor-job path for comparison. + +This is the officially supported route for documents over 5 pages (up to +2,000 pages / 500 MB). It is benchmark-only here: the job is queued and +polled, so end-to-end latency is dominated by queueing rather than +processing — which is exactly what the parallel synchronous pipeline avoids. + +Requires an existing bucket in a Document Understanding region: + export DU_BUCKET_NAME= # required + export DU_BUCKET_NAMESPACE= # optional, discovered if unset + +Usage: + python benchmarks/run_async_job.py document.pdf +""" + +import argparse +import os +import sys +import time + +import oci +import oci.ai_document.models as ai_document_models + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from du_pipeline.config import Settings + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("pdf", help="Path to a PDF (any page count up to 2,000)") + parser.add_argument("--features", default="table,text") + parser.add_argument("--profile", help="OCI config profile (default: DEFAULT)") + parser.add_argument( + "--prefix", default="du-async-benchmark", help="Output prefix in the bucket" + ) + args = parser.parse_args() + + bucket = os.environ.get("DU_BUCKET_NAME") + if not bucket: + sys.exit("Set DU_BUCKET_NAME to a bucket in a Document Understanding region.") + + settings = Settings.from_env(profile=args.profile) + oci_config = oci.config.from_file(settings.config_file, settings.profile) + + object_storage = oci.object_storage.ObjectStorageClient(oci_config) + namespace = os.environ.get("DU_BUCKET_NAMESPACE") or object_storage.get_namespace().data + object_name = f"{args.prefix}/{os.path.basename(args.pdf)}" + + feature_models = { + "table": ai_document_models.DocumentTableExtractionFeature( + feature_type="TABLE_EXTRACTION" + ), + "text": ai_document_models.DocumentTextExtractionFeature( + feature_type="TEXT_EXTRACTION" + ), + "key_value": ai_document_models.DocumentKeyValueExtractionFeature( + feature_type="KEY_VALUE_EXTRACTION" + ), + } + features = [feature_models[name.strip()] for name in args.features.split(",")] + + started = time.perf_counter() + + print(f"Uploading {args.pdf} to {bucket}/{object_name}...") + with open(args.pdf, "rb") as handle: + object_storage.put_object( + namespace_name=namespace, + bucket_name=bucket, + object_name=object_name, + put_object_body=handle, + ) + upload_done = time.perf_counter() + + client = oci.ai_document.AIServiceDocumentClient(oci_config) + composite = oci.ai_document.AIServiceDocumentClientCompositeOperations(client) + + print("Creating processor job and waiting for completion...") + response = composite.create_processor_job_and_wait_for_state( + create_processor_job_details=ai_document_models.CreateProcessorJobDetails( + compartment_id=settings.compartment_id, + input_location=ai_document_models.ObjectStorageLocations( + source_type="OBJECT_STORAGE_LOCATIONS", + object_locations=[ + ai_document_models.ObjectLocation( + namespace_name=namespace, + bucket_name=bucket, + object_name=object_name, + ) + ], + ), + output_location=ai_document_models.OutputLocation( + namespace_name=namespace, + bucket_name=bucket, + prefix=f"{args.prefix}-results", + ), + processor_config=ai_document_models.GeneralProcessorConfig( + processor_type="GENERAL", + features=features, + ), + ), + wait_for_states=["SUCCEEDED", "FAILED"], + waiter_kwargs={"max_wait_seconds": 3600}, + ) + finished = time.perf_counter() + + job = response.data + print(f"\nJob {job.id} finished with state: {job.lifecycle_state}") + print(f"Upload time: {upload_done - started:.1f}s") + print(f"Job time: {finished - upload_done:.1f}s (queue + processing + polling)") + print(f"Total async time: {finished - started:.1f}s") + print(f"Results are under {bucket}/{args.prefix}-results/ in Object Storage.") + + +if __name__ == "__main__": + main() diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/benchmarks/run_benchmark.py b/ai/ai-document-understanding/du-parallel-pipeline/files/benchmarks/run_benchmark.py new file mode 100644 index 0000000000..df0cefc239 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/benchmarks/run_benchmark.py @@ -0,0 +1,65 @@ +"""Benchmark sequential vs parallel synchronous processing of one PDF. + +Runs the identical chunk set twice — once with 1 worker, once with N — +and prints a comparison table for the README. + +Usage: + python benchmarks/run_benchmark.py document.pdf --workers 4 +""" + +import argparse +import os +import sys +import time + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from du_pipeline.config import Settings +from du_pipeline.executor import ParallelDocumentAnalyzer +from du_pipeline.merger import merge_results +from du_pipeline.splitter import split_pdf + + +def timed_run(analyzer, chunks, features, workers): + started = time.perf_counter() + results, timings = analyzer.analyze(chunks, features=features, max_workers=workers) + wall_time = time.perf_counter() - started + merge_results(results) # include merge cost in the measurement + return wall_time, timings + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("pdf", help="Path to a multi-page PDF") + parser.add_argument("--workers", type=int, default=4) + parser.add_argument("--features", default="table,text") + parser.add_argument("--profile", help="OCI config profile (default: DEFAULT)") + args = parser.parse_args() + + settings = Settings.from_env(profile=args.profile) + features = [name.strip() for name in args.features.split(",")] + + with open(args.pdf, "rb") as handle: + chunks = split_pdf(handle.read()) + total_pages = sum(chunk.page_count for chunk in chunks) + print(f"{args.pdf}: {total_pages} pages, {len(chunks)} chunks, features: {args.features}\n") + + analyzer = ParallelDocumentAnalyzer(settings) + + print("Sequential run (1 worker)...") + seq_wall, seq_timings = timed_run(analyzer, chunks, features, workers=1) + + print(f"Parallel run ({args.workers} workers)...") + par_wall, par_timings = timed_run(analyzer, chunks, features, workers=args.workers) + + speedup = seq_wall / par_wall if par_wall else float("inf") + print() + print(f"| Mode | Wall time | Avg per chunk |") + print(f"|---|---|---|") + print(f"| Sequential (1 worker) | {seq_wall:.1f}s | {sum(seq_timings)/len(seq_timings):.1f}s |") + print(f"| Parallel ({args.workers} workers) | {par_wall:.1f}s | {sum(par_timings)/len(par_timings):.1f}s |") + print(f"\nSpeedup: {speedup:.1f}x") + + +if __name__ == "__main__": + main() diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/__init__.py b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/__init__.py new file mode 100644 index 0000000000..0839e37047 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/__init__.py @@ -0,0 +1,21 @@ +"""Parallel processing pipeline for OCI Document Understanding. + +Splits documents larger than the 5-page synchronous limit into chunks, +analyzes all chunks concurrently, and merges the results back into a +single response with document-level page numbering. +""" + +from du_pipeline.splitter import split_pdf, Chunk +from du_pipeline.executor import ParallelDocumentAnalyzer, SUPPORTED_FEATURES +from du_pipeline.merger import merge_results, export_tables_csv +from du_pipeline.config import Settings + +__all__ = [ + "split_pdf", + "Chunk", + "ParallelDocumentAnalyzer", + "SUPPORTED_FEATURES", + "merge_results", + "export_tables_csv", + "Settings", +] diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/cli.py b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/cli.py new file mode 100644 index 0000000000..0af7809a3b --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/cli.py @@ -0,0 +1,125 @@ +"""Command-line entry point for the parallel Document Understanding pipeline. + +Example: + python -m du_pipeline.cli invoice_40_pages.pdf \ + --features table,text --workers 4 \ + --output result.json --tables-dir tables/ +""" + +import argparse +import json +import sys +import time + +from du_pipeline.config import Settings +from du_pipeline.executor import DEFAULT_WORKERS, SUPPORTED_FEATURES, ParallelDocumentAnalyzer +from du_pipeline.merger import export_tables_csv, merge_results +from du_pipeline.splitter import split_pdf + + +def build_parser(): + parser = argparse.ArgumentParser( + prog="du_pipeline", + description=( + "Analyze PDFs of any length with OCI Document Understanding by " + "splitting them into 5-page chunks and processing the chunks in " + "parallel synchronous requests." + ), + ) + parser.add_argument("pdf", help="Path to the PDF document") + parser.add_argument( + "--features", + default="table,text", + help=f"Comma-separated features to run: {','.join(sorted(SUPPORTED_FEATURES))} " + "(default: table,text)", + ) + parser.add_argument( + "--workers", + type=int, + default=DEFAULT_WORKERS, + help=f"Concurrent requests (default: {DEFAULT_WORKERS}; use 1 for sequential)", + ) + parser.add_argument("--output", help="Write the merged result JSON to this file") + parser.add_argument( + "--tables-dir", help="Export every extracted table as CSV into this directory" + ) + parser.add_argument( + "--language", + help="Document language code, e.g. ENG (English models) or ARA (multilingual models)", + ) + parser.add_argument( + "--doc-type", + help="Document type hint for key_value extraction, e.g. INVOICE, RECEIPT", + ) + parser.add_argument("--compartment-id", help="Compartment OCID (default: $OCI_COMPARTMENT_ID)") + parser.add_argument("--config-file", help="OCI config file (default: ~/.oci/config)") + parser.add_argument("--profile", help="OCI config profile (default: DEFAULT)") + return parser + + +def run(args): + settings = Settings.from_env( + compartment_id=args.compartment_id, + config_file=args.config_file, + profile=args.profile, + ) + features = [name.strip() for name in args.features.split(",") if name.strip()] + + with open(args.pdf, "rb") as handle: + pdf_bytes = handle.read() + + chunks = split_pdf(pdf_bytes) + total_pages = sum(chunk.page_count for chunk in chunks) + print( + f"{args.pdf}: {total_pages} pages -> {len(chunks)} chunk(s), " + f"{args.workers} worker(s), features: {', '.join(features)}" + ) + + analyzer = ParallelDocumentAnalyzer(settings) + + def report(chunk, elapsed): + last_page = chunk.first_page + chunk.page_count - 1 + print(f" chunk {chunk.index + 1}/{len(chunks)} " + f"(pages {chunk.first_page}-{last_page}) done in {elapsed:.1f}s") + + started = time.perf_counter() + results, timings = analyzer.analyze( + chunks, + features=features, + language=args.language, + document_type=args.doc_type, + max_workers=args.workers, + on_chunk_done=report, + ) + wall_time = time.perf_counter() - started + + merged = merge_results(results) + table_count = sum(len(page.get("tables") or []) for page in merged["pages"]) + print( + f"Done in {wall_time:.1f}s wall time " + f"(sum of individual requests: {sum(timings):.1f}s). " + f"Extracted {table_count} table(s) across {len(merged['pages'])} page(s)." + ) + + if args.output: + with open(args.output, "w", encoding="utf-8") as handle: + json.dump(merged, handle, indent=2, default=str) + print(f"Merged result written to {args.output}") + + if args.tables_dir: + written = export_tables_csv(merged, args.tables_dir) + print(f"{len(written)} CSV file(s) written to {args.tables_dir}/") + + return merged + + +def main(argv=None): + args = build_parser().parse_args(argv) + try: + run(args) + except (ValueError, FileNotFoundError) as error: + sys.exit(str(error)) + + +if __name__ == "__main__": + main() diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/config.py b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/config.py new file mode 100644 index 0000000000..a8557cd948 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/config.py @@ -0,0 +1,29 @@ +import os +from dataclasses import dataclass + + +@dataclass +class Settings: + """Connection settings for the OCI Document Understanding client. + + The compartment OCID is never hardcoded: it comes from the + OCI_COMPARTMENT_ID environment variable or the --compartment-id flag. + """ + + compartment_id: str + config_file: str = "~/.oci/config" + profile: str = "DEFAULT" + + @classmethod + def from_env(cls, compartment_id=None, config_file=None, profile=None): + compartment_id = compartment_id or os.environ.get("OCI_COMPARTMENT_ID") + if not compartment_id: + raise SystemExit( + "No compartment OCID given. Set the OCI_COMPARTMENT_ID environment " + "variable or pass --compartment-id." + ) + return cls( + compartment_id=compartment_id, + config_file=config_file or os.environ.get("OCI_CONFIG_FILE", "~/.oci/config"), + profile=profile or os.environ.get("OCI_CONFIG_PROFILE", "DEFAULT"), + ) diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/executor.py b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/executor.py new file mode 100644 index 0000000000..b2f926fe4f --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/executor.py @@ -0,0 +1,100 @@ +"""Run synchronous analyze_document calls for many chunks in parallel. + +The synchronous API is stateless, so independent chunks can be analyzed +concurrently with a thread pool. The OCI SDK's default retry strategy +handles 429 throttling and transient 5xx errors with exponential backoff, +which keeps the pipeline polite against the per-tenancy rate limit. +""" + +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + +import oci +import oci.ai_document.models as ai_document_models +from oci.util import to_dict + +SUPPORTED_FEATURES = { + "table": lambda: ai_document_models.DocumentTableExtractionFeature( + feature_type="TABLE_EXTRACTION" + ), + "text": lambda: ai_document_models.DocumentTextExtractionFeature( + feature_type="TEXT_EXTRACTION" + ), + "key_value": lambda: ai_document_models.DocumentKeyValueExtractionFeature( + feature_type="KEY_VALUE_EXTRACTION" + ), +} + +DEFAULT_WORKERS = 4 + + +class ParallelDocumentAnalyzer: + def __init__(self, settings): + oci_config = oci.config.from_file(settings.config_file, settings.profile) + self.client = oci.ai_document.AIServiceDocumentClient( + config=oci_config, + retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY, + ) + self.compartment_id = settings.compartment_id + + def _analyze_chunk(self, chunk, features, language, document_type): + kwargs = {} + if language: + kwargs["language"] = language + if document_type: + kwargs["document_type"] = document_type + + started = time.perf_counter() + response = self.client.analyze_document( + analyze_document_details=ai_document_models.AnalyzeDocumentDetails( + features=[SUPPORTED_FEATURES[name]() for name in features], + document=ai_document_models.InlineDocumentDetails( + source="INLINE", data=chunk.data + ), + compartment_id=self.compartment_id, + **kwargs, + ) + ) + elapsed = time.perf_counter() - started + return to_dict(response.data), elapsed + + def analyze( + self, + chunks, + features=("table", "text"), + language=None, + document_type=None, + max_workers=DEFAULT_WORKERS, + on_chunk_done=None, + ): + """Analyze all chunks and return (results, timings), both in chunk order. + + results[i] pairs chunks[i] with its analyze_document response as a dict; + timings[i] is that chunk's request wall time in seconds. Use + max_workers=1 for a sequential baseline. on_chunk_done, if given, is + called as chunks finish (out of order) for progress reporting. + """ + unknown = set(features) - set(SUPPORTED_FEATURES) + if unknown: + raise ValueError( + f"Unsupported features: {sorted(unknown)}. " + f"Choose from {sorted(SUPPORTED_FEATURES)}." + ) + + results = [None] * len(chunks) + timings = [0.0] * len(chunks) + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = { + pool.submit( + self._analyze_chunk, chunk, features, language, document_type + ): chunk + for chunk in chunks + } + for future in as_completed(futures): + chunk = futures[future] + result, elapsed = future.result() + results[chunk.index] = (chunk, result) + timings[chunk.index] = elapsed + if on_chunk_done: + on_chunk_done(chunk, elapsed) + return results, timings diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/merger.py b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/merger.py new file mode 100644 index 0000000000..91b7b73513 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/merger.py @@ -0,0 +1,91 @@ +"""Merge per-chunk analyze_document results into one document-level result. + +Each chunk is analyzed in isolation, so the service numbers its pages 1..5. +The merger restores the original page numbers from the chunk offsets and +returns a single response-shaped dict, as if one call had processed the +whole document. Tables can additionally be exported to CSV files named by +their true page number. +""" + +import csv +import os + + +def merge_results(chunk_results): + """Merge ordered (chunk, result_dict) pairs into one result dict. + + Page-level content (text lines, words, tables) merges losslessly. + Document-level classifications (detected_document_types, + detected_languages) are taken from the first chunk, since chunked + processing produces one answer per chunk. + """ + if not chunk_results: + raise ValueError("No chunk results to merge") + + merged_pages = [] + for chunk, result in chunk_results: + for page in result.get("pages") or []: + page = dict(page) + local_number = page.get("page_number") or 1 + page["page_number"] = chunk.first_page + (local_number - 1) + merged_pages.append(page) + merged_pages.sort(key=lambda page: page["page_number"]) + + first_result = chunk_results[0][1] + metadata = dict(first_result.get("document_metadata") or {}) + metadata["page_count"] = len(merged_pages) + + return { + "document_metadata": metadata, + "detected_document_types": first_result.get("detected_document_types"), + "detected_languages": first_result.get("detected_languages"), + "pages": merged_pages, + } + + +def _table_to_grid(table): + """Convert a DU table dict into a 2D list of cell texts.""" + cells = [] + for section in ("header_rows", "body_rows", "footer_rows"): + for row in table.get(section) or []: + for cell in row.get("cells") or []: + if cell.get("row_index") is not None and cell.get("column_index") is not None: + cells.append(cell) + if not cells: + return [] + + # Normalize indices so both 0- and 1-based numbering produce a full grid. + min_row = min(cell["row_index"] for cell in cells) + min_col = min(cell["column_index"] for cell in cells) + n_rows = max(cell["row_index"] for cell in cells) - min_row + 1 + n_cols = max(cell["column_index"] for cell in cells) - min_col + 1 + + grid = [[""] * n_cols for _ in range(n_rows)] + for cell in cells: + grid[cell["row_index"] - min_row][cell["column_index"] - min_col] = ( + cell.get("text") or "" + ) + return grid + + +def export_tables_csv(merged_result, output_dir): + """Write every extracted table to output_dir as one CSV per table. + + Files are named page_table.csv using document-level page + numbers. Returns the list of written paths. + """ + os.makedirs(output_dir, exist_ok=True) + written = [] + for page in merged_result["pages"]: + for table_index, table in enumerate(page.get("tables") or [], start=1): + grid = _table_to_grid(table) + if not grid: + continue + path = os.path.join( + output_dir, + f"page{page['page_number']:03d}_table{table_index}.csv", + ) + with open(path, "w", newline="", encoding="utf-8") as handle: + csv.writer(handle).writerows(grid) + written.append(path) + return written diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/splitter.py b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/splitter.py new file mode 100644 index 0000000000..25831a8014 --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/du_pipeline/splitter.py @@ -0,0 +1,68 @@ +"""Split a PDF into chunks that fit the synchronous analyze_document limits. + +OCI Document Understanding accepts at most 5 pages and 8 MB per synchronous +request. The service keeps no state between requests, so the pipeline tracks +each chunk's position in the original document and restores it when merging. +""" + +import base64 +from dataclasses import dataclass +from io import BytesIO + +from pypdf import PdfReader, PdfWriter + +MAX_PAGES_PER_CHUNK = 5 +MAX_CHUNK_BYTES = 8 * 1024 * 1024 + + +@dataclass +class Chunk: + index: int + first_page: int # 1-based page number in the original document + page_count: int + data: str # base64-encoded PDF bytes, ready for InlineDocumentDetails + + +def _write_range(reader, start, count): + writer = PdfWriter() + for page in reader.pages[start:start + count]: + writer.add_page(page) + buffer = BytesIO() + writer.write(buffer) + return buffer.getvalue() + + +def split_pdf(pdf_bytes, max_pages=MAX_PAGES_PER_CHUNK): + """Split raw PDF bytes into base64 chunks of at most max_pages pages. + + A chunk that exceeds the 8 MB request limit is halved until it fits, + so image-heavy documents degrade to smaller chunks instead of failing. + """ + if not 1 <= max_pages <= MAX_PAGES_PER_CHUNK: + raise ValueError(f"max_pages must be between 1 and {MAX_PAGES_PER_CHUNK}") + + reader = PdfReader(BytesIO(pdf_bytes)) + total_pages = len(reader.pages) + chunks = [] + start = 0 + while start < total_pages: + count = min(max_pages, total_pages - start) + chunk_bytes = _write_range(reader, start, count) + while len(chunk_bytes) > MAX_CHUNK_BYTES and count > 1: + count = max(1, count // 2) + chunk_bytes = _write_range(reader, start, count) + if len(chunk_bytes) > MAX_CHUNK_BYTES: + raise ValueError( + f"Page {start + 1} alone exceeds the 8 MB synchronous request limit; " + "use the asynchronous Object Storage path for this document." + ) + chunks.append( + Chunk( + index=len(chunks), + first_page=start + 1, + page_count=count, + data=base64.b64encode(chunk_bytes).decode("ascii"), + ) + ) + start += count + return chunks diff --git a/ai/ai-document-understanding/du-parallel-pipeline/files/requirements.txt b/ai/ai-document-understanding/du-parallel-pipeline/files/requirements.txt new file mode 100644 index 0000000000..1dfbfc58cd --- /dev/null +++ b/ai/ai-document-understanding/du-parallel-pipeline/files/requirements.txt @@ -0,0 +1,2 @@ +oci +pypdf