Function _probe_embedding_provider

openviking_cli/doctor.py:198–268

Create the configured embedder, make one request, and validate vector shape.

(
    embedding: dict[str, Any],
    dense: dict[str, Any],
)

def _probe_embedding_provider(
    embedding: dict[str, Any],
    dense: dict[str, Any],
) -> tuple[bool, str, Optional[str]]:
    """Create the configured embedder, make one request, and validate vector shape."""
    from openviking.models.embedder.base import embed_compat
    from openviking_cli.utils.config.embedding_config import EmbeddingConfig, EmbeddingModelConfig

    provider = dense.get("provider", "local")
    model = dense.get("model", "bge-small-zh-v1.5-f16")
    label = _embedding_probe_label(provider, model, dense)

    try:
        model_config = EmbeddingModelConfig(**dense)
        config_kwargs = {
            key: embedding[key]
            for key in ("max_concurrent", "max_retries", "text_source", "max_input_tokens")
            if key in embedding
        }
        embedding_config = EmbeddingConfig(dense=model_config, **config_kwargs)
        embedder = embedding_config.get_embedder()
        expected_dimension = (
            model_config.dimension
            if model_config.dimension is not None
            else embedder.get_dimension()
        )
    except Exception as exc:
        return (
            False,
            f"{label} (invalid embedding config: {exc})",
            "Fix embedding.dense provider/model/api_base/dimension in ov.conf",
        )

    async def _run_probe():
        return await asyncio.wait_for(
            embed_compat(embedder, "OpenViking doctor embedding probe", is_query=True),
            timeout=10.0,
        )

    try:
        result = asyncio.run(_run_probe())
    except TimeoutError:
        return (
            False,
            f"{label} (probe timed out)",
            "Check embedding.dense.api_base and make sure the embedding provider is reachable",
        )
    except Exception as exc:
        return (
            False,
            f"{label} (probe failed: {exc})",
            "Check the embedding provider service, model name, API key, and api_base",
        )

    vector = getattr(result, "dense_vector", None)
    if not vector:
        return (
            False,