LLM clients¶

A large language model takes text in and produces text out. Every provider wraps that in its own SDK, its own request shape and its own way of reporting what a call cost. Kaval.AI puts one small async interface in front of all of them, so switching providers is a change of one string.

Three methods do the work:

Each takes an optional response_model — a Pydantic model — to get a validated object back instead of prose you have to parse.

The clients are standalone: use them directly for a single call, or let a workflow build them for you.

Setup¶

A client needs an API key, read from the environment: OPENAI_API_KEY, GEMINI_API_KEY, ANTHROPIC_API_KEY, or OLLAMA_HOST for a local Ollama. This notebook lives in notebooks/, so it loads the project .env next door.

Kaval.AI logs every model call through loguru. That is useful in an application and noisy in a tutorial, so the setup cell lowers it to warnings and the relevant values are printed explicitly.

import sys

import dotenv
from loguru import logger

dotenv.load_dotenv("../.env")

logger.remove()
_ = logger.add(sys.stderr, level="WARNING")

Your first call¶

make_client builds a client from a "provider/model" id and picks up the matching API key on its own.

Note what prompt returns: a plain str. Nothing is wrapped, and there is no response envelope to dig through.

from kavalai import make_client

client = make_client("openai/gpt-5.6-luna")

answer = await client.prompt(
    "What is the capital of Estonia? Answer in one sentence."
)
print(answer)

The capital of Estonia is Tallinn.

The same code, a different provider¶

The provider lives entirely in that id string. Swap it and nothing else changes — which is what makes it cheap to compare models, or to fall back to a second provider when the first one is unavailable.

gemini = make_client("gemini/gemini-3.1-flash-lite")

print(await gemini.prompt(
    "What is the capital of Estonia? Answer in one sentence."
))

The capital of Estonia is Tallinn.

Structured output¶

Prose is a poor interface for a program. The moment you need a field — a price, a category, a date — you are writing a parser against text that changes wording between calls.

Pass a response_model instead and you get a validated instance of your own Pydantic type. The model is constrained to that schema, and the result is checked before you ever see it.

import textwrap

from pydantic import BaseModel, Field


class City(BaseModel):
    name: str
    country: str
    population: int = Field(description="Approximate number of residents.")
    fun_fact: str


city = await client.prompt("Describe Tallinn.", response_model=City)

print(type(city).__name__)
print("country   :", city.country)
print("population:", city.population)
print("fun fact  :", textwrap.fill(city.fun_fact, 80,
                                   subsequent_indent=" " * 12))

City
country   : Estonia
population: 460000
fun fact  : Tallinn’s remarkably well-preserved medieval Old Town is a UNESCO World Heritage
            site, and the city is widely regarded as one of Europe’s most
            digitally advanced capitals.

city.population is an int, not the string "about 460,000". That is the purpose: you declare the shape once, and every field arrives typed.

Structured output is not provider-specific either — the same City model works against Gemini:

city_gemini = await gemini.prompt("Describe Tallinn.", response_model=City)
print(f"{city_gemini.name}, {city_gemini.country} — "
      f"{city_gemini.population:,} residents")

Tallinn, Estonia — 445,000 residents

Nested models and lists work too, so one call can return a whole small document. Here the fictional Green Village gets an itinerary:

class Stop(BaseModel):
    name: str
    why_visit: str


class Itinerary(BaseModel):
    village: str
    stops: list[Stop]
    advice: str


itinerary = await client.prompt(
    "Plan a half-day visit to Green Village, a tiny village of 104 residents "
    "with a 23-metre grain tower, a pond called Lake Miller and one pub, "
    "The Rusty Anchor. Suggest three stops.",
    response_model=Itinerary,
)

import textwrap

for stop in itinerary.stops:
    print(textwrap.fill(f"- {stop.name}: {stop.why_visit}", 80,
                        subsequent_indent="  "))
print("\nAdvice:", textwrap.fill(itinerary.advice, 80))

- Green Village Grain Tower: Begin with the village’s landmark 23-metre grain
  tower. Allow time to admire its unusual scale and take photographs from around
  its base.
- Lake Miller: Take a peaceful walk beside the village pond, enjoying the water,
  birdlife and quiet countryside atmosphere. This is a good spot for a short
  rest or packed snack.
- The Rusty Anchor: Finish at Green Village’s only pub for a drink and a meal,
  giving you a chance to relax and experience the social hub of this tiny
  community.

Advice: Allow roughly 45–60 minutes at the grain tower, 60–90 minutes around Lake
Miller, and 90 minutes to two hours at The Rusty Anchor. Check the pub’s opening
hours and whether the grain tower can be entered before travelling; with only
104 residents, facilities may be limited.

Conversations¶

prompt sends a single message and keeps no state — like every LLM API, the client is stateless. To hold a conversation you pass the messages so far as a ChatHistory, and the model sees the whole thread.

Roles carry meaning: system sets behaviour, user is the person, assistant is what the model said previously.

from kavalai import ChatHistory, ChatMessage

history = ChatHistory(
    messages=[
        ChatMessage(role="system",
                    content="You are a terse village guide. "
                            "One short sentence."),
        ChatMessage(role="user",
                    content="I'm visiting Green Village tomorrow."),
        ChatMessage(role="assistant",
                    content="Bring boots — Lake Miller is muddy."),
        ChatMessage(role="user", content="What did you just tell me to bring?"),
    ]
)

print(await client.chat_completions(chat_history=history))

I told you to bring boots.

Appending every turn by hand gets tedious. Workflows do it for you: the engine records each turn and replays it on the next one, which is what gives a chatbot memory. See Observability & storage.

Streaming responses¶

A long answer takes seconds to generate. Streaming shows it as it arrives, so the reader sees the first words immediately instead of watching a spinner.

stream_prompt returns a Streamer you iterate. Each chunk has a type — partial while generating, complete at the end.

streamer = await client.stream_prompt(
    "In three sentences, describe autumn in Tallinn's Old Town."
)

chunks = 0
async for chunk in streamer:
    if chunk.type == "partial":
        chunks += 1
    elif chunk.type == "complete":
        print(textwrap.fill(chunk.value, 80))

print(f"\n[arrived in {chunks} partial chunks]")

Autumn paints Tallinn’s Old Town in russet and gold, with leaves swirling along
its cobbled streets. The medieval towers and pastel facades glow softly beneath
cool, misty skies. Cafés spill warmth and the scent of cinnamon into the crisp
air as the city settles into a quieter rhythm.

[arrived in 60 partial chunks]

By default each partial carries the full text accumulated so far, so a UI can render it directly with no reassembly. For long outputs prefer stream_delta, where each chunk carries only what is new. Structured output streams too — every partial is still valid JSON. The Streaming tutorial covers both.

Model statistics and observability¶

Every call reports a ModelCallStat: the model, the prompt and completion token counts, and the wall-clock duration.

Tokens are the units a model reads and writes — roughly ¾ of a word each — and they are what you are billed for. Prompt tokens are what you sent; completion tokens are what came back.

By default these are logged through loguru. Pass your own ModelStatsReceiver to send them anywhere — a metrics backend, a database, or a list you inspect later.

from kavalai import ModelCallStat, ModelStatsReceiver, OpenAIClient


class CollectStats(ModelStatsReceiver):
    def __init__(self):
        self.calls: list[ModelCallStat] = []

    def receive_model_stats(self, stats: ModelCallStat):
        self.calls.append(stats)
        print(
            f"{stats.model}: {stats.prompt_tokens} prompt + "
            f"{stats.completion_tokens} completion = "
            f"{stats.total_tokens} tokens "
            f"in {stats.duration_seconds:.2f}s"
        )


collector = CollectStats()
counted = OpenAIClient("gpt-5.6-luna", model_stats_receiver=collector)

await counted.prompt("Name the three streets of Green Village.")
await counted.prompt("How deep is Lake Miller?")

total = sum(c.total_tokens or 0 for c in collector.calls)
print(f"\ntotal so far: {total} tokens "
      f"over {len(collector.calls)} calls")

openai/gpt-5.6-luna: 14 prompt + 545 completion = 559 tokens in 8.03s

openai/gpt-5.6-luna: 12 prompt + 89 completion = 101 tokens in 3.26s

total so far: 660 tokens over 2 calls

Inside a workflow these are aggregated per run as WorkflowState.token_usage and written to the model_call_stats table, where the backoffice UI shows them per call — see Observability. Multiply the token counts by your provider’s published price to turn them into money; the runtime records usage, not cost.

Timeouts and retries¶

Reliability and sampling live on LlmClientParameters.

timeout_seconds bounds each attempt (default 30s). On transient failures — rate limits, timeouts, dropped connections, 5xx — the client retries with exponential backoff and jitter, up to five retries. It does not retry what you should fix yourself: authentication failures, 404s and other bad requests raise immediately.

from kavalai import LlmClientParameters

careful = OpenAIClient(
    "gpt-5.6-luna",
    llm_client_parameters=LlmClientParameters(
        reasoning_effort="low",   # spend little on thinking
        timeout_seconds=60,       # per attempt
    ),
)

print(await careful.prompt(
    "The official flower of Green Village is the marsh marigold. "
    "Repeat its name and nothing else."
))

marsh marigold

Sampling parameters (temperature, top_p, reasoning_effort, …) default to None and are only sent when you set them, so each provider’s own defaults apply otherwise. That is deliberate: the current OpenAI and Claude reasoning models reject temperature outright, so leaving it unset is what keeps one LlmClientParameters usable across providers. A model that rejects a parameter fails the call with the provider’s own error — the runtime drops nothing.

When you stream, a second timeout guards the gap between chunks: stream_timeout_seconds, defaulting to twice timeout_seconds so a stream survives one timed-out attempt plus its retry backoff. Each retry pushes a restart chunk, telling consumers to discard what they accumulated — the new attempt re-sends from the beginning.

Capping the output¶

max_output_tokens caps what a call may generate. Each client sends it under its provider’s own name — max_output_tokens for OpenAI and Gemini, max_tokens for Anthropic and WebLLM, num_predict for Ollama — and on reasoning models the cap includes the reasoning tokens. Like the sampling parameters it defaults to None, which leaves the provider’s limit in force; Anthropic requires a value and is sent 64,000.

A call that reaches the cap raises OutputTruncatedError rather than returning what was generated. For structured output that text is JSON cut off mid-value, which either fails to parse or, repaired leniently, validates into a model with content missing. The error is not retried, since the same cap cuts the same answer and the retry is billed again, and the truncated call is still reported as a ModelCallStat with its token counts.

from kavalai.llm_clients.base_client import OutputTruncatedError


class Chronicle(BaseModel):
    title: str
    paragraphs: list[str]


capped = OpenAIClient(
    "gpt-5.6-luna",
    llm_client_parameters=LlmClientParameters(
        reasoning_effort="low", max_output_tokens=40
    ),
)

try:
    await capped.prompt(
        "Write the history of Tallinn in five paragraphs.",
        response_model=Chronicle,
    )
except OutputTruncatedError as error:
    print(textwrap.fill(str(error), 78), end="\n\n")
    print(textwrap.fill(f"partial output: {error.partial_output}", 78))

The output of 'openai/gpt-5.6-luna' was cut off at the output cap of 40 tokens
after 40 output tokens (stop reason 'max_output_tokens'). The partial output
is not returned. Raise max_output_tokens (LlmClientParameters, or llm_kwargs
in a workflow) or ask for a shorter answer.

partial output: {"title":"A Brief History of Tallinn","paragraphs":["Tallinn,
the capital of Estonia, lies on the southern shore of the Gulf of Finland. Its
strategic

The partial output stays on the error for diagnosis and never reaches the caller as a Chronicle. A streaming consumer receives the partial chunks that arrived, then the same error; no complete chunk is sent. In a workflow the cap is set per node or per graph through llm_kwargs, or for every call the agent server makes through KAVALAI_LLM_MAX_OUTPUT_TOKENS.

Embeddings¶

An embedding turns text into a list of numbers — a vector — positioned so that similar meanings land near each other. That is what makes semantic search possible: you compare vectors instead of matching words.

make_embedding_client mirrors make_client. fastembed runs locally with no API key, which is why this notebook uses it; openai, gemini, ollama and browser work the same way. Which embedding models each of them offers — and how many dimensions those models produce, which is what a RAG collection is sized on — is in Model providers.

compute_embeddings returns (vectors, stats). With normalize=True the vectors have unit length, so cosine similarity reduces to a dot product.

from kavalai import make_embedding_client

embedder = make_embedding_client("fastembed/BAAI/bge-small-en-v1.5")

texts = [
    "Tallinn is the capital of Estonia.",
    "Estonia's capital city is Tallinn.",
    "Green Village has one pub, The Rusty Anchor.",
]
vectors, stats = await embedder.compute_embeddings(texts, normalize=True)

print(f"{len(vectors)} vectors of dimension {len(vectors[0])}")


def similarity(a, b):
    return sum(x * y for x, y in zip(a, b))


print(f"sim(0, 1) = {similarity(vectors[0], vectors[1]):.3f}"
      "  <- same meaning, different words")
print(f"sim(0, 2) = {similarity(vectors[0], vectors[2]):.3f}  <- unrelated")

3 vectors of dimension 384
sim(0, 1) = 0.976  <- same meaning, different words
sim(0, 2) = 0.427  <- unrelated

2026-09-12 16:59:38.419972035 [E:onnxruntime:Default, provider_bridge_ort.cc:2195 TryGetProviderInfo_CUDA] /onnxruntime_src/onnxruntime/core/session/provider_bridge_ort.cc:1778 onnxruntime::Provider& onnxruntime::ProviderLibrary::Get() [ONNXRuntimeError] : 1 : FAIL : Failed to load library libonnxruntime_providers_cuda.so with error: libcublasLt.so.12: cannot open shared object file: No such file or directory

2026-09-12 16:59:38.419990255 [W:onnxruntime:Default, onnxruntime_pybind_state.cc:1055 CreateExecutionProviderInstance] Failed to create CUDAExecutionProvider. Require cuDNN 9.* and CUDA 12.*. Please install all dependencies as mentioned in the GPU requirements page (https://onnxruntime.ai/docs/execution-providers/CUDA-ExecutionProvider.html#requirements), make sure they're in the PATH, and that your GPU is supported.
/home/timo/projects/kaval.ai/.venv/lib/python3.12/site-packages/fastembed/common/onnx_model.py:120: RuntimeWarning: Attempt to set CUDAExecutionProvider failed. Current providers: ['CPUExecutionProvider'].If you are using CUDA 12.x, install onnxruntime-gpu via `pip install onnxruntime-gpu --extra-index-url https://aiinfra.pkgs.visualstudio.com/PublicPackages/_packaging/onnxruntime-cuda-12/pypi/simple/`
  warnings.warn(

The first two sentences express the same proposition with different structure and score 0.98; the third shares the word “Village” with nothing else in common and scores 0.43. The vector captures meaning, not spelling — which is why this beats keyword search for questions phrased in the user’s own words. Indexing a corpus this way and querying it is exactly what Retrieval-augmented generation (RAG) does for you.

Provider clients: OpenAI, Gemini, Anthropic and Ollama¶

You can also construct a client directly and pass the key yourself:

from kavalai import OpenAIClient, GeminiClient, AnthropicClient, OllamaClient

openai = OpenAIClient("gpt-5.6-luna", api_key="sk-...")
gemini = GeminiClient("gemini-3.1-flash-lite")   # reads GEMINI_API_KEY
claude = AnthropicClient("claude-sonnet-5")      # reads ANTHROPIC_API_KEY
ollama = OllamaClient("llama3")                  # local, reads OLLAMA_HOST

Which models each provider currently offers is the provider’s own list, not Kaval.AI’s: Model providers collects the catalogue links and the one-liner that asks each SDK directly, for chat and embedding models alike.

Inside a workflow you rarely build a client at all — you name the model and the engine builds it, per node or as the workflow default via llm_model="openai/gpt-5.6-luna". Set KAVALAI_DEFAULT_LLM_MODEL and you can omit it entirely.

Bringing your own provider¶

The table above is a starting set, not a closed one. register_llm_provider adds a provider of your own under a name of your choosing, and from that point it is indistinguishable from a built-in: mycorp/model-x works in make_client, as a workflow’s llm_model, in a YAML node and in KAVALAI_DEFAULT_LLM_MODEL.

Most third-party providers — DeepSeek, Groq, Together, Fireworks, OpenRouter — speak the OpenAI wire format, so there is nothing to implement. Point the built-in client at their base URL and give the combination a name:

from kavalai import OpenAIClient, register_llm_provider

# Any endpoint speaking the OpenAI dialect. This one is OpenAI's own,
# because that is the key this notebook has.
register_llm_provider(
    "mycorp", OpenAIClient, base_url="https://api.openai.com/v1"
)

mycorp = make_client("mycorp/gpt-5.6-luna")
print(await mycorp.prompt("Name the capital of Estonia in one word."))

Tallinn

Keyword arguments given at registration are bound to the name and handed to the client on every use, so one class can serve several endpoints:

register_llm_provider(
    "deepseek", OpenAIClient,
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
)

A registration can also name a class that has not been imported yet. The dotted path is resolved on first use, which is how Kaval.AI registers its own clients — so import kavalai costs nothing on a machine where no provider SDK is installed:

register_llm_provider("mycorp", "mycorp_sdk.client.MyCorpClient")

A provider with its own protocol¶

When the API is not OpenAI-shaped, implement BaseLlmClient. There is one method to write, and it does three things: send the request, push text into the value streamer as it arrives, and report what the call cost.

The client below speaks Anthropic’s Messages API over plain HTTP. Kaval.AI already ships AnthropicClient, which is what makes this a runnable example rather than a sketch: execute it, then read the shipped class to see what a production client adds — real streaming, retries, refusal and truncation handling, and cached-token accounting.

import json
import os
import time

import httpx

from kavalai import BaseLlmClient, ModelCallStat, ensure_user_turn


class MinimalAnthropicClient(BaseLlmClient):
    """A chat client for a provider with its own request format."""

    provider = "mycorp-native"

    def __init__(self, model, llm_client_parameters=None,
                 model_stats_receiver=None, api_key=None):
        super().__init__(llm_client_parameters, model_stats_receiver)
        self.model = model
        self.api_key = api_key or os.environ["ANTHROPIC_API_KEY"]

    async def _run_chat_completions(self, chat_history, response_model,
                                    streamer):
        started = time.perf_counter()
        value_streamer = streamer.get_value_streamer(
            "response", response_model=response_model
        )

        payload = {
            "model": self.model,
            # The Messages API requires a cap; the caller's wins.
            "max_tokens": self.parameters.max_output_tokens or 1024,
            # prompt() and llm nodes send system-only histories, which this
            # API rejects; ensure_user_turn is the shared fix.
            "messages": [
                {"role": message.role, "content": message.content}
                for message in ensure_user_turn(chat_history.messages)
            ],
        }
        if response_model:
            schema = response_model.model_json_schema()
            schema["additionalProperties"] = False
            payload["output_config"] = {
                "format": {"type": "json_schema", "schema": schema}
            }

        async with httpx.AsyncClient(timeout=60.0) as http:
            response = await http.post(
                "https://api.anthropic.com/v1/messages",
                headers={"x-api-key": self.api_key,
                         "anthropic-version": "2023-06-01"},
                json=payload,
            )
            response.raise_for_status()
            body = response.json()

        text = "".join(block["text"] for block in body["content"]
                       if block["type"] == "text")
        await value_streamer.stream_partial(text)
        await value_streamer.stream_complete()

        usage = body.get("usage", {})
        prompt_tokens = usage.get("input_tokens", 0)
        completion_tokens = usage.get("output_tokens", 0)
        await self._send_model_call_stats(ModelCallStat(
            call_type="llm",
            model=self.stat_model_name(),
            request_data=json.dumps(payload, default=str),
            response_data=text,
            duration_seconds=time.perf_counter() - started,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
            response_code=response.status_code,
        ))


register_llm_provider("mycorp-native", MinimalAnthropicClient)

claude = make_client("mycorp-native/claude-sonnet-5")
print(await claude.prompt("Reply with exactly: it works."))

it works.

Reporting a ModelCallStat is what puts a custom provider’s token usage in the same place as the built-ins: WorkflowState.token_usage, the model_call_stats table and the backoffice, with nothing else needing to be told the provider exists.

Structured output needs no extra support either. The value streamer validates whatever comes back against the model, so a custom client gets typed answers on the same terms as openai/:

class Village(BaseModel):
    name: str
    residents: int


village = await claude.prompt(
    "Green Village has 104 residents. Describe it.", response_model=Village
)
print(repr(village))

Village(name='Green Village', residents=104)

Embedding providers¶

register_embedding_provider mirrors it. An embedding client implements compute_embeddings, which returns the vectors and a ModelCallStat.

The one below calls no provider at all: it hashes words into a fixed number of dimensions with scikit-learn, downloading nothing and needing no key. Be clear about what that buys — the vectors are lexical rather than semantic. They match shared words, which is useful for exact terms, offline runs and tests, and no substitute for an embedding model when the question is worded differently from the answer.

from sklearn.feature_extraction.text import HashingVectorizer

from kavalai import (
    BaseEmbeddingClient,
    make_embedding_client,
    register_embedding_provider,
)


class LexicalEmbeddingClient(BaseEmbeddingClient):
    """Hashed bag-of-words vectors: no download, no key, no network."""

    def __init__(self, model, dimensions=512):
        super().__init__(model)
        self.vectorizer = HashingVectorizer(
            n_features=dimensions, norm="l2", alternate_sign=False
        )

    async def compute_embeddings(self, texts, normalize=False,
                                 normalizer=None, **kwargs):
        started = time.perf_counter()
        vectors = self.vectorizer.transform(texts).toarray().tolist()
        stats = ModelCallStat(
            call_type="embedding",
            model=f"lexical/{self.model}",
            batch_size=len(texts),
            total_tokens=0,
            duration_seconds=time.perf_counter() - started,
        )
        return vectors, stats


register_embedding_provider("lexical", LexicalEmbeddingClient, dimensions=512)

embedder = make_embedding_client("lexical/word-hash")
vectors, stats = await embedder.compute_embeddings(["a test sentence"])
print(f"{len(vectors[0])} dimensions, batch of {stats.batch_size}")

512 dimensions, batch of 1

The registry is what carries that beyond the client itself. A RAG service builds its embedding client by name through the same lookup, so a provider registered in one line reaches indexing, retrieval and every rag_query node without either of them being changed:

from kavalai.rag import SqliteRagService

FACTS = [
    "Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).",
    "Green Village has 104 residents.",
    "The village pond, Lake Miller, is 1.2 metres deep at its deepest "
    "point.",
    "Green Village's only pub, The Rusty Anchor, has been operating "
    "since 1923.",
]

village_index = SqliteRagService(":memory:", model="lexical/word-hash")
await village_index.index_batch(
    texts=FACTS,
    metadata_list=[{}] * len(FACTS),
    source_ids=[f"fact-{i}" for i in range(len(FACTS))],
)

for hit in await village_index.query("How deep is Lake Miller?", top_k=2):
    print(f"{hit.similarity:.3f}  {hit.content}")

0.516  The village pond, Lake Miller, is 1.2 metres deep at its deepest point.
0.135  Green Village's oldest resident is Agnes Whitlow (born 02.06.1929).

Two rules matter in practice.

Register before the workflow loads. Model names are checked when the graph is parsed, and a rag_query node resolves its service then too. An import two lines too late looks exactly like a typo.

A workflow names a registration, never a Python path. Workflow documents are served over HTTP and edited in the backoffice, so a dotted path in one would turn “edit a workflow” into “run code on the agent server”. Both that and connection strings are rejected when the workflow loads.

Under python -m kavalai.server nobody constructs the engine in Python, so name the modules that do the registering and let the server import them first:

KAVALAI_PROVIDER_MODULES=mycorp.providers,mycorp.rag

Every dotted registration is resolved directly afterwards, so a mistyped path stops the container at boot instead of failing the first request that reaches that node.

Registering a name that already exists raises, as duplicate tool names do; replace=True overrides it and logs a warning, because it changes what every later lookup means. register_rag_service completes the set — see Retrieval-augmented generation (RAG).

Where to next¶

• Streaming — streaming text and structured values in depth.

• Agents & tools — give a model tools and let it act.

• Workflows — orchestrate several steps into a typed graph.

• Retrieval-augmented generation (RAG) — answer from your own documents.

• Observability — tracing, stats and the backoffice UI.