# DSOL - Technical Specification: Knowledge Base & Indexing

**Author:** TextIQ
**Date:** 2025-11-27
**Project Level:** Complex (BMad Method)
**Change Type:** New Feature - Epic 4 Implementation
**Development Context:** Brownfield - Building on existing extraction pipeline

---

## Context

### Available Documents

**Product Requirements Document (`docs/prd.md`):**
- DSOL is an LLM-powered Excel extraction service for AI Q&A
- 119 Excel files with complex tables, symbols, and cross-references
- Hybrid retrieval required: exact lookups (codes, symbols) + semantic search (conceptual queries)
- FR18-21: Knowledge Base requirements (store, index exact, index semantic, maintain sources)
- Target: ≥90% answer accuracy with 100% source traceability

**Architecture Document (`docs/architecture.md`):**
- Dual-store architecture: PostgreSQL (structured) + Milvus (vector)
- PostgreSQL: JSONB columns for flexible content, GIN indexes for fast JSON queries
- Milvus: Vector similarity search with metadata filtering
- Azure OpenAI text-embedding-3-large (3072 dimensions)
- Module structure: `src/knowledge/` for indexing, `src/retrieval/` for query (future)

**Epic Breakdown (`docs/epics.md`):**
- Epic 4: Knowledge Base & Indexing (5 stories)
- Story 4.1: Structured Store (PostgreSQL persistence)
- Story 4.2: Source Metadata Preservation (full traceability)
- Story 4.3: Exact Lookup Indexing (codes, symbols, sheet names)
- Story 4.4: Vector Embedding and Indexing (semantic search via Milvus)
- Story 4.5: Incremental Indexing (add/delete without full rebuild)

**Sprint Artifacts (`docs/sprint-artifacts/`):**
- Stories 3.1-3.4 COMPLETE: Extraction pipeline produces `ExtractedRecord` objects
- Each record has: content (dict), headers (list), _source (metadata), _merge_metadata (optional)
- Pipeline: TableDetector → HeaderExtractor → MergedCellResolver → RecordBuilder
- Ready to consume extracted records and index them

### Project Stack

**Runtime & Framework:**
- Python 3.11+
- FastAPI 0.122 (async web framework)
- Uvicorn 0.30+ (ASGI server)

**Data Layer:**
- **PostgreSQL 18** (structured store)
  - SQLAlchemy 2.0 (async ORM)
  - asyncpg 0.29+ (async driver)
  - Alembic 1.13 (migrations)
  - JSONB columns for flexible content storage
  - GIN indexes for fast JSON queries

- **Milvus 1.16** (vector store)
  - milvus-client 1.16+
  - Cosine distance for similarity
  - Metadata filtering support
  - Local deployment (dev), cloud (prod)

**LLM & Embeddings:**
- Azure OpenAI (via openai SDK 1.0+)
- text-embedding-3-large model (3072 dimensions)
- Batch embedding API (up to 100 texts per call)

**Processing & Queue:**
- Celery 5.3+ (async task processing)
- Redis 7+ (task queue backend)

**Utilities:**
- structlog 24.0 (structured logging)
- Pydantic 2.0+ (data validation)
- httpx 0.27+ (async HTTP)

**Development Tools:**
- pytest 8.0+ with pytest-asyncio 0.23+
- Black 24.0 (formatter, line-length: 100)
- Ruff 0.5+ (linter)
- mypy 1.10+ (type checker, strict mode)

### Existing Codebase Structure

**✅ Extraction Pipeline (Stories 3.1-3.4 COMPLETE):**
```
src/extraction/
├── models.py                  # TableBoundary, HeaderInfo, MergeMetadata, ExtractedRecord
├── table_detector.py          # Story 3.1: Table boundary detection
├── header_extractor.py        # Story 3.2: Multi-level header extraction
├── merged_cell_resolver.py    # Story 3.3: Merged cell propagation
└── record_builder.py          # Story 3.4: Structured record generation
```

**✅ Database Infrastructure:**
```
src/db/
├── models.py                  # ORM models: documents, extracted_records, etc.
├── session.py                 # SQLAlchemy async session factory
└── migrations/                # Alembic migrations (schema exists from Story 1.2)
```

**✅ API & Services:**
```
src/api/                       # FastAPI routes, schemas, middleware
src/services/                  # document_service, file_storage, auth_service, etc.
src/workers/                   # Celery app, background tasks
src/core/                      # config, exceptions, logging
```

**🎯 TO BE IMPLEMENTED (This Tech Spec):**
```
src/knowledge/                 # ← NEW: Indexing layer
├── structured_store.py        # Story 4.1: PostgreSQL operations
├── indexer.py                 # Story 4.3: Index management
├── embeddings.py              # Story 4.4: Text → vector conversion
└── vector_store.py            # Story 4.4: Milvus operations (stub exists, needs implementation)

src/retrieval/                 # ← FUTURE: Query processing (Epic 5)
src/qa/                        # ← FUTURE: Answer generation (Epic 5)
src/llm/                       # ← FUTURE: Azure OpenAI client (Epic 5)
```

---

## The Change

### Problem Statement

The extraction pipeline (Stories 3.1-3.4) produces high-quality structured records from Excel files, but these records are not yet stored or indexed for retrieval. Without a knowledge base layer, we cannot:

1. **Persist extracted data** - Records exist only in memory during processing
2. **Perform exact lookups** - Cannot query by item code ("find 2024-4_0019"), symbol ("what does ○ mean?"), or sheet name
3. **Enable semantic search** - Cannot answer conceptual queries ("tell me about construction results")
4. **Maintain source traceability** - Need to reference exact file/sheet/row for every answer
5. **Support incremental updates** - Need to add/delete documents without re-indexing everything

The Q&A feature (Epic 5) depends entirely on this indexing layer. Without it, DSOL cannot answer user questions.

### Proposed Solution

Implement **Epic 4: Knowledge Base & Indexing** - a dual-store indexing system that:

**1. Structured Store (PostgreSQL) - for Exact Lookups:**
- Store extracted records in `extracted_records` table with JSONB columns
- Create GIN indexes on `content` and `resolved_content` for fast JSON queries
- Enable queries like: "find item code 2024-4_0019", "find all records with ○ in Screen column"
- Query performance: < 50ms for exact matches

**2. Vector Store (Milvus) - for Semantic Search:**
- Convert each record to text representation
- Embed using Azure OpenAI text-embedding-3-large (3072 dimensions)
- Store vectors in Milvus with metadata (document_id, document_type, filename, metadata JSON)
- Enable similarity search for conceptual queries
- Query performance: < 200ms for semantic search

**3. Source Metadata Preservation:**
- Every record includes complete source metadata: document_id, filename, document_type, and type-specific metadata (Excel: sheet_name/row_number, Word: page_number)
- Enables exact citation: "Found in distribution_spec.xlsx, sheet 'Item Mapping', row 42"
- 100% source traceability for all answers

**4. Incremental Indexing:**
- Add new documents without re-indexing existing ones
- Delete documents cleanly (PostgreSQL CASCADE + Milvus filter delete)
- Re-process documents atomically (delete old, insert new)

**Architecture Flow:**
```
┌─────────────────────────────────────────────────────────────────┐
│                    EXTRACTION PIPELINE                          │
│  (Stories 3.1-3.4 COMPLETE)                                     │
│  TableDetector → HeaderExtractor → MergedCellResolver →         │
│  RecordBuilder                                                  │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
                    List[ExtractedRecord]
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    INDEXING LAYER (Epic 4)                      │
│  ┌───────────────────────┐  ┌────────────────────────────────┐ │
│  │ Structured Store      │  │ Vector Store                   │ │
│  │ (PostgreSQL)          │  │ (Milvus)                       │ │
│  │                       │  │                                │ │
│  │ • Store records       │  │ • Embed records                │ │
│  │ • Create GIN indexes  │  │ • Store vectors + metadata     │ │
│  │ • Exact lookups       │  │ • Semantic similarity search   │ │
│  └───────────────────────┘  └────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
                                │
                                ▼
                        Knowledge Base
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Q&A PIPELINE (Epic 5 - Future)               │
│  QueryRouter → Retrieval → AnswerGenerator                      │
└─────────────────────────────────────────────────────────────────┘
```

### Scope

**In Scope:**

1. **Story 4.1: Structured Store (PostgreSQL)**
   - Persist `ExtractedRecord` objects to `extracted_records` table
   - Store content as JSONB, resolved_content as JSONB, headers as JSONB
   - Batch insert optimization (500 records per transaction)
   - Full source metadata storage

2. **Story 4.2: Source Metadata Preservation**
   - Ensure every record has complete source metadata
   - Store: document_id, filename, document_type, and metadata JSON (Excel: sheet_name/row_number/has_symbols, Word: page_number/contains_table)
   - Enable exact citation and human verification

3. **Story 4.3: Exact Lookup Indexing**
   - Create GIN indexes on `content` and `resolved_content` JSONB columns
   - Optimize queries: item code lookups, symbol searches, sheet name filters
   - Target: < 50ms for exact matches, < 100ms for JSONB queries

4. **Story 4.4: Vector Embedding and Indexing**
   - Convert records to text representation (flatten content + resolved + source)
   - Batch embed using Azure OpenAI text-embedding-3-large (100 texts/call)
   - Store vectors in Milvus collection with metadata
   - Target: ≤ 30s per 1000 records

5. **Story 4.5: Incremental Indexing**
   - Add new documents without re-indexing existing ones
   - Delete documents cleanly (CASCADE + Milvus filter)
   - Re-process documents atomically (transactional)

**Out of Scope:**

- **Retrieval logic** (Epic 5, Story 5.2): Query routing, retrieval strategies
- **Answer generation** (Epic 5, Story 5.3): LLM answer synthesis
- **Symbol resolution** (Epic 3, Stories 3.5-3.6): Already handled by extraction pipeline
- **Cross-reference resolution** (Epic 3, Story 3.8): Future enhancement
- **Query endpoint** (Epic 5, Story 5.1): API endpoint for Q&A
- **Performance optimization** beyond targets: Query caching, materialized views (future)

---

## Implementation Details

### Source Tree Changes

**CREATE - Core Indexing Modules:**
```
src/knowledge/structured_store.py    # Story 4.1: PostgreSQL operations
src/knowledge/embeddings.py          # Story 4.4: Text → vector conversion
src/knowledge/indexer.py             # Story 4.3, 4.5: Index management orchestrator
```

**MODIFY - Existing Files:**
```
src/knowledge/vector_store.py        # Story 4.4: Complete Milvus implementation (currently stub)
src/workers/tasks.py                 # Add indexing task after extraction completes
src/db/models.py                     # Add helper methods for record queries (if needed)
```

**CREATE - Tests:**
```
tests/knowledge/test_structured_store.py   # Story 4.1 unit tests
tests/knowledge/test_embeddings.py         # Story 4.4 unit tests
tests/knowledge/test_vector_store.py       # Story 4.4 unit tests
tests/knowledge/test_indexer.py            # Integration tests
tests/knowledge/test_incremental.py        # Story 4.5 tests
```

### Technical Approach

**Story 4.1: Structured Store (PostgreSQL)**

**Goal:** Persist `ExtractedRecord` objects to PostgreSQL with optimized JSONB storage.

**Approach:**
1. **Batch Insert Optimization:**
   - Use SQLAlchemy async bulk insert: `session.execute(insert(ExtractedRecords), records_batch)`
   - Batch size: 500 records per transaction
   - Use `RETURNING id` to capture inserted record IDs for Milvus linkage

2. **JSONB Storage Strategy:**
   ```python
   # extracted_records table schema (already exists from Story 1.2)
   class ExtractedRecords(Base):
       id: UUID
       document_id: UUID (FK to documents)
       sheet_name: str
       row_number: int
       col_range: str            # "A:F"
       content: JSONB             # Original extracted content
       resolved_content: JSONB    # After symbol resolution
       headers: JSONB             # Column headers array
       created_at: timestamp
   ```

3. **Data Transformation:**
   ```python
   # Convert extraction.models.ExtractedRecord → db.models.ExtractedRecords
   def to_db_model(extracted_record: ExtractedRecord, document_id: UUID) -> dict:
       return {
           "document_id": document_id,
           "sheet_name": extracted_record._source["sheet"],
           "row_number": extracted_record._source["row"],
           "col_range": extracted_record._source["col_range"],
           "content": extracted_record.content,                    # dict → JSONB
           "resolved_content": extracted_record.resolved_content,  # dict → JSONB (if exists)
           "headers": extracted_record.headers,                    # list → JSONB
       }
   ```

4. **Error Handling:**
   - Wrap in transaction: rollback on any error
   - Retry on transient DB errors (connection loss)
   - Log failed records with document_id and row_number

**Module:** `src/knowledge/structured_store.py`

```python
class StructuredStore:
    """Manages PostgreSQL storage for extracted records."""

    def __init__(self, db_session: AsyncSession):
        self.db = db_session
        self.logger = structlog.get_logger()

    async def store_records(
        self,
        records: list[ExtractedRecord],
        document_id: UUID
    ) -> int:
        """Store extracted records in PostgreSQL.

        Args:
            records: List of ExtractedRecord from extraction pipeline
            document_id: UUID of source document

        Returns:
            Count of successfully stored records

        Raises:
            DSOLError: If storage fails
        """
        # Implementation in Story 4.1
```

---

**Story 4.3: Exact Lookup Indexing**

**Goal:** Create database indexes for fast exact lookups (item codes, symbols, sheet names).

**Approach:**
1. **GIN Indexes on JSONB Columns:**
   ```sql
   -- Already created in Alembic migration from Story 1.2
   CREATE INDEX idx_records_content ON extracted_records USING GIN (content);
   CREATE INDEX idx_records_resolved ON extracted_records USING GIN (resolved_content);
   ```

2. **B-tree Indexes on Common Filters:**
   ```sql
   CREATE INDEX idx_records_document ON extracted_records(document_id);
   CREATE INDEX idx_records_sheet ON extracted_records(sheet_name);
   CREATE INDEX idx_records_row ON extracted_records(row_number);
   ```

3. **Query Patterns to Optimize:**
   ```python
   # Exact item code lookup
   SELECT * FROM extracted_records
   WHERE content->>'Item Code' = '2024-4_0019'
   # Uses GIN index, < 50ms

   # Find all records with specific symbol
   SELECT * FROM extracted_records
   WHERE content @> '{"Screen": "○"}'
   # Uses GIN index, < 100ms

   # Find all records from sheet
   SELECT * FROM extracted_records
   WHERE document_id = 'uuid' AND sheet_name = 'Sheet1'
   # Uses composite index, < 20ms
   ```

4. **Index Management:**
   - Indexes created during Alembic migration (Story 1.2)
   - Monitor index usage: `pg_stat_user_indexes`
   - Consider additional indexes based on query patterns (future optimization)

**Module:** `src/knowledge/indexer.py` (orchestrates indexing, doesn't create indexes directly)

---

**Story 4.4: Vector Embedding and Indexing**

**Goal:** Embed records as vectors and store in Milvus for semantic similarity search.

**Approach:**

1. **Text Representation Generation:**
   ```python
   def record_to_text(record: ExtractedRecord) -> str:
       """Convert ExtractedRecord to text for embedding.

       Format:
       [Headers] Key1: Value1. Key2: Value2. ...
       [Source] From: filename, Sheet: sheet_name, Row: row_number
       [Resolved] Key1_resolved: Resolved1 (if symbols resolved)
       """
       content_parts = []

       # Add content fields
       for key, value in record.content.items():
           if value is not None:
               content_parts.append(f"{key}: {value}")

       # Add resolved fields (symbol meanings)
       if record.resolved_content:
           for key, value in record.resolved_content.items():
               if "_resolved" in key and value:
                   content_parts.append(f"{key}: {value}")

       # Add source context
       source = record._source
       source_text = f"From: {source['filename']}, Sheet: {source['sheet']}, Row: {source['row']}"

       return f"{'. '.join(content_parts)}. {source_text}"
   ```

2. **Batch Embedding with Azure OpenAI:**
   ```python
   async def embed_batch(texts: list[str], client: AzureOpenAI) -> list[list[float]]:
       """Embed batch of texts using Azure OpenAI.

       Args:
           texts: List of text representations (max 100)
           client: Azure OpenAI client

       Returns:
           List of embedding vectors (3072 dimensions each)
       """
       response = await client.embeddings.create(
           model="text-embedding-3-large",  # Azure deployment name
           input=texts[:100],  # Max 100 texts per call
           dimensions=3072
       )
       return [item.embedding for item in response.data]
   ```

3. **Milvus Collection Setup:**
   ```python
   # Collection configuration (created once)
   # Unified schema supporting both Excel and Word documents
   collection_config = {
       "name": "dsol_records",
       "vectors": {
           "size": 3072,
           "distance": "Cosine"
       },
       "optimizers_config": {
           "indexing_threshold": 10000  # Index after 10k vectors
       }
   }
   ```

4. **Milvus Payload Structure (Unified for Excel and Word):**
   ```python
   # For Excel documents
   payload = {
       "record_id": str(record_id),           # Link to PostgreSQL
       "document_id": str(document_id),
       "filename": source["filename"],
       "document_type": "excel",               # NEW: Document type
       "text_content": text_representation,   # For display in results
       "metadata": {                           # NEW: Type-specific metadata
           "sheet_name": source["sheet"],
           "row_number": source["row"],
           "has_symbols": bool(record.resolved_content)
       }
   }
   
   # For Word documents
   payload = {
       "record_id": str(record_id),
       "document_id": str(document_id),
       "filename": filename,
       "document_type": "word",                # NEW: Document type
       "text_content": text_representation,
       "metadata": {                           # NEW: Type-specific metadata
           "page_number": chunk_metadata.get("page_number"),
           "contains_table": chunk_metadata.get("contains_table", False)
       }
   }
   ```

5. **Batch Upload to Milvus:**
   ```python
   # Upload in batches of 100 vectors
   points = [
       PointStruct(
           id=str(uuid.uuid4()),           # Milvus point ID
           vector=embedding,
           payload=payload
       )
       for embedding, payload in zip(embeddings, payloads)
   ]

   await milvus_client.upsert(
       collection_name="dsol_records",
       points=points,
       wait=True  # Wait for indexing
   )
   ```

6. **Performance Targets:**
   - Batch size: 100 records per embedding call
   - Embedding latency: ~2-3s per 100 records
   - Milvus upload: ~500ms per 100 vectors
   - **Total: ≤ 30s per 1000 records** (within target)

**Modules:**
- `src/knowledge/embeddings.py`: Text representation + Azure OpenAI embedding
- `src/knowledge/vector_store.py`: Milvus operations (collection management, upsert, delete)

---

**Story 4.5: Incremental Indexing**

**Goal:** Support adding, deleting, and re-processing documents without full rebuild.

**Approach:**

1. **Add New Document:**
   ```python
   async def index_document(document_id: UUID, records: list[ExtractedRecord]):
       """Index a new document's records.

       Steps:
       1. Insert records into PostgreSQL (Story 4.1)
       2. Embed and upload to Milvus (Story 4.4)
       3. Update document status to 'completed'
       """
       # PostgreSQL: Standard batch insert
       await structured_store.store_records(records, document_id)

       # Milvus: Batch embed + upload
       await vector_store.index_records(records, document_id)

       # No impact on existing documents ✓
   ```

2. **Delete Document:**
   ```python
   async def delete_document(document_id: UUID):
       """Remove all data for a document.

       Steps:
       1. Delete from PostgreSQL (CASCADE handles extracted_records)
       2. Delete from Milvus (filter by document_id)
       """
       # PostgreSQL: CASCADE delete
       await db.execute(
           delete(Documents).where(Documents.id == document_id)
       )
       # Deletes extracted_records automatically via FK CASCADE

       # Milvus: Filter delete
       await milvus_client.delete(
           collection_name="dsol_records",
           points_selector=FilterSelector(
               filter=Filter(
                   must=[
                       FieldCondition(
                           key="document_id",
                           match=MatchValue(value=str(document_id))
                       )
                   ]
               )
           )
       )
   ```

3. **Re-process Document:**
   ```python
   async def reindex_document(document_id: UUID, new_records: list[ExtractedRecord]):
       """Re-index a document (e.g., after symbol dictionary update).

       Steps:
       1. Delete old data (transactional)
       2. Insert new data
       3. Atomic operation - no partial state
       """
       async with db.begin():  # Transaction
           await delete_document(document_id)
           await index_document(document_id, new_records)
       # Rollback on any error ✓
   ```

4. **Idempotency:**
   - Safe to call `index_document()` multiple times with same document_id
   - Use `ON CONFLICT` for PostgreSQL: `ON CONFLICT (document_id, sheet_name, row_number) DO UPDATE`
   - Use Milvus `upsert()` (not insert) to replace existing points

---

### Existing Patterns to Follow

**From Extraction Pipeline (Stories 3.1-3.4):**

1. **Structured Logging with Context:**
   ```python
   logger.info(
       "records_indexed",
       document_id=doc_id,
       filename=filename,
       record_count=len(records),
       postgres_duration_ms=pg_duration,
       milvus_duration_ms=milvus_duration,
       total_duration_ms=total_duration
   )
   ```

2. **Error Handling Pattern:**
   ```python
   from src.core.exceptions import DSOLError

   class IndexingError(DSOLError):
       """Raised when indexing fails."""
       pass

   try:
       await index_records(records, document_id)
   except Exception as e:
       logger.error(
           "indexing_failed",
           document_id=doc_id,
           error=str(e),
           error_type=type(e).__name__
       )
       raise IndexingError(
           message=f"Failed to index document {doc_id}",
           code="INDEXING_FAILED",
           details={"document_id": str(doc_id), "error": str(e)}
       ) from e
   ```

3. **Type Annotations (mypy strict):**
   ```python
   from typing import Optional
   from uuid import UUID
   from src.extraction.models import ExtractedRecord

   async def store_records(
       self,
       records: list[ExtractedRecord],
       document_id: UUID
   ) -> int:
       """Type hints for all parameters and return values."""
   ```

4. **Data Class Pattern:**
   ```python
   from dataclasses import dataclass

   @dataclass
   class IndexingResult:
       """Result of indexing operation."""
       document_id: UUID
       records_stored: int
       postgres_duration_ms: int
       milvus_duration_ms: int
       total_duration_ms: int
   ```

### Integration Points

**1. Extraction Pipeline (Stories 3.1-3.4) → Indexing:**
```
RecordBuilder.build_records()
  → List[ExtractedRecord]
    → Indexer.index_document(document_id, records)
      → StructuredStore.store_records() + VectorStore.index_records()
```

**2. Document Upload (Story 2.2) → Celery Task → Indexing:**
```
POST /documents
  → Celery task: process_document_task(document_id)
    → ExtractionPipeline.process(document_id)
      → Indexer.index_document(document_id, extracted_records)
        → Update document status: "completed"
```

**3. Document Deletion (Story 2.6) → Indexing:**
```
DELETE /documents/{id}
  → DocumentService.delete_document(document_id)
    → Indexer.delete_document(document_id)
      → PostgreSQL CASCADE + Milvus filter delete
```

**4. Indexing → Future Q&A (Epic 5):**
```
Indexing creates knowledge base
  → Retrieval (Story 5.2) queries PostgreSQL + Milvus
    → Answer Generator (Story 5.3) synthesizes answer with sources
```

---

## Development Context

### Relevant Existing Code

**Extraction Models (`src/extraction/models.py:151-215`):**
```python
@dataclass
class ExtractedRecord:
    """Structured record from table extraction."""
    content: dict                        # Column → value mapping
    headers: list[str]                   # Column headers
    _source: dict                        # {document_id, filename, sheet, table_id, row, col_range}
    _merge_metadata: Optional[dict]      # Optional: merged cell tracking
    resolved_content: Optional[dict]     # Optional: after symbol resolution

    def to_dict(self) -> dict:
        """Convert to JSON-serializable dict."""
```

**Database Models (`src/db/models.py`):**
```python
class ExtractedRecords(Base):
    __tablename__ = "extracted_records"
    id = Column(UUID, primary_key=True, default=uuid.uuid4)
    document_id = Column(UUID, ForeignKey("documents.id", ondelete="CASCADE"))
    sheet_name = Column(String(255), nullable=False)
    row_number = Column(Integer, nullable=False)
    col_range = Column(String(50))
    content = Column(JSONB, nullable=False)         # Use this for storage
    resolved_content = Column(JSONB)
    headers = Column(JSONB)
    created_at = Column(DateTime, default=func.now())

    __table_args__ = (
        UniqueConstraint("document_id", "sheet_name", "row_number", name="unique_record"),
    )
```

**Celery Worker Pattern (`src/workers/tasks.py`):**
```python
@celery_app.task(bind=True, max_retries=3)
def process_document_task(self, document_id: str):
    """Background task for document processing."""
    try:
        # 1. Extract records (Stories 3.1-3.4)
        # 2. Index records (Epic 4 - THIS SPEC) ← Add here
        # 3. Update document status
        pass
    except Exception as e:
        logger.error("task_failed", task_id=self.request.id, error=str(e))
        self.retry(countdown=60, exc=e)
```

**Configuration (`src/core/config.py`):**
```python
class Settings(BaseSettings):
    # Database
    database_url: str

    # Milvus
    milvus_url: str = "http://localhost:6333"
    milvus_collection: str = "dsol_records"

    # Azure OpenAI
    azure_openai_api_key: str
    azure_openai_endpoint: str
    azure_openai_api_version: str = "2024-02-15-preview"
    azure_openai_embedding_deployment: str  # Deployment name for text-embedding-3-large

    class Config:
        env_file = ".env"
```

### Framework Dependencies

**Core Dependencies (from `pyproject.toml`):**
- **sqlalchemy[asyncio]==2.0+** - Async ORM for PostgreSQL
- **asyncpg==0.29+** - Async PostgreSQL driver
- **milvus-client==1.16+** - Milvus vector store client
- **openai==1.0+** - Azure OpenAI SDK (embeddings)
- **celery==5.3+** - Async task processing
- **structlog==24.0** - Structured logging
- **pydantic==2.0+** - Data validation

**Key APIs:**
```python
# SQLAlchemy async
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, insert, delete

# Milvus
from milvus_client import AsyncMilvusClient
from milvus_client.models import PointStruct, Filter, FieldCondition, MatchValue

# Azure OpenAI
from openai import AsyncAzureOpenAI

# Structlog
import structlog
logger = structlog.get_logger()
```

### Internal Dependencies

**Modules Used:**
- `src.extraction.models.ExtractedRecord` - Input data structure
- `src.db.models.ExtractedRecords` - ORM model for storage
- `src.db.session.get_db_session()` - Async session factory
- `src.core.config.get_settings()` - Configuration access
- `src.core.exceptions.DSOLError` - Base exception class
- `src.core.logging` - Structured logging setup

**Import Example:**
```python
from uuid import UUID
from src.extraction.models import ExtractedRecord
from src.db.models import ExtractedRecords
from src.db.session import get_db_session
from src.core.config import get_settings
from src.core.exceptions import DSOLError
import structlog
```

### Configuration Changes

**Environment Variables (`.env`):**
```bash
# No new variables needed - already defined in architecture.md
# Just ensure these are set:

# Milvus
MILVUS_HOST=http://localhost:6333
MILVUS_COLLECTION=dsol_records

# Azure OpenAI
AZURE_OPENAI_API_KEY=your-key-here
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_API_VERSION=2024-02-15-preview
AZURE_OPENAI_EMBEDDING_DEPLOYMENT=text-embedding-3-large  # Your deployment name
```

**Database (Alembic Migration - Already Exists from Story 1.2):**
- No new migration needed
- Indexes already created: `idx_records_content`, `idx_records_resolved`, `idx_records_document`, `idx_records_sheet`
- Schema from architecture.md is complete

**Milvus Collection (Create on First Run):**
```python
# Create collection if not exists (in indexer.py initialization)
try:
    await milvus_client.get_collection("dsol_records")
except Exception:
    await milvus_client.create_collection(
        collection_name="dsol_records",
        vectors_config=VectorParams(size=3072, distance=Distance.COSINE)
    )
```

### Existing Conventions (Brownfield)

**Code Style (from `pyproject.toml`):**
- Black formatter: 100-character line length
- Ruff linter: pycodestyle, pyflakes, isort, flake8-bugbear
- mypy strict mode: Full type annotations required
- Python 3.11 type hints

**Naming Conventions (from `architecture.md`):**
- Modules: snake_case (`structured_store.py`)
- Classes: PascalCase (`StructuredStore`)
- Functions: snake_case (`store_records()`)
- Constants: UPPER_SNAKE (`MAX_BATCH_SIZE = 500`)
- Private methods: _prefix (`_convert_to_db_model()`)

**Async Patterns:**
```python
# All I/O operations are async
async def store_records(self, records: list[ExtractedRecord]) -> int:
    async with self.db.begin():  # Transaction
        result = await self.db.execute(...)
        return result.rowcount
```

**Error Handling:**
```python
# Use DSOLError hierarchy
class IndexingError(DSOLError):
    """Raised when indexing fails."""
    pass

# Log before raising
logger.error("operation_failed", context=...)
raise IndexingError(message=..., code=..., details=...)
```

**Logging Style:**
```python
# Structured logging with context
logger.info(
    "event_name",
    document_id=doc_id,
    metric_value=123,
    duration_ms=duration
)
```

### Test Framework & Standards

**From `pyproject.toml` and existing tests:**
- **pytest 8.0+** with pytest-asyncio 0.23+
- Test file naming: `test_*.py`
- Test function naming: `test_*`
- Async mode: auto (pytest-asyncio)
- Coverage target: Aim for 80%+ coverage of new code

**Test Organization:**
```
tests/knowledge/
├── __init__.py
├── conftest.py                        # Fixtures (db session, milvus client, etc.)
├── test_structured_store.py           # Story 4.1 unit tests
├── test_embeddings.py                 # Story 4.4 unit tests
├── test_vector_store.py               # Story 4.4 unit tests
├── test_indexer.py                    # Integration tests (all stories)
└── test_incremental.py                # Story 4.5 tests
```

**Test Patterns (from `tests/extraction/`):**
```python
import pytest
from uuid import uuid4
from src.extraction.models import ExtractedRecord
from src.knowledge.structured_store import StructuredStore

@pytest.mark.asyncio
async def test_store_records_success(db_session):
    """Test storing records in PostgreSQL."""
    # Arrange
    store = StructuredStore(db_session)
    records = [
        ExtractedRecord(
            content={"Item Code": "2024-4_0019", "Screen": "○"},
            headers=["Item Code", "Screen"],
            _source={
                "document_id": str(uuid4()),
                "filename": "test.xlsx",
                "sheet": "Sheet1",
                "table_id": 1,
                "row": 5,
                "col_range": "A:B"
            }
        )
    ]

    # Act
    count = await store.store_records(records, uuid4())

    # Assert
    assert count == 1
```

---

## Implementation Stack

**Runtime:**
- Python 3.11+ (required by project)

**Database:**
- PostgreSQL 18 (structured store, JSONB)
  - SQLAlchemy 2.0 (async ORM)
  - asyncpg 0.29+ (async driver)
  - Alembic 1.13 (migrations)

**Vector Store:**
- Milvus 1.16 (vector similarity search)
  - milvus-client 1.16+ (Python SDK)
  - Cosine distance metric
  - 1536 dimensions (text-embedding-3-large)

**LLM Integration:**
- Azure OpenAI (embeddings)
  - openai SDK 1.0+ (Azure support)
  - text-embedding-3-large model (3072 dims)
  - Batch API: up to 100 texts per call

**Task Queue:**
- Celery 5.3+ (async processing)
- Redis 7+ (broker)

**Utilities:**
- structlog 24.0 (structured logging)
- Pydantic 2.0+ (validation)
- httpx 0.27+ (async HTTP)

**Development:**
- pytest 8.0+ (testing)
- pytest-asyncio 0.23+ (async tests)
- Black 24.0 (formatting)
- Ruff 0.5+ (linting)
- mypy 1.10+ (type checking)

---

## Technical Details

### Batch Processing Strategy

**Why Batch Processing:**
- Reduce API calls to Azure OpenAI (cost + latency)
- Optimize PostgreSQL inserts (transaction overhead)
- Improve Milvus upload performance (bulk indexing)

**Batch Sizes:**
- **PostgreSQL:** 500 records per transaction
  - Large enough for efficiency, small enough to avoid long locks
  - Typical document: 1000-5000 records → 2-10 transactions

- **Azure OpenAI Embeddings:** 100 texts per API call
  - API limit: 100 inputs per request
  - Reduces API latency: 10 calls vs 1000 calls for 1000 records

- **Milvus Upload:** 100 vectors per upsert
  - Optimal batch size for Milvus indexing
  - Parallel uploads: Use asyncio.gather for multiple batches

**Example Flow for 1000 Records:**
```
1000 records → 2 PostgreSQL transactions (500 each)        [~2-3s]
             → 10 embedding API calls (100 each)           [~20-30s]
             → 10 Milvus upserts (100 each)                [~5s]
                                                   Total: ~30s ✓ (within target)
```

### Embedding Text Representation Format

**Design Principles:**
1. **Include all searchable content** - field names + values
2. **Include resolved symbols** - semantic meaning, not just "○"
3. **Include source context** - helps LLM understand provenance
4. **Keep concise** - avoid overwhelming the embedding model
5. **Consistent format** - enables better semantic matching

**Format:**
```
{Field1}: {Value1}. {Field2}: {Value2}. ... {Field1_resolved}: {Resolved1}. From: {filename}, Sheet: {sheet_name}, Row: {row_number}
```

**Example:**
```python
# Input ExtractedRecord:
{
    "content": {
        "Item Code": "2024-4_0019",
        "Report": null,
        "Screen": "○"
    },
    "resolved_content": {
        "Screen": "○",
        "Screen_resolved": "Applicable/Yes"
    },
    "_source": {
        "filename": "distribution_spec.xlsx",
        "sheet": "Item Mapping",
        "row": 42,
        ...
    }
}

# Output text for embedding:
"Item Code: 2024-4_0019. Screen: ○. Screen_resolved: Applicable/Yes. From: distribution_spec.xlsx, Sheet: Item Mapping, Row: 42"
```

**Why This Format Works:**
- **Semantic richness:** Includes both raw symbols (○) and meanings (Applicable/Yes)
- **Context-aware:** LLM can understand "this is about item 2024-4_0019 from distribution spec"
- **Queryable:** User query "what items are applicable for screen?" matches "Applicable/Yes"
- **Concise:** ~100-200 chars per record, well within embedding model limits

### PostgreSQL JSONB Query Patterns

**Pattern 1: Exact Value Match**
```sql
-- Find record with specific item code
SELECT * FROM extracted_records
WHERE content->>'Item Code' = '2024-4_0019';
-- Uses GIN index, < 50ms
```

**Pattern 2: JSONB Containment**
```sql
-- Find all records with ○ in Screen column
SELECT * FROM extracted_records
WHERE content @> '{"Screen": "○"}';
-- Uses GIN index, < 100ms
```

**Pattern 3: Resolved Symbol Search**
```sql
-- Find all "applicable" items via resolved content
SELECT * FROM extracted_records
WHERE resolved_content->>'Screen_resolved' LIKE '%Applicable%';
-- Uses GIN index, < 200ms
```

**Pattern 4: Sheet Filter + Content Match**
```sql
-- Find item in specific sheet
SELECT * FROM extracted_records
WHERE document_id = 'uuid'
  AND sheet_name = 'Item Mapping'
  AND content->>'Item Code' = '2024-4_0019';
-- Uses composite index, < 20ms
```

### Milvus Search Strategy

**Metadata Filtering Before Vector Search:**
```python
# Strategy: Filter → Search (more efficient than Search → Filter)
search_result = await milvus_client.search(
    collection_name="dsol_records",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="document_id", match=MatchValue(value=str(doc_id))),
            FieldCondition(key="sheet_name", match=MatchValue(value="Item Mapping"))
        ]
    ),
    limit=10,
    score_threshold=0.7  # Only return matches with similarity > 0.7
)
```

**Why Filter First:**
- Reduces search space (e.g., 10M vectors → 10K vectors for specific document)
- Faster search: fewer vectors to compare
- More relevant results: already scoped to user's documents

### Error Handling & Retry Strategy

**Transient Errors (Retry):**
- PostgreSQL connection loss → Retry 3x with exponential backoff (1s, 2s, 4s)
- Azure OpenAI rate limit (429) → Retry 3x with backoff
- Milvus connection timeout → Retry 3x

**Permanent Errors (Fail Fast):**
- Invalid record data (e.g., missing required field) → Log + skip record
- PostgreSQL unique constraint violation → Log + skip (already indexed)
- Azure OpenAI authentication error → Fail entire batch

**Transaction Rollback:**
```python
async with db.begin():  # Transaction
    try:
        await store_records_batch(records_batch)
        await commit_metadata(batch_id)
    except Exception:
        # Automatic rollback on exception
        raise
```

### Performance Targets & Metrics

**From NFRs (architecture.md):**
- **Document Processing:** ≤ 60s per Excel file (includes extraction + indexing)
- **Indexing Target:** ≤ 30s per 1000 records (derived from 60s total - 30s extraction)

**Breakdown for 1000 Records:**
- PostgreSQL storage: ~2-3s (batch inserts)
- Azure OpenAI embedding: ~20-30s (10 API calls × 2-3s each)
- Milvus upload: ~5s (bulk upsert)
- **Total: ~30s** ✓

**Query Performance Targets (for Epic 5):**
- Exact lookup (PostgreSQL): < 50ms
- JSONB query (PostgreSQL): < 100ms
- Semantic search (Milvus): < 200ms
- Full Q&A pipeline: < 5s (includes retrieval + LLM generation)

**Monitoring Metrics:**
```python
logger.info(
    "indexing_complete",
    document_id=doc_id,
    record_count=count,
    postgres_duration_ms=pg_ms,      # Track separately
    embedding_duration_ms=embed_ms,  # Track separately
    milvus_duration_ms=milvus_ms,    # Track separately
    total_duration_ms=total_ms,      # Overall
    records_per_second=count / (total_ms / 1000)
)
```

### Idempotency & Atomicity

**Idempotent Operations:**
1. **PostgreSQL:** Use `ON CONFLICT` clause
   ```sql
   INSERT INTO extracted_records (document_id, sheet_name, row_number, ...)
   VALUES (...)
   ON CONFLICT (document_id, sheet_name, row_number)
   DO UPDATE SET content = EXCLUDED.content, updated_at = NOW();
   ```

2. **Milvus:** Use `upsert()` instead of `insert()`
   - If point ID exists → update vector and payload
   - If point ID doesn't exist → insert new point
   - Safe to call multiple times

**Atomic Re-indexing:**
```python
async def reindex_document(document_id: UUID, new_records: list[ExtractedRecord]):
    """Atomically replace document's indexed data."""
    async with db.begin():  # Single transaction
        # Delete old data
        await db.execute(delete(ExtractedRecords).where(ExtractedRecords.document_id == document_id))
        await milvus_client.delete(collection_name="dsol_records",
                                    points_selector=Filter(...))

        # Insert new data
        await structured_store.store_records(new_records, document_id)
        await vector_store.index_records(new_records, document_id)

        # Commit or rollback together
```

### Security Considerations

**Data Isolation:**
- Records are scoped by `api_key_id` (FK from documents table)
- Retrieval queries MUST filter by authenticated user's documents
- No cross-tenant data leakage

**Sensitive Content:**
- Do NOT log document content (only metadata: document_id, filename, record_count)
- Do NOT log query answers (only metadata: query_text, confidence, processing_time)
- Embeddings are stored in Milvus - ensure Milvus is not publicly accessible

**API Key Security:**
- Azure OpenAI API key stored in environment variables (not in code)
- Use Azure Key Vault in production (future enhancement)

---

## Development Setup

**Prerequisites:**
- Python 3.11+ installed
- Docker & Docker Compose installed
- Poetry installed (dependency management)
- DSOL repository cloned (Stories 1-3 already implemented)

**Setup Steps:**

```bash
# 1. Navigate to project
cd /path/to/DSOL

# 2. Install dependencies (if not already)
poetry install

# 3. Start infrastructure (PostgreSQL, Redis, Milvus)
docker-compose up -d postgres redis milvus

# 4. Verify Milvus is running
curl http://localhost:6333/collections
# Should return: {"result": {"collections": []}, "status": "ok"}

# 5. Run database migrations (if not already)
poetry run alembic upgrade head

# 6. Set environment variables
cp .env.example .env
# Edit .env and add:
# - AZURE_OPENAI_API_KEY=<your-key>
# - AZURE_OPENAI_ENDPOINT=<your-endpoint>
# - AZURE_OPENAI_EMBEDDING_DEPLOYMENT=<deployment-name>

# 7. Start API server (development mode)
poetry run uvicorn src.api.main:app --reload --port 8000

# 8. Start Celery worker (separate terminal)
poetry run celery -A src.workers.celery_app worker --loglevel=info

# 9. Run tests
poetry run pytest tests/knowledge/ -v
```

**Docker Compose Services:**
```yaml
# Already exists in docker-compose.yml
services:
  postgres:
    image: postgres:18
    ports: ["5432:5432"]
    environment:
      POSTGRES_DB: dsol
      POSTGRES_USER: dsol
      POSTGRES_PASSWORD: dsol

  redis:
    image: redis:7
    ports: ["6379:6379"]

  milvus:
    image: milvus/milvus:v1.16.0
    ports: ["6333:6333"]
    volumes:
      - milvus_storage:/milvus/storage
```

**Verify Setup:**
```bash
# Test PostgreSQL connection
poetry run python -c "from src.db.session import get_db_session; import asyncio; asyncio.run(get_db_session().__anext__())"

# Test Milvus connection
poetry run python -c "from milvus_client import MilvusClient; client = MilvusClient('localhost', port=6333); print(client.get_collections())"

# Test Azure OpenAI connection
poetry run python -c "from openai import AsyncAzureOpenAI; import asyncio; import os; client = AsyncAzureOpenAI(api_key=os.getenv('AZURE_OPENAI_API_KEY'), api_version='2024-02-15-preview', azure_endpoint=os.getenv('AZURE_OPENAI_ENDPOINT')); asyncio.run(client.embeddings.create(model=os.getenv('AZURE_OPENAI_EMBEDDING_DEPLOYMENT'), input=['test']))"
```

---

## Implementation Guide

### Setup Steps

**Step 1: Create Module Structure**
```bash
# Create knowledge module files
touch src/knowledge/structured_store.py
touch src/knowledge/embeddings.py
touch src/knowledge/indexer.py
# vector_store.py already exists (stub)

# Create test files
touch tests/knowledge/conftest.py
touch tests/knowledge/test_structured_store.py
touch tests/knowledge/test_embeddings.py
touch tests/knowledge/test_vector_store.py
touch tests/knowledge/test_indexer.py
touch tests/knowledge/test_incremental.py
```

**Step 2: Verify Database Schema**
```sql
-- Verify extracted_records table exists with correct schema
\d extracted_records

-- Verify indexes exist
\di extracted_records
-- Should see: idx_records_content, idx_records_resolved, idx_records_document, idx_records_sheet
```

**Step 3: Initialize Milvus Collection**
```python
# Run once to create collection (add to src/knowledge/vector_store.py __init__)
from milvus_client import AsyncMilvusClient
from milvus_client.models import Distance, VectorParams

client = AsyncMilvusClient(url="http://localhost:6333")

# Create collection if not exists
try:
    await client.get_collection("dsol_records")
except:
    await client.create_collection(
        collection_name="dsol_records",
        vectors_config=VectorParams(size=3072, distance=Distance.COSINE)
    )
```

### Implementation Steps

**Phase 1: Story 4.1 - Structured Store (PostgreSQL)**

1. **Create `src/knowledge/structured_store.py`:**
   ```python
   class StructuredStore:
       def __init__(self, db_session: AsyncSession):
           self.db = db_session
           self.logger = structlog.get_logger()

       async def store_records(
           self,
           records: list[ExtractedRecord],
           document_id: UUID
       ) -> int:
           """Store extracted records in PostgreSQL."""
           # TODO: Implement batch insert (500 records per transaction)
           # TODO: Convert ExtractedRecord → db model dict
           # TODO: Use SQLAlchemy bulk insert
           # TODO: Return count of stored records
   ```

2. **Write Unit Tests:**
   - Test successful batch insert
   - Test duplicate handling (ON CONFLICT)
   - Test transaction rollback on error
   - Test JSONB storage (content, resolved_content, headers)

3. **Integration Test:**
   - Store 1000 mock records
   - Verify all records in database
   - Query by item code, symbol, sheet name
   - Measure performance (< 3s for 1000 records)

**Phase 2: Story 4.4 - Vector Embedding & Indexing**

1. **Create `src/knowledge/embeddings.py`:**
   ```python
   class EmbeddingService:
       def __init__(self, azure_client: AsyncAzureOpenAI):
           self.client = azure_client
           self.deployment = get_settings().azure_openai_embedding_deployment

       def record_to_text(self, record: ExtractedRecord) -> str:
           """Convert record to text representation."""
           # TODO: Implement format from Technical Details section

       async def embed_batch(self, texts: list[str]) -> list[list[float]]:
           """Embed batch of texts (max 100)."""
           # TODO: Call Azure OpenAI embeddings API
           # TODO: Return list of 3072-dim vectors
   ```

2. **Implement `src/knowledge/vector_store.py`:**
   ```python
   class VectorStore:
       def __init__(self, milvus_client: AsyncMilvusClient):
           self.client = milvus_client
           self.collection = "dsol_records"

       async def index_records(
           self,
           records: list[ExtractedRecord],
           embeddings: list[list[float]],
           document_id: UUID
       ) -> int:
           """Upload vectors to Milvus."""
           # TODO: Build PointStruct objects with vectors + metadata
           # TODO: Batch upsert (100 points per call)
           # TODO: Return count of indexed records
   ```

3. **Write Unit Tests:**
   - Test text representation format
   - Test Azure OpenAI embedding call (mock API)
   - Test Milvus upsert
   - Test batch processing (split into chunks of 100)

4. **Integration Test:**
   - End-to-end: ExtractedRecord → embedding → Milvus
   - Verify vectors stored with correct metadata
   - Test similarity search (query vector → retrieve similar records)

**Phase 3: Story 4.3 & 4.5 - Indexer Orchestrator**

1. **Create `src/knowledge/indexer.py`:**
   ```python
   class Indexer:
       def __init__(
           self,
           structured_store: StructuredStore,
           embedding_service: EmbeddingService,
           vector_store: VectorStore
       ):
           self.structured_store = structured_store
           self.embedding_service = embedding_service
           self.vector_store = vector_store

       async def index_document(
           self,
           document_id: UUID,
           records: list[ExtractedRecord]
       ) -> IndexingResult:
           """Index document in both PostgreSQL and Milvus."""
           # TODO: 1. Store in PostgreSQL (Story 4.1)
           # TODO: 2. Embed records (Story 4.4)
           # TODO: 3. Index in Milvus (Story 4.4)
           # TODO: Return IndexingResult with metrics

       async def delete_document(self, document_id: UUID) -> None:
           """Delete document from both stores."""
           # TODO: Delete from PostgreSQL (CASCADE)
           # TODO: Delete from Milvus (filter by document_id)

       async def reindex_document(
           self,
           document_id: UUID,
           new_records: list[ExtractedRecord]
       ) -> IndexingResult:
           """Atomically re-index document."""
           # TODO: Wrap in transaction
           # TODO: delete_document() then index_document()
   ```

2. **Integration Tests:**
   - Test full indexing flow (1000 records end-to-end)
   - Test delete document (verify PostgreSQL + Milvus cleanup)
   - Test re-index (verify atomic replace)
   - Test idempotency (index same document twice)
   - Performance test (measure 1000 records, target ≤ 30s)

**Phase 4: Integration with Celery Worker**

1. **Modify `src/workers/tasks.py`:**
   ```python
   @celery_app.task(bind=True, max_retries=3)
   def process_document_task(self, document_id: str):
       """Background task for document processing."""
       try:
           # 1. Run extraction pipeline (Stories 3.1-3.4) - Already exists
           extracted_records = extraction_pipeline.process(document_id)

           # 2. Index records (NEW - This Tech Spec)
           indexer = Indexer(...)
           result = await indexer.index_document(UUID(document_id), extracted_records)

           # 3. Update document status
           await update_document_status(document_id, "completed", record_count=result.records_stored)

       except Exception as e:
           logger.error("task_failed", task_id=self.request.id, error=str(e))
           await update_document_status(document_id, "failed", error=str(e))
           self.retry(countdown=60, exc=e)
   ```

2. **End-to-End Test:**
   - Upload Excel file via API (`POST /documents`)
   - Verify Celery task executes
   - Verify extraction → indexing → completion
   - Query PostgreSQL for records
   - Query Milvus for vectors
   - Check document status (`GET /documents/{id}`) shows "completed"

### Testing Strategy

**Unit Tests (Per Module):**
- `test_structured_store.py`: PostgreSQL operations, batch insert, error handling
- `test_embeddings.py`: Text representation, Azure OpenAI calls (mocked)
- `test_vector_store.py`: Milvus operations, batch upsert, filter delete
- `test_indexer.py`: Orchestration logic, transaction handling

**Integration Tests:**
- `test_indexer.py`: Full flow with real PostgreSQL + Milvus (use test database)
- `test_incremental.py`: Add/delete/reindex scenarios

**Performance Tests:**
- Benchmark 1000 records end-to-end (target: ≤ 30s)
- Measure PostgreSQL insert time (< 3s)
- Measure embedding time (< 30s)
- Measure Milvus upload time (< 5s)

**Test Data:**
- Use existing test Excel files from `tests/fixtures/sample_excel/`
- Mock `ExtractedRecord` objects for unit tests
- Small dataset (100 records) for fast unit tests
- Large dataset (1000 records) for performance tests

**Coverage Target:**
- Aim for 80%+ coverage of new code
- Critical paths: 100% coverage (error handling, transactions)

### Acceptance Criteria

**Story 4.1: Structured Store (PostgreSQL)**
- ✅ Records stored in `extracted_records` table with JSONB columns
- ✅ Batch insert handles 500 records per transaction
- ✅ Duplicate handling with ON CONFLICT
- ✅ Transaction rollback on error
- ✅ Performance: < 3s for 1000 records

**Story 4.2: Source Metadata Preservation**
- ✅ Every record has complete source metadata
- ✅ Can query by document_id, sheet_name, row_number
- ✅ Source metadata enables exact citation

**Story 4.3: Exact Lookup Indexing**
- ✅ GIN indexes on content, resolved_content (already exist from Story 1.2)
- ✅ Query performance: < 50ms exact match, < 100ms JSONB query
- ✅ Can query by item code, symbol, sheet name

**Story 4.4: Vector Embedding and Indexing**
- ✅ Records embedded using Azure OpenAI text-embedding-3-large
- ✅ Vectors stored in Milvus with metadata
- ✅ Batch processing: 100 texts per API call, 100 vectors per Milvus upsert
- ✅ Performance: ≤ 30s per 1000 records total

**Story 4.5: Incremental Indexing**
- ✅ Add new documents without re-indexing existing
- ✅ Delete documents cleanly (PostgreSQL CASCADE + Milvus filter)
- ✅ Re-index atomically (transaction-wrapped)
- ✅ Idempotent operations (safe to retry)

---

## Developer Resources

### File Paths Reference

**NEW Files (To Be Created):**
- `src/knowledge/structured_store.py` - PostgreSQL operations (Story 4.1)
- `src/knowledge/embeddings.py` - Text representation + Azure OpenAI embedding (Story 4.4)
- `src/knowledge/indexer.py` - Orchestrates indexing (Stories 4.3, 4.5)

**MODIFIED Files:**
- `src/knowledge/vector_store.py` - Complete Milvus implementation (currently stub)
- `src/workers/tasks.py` - Add indexing after extraction

**TEST Files (To Be Created):**
- `tests/knowledge/conftest.py` - Fixtures (db session, milvus client, mock Azure OpenAI)
- `tests/knowledge/test_structured_store.py` - Story 4.1 tests
- `tests/knowledge/test_embeddings.py` - Story 4.4 tests (text representation)
- `tests/knowledge/test_vector_store.py` - Story 4.4 tests (Milvus)
- `tests/knowledge/test_indexer.py` - Integration tests
- `tests/knowledge/test_incremental.py` - Story 4.5 tests

### Key Code Locations

**Extraction Models (Input to Indexing):**
- `src/extraction/models.py:151-215` - `ExtractedRecord` dataclass (input format)

**Database Models:**
- `src/db/models.py` - `ExtractedRecords` ORM model (lines TBD - check file)
- `src/db/session.py` - Async session factory

**Configuration:**
- `src/core/config.py` - Settings (Azure OpenAI, Milvus URLs)
- `.env` - Environment variables

**Worker Integration:**
- `src/workers/tasks.py` - Background task for document processing

### Testing Locations

**Test Files:**
- `tests/knowledge/` - All indexing tests (new directory)

**Fixtures:**
- `tests/conftest.py` - Project-wide fixtures
- `tests/knowledge/conftest.py` - Knowledge-specific fixtures

**Test Data:**
- `tests/fixtures/sample_excel/` - Sample Excel files (from Stories 3.1-3.4)

### Documentation to Update

**After Implementation:**
1. **README.md** - Add section on indexing layer setup
2. **docs/architecture.md** - Mark Epic 4 modules as ✅ IMPLEMENTED
3. **docs/epics.md** - Update Story 4.1-4.5 status to DONE
4. **API Documentation** - No changes (indexing is internal, no new endpoints)

---

## UX/UI Considerations

No UI/UX impact - backend/infrastructure change only.

**User-Facing Changes:**
- Document processing status now includes "indexing" stage
- Processing time increases by ~30s per 1000 records (for indexing)
- Document status endpoint (`GET /documents/{id}`) shows indexing metrics:
  ```json
  {
    "status": "completed",
    "record_count": 1250,
    "indexing_duration_ms": 28000
  }
  ```

---

## Testing Approach

### Test Strategy

**Unit Tests (Fast, Isolated):**
- Mock external dependencies (Azure OpenAI, Milvus)
- Test individual functions and classes
- Run on every commit (CI/CD)
- Target: < 10s total runtime for knowledge/ unit tests

**Integration Tests (Real Dependencies):**
- Use test PostgreSQL database (separate from dev)
- Use test Milvus collection
- Mock Azure OpenAI (avoid API costs in CI)
- Test full indexing flow end-to-end
- Run before merge to main

**Performance Tests (Benchmark):**
- Measure indexing time for 1000 records
- Verify targets: PostgreSQL < 3s, embedding < 30s, Milvus < 5s
- Run manually before release

**Coverage Requirements:**
- Overall: 80%+ coverage for new code
- Critical paths: 100% coverage (transactions, error handling)

### Test Framework (Already Configured)

From `pyproject.toml`:
```toml
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = "-v --tb=short"
```

**Key Testing Libraries:**
- pytest 8.0+
- pytest-asyncio 0.23+ (async test support)
- pytest-mock (mocking)

### Sample Test Structure

**Fixture Example (`tests/knowledge/conftest.py`):**
```python
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from milvus_client import AsyncMilvusClient
from src.knowledge.structured_store import StructuredStore
from src.knowledge.embeddings import EmbeddingService
from src.knowledge.vector_store import VectorStore

@pytest.fixture
async def db_session():
    """Async database session for testing."""
    engine = create_async_engine("postgresql+asyncpg://test:test@localhost/dsol_test")
    async with AsyncSession(engine) as session:
        yield session
        await session.rollback()  # Clean up after test

@pytest.fixture
async def milvus_client():
    """Milvus client for testing."""
    client = AsyncMilvusClient(url="http://localhost:6333")
    collection = "dsol_test_records"

    # Create test collection
    try:
        await client.delete_collection(collection)
    except:
        pass
    await client.create_collection(
        collection_name=collection,
        vectors_config=VectorParams(size=3072, distance=Distance.COSINE)
    )

    yield client

    # Cleanup
    await client.delete_collection(collection)

@pytest.fixture
def mock_azure_openai(mocker):
    """Mock Azure OpenAI client."""
    mock_client = mocker.AsyncMock()
    mock_client.embeddings.create.return_value = mocker.Mock(
        data=[mocker.Mock(embedding=[0.1] * 3072)]
    )
    return mock_client
```

**Unit Test Example (`tests/knowledge/test_structured_store.py`):**
```python
import pytest
from uuid import uuid4
from src.extraction.models import ExtractedRecord
from src.knowledge.structured_store import StructuredStore

@pytest.mark.asyncio
async def test_store_records_success(db_session):
    """Test storing records in PostgreSQL."""
    # Arrange
    store = StructuredStore(db_session)
    document_id = uuid4()
    records = [
        ExtractedRecord(
            content={"Item Code": "2024-4_0019", "Screen": "○"},
            headers=["Item Code", "Screen"],
            _source={
                "document_id": str(document_id),
                "filename": "test.xlsx",
                "sheet": "Sheet1",
                "table_id": 1,
                "row": 5,
                "col_range": "A:B"
            }
        )
    ]

    # Act
    count = await store.store_records(records, document_id)

    # Assert
    assert count == 1

    # Verify stored in database
    result = await db_session.execute(
        select(ExtractedRecords).where(ExtractedRecords.document_id == document_id)
    )
    stored_record = result.scalars().first()
    assert stored_record is not None
    assert stored_record.content == {"Item Code": "2024-4_0019", "Screen": "○"}

@pytest.mark.asyncio
async def test_store_records_duplicate_handling(db_session):
    """Test ON CONFLICT handling for duplicate records."""
    # Arrange
    store = StructuredStore(db_session)
    document_id = uuid4()
    record = ExtractedRecord(...)  # Same record

    # Act
    await store.store_records([record], document_id)
    await store.store_records([record], document_id)  # Duplicate

    # Assert - Should update, not error
    result = await db_session.execute(
        select(func.count()).select_from(ExtractedRecords).where(
            ExtractedRecords.document_id == document_id
        )
    )
    count = result.scalar()
    assert count == 1  # Only one record (updated, not duplicated)
```

**Integration Test Example (`tests/knowledge/test_indexer.py`):**
```python
@pytest.mark.asyncio
async def test_index_document_end_to_end(db_session, milvus_client, mock_azure_openai):
    """Test full indexing flow."""
    # Arrange
    structured_store = StructuredStore(db_session)
    embedding_service = EmbeddingService(mock_azure_openai)
    vector_store = VectorStore(milvus_client)
    indexer = Indexer(structured_store, embedding_service, vector_store)

    document_id = uuid4()
    records = [ExtractedRecord(...) for _ in range(10)]  # 10 test records

    # Act
    result = await indexer.index_document(document_id, records)

    # Assert
    assert result.records_stored == 10
    assert result.total_duration_ms > 0

    # Verify PostgreSQL
    pg_count = await db_session.execute(
        select(func.count()).select_from(ExtractedRecords).where(
            ExtractedRecords.document_id == document_id
        )
    )
    assert pg_count.scalar() == 10

    # Verify Milvus
    search_result = await milvus_client.scroll(
        collection_name="dsol_test_records",
        scroll_filter=Filter(must=[
            FieldCondition(key="document_id", match=MatchValue(value=str(document_id)))
        ])
    )
    assert len(search_result[0]) == 10
```

---

## Deployment Strategy

### Deployment Steps

**Pre-Deployment:**
1. ✅ Verify all tests pass (unit + integration)
2. ✅ Run performance benchmark (1000 records ≤ 30s)
3. ✅ Code review approved
4. ✅ Database migrations up to date (Alembic)

**Deployment (Rolling Update):**
1. **Deploy Database Changes:**
   - No new migrations needed (schema exists from Story 1.2)
   - Verify indexes exist: `idx_records_content`, `idx_records_resolved`

2. **Deploy Milvus Collection:**
   ```bash
   # Create production Milvus collection (if not exists)
   curl -X PUT "https://milvus-prod.example.com/collections/dsol_records" \
     -H "Content-Type: application/json" \
     -d '{
       "vectors": {
         "size": 3072,
         "distance": "Cosine"
       }
     }'
   ```

3. **Deploy Application Code:**
   ```bash
   # Build Docker image
   docker build -t dsol:epic4 .

   # Deploy API service (rolling update)
   kubectl set image deployment/dsol-api dsol-api=dsol:epic4

   # Deploy Celery worker (rolling update)
   kubectl set image deployment/dsol-worker dsol-worker=dsol:epic4
   ```

4. **Verify Deployment:**
   - Check health endpoint: `GET /health` (database, milvus should be "healthy")
   - Upload test document: `POST /documents`
   - Verify indexing completes: `GET /documents/{id}` (status: "completed")
   - Check logs for errors

**Post-Deployment:**
1. Monitor indexing metrics (processing_time_ms, records_per_second)
2. Monitor Milvus collection size (should grow with new documents)
3. Monitor PostgreSQL table size (`extracted_records` row count)

### Rollback Plan

**Scenario 1: Indexing Fails (Bug in Code)**
```bash
# Revert to previous version
kubectl rollout undo deployment/dsol-api
kubectl rollout undo deployment/dsol-worker

# No data loss: Extraction still works, just no indexing
# Documents remain in "processing" state, can re-process after fix
```

**Scenario 2: Performance Issues (Indexing Too Slow)**
```bash
# Quick fix: Increase batch sizes or add more Celery workers
kubectl scale deployment/dsol-worker --replicas=5

# If still slow: Revert deployment and optimize code offline
```

**Scenario 3: Milvus Issues**
```bash
# Fallback: Disable Milvus indexing temporarily
# Set env var: ENABLE_VECTOR_INDEXING=false
# Documents still indexed in PostgreSQL (exact lookups work)
# Semantic search unavailable until Milvus fixed
```

**Data Recovery:**
- No data loss risk: Extraction pipeline stores data during processing
- Worst case: Re-run indexing for all documents
  ```python
  # Admin script: reindex all completed documents
  documents = await db.execute(
      select(Documents).where(Documents.status == "completed")
  )
  for doc in documents:
      await indexer.reindex_document(doc.id, extracted_records)
  ```

### Monitoring

**Key Metrics (CloudWatch / Datadog):**
- `indexing.duration_ms` - Total indexing time per document
- `indexing.postgres.duration_ms` - PostgreSQL insert time
- `indexing.embedding.duration_ms` - Azure OpenAI embedding time
- `indexing.milvus.duration_ms` - Milvus upload time
- `indexing.records_per_second` - Throughput
- `indexing.errors` - Error count (with document_id tags)

**Alerts:**
- ⚠️ Warning: Indexing > 45s per 1000 records (target: 30s)
- 🚨 Critical: Indexing errors > 5% of documents
- 🚨 Critical: Milvus collection size not growing (upload failures)
- ⚠️ Warning: PostgreSQL `extracted_records` table > 10M rows (consider partitioning)

**Dashboards:**
- Indexing performance over time (duration trend)
- Error rate by error type (PostgreSQL, Milvus, Azure OpenAI)
- Knowledge base growth (records added per day)

**Logging (Structured with structlog):**
```python
logger.info(
    "document_indexed",
    document_id=doc_id,
    filename=filename,
    record_count=count,
    postgres_duration_ms=pg_ms,
    embedding_duration_ms=embed_ms,
    milvus_duration_ms=milvus_ms,
    total_duration_ms=total_ms
)

logger.error(
    "indexing_failed",
    document_id=doc_id,
    error_type="MilvusConnectionError",
    error_message=str(e),
    retry_attempt=retry_count
)
```

---

## Summary

**What This Tech Spec Covers:**
- Implementation of **Epic 4: Knowledge Base & Indexing** (5 stories)
- Dual-store indexing: PostgreSQL (exact lookups) + Milvus (semantic search)
- Batch processing optimization for performance (≤ 30s per 1000 records)
- Incremental indexing (add/delete/reindex without full rebuild)
- Integration with existing extraction pipeline (Stories 3.1-3.4)
- Complete test strategy and deployment plan

**Deliverables:**
- `src/knowledge/structured_store.py` - PostgreSQL operations (Story 4.1)
- `src/knowledge/embeddings.py` - Text → vector conversion (Story 4.4)
- `src/knowledge/indexer.py` - Orchestration (Stories 4.3, 4.5)
- `src/knowledge/vector_store.py` - Milvus operations (Story 4.4)
- Integration with Celery worker (`src/workers/tasks.py`)
- Comprehensive test suite (unit + integration tests)

**Success Criteria:**
- ✅ All 5 stories (4.1-4.5) acceptance criteria met
- ✅ Performance target achieved (≤ 30s per 1000 records)
- ✅ 80%+ test coverage for new code
- ✅ All tests pass (unit + integration)
- ✅ Zero data loss or corruption
- ✅ Enables Epic 5 (Q&A) implementation

**Next Steps After This Tech Spec:**
1. Implement Story 4.1 (Structured Store)
2. Implement Story 4.4 (Vector Embedding)
3. Implement Story 4.3 & 4.5 (Indexer Orchestrator)
4. Integration testing (end-to-end)
5. Performance testing (1000 records benchmark)
6. Deploy to production
7. **Then:** Start Epic 5 (Q&A & Answer Generation)

---

_This technical specification provides everything needed to implement Epic 4: Knowledge Base & Indexing for DSOL. All architectural decisions align with the PRD, Architecture, and existing extraction pipeline implementation._

_Ready for development! 🚀_