---
title: 'Migrate to LlamaIndex Hierarchical Chunking'
slug: 'hierarchical-chunking-migration'
created: '2026-01-30'
status: 'ready-for-dev'
stepsCompleted: [1, 2, 3, 4]
tech_stack:
  - Python 3.x (asyncio)
  - Airflow (DAG orchestration)
  - llama-index-core
  - llama-index-vector-stores-milvus
  - llama-index-storage-docstore-postgres
  - Milvus (vector store)
  - PostgreSQL (docstore + extracted_records)
  - Azure OpenAI (text-embedding-3-large, 3072 dims)
  - tiktoken (cl100k_base tokenizer)
  - LibreOffice (DOCX/PPTX → PDF conversion)
  - openpyxl (Excel detailed pipeline)
files_to_modify:
  - airflow/dags/tasks/processing_tasks.py
  - airflow/dags/tasks/parse_tasks.py
  - airflow/dags/document_extraction_dag.py
  - src/extraction_v2/excel_parser_service.py
files_to_delete:
  - src/extraction_v2/semantic_chunker.py
  - src/knowledge/vector_store.py
  - src/knowledge/embeddings.py
files_to_create:
  - airflow/dags/tasks/preprocess_tasks.py (or add to existing)
code_patterns:
  - DAG uses external_python tasks with pickle serialization
  - Branching by document type via detect_format task
  - Excel dual pipelines merge at merge_pipeline_results task
  - Domain-aware RAG with domain_category, domain_terms, tags
  - Semantic tagging with ISO 29148 + risk + info + topics
  - Package management via uv (use `uv add` for dependencies)
test_patterns:
  - Integration tests via Airflow DAG execution
  - Sample files for each document type
  - Success validation via DAG status + DB verification
---

# Tech-Spec: Migrate to LlamaIndex Hierarchical Chunking

**Created:** 2026-01-30

## Overview

### Problem Statement

The current document chunking uses Docling HybridChunker which produces flat, single-level chunks (8000 tokens max). This lacks parent-child relationships needed for better context retrieval and doesn't support query-time context expansion.

### Solution

Replace Docling HybridChunker with LlamaIndex HierarchicalNodeParser across all document types (PDF, Word, PowerPoint, Excel detailed pipeline). Implement 3-level hierarchy (2048/512/256 tokens) with:
- **LlamaIndex MilvusVectorStore** for leaf node storage (replaces custom `VectorStore`)
- **LlamaIndex PostgresDocumentStore** for all nodes (parent + leaf)

Data flow change:
```
Current:  chunks → custom VectorStore → Milvus (pymilvus)
Target:   nodes → LlamaIndex MilvusVectorStore → Milvus
                → LlamaIndex PostgresDocumentStore → PostgreSQL
```

**Unified Storage Architecture:**
```
┌─────────────────────────────────────────────────────────────────┐
│                   LlamaIndex MilvusVectorStore                  │
├─────────────────────────────────────────────────────────────────┤
│  PDF/Word/PPTX           │  Excel Detailed    │  Excel Large   │
│  ───────────────         │  ───────────────   │  ────────────  │
│  Hierarchical chunks     │  Hierarchical      │  1 row = 1 doc │
│  [2048/512/256 tokens]   │  (images→desc)     │  No hierarchy  │
│  Parent nodes→Docstore   │  Parent→Docstore   │  Direct store  │
│  Leaf nodes→Milvus       │  Leaf→Milvus       │  →Milvus       │
└─────────────────────────────────────────────────────────────────┘
```

### Scope

**In Scope:**
- New `HierarchicalChunker` class using LlamaIndex `HierarchicalNodeParser`
- 3-level token-based hierarchy: [2048, 512, 256] tokens
- LlamaIndex MilvusVectorStore for leaf node storage
- LlamaIndex PostgresDocumentStore for all nodes (parent + leaf)
- Milvus schema update (`textiq_records` collection) with hierarchy fields
- All document types: PDF, Word, PowerPoint, Excel (both pipelines)
- Image/Shape/Flowchart preprocessing: replace with AI descriptions before chunking
- LibreOffice conversion for DOCX/PPTX page metadata
- DAG task updates: new `hierarchical_chunk_task`, unified storage task
- **Excel Large Table Pipeline Update:**
  - Update store record task to use LlamaIndex MilvusVectorStore
  - 1 record = 1 Document (no hierarchical parsing)
  - Text = natural language description of the row
  - `record_json` in metadata = raw JSON for reconstruction
- Remove old chunking code:
  - Delete `SemanticChunker` class (`src/extraction_v2/semantic_chunker.py`)
  - Delete `chunk_text_task` from DAG
  - Delete custom `VectorStore` class (`src/knowledge/vector_store.py`)
  - Remove `RecursiveCharacterTextSplitter` usage in Excel detailed pipeline
  - Clean up related imports and dead code

**Out of Scope:**
- AutoMergingRetriever
- Query layer changes (future work)
- Data migration (recreate collection fresh)

## Context for Development

### Codebase Patterns

**DAG Architecture:**
- Entry: `validate_event` → `fetch_document` → `save_document_metadata` → `detect_format` (branch)
- Text path (Word/PDF/PPTX): `parse_*` → `chunk_text` → `extract_domain_terms_text` → `tag_records_text` → `generate_embeddings_text` → `save_extracted_records_text` → `store_vectors_text`
- Excel path: Dual pipelines (Large Table + Detailed) → `merge_pipeline_results` → `save_extracted_records_excel` → `store_vectors_excel`
- Exit: `update_document_status` → `cleanup_temp_files`

**Task Execution Pattern:**
- All processing tasks use `@task.external_python` decorator
- Data passed via pickle files (XCom for metadata only)
- Timeouts: Parse=120min, Embedding=60min, Store=30min
- Concurrency pools: `llm_api_pool: 3`, `embedding_pool: 2`

**Current Chunking (SemanticChunker):**
- Uses Docling `HybridChunker` with 8000 token max
- `OpenAITokenizer` with tiktoken for text-embedding-3-large
- Custom `AnnotationPictureSerializer` for image captions
- Produces flat chunks with metadata: `page_number`, `source_filename`, `contains_table`

**Excel Large Table Pipeline:**
- `LargeTableExtractor` extracts rows as JSON records
- `RecordToNaturalLanguage.convert_batch()` generates NL descriptions
- 1 row = 1 record with `_source` metadata (sheet, row, filename)
- Currently uses custom `VectorStore` for Milvus storage

**Excel Detailed Pipeline:**
- `ExcelParserService.parse_excel_file()` extracts images/flowcharts/shapes
- `ExcelImageExtractionService` generates AI descriptions via Azure Vision
- Images replaced in-place at cell location (ZIP manipulation)
- `RecursiveCharacterTextSplitter` for text-by-sheet (4000 char max)
- Produces `ChunkedElementDTO` objects

### Files to Reference

| File | Purpose |
| ---- | ------- |
| `scripts/hierarchical_llamaindex_milvus.py` | **Reference implementation** - HierarchicalNodeParser + MilvusVectorStore + PostgresDocumentStore |
| `src/extraction_v2/semantic_chunker.py` | Current chunker (DELETE) - understand metadata schema |
| `src/knowledge/vector_store.py` | Current Milvus storage (DELETE) - understand MilvusEntity schema |
| `src/knowledge/embeddings.py` | Current EmbeddingService (DELETE) - LlamaIndex handles this |
| `airflow/dags/tasks/processing_tasks.py` | **Main update target** - chunk_text_task, store_vectors_task |
| `airflow/dags/tasks/parse_tasks.py` | Excel parse tasks - parse_excel_large_tables, parse_excel_detailed |
| `airflow/dags/document_extraction_dag.py` | DAG flow definition - task dependencies |
| `src/extraction_v2/excel_parser_service.py` | Detailed pipeline - image preprocessing location |
| `src/extraction_v2/large_table_extractor.py` | Large table extraction logic |
| `src/extraction_v2/record_to_natural_language.py` | NL description generation |
| `src/adapters/legacy_to_milvus_adapter.py` | DETAILED pipeline Milvus adapter (update for LlamaIndex) |

### Key Function Signatures to Update

**Text Pipeline:**
```python
# Current: processing_tasks.py
def chunk_text_task(document_id, parse_result, max_tokens=8000) -> Dict
def store_vectors_task(document_id, embedded_result, pg_records_result) -> Dict

# Target: Replace with hierarchical chunking + LlamaIndex storage
```

**Excel Large Table Pipeline:**
```python
# KEEP: processing_tasks.py
def convert_large_tables_to_natural_language(chunks_result) -> Dict
# ↑ Generates NL descriptions from JSON records - ESSENTIAL for search

# UPDATE: After NL conversion, create LlamaIndex Documents
# Document(text=NL_description, metadata={"record_json": original_JSON, ...})

# DELETE: store_vectors_task (uses custom VectorStore)
# REPLACE WITH: LlamaIndex MilvusVectorStore.add()
```

**Excel Detailed Pipeline:**
```python
# Current: legacy_to_milvus_adapter.py
class LegacyToMilvusAdapter:
    async def adapt_and_store(chunks, document_id, ...) -> Dict

# Target: Convert ChunkedElementDTO → LlamaIndex Document → hierarchical chunking
```

### DAG Flow (Current → Target)

**Current Text Pipeline:**
```
parse_* → chunk_text → extract_domain_terms → tag_records → generate_embeddings → save_records → store_vectors
```

**Target Text Pipeline:**
```
parse_* → hierarchical_chunk → extract_domain_terms → tag_records → save_to_llamaindex_stores
         (HierarchicalNodeParser)                                   (MilvusVectorStore + PostgresDocumentStore)
```

**Current Excel Large Table:**
```
parse_excel_large_tables → detect_headers → chunk_records → convert_to_nl → domain_terms → tag → embeddings → merge → save → store_vectors
```

**Target Excel Large Table:**
```
parse_excel_large_tables → detect_headers → chunk_records → convert_to_nl → domain_terms → tag → save_to_llamaindex_stores
                                                           ↓ (KEEP)                            (MilvusVectorStore only, no hierarchy)
                                            NL description = Document.text
                                            JSON record = Document.metadata["record_json"]
```

**IMPORTANT:** `convert_large_tables_to_natural_language()` is **KEPT** - it generates the searchable text content from JSON records.

**Output format of `convert_large_tables_to_natural_language()`:**
```python
{
    'records': [
        {
            'text': "Natural language description of the row...",  # NL for embedding
            'content': {'Column A': 'val1', 'Column B': 'val2'},   # Original JSON
            '_source': {'filename': 'file.xlsx', 'sheet': 'Sheet1', 'row': 42},
            'headers': ['Column A', 'Column B'],
            'tags': [],  # Populated by tagging task
            'domain_category': 'general',  # Populated by domain terms task
            'domain_terms': [],  # Populated by domain terms task
        },
        ...
    ]
}
```

**Flow:**
1. Extract rows as JSON records
2. **Convert to NL descriptions** (existing logic - KEEP, returns records with 'text' field)
3. **Domain terms extraction** (AFTER NL conversion, adds domain_category/domain_terms to records)
4. **Tagging** (AFTER domain terms, adds tags to records)
5. **Create LlamaIndex Documents** in `store_large_table_to_llamaindex()`: `text=record['text']`, `metadata.record_json=record['content']`
6. Store via MilvusVectorStore (no hierarchical parsing needed)

### Reference Implementation Patterns

**From `scripts/hierarchical_llamaindex_milvus.py`:**

```python
# 1. Tokenizer setup (reuse)
TIKTOKEN_ENCODER = tiktoken.get_encoding("cl100k_base")

# 2. HierarchicalNodeParser configuration
node_parser_map = {}
for chunk_size in [2048, 512, 256]:
    splitter = SentenceSplitter(
        chunk_size=chunk_size,
        chunk_overlap=int(chunk_size * 0.2),  # 20% overlap
        tokenizer=tiktoken_tokenizer,
        paragraph_separator="\n\n",
    )
    node_parser_map[f"chunk_size_{chunk_size}"] = splitter

node_parser = HierarchicalNodeParser(
    node_parser_ids=[f"chunk_size_{s}" for s in chunk_sizes],
    node_parser_map=node_parser_map,
)

# 3. Storage setup
vector_store = MilvusVectorStore(
    uri=f"http://{host}:{port}",
    collection_name="textiq_records",
    dim=3072,
    overwrite=True,  # Recreate collection
)

docstore = PostgresDocumentStore.from_uri(
    uri=postgres_url,
    table_name="hierarchical_nodes",
    namespace="hierarchical",
    perform_setup=True,
    use_jsonb=True,
)

storage_context = StorageContext.from_defaults(
    vector_store=vector_store,
    docstore=docstore,
)

# 4. Critical pattern: ALL nodes → docstore, LEAF nodes → Milvus
storage_context.docstore.add_documents(nodes)
index = VectorStoreIndex(leaf_nodes, storage_context=storage_context)
```

### Technical Decisions

1. **Chunk Sizes**: [2048, 512, 256] tokens with 20% overlap per level
2. **Docstore Backend**: PostgreSQL with **new tables created by LlamaIndex** (not reusing `extracted_records`)
3. **Milvus Collection**: Recreate `textiq_records` manually with verification to ensure success
4. **Image Handling**: Replace images with AI descriptions **in-place at their cell location** before chunking
5. **Page Metadata**: Use LibreOffice for DOCX/PPTX to PDF conversion (accurate page numbers)
6. **Dual-Store Consistency**: Trust LlamaIndex library to handle MilvusVectorStore + PostgresDocumentStore writes
7. **Metadata Strategy**: Store `domain_category`, `domain_terms`, `tags` in LlamaIndex `Document.metadata` dict - flows automatically to nodes and storage
8. **Embedding Handling**: LlamaIndex MilvusVectorStore handles embedding automatically (no manual `EmbeddingService` calls)
9. **Excel Large Table**: Update existing store task to use LlamaIndex MilvusVectorStore with new schema
10. **Azure Embedding Config**: Reuse existing Azure OpenAI credentials from `settings.py`
11. **PostgresDocumentStore**: Reuse existing `DATABASE_URL` from settings
12. **Domain Terms & Tagging**: Keep existing tasks unchanged - they operate on node.metadata AFTER chunking (enrichment happens post-chunking)
13. **merge_pipeline_results**: Update to work with new LlamaIndex document format
14. **Image Preprocessing**: Runs BEFORE parsing (new `preprocess_excel_images` task)

### Metadata Flow

```
Document(metadata={...})
    ↓ HierarchicalNodeParser
Node(metadata={...})  ← inherits + adds node_id, parent_id, etc.
    ↓ MilvusVectorStore.add()
    ↓ (auto-embeds text using configured embedding model)
Milvus record with vector + metadata JSON field
```

**Note:** LlamaIndex MilvusVectorStore handles embedding internally. Configure once:
```python
from llama_index.vector_stores.milvus import MilvusVectorStore

vector_store = MilvusVectorStore(
    uri="http://milvus:19530",
    collection_name="textiq_records",
    dim=3072,  # text-embedding-3-large dimensions
)
# Embeddings generated automatically on add()
```

**Document.metadata schema:**
```python
{
    "document_id": "uuid",
    "filename": "report.docx",
    "document_type": "word|excel|pdf|pptx",
    "page_number": int,  # or start_page/end_page for multi-page chunks
    # Domain-aware RAG fields
    "domain_category": "finance|hr|engineering|general",
    "domain_terms": ["term1", "term2"],
    "tags": ["req:functional", "topic:reporting"],
    # Excel-specific (detailed pipeline)
    "sheet_name": "Sheet1",
    "record_json": {...}  # original row data for reconstruction
}
```

### Excel Image Preprocessing Flow (Detailed Pipeline)

**IMPORTANT:** Image preprocessing happens BEFORE parsing, not during.

```
┌─────────────────────┐
│ 1. Extract Images   │  ← ExcelImageExtractionService (new preprocessing task)
│    from Excel       │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 2. Generate AI      │  ← Azure Vision API
│    Descriptions     │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 3. Replace Images   │  ← Modify Excel file (ZIP manipulation)
│    with Text at     │     Image at B5 → "Description text" in B5
│    Cell Location    │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 4. Parse Modified   │  ← parse_excel_detailed (now sees text, not images)
│    Excel File       │
└─────────┬───────────┘
          ↓
┌─────────────────────┐
│ 5. Hierarchical     │  ← HierarchicalNodeParser
│    Chunking         │
└─────────────────────┘
```

**Implementation Note:** Create new `preprocess_excel_images` task that runs BEFORE `parse_excel_detailed`.

### Excel Large Table Pipeline (Updated)

```
┌─────────────────┐    ┌──────────────────┐    ┌─────────────────────────┐
│ Extract Row     │ →  │ Generate NL Desc │ →  │ Create LlamaIndex Doc   │
│ as JSON record  │    │ (existing logic) │    │ text=desc, meta=json    │
└─────────────────┘    └──────────────────┘    └─────────────────────────┘
                                                           ↓
                                               ┌─────────────────────────┐
                                               │ LlamaIndex MilvusStore  │
                                               │ (1 record = 1 vector)   │
                                               └─────────────────────────┘
```

**Large Table Document Schema:**
```python
Document(
    text="Natural language description of the row...",
    metadata={
        "document_id": "uuid",
        "filename": "data.xlsx",
        "document_type": "excel",
        "sheet_name": "Sheet1",
        "row_number": 42,
        "record_json": {"Column A": "val1", "Column B": "val2", ...},
        "domain_category": "finance",
        "domain_terms": ["revenue", "margin"],
        "tags": ["data:tabular"]
    }
)
```

## Implementation Plan

### Tasks

#### Phase 1: Dependencies

- [ ] **Task 1: Add LlamaIndex dependencies**
  - File: `pyproject.toml` (managed by uv)
  - Action: Use `uv add` to install packages:
    ```bash
    uv add llama-index-core
    uv add llama-index-vector-stores-milvus
    uv add llama-index-storage-docstore-postgres
    uv add llama-index-embeddings-azure-openai
    ```
  - Notes: Verify imports work after installation. Check `uv.lock` is updated.

#### Phase 2: Excel Detailed Pipeline Updates

- [ ] **Task 2: Create Excel image preprocessing task**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Add new task function:
    ```python
    @task.external_python(...)
    def preprocess_excel_images(fetch_result: Dict) -> Dict:
        # 1. Load Excel file
        # 2. Extract images with ExcelImageExtractionService
        # 3. Generate AI descriptions via Azure Vision
        # 4. Replace images with text at cell locations (ZIP manipulation)
        # 5. Save modified Excel file
        # 6. Return path to modified file
    ```
  - Notes: Extract logic from ExcelParserService, run BEFORE parse_excel_detailed

- [ ] **Task 3: Update parse_excel_detailed task**
  - File: `airflow/dags/tasks/parse_tasks.py`
  - Action: Modify to:
    - Accept preprocessed Excel file (images already replaced with text)
    - Remove image extraction logic (moved to Task 4)
    - Parse with Docling to get DoclingDocument
    - Return DoclingDocument for hierarchical chunking
  - Notes: No longer produces ChunkedElementDTO directly

- [ ] **Task 4: Add hierarchical chunking task for Excel Detailed**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Add task using LlamaIndex directly:
    ```python
    import tiktoken
    from llama_index.core import Document
    from llama_index.core.node_parser import HierarchicalNodeParser, SentenceSplitter

    TIKTOKEN_ENCODER = tiktoken.get_encoding("cl100k_base")

    @task.external_python(...)
    def hierarchical_chunk_excel_detailed(parse_result: Dict, document_id: str) -> Dict:
        # Setup HierarchicalNodeParser directly
        chunk_sizes = [2048, 512, 256]
        node_parser = HierarchicalNodeParser.from_defaults(
            chunk_sizes=chunk_sizes,
            chunk_overlap=int(chunk_sizes[-1] * 0.2),  # 20% overlap
        )

        # Convert parse result to LlamaIndex Documents
        documents = []
        for sheet_data in parse_result['sheets']:
            doc = Document(
                text=sheet_data['markdown'],
                metadata={
                    "document_id": document_id,
                    "sheet_name": sheet_data['sheet_name'],
                    "document_type": "excel",
                }
            )
            documents.append(doc)

        # Apply hierarchical parsing
        nodes = node_parser.get_nodes_from_documents(documents)
        return {"nodes": nodes, "document_id": document_id}
    ```

#### Phase 3: Text Pipeline Updates (Word/PDF/PPTX)

- [ ] **Task 5: Replace chunk_text_task with hierarchical_chunk_task**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Create new task using LlamaIndex directly:
    ```python
    import tiktoken
    from llama_index.core import Document
    from llama_index.core.node_parser import HierarchicalNodeParser

    @task.external_python(...)
    def hierarchical_chunk_text(parse_result: Dict, document_id: str, filename: str) -> Dict:
        """Hierarchically chunk Word/PDF/PPTX documents using LlamaIndex."""
        # Setup HierarchicalNodeParser directly
        chunk_sizes = [2048, 512, 256]
        node_parser = HierarchicalNodeParser.from_defaults(
            chunk_sizes=chunk_sizes,
            chunk_overlap=int(chunk_sizes[-1] * 0.2),
        )

        # Convert DoclingDocument pages to LlamaIndex Documents
        documents = []
        for page_num, page_content in enumerate(parse_result['pages'], 1):
            doc = Document(
                text=page_content,
                metadata={
                    "document_id": document_id,
                    "filename": filename,
                    "document_type": parse_result['document_type'],
                    "page_number": page_num,
                }
            )
            documents.append(doc)

        # Apply hierarchical parsing
        nodes = node_parser.get_nodes_from_documents(documents)
        return {"nodes": nodes, "document_id": document_id}
    ```
  - Notes: Keep function signature compatible for domain_terms/tagging tasks

- [ ] **Task 6: Update domain terms extraction for nodes**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Modify `extract_domain_terms_text` to:
    - Accept nodes instead of chunks
    - Extract domain_category and domain_terms
    - Store in node.metadata dict
    - Return enriched nodes
  - Notes: Logic stays same, just operates on node.metadata

- [ ] **Task 7: Update tagging task for nodes**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Modify `tag_records_text` to:
    - Accept nodes instead of chunks
    - Apply tags to node.metadata["tags"]
    - Return tagged nodes
  - Notes: Logic stays same, just operates on node.metadata

#### Phase 4: Excel Large Table Pipeline Updates

- [ ] **Task 8: Update store task for Large Table pipeline**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Create/modify task with explicit Document conversion using LlamaIndex directly:
    ```python
    from llama_index.core import Document, VectorStoreIndex, StorageContext
    from llama_index.vector_stores.milvus import MilvusVectorStore
    from src.core.config import settings

    @task.external_python(...)
    def store_large_table_to_llamaindex(nl_result: Dict, document_id: str) -> Dict:
        """Convert NL records to LlamaIndex Documents and store.

        Input: nl_result from convert_large_tables_to_natural_language()
               Contains list of records with 'text' (NL description) and 'content' (JSON)
        """
        # Initialize LlamaIndex stores directly
        vector_store = MilvusVectorStore(
            uri=f"http://{settings.milvus_host}:{settings.milvus_port}",
            collection_name=settings.milvus_collection,
            dim=3072,
        )
        storage_context = StorageContext.from_defaults(vector_store=vector_store)

        documents = []
        for record in nl_result['records']:
            nl_text = record['text']
            source = record.get('_source', {})

            doc = Document(
                text=nl_text,
                metadata={
                    "document_id": document_id,
                    "filename": source.get('filename', ''),
                    "document_type": "excel",
                    "sheet_name": source.get('sheet', ''),
                    "row_number": source.get('row', 0),
                    "record_json": record.get('content', {}),
                    "domain_category": record.get('domain_category', 'general'),
                    "domain_terms": record.get('domain_terms', []),
                    "tags": record.get('tags', []),
                }
            )
            documents.append(doc)

        # Store directly via VectorStoreIndex (no hierarchy)
        index = VectorStoreIndex.from_documents(documents, storage_context=storage_context)
        return {"stored_count": len(documents), "document_id": document_id}
    ```
  - Notes: No hierarchical parsing - 1 record = 1 document. Uses LlamaIndex directly, no wrapper class.

- [ ] **Task 9: Update merge_pipeline_results**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Modify to:
    - Accept nodes from Detailed pipeline
    - Accept documents from Large Table pipeline
    - Combine into unified format for storage
  - Notes: Both pipelines now produce LlamaIndex-compatible objects

#### Phase 5: Unified Storage Task

- [ ] **Task 10: Create unified LlamaIndex storage task**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Create new task using LlamaIndex directly:
    ```python
    from llama_index.core import VectorStoreIndex, StorageContext
    from llama_index.core.node_parser import get_leaf_nodes
    from llama_index.vector_stores.milvus import MilvusVectorStore
    from llama_index.storage.docstore.postgres import PostgresDocumentStore
    from src.core.config import settings

    @task.external_python(...)
    def store_to_llamaindex(nodes_result: Dict, document_id: str) -> Dict:
        """Store hierarchical nodes to PostgreSQL docstore + Milvus.

        Pattern: ALL nodes → docstore, LEAF nodes only → Milvus
        """
        nodes = nodes_result['nodes']

        # Initialize LlamaIndex stores directly
        vector_store = MilvusVectorStore(
            uri=f"http://{settings.milvus_host}:{settings.milvus_port}",
            collection_name=settings.milvus_collection,
            dim=3072,
        )
        docstore = PostgresDocumentStore.from_uri(
            uri=settings.database_url,
            table_name="hierarchical_nodes",
            namespace="hierarchical",
            perform_setup=True,
            use_jsonb=True,
        )
        storage_context = StorageContext.from_defaults(
            vector_store=vector_store,
            docstore=docstore,
        )

        # ALL nodes → docstore
        storage_context.docstore.add_documents(nodes)

        # LEAF nodes only → Milvus
        leaf_nodes = get_leaf_nodes(nodes)
        index = VectorStoreIndex(leaf_nodes, storage_context=storage_context)

        return {"stored_count": len(nodes), "leaf_count": len(leaf_nodes), "document_id": document_id}
    ```
  - Notes: Replaces store_vectors_text, store_vectors_excel. Uses LlamaIndex directly, no wrapper class.

#### Phase 6: DAG Updates

- [ ] **Task 11: Update document_extraction_dag.py**
  - File: `airflow/dags/document_extraction_dag.py`
  - Action: Update task dependencies:
    - Text path: `parse_*` → `hierarchical_chunk_text` → `extract_domain_terms` → `tag_records` → `store_to_llamaindex`
    - Excel Detailed: `preprocess_excel_images` → `parse_excel_detailed` → `hierarchical_chunk_excel_detailed` → ... → `store_to_llamaindex`
    - Excel Large Table: ... → `convert_to_nl` → `domain_terms` → `tag` → `store_large_table_to_llamaindex`
    - Remove: `generate_embeddings_*` tasks (LlamaIndex handles)
    - Remove: `save_extracted_records_*` tasks (docstore handles)
    - Remove: `store_vectors_*` tasks (replaced by store_to_llamaindex)
  - Notes: Keep merge_pipeline_results for Excel

#### Phase 7: Cleanup

- [ ] **Task 12: Delete old chunking code**
  - Files to DELETE:
    - `src/extraction_v2/semantic_chunker.py`
    - `src/knowledge/vector_store.py`
    - `src/knowledge/embeddings.py`
  - Action: Remove files and all imports referencing them
  - Notes: Search codebase for imports to clean up

- [ ] **Task 13: Clean up legacy adapter**
  - File: `src/adapters/legacy_to_milvus_adapter.py`
  - Action: Delete or update to use LlamaIndex
  - Notes: Used by Excel Detailed pipeline

- [ ] **Task 14: Remove unused DAG tasks**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Delete:
    - `chunk_text_task` (replaced by hierarchical_chunk_text)
    - `generate_embeddings_text` (LlamaIndex handles)
    - `generate_embeddings_large_tables` (LlamaIndex handles)
    - `generate_embeddings_chunks` (LlamaIndex handles)
    - `store_vectors_text` (replaced by store_to_llamaindex)
    - `store_vectors_excel` (replaced by store_to_llamaindex)
  - Notes: Keep save_extracted_records if needed for PostgreSQL record linking

#### Phase 8: Manual Operations

- [ ] **Task 15: Recreate Milvus collection**
  - Action: Manual operation:
    1. Connect to Milvus
    2. Drop `textiq_records` collection
    3. Let LlamaIndex MilvusVectorStore recreate with new schema on first write
    4. Verify collection exists with correct fields
  - Notes: Coordinate with team, ensure no production data loss

### Acceptance Criteria

#### Core Functionality

- [ ] **AC1:** Given a Word document (.docx), when processed through the DAG, then hierarchical nodes are created with 3 levels [2048/512/256 tokens] and stored in PostgreSQL docstore with parent-child relationships

- [ ] **AC2:** Given a PDF document, when processed through the DAG, then leaf nodes are stored in Milvus with correct metadata (document_id, filename, page_number, domain_category, domain_terms, tags)

- [ ] **AC3:** Given a PowerPoint document (.pptx), when processed through the DAG, then slide-level hierarchy is preserved in node metadata

- [ ] **AC4:** Given an Excel file with images (detailed pipeline), when processed, then images are replaced with AI descriptions at their cell locations BEFORE parsing, and the resulting text is hierarchically chunked

- [ ] **AC5:** Given an Excel file with large tables (>1000 rows), when processed through large table pipeline, then each row is stored as 1 LlamaIndex Document with text=NL_description and metadata.record_json=original_JSON

#### Metadata Preservation

- [ ] **AC6:** Given any document, when chunked and stored, then domain_category, domain_terms, and tags are preserved in node.metadata and queryable in Milvus

- [ ] **AC7:** Given an Excel document, when stored, then sheet_name and row_number (for large tables) or sheet_name (for detailed) are preserved in metadata

#### Storage Verification

- [ ] **AC8:** Given hierarchical nodes stored, when querying PostgreSQL docstore, then all nodes (parent + leaf) are present with valid parent_id references

- [ ] **AC9:** Given hierarchical nodes stored, when querying Milvus, then ONLY leaf nodes are present with 3072-dimensional vectors

- [ ] **AC10:** Given the DAG completes successfully, when checking Airflow UI, then all tasks show success status with no errors

#### Backward Compatibility

- [ ] **AC11:** Given the old code is deleted, when running the DAG, then no import errors occur and all document types process successfully

- [ ] **AC12:** Given the Milvus collection is recreated, when querying with existing query patterns, then results are returned (schema compatible)

## Additional Context

### Dependencies

**New packages (install via `uv add`):**
- `llama-index-core`
- `llama-index-vector-stores-milvus`
- `llama-index-storage-docstore-postgres`
- `llama-index-embeddings-azure-openai`
- `tiktoken` (may already exist)

**Existing (no changes):**
- LibreOffice (Docker image must have it installed)
- Azure OpenAI (text-embedding-3-large deployment)
- Milvus server
- PostgreSQL server

### Testing Strategy

**Test Files Required:**
| Format | Extensions | Test Coverage |
|--------|------------|---------------|
| Word | `.docx`, `.doc` | Hierarchical chunks, page metadata |
| PowerPoint | `.pptx` | Slide-level hierarchy |
| Excel | `.xlsx`, `.xls`, `.xlsm` | Both pipelines: Large Table + Detailed with images |
| PDF | `.pdf` | Hierarchical chunks, page metadata |

**Unit Tests:**
- `HierarchicalChunker.chunk_document()` produces correct node hierarchy
- `LlamaIndexStore.store_hierarchical_nodes()` stores to both Milvus and PostgreSQL
- `LlamaIndexStore.store_flat_documents()` stores to Milvus only
- Metadata preservation through chunking pipeline

**Integration Tests:**
- Full DAG execution for each document type
- Verify Airflow task success status
- Query Milvus to verify leaf nodes present
- Query PostgreSQL docstore to verify all nodes present

**Manual Verification:**
- Milvus collection schema matches expected fields
- Parent-child relationships are valid (no orphan nodes)
- Domain metadata (domain_category, domain_terms, tags) queryable

### Risk Assessment

| Risk | Likelihood | Impact | Mitigation |
|------|------------|--------|------------|
| Milvus collection recreation data loss | Low | High | Manual operation with team coordination |
| LlamaIndex API breaking changes | Low | Medium | Pin package versions in requirements.txt |
| PostgreSQL docstore connection issues | Low | Medium | Reuse existing DATABASE_URL, test connectivity |
| Image preprocessing breaks Excel parsing | Medium | Medium | Thorough testing with various Excel files |
| Domain terms/tagging tasks incompatible with nodes | Low | Low | Minimal changes - just operate on metadata dict |

### Notes

- Query layer will be updated in future work to leverage hierarchy
- Collection recreation is a manual operation with verification - no automated rollback
- LibreOffice dependency accepted for accurate page metadata (Docker image has it installed)
- Excel large table pipeline updated to use LlamaIndex MilvusVectorStore (unified storage layer)
- Hierarchy verification utility deferred to query layer work (not in scope)
- `EmbeddingService` class will be deleted - LlamaIndex handles embeddings internally
- `convert_large_tables_to_natural_language()` is KEPT - essential for generating searchable text (returns records with 'text' field containing NL descriptions)
- Existing domain terms and tagging tasks require minimal changes (operate on metadata dict)
- **Async/Sync Handling:** LlamaIndex classes used directly in tasks (no wrapper). LlamaIndex sync operations are compatible with Airflow `@task.external_python`
- **Metadata Enrichment Order:** Domain terms and tags are added AFTER chunking (post-processing on nodes/records)
- **Milvus/PostgreSQL tables:** Created automatically by LlamaIndex libraries on first use (no manual schema definition needed)