Add du-parallel-pipeline: parallel synchronous processing for long do…

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.

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.

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=<compartment_ocid>
+   # 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=<bucket_in_a_DU_region>
+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.

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=<bucket>          # required
+    export DU_BUCKET_NAMESPACE=<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()

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()