---
title: 'Dynamic Domain Term Dictionary with Multilingual Support'
slug: 'dynamic-domain-term-dictionary'
created: '2026-01-14T16:13:37+07:00'
status: 'ready-for-dev'
stepsCompleted: [1, 2, 3, 4]
tech_stack: ['PostgreSQL', 'SQLAlchemy', 'Azure OpenAI', 'Alembic']
files_to_modify:
  - src/db/models.py
  - src/extraction_v2/domain_term_extractor.py
  - airflow/dags/tasks/processing_tasks.py
files_to_create:  # F15 fix: explicitly list new files
  - src/knowledge/domain_dictionary_service.py
  - src/extraction_v2/translation_service.py
  - migrations/versions/xxx_add_domain_term_dictionary.py
test_files_to_create:  # F16 fix: track test files
  - tests/extraction_v2/test_translation_service.py
  - tests/knowledge/test_domain_dictionary_service.py
  - tests/integration/test_domain_dictionary_pipeline.py
code_patterns:
  - 'SQLAlchemy ORM with DeclarativeBase'
  - 'Lazy AzureOpenAI client initialization'
  - 'Pickle-based XCom data passing in Airflow'
test_patterns:
  - 'pytest with unittest.mock for LLM mocking'
  - 'tests/extraction_v2/test_domain_term_extractor.py as reference'
---

# Tech-Spec: Dynamic Domain Term Dictionary with Multilingual Support

**Created:** 2026-01-14

## Overview

### Problem Statement

The current system extracts domain terms from documents but stores them ephemerally per-chunk in Milvus. There is no persistent, centralized dictionary that:
- Accumulates domain-specific vocabulary across all processed documents
- Provides multilingual translations (Japanese, English, Vietnamese)
- Automatically grows as new documents introduce new terminology

### Solution

Build a dynamic domain term dictionary that:
1. **Auto-extracts** new domain terms during document processing via LLM
2. **Auto-translates** each term to Japanese, English, and Vietnamese at extraction time
3. **Deduplicates** by checking if the extracted term already exists in the dictionary
4. **Stores persistently** in PostgreSQL for durability and future API exposure

### Scope

**In Scope:**
- New PostgreSQL table for domain term dictionary with multilingual fields
- Integration with `DomainTermExtractor` to auto-add new terms during document processing
- LLM-powered translation to generate Japanese/English/Vietnamese variants
- Simple deduplication: skip if `term_original` already exists in the dictionary (F18 fix: clarified)
- Global dictionary (single shared dictionary for all tenants/projects)

**Out of Scope:**
- API endpoints for dictionary management (future phase)
- Per-tenant/per-project dictionary isolation (future feature, but schema should support it)
- RAG query expansion using dictionary (future phase)
- User approval workflow before adding terms (auto-add for now)
- Synonym/alias management within a single term entry
- UI implementation

## Context for Development

### Codebase Patterns

1. **SQLAlchemy ORM**: Models in `src/db/models.py` use `DeclarativeBase`, `Mapped`, `mapped_column` patterns with PostgreSQL-specific types (UUID, JSONB)

2. **Lazy Client Initialization**: `DomainTermExtractor._get_client()` lazily creates AzureOpenAI client to avoid fork issues in Airflow workers - same pattern should be used for translation

3. **Airflow External Python**: Processing tasks use `@task.external_python` decorator and pickle files for large data (XCom size limits)

4. **Domain Extraction Flow**:
   ```
   chunk_task → extract_domain_terms_task → generate_embeddings_task → store_vectors_task
   ```
   Integration point: `extract_domain_terms_task` in `processing_tasks.py`

### Files to Reference

| File | Purpose |
| ---- | ------- |
| [models.py](file:///home/neil/Documents/textiq-doc-extraction/src/db/models.py) | SQLAlchemy ORM patterns, existing table structures |
| [domain_term_extractor.py](file:///home/neil/Documents/textiq-doc-extraction/src/extraction_v2/domain_term_extractor.py) | Current term extraction logic, LLM patterns |
| [processing_tasks.py](file:///home/neil/Documents/textiq-doc-extraction/airflow/dags/tasks/processing_tasks.py) | Airflow task integration point (lines 372-533) |
| [vector_store.py](file:///home/neil/Documents/textiq-doc-extraction/src/knowledge/vector_store.py) | Milvus entity patterns, domain_terms storage |
| [test_domain_term_extractor.py](file:///home/neil/Documents/textiq-doc-extraction/tests/extraction_v2/test_domain_term_extractor.py) | Test patterns reference |

### Technical Decisions

**Storage: PostgreSQL**
- CRUD-friendly for future API exposure
- Easy to add `project_id` column for per-project isolation later
- Matches existing `symbol_dictionaries` table pattern

**Language Detection (F1 fix):**
- Use Unicode range detection:
  - Japanese: Hiragana (\u3040-\u309F), Katakana (\u30A0-\u30FF), or mixed with Kanji
  - Vietnamese: Latin with diacritics (ắ, ế, ơ, etc.)
  - English: ASCII-only Latin characters
- Reuse existing `DomainTermExtractor._is_japanese()` method as reference
- Default to "en" if detection is ambiguous

**Deduplication Strategy: Batch lookup with unique constraint (F4, F7 fix)**
- Add unique constraint on `(term_original, project_id)` to prevent race conditions
- Batch dedup: Load all existing `term_original` values into a set before processing
- O(1) lookup per term instead of O(n) database queries
- Skip insertion if term exists in the in-memory set

**Translation: LLM-based (Azure OpenAI GPT-4o)**
- Single prompt to translate term to all 3 languages
- Handle untranslatable terms (store NULL for missing languages)
- Batch translations for efficiency (future optimization)

**Error Handling Strategy (F2 fix):**
- If LLM translation fails: Store term with `term_original` only, leave `term_ja/en/vi` as NULL
- Log warning but do NOT block document processing
- Retry logic: 1 retry with exponential backoff (2s delay)
- On persistent failure: Mark term as `translation_failed=True` (optional column) for later retry

**Airflow Session Management (F3 fix):**
- Create SQLAlchemy engine/session inside `@task.external_python` function (not passed via XCom)
- Use `settings.database_url` to create session within the Airflow worker process
- Session lifecycle: create → use → commit/rollback → close within task

## Implementation Plan

### Tasks

#### 1. [NEW] Create PostgreSQL Table Model

**File:** `src/db/models.py`

Add new `DomainTermDictionary` model:

```python
class DomainTermDictionary(Base):
    """Domain term dictionary with multilingual support."""
    
    __tablename__ = "domain_term_dictionary"
    
    id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    term_original: Mapped[str] = mapped_column(String(255), nullable=False)  # Original extracted term
    term_ja: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)   # Japanese
    term_en: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)   # English
    term_vi: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)   # Vietnamese
    source_language: Mapped[str] = mapped_column(String(10), nullable=False)  # ja, en, vi
    domain_category: Mapped[str] = mapped_column(String(100), default="general")
    source_document_id: Mapped[Optional[UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
    project_id: Mapped[Optional[UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)  # Future: per-project
    translation_failed: Mapped[bool] = mapped_column(Boolean, default=False)  # F10 fix: mark for retry
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    
    # Indexes for fast dedup lookups + unique constraint (F7 fix)
    __table_args__ = (
        Index("idx_domain_dict_term_ja", "term_ja"),
        Index("idx_domain_dict_term_en", "term_en"),
        Index("idx_domain_dict_term_vi", "term_vi"),
        Index("idx_domain_dict_project_id", "project_id"),
        UniqueConstraint("term_original", "project_id", name="uq_term_original_project"),
    )
```

#### 2. [NEW] Create Alembic Migration

**File:** `migrations/versions/xxx_add_domain_term_dictionary.py`

Generate and run migration for the new table.

#### 3. [NEW] Create DomainDictionaryService

**File:** `src/knowledge/domain_dictionary_service.py`

Service for dictionary CRUD operations:

```python
# F8 fix: Required imports
from sqlalchemy import or_, func
from sqlalchemy.orm import Session
from uuid import UUID
from src.db.models import DomainTermDictionary

class DomainDictionaryService:
    """Service for managing domain term dictionary."""
    
    def __init__(self, session: Session):
        self.session = session
    
    # F14 fix: Removed unused term_exists() method - use get_existing_terms_set() instead
    
    def add_term(self, entry: DomainTermDictionary) -> DomainTermDictionary:
        """Add new term to dictionary."""
        self.session.add(entry)
        self.session.commit()
        return entry
    
    def get_all_terms(self, project_id: UUID | None = None) -> list[DomainTermDictionary]:
        """Get all dictionary terms."""
        query = self.session.query(DomainTermDictionary)
        if project_id:
            query = query.filter(DomainTermDictionary.project_id == project_id)
        return query.all()
    
    def get_existing_terms_set(self, project_id: UUID | None = None) -> set[str]:
        """Load all term_original values as a set for O(1) dedup lookup (F4 fix)."""
        query = self.session.query(DomainTermDictionary.term_original)
        if project_id:
            query = query.filter(DomainTermDictionary.project_id == project_id)
        return {row[0].lower() for row in query.all()}
    
    def add_terms_batch(self, entries: list[DomainTermDictionary]) -> int:
        """Add multiple terms in a single transaction."""
        self.session.add_all(entries)
        self.session.commit()
        return len(entries)
```

#### 4. [NEW] Create TranslationService

**File:** `src/extraction_v2/translation_service.py`

LLM-based translation service:

```python
# F8 fix: Required imports
import json
import time
import structlog
from openai import AzureOpenAI
from src.core.config import settings

logger = structlog.get_logger(__name__)

TRANSLATION_PROMPT = """Translate the following domain term to Japanese, English, and Vietnamese.

Term: {term}
Source language: {source_lang}

Rules:
- If the term is already in a target language, copy it as-is
- If the term cannot be accurately translated, return null for that language
- Return ONLY a JSON object with keys: ja, en, vi

Output ONLY valid JSON:
{"ja": "...", "en": "...", "vi": "..."}
"""

class TranslationService:
    """LLM-powered term translation service (F5 fix)."""
    
    def __init__(self, client: AzureOpenAI | None = None):
        self._client = client
        self._client_initialized = client is not None
    
    def _get_client(self) -> AzureOpenAI:
        """Lazy client initialization (same pattern as DomainTermExtractor)."""
        if not self._client_initialized:
            import httpx
            http_client = httpx.Client(http2=False, timeout=httpx.Timeout(60.0))
            self._client = AzureOpenAI(
                api_key=settings.azure_openai_api_key,
                api_version=settings.azure_openai_api_version,
                azure_endpoint=settings.azure_openai_endpoint,
                http_client=http_client,
            )
            self._client_initialized = True
        return self._client
    
    def translate_term(self, term: str, source_lang: str, max_retries: int = 1) -> tuple[dict[str, str | None], bool]:
        """Translate term to Japanese, English, Vietnamese.
        
        Args:
            term: Term to translate
            source_lang: Source language code (ja, en, vi)
            max_retries: Number of retries on failure (F9 fix)
        
        Returns:
            Tuple of (translations dict, success bool)
            translations: {"ja": "...", "en": "...", "vi": "..."}
            success: True if translation succeeded, False if failed (for translation_failed column)
        """
        for attempt in range(max_retries + 1):
            try:
                client = self._get_client()
                prompt = TRANSLATION_PROMPT.format(term=term, source_lang=source_lang)
                
                response = client.chat.completions.create(
                    model=settings.azure_openai_deployment,
                    messages=[{"role": "user", "content": prompt}],
                    temperature=0.0,
                    max_tokens=200
                )
                
                result = json.loads(response.choices[0].message.content.strip())
                return {
                    "ja": result.get("ja"),
                    "en": result.get("en"),
                    "vi": result.get("vi"),
                }, True  # Success
                
            except Exception as e:
                if attempt < max_retries:
                    backoff = 2 ** attempt  # F9 fix: exponential backoff (1s, 2s)
                    logger.warning(f"Translation attempt {attempt + 1} failed for '{term}': {e}. Retrying in {backoff}s...")
                    time.sleep(backoff)
                else:
                    logger.warning(f"Translation failed for '{term}' after {max_retries + 1} attempts: {e}")
                    return {"ja": None, "en": None, "vi": None}, False  # Failed
        
        # F17 fix: Defensive return (unreachable but ensures type safety)
        return {"ja": None, "en": None, "vi": None}, False
```

#### 5. [MODIFY] Update DomainTermExtractor

**File:** `src/extraction_v2/domain_term_extractor.py`

**Depends on:** Task 3 (DomainDictionaryService), Task 4 (TranslationService) ← (F6 fix)

Add method to persist new terms to dictionary:

```python
def detect_language(self, term: str) -> str:
    """Detect language of term using Unicode ranges (F1 fix)."""
    # Check for Japanese characters (Hiragana, Katakana, CJK)
    if self._is_japanese(term):  # Existing method
        return "ja"
    # Check for Vietnamese diacritics
    vietnamese_chars = set("àáảãạăằắẳẵặâầấẩẫậèéẻẽẹêềếểễệìíỉĩịòóỏõọôồốổỗộơờớởỡợùúủũụưừứửữựỳýỷỹỵđ")
    if any(c in vietnamese_chars for c in term.lower()):
        return "vi"
    return "en"  # Default

def persist_new_terms(
    self,
    terms: list[str],
    domain_category: str,
    document_id: UUID | None = None,
    project_id: UUID | None = None
) -> int:
    """Persist new terms to dictionary with translations.
    
    Returns count of new terms added.
    """
    from src.db.models import DomainTermDictionary
    from src.knowledge.domain_dictionary_service import DomainDictionaryService
    from src.extraction_v2.translation_service import TranslationService
    from sqlalchemy import create_engine
    from sqlalchemy.orm import Session
    
    # F3 fix: Create session inside function for Airflow compatibility
    engine = create_engine(settings.database_url)
    with Session(engine) as session:
        service = DomainDictionaryService(session)
        translator = TranslationService()
        
        # F4 fix: Load existing terms once for O(1) lookup
        existing = service.get_existing_terms_set(project_id)
        
        new_entries = []
        for term in terms:
            if term.lower() in existing:
                continue  # Skip duplicate
            
            source_lang = self.detect_language(term)
            translations, success = translator.translate_term(term, source_lang)  # F9: unpack tuple
            
            entry = DomainTermDictionary(
                term_original=term,
                term_ja=translations["ja"],
                term_en=translations["en"],
                term_vi=translations["vi"],
                source_language=source_lang,
                domain_category=domain_category,
                source_document_id=document_id,
                project_id=project_id,
                translation_failed=not success,  # F10 fix: track failures
            )
            new_entries.append(entry)
            existing.add(term.lower())  # Update local set
        
        if new_entries:
            service.add_terms_batch(new_entries)
        
        return len(new_entries)
```

#### 6. [MODIFY] Update extract_domain_terms_task

**File:** `airflow/dags/tasks/processing_tasks.py`

**Depends on:** Task 5 (persist_new_terms method) ← (F6 fix)

After extracting terms, call persist_new_terms to add to dictionary:

```python
# F11 fix: Complete integration code (add after line ~530, before return statement)

# Collect unique terms across all chunks
all_terms = set()
for data in domain_data:
    all_terms.update(data['domain_terms'])

# Infer dominant category from chunk categories
from collections import Counter
category_counts = Counter(data['domain_category'] for data in domain_data)
dominant_category = category_counts.most_common(1)[0][0] if category_counts else 'general'

# Get document_id from chunks_result metadata (passed via XCom)
# Note: document_id is available in chunks_result from upstream task
document_id = chunks_result.get('document_id')  # May be None for some pipelines
if document_id:
    from uuid import UUID
    document_id = UUID(document_id) if isinstance(document_id, str) else document_id

# Persist new terms to dictionary (skip duplicates, translate via LLM)
if all_terms:
    new_count = extractor.persist_new_terms(
        terms=list(all_terms),
        domain_category=dominant_category,
        document_id=document_id,
        project_id=None  # Global dictionary for MVP
    )
    log.info(f"Persisted {new_count} new terms to domain dictionary")
```

### Acceptance Criteria

**AC1: Database Schema**
- Given the migration is run
- When I query the `domain_term_dictionary` table
- Then it exists with columns: id, term_original, term_ja, term_en, term_vi, source_language, domain_category, source_document_id, project_id, translation_failed, created_at

**AC2: Term Deduplication**
- Given a term "revenue" already exists in the dictionary
- When a new document extracts "revenue" again
- Then the term is NOT added again (dedup check passes)

**AC3: Multilingual Translation**
- Given a new term "売上" (Japanese) is extracted
- When the term is persisted to dictionary
- Then it is translated to English and Vietnamese via LLM (F19 fix: removed specific translation example as LLM output varies)

**AC4: Missing Translation Handling**
- Given a term that cannot be translated to a language
- When persisting to dictionary
- Then that language column is NULL (not empty string)

**AC5: Integration with Pipeline**
- Given a document is processed through the Airflow DAG
- When `extract_domain_terms_task` completes
- Then new unique terms are added to the dictionary

**AC6: Language Detection (F13 fix)**
- Given a Japanese term "売上" is extracted
- When the term is persisted to dictionary
- Then `source_language` is "ja"
- Given a Vietnamese term "doanh thu" is extracted
- When the term is persisted to dictionary
- Then `source_language` is "vi"
- Given an English term "revenue" is extracted
- When the term is persisted to dictionary
- Then `source_language` is "en"

**AC7: Translation Failure Tracking**
- Given the LLM translation fails for a term
- When the term is persisted to dictionary
- Then `translation_failed` is True and `term_ja/en/vi` are NULL

## Additional Context

### Dependencies

- `sqlalchemy` (existing)
- `alembic` (existing, for migrations)
- `openai` (existing, for Azure OpenAI)

No new dependencies required.

### Testing Strategy

**Unit Tests:**
- `tests/extraction_v2/test_translation_service.py` - Mock LLM, test translation parsing
- `tests/knowledge/test_domain_dictionary_service.py` - Test dedup logic, CRUD operations

**Integration Test:**
- `tests/integration/test_domain_dictionary_pipeline.py` - End-to-end test with real DB

**Manual Verification:**
1. Run migration: `alembic upgrade head`
2. Verify table exists: `psql -c "\\d domain_term_dictionary"`
3. Process a test document through DAG
4. Query dictionary: `SELECT * FROM domain_term_dictionary LIMIT 10;`

### Notes

- **Future: Per-project isolation** - `project_id` column is nullable now, can be populated when multi-project support is added
- **Future: RAG query expansion** - Dictionary can be used to expand queries with multilingual synonyms
- **Performance**: Batch translations could be optimized by grouping terms, but single-term translation is fine for MVP