# Document Extraction Engine - Architecture Specification

**Document:** Document Extraction Engine (Metis Platform Component)
**Project:** Metis Platform - AI-Powered Codebase Intelligence System
**Date:** 2025-12-08
**Author:** Hieu

---

## Executive Summary

This document provides comprehensive architecture documentation for the Document Extraction Engine, a core component of the Metis Platform's Knowledge Extraction Engine. It covers the container architecture, component design, and implementation guidelines for a DAG-based document processing system using Apache Airflow.

**Documentation Scope:**
- **Level 2 (Containers):** Airflow, Workers, MinIO, Redis, PostgreSQL, Milvus, Azure OpenAI
- **Level 3 (Components):** DAG definitions, Operators, Parsers, Event handlers

**Focus Areas:**
- Event Handling (MinIO webhook triggers)
- Authentication (MinIO access keys, environment variables)
- Output Storage (PostgreSQL for metadata, Milvus for vectors)

**PoC Scope:**
- Single document source: MinIO (S3-compatible object storage)
- Single-node Airflow deployment with LocalExecutor
- Streaming output pattern for chunk persistence

---

## Architecture Requirements & Constraints

### Functional Requirements

| Category | Requirement | Details |
|----------|-------------|---------|
| **Event Handling** | Webhook-based triggers | MinIO bucket notifications via webhooks (real-time) |
| **Event Types** | New file uploads only | `s3:ObjectCreated:*` events |
| **Document Formats** | Multi-format support | PDF, DOCX, Markdown, TXT, HTML |
| **Output** | Structured chunks + embeddings | Stream to PostgreSQL and Milvus |

### Non-Functional Requirements

| Category | Requirement | Details |
|----------|-------------|---------|
| **Authentication** | Access Key / Secret Key | MinIO credentials for bucket access |
| **Secrets Management** | Environment Variables | Simple PoC approach |
| **Metadata Storage** | PostgreSQL | Document chunks, metadata, relationships |
| **Vector Storage** | Milvus | Embeddings for semantic search |
| **Output Pattern** | Streaming | Stream chunks as parsed (not batch) |

### Technology Stack

| Component | Technology | Version | Notes |
|-----------|------------|---------|-------|
| Runtime | Python | 3.11+ | Required >=3.10, <3.14 for Airflow 3.x |
| Orchestration | Apache Airflow | 3.x (latest: 3.1.3) | Major rewrite with Task SDK, DAG Versioning |
| Document Parsing | Docling | Latest | Unified PDF, Office, Markdown parsing with AI |
| Document Storage | MinIO | Latest (RELEASE.2025-*) | S3-compatible object storage |
| Message Broker | Redis | 8.x (latest: 8.4) | New I/O threading, 30%+ throughput improvement |
| Metadata DB | PostgreSQL | 17+ (latest: 18.1) | New I/O subsystem, uuidv7() support |
| Vector DB | Milvus | 2.6.x (latest: 2.6.4) | RaBitQ 1-bit quantization, JSON path index |
| Embedding Provider | Azure OpenAI | text-embedding-ada-002 / text-embedding-3-small | 1536 dimensions |

**Version Sources:**
- [Apache Airflow Releases](https://github.com/apache/airflow/releases)
- [Docling GitHub](https://github.com/docling-project/docling)
- [Milvus Releases](https://github.com/milvus-io/milvus/releases)
- [Redis Downloads](https://redis.io/downloads/)
- [PostgreSQL Releases](https://www.postgresql.org/about/news/)

---

## Containers (Level 2)

### Container Architecture Overview

The Document Extraction Engine uses Apache Airflow as the orchestration backbone with a single-node deployment for PoC, designed to scale horizontally later.

### Containers

#### 1. Apache Airflow (Orchestrator)

**Technology:** Apache Airflow 3.x (Single-node with LocalExecutor)

**Responsibility:**
- Orchestrate document extraction DAGs
- Receive webhook triggers from MinIO via REST API
- Schedule and monitor extraction tasks
- Manage task dependencies and retries
- Expose REST API for external triggers

**Key Endpoints:**
- `POST /api/v1/dags/{dag_id}/dagRuns` - Trigger DAG execution
- `GET /api/v1/dags/{dag_id}/dagRuns/{run_id}` - Check DAG status
- `POST /api/v1/connections` - Manage MinIO/DB connections

**Communication:**
- Inbound: MinIO webhooks, API Gateway triggers, scheduled intervals
- Outbound: MinIO (fetch files), PostgreSQL (metadata), Milvus (vectors), Azure OpenAI (embeddings)

#### 2. MinIO (Document Source)

**Technology:** MinIO (S3-compatible object storage)

**Responsibility:**
- Store uploaded documents (PDF, DOCX, MD, etc.)
- Emit bucket notifications on file uploads
- Provide S3-compatible API for file retrieval

**Bucket Configuration:**
- `documents/` - Raw uploaded documents
- Webhook notifications enabled for `s3:ObjectCreated:*`

**Communication:**
- Outbound: Webhook POST to Airflow REST API on new uploads
- Inbound: S3 GET requests from Airflow workers

#### 3. Redis (Message Broker & Cache)

**Technology:** Redis 8.x

**Responsibility:**
- Airflow task queue (Celery broker for future scaling)
- Cache for processing state and deduplication
- Pub/Sub for real-time status updates

**Data Stored:**
- Task execution state
- Document processing cache (content hashes for deduplication)
- Temporary extraction results before streaming to DB

**Communication:**
- Bidirectional with Airflow workers
- Pub/Sub channels for progress events

#### 4. PostgreSQL (Metadata Storage)

**Technology:** PostgreSQL 17+ (latest: 18.x)

**Responsibility:**
- Store extracted document chunks with metadata
- Maintain document lineage and source attribution
- Store Airflow metadata (DAG runs, task instances)
- Knowledge graph edges (document relationships)

**Key Tables:**
- `document_sources` - MinIO bucket/object references
- `document_chunks` - Extracted text chunks with metadata
- `extraction_jobs` - Job tracking and audit log
- `knowledge_edges` - Relationships between chunks and code

**Communication:**
- Inbound: Writes from Airflow workers (streaming chunks)
- Queries from Knowledge Base API

#### 5. Milvus (Vector Storage)

**Technology:** Milvus 2.6.x (standalone for PoC)

**Responsibility:**
- Store document chunk embeddings
- Enable semantic similarity search
- Support hybrid search (vector + metadata filtering)

**Collections:**
- `document_embeddings` - Vector embeddings with chunk_id reference

**Communication:**
- Inbound: Embedding inserts from Airflow workers
- Queries from Semantic Search service

#### 6. Azure OpenAI (Embedding Provider)

**Technology:** Azure OpenAI Service (text-embedding-ada-002 or text-embedding-3-small)

**Responsibility:**
- Generate embeddings for document chunks
- Provide consistent vector dimensions for Milvus

**Communication:**
- Outbound API calls from Airflow embedding tasks
- Rate-limited and batched for cost efficiency

---

### Container Communication Diagram

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                    DOCUMENT EXTRACTION ENGINE (PoC)                         │
└─────────────────────────────────────────────────────────────────────────────┘

                              ┌──────────────────┐
                              │   API Gateway    │
                              │ (External Trigger)│
                              └────────┬─────────┘
                                       │ POST /api/v1/dags/.../dagRuns
                                       ▼
┌──────────────┐  Webhook     ┌──────────────────────────────────────────────┐
│              │  (s3:ObjectCreated)   │                                      │
│    MinIO     │─────────────────────▶│         Apache Airflow               │
│   (Documents)│                      │      ┌─────────────────────┐         │
│              │◀─────────────────────│      │   Webserver (UI)    │         │
│  buckets:    │   S3 GET (fetch)     │      ├─────────────────────┤         │
│  - documents/│                      │      │   Scheduler         │         │
└──────────────┘                      │      ├─────────────────────┤         │
                                      │      │   LocalExecutor     │         │
                                      │      │   (Workers)         │         │
                                      │      └─────────────────────┘         │
                                      └──────────────┬───────────────────────┘
                                                     │
                      ┌──────────────────────────────┼──────────────────────────────┐
                      │                              │                              │
                      ▼                              ▼                              ▼
            ┌──────────────────┐          ┌──────────────────┐          ┌──────────────────┐
            │      Redis       │          │    PostgreSQL    │          │     Milvus       │
            │  (Broker/Cache)  │          │   (Metadata)     │          │   (Vectors)      │
            │                  │          │                  │          │                  │
            │ - Task queue     │          │ - document_chunks│          │ - embeddings     │
            │ - Processing     │          │ - extraction_jobs│          │ - similarity     │
            │   state cache    │          │ - knowledge_edges│          │   search         │
            └──────────────────┘          └──────────────────┘          └──────────────────┘
                                                     │
                                                     ▼
                                          ┌──────────────────┐
                                          │  Azure OpenAI    │
                                          │  (Embeddings)    │
                                          │                  │
                                          │ text-embedding-  │
                                          │ ada-002          │
                                          └──────────────────┘
```

---

### Container Architecture Diagram (C4 Level 2 - Mermaid)

```mermaid
flowchart TB
    subgraph External["External Systems"]
        APIGateway["🚪 API Gateway<br/>Manual Triggers"]
        KnowledgeBase["🗄️ Knowledge Base<br/>Query Interface"]
    end

    subgraph DocExtractor["<b>DOCUMENT EXTRACTION ENGINE</b>"]
        subgraph Airflow["Apache Airflow (Single-node)"]
            Webserver["🌐 Webserver<br/>REST API + UI"]
            Scheduler["⏰ Scheduler<br/>DAG Orchestration"]
            Worker["⚙️ LocalExecutor<br/>Task Workers"]
        end

        MinIO["📦 MinIO<br/>Document Storage<br/>Bucket Notifications"]
        Redis["🔴 Redis<br/>Task Queue<br/>State Cache"]
        PostgreSQL["🐘 PostgreSQL<br/>Chunks & Metadata<br/>Extraction Jobs"]
        Milvus["🔷 Milvus<br/>Vector Embeddings<br/>Semantic Search"]
    end

    AzureOpenAI["☁️ Azure OpenAI<br/>Embedding API"]

    %% External triggers
    APIGateway -->|"POST /dagRuns"| Webserver
    MinIO -->|"Webhook<br/>s3:ObjectCreated"| Webserver

    %% Internal Airflow
    Webserver --> Scheduler
    Scheduler --> Worker

    %% Worker connections
    Worker -->|"S3 GET<br/>Fetch Documents"| MinIO
    Worker -->|"Task State<br/>Caching"| Redis
    Worker -->|"Stream Chunks<br/>Metadata"| PostgreSQL
    Worker -->|"Store Vectors"| Milvus
    Worker -->|"Generate<br/>Embeddings"| AzureOpenAI

    %% Output
    PostgreSQL -->|"Query"| KnowledgeBase
    Milvus -->|"Search"| KnowledgeBase

    style Airflow fill:#E8F5E9,stroke:#2E7D32
    style MinIO fill:#FFF3E0,stroke:#E65100
    style Redis fill:#FFEBEE,stroke:#C62828
    style PostgreSQL fill:#E3F2FD,stroke:#1565C0
    style Milvus fill:#F3E5F5,stroke:#7B1FA2
    style AzureOpenAI fill:#E0F7FA,stroke:#00838F
```

---

### Scaling Strategy (Future)

| Container | PoC (Current) | Production (Future) |
|-----------|---------------|---------------------|
| Airflow 3.x | LocalExecutor (single-node) | CeleryExecutor or KubernetesExecutor |
| Workers | Single worker | Multiple worker pools (parsing, embedding) |
| Redis 8.x | Single instance | Redis Cluster or Sentinel |
| PostgreSQL 17+ | Single instance | Read replicas, connection pooling |
| Milvus 2.6.x | Standalone | Milvus Cluster with multiple nodes |

---

## Components (Level 3)

This section details the major structural building blocks within the Document Extraction Engine containers.

---

### Airflow DAG Components

#### 1. Document Extraction DAG (`document_extraction_dag`)

**Purpose:** Main orchestration DAG triggered by MinIO webhooks or API calls

**DAG Configuration:**
```python
dag_id: "document_extraction_dag"
schedule_interval: None  # Event-triggered only
catchup: False
max_active_runs: 10
default_args:
  retries: 3
  retry_delay: timedelta(minutes=2)
```

**Task Flow:**
```
┌─────────────────────────────────────────────────────────────────────────┐
│                     document_extraction_dag                              │
└─────────────────────────────────────────────────────────────────────────┘

  ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
  │   validate   │────▶│    fetch     │────▶│    parse     │
  │   _event     │     │  _document   │     │  _document   │
  └──────────────┘     └──────────────┘     └──────────────┘
                                                   │
                                                   ▼
                                           ┌──────────────┐
                                           │    chunk     │
                                           │  _content    │
                                           └──────────────┘
                                                   │
                              ┌────────────────────┼────────────────────┐
                              ▼                    ▼                    ▼
                       ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
                       │   generate   │     │    store     │     │   extract    │
                       │  _embeddings │     │   _chunks    │     │ _relationships│
                       └──────────────┘     └──────────────┘     └──────────────┘
                              │                    │                    │
                              ▼                    ▼                    ▼
                       ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
                       │    store     │     │  PostgreSQL  │     │    store     │
                       │  _vectors    │     │   (done)     │     │   _edges     │
                       └──────────────┘     └──────────────┘     └──────────────┘
                              │                                        │
                              ▼                                        ▼
                       ┌──────────────┐                         ┌──────────────┐
                       │    Milvus    │                         │  PostgreSQL  │
                       │   (done)     │                         │   (done)     │
                       └──────────────┘                         └──────────────┘
```

---

### Task Components (Airflow Operators)

#### 1. ValidateEventOperator

**Purpose:** Validate incoming webhook payload and extract event metadata

**Responsibilities:**
- Parse MinIO webhook JSON payload
- Validate event type is `s3:ObjectCreated:*`
- Extract bucket name, object key, content type, size
- Check for duplicate events (idempotency via Redis)
- Reject unsupported file types

**Input:** `dag_run.conf` (webhook payload)
**Output:** `validated_event` XCom with event metadata

```python
@dataclass
class ValidateEventOperator:
    task_id: str = "validate_event"

    def parse_webhook_payload(self, conf: dict) -> dict: ...
    def validate_event_type(self, event: dict) -> bool: ...
    def check_duplicate(self, event_id: str) -> bool: ...
    def extract_metadata(self, event: dict) -> EventMetadata: ...

    # Output
    @dataclass
    class Output:
        bucket: str
        object_key: str
        content_type: str
        size_bytes: int
        event_id: str
        timestamp: datetime
```

#### 2. FetchDocumentOperator

**Purpose:** Download document from MinIO to worker local storage

**Responsibilities:**
- Authenticate with MinIO using access key/secret
- Download object to temporary local path
- Validate file integrity (size, checksum)
- Handle large files with streaming download

**Input:** `validated_event` XCom
**Output:** `local_file_path` XCom

```python
@dataclass
class FetchDocumentOperator:
    task_id: str = "fetch_document"
    connection_id: str = "minio_default"

    def get_minio_client(self) -> Minio: ...
    def download_object(self, bucket: str, key: str) -> Path: ...
    def validate_download(self, path: Path, expected_size: int) -> bool: ...

    # Output
    @dataclass
    class Output:
        local_path: Path
        content_type: str
        original_filename: str
```

#### 3. ParseDocumentOperator

**Purpose:** Extract raw content from document using format-specific parsers

**Responsibilities:**
- Detect document format (PDF, DOCX, MD, TXT, HTML, etc.)
- Route to appropriate parser
- Extract text content with structure preservation
- Extract metadata (title, author, dates, headings)

**Supported Formats (PoC):**

| Format | InputFormat | Notes |
|--------|-------------|-------|
| PDF | `InputFormat.PDF` | OCR support, table structure detection |
| DOCX | `InputFormat.DOCX` | Word documents with formatting preserved |
| XLSX | `InputFormat.XLSX` | Excel spreadsheets |
| PPTX | `InputFormat.PPTX` | PowerPoint presentations |
| Markdown | `InputFormat.MD` | Preserves structure, code blocks |

**Parser Library:** [Docling](https://github.com/docling-project/docling) - Unified document processing library supporting PDF, Office formats, and more with advanced AI-powered extraction.

```python
@dataclass
class ParseDocumentOperator:
    task_id: str = "parse_document"

    def detect_format(self, path: Path, content_type: str) -> DocumentFormat: ...
    def get_parser(self, format: DocumentFormat) -> BaseParser: ...
    def parse(self, path: Path) -> RawDocument: ...

    # Output
    @dataclass
    class Output:
        raw_content: str
        format: DocumentFormat
        metadata: DocumentMetadata
        structure: list[Section]
```

#### 4. ChunkContentOperator

**Purpose:** Split parsed content into semantic chunks for embedding

**Responsibilities:**
- Apply chunking strategy based on document type
- Maintain chunk overlap for context continuity
- Preserve section hierarchy in chunk metadata
- Generate chunk IDs for reference

**Chunking Strategies:**

| Strategy | Use Case | Parameters |
|----------|----------|------------|
| `semantic` | Long documents | max_tokens=512, overlap=50 |
| `section` | Structured docs (MD, DOCX) | split_on_headings=True |
| `fixed` | Plain text | chunk_size=1000, overlap=100 |

```python
@dataclass
class ChunkContentOperator:
    task_id: str = "chunk_content"

    def select_strategy(self, doc_type: str, structure: list) -> ChunkStrategy: ...
    def chunk_document(self, content: str, strategy: ChunkStrategy) -> list[Chunk]: ...
    def generate_chunk_id(self, chunk: Chunk, doc_id: str) -> str: ...

    # Output
    @dataclass
    class DocumentChunk:
        chunk_id: str
        content: str
        token_count: int
        section_path: str  # "Chapter 1 > Section 2"
        order_index: int
        metadata: dict
```

#### 5. GenerateEmbeddingsOperator

**Purpose:** Generate vector embeddings for each chunk via Azure OpenAI

**Responsibilities:**
- Batch chunks for API efficiency (max 16 per request)
- Call Azure OpenAI embedding endpoint
- Handle rate limiting with exponential backoff
- Cache embeddings by content hash (deduplication)

**Configuration:**
```python
@dataclass
class GenerateEmbeddingsOperator:
    task_id: str = "generate_embeddings"
    connection_id: str = "azure_openai_default"
    model: str = "text-embedding-ada-002"
    batch_size: int = 16
    max_retries: int = 5

    def batch_chunks(self, chunks: list[Chunk], size: int) -> list[list[Chunk]]: ...
    def generate_batch(self, batch: list[Chunk]) -> list[Vector]: ...
    def check_cache(self, content_hash: str) -> Vector | None: ...
    def cache_embedding(self, content_hash: str, vector: Vector) -> None: ...

    # Output
    @dataclass
    class ChunkEmbedding:
        chunk_id: str
        vector: list[float]  # 1536 dimensions
        model: str
        cached: bool
```

#### 6. StoreChunksOperator

**Purpose:** Stream document chunks to PostgreSQL

**Responsibilities:**
- Insert chunks with metadata to `document_chunks` table
- Update `document_sources` with processing status
- Handle conflicts (upsert on re-processing)
- Maintain transaction integrity

```python
@dataclass
class StoreChunksOperator:
    task_id: str = "store_chunks"
    connection_id: str = "postgres_default"

    def stream_insert(self, chunks: list[Chunk]) -> int: ...
    def upsert_source(self, source_metadata: SourceMetadata) -> None: ...
    def update_job_status(self, job_id: str, status: str) -> None: ...

    # Output
    @dataclass
    class Output:
        chunks_stored: int
        source_id: str
```

#### 7. StoreVectorsOperator

**Purpose:** Insert embeddings into Milvus vector database

**Responsibilities:**
- Connect to Milvus collection
- Batch insert vectors with chunk_id reference
- Create index if not exists
- Flush to ensure persistence

```python
@dataclass
class StoreVectorsOperator:
    task_id: str = "store_vectors"
    connection_id: str = "milvus_default"
    collection: str = "document_embeddings"

    def get_collection(self) -> Collection: ...
    def batch_insert(self, embeddings: list[ChunkEmbedding]) -> int: ...
    def ensure_index(self) -> None: ...
    def flush(self) -> None: ...

    # Output
    @dataclass
    class Output:
        vectors_stored: int
```

#### 8. ExtractRelationshipsOperator

**Purpose:** Identify and emit knowledge graph edges between chunks

**Responsibilities:**
- Detect cross-references within document
- Link to existing code entities (if referenced)
- Build document hierarchy edges (parent-child sections)
- Emit edges to PostgreSQL knowledge graph

```python
@dataclass
class ExtractRelationshipsOperator:
    task_id: str = "extract_relationships"

    def find_internal_refs(self, chunks: list[Chunk]) -> list[Edge]: ...
    def find_code_refs(self, chunks: list[Chunk], code_entities: list) -> list[Edge]: ...
    def build_hierarchy(self, chunks: list[Chunk]) -> list[Edge]: ...
    def emit_edges(self, edges: list[Edge]) -> int: ...

    # Output
    @dataclass
    class Output:
        edges_created: int
```

---

### Webhook Handler Component

#### MinIO Event Receiver

**Location:** Built into Airflow REST API
**Endpoint:** `POST /api/v1/dags/document_extraction_dag/dagRuns`

**Webhook Payload (MinIO → Airflow):**
```json
{
  "conf": {
    "EventName": "s3:ObjectCreated:Put",
    "Key": "documents/reports/quarterly-report.pdf",
    "Records": [
      {
        "s3": {
          "bucket": { "name": "documents" },
          "object": {
            "key": "reports/quarterly-report.pdf",
            "size": 1048576,
            "contentType": "application/pdf",
            "eTag": "abc123"
          }
        },
        "eventTime": "2025-12-08T10:30:00Z"
      }
    ]
  }
}
```

**MinIO Bucket Notification Config:**
```json
{
  "QueueConfigurations": [],
  "TopicConfigurations": [],
  "CloudFunctionConfigurations": [
    {
      "Id": "airflow-trigger",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            { "Name": "prefix", "Value": "documents/" }
          ]
        }
      },
      "CloudFunction": "http://airflow:8080/api/v1/dags/document_extraction_dag/dagRuns"
    }
  ]
}
```

---

### Parser Components (Docling-based)

#### Docling Document Converter

The Document Extraction Engine uses **Docling** as the unified parsing library. Docling provides a single `DocumentConverter` that handles all supported formats with consistent output.

```python
from docling.document_converter import (
    DocumentConverter,
    PdfFormatOption,
    WordFormatOption,
    ExcelFormatOption,
    PowerpointFormatOption,
    MarkdownFormatOption,
)
from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import PdfPipelineOptions

# Configure PDF options with OCR and table detection
pdf_options = PdfPipelineOptions()
pdf_options.do_ocr = True
pdf_options.do_table_structure = True

# Create converter with format-specific configurations
converter = DocumentConverter(
    allowed_formats=[
        InputFormat.PDF,
        InputFormat.DOCX,
        InputFormat.XLSX,
        InputFormat.PPTX,
        InputFormat.MD,
    ],
    format_options={
        InputFormat.PDF: PdfFormatOption(pipeline_options=pdf_options),
        InputFormat.DOCX: WordFormatOption(),
        InputFormat.XLSX: ExcelFormatOption(),
        InputFormat.PPTX: PowerpointFormatOption(),
        InputFormat.MD: MarkdownFormatOption(),
    }
)

# Convert document - format auto-detected
result = converter.convert("document.pdf")
markdown_output = result.document.export_to_markdown()
```

#### Supported Formats

| Format | InputFormat | Extensions | Features |
|--------|-------------|------------|----------|
| PDF | `InputFormat.PDF` | `.pdf` | OCR, table structure, layout detection |
| Word | `InputFormat.DOCX` | `.docx` | Headings, tables, formatting preserved |
| Excel | `InputFormat.XLSX` | `.xlsx` | Sheet extraction, cell data |
| PowerPoint | `InputFormat.PPTX` | `.pptx` | Slide content, speaker notes |
| Markdown | `InputFormat.MD` | `.md`, `.markdown` | Code blocks, links preserved |

#### Docling Chunking Integration

Docling provides built-in chunking via `HybridChunker` for semantic-aware document splitting:

```python
from docling.chunking import HybridChunker
from docling_core.transforms.chunker.tokenizer.huggingface import HuggingFaceTokenizer
from transformers import AutoTokenizer

# Setup tokenizer for chunk size control
EMBED_MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2"
tokenizer = HuggingFaceTokenizer(
    tokenizer=AutoTokenizer.from_pretrained(EMBED_MODEL_ID),
)

# Create chunker with tokenizer
chunker = HybridChunker(
    tokenizer=tokenizer,
    merge_peers=True,  # Merge small adjacent chunks
)

# Chunk the converted document
chunk_iter = chunker.chunk(dl_doc=result.document)

for chunk in chunk_iter:
    # Raw chunk text
    text = chunk.text
    # Context-enriched text (includes section headers)
    enriched_text = chunker.contextualize(chunk=chunk)
```

#### Installation

```bash
# Core Docling
pip install docling

# With chunking support (HuggingFace tokenizer)
pip install 'docling-core[chunking]'

# With OpenAI tokenizer
pip install 'docling-core[chunking-openai]'
```

---

### Data Models

#### DocumentChunk

```python
from dataclasses import dataclass
from datetime import datetime
from typing import Any

@dataclass
class DocumentChunk:
    chunk_id: str           # UUID
    source_id: str          # Reference to document_sources
    content: str            # Chunk text
    token_count: int        # For embedding cost tracking
    section_path: str       # "Chapter 1 > Section 2 > Subsection A"
    order_index: int        # Position in document
    content_hash: str       # SHA-256 for deduplication
    metadata: dict[str, Any]
    created_at: datetime
```

#### ChunkEmbedding

```python
@dataclass
class ChunkEmbedding:
    chunk_id: str           # Reference to DocumentChunk
    vector: list[float]     # 1536 dimensions (ada-002)
    model: str              # "text-embedding-ada-002"
    created_at: datetime
```

#### KnowledgeEdge

```python
@dataclass
class KnowledgeEdge:
    source_id: str          # From entity (chunk, code element)
    target_id: str          # To entity
    relationship_type: str  # "references", "parent_of", "related_to"
    confidence: float       # 0.0 - 1.0
    evidence: str           # Why this relationship exists
    created_at: datetime
```

---

### Component Interaction Diagram (Level 3)

```mermaid
flowchart TB
    subgraph Webhook["Webhook Handler"]
        EventReceiver["🔔 Event Receiver<br/>(Airflow REST API)"]
    end

    subgraph DAG["document_extraction_dag"]
        Validate["✅ ValidateEventOperator"]
        Fetch["📥 FetchDocumentOperator"]
        Parse["📄 ParseDocumentOperator"]
        Chunk["✂️ ChunkContentOperator"]
        Embed["🧠 GenerateEmbeddingsOperator"]
        StoreChunks["💾 StoreChunksOperator"]
        StoreVectors["📊 StoreVectorsOperator"]
        ExtractRels["🔗 ExtractRelationshipsOperator"]
    end

    subgraph Parsers["Parser Factory"]
        PdfParser["PDF Parser"]
        DocxParser["DOCX Parser"]
        MdParser["Markdown Parser"]
        HtmlParser["HTML Parser"]
    end

    subgraph Storage["Storage Layer"]
        MinIO["📦 MinIO"]
        Redis["🔴 Redis<br/>(Cache)"]
        PostgreSQL["🐘 PostgreSQL"]
        Milvus["🔷 Milvus"]
    end

    AzureOpenAI["☁️ Azure OpenAI"]

    EventReceiver -->|"dag_run.conf"| Validate
    Validate -->|"XCom: event_meta"| Fetch
    Fetch -->|"S3 GET"| MinIO
    Fetch -->|"XCom: local_path"| Parse
    Parse --> Parsers
    Parsers -->|"RawDocument"| Parse
    Parse -->|"XCom: raw_content"| Chunk
    Chunk -->|"XCom: chunks[]"| Embed
    Chunk -->|"XCom: chunks[]"| StoreChunks
    Chunk -->|"XCom: chunks[]"| ExtractRels

    Embed -->|"Check cache"| Redis
    Embed -->|"API call"| AzureOpenAI
    Embed -->|"XCom: embeddings[]"| StoreVectors

    StoreChunks -->|"INSERT"| PostgreSQL
    StoreVectors -->|"INSERT"| Milvus
    ExtractRels -->|"INSERT edges"| PostgreSQL

    style Webhook fill:#E8F5E9,stroke:#2E7D32
    style DAG fill:#FFF8E1,stroke:#FF8F00
    style Parsers fill:#E3F2FD,stroke:#1565C0
    style Storage fill:#F3E5F5,stroke:#7B1FA2
```

---

### Database Schema (PostgreSQL)

```sql
-- Document sources (MinIO objects)
CREATE TABLE document_sources (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    bucket VARCHAR(255) NOT NULL,
    object_key VARCHAR(1024) NOT NULL,
    content_type VARCHAR(255),
    size_bytes BIGINT,
    etag VARCHAR(255),
    status VARCHAR(50) DEFAULT 'pending',  -- pending, processing, completed, failed
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(bucket, object_key, etag)
);

-- Document chunks
CREATE TABLE document_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    source_id UUID REFERENCES document_sources(id),
    content TEXT NOT NULL,
    token_count INTEGER,
    section_path VARCHAR(1024),
    order_index INTEGER,
    content_hash VARCHAR(64),  -- SHA-256
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(source_id, order_index)
);

-- Knowledge edges
CREATE TABLE knowledge_edges (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    source_entity_id UUID NOT NULL,
    target_entity_id UUID NOT NULL,
    relationship_type VARCHAR(100),
    confidence FLOAT,
    evidence TEXT,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Extraction jobs (audit)
CREATE TABLE extraction_jobs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    dag_run_id VARCHAR(255),
    source_id UUID REFERENCES document_sources(id),
    status VARCHAR(50),
    chunks_created INTEGER,
    vectors_created INTEGER,
    edges_created INTEGER,
    error_message TEXT,
    started_at TIMESTAMP,
    completed_at TIMESTAMP
);

-- Indexes
CREATE INDEX idx_chunks_source ON document_chunks(source_id);
CREATE INDEX idx_chunks_hash ON document_chunks(content_hash);
CREATE INDEX idx_edges_source ON knowledge_edges(source_entity_id);
CREATE INDEX idx_edges_target ON knowledge_edges(target_entity_id);
```

### Milvus Collection Schema

```python
from pymilvus import DataType

collection_schema = {
    "collection_name": "document_embeddings",
    "fields": [
        {"name": "id", "type": DataType.VARCHAR, "max_length": 36, "is_primary": True},
        {"name": "chunk_id", "type": DataType.VARCHAR, "max_length": 36},
        {"name": "vector", "type": DataType.FLOAT_VECTOR, "dim": 1536},
        {"name": "source_id", "type": DataType.VARCHAR, "max_length": 36},
        {"name": "created_at", "type": DataType.INT64}
    ],
    "index": {
        "field_name": "vector",
        "index_type": "IVF_FLAT",
        "metric_type": "COSINE",
        "params": {"nlist": 1024}
    }
}
```

---

## Design Patterns & Best Practices

### Architectural Patterns

#### 1. DAG-Based Orchestration Pattern

**Pattern:** Directed Acyclic Graph (DAG) for workflow orchestration

**Implementation:** Apache Airflow DAGs define task dependencies and execution order

**Benefits:**
- Visual representation of extraction pipeline
- Built-in retry and failure handling
- Task-level parallelism where dependencies allow
- Audit trail of all executions

#### 2. Event-Driven Architecture Pattern

**Pattern:** Webhook-triggered processing with event sourcing

**Implementation:** MinIO bucket notifications trigger Airflow DAG runs

**Benefits:**
- Real-time processing on document upload
- Decoupled producers (MinIO) and consumers (Airflow)
- No polling overhead
- Natural backpressure via Airflow concurrency limits

#### 3. Factory Pattern (Parser Selection)

**Pattern:** Factory Method for parser instantiation

**Implementation:** `ParserFactory` creates appropriate parser based on file extension

**Benefits:**
- Easy to add new document formats
- Single responsibility per parser
- Runtime format detection
- Consistent interface across parsers

#### 4. Operator Pattern (Airflow Tasks)

**Pattern:** Command Pattern via Airflow Operators

**Implementation:** Each task encapsulated in a custom Operator class

**Benefits:**
- Reusable task logic
- Testable in isolation
- Consistent interface for Airflow scheduler
- Built-in XCom for inter-task communication

#### 5. Streaming Output Pattern

**Pattern:** Stream processing for chunk persistence

**Implementation:** Chunks written to PostgreSQL as they're generated, not batched at end

**Benefits:**
- Reduced memory footprint
- Partial results available on failure
- Better progress visibility
- Lower latency to first result

---

### Cross-Cutting Concerns

#### 1. Authentication & Secrets Management

**Pattern:** Airflow Connections for credential management

**Implementation:**

| Service | Connection ID | Auth Method |
|---------|---------------|-------------|
| MinIO | `minio_default` | Access Key / Secret Key |
| PostgreSQL | `postgres_default` | Username / Password |
| Milvus | `milvus_default` | Token (optional) |
| Azure OpenAI | `azure_openai_default` | API Key + Endpoint |

**Environment Variables (PoC):**
```bash
# MinIO
MINIO_ACCESS_KEY=minioadmin
MINIO_SECRET_KEY=minioadmin123
MINIO_ENDPOINT=minio:9000

# PostgreSQL
POSTGRES_HOST=postgres
POSTGRES_USER=metis
POSTGRES_PASSWORD=metis123
POSTGRES_DB=document_extraction

# Milvus
MILVUS_HOST=milvus
MILVUS_PORT=19530

# Azure OpenAI
AZURE_OPENAI_API_KEY=your-api-key
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com
AZURE_OPENAI_API_VERSION=2024-02-15-preview
```

#### 2. Error Handling & Resilience

**Pattern:** Retry with exponential backoff + Dead Letter Queue

| Error Type | Handling Strategy |
|------------|-------------------|
| Transient (network, rate limit) | Retry 3x with exponential backoff |
| Permanent (unsupported format) | Fail task, log error, continue pipeline |
| Partial failure | Store successful chunks, mark source as `partial` |
| Critical (DB down) | Fail DAG, alert, manual intervention |

**Airflow Retry Configuration:**
```python
default_args = {
    'retries': 3,
    'retry_delay': timedelta(minutes=2),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=30),
}
```

#### 3. Idempotency & Deduplication

**Pattern:** Content-hash based deduplication

**Implementation:**
- Each chunk gets SHA-256 hash of content
- Embeddings cached by content hash in Redis
- Database UNIQUE constraints prevent duplicate chunks
- Event ID tracking prevents duplicate DAG runs

#### 4. Observability & Monitoring

**Key Metrics (Prometheus):**

| Metric | Type | Description |
|--------|------|-------------|
| `documents_processed_total` | Counter | Total documents processed |
| `chunks_created_total` | Counter | Total chunks created |
| `embeddings_generated_total` | Counter | Total embeddings generated |
| `embedding_cache_hits` | Counter | Cache hit count |
| `embedding_latency_seconds` | Histogram | Azure OpenAI call latency |
| `dag_run_duration_seconds` | Histogram | End-to-end DAG duration |

---

### Security Patterns

#### 1. Principle of Least Privilege

- MinIO: Read-only access to `documents/` bucket
- PostgreSQL: INSERT/SELECT only, no DROP/TRUNCATE
- Milvus: Collection-specific permissions
- Azure OpenAI: Embedding model only, no completion

#### 2. Input Validation

```python
ALLOWED_CONTENT_TYPES = {
    'application/pdf',
    'application/vnd.openxmlformats-officedocument.wordprocessingml.document',  # DOCX
    'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',         # XLSX
    'application/vnd.openxmlformats-officedocument.presentationml.presentation', # PPTX
    'text/markdown',
}

MAX_FILE_SIZE = 100 * 1024 * 1024  # 100MB

def validate_event(event: dict) -> bool:
    if event['content_type'] not in ALLOWED_CONTENT_TYPES:
        raise UnsupportedFormatError()
    if event['size_bytes'] > MAX_FILE_SIZE:
        raise FileTooLargeError()
    return True
```

---

## Implementation Guidelines

### Project Structure

```
document-extraction-engine/
├── dags/
│   ├── __init__.py
│   ├── document_extraction_dag.py      # Main DAG definition
│   └── config.py                       # DAG configuration constants
├── operators/
│   ├── __init__.py
│   ├── validate_event.py               # ValidateEventOperator
│   ├── fetch_document.py               # FetchDocumentOperator
│   ├── parse_document.py               # ParseDocumentOperator
│   ├── chunk_content.py                # ChunkContentOperator
│   ├── generate_embeddings.py          # GenerateEmbeddingsOperator
│   ├── store_chunks.py                 # StoreChunksOperator
│   ├── store_vectors.py                # StoreVectorsOperator
│   └── extract_relationships.py        # ExtractRelationshipsOperator
├── parsers/
│   ├── __init__.py
│   ├── docling_converter.py            # Docling DocumentConverter wrapper
│   ├── chunker.py                      # HybridChunker configuration
│   └── format_options.py               # Format-specific options (PDF, Office, MD)
├── models/
│   ├── __init__.py
│   ├── document.py                     # DocumentChunk, RawDocument
│   ├── embedding.py                    # ChunkEmbedding
│   └── edge.py                         # KnowledgeEdge
├── services/
│   ├── __init__.py
│   ├── minio_client.py                 # MinIO wrapper
│   ├── postgres_client.py              # PostgreSQL wrapper
│   ├── milvus_client.py                # Milvus wrapper
│   ├── azure_openai_client.py          # Azure OpenAI wrapper
│   └── redis_client.py                 # Redis cache wrapper
├── utils/
│   ├── __init__.py
│   ├── chunking.py                     # Chunking strategies
│   ├── hashing.py                      # Content hashing
│   └── validation.py                   # Input validation
├── migrations/
│   ├── 001_create_document_sources.sql
│   ├── 002_create_document_chunks.sql
│   ├── 003_create_knowledge_edges.sql
│   └── 004_create_extraction_jobs.sql
├── tests/
│   ├── __init__.py
│   ├── test_operators/
│   ├── test_parsers/
│   ├── test_services/
│   └── fixtures/
├── docker/
│   ├── Dockerfile.airflow
│   ├── docker-compose.yml
│   └── .env.example
├── scripts/
│   ├── setup_minio_webhook.sh
│   ├── create_milvus_collection.py
│   └── run_migrations.py
├── pyproject.toml
└── README.md
```

---

### Getting Started

#### 1. Prerequisites

```bash
# Required software
- Docker & Docker Compose v2.x+
- Python 3.11+ (required: >=3.10, <3.14 for Airflow 3.x)
- MinIO CLI (mc)

# Azure OpenAI access
- Azure subscription
- Azure OpenAI resource with text-embedding-ada-002 or text-embedding-3-small deployed
```

#### 2. Local Development Setup

```bash
# Clone repository
git clone <repo-url>
cd document-extraction-engine

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac
# or: venv\Scripts\activate  # Windows

# Install dependencies
pip install -r requirements.txt

# Copy environment template
cp docker/.env.example docker/.env
# Edit docker/.env with your Azure OpenAI credentials

# Start infrastructure
docker-compose -f docker/docker-compose.yml up -d

# Run database migrations
python scripts/run_migrations.py

# Create Milvus collection
python scripts/create_milvus_collection.py

# Setup MinIO webhook
./scripts/setup_minio_webhook.sh
```

#### 3. Docker Compose Configuration

```yaml
# docker/docker-compose.yml
version: '3.8'

services:
  airflow:
    build:
      context: ..
      dockerfile: docker/Dockerfile.airflow
    ports:
      - "8080:8080"
    environment:
      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
      - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
      - AIRFLOW__WEBSERVER__SECRET_KEY=your-secret-key
    env_file:
      - .env
    volumes:
      - ../dags:/opt/airflow/dags
      - ../operators:/opt/airflow/plugins/operators
      - ../parsers:/opt/airflow/plugins/parsers
      - ../services:/opt/airflow/plugins/services
    depends_on:
      - postgres
      - redis
      - minio
      - milvus

  postgres:
    image: postgres:17
    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:8
    ports:
      - "6379:6379"

  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      - MINIO_ROOT_USER=minioadmin
      - MINIO_ROOT_PASSWORD=minioadmin123
    volumes:
      - minio_data:/data

  milvus:
    image: milvusdb/milvus:v2.6.4
    ports:
      - "19530:19530"
    environment:
      - ETCD_ENDPOINTS=etcd:2379
      - MINIO_ADDRESS=minio:9000
    depends_on:
      - etcd

  etcd:
    image: quay.io/coreos/etcd:v3.5.0
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000

volumes:
  postgres_data:
  minio_data:
```

---

### Common Development Tasks

#### Adding a New Document Format (Docling)

Docling already supports PDF, DOCX, XLSX, PPTX, and Markdown. To add custom processing or a new format:

1. **Add format to allowed formats:**

```python
# services/docling_converter.py
from docling.document_converter import DocumentConverter
from docling.datamodel.base_models import InputFormat

def create_converter() -> DocumentConverter:
    return DocumentConverter(
        allowed_formats=[
            InputFormat.PDF,
            InputFormat.DOCX,
            InputFormat.XLSX,
            InputFormat.PPTX,
            InputFormat.MD,
            InputFormat.HTML,  # Add new format
        ],
        format_options={
            # Add format-specific options as needed
        }
    )
```

2. **Configure format-specific options:**

```python
from docling.document_converter import HTMLFormatOption
from docling.datamodel.backend_options import HTMLBackendOptions

# Custom HTML options
html_options = HTMLBackendOptions(parse_images=True)

converter = DocumentConverter(
    format_options={
        InputFormat.HTML: HTMLFormatOption(backend_options=html_options),
    }
)
```

3. **Add tests:**

```python
# tests/test_services/test_docling_converter.py
import pytest
from services.docling_converter import create_converter

def test_html_conversion():
    converter = create_converter()
    result = converter.convert("tests/fixtures/sample.html")
    assert result.status.value == "success"
    assert result.document.export_to_markdown()
```

4. **Update validation allowlist:**

```python
# utils/validation.py
ALLOWED_CONTENT_TYPES = {
    'application/pdf',
    'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
    'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
    'application/vnd.openxmlformats-officedocument.presentationml.presentation',
    'text/markdown',
    'text/html',  # Add new content type
}
```

---

#### Triggering DAG Manually (Testing)

```bash
# Via Airflow CLI
airflow dags trigger document_extraction_dag \
  --conf '{"Records": [{"s3": {"bucket": {"name": "documents"}, "object": {"key": "test/sample.pdf", "size": 1024, "contentType": "application/pdf"}}}]}'

# Via REST API
curl -X POST "http://localhost:8080/api/v1/dags/document_extraction_dag/dagRuns" \
  -H "Content-Type: application/json" \
  -u "admin:admin" \
  -d '{
    "conf": {
      "Records": [{
        "s3": {
          "bucket": {"name": "documents"},
          "object": {"key": "test/sample.pdf", "size": 1024, "contentType": "application/pdf"}
        }
      }]
    }
  }'
```

---

#### Uploading Test Document to MinIO

```bash
# Install MinIO client
brew install minio/stable/mc  # Mac
# or download from https://min.io/download

# Configure client
mc alias set local http://localhost:9000 minioadmin minioadmin123

# Create bucket
mc mb local/documents

# Upload file (triggers webhook)
mc cp sample.pdf local/documents/

# List files
mc ls local/documents/
```

---

### Troubleshooting Guide

| Problem | Symptoms | Solution |
|---------|----------|----------|
| **DAG not triggering** | No new DAG runs after upload | Check MinIO webhook config; verify Airflow API is accessible; check `mc admin trace local` |
| **Parse failure** | Task fails at `parse_document` | Check file format is supported; verify file isn't corrupted; check parser logs |
| **Embedding timeout** | `generate_embeddings` times out | Azure OpenAI rate limited; increase retry delay; reduce batch size |
| **Milvus insert fails** | Vector insert errors | Check collection exists; verify vector dimensions (1536); check Milvus logs |
| **Duplicate processing** | Same file processed multiple times | Check Redis cache connection; verify event deduplication logic |
| **Memory issues** | Worker OOM on large files | Increase worker memory; implement streaming for large documents |

#### Viewing Logs

```bash
# Airflow task logs
docker-compose logs -f airflow

# Specific DAG run logs
airflow tasks logs document_extraction_dag parse_document <run_id>

# MinIO webhook events
mc admin trace local --verbose

# PostgreSQL queries
docker-compose exec postgres psql -U airflow -c "SELECT * FROM extraction_jobs ORDER BY started_at DESC LIMIT 10;"
```

---

### Testing Strategy

#### Unit Tests

```bash
# Run all unit tests
pytest tests/ -v

# Run specific test file
pytest tests/test_parsers/test_pdf_parser.py -v

# Run with coverage
pytest tests/ --cov=operators --cov=parsers --cov-report=html
```

#### Integration Tests

```bash
# Start test infrastructure
docker-compose -f docker/docker-compose.test.yml up -d

# Run integration tests
pytest tests/integration/ -v --integration

# Cleanup
docker-compose -f docker/docker-compose.test.yml down -v
```

---

### Configuration Reference

#### Environment Variables

| Variable | Description | Default |
|----------|-------------|---------|
| `MINIO_ENDPOINT` | MinIO server address | `minio:9000` |
| `MINIO_ACCESS_KEY` | MinIO access key | Required |
| `MINIO_SECRET_KEY` | MinIO secret key | Required |
| `MINIO_BUCKET` | Document bucket name | `documents` |
| `POSTGRES_HOST` | PostgreSQL host | `postgres` |
| `POSTGRES_PORT` | PostgreSQL port | `5432` |
| `POSTGRES_USER` | PostgreSQL user | `airflow` |
| `POSTGRES_PASSWORD` | PostgreSQL password | Required |
| `POSTGRES_DB` | PostgreSQL database | `document_extraction` |
| `MILVUS_HOST` | Milvus host | `milvus` |
| `MILVUS_PORT` | Milvus port | `19530` |
| `MILVUS_COLLECTION` | Milvus collection name | `document_embeddings` |
| `REDIS_HOST` | Redis host | `redis` |
| `REDIS_PORT` | Redis port | `6379` |
| `AZURE_OPENAI_API_KEY` | Azure OpenAI API key | Required |
| `AZURE_OPENAI_ENDPOINT` | Azure OpenAI endpoint | Required |
| `AZURE_OPENAI_API_VERSION` | API version | `2024-02-15-preview` |
| `AZURE_OPENAI_EMBEDDING_MODEL` | Embedding model deployment name | `text-embedding-ada-002` |
| `EMBEDDING_BATCH_SIZE` | Chunks per embedding API call | `16` |
| `CHUNK_MAX_TOKENS` | Maximum tokens per chunk | `512` |
| `CHUNK_OVERLAP` | Token overlap between chunks | `50` |

---

### Extending for Production

#### 1. Scale Airflow Workers

```python
# Change executor in airflow.cfg or environment
AIRFLOW__CORE__EXECUTOR=CeleryExecutor
AIRFLOW__CELERY__BROKER_URL=redis://redis:6379/0
AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql://airflow:airflow@postgres/airflow
```

#### 2. Add New Document Source

When extending beyond MinIO:

1. Create new connector in `services/`
2. Add authentication handling
3. Create webhook receiver endpoint
4. Update `validate_event` operator
5. Add source-specific tests

#### 3. Add Monitoring

```python
from prometheus_client import Counter, Histogram

documents_processed = Counter('documents_processed_total', 'Total documents processed')
embedding_latency = Histogram('embedding_latency_seconds', 'Embedding generation latency')
```

---

### Onboarding Checklist

For new developers joining the project:

- [ ] Read this architecture document completely
- [ ] Set up local development environment (Python 3.11+)
- [ ] Run the test suite successfully
- [ ] Upload a test document and trace the full pipeline
- [ ] Review existing DAG code and operators
- [ ] Add a simple enhancement (e.g., new parser, metric)
- [ ] Submit first PR with tests

---

## How to Use This Documentation

**For Architects:**
- Review Containers (Level 2) for system boundaries
- Check Components (Level 3) for internal structure
- Reference Design Patterns for consistency

**For Developers (Implementation):**
- Start with Implementation Guidelines
- Follow Project Structure for code organization
- Use Common Development Tasks for quick reference

**For Developers (Maintenance):**
- Use Troubleshooting Guide for debugging
- Check Configuration Reference for environment setup
- Review Testing Strategy for validation

---

## Related Documentation

- **[Main Architecture](./metis-platform-design.md)** - Complete Metis Platform architecture
- **[Containers Level 2](./metis-containers-level-2.md)** - Parent container architecture
- **[System Context Level 1](./metis-system-context-level-1.md)** - High-level system boundary

---

_Last Updated: 2025-12-08_