Evaluations Module Developer Guide

Table of Contents

• Architecture Overview
• Module Structure
• Database Schema
• Internal API Usage
• Adding New Evaluation Types
• Extending the Evaluation Runner
• Integration Points
• Testing
• Performance Considerations

Architecture Overview

The Evaluations module follows a layered architecture pattern with clear separation of concerns:

┌─────────────────────────────────────┐
│         API Layer (FastAPI)         │
│   /app/api/v1/endpoints/            │
├─────────────────────────────────────┤
│        Business Logic Layer         │
│   /app/core/Evaluations/            │
├─────────────────────────────────────┤
│        Database Layer               │
│   /app/core/DB_Management/          │
├─────────────────────────────────────┤
│        Storage Layer                │
│   SQLite (evaluations.db)           │
└─────────────────────────────────────┘

Key Components

1. Unified API Endpoints (evaluations_unified.py)
2. Unified REST API under /api/v1/evaluations (OpenAI-compatible + tldw-specific)
3. Request/response validation with unified schemas

4. Endpoint tag: evaluations

5. Unified Evaluation Service (unified_evaluation_service.py)

6. Central orchestration (DB + runner + evaluators + webhooks)

7. Create/list/get/update/delete evaluations, runs, datasets

8. Evaluation Runner (eval_runner.py)

9. Async task orchestration for runs

10. Progress tracking and aggregation

11. Evaluators (various *_evaluator.py files)

12. Implementations: G-Eval, RAG, response quality, OCR, etc.

13. Metric calculations and LLM integration

14. Database Manager (Evaluations_DB.py)

15. CRUD operations (evaluations, runs, datasets)
16. Additional registries (webhooks, pipeline presets, embeddings A/B tests)

Module Structure

app/core/Evaluations/
├── unified_evaluation_service.py   # Unified evaluation orchestration
├── eval_runner.py                  # Evaluation run executor
├── ms_g_eval.py                    # G-Eval implementation
├── rag_evaluator.py                # RAG evaluator
├── response_quality_evaluator.py   # Response quality evaluator
└── evaluation_manager.py           # Legacy manager (kept for compatibility)

app/api/v1/
├── endpoints/
│   └── evaluations_unified.py      # Unified Evaluations API
└── schemas/
    ├── evaluation_schemas_unified.py  # Unified request/response models
    ├── embeddings_abtest_schemas.py   # Embeddings A/B test models
    └── evaluation_schema.py           # Legacy schemas (compat)

app/core/DB_Management/
└── Evaluations_DB.py              # Database operations (unified schema)

Database Schema

Tables

evaluations

CREATE TABLE evaluations (
    id TEXT PRIMARY KEY,           -- eval_xxxxxxxxxxxx
    name TEXT NOT NULL,
    description TEXT,
    eval_type TEXT NOT NULL,        -- model_graded, exact_match, etc.
    eval_spec TEXT NOT NULL,        -- JSON configuration
    dataset_id TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_by TEXT,
    metadata TEXT,                  -- JSON metadata
    deleted_at TIMESTAMP NULL       -- Soft delete
);

evaluation_runs

CREATE TABLE evaluation_runs (
    id TEXT PRIMARY KEY,           -- run_xxxxxxxxxxxx
    eval_id TEXT NOT NULL,
    status TEXT NOT NULL,          -- pending, running, completed, failed
    target_model TEXT,
    config TEXT,                   -- JSON run configuration
    progress TEXT,                 -- JSON progress info
    results TEXT,                  -- JSON results
    error_message TEXT,
    started_at TIMESTAMP,
    completed_at TIMESTAMP,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    webhook_url TEXT,
    usage TEXT,                    -- JSON token usage
    FOREIGN KEY (eval_id) REFERENCES evaluations(id)
);

datasets

CREATE TABLE datasets (
    id TEXT PRIMARY KEY,           -- dataset_xxxxxxxxxxxx
    name TEXT NOT NULL,
    description TEXT,
    samples TEXT NOT NULL,         -- JSON array of samples
    sample_count INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_by TEXT,
    metadata TEXT                  -- JSON metadata
);

Additional Tables (Unified)

The unified service also manages auxiliary tables used by evaluation features:

• internal_evaluations: tldw-specific evaluations (RAG, response quality) with lifecycle/state
• pipeline_presets: saved RAG pipeline configurations for evaluation runs
• ephemeral_collections: TTL registry for temporary vector collections
• webhook_registrations: per-user webhook registrations and delivery stats
• Embeddings A/B tests:
• embedding_abtests: test metadata/config/stats
• embedding_abtest_arms: per-arm provider/model settings and stats
• embedding_abtest_queries: queries and optional ground-truth ids
• embedding_abtest_results: per-arm per-query results/metrics/latency
• Persistence: SQLite uses the SQLAlchemy repository (embeddings_abtest_repository.py) with create_all (no separate migrations yet); Postgres falls back to the legacy adapter. Override with EVALS_ABTEST_PERSISTENCE.
• v2 highlights: Jobs backend with retry/backoff, deterministic collection reuse via hashing, allowlist/quota enforcement at API/worker, structured logs + Prometheus metrics, stable JSON/CSV export schema.

Internal API Usage

Using Evaluations Programmatically

Option A - Unified service (recommended):

from tldw_Server_API.app.core.Evaluations.unified_evaluation_service import UnifiedEvaluationService

svc = UnifiedEvaluationService(db_path="Databases/evaluations.db")
await svc.initialize()

# (Optional) create a dataset inline
dataset = await svc.create_dataset(
    name="demo_ds",
    description="Sample eval dataset",
    samples=[{"input": {"text": "hello"}, "expected": {"label": "greeting"}}],
    created_by="dev"
)

# Create an evaluation definition
evaluation = await svc.create_evaluation(
    name="my_eval",
    eval_type="model_graded",
    eval_spec={"sub_type": "summarization", "metrics": ["fluency", "relevance"], "model": "gpt-4"},
    dataset_id=dataset["id"],
    created_by="dev"
)

# Start a run
run = await svc.create_run(
    eval_id=evaluation["id"],
    target_model="gpt-4o",
    config={"temperature": 0.0},
    created_by="dev"
)

# Poll run
run_status = await svc.get_run(run["id"])

Option B - Direct DB + runner (supported):

from tldw_Server_API.app.core.DB_Management.Evaluations_DB import EvaluationsDatabase
from tldw_Server_API.app.core.Evaluations.eval_runner import EvaluationRunner

db = EvaluationsDatabase("Databases/evaluations.db")
runner = EvaluationRunner("Databases/evaluations.db")

eval_id = db.create_evaluation(
    name="my_evaluation",
    eval_type="model_graded",
    eval_spec={"sub_type": "summarization", "metrics": ["fluency", "relevance"], "model": "gpt-4"}
)
run_id = db.create_run(eval_id=eval_id, target_model="gpt-4o", config={"temperature": 0.0})
await runner.run_evaluation(run_id=run_id, eval_id=eval_id, eval_config={"eval_type": "model_graded"})
results = db.get_run_results(run_id)

Direct Evaluator Usage

from tldw_Server_API.app.core.Evaluations.rag_evaluator import RAGEvaluator

# For RAG evaluation
rag_eval = RAGEvaluator()
results = await rag_eval.evaluate(
    query="What is the capital of France?",
    contexts=["Paris is the capital of France.", "France is in Europe."],
    response="The capital of France is Paris.",
    ground_truth="Paris",
    metrics=["relevance", "faithfulness"],
    api_name="openai"
)

# For G-Eval summarization
from tldw_Server_API.app.core.Evaluations.ms_g_eval import run_geval

result = run_geval(
    transcript="Long document text...",
    summary="Summary text...",
    api_key="your-api-key",
    api_name="openai",
    save=False
)

Adding New Evaluation Types

Step 1: Create Evaluator Class

Create a new file in /app/core/Evaluations/:

# my_custom_evaluator.py
from typing import Dict, List, Any, Optional
from loguru import logger

class MyCustomEvaluator:
    """Custom evaluation implementation"""

    async def evaluate(
        self,
        input_data: Dict[str, Any],
        expected: Dict[str, Any],
        config: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Evaluate input against expected output.

        Returns:
            Dict with scores and metadata
        """
        # Your evaluation logic here
        score = self._calculate_score(input_data, expected)

        return {
            "scores": {"custom_metric": score},
            "passed": score >= config.get("threshold", 0.7),
            "avg_score": score,
            "metadata": {"evaluator": "custom"}
        }

    def _calculate_score(self, input_data, expected):
        # Implement scoring logic
        return 0.85

Step 2: Register in Evaluation Runner

Update /app/core/Evaluations/eval_runner.py:

# In _get_evaluation_function method
def _get_evaluation_function(self, eval_type: str, eval_spec: Dict[str, Any]) -> Callable:
    if eval_type == "model_graded":
        sub_type = eval_spec.get("sub_type")
        # ... existing code ...
    elif eval_type == "custom_type":  # Add your type
        return self._eval_custom
    # ... rest of the code ...

# Add evaluation method
async def _eval_custom(
    self,
    sample: Dict[str, Any],
    eval_spec: Dict[str, Any],
    config: Dict[str, Any],
    sample_id: str
) -> Dict[str, Any]:
    """Custom evaluation implementation"""
    try:
        from .my_custom_evaluator import MyCustomEvaluator
        evaluator = MyCustomEvaluator()

        result = await evaluator.evaluate(
            input_data=sample["input"],
            expected=sample.get("expected", {}),
            config=eval_spec
        )

        return {
            "sample_id": sample_id,
            **result
        }
    except Exception as e:
        logger.error(f"Custom eval failed for {sample_id}: {e}")
        return {"sample_id": sample_id, "error": str(e)}

Step 3: Update API Schema (Optional)

If your evaluation needs special parameters, update unified schemas:

# In evaluation_schemas_unified.py
from pydantic import BaseModel
from typing import Optional

# Extend the unified spec with custom knobs
class EvaluationSpec(BaseModel):
    # existing fields ...
    custom_param: Optional[str] = None
    another_param: Optional[int] = 10

# If introducing a new top-level type, add it to EvaluationType enum
# class EvaluationType(str, Enum):
#     CUSTOM = "custom"

Extending the Evaluation Runner

Adding Progress Callbacks

class ExtendedEvaluationRunner(EvaluationRunner):
    def __init__(self, db_path: str, progress_callback=None):
        super().__init__(db_path)
        self.progress_callback = progress_callback

    async def _process_batch(self, batch, eval_fn, eval_spec, eval_config, max_workers):
        results = await super()._process_batch(
            batch, eval_fn, eval_spec, eval_config, max_workers
        )

        # Call progress callback
        if self.progress_callback:
            await self.progress_callback(len(results))

        return results

Custom Result Aggregation

def _calculate_custom_aggregate(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Custom aggregation logic"""
    # Calculate percentiles
    import numpy as np
    scores = [r["avg_score"] for r in results if "avg_score" in r]

    return {
        "mean_score": np.mean(scores),
        "median_score": np.median(scores),
        "p25": np.percentile(scores, 25),
        "p75": np.percentile(scores, 75),
        "p95": np.percentile(scores, 95)
    }

Integration Points

With LLM Module

from tldw_Server_API.app.core.LLM_Calls.Summarization_General_Lib import analyze
from tldw_Server_API.app.core.Chat.chat_orchestrator import chat_api_call

# Use for model-graded evaluations
response = await analyze(
    input_data=text,
    custom_prompt=evaluation_prompt,
    api_name="openai",
    api_key=api_key,
    temp=0.0,
    system_message="You are an evaluation expert."
)

With Embeddings Module (Future)

# When embeddings module is ready
from tldw_Server_API.app.core.Embeddings import EmbeddingsServiceWrapper

class EmbeddingEvaluator:
    def __init__(self):
        self.embeddings = EmbeddingsServiceWrapper()

    async def calculate_similarity(self, text1: str, text2: str) -> float:
        emb1 = await self.embeddings.get_embedding(text1)
        emb2 = await self.embeddings.get_embedding(text2)
        return cosine_similarity(emb1, emb2)

With Media Processing

from tldw_Server_API.app.core.Ingestion_Media_Processing import process_media

# Evaluate transcription quality
async def evaluate_transcription(media_file: str):
    # Process media
    transcription = await process_media(media_file)

    # Evaluate against ground truth
    eval_result = await evaluator.evaluate(
        transcription=transcription,
        ground_truth=reference_text
    )
    return eval_result

Testing

Unit Testing Evaluators

import pytest
from unittest.mock import Mock, patch

@pytest.mark.asyncio
async def test_custom_evaluator():
    from tldw_Server_API.app.core.Evaluations.my_custom_evaluator import MyCustomEvaluator

    evaluator = MyCustomEvaluator()
    result = await evaluator.evaluate(
        input_data={"text": "test"},
        expected={"output": "expected"},
        config={"threshold": 0.5}
    )

    assert "scores" in result
    assert result["passed"] == True
    assert result["avg_score"] >= 0.5

Integration Testing

@pytest.mark.asyncio
async def test_evaluation_workflow():
    from fastapi.testclient import TestClient
    from tldw_Server_API.app.main import app

    client = TestClient(app)

    # Create evaluation (inline dataset)
    resp = client.post("/api/v1/evaluations", json={
        "name": "test_eval",
        "eval_type": "exact_match",
        "eval_spec": {"metrics": ["exact_match"]},
        "dataset": [{"input": {"text": "test"}, "expected": {"text": "test"}}]
    })
    assert resp.status_code in (200, 201)
    eval_id = resp.json()["id"]

    # Start a run
    resp = client.post(f"/api/v1/evaluations/{eval_id}/runs", json={
        "target_model": "test",
        "config": {"temperature": 0}
    })
    assert resp.status_code in (200, 202)
    run_id = resp.json()["id"]

    # Poll run (results included on completion)
    resp = client.get(f"/api/v1/evaluations/runs/{run_id}")
    assert resp.status_code == 200

Mocking External Services

@patch('tldw_Server_API.app.core.Chat.chat_orchestrator.chat_api_call')
async def test_geval_with_mock(mock_chat):
    """Test G-Eval with mocked LLM"""
    mock_chat.return_value = "4.5"  # Mock score response

    from tldw_Server_API.app.core.Evaluations.ms_g_eval import run_geval

    result = run_geval(
        transcript="Test document",
        summary="Test summary",
        api_key="mock-key",
        api_name="openai",
        save=False
    )

    assert "Coherence:" in result
    mock_chat.assert_called()

Performance Considerations

Database Optimization

# Use transactions for bulk operations
with db.get_connection() as conn:
    cursor = conn.cursor()
    cursor.executemany(
        "INSERT INTO evaluation_runs ...",
        batch_data
    )
    conn.commit()

# Add indexes for common queries
CREATE INDEX idx_runs_eval_status ON evaluation_runs(eval_id, status);
CREATE INDEX idx_runs_created ON evaluation_runs(created_at DESC);

Async Processing

# Process evaluations concurrently
async def process_evaluations_batch(evaluations: List[Dict]):
    tasks = [
        evaluate_single(eval_data)
        for eval_data in evaluations
    ]

    # Limit concurrency
    semaphore = asyncio.Semaphore(10)
    async def bounded_evaluate(eval_data):
        async with semaphore:
            return await evaluate_single(eval_data)

    results = await asyncio.gather(
        *[bounded_evaluate(e) for e in evaluations]
    )
    return results

Caching Strategy

from functools import lru_cache
from typing import Tuple

class CachedEvaluator:
    @lru_cache(maxsize=1000)
    def _get_cached_score(self, input_hash: str) -> float:
        """Cache evaluation results for identical inputs"""
        return self._calculate_score(input_hash)

    async def evaluate(self, input_data: Dict) -> Dict:
        # Create deterministic hash
        input_hash = self._hash_input(input_data)

        # Check cache
        if cached_score := self._get_cached_score(input_hash):
            return {"score": cached_score, "cached": True}

        # Calculate and cache
        score = await self._calculate_score_async(input_data)
        return {"score": score, "cached": False}

Memory Management

# Stream large datasets
async def stream_dataset(dataset_id: str):
    """Stream dataset samples instead of loading all at once"""
    offset = 0
    batch_size = 100

    while True:
        samples = db.get_dataset_samples(
            dataset_id,
            offset=offset,
            limit=batch_size
        )
        if not samples:
            break

        for sample in samples:
            yield sample

        offset += batch_size

# Process with streaming
async for sample in stream_dataset(dataset_id):
    result = await evaluate_sample(sample)
    await store_result(result)

Configuration Best Practices

Environment-Specific Settings

# config.py
from pydantic import BaseSettings

class EvaluationSettings(BaseSettings):
    max_workers: int = 4
    default_timeout: int = 300
    batch_size: int = 10
    cache_ttl: int = 3600
    max_retries: int = 3

    class Config:
        env_prefix = "EVAL_"

settings = EvaluationSettings()

Feature Flags

# Enable/disable features dynamically
FEATURE_FLAGS = {
    "enable_embeddings": False,  # Until embeddings module ready
    "enable_caching": True,
    "enable_webhooks": True,
    "enable_streaming": True
}

def is_feature_enabled(feature: str) -> bool:
    return FEATURE_FLAGS.get(feature, False)

# Usage
if is_feature_enabled("enable_embeddings"):
    similarity = await calculate_embedding_similarity(text1, text2)
else:
    similarity = calculate_text_similarity(text1, text2)

Debugging and Monitoring

Logging Best Practices

from loguru import logger

# Structured logging
logger.info(
    "Evaluation started",
    eval_id=eval_id,
    run_id=run_id,
    samples=len(dataset),
    config=config
)

# Performance logging
with logger.contextualize(run_id=run_id):
    start = time.time()
    result = await evaluate()
    logger.info(f"Evaluation completed in {time.time()-start:.2f}s")

Error Tracking

class EvaluationError(Exception):
    """Base exception for evaluation errors"""
    pass

class EvaluatorNotFoundError(EvaluationError):
    """Raised when evaluator type is not found"""
    pass

class EvaluationTimeoutError(EvaluationError):
    """Raised when evaluation times out"""
    pass

# Usage with context
try:
    result = await evaluator.evaluate(sample)
except EvaluationTimeoutError as e:
    logger.error(f"Evaluation timeout: {e}", sample_id=sample_id)
    # Store partial results
    db.update_run_status(run_id, "partial", error=str(e))

Migration Guide

From Legacy to Unified Evaluations API

# Old API call
from tldw_Server_API.app.core.Evaluations.evaluation_manager import EvaluationManager

manager = EvaluationManager()
eval_id = await manager.store_evaluation(
    evaluation_type="geval",
    input_data={"source": text, "summary": summary},
    results=results
)

# New API call (unified service)
from tldw_Server_API.app.core.Evaluations.unified_evaluation_service import UnifiedEvaluationService

svc = UnifiedEvaluationService()
evaluation = await svc.create_evaluation(
    name="geval_summary",
    eval_type="model_graded",
    eval_spec={"sub_type": "summarization", "model": "gpt-4"}
)

Future Enhancements

Planned Features

1. Embeddings Integration: Full vector similarity support
2. Custom Metrics UI: Web interface for creating evaluations
3. Evaluation Templates: Pre-built evaluation configurations
4. Comparison Tools: A/B testing framework
5. Export Formats: JSON, CSV, Markdown reports
6. Scheduling: Automated periodic evaluations
7. Notifications: Email/Slack integration for results

Extension Points

• Custom storage backends (PostgreSQL, MongoDB)
• Alternative evaluation frameworks (HELM, BigBench)
• Multi-language support for evaluations
• Distributed evaluation processing
• Real-time collaboration features

Resources

• Unified API Reference
• User Guide
• OpenAI Evals - Compatible format
• Test Suite - Example implementations