---
code_patterns:
- '@task.external_python with all imports inside function'
- LangChain AzureChatService for LLM calls
- Async batch processing with checkpointing
- Graceful degradation with fallbacks
- Pickle files for large data in Airflow
created: '2026-01-20'
files_to_modify:
- src/extraction_v2/record_to_natural_language.py
- airflow/dags/tasks/processing_tasks.py
- src/knowledge/vector_store.py
- tests/extraction_v2/test_record_to_natural_language.py
- tests/knowledge/test_vector_store.py
jira_id: TEXTIQ-35
slug: llm-natural-language-large-tables
status: ready-for-dev
stepsCompleted:
- 1
- 2
- 3
- 4
tech_stack:
- Python 3.11
- Airflow 2.x
- Azure OpenAI GPT-4o
- LangChain
- Milvus
- Pydantic
test_patterns:
- pytest with async support
- Mock LLM responses with unittest.mock
- Parametrized tests for edge cases
- Integration tests marked with @pytest.mark.integration
title: LLM-based Natural Language Conversion for Large Table Records
ticket_id: "TEXTIQ-35"
jira_url: "https://insight.fsoft.com.vn/jiradc/browse/TEXTIQ-35"
current_phase: "implementation"
delegated_to: "quick-dev"
page_id: ""
confluence_url: ""
parent_page_id: ""
micro_context:
  last_file: ""
  last_error: ""
  notes: ""
lastStep: "step-01-init"
lastContinued: "2026-01-21"
---

# Tech-Spec: LLM-based Natural Language Conversion for Large Table Records

**Created:** 2026-01-20

## Overview

### Problem Statement

Current large table records are embedded as simple pipe-delimited text (`"key: value | key: value"`), which produces suboptimal embeddings for semantic search. The original JSON structure is also lost after embedding, making precise field retrieval impossible.

### Solution

Use LLM (Azure OpenAI) to convert large table JSON records to natural language narratives before embedding, while preserving the original JSON in Milvus metadata field `record_json` for exact retrieval.

### Scope

**In Scope:**

- Create new LLM-based natural language converter for Excel table records **from LARGE TABLE pipeline ONLY**
- Integrate converter into `generate_embeddings_large_tables` task (line 565-747 in processing_tasks.py)
- Add `record_json` field to Milvus metadata for **Excel records WITHOUT `_legacy_metadata`** (standard table records only)
- Apply to new documents only (no migration of existing data)
- Maintain source context (sheet name, row number) in natural language output

**Out of Scope:**

- **DETAILED pipeline records** (images, flowcharts, shapes from `parse_excel_detailed`) - NO CHANGES to line 751-857
- **DETAILED pipeline metadata** - Records with `_legacy_metadata` field are NOT touched
- Word/PDF/PPT documents - keep existing semantic chunking
- Migration/re-indexing of existing large table records
- Modifying `record_to_text()` function used elsewhere in the system
- Performance optimization for LLM calls (can be addressed later if needed)

## Context for Development

### Codebase Patterns

**Airflow Patterns:**
- External Python tasks use `@task.external_python(python='/opt/airflow/.venv/bin/python')` decorator
- All imports MUST be inside external Python task functions (different Python environment)
- Large data stored in pickle files to avoid XCom size limits (max ~48KB)
- Cleanup pattern: Each task deletes temp pickle files after loading them

**LLM Service Patterns:**
- Use `AzureChatService` from `src/extraction_v2/azure_chat_service.py` (LangChain-based)
- Factory function `get_azure_chat_service()` returns cached instance
- Async invocation with `await service.ainvoke(messages)` for better performance
- Messages format: `[{"role": "system", "content": "..."}, {"role": "user", "content": "..."}]`

**Data Flow Patterns:**
- Large table chunks have structure: `{'content': {dict}, 'headers': [list], '_source': {dict}}`
- Text conversion happens in `generate_embeddings_task()` before embedding (line 636-641)
- Milvus metadata prepared in `_prepare_milvus_entities()` (vector_store.py:527+)
- Batch processing pattern: Process in batches with progress logging

**Error Handling:**
- Graceful degradation: Fall back to simple pipe format if LLM fails
- Checkpoint/retry pattern: Save progress for long-running operations
- Structured logging with `structlog` for observability

### Files to Reference

| File | Purpose | Key Details |
| ---- | ------- | ----------- |
| `airflow/dags/tasks/processing_tasks.py` | Current text conversion location | Line 636-641: Simple pipe format that needs replacement |
| `src/extraction_v2/azure_chat_service.py` | LLM service wrapper | LangChain-based, has `ainvoke()` for async calls |
| `src/extraction_v2/llm_header_detector.py` | Reference for LLM prompt patterns | Shows how to structure prompts for structured JSON output |
| `src/knowledge/vector_store.py` | Metadata storage in Milvus | Line 527+: `_prepare_milvus_entities()` builds metadata dict |
| `src/knowledge/embeddings.py` | Current `record_to_text()` function | Reference only - NOT modifying this |
| `src/core/config.py` | Azure OpenAI settings | GPT deployment: `gpt-4o`, already configured |
| `tests/extraction_v2/test_azure_chat_service.py` | Test patterns for LLM services | Mock patterns, factory function testing |
| `tests/knowledge/test_embeddings.py` | Test patterns for text conversion | Parametrized tests, edge case handling |

### Technical Decisions

**1. LLM Service Choice:**
- ✅ Use `AzureChatService` (LangChain-based) for consistency with codebase
- ❌ Not using direct OpenAI SDK (older pattern being phased out)

**2. Batch Processing:**
- Process records in batches (e.g., 10-20 at a time) to balance latency vs. throughput
- Use async `ainvoke()` for non-blocking LLM calls
- Add progress logging every N batches

**3. Prompt Design:**
- System prompt: Define role and output format (natural language narrative)
- User prompt: Provide record JSON + context (sheet name, row number)
- Request structured output: Natural language text that includes all field values

**4. Fallback Strategy:**
- If LLM call fails: Fall back to simple pipe format `' | '.join(...)`
- Log warnings for failed conversions (monitoring/debugging)
- Don't fail entire pipeline - continue with degraded quality

**5. Metadata Storage:**
- Add new field `record_json` to existing metadata dict
- Store only `record.content` dict (not internal fields like `_source`, `_merge_metadata`)
- Preserve existing metadata fields (`sheet_name`, `row_number`, `has_symbols`, `cell_colors`)

**6. Performance Considerations:**
- Accept increased latency for better embedding quality (LLM calls are slower)
- Future optimization: Can batch multiple records into single LLM prompt
- No caching for now (each record converted fresh)

## Implementation Plan

### Tasks

- [ ] **Task 1: Create RecordToNaturalLanguage service**
  - File: `src/extraction_v2/record_to_natural_language.py`
  - Action: Create new service class `RecordToNaturalLanguage` with async method `convert_batch()`
  - Details:
    - **Required imports:**
      ```python
      import asyncio
      import json
      from typing import List, Dict, Any, Optional
      import structlog
      from src.extraction_v2.azure_chat_service import get_azure_chat_service
      from src.knowledge.embeddings import record_to_text
      from src.extraction.models import ExtractedRecord
      ```

    - **Method signature:**
      ```python
      async def convert_batch(
          self,
          records: List[Dict[str, Any]],  # List of record.content dicts
          sheet_names: List[str],          # Corresponding sheet names
          row_numbers: List[int],          # Corresponding row numbers
          batch_size: int = 10             # Process N records per LLM call
      ) -> List[str]:
          """Convert records to natural language, returns list of strings in same order."""
      ```

    - **LLM Prompt Template (EXACT):**
      ```python
      SYSTEM_PROMPT = """You are a data converter. Convert Excel table records into clear, natural language sentences.

Output format: Start with "In [SheetName] (row [N]), " then describe all field values naturally in one sentence.

Example:
Input - Sheet: "Employee Data", Row: 42, Record: {"Name": "Alice", "Dept": "Eng", "Level": "Sr"}
Output - "In Employee Data (row 42), the name is Alice, the department is Eng, and the level is Sr."

Rules:
- Include ALL field values
- Use natural connectors: "the", "is", "and"
- Keep it one sentence
- Preserve exact field names from the record
"""

      USER_PROMPT_TEMPLATE = """Convert this record to natural language:

Sheet: {sheet_name}
Row: {row_number}
Record:
{record_json}
"""
      ```

    - **Timeout handling:** Wrap LLM call in `asyncio.wait_for(service.ainvoke(...), timeout=10.0)`

    - **Graceful fallback:** On exception, use `record_to_text()` from embeddings.py to preserve source context and resolved content:
      ```python
      try:
          response = await asyncio.wait_for(...)
      except Exception as e:
          logger.warning(f"LLM conversion failed for {sheet} row {row}: {e}")
          # Fallback: use existing record_to_text() to maintain quality
          temp_record = ExtractedRecord(
              content=record,
              headers=list(record.keys()),
              _source={'filename': 'unknown', 'sheet': sheet, 'row': row, ...}
          )
          return record_to_text(temp_record)
      ```

    - **Structured logging:** Log conversion success/failure rates per batch

- [ ] **Task 2: Integrate natural language conversion into generate_embeddings_large_tables**
  - File: `airflow/dags/tasks/processing_tasks.py`
  - Action: Replace simple pipe format (line 636-641) with LLM-based conversion in `generate_embeddings_large_tables` ONLY
  - Details:
    - **CRITICAL: This modifies ONLY the LARGE TABLE pipeline (line 565-747), NOT the DETAILED pipeline (line 751-857)**
    - Import `RecordToNaturalLanguage` inside `generate_embeddings_large_tables()` function (external Python requirement)
    - Create async helper function inside the task (similar to existing `embed_with_checkpoints` pattern at line 672-717):
      ```python
      async def convert_with_llm():
          from src.extraction_v2.record_to_natural_language import RecordToNaturalLanguage
          converter = RecordToNaturalLanguage()

          # Batch records for LLM conversion
          records_batch = []
          sheets_batch = []
          rows_batch = []

          for chunk in chunks:
              content = chunk.get('content') or chunk.get('text') or ...
              if isinstance(content, dict):  # Excel record
                  records_batch.append(content)
                  sheets_batch.append(chunk.get('_source', {}).get('sheet', 'Sheet1'))
                  rows_batch.append(chunk.get('_source', {}).get('row', 0))

          # Convert using LLM
          texts = await converter.convert_batch(records_batch, sheets_batch, rows_batch)
          return texts

      # Call from synchronous context
      texts = asyncio.run(convert_with_llm())
      ```
    - Keep existing non-Excel text extraction unchanged (Word/PDF/PPT)
    - Add progress logging: Log every 100 records converted with counter management
    - **DO NOT modify** `generate_embeddings_chunks` task (DETAILED pipeline)

- [ ] **Task 3: Add record_json to Milvus metadata with size validation**
  - File: `src/knowledge/vector_store.py`
  - Action: Update `_prepare_milvus_entities()` method to add `record_json` field for **standard Excel records ONLY** (not DETAILED pipeline)
  - Details:
    - Locate Excel record handling block (line 515-556, `if document_type == "excel"`)
    - **ONLY add `record_json` for records WITHOUT `_legacy_metadata`** (standard table records from LARGE TABLE pipeline)
    - After building base `metadata` dict (line 527-531), add conditional logic:
      ```python
      # Only add record_json for standard table records (not DETAILED pipeline)
      if not (hasattr(record, '_legacy_metadata') and record._legacy_metadata):
          # Validate JSON serialization and size
          try:
              record_json_str = json.dumps(record.content, ensure_ascii=False, default=str)
              json_size_kb = len(record_json_str.encode('utf-8')) / 1024

              # Milvus JSON field limit: 65KB (leave margin)
              if json_size_kb > 50:
                  logger.warning(f"Record JSON too large ({json_size_kb:.1f}KB), truncating to first 20 fields")
                  # Store first 20 fields as fallback
                  truncated = dict(list(record.content.items())[:20])
                  metadata['record_json'] = truncated
                  metadata['record_json_truncated'] = True
              else:
                  metadata['record_json'] = record.content

          except (TypeError, ValueError) as e:
              logger.warning(f"Record content not JSON serializable: {e}, storing as string")
              metadata['record_json'] = str(record.content)
      ```
    - Ensure this happens BEFORE legacy metadata handling (line 538-554)
    - **DO NOT add `record_json` to DETAILED pipeline records** (those with `_legacy_metadata`)

- [ ] **Task 4: Write unit tests for RecordToNaturalLanguage**
  - File: `tests/extraction_v2/test_record_to_natural_language.py`
  - Action: Create comprehensive test suite
  - Details:
    - Test successful conversion with mocked LLM response
    - Test graceful fallback when LLM fails (mock exception)
    - Test batch processing with multiple records
    - Test empty record handling
    - Test special characters in record values
    - Parametrize test for different record structures
    - Mock `AzureChatService.ainvoke()` using `unittest.mock.AsyncMock`
    - Follow existing test patterns from `tests/extraction_v2/test_azure_chat_service.py`

- [ ] **Task 5: Write integration tests for vector_store metadata**
  - File: `tests/knowledge/test_vector_store.py`
  - Action: Add tests for `record_json` field in metadata
  - Details:
    - Test that `_prepare_milvus_entities()` includes `record_json` for Excel records
    - Verify `record_json` contains only `record.content` dict
    - Verify existing metadata fields still present (`sheet_name`, `row_number`, etc.)
    - Test that Word records do NOT have `record_json` field
    - Use existing test fixtures and patterns from the file

### Acceptance Criteria

- [ ] **AC1: Natural language conversion produces readable text**
  - Given a large table record with fields `{"Name": "Alice", "Age": "30", "Department": "Engineering"}`
  - When the record is processed through `RecordToNaturalLanguage.convert_records_batch()`
  - Then the output is natural language like "In Sheet1 (row 5), the name is Alice, age is 30, and department is Engineering"
  - And the text includes all field values from the original record
  - And the text includes source context (sheet name and row number)

- [ ] **AC2: LLM failure gracefully falls back to pipe format**
  - Given the LLM service raises an exception during conversion
  - When `RecordToNaturalLanguage.convert_records_batch()` is called
  - Then the service logs a warning about the failure
  - And returns the simple pipe format: "Name: Alice | Age: 30 | Department: Engineering"
  - And the pipeline continues without failing

- [ ] **AC3: Original JSON is stored in Milvus metadata for LARGE TABLE records ONLY**
  - Given a large table record (from `generate_embeddings_large_tables`) is processed and stored in Milvus
  - When querying the Milvus entity
  - Then the metadata field contains `record_json` key
  - And `record_json` value equals the original `record.content` dict
  - And existing metadata fields (`sheet_name`, `row_number`, `has_symbols`) are still present
  - And **DETAILED pipeline records** (with `_legacy_metadata` - images, flowcharts, shapes) do NOT have `record_json` field

- [ ] **AC4: Integration with large table pipeline works end-to-end**
  - Given a large Excel table (>1000 rows) is uploaded and processed
  - When the `generate_embeddings_task` runs in Airflow
  - Then natural language conversion is applied to all large table records
  - And embeddings are generated from the natural language text
  - And vectors are stored in Milvus with `record_json` metadata
  - And no records are lost or skipped due to conversion errors

- [ ] **AC5: DETAILED pipeline and Word/PDF/PPT documents are unaffected**
  - Given a Word/PDF/PPT document is processed OR an Excel DETAILED pipeline record (image/flowchart/shape)
  - When the `generate_embeddings_chunks` task runs (DETAILED pipeline)
  - Then the existing text extraction logic is used (NOT natural language conversion)
  - And the metadata does NOT contain `record_json` field
  - And DETAILED pipeline processing works exactly as before (no changes)

- [ ] **AC6: Batch processing handles multiple records efficiently**
  - Given 100 large table records need conversion
  - When `RecordToNaturalLanguage.convert_records_batch()` is called
  - Then records are processed in batches (e.g., 10-20 at a time)
  - And progress is logged every 100 records
  - And total conversion time is reasonable (< 5 seconds per record average)

- [ ] **AC7: Special characters and edge cases handled correctly**
  - Given a record contains special characters (`&`, `"`, `'`, unicode)
  - When the record is converted to natural language
  - Then all characters are preserved correctly
  - And the JSON stored in `record_json` is valid and parseable
  - And empty/null values are handled gracefully (skipped or marked as "not provided")

## Additional Context

### Dependencies

**External Services:**
- Azure OpenAI API (already configured via `settings.azure_openai_*` in `.env`)
- GPT-4o deployment (`settings.azure_openai_deployment`)

**Internal Dependencies:**
- `AzureChatService` from `src/extraction_v2/azure_chat_service.py` (LangChain wrapper)
- Airflow external Python environment at `/opt/airflow/.venv` with all dependencies installed
- Milvus collection schema already supports JSON metadata fields

**No New External Dependencies Required:**
- All required libraries already in use (LangChain, OpenAI SDK, structlog)

### Testing Strategy

**Unit Tests:**
- Mock `AzureChatService.ainvoke()` to avoid real LLM calls during tests
- Use `unittest.mock.AsyncMock` for async method mocking
- Parametrize tests for different record structures (simple, complex, edge cases)
- Test error handling and fallback scenarios explicitly

**Integration Tests:**
- Mark with `@pytest.mark.integration` decorator
- Test full flow: record → natural language → embedding → Milvus storage
- Verify `record_json` metadata persists correctly in Milvus
- Use test fixtures for sample Excel records

**Manual Testing:**
- Upload a large Excel file (>1000 rows) via API
- Monitor Airflow DAG execution logs for natural language conversion
- Query Milvus to verify `record_json` field is present
- Compare search quality before/after (qualitative assessment)

**Performance Testing:**
- Measure conversion time for 100 records batch
- Ensure total pipeline latency increase is acceptable (<2x current time)
- Monitor LLM API rate limits and costs

### Notes

**Current State:**
- Text conversion location: `airflow/dags/tasks/processing_tasks.py:641`
- Current format: `text = ' | '.join(f"{k}: {v}" for k, v in content.items() if v)`
- Large table classification: Records from tables with `>1000 rows`

**Implementation Risks:**
- **High Risk:** LLM latency could significantly slow down pipeline (mitigation: batch processing, async calls)
- **Medium Risk:** LLM costs could increase substantially (mitigation: monitor usage, implement rate limiting if needed)
- **Low Risk:** Prompt engineering may need iteration to get good natural language output (mitigation: start simple, refine based on results)

**Known Limitations:**
- No caching of LLM conversions (each record converted fresh every time)
- No retry logic for transient LLM failures (falls back immediately)
- Batch size is fixed (not dynamically adjusted based on performance)

**Future Considerations (Out of Scope):**
- Optimize prompt to batch multiple records in single LLM call
- Add caching layer for repeated conversions of identical records
- Implement adaptive batch sizing based on LLM latency
- Add metrics/monitoring for conversion quality and costs
- Support custom prompts per domain/document type