Function check_embedding

src/cocoindex_code/shared.py:61–80

Run a single embed call against *embedder* and report dim or error. *params* are spread into ``embed()`` so callers can verify indexing vs query params separately (they may use different keys at runtime). Never raises. Used by the daemon's doctor path (`daemon._check_model`).

(
    embedder: Embedder,
    params: dict[str, Any] | None = None,
)

async def check_embedding(
    embedder: Embedder,
    params: dict[str, Any] | None = None,
) -> EmbeddingCheckResult:
    """Run a single embed call against *embedder* and report dim or error.

    *params* are spread into ``embed()`` so callers can verify indexing vs
    query params separately (they may use different keys at runtime).

    Never raises. Used by the daemon's doctor path (`daemon._check_model`).
    """
    kwargs = dict(params) if params else {}
    try:
        vec = await embedder.embed("hello world", **kwargs)
        return EmbeddingCheckResult(dim=len(vec), error=None)
    except Exception as e:
        msg = " ".join(f"{type(e).__name__}: {e}".splitlines())
        if len(msg) > 500:
            msg = msg[:500] + "…"
        return EmbeddingCheckResult(dim=None, error=msg, traceback=_tb.format_exc())


def create_embedder(