Testing API¶

kavalai.testing replaces the model and nothing else, so a test runs the real WorkflowEngine, streamer and validation without a network connection or an API key. The module is part of the base install and imports neither a provider SDK nor pytest, so it also runs under Pyodide.

The recipe “Testing a workflow without calling a model” in Cookbook shows the clients in use, and Safety explains why a scripted model is the right substitute.

Fake LLM and embedding clients for testing without a provider.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

A workflow test should exercise the real engine, the real streamer and the real validation, and replace only the model. This module provides that replacement:

ScriptedLlmClient answers from a script of replies, streams each reply in partials through the base client’s machinery, and reports a ModelCallStat for every call.
FakeEmbeddingClient embeds any text deterministically by hashing its words, so texts that share words lie close together.
fake_providers() registers both under a provider name for the duration of a with block, so llm_model: fake/anything in a workflow and model="fake/tiny" for a RAG service resolve to them.

The module is pure Python and part of the base install. It imports neither a provider SDK nor pytest, so it also runs under Pyodide.

Token counts are estimated as one token per four characters, rounded up. The estimate is deterministic, which is what a test asserting on state.token_usage needs; it is not a tokeniser.

class kavalai.testing.ScriptedLlmClient(replies: Sequence[Any] | Callable[[list[ChatMessage], type[BaseModel] | None], Any], *, chunk_size: int = 8, model: str = 'scripted', provider: str = 'fake', llm_client_parameters: LlmClientParameters | None = None, model_stats_receiver: ModelStatsReceiver | None = None)[source]¶

Bases: BaseLlmClient

An LLM client that answers from a script instead of a provider.

Only the provider call is replaced. The reply is streamed through the base client’s Streamer, retry and restart handling, so stream_output, stream_delta and the partial-JSON parsing of structured output run as they do against a real model.

The client does not validate a reply. Like a provider, it sends text, and the consumer — the engine, chat_completions() or the agent — validates it into the requested response model. A scripted reply that does not fit the model therefore fails with the same error a model’s reply would.

A reply is one of:

a string, streamed verbatim;
a dict, list or other JSON value, or a Pydantic model instance, streamed as its JSON;
an exception instance or class, raised in place of an answer, and recorded as a failed call;
an Interrupted, which streams its text and then raises.

An exception the retry policy treats as transient (a provider SDK’s rate-limit or connection error) is retried after the policy’s backoff, and the next reply answers the retry. Any other exception fails the call.

max_output_tokens in the parameters is honoured as a provider honours it: a reply estimated at more tokens than the cap is streamed up to the cap, and the call then raises OutputTruncatedError.

The instance is also a client factory with the signature the engine’s client_factory= expects: calling it returns a client bound to the model, parameters and statistics receiver of that call, which shares this client’s script. The engine builds a client for every node it executes, so the replies are consumed in the order the nodes run, and calls lists every ScriptedCall received by this client and the clients bound from it, in order.

Parameters:¶

replies: Sequence[Any] | Callable[[list[ChatMessage], type[BaseModel] | None], Any]¶: A sequence of replies consumed in order, or a function (messages, response_model) -> reply called for every call. The function may be a coroutine function.
chunk_size: int = 8¶: Characters per streamed partial.
model: str = 'scripted'¶: Model name recorded in the statistics.
provider: str = 'fake'¶: Provider prefix recorded in the statistics.
llm_client_parameters: LlmClientParameters | None = None¶: Parameters recorded with each call.
model_stats_receiver: ModelStatsReceiver | None = None¶: Where each call’s statistics are reported.

Raises:¶

TypeError – replies is a single reply rather than a sequence of them.
ValueError – chunk_size is smaller than one.

classmethod from_model(model: str, *args: Any, **defaults: Any)[source]¶

Refuse construction from a model name alone.

A scripted client is nothing without its script, which a registry cannot supply. Register an instance with fake_providers().

property remaining : int | None¶: Replies not yet consumed; None when a function answers.

bind(model: str, parameters: LlmClientParameters | None = None, stats_receiver: ModelStatsReceiver | None = None, *, provider: str | None = None) → ScriptedLlmClient[source]¶

A client sharing this one’s script and calls, for another model.

Parameters and the statistics receiver not given are this client’s.

class kavalai.testing.ScriptedCall(model: str, messages: list[ChatMessage], response_model: type[BaseModel] | None, parameters: LlmClientParameters, reply: Any = None)[source]¶

Bases: object

One request a ScriptedLlmClient received.

Every attempt is a call, a retried one included, because every attempt consumes a reply.

Variables:¶

model : str¶: The provider/model name the call was recorded under.
messages : list[kavalai.llm_clients.base_client.ChatMessage]¶: The chat history the client was given.
response_model : type[pydantic.main.BaseModel] | None¶: The structured-output model requested, or None.
parameters : kavalai.llm_clients.base_client.LlmClientParameters¶: The parameters the client was built with, exactly as passed — sampling, timeouts and anything added later.
reply : Any¶: The scripted reply that answered the call.

model : str¶

messages : list[ChatMessage]¶

response_model : type[BaseModel] | None¶

parameters : LlmClientParameters¶

reply : Any = None¶

property prompt : str¶: The text of every message, joined by newlines.

class kavalai.testing.Interrupted(text: str, error: BaseException)[source]¶

Bases: object

A reply that streams text and then fails with error.

It stands for a provider connection that drops mid-answer. When error is one the retry policy treats as transient, the base client emits a restart chunk, consumers discard text, and the next scripted reply answers the retry.

Variables:¶

text : str¶: The partial output streamed before the failure.
error : BaseException¶: The exception raised once text has been streamed.

text : str¶

error : BaseException¶

class kavalai.testing.FakeEmbeddingClient(model: str = 'hashing', dimension: int = 8, *, provider: str = 'fake')[source]¶

Bases: BaseEmbeddingClient

An embedding client that hashes words into a fixed number of dimensions.

Each word, lowercased, is hashed to one dimension and counted there; the count vector is scaled to unit length, as provider embeddings are. The vectors are deterministic across processes and exist for any text, and a query that shares words with a document lies closer to it than to one that does not. They are lexical, not semantic: a paraphrase with no word in common is not recognised.

Normalisation follows the real clients: with normalize=True the vectors pass through normalizer, or the default normaliser when none is given.

calls lists the texts of every batch embedded by this client and the clients bound from it, in order.

Parameters:¶

model: str = 'hashing'¶: Model name recorded in the statistics.
dimension: int = 8¶: Length of every vector.
provider: str = 'fake'¶: Provider prefix recorded in the statistics.

Raises:¶

ValueError – dimension is smaller than one.

bind(model: str, *, provider: str | None = None) → FakeEmbeddingClient[source]¶: A client sharing this one’s dimension and calls, for another model.

vector(text: str) → list[float][source]¶

The unit-length embedding of text.

A text without a word character is hashed whole, so no text maps to the zero vector, whose cosine distance is undefined.

async compute_embeddings(texts: list[str], normalize: bool = False, normalizer: Normalizer | None = None, **kwargs) → tuple[list[list[float]], ModelCallStat][source]¶

class kavalai.testing.FakeProviders(llm: ScriptedLlmClient | None, embedding: FakeEmbeddingClient)[source]¶

Bases: NamedTuple

The clients fake_providers() registered.

Variables:¶

llm : kavalai.testing.ScriptedLlmClient | None¶: The scripted LLM client, or None when none was registered.
embedding : kavalai.testing.FakeEmbeddingClient¶: The embedding client.

llm : ScriptedLlmClient | None¶: Alias for field number 0

embedding : FakeEmbeddingClient¶: Alias for field number 1

kavalai.testing.fake_providers(llm: ScriptedLlmClient | None = None, embedding: FakeEmbeddingClient | None = None, name: str = 'fake') → Iterator[FakeProviders][source]¶

Inside the block name/<model> resolves to llm wherever an LLM model is named — a workflow’s llm_model, make_client() — and to embedding wherever an embedding model is, such as a RAG service’s model. Each resolution returns a client bound to the named model that shares the original’s script and calls.

On exit the registries are restored exactly: a name that was registered before, a built-in included, gets its previous registration back, and one that was not is removed. name="openai" therefore runs an unchanged workflow that names openai/... against the script.

The function is also the body of a pytest fixture:

@pytest.fixture
def providers():
    with fake_providers(llm=ScriptedLlmClient([...])) as fakes:
        yield fakes

Parameters:¶

llm: ScriptedLlmClient | None = None¶: The LLM client to register. None registers no LLM provider.
embedding: FakeEmbeddingClient | None = None¶: The embedding client to register. None registers a new FakeEmbeddingClient.
name: str = 'fake'¶: The provider name to register both under.

Yields:¶

The registered clients.

Raises:¶

RegistryError – name is not a valid provider name.