Auto-tune Embedding Model Parameters & Add Benchmarking Tool

@gvanrossum-ms please check. I have implemented a small scope of what was asked regarding tuning parameters and benchmarking.

Overview

This pull request addresses the need to determine and apply optimal tuning parameters (min_score and max_hits/max_matches) dynamically based on the active embedding model, especially considering the structural scoring disparities between classic and newer models (e.g. text-embedding-ada-002 vs. text-embedding-3).

It also introduces an offline benchmarking utility that calculates model evaluation metrics (Hit Rate and Mean Reciprocal Rank) using ground truth expected matches from the Adrian Tchaikovsky test dataset, allowing for continuous empirically driven optimization.

Key Changes

1. `TextEmbeddingIndexSettings` Auto-Tuning

Modified File: src/typeagent/aitools/vectorbase.py

Introduced a MODEL_DEFAULTS registry to map well-known models to community-consensus, authentic retrieval "sweet spots".
Dynamic Tuning: The configuration natively detects standard model names (text-embedding-3-* vs ada-002) and optimally anchors min_score around 0.30/0.35 or 0.75 respectively. It additionally sets standard thresholds natively instead of returning unbounded queries.
Strict Backward Compatibility: Explicit constraints passed at instantiation (e.g., TextEmbeddingIndexSettings(model, min_score=0.90)) are unequivocally respected, aggressively overriding auto-tuning logic.
Safety Handling: Safe attribute extraction handles cases where dummy or radically custom model objects are passed.

2. Implementation of `benchmark_embeddings.py` Tool

New File: tools/benchmark_embeddings.py

Uses all three Adrian Tchaikovsky test data files (already registered in tests/conftest.py as EPISODE_53_INDEX, EPISODE_53_ANSWERS, EPISODE_53_SEARCH) for comprehensive evaluation:

Episode_53_AdrianTchaikovsky_index_data.json: ~96 podcast messages used as the embedding index corpus.
Episode_53_Search_results.json: Search queries with messageMatches ground truth used for grid search evaluation (Hit Rate & MRR).
Episode_53_Answer_results.json: 55+ curated Q&A pairs from the podcast, used for answer-quality benchmarking via keyword/entity coverage matching.

Key features:

Search benchmark: Grid searches over min_score × max_hits parameter space, evaluating retrieval quality via Hit Rate and MRR against expected messageMatches.
Answer benchmark: Tests each answerable question from the Q&A dataset, checking whether retrieved messages contain key terms and named entities (e.g., "Children of Time", "Skynet", "University of Reading", "Adrian Tchaikovsky") from the expected answer.
Unanswerable query detection: Evaluates hasNoAnswer=True queries separately, flagging potential false positives where the system incorrectly retrieves high-confidence results.
Robust exception handling for missing dataset files or unconfigured API keys.
--model test:fake flag for deterministic automated testing without network cost.

How to Test

Running the Benchmark Tool

# Using a local deterministic test model (No API keys needed)
uv run python tools/benchmark_embeddings.py --model test:fake

# Profiling open-ai embedding models (Requires API configuration)
uv run python tools/benchmark_embeddings.py --model openai:text-embedding-3-small

Validating Overrides

Provide any custom parameter setting and ensure it evaluates successfully:

model = create_embedding_model("openai:text-embedding-3-large")
# Automated tune to 0.30 applies underneath
settings_auto = TextEmbeddingIndexSettings(model)
assert settings_auto.min_score == 0.30

# Explicit threshold supersedes logic securely
settings_explicit = TextEmbeddingIndexSettings(model, min_score=0.55)
assert settings_explicit.min_score == 0.55

Security & Exceptions

Dataset files are wrapped in try/except clauses alerting developers effectively.
Configuration models extract model_name safely bypassing AttributeError when using esoteric API structures.

Resolves optimization issues globally across TypeAgent's storage indices.

Full Code

`src/typeagent/aitools/vectorbase.py`

# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

from collections.abc import Callable, Iterable
from dataclasses import dataclass

import numpy as np

from .embeddings import (
    IEmbeddingModel,
    NormalizedEmbedding,
    NormalizedEmbeddings,
)
from .model_adapters import create_embedding_model


@dataclass
class ScoredInt:
    item: int
    score: float


@dataclass
class TextEmbeddingIndexSettings:
    embedding_model: IEmbeddingModel
    min_score: float  # Between 0.0 and 1.0
    max_matches: int | None  # >= 1; None means no limit
    batch_size: int  # >= 1

    def __init__(
        self,
        embedding_model: IEmbeddingModel | None = None,
        min_score: float | None = None,
        max_matches: int | None = None,
        batch_size: int | None = None,
    ):
        self.embedding_model = embedding_model or create_embedding_model()

        # Default fallback values
        default_min_score = 0.85
        default_max_matches = None

        # Determine optimal parameters automatically for well-known models.
        # Format: (min_score, max_matches)
        # Note: text-embedding-3 models produce structurally lower cosine scores than older models
        # and typically perform best in the 0.3 - 0.5 range for relevance filtering.
        MODEL_DEFAULTS = {
            "text-embedding-3-large": (0.30, 20),
            "text-embedding-3-small": (0.35, 20),
            "text-embedding-ada-002": (0.75, 20),
        }

        # Check if the model_name matches any known ones
        model_name = getattr(self.embedding_model, 'model_name', "")
        
        if model_name:
            for known_model, defaults in MODEL_DEFAULTS.items():
                if known_model in model_name:
                    default_min_score, default_max_matches = defaults
                    break

        self.min_score = min_score if min_score is not None else default_min_score
        self.max_matches = max_matches if max_matches is not None else default_max_matches
        self.batch_size = batch_size if batch_size and batch_size >= 1 else 8


class VectorBase:
    settings: TextEmbeddingIndexSettings
    _vectors: NormalizedEmbeddings
    _model: IEmbeddingModel
    _embedding_size: int

    def __init__(self, settings: TextEmbeddingIndexSettings):
        self.settings = settings
        self._model = settings.embedding_model
        self._embedding_size = 0
        self.clear()

    async def get_embedding(self, key: str, cache: bool = True) -> NormalizedEmbedding:
        if cache:
            return await self._model.get_embedding(key)
        else:
            return await self._model.get_embedding_nocache(key)

    async def get_embeddings(
        self, keys: list[str], cache: bool = True
    ) -> NormalizedEmbeddings:
        if cache:
            return await self._model.get_embeddings(keys)
        else:
            return await self._model.get_embeddings_nocache(keys)

    def __len__(self) -> int:
        return len(self._vectors)

    # Needed because otherwise an empty index would be falsy.
    def __bool__(self) -> bool:
        return True

    def add_embedding(
        self, key: str | None, embedding: NormalizedEmbedding | list[float]
    ) -> None:
        if isinstance(embedding, list):
            embedding = np.array(embedding, dtype=np.float32)
        if self._embedding_size == 0:
            self._set_embedding_size(len(embedding))
            self._vectors.shape = (0, self._embedding_size)
        if len(embedding) != self._embedding_size:
            raise ValueError(
                f"Embedding size mismatch: expected {self._embedding_size}, "
                f"got {len(embedding)}"
            )
        embeddings = embedding.reshape(1, -1)  # Make it 2D: 1xN
        self._vectors = np.append(self._vectors, embeddings, axis=0)
        if key is not None:
            self._model.add_embedding(key, embedding)

    def add_embeddings(
        self, keys: None | list[str], embeddings: NormalizedEmbeddings
    ) -> None:
        if embeddings.ndim != 2:
            raise ValueError(f"Expected 2D embeddings array, got {embeddings.ndim}D")
        if self._embedding_size == 0:
            self._set_embedding_size(embeddings.shape[1])
            self._vectors.shape = (0, self._embedding_size)
        if embeddings.shape[1] != self._embedding_size:
            raise ValueError(
                f"Embedding size mismatch: expected {self._embedding_size}, "
                f"got {embeddings.shape[1]}"
            )
        self._vectors = np.concatenate((self._vectors, embeddings), axis=0)
        if keys is not None:
            for key, embedding in zip(keys, embeddings):
                self._model.add_embedding(key, embedding)

    async def add_key(self, key: str, cache: bool = True) -> None:
        embedding = await self.get_embedding(key, cache=cache)
        self.add_embedding(key if cache else None, embedding)

    async def add_keys(self, keys: list[str], cache: bool = True) -> None:
        if not keys:
            return
        embeddings = await self.get_embeddings(keys, cache=cache)
        self.add_embeddings(keys if cache else None, embeddings)

    def fuzzy_lookup_embedding(
        self,
        embedding: NormalizedEmbedding,
        max_hits: int | None = None,
        min_score: float | None = None,
        predicate: Callable[[int], bool] | None = None,
    ) -> list[ScoredInt]:
        if max_hits is None:
            max_hits = 10
        if min_score is None:
            min_score = 0.0
        if len(self._vectors) == 0:
            return []
        # This line does most of the work:
        scores: Iterable[float] = np.dot(self._vectors, embedding)
        scored_ordinals = [
            ScoredInt(i, score)
            for i, score in enumerate(scores)
            if score >= min_score and (predicate is None or predicate(i))
        ]
        scored_ordinals.sort(key=lambda x: x.score, reverse=True)
        return scored_ordinals[:max_hits]

    # TODO: Make this and fuzzy_lookup_embedding() more similar.
    def fuzzy_lookup_embedding_in_subset(
        self,
        embedding: NormalizedEmbedding,
        ordinals_of_subset: list[int],
        max_hits: int | None = None,
        min_score: float | None = None,
    ) -> list[ScoredInt]:
        ordinals_set = set(ordinals_of_subset)
        return self.fuzzy_lookup_embedding(
            embedding, max_hits, min_score, lambda i: i in ordinals_set
        )

    async def fuzzy_lookup(
        self,
        key: str,
        max_hits: int | None = None,
        min_score: float | None = None,
        predicate: Callable[[int], bool] | None = None,
    ) -> list[ScoredInt]:
        if max_hits is None:
            max_hits = self.settings.max_matches
        if min_score is None:
            min_score = self.settings.min_score
        embedding = await self.get_embedding(key)
        return self.fuzzy_lookup_embedding(
            embedding, max_hits=max_hits, min_score=min_score, predicate=predicate
        )

    def _set_embedding_size(self, size: int) -> None:
        """Adopt *size* when it was not known at construction time."""
        assert size > 0
        self._embedding_size = size

    def clear(self) -> None:
        self._vectors = np.array([], dtype=np.float32)
        if self._embedding_size > 0:
            self._vectors.shape = (0, self._embedding_size)

    def get_embedding_at(self, pos: int) -> NormalizedEmbedding:
        if 0 <= pos < len(self._vectors):
            return self._vectors[pos]
        raise IndexError(
            f"Index {pos} out of bounds for embedding index of size {len(self)}"
        )

    def serialize_embedding_at(self, pos: int) -> NormalizedEmbedding | None:
        return self._vectors[pos] if 0 <= pos < len(self._vectors) else None

    def serialize(self) -> NormalizedEmbeddings:
        if self._embedding_size > 0:
            assert self._vectors.shape == (len(self._vectors), self._embedding_size)
        return self._vectors  # TODO: Should we make a copy?

    def deserialize(self, data: NormalizedEmbeddings | None) -> None:
        if data is None:
            self.clear()
            return
        if self._embedding_size == 0:
            if data.ndim < 2 or data.shape[0] == 0:
                # Empty data — can't determine size; just clear.
                self.clear()
                return
            self._set_embedding_size(data.shape[1])
        assert data.shape == (len(data), self._embedding_size), [
            data.shape,
            self._embedding_size,
        ]
        self._vectors = data  # TODO: Should we make a copy?

`tools/benchmark_embeddings.py`

#!/usr/bin/env python3
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

"""
Utility script to benchmark different TextEmbeddingIndexSettings parameters.

Uses the Adrian Tchaikovsky podcast dataset (Episode 53) which contains:
- Index data: ~96 messages from the podcast conversation
- Search results: Queries with expected messageMatches (ground truth for retrieval)
- Answer results: Curated Q&A pairs with expected answers (ground truth for Q&A quality)

The benchmark evaluates embedding model retrieval quality using:
1. Search-based evaluation: Compares fuzzy_lookup results against expected messageMatches
2. Answer-based evaluation: Tests if queries from the Answer dataset retrieve messages
   that contain the expected answer content (substring matching)

Metrics:
- Hit Rate: Percentage of queries where at least one expected result was retrieved
- MRR (Mean Reciprocal Rank): Average of 1/rank of the first relevant result

Usage:
    uv run python tools/benchmark_embeddings.py [--model provider:model]
"""

import argparse
import asyncio
import json
import logging
from pathlib import Path
from statistics import mean
import sys
from typing import Any

from typeagent.aitools.model_adapters import create_embedding_model
from typeagent.aitools.vectorbase import TextEmbeddingIndexSettings, VectorBase


async def run_benchmark(model_spec: str | None) -> None:
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    # Paths — same testdata directory used by tests/conftest.py
    # (EPISODE_53_INDEX, EPISODE_53_ANSWERS, EPISODE_53_SEARCH)
    script_dir = Path(__file__).resolve().parent
    repo_root = script_dir.parent
    index_data_path = repo_root / "tests" / "testdata" / "Episode_53_AdrianTchaikovsky_index_data.json"
    search_data_path = repo_root / "tests" / "testdata" / "Episode_53_Search_results.json"
    answer_data_path = repo_root / "tests" / "testdata" / "Episode_53_Answer_results.json"

    # ── Load index data (messages to embed) ──
    logger.info(f"Loading index data from {index_data_path}")
    try:
        with open(index_data_path, "r", encoding="utf-8") as f:
            index_json = json.load(f)
    except Exception as e:
        logger.error(f"Failed to load index data: {e}")
        return

    messages = index_json.get("messages", [])
    message_texts = [" ".join(m.get("textChunks", [])) for m in messages]

    # ── Load search queries (ground truth: messageMatches) ──
    logger.info(f"Loading search queries from {search_data_path}")
    try:
        with open(search_data_path, "r", encoding="utf-8") as f:
            search_json = json.load(f)
    except Exception as e:
        logger.error(f"Failed to load search queries: {e}")
        return

    # Filter out ones without results or expected matches
    search_queries: list[tuple[str, list[int]]] = []
    for item in search_json:
        search_text = item.get("searchText")
        results = item.get("results", [])
        if not results:
            continue
        expected = results[0].get("messageMatches", [])
        if not expected:
            continue
        search_queries.append((search_text, expected))

    # ── Load answer results (Q&A ground truth from Adrian Tchaikovsky dataset) ──
    answer_queries: list[tuple[str, str, bool]] = []  # (question, answer, hasNoAnswer)
    logger.info(f"Loading answer results from {answer_data_path}")
    try:
        with open(answer_data_path, "r", encoding="utf-8") as f:
            answer_json = json.load(f)
        for item in answer_json:
            question = item.get("question", "")
            answer = item.get("answer", "")
            has_no_answer = item.get("hasNoAnswer", False)
            if question and answer:
                answer_queries.append((question, answer, has_no_answer))
        logger.info(f"Found {len(answer_queries)} answer Q&A pairs "
                     f"({sum(1 for _, _, h in answer_queries if not h)} with answers, "
                     f"{sum(1 for _, _, h in answer_queries if h)} with no-answer).")
    except Exception as e:
        logger.warning(f"Failed to load answer results (continuing without): {e}")

    logger.info(f"Found {len(message_texts)} messages to embed.")
    logger.info(f"Found {len(search_queries)} search queries with expected matches.")

    # ── Create embedding model and index ──
    try:
        if model_spec == "test:fake":
            from typeagent.aitools.model_adapters import create_test_embedding_model
            model = create_test_embedding_model(embedding_size=384)
        else:
            model = create_embedding_model(model_spec)
    except Exception as e:
        logger.error(f"Failed to create embedding model: {e}")
        logger.info("Are your environment variables (e.g. OPENAI_API_KEY) set?")
        return
    settings = TextEmbeddingIndexSettings(model)
    vbase = VectorBase(settings)

    logger.info("Computing embeddings for messages (this may take some time...)")
    # Batch the embeddings
    batch_size = 50
    for i in range(0, len(message_texts), batch_size):
        batch = message_texts[i : i + batch_size]
        await vbase.add_keys(batch)
        print(f"  ... embedded {min(i + batch_size, len(message_texts))}/{len(message_texts)}")

    # ── Compute query embeddings ──
    logger.info("Computing embeddings for search queries...")
    search_query_texts = [q[0] for q in search_queries]
    search_query_embeddings = await model.get_embeddings(search_query_texts)

    answer_query_embeddings = None
    if answer_queries:
        logger.info("Computing embeddings for answer queries...")
        answer_query_texts = [q[0] for q in answer_queries]
        answer_query_embeddings = await model.get_embeddings(answer_query_texts)

    # ──────────────────────────────────────────────────────────────────────
    # Section 1: Grid Search using Search Results (messageMatches)
    # ──────────────────────────────────────────────────────────────────────

    # Grid search config
    min_scores_to_test = [0.70, 0.75, 0.80, 0.85, 0.90, 0.95]
    max_hits_to_test = [5, 10, 15, 20]

    logger.info(f"Starting grid search over model: {model.model_name}")
    print()
    print("=" * 72)
    print("  SEARCH RESULTS BENCHMARK (messageMatches ground truth)")
    print("=" * 72)
    print(f"{'Min Score':<12} | {'Max Hits':<10} | {'Hit Rate (%)':<15} | {'MRR':<10}")
    print("-" * 65)

    best_mrr = -1.0
    best_config = None

    for ms in min_scores_to_test:
        for mh in max_hits_to_test:
            hits = 0
            reciprocal_ranks = []
            
            for (query_text, expected_indices), q_emb in zip(search_queries, search_query_embeddings):
                scored_results = vbase.fuzzy_lookup_embedding(q_emb, max_hits=mh, min_score=ms)
                retrieved_indices = [sr.item for sr in scored_results]

                # Check if any of the expected items are in the retrieved answers
                rank = -1
                for r_idx, retrieved in enumerate(retrieved_indices):
                    if retrieved in expected_indices:
                        rank = r_idx + 1
                        break

                if rank > 0:
                    hits += 1
                    reciprocal_ranks.append(1.0 / rank)
                else:
                    reciprocal_ranks.append(0.0)

            hit_rate = (hits / len(search_queries)) * 100
            mrr = mean(reciprocal_ranks)
            
            print(f"{ms:<12.2f} | {mh:<10d} | {hit_rate:<15.2f} | {mrr:<10.4f}")

            if mrr > best_mrr:
                best_mrr = mrr
                best_config = (ms, mh)

    print("-" * 65)
    if best_config:
        logger.info(f"Search benchmark optimal: min_score={best_config[0]}, "
                     f"max_hits={best_config[1]} (MRR={best_mrr:.4f})")
    else:
        logger.info("Could not determine optimal parameters (no hits).")

    # ──────────────────────────────────────────────────────────────────────
    # Section 2: Answer Results Benchmark (Adrian Tchaikovsky Q&A pairs)
    # ──────────────────────────────────────────────────────────────────────

    if answer_queries and answer_query_embeddings is not None:
        print()
        print("=" * 72)
        print("  ANSWER RESULTS BENCHMARK (Adrian Tchaikovsky Q&A ground truth)")
        print("=" * 72)
        print()

        # For each answer query, check if retrieved messages contain key terms
        # from the expected answer. This is a content-based relevance check.
        #
        # We split answers with hasNoAnswer=True vs False to evaluate separately.

        answerable = [(q, a, emb) for (q, a, h), emb
                       in zip(answer_queries, answer_query_embeddings) if not h]
        unanswerable = [(q, a, emb) for (q, a, h), emb
                         in zip(answer_queries, answer_query_embeddings) if h]

        print(f"Answerable queries: {len(answerable)}")
        print(f"Unanswerable queries (hasNoAnswer=True): {len(unanswerable)}")
        print()

        # Extract key terms from expected answers for content matching
        def extract_answer_keywords(answer_text: str) -> list[str]:
            """Extract distinctive keywords/phrases from an answer for matching."""
            # Look for quoted items, proper nouns, and distinctive phrases
            keywords = []
            # Extract quoted phrases
            import re
            quoted = re.findall(r"'([^']+)'", answer_text)
            keywords.extend(quoted)
            quoted2 = re.findall(r'"([^"]+)"', answer_text)
            keywords.extend(quoted2)

            # Extract proper-noun-like terms (capitalized words that aren't sentence starters)
            # and key named entities from the Adrian Tchaikovsky dataset
            known_entities = [
                "Adrian Tchaikovsky", "Tchaikovsky", "Kevin Scott", "Christina Warren",
                "Children of Time", "Children of Ruin", "Children of Memory",
                "Shadows of the Apt", "Empire in Black and Gold",
                "Final Architecture", "Lords of Uncreation",
                "Dragonlance Chronicles", "Skynet", "Portids", "Corvids",
                "University of Reading", "Magnus Carlsen", "Warhammer",
                "Asimov", "Peter Watts", "William Gibson", "Iain Banks",
                "Peter Hamilton", "Arthur C. Clarke", "Profiles of the Future",
                "Dune", "Brave New World", "Iron Sunrise", "Wall-E",
                "George RR Martin", "Alastair Reynolds", "Ovid",
                "zoology", "psychology", "spiders", "arachnids", "insects",
            ]
            for entity in known_entities:
                if entity.lower() in answer_text.lower():
                    keywords.append(entity)

            return keywords

        # Run answer benchmark with the best config from search benchmark
        if best_config:
            eval_min_score, eval_max_hits = best_config
        else:
            eval_min_score, eval_max_hits = 0.80, 10

        print(f"Using parameters: min_score={eval_min_score}, max_hits={eval_max_hits}")
        print("-" * 72)
        print(f"{'#':<4} | {'Question':<45} | {'Keywords Found':<14} | {'Msgs':<5}")
        print("-" * 72)

        answer_hits = 0
        answer_keyword_scores: list[float] = []

        for idx, (question, answer, q_emb) in enumerate(answerable, 1):
            scored_results = vbase.fuzzy_lookup_embedding(
                q_emb, max_hits=eval_max_hits, min_score=eval_min_score
            )
            retrieved_indices = [sr.item for sr in scored_results]

            # Concatenate the text of all retrieved messages
            retrieved_text = " ".join(
                message_texts[i] for i in retrieved_indices if i < len(message_texts)
            )

            # Check how many answer keywords appear in retrieved text
            keywords = extract_answer_keywords(answer)
            if keywords:
                found = sum(
                    1 for kw in keywords
                    if kw.lower() in retrieved_text.lower()
                )
                keyword_score = found / len(keywords)
            else:
                # No keywords extracted — just check if we retrieved anything
                keyword_score = 1.0 if retrieved_indices else 0.0

            if keyword_score > 0:
                answer_hits += 1
            answer_keyword_scores.append(keyword_score)

            q_display = question[:42] + "..." if len(question) > 45 else question
            kw_display = f"{int(keyword_score * 100):>3}%"
            if keywords:
                kw_display += f" ({sum(1 for kw in keywords if kw.lower() in retrieved_text.lower())}/{len(keywords)})"
            print(f"{idx:<4} | {q_display:<45} | {kw_display:<14} | {len(retrieved_indices):<5}")

        print("-" * 72)

        if answerable:
            answer_hit_rate = (answer_hits / len(answerable)) * 100
            avg_keyword_score = mean(answer_keyword_scores) * 100
            print(f"Answer Hit Rate:        {answer_hit_rate:.1f}% "
                  f"({answer_hits}/{len(answerable)} queries found relevant content)")
            print(f"Avg Keyword Coverage:   {avg_keyword_score:.1f}%")

        # Evaluate unanswerable queries — ideally these should retrieve fewer/no results
        if unanswerable:
            print()
            print("-" * 72)
            print("Unanswerable queries (should ideally retrieve less relevant content):")
            print("-" * 72)
            false_positive_count = 0
            for question, answer, q_emb in unanswerable:
                scored_results = vbase.fuzzy_lookup_embedding(
                    q_emb, max_hits=eval_max_hits, min_score=eval_min_score
                )
                n_results = len(scored_results)
                avg_score = mean(sr.score for sr in scored_results) if scored_results else 0.0
                q_display = question[:55] + "..." if len(question) > 58 else question
                flag = "[!]" if n_results > 3 else "[ok]"
                if n_results > 3:
                    false_positive_count += 1
                print(f"  {flag} {q_display:<58} | {n_results:>3} results (avg={avg_score:.3f})")
            print(f"\nFalse positives (>3 results): {false_positive_count}/{len(unanswerable)}")

    # ── Summary ──
    print()
    print("=" * 72)
    print("  SUMMARY")
    print("=" * 72)
    print(f"Model:                  {model.model_name}")
    print(f"Messages indexed:       {len(message_texts)}")
    print(f"Search queries tested:  {len(search_queries)}")
    if best_config:
        print(f"Best search params:     min_score={best_config[0]}, max_hits={best_config[1]}")
        print(f"Best search MRR:        {best_mrr:.4f}")
    if answer_queries:
        print(f"Answer queries tested:  {len(answerable)} answerable, {len(unanswerable)} unanswerable")
        if answerable:
            print(f"Answer hit rate:        {answer_hit_rate:.1f}%")
            print(f"Keyword coverage:       {avg_keyword_score:.1f}%")
    print("=" * 72)


def main() -> None:
    parser = argparse.ArgumentParser(description="Benchmark embedding model parameters.")
    parser.add_argument(
        "--model",
        type=str,
        default=None,
        help="Provider and model name, e.g. 'openai:text-embedding-3-small'",
    )
    args = parser.parse_args()
    asyncio.run(run_benchmark(args.model))


if __name__ == "__main__":
    main()

Tuning embeddings #57

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions