qmd_py.llm.client¶

async def list_models(self) -> list[str]:
    """Model ids the router currently has presets for - backs `doctor`'s
    check that the configured embed/generate/rerank models actually
    exist on the router, not just in qmd-py's own settings."""
    response = await self._client.get("/v1/models")
    response.raise_for_status()
    return [item["id"] for item in response.json()["data"]]

async def embed(self, texts: list[str], model: str) -> list[list[float]]:
    """Embed a batch of texts in one request.

    Returns:
        Vectors positionally matching `texts`. The router keys its
        response by `index` rather than guaranteeing order, so the
        results are re-sorted here - without that, every chunk in a
        batch could get another chunk's vector.

    Raises:
        httpx.HTTPStatusError: The router returned an error status.
        KeyError: The response had no `data` array.
    """
    response = await self._client.post(
        "/v1/embeddings", json={"model": model, "input": texts}
    )
    response.raise_for_status()
    data = response.json()["data"]
    return [item["embedding"] for item in sorted(data, key=lambda d: d["index"])]

async def chat_json(
    self,
    messages: list[dict[str, str]],
    model: str,
    json_schema: dict[str, Any],
    max_tokens: int = 300,
    temperature: float = 0.7,
) -> dict[str, Any]:
    """One chat completion constrained to `json_schema` via the
    router's `response_format` (an OpenAI-compatible, increasingly
    standard llama.cpp server feature) - returns the parsed JSON
    object. Used for query expansion (see search/hybrid.py)."""
    response = await self._client.post(
        "/v1/chat/completions",
        json={
            "model": model,
            "messages": messages,
            "response_format": {"type": "json_schema", "json_schema": json_schema},
            "max_tokens": max_tokens,
            "temperature": temperature,
        },
    )
    response.raise_for_status()
    content: dict[str, Any] = json.loads(response.json()["choices"][0]["message"]["content"])
    return content

async def rerank(self, query: str, documents: list[str], model: str) -> list[float]:
    """Relevance scores in the same order as `documents` - the
    router's response is keyed by `index` rather than guaranteed to
    preserve input order, so results are re-sorted into place here.

    Always returns one score per document (unscored ones stay 0.0);
    callers rely on that length, hybrid.py zipping it against its
    candidate chunks with `strict=True`. Out-of-range indices are
    dropped rather than assigned: a stray large index would raise
    IndexError, and - worse, because it fails silently - a negative
    one would write onto the wrong document from the end.
    """
    response = await self._client.post(
        "/rerank", json={"model": model, "query": query, "documents": documents}
    )
    response.raise_for_status()
    results = response.json()["results"]
    scores = [0.0] * len(documents)
    for r in results:
        index = r["index"]
        if 0 <= index < len(scores):
            scores[index] = r["relevance_score"]
    return scores

async def tokenize(self, text: str, model: str) -> list[int]:
    """Count tokens using the model's own tokenizer.

    Used to fit text to the reranker's per-pair budget, where a
    chars-per-token estimate proved unsafe for dense code.

    Returns:
        Token ids; callers generally only need the length.

    Raises:
        httpx.HTTPStatusError: The router returned an error status.
    """
    response = await self._client.post("/tokenize", json={"model": model, "content": text})
    response.raise_for_status()
    tokens: list[int] = response.json()["tokens"]
    return tokens

async def aclose(self) -> None:
    """Close the underlying HTTP connection pool.

    Prefer the async context manager. Once closed the client cannot
    be reused - further calls raise `RuntimeError`.
    """
    await self._client.aclose()