A Databricks-native governed document intelligence pipeline — the upstream preparation layer that converts unstructured enterprise documents into traceable, schema-validated AI-ready assets before downstream Bedrock retrieval and agent reasoning begins.
At a glance: this project ingests public or enterprise documents, parses them through Databricks AI Function surfaces, validates structured extraction with Pydantic contracts, classifies/routes Gold records, evaluates every layer, and prepares a schema-versioned handoff payload for downstream Bedrock CaseOps.
| Public status | Current state |
|---|---|
| Pipeline boundary | Upstream document preparation only: raw document in, contract-valid Gold record out |
| Local verification | 1,510 tests passing with .venv/bin/python -m pytest -q |
| Runtime validation | Personal Databricks dev workspace smoke closeout on 2026-06-08 |
| Delivery posture | Producer-side Delta Sharing manifests and delivery events; no Bedrock consumer runtime |
| Production posture | Portfolio-safe, non-production; no enterprise deployment, credentials, or private workspace identifiers |
Public proof record: this repository is also summarized on navidbr.me/work/databricks-caseops-lakehouse as part of Navid's public NAVIDBR Applied AI Systems work record. The site keeps the same boundary: upstream source preparation, not an enterprise deployment or live downstream Bedrock runtime.
Enterprise operations accumulate large volumes of unstructured documents — regulatory notices, incident reports, standard operating procedures, quality reviews, and technical advisories. These documents carry operationally significant information, but without structured, governed transformation, they cannot be reliably extracted from, classified, or consumed by downstream AI retrieval and agent systems.
This repo implements the governed upstream transformation layer on Databricks — the system that must exist before any downstream retrieval, RAG, or agent reasoning begins:
- Ingests unstructured documents into governed Unity Catalog Volumes with full source provenance
- Parses raw document content using
ai_parse_documentand normalizes it into a Bronze layer - Extracts structured fields from parsed content using
ai_extractinto a Silver layer - Classifies and routes documents using
ai_classifyinto a Gold layer of AI-ready assets - Evaluates every stage for extraction quality, schema validity, and cross-layer traceability using MLflow
- Exports contract-enforced Gold-tier assets to downstream Bedrock retrieval and agent systems
This repo is the upstream governed preparation layer of the Bedrock CaseOps architecture. It owns the complete ingestion-to-handoff pipeline — governed document transformation, structured extraction, schema validation, classification, routing, and AI-ready asset materialization. Retrieval, agent reasoning, escalation, and case-support belong to the Bedrock CaseOps Control Tower.
| Concern | This Repo | Bedrock CaseOps |
|---|---|---|
| Raw document ingestion | Yes | No |
| Parsing and extraction | Yes | No |
| Schema validation and traceability | Yes | No |
| Classification and routing | Yes | No |
| Governed AI-ready asset preparation | Yes | No |
| Gold export payload delivery | Yes (file/Delta) | Consumes |
| Retrieval and RAG | No | Yes |
| Agent reasoning and orchestration | No | Yes |
| Escalation and case-support workflows | No | Yes |
| KPI reporting or cross-case analytics | No | Out of scope for both |
This repo is the governed upstream structuring layer. It does not reason over documents, orchestrate decisions, or produce operational dashboards. Its contract is: raw document in, structured AI-ready record out.
The pipeline is designed for document-heavy operational and regulatory workflows:
- FDA warning letters and safety notices
- CISA advisories and cybersecurity bulletins
- Incident reports and post-mortems
- Standard operating procedures (SOPs)
- Quality review and audit records
- Technical support and case review documents
| Layer | Responsibility | Contract output |
|---|---|---|
| Raw Volume | Immutable source document storage in Unity Catalog | Source metadata and file provenance |
| Bronze | Parse source content with ai_parse_document surfaces |
Parsed text, parse status, source hash, lineage IDs |
| Silver | Extract structured domain fields with ai_extract surfaces |
Pydantic-validated domain records and validation status |
| Gold | Classify, route, package, and gate export readiness | AI-ready payloads, quarantine outcomes, handoff artifacts |
| Delivery | Prepare producer-side handoff for Bedrock CaseOps | JSON exports, batch reports, bundles, delivery events, Delta Share manifest |
All layers are governed by Unity Catalog. All transformations are traceable via MLflow. See ARCHITECTURE.md for full design detail.
| Component | Technology |
|---|---|
| Platform | Databricks |
| Storage governance | Unity Catalog, Volumes |
| Parsing | ai_parse_document |
| Extraction | ai_extract |
| Classification | ai_classify |
| Evaluation & tracing | MLflow |
| Language (pipelines) | Python, SQL |
| Config | YAML |
| Docs | Markdown |
V1 and V2 are both complete.
Phase 1 (V1 — Core Governed Pipeline) delivered the full Bronze → Silver → Gold document intelligence pipeline: Unity Catalog-governed ingestion, ai_parse_document-based Bronze parsing, ai_extract-based Silver structured field extraction, ai_classify-based Gold classification and routing, MLflow evaluation across all four quality dimensions (parse quality, extraction quality, classification quality, traceability), and the complete Gold → Bedrock handoff preparation layer — including contract definition, schema enforcement, export materialization, batch bundle packaging, and integrity validation. V1 was validated end-to-end in a personal Databricks workspace using real AI Functions against public FDA sample documents.
Phase 2 (V2 — Hardening and Expansion) added producer-side handoff preparation via Delta Sharing manifests and delivery events (Phase C), multi-domain pipeline expansion across three active document domains — FDA warning letters, CISA cybersecurity advisories, and incident reports (Phase D), and enterprise operational hardening: structured human review queue and reprocessing, multi-environment configuration separation, and governance monitoring (Phase E).
Current state: Portfolio-safe and non-production. No enterprise deployment, no production credentials, no live Bedrock integration beyond the producer-side delivery preparation layer. The pipeline is fully functional locally, was validated in a personal Databricks workspace, Phase 3 runtime handoff validation reached validated with sanitized personal-workspace evidence on 2026-06-07, and Phase 4 reached ready_for_phase4_closeout with sanitized personal Databricks dev workspace smoke evidence on 2026-06-08. Total test coverage: 1,510 tests across all pipeline stages, contract layers, export boundaries, delivery validation, multi-domain framework, operational hardening, runtime adapter scaffolding, Phase 3 evidence validation, and Phase 4 runtime smoke support.
Phase 4 is closed for the personal Databricks dev workspace smoke scope. Runtime
adapter wiring for ai_parse_document, ai_extract, and ai_classify, Delta
table read/write support, Databricks bundle scaffolding, check-only preflight,
write-permission preflight, delivery validation, and sanitized runtime smoke
package validation are implemented and validated. This is not a staging/prod
enterprise deployment and does not add Bedrock runtime behavior.
For the full delivery history, phase-by-phase detail, and roadmap, see PROJECT_SPEC.md and docs/roadmap.md.
To run the local Gold demo, see the Running the Gold Demo section below.
The audit found this repo is coherent as an upstream governed preparation layer: it ingests documents, creates Bronze/Silver/Gold records, validates contracts, prepares Gold handoff artifacts, and exposes producer-side delivery and governance outputs.
It does not implement Bedrock runtime behavior, retrieval, RAG, agent reasoning, escalation, or case-support workflows. Those belong in the downstream Bedrock CaseOps Control Tower.
The next milestone is broader deployment hardening outside this personal
workspace closeout: staging/prod Databricks promotion if needed, plus the
downstream Bedrock-side consumer adapter for this repo's Gold export or Delta
Share surface. See
docs/databricks-runtime-productionization.md
for the Phase 4 execution details.
This repository is public-safe as a reference implementation. It does not include Databricks workspace URLs, account IDs, personal identifiers, PATs, tokens, customer data, generated runtime evidence, or committed local outputs.
Another user can reuse the local pipeline, tests, contracts, Databricks runtime entrypoints, and Asset Bundle scaffold in their own workspace. They must provide their own Databricks environment, Unity Catalog resources, AI Function access, workspace credentials, and downstream Bedrock consumer.
Use Python 3.10+. The local shell may not expose python, so commands use .venv/bin/python after the virtual environment is created. If you do not use a virtual environment, replace .venv/bin/python with python3.
python3 -m venv .venv
.venv/bin/python -m pip install -r requirements-dev.txt
.venv/bin/python -m pytest -qEquivalent Make targets:
make install-dev
make test
make auditmake audit currently runs the full local test suite. Runtime dependencies are in requirements.txt; development and test dependencies are in requirements-dev.txt. Generated local run artifacts are written under output/ and should not be committed. Curated fixtures belong under examples/.
Requires Python 3.10+ and pydantic (v2). No Databricks workspace needed. Install dependencies with Local Setup and Verification first.
# 1. Ingest the sample FDA warning letter → produces a Bronze JSON artifact
.venv/bin/python src/pipelines/ingest_bronze.py \
--input examples/fda_warning_letter_sample.md \
--document-class-hint fda_warning_letter \
--source-system local_dev
# Artifact is written to output/bronze/<bronze_record_id>.json
# 2. Run Bronze evaluation against all artifacts in the output directory
.venv/bin/python src/evaluation/eval_bronze.py --input-dir output/bronze
# Optional: evaluate a single artifact
.venv/bin/python src/evaluation/eval_bronze.py --input output/bronze/<bronze_record_id>.jsonThe evaluation script prints a parse quality summary and writes a JSON evaluation artifact to output/eval/.
Requires Python 3.10+ and pydantic (v2). No Databricks workspace needed. Install dependencies with Local Setup and Verification first.
If you have already run the Bronze demo, skip the ingest step.
# 1. Ingest the sample FDA warning letter → produces a Bronze JSON artifact
.venv/bin/python src/pipelines/ingest_bronze.py \
--input examples/fda_warning_letter_sample.md \
--document-class-hint fda_warning_letter \
--source-system local_dev
# Artifact is written to output/bronze/<bronze_record_id>.json
# 2. Extract structured fields from Bronze → produces a Silver JSON artifact
.venv/bin/python src/pipelines/extract_silver.py --input-dir output/bronze
# Artifact is written to output/silver/<extraction_id>.json
# 3. Run Silver evaluation against all artifacts in the output directory
.venv/bin/python src/evaluation/eval_silver.py --input-dir output/silver
# Optional: evaluate a single artifact
.venv/bin/python src/evaluation/eval_silver.py --input output/silver/<extraction_id>.jsonThe evaluation script prints an extraction quality summary (validity rate, field
coverage, required-field null rate) and writes a JSON evaluation artifact to
output/eval/. See examples/expected_silver_fda_warning_letter.json for a
reference fixture showing a successful extraction result.
Requires Python 3.10+ and pydantic (v2). No Databricks workspace needed. Install dependencies with Local Setup and Verification first.
If you have already run the Bronze and Silver demos, skip the ingest and extraction steps.
# 1. Ingest the sample FDA warning letter → produces a Bronze JSON artifact
.venv/bin/python src/pipelines/ingest_bronze.py \
--input examples/fda_warning_letter_sample.md \
--document-class-hint fda_warning_letter \
--source-system local_dev
# Artifact is written to output/bronze/<bronze_record_id>.json
# 2. Extract structured fields from Bronze → produces a Silver JSON artifact
.venv/bin/python src/pipelines/extract_silver.py --input-dir output/bronze
# Artifact is written to output/silver/<extraction_id>.json
# 3. Classify Silver records → produces Gold artifacts and export payloads
.venv/bin/python src/pipelines/classify_gold.py \
--input-dir output/silver \
--bronze-dir output/bronze
# Gold record: output/gold/<gold_record_id>.json
# Export payload (if export-ready and contract-valid): output/gold/exports/regulatory_review/<document_id>.json
# Invalid payloads are blocked before write — see contract_validation_errors in pipeline output
# Optional: produce a handoff batch report
.venv/bin/python src/pipelines/classify_gold.py \
--input-dir output/silver \
--bronze-dir output/bronze \
--report-dir output/reports
# Report artifacts: output/reports/handoff_report_<run_id>.json (machine-readable)
# output/reports/handoff_report_<run_id>.txt (human-readable)
# Optional: produce a batch manifest and review bundle (may be combined with --report-dir)
.venv/bin/python src/pipelines/classify_gold.py \
--input-dir output/silver \
--bronze-dir output/bronze \
--report-dir output/reports \
--bundle-dir output/reports
# Bundle artifacts: output/reports/handoff_bundle_<run_id>.json (machine-readable manifest)
# output/reports/handoff_bundle_<run_id>.txt (human-readable review summary)
# The bundle references all per-record artifact paths and the handoff report when both flags are used.
# Optional: run bundle integrity validation against the generated bundle
.venv/bin/python -c "
from pathlib import Path
from src.pipelines.handoff_bundle_validation import validate_handoff_bundle, format_validation_result_text
import glob
bundles = sorted(glob.glob('output/reports/handoff_bundle_*.json'))
if bundles:
result = validate_handoff_bundle(Path(bundles[-1]))
print(format_validation_result_text(result))
"
# 4. Run Gold evaluation against all artifacts in the output directory
.venv/bin/python src/evaluation/eval_gold.py --input-dir output/gold
# Optional: evaluate a single artifact
.venv/bin/python src/evaluation/eval_gold.py --input output/gold/<gold_record_id>.jsonThe evaluation script prints a classification quality summary (success rate,
export-ready rate, confidence distribution, label distribution) and writes a
JSON evaluation artifact to output/eval/. See
examples/expected_gold_fda_warning_letter.json for a reference fixture showing
a successful classification and export-ready result.
Requires Python 3.10+ and pydantic (v2). Run the full Gold Demo first to generate artifacts.
# 1–3. Run the Bronze, Silver, Gold demos (if not already done)
.venv/bin/python src/pipelines/ingest_bronze.py \
--input examples/fda_warning_letter_sample.md \
--document-class-hint fda_warning_letter \
--source-system local_dev
.venv/bin/python src/pipelines/extract_silver.py --input-dir output/bronze
# 4. Classify Gold with delivery augmentation enabled
.venv/bin/python src/pipelines/classify_gold.py \
--input-dir output/silver \
--bronze-dir output/bronze \
--report-dir output/reports \
--bundle-dir output/reports \
--delivery-dir output/delivery
# Delivery artifacts:
# output/delivery/delivery_event_<run_id>.json — DeliveryEvent record (schema v0.2.0)
# output/delivery/delivery_event_<run_id>.txt — Human-readable delivery event summary
# output/delivery/delta_share_preparation_manifest.json — Share config + Unity Catalog SQL DDLThe delivery event JSON carries status: "prepared" — the producer-side layer is complete.
The Delta Share preparation manifest contains the Unity Catalog SQL DDL to provision the share
in a Databricks workspace. See Running Delivery Validation to
validate the delivery artifacts locally.
When --delivery-dir is active, export payloads are written at schema_version: v0.2.0 with
three optional provenance fields: delivery_mechanism, delta_share_name, delivery_event_id.
When --delivery-dir is omitted, baseline export behavior (v0.1.0) is fully preserved.
Requires Python 3.10+ and pydantic (v2). Run the Gold Demo first to generate pipeline artifacts.
# Run the Gold pipeline with full report, bundle, and review queue output
.venv/bin/python src/pipelines/classify_gold.py \
--input-dir output/silver \
--bronze-dir output/bronze \
--report-dir output/reports \
--bundle-dir output/reports \
--review-queue-dir output/review_queue
# Review queue artifacts:
# output/review_queue/review_queue_<run_id>.json — ReviewQueueArtifact (machine-readable)
# output/review_queue/review_queue_<run_id>.txt — Human-readable review queue summaryThe review queue collects records with:
outcome_category == 'quarantined'→ reason:quarantinedoutcome_category == 'contract_blocked'→ reason:contract_blockedoutcome_category == 'skipped_not_export_ready'ANDdocument_type_label == 'unknown'→ reason:extraction_failed
To record a review decision and produce a reprocessing request:
from src.schemas.review_decision import (
ReviewDecision,
DECISION_REQUEST_REPROCESSING,
REVIEW_DECISION_SCHEMA_VERSION,
build_reprocessing_request,
make_decision_id,
validate_review_decision,
validate_reprocessing_request,
)
from datetime import datetime, timezone
# Load a queue entry (from the review queue JSON)
# ... queue_entry = loaded_queue["queue_entries"][0] ...
decision = ReviewDecision(
decision_id=make_decision_id(),
queue_entry_id=queue_entry["queue_entry_id"],
document_id=queue_entry["document_id"],
gold_record_id=queue_entry["gold_record_id"],
pipeline_run_id=queue_entry["pipeline_run_id"],
decided_at=datetime.now(tz=timezone.utc).isoformat(),
schema_version=REVIEW_DECISION_SCHEMA_VERSION,
decision=DECISION_REQUEST_REPROCESSING,
decision_rationale="Reviewer identified this as an FDA warning letter. Re-extract with explicit class hint.",
reprocessing_request_id="placeholder-to-be-replaced",
)
reprocessing_req = build_reprocessing_request(
decision=decision,
reprocessing_reason="Document classified as unknown but contains FDA warning letter structure.",
suggested_document_class_hint="fda_warning_letter",
)
# Update decision with the real reprocessing_request_id
decision.reprocessing_request_id = reprocessing_req.reprocessing_request_id
errors = validate_review_decision(decision)
assert not errors, errors
errors = validate_reprocessing_request(reprocessing_req)
assert not errors, errorsSee examples/expected_review_queue.json, examples/expected_review_decision.json, and examples/expected_reprocessing_request.json for reference artifacts.
Requires Python 3.10+ and pydantic (v2). Run the Delivery Demo first to generate delivery artifacts.
# Run delivery validation against the generated delivery artifacts
.venv/bin/python -c "
from pathlib import Path
import glob
from src.pipelines.delivery_validation import (
validate_delivery_layer,
format_validation_result_text,
write_validation_result,
)
# Locate delivery event (adjust run_id to match your run)
events = sorted(glob.glob('output/delivery/delivery_event_*.json'))
if not events:
print('No delivery event found. Run the Delivery Demo first.')
else:
# Extract run_id from filename
import re
run_id = re.sub(r'^delivery_event_|\.json$', '', Path(events[-1]).name)
result = validate_delivery_layer(
pipeline_run_id=run_id,
delivery_event_path=Path(events[-1]),
share_manifest_path=Path('output/delivery/delta_share_preparation_manifest.json'),
workspace_mode='local_repo_only',
)
print(format_validation_result_text(result))
json_path, text_path = write_validation_result(result, Path('output/validation'))
print(f'Written: {json_path}')
"
# Expected: status = 'not_provisioned' — correct and honest.
# This means: delivery artifacts are correct producer-side; Delta Share not yet
# executed in Unity Catalog. Run setup_sql from the manifest in Databricks SQL
# to proceed toward 'validated' status (see docs/delivery-runtime-validation.md).Integration health states:
not_provisioned— Share designed in repo, not yet executed in Unity Catalog (honest default)partially_validated— Producer-side correct; share provisioned; no live queries runvalidated— Confirmed in personal Databricks workspacefailed— Schema error, ID mismatch, or parse failure
See docs/delivery-runtime-validation.md for the full delivery validation design, check catalogue, and personal Databricks runtime validation runbook.
Phase 4 runtime smoke evidence and deployment scaffolding are documented in
docs/databricks-runtime-productionization.md.
Requires Python 3.10+ and pydantic (v2). Run the pipeline demos first to generate artifacts.
# Run the full evaluation pass across all three layers
.venv/bin/python src/evaluation/run_evaluation.py \
--bronze-dir output/bronze \
--silver-dir output/silver \
--gold-dir output/gold
# Reports are written to output/eval/:
# report_<id>.json — machine-readable full report
# report_<id>.txt — human-readable summaryOr run individual evaluators:
# Bronze: parse quality
.venv/bin/python src/evaluation/eval_bronze.py --input-dir output/bronze
# Silver: extraction quality
.venv/bin/python src/evaluation/eval_silver.py --input-dir output/silver
# Gold: classification quality (null-confidence safe)
.venv/bin/python src/evaluation/eval_gold.py --input-dir output/gold
# Cross-layer traceability
.venv/bin/python src/evaluation/eval_traceability.py \
--bronze-dir output/bronze \
--silver-dir output/silver \
--gold-dir output/goldOptional MLflow logging (requires mlflow installed):
.venv/bin/python src/evaluation/run_evaluation.py \
--bronze-dir output/bronze \
--silver-dir output/silver \
--gold-dir output/gold \
--mlflowSee examples/evaluation/README.md for the full evaluation usage guide, including bootstrap-path context notes on null confidence and placeholder run IDs.
databricks-caseops-lakehouse/
├── README.md
├── PROJECT_SPEC.md # Scope and roadmap source of truth
├── ARCHITECTURE.md # Technical design source of truth
├── config/
│ ├── databricks.resources.example.yml # Unity Catalog layout reference (no credentials)
│ ├── databricks.resources.dev.example.yml # Dev environment layout (no credentials)
│ ├── databricks.resources.staging.example.yml # Staging environment layout (no credentials)
│ └── databricks.resources.prod.example.yml # Prod environment layout reference (no credentials)
├── docs/
│ ├── roadmap.md
│ ├── data-contracts.md
│ ├── evaluation-plan.md
│ ├── databricks-bootstrap.md # Personal Databricks bootstrap validation record
│ └── prompts/ # Excluded from version control
├── src/
│ ├── schemas/ # Pydantic / JSON Schema definitions
│ │ ├── bedrock_contract.py # Gold export payload contract validator
│ │ ├── delivery_event.py # Delivery event schema (v0.2.0)
│ │ ├── review_queue.py # Human review queue schema
│ │ ├── review_decision.py # Review decision and reprocessing request schemas
│ │ └── governance_monitoring.py # Governance monitoring schema and flag vocabulary
│ ├── pipelines/ # Bronze → Silver → Gold pipeline logic
│ │ ├── export_handoff.py # Export packaging and handoff service boundary
│ │ ├── handoff_report.py # Export outcome observability and batch reporting
│ │ ├── handoff_bundle.py # Batch manifest and review bundle packaging
│ │ ├── handoff_bundle_validation.py # Bundle integrity and consistency validation
│ │ ├── delivery_events.py # Delivery event materialization
│ │ ├── delta_share_handoff.py # Delta Sharing producer-side preparation layer
│ │ ├── review_queue.py # Human review queue derivation and materialization
│ │ └── governance_monitoring.py # Governance monitoring aggregation and reporting
│ ├── evaluation/ # Evaluation runners and report infrastructure
│ │ ├── eval_bronze.py
│ │ ├── eval_silver.py
│ │ ├── eval_gold.py # Null-confidence safe
│ │ ├── eval_traceability.py # Cross-layer traceability
│ │ ├── run_evaluation.py # Full-pipeline evaluation orchestrator
│ │ ├── report_models.py # Structured report dataclasses
│ │ ├── report_writer.py # JSON + text report writer
│ │ └── mlflow_experiment_paths.py # Environment-aware MLflow path resolution
│ └── utils/ # Shared helpers
│ └── environment_config.py # Environment model and resource naming
├── notebooks/
│ └── bootstrap/ # Validated Databricks bootstrap SQL
├── tests/ # 1,510 tests across all pipeline stages and contract layers
└── examples/
├── evaluation/ # Evaluation usage guide
├── expected_delivery_event.json # Reference delivery event fixture
└── ... # Sample documents and expected output fixtures
This repo does not include:
- A frontend or UI of any kind
- A standalone chatbot or conversational interface
- Generic Databricks demo notebooks
- Scala code
- Anything that requires production credentials to demonstrate
- Cross-case analytics, trend reporting, or KPI dashboards
- Historical operational intelligence or aggregate performance metrics
- Downstream agent orchestration, reasoning, or decision runtime (that is Bedrock CaseOps)
This repo is not yet, and does not aim to be, a mature analytics backbone or operational intelligence platform. It is the governed upstream document intelligence and AI-ready asset preparation layer.
This repo is the upstream governed preparation layer in a two-repository architecture. The boundary between them is intentional and non-negotiable.
| Repository | Role |
|---|---|
| databricks-caseops-lakehouse (this repo) | Upstream: governed document ingestion, parsing, extraction, classification, and AI-ready asset preparation on Databricks |
| bedrock-caseops-control-tower | Downstream: retrieval, agent reasoning, validation, and escalation workflows on AWS / Amazon Bedrock |
The handoff point between these systems is the formal Gold export payload — a schema-versioned, contract-enforced structured record produced by this repo and consumed by the Bedrock repo. The interface contract is defined in docs/bedrock-handoff-contract.md.
This repo does not own retrieval, generation, agent orchestration, or escalation logic. Those concerns belong entirely to the Bedrock CaseOps Control Tower.
If you're exploring this project, interested in governed AI data pipelines, or open to discussing data platform and AI engineering roles — feel free to reach out.
This project was developed with AI-assisted workflows. The system architecture, schema contracts, evaluation framework, and safety boundaries were intentionally designed and directed by the author, with AI tooling used to support and accelerate implementation as part of a modern engineering workflow.