API Reference

Protocols

`LLMCacheProtocol`

Protocol for LLM cache implementations.

get

async def get(key: str | dict[str, Any]) -> Any | None

source

Get value from cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key (string or structured dict).

Returns

Type	Description
Any \| None	Cached value, or ``None`` if not present.

set

async def set(
    key: str | dict[str, Any],
    value: Any,
    ttl: float | None = None
) -> None

source

Set value in cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key (string or structured dict).
`value`	Any	Value to store.
`ttl`	float \| None	Optional time-to-live in seconds.

delete

async def delete(key: str | dict[str, Any]) -> bool

source

Delete entry from cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key to remove.

Returns

Type	Description
bool	``True` `if the key existed and was removed,` `False`` otherwise.

clear

async def clear() -> None

source

Clear all entries.

get_stats

def get_stats() -> dict[str, Any]

source

Return cache statistics.

Returns

Type	Description
dict[str, Any]	Mapping of statistic name to value.

Classes

`APIPricingSource`

source

Pricing source from HTTP API endpoint.

Fetches pricing data from a remote API. Useful for getting the latest pricing updates, but requires network connectivity.

Attributes: endpoint: API endpoint URL. timeout: Request timeout in seconds.

Example

source = APIPricingSource(
"https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json"
)
pricing = await source.get_pricing("gpt-4")

__init__

def __init__(
    endpoint: str,
    timeout: float = 10.0
)

source

Initialize API pricing source.

Parameters

Parameter	Type	Description
`endpoint`	str	URL to fetch pricing from.
`timeout`	float	Request timeout in seconds (default: 10).

get_pricing

async def get_pricing(model: str) -> ModelPricing | None

source

Get pricing for a specific model.

Parameters

Parameter	Type	Description
`model`	str	Model identifier.

Returns

Type	Description
ModelPricing \| None	ModelPricing if found, None otherwise.

get_all_pricing

async def get_all_pricing() -> dict[str, ModelPricing]

source

Get all pricing data.

Returns

Type	Description
dict[str, ModelPricing]	All pricing data from API.

source_name

property source_name() -> str

source

Get source name.

invalidate_cache

def invalidate_cache() -> None

source

Clear cached pricing data to force refresh.

`AbstractPricingSource`

source

Abstract base class for pricing data sources.

All pricing sources must implement get_pricing() to return ModelPricing for a given model name, or None if not found.

get_pricing

async def get_pricing(model: str) -> ModelPricing | None

source

Get pricing for a specific model.

Parameters

Parameter	Type	Description
`model`	str	Model identifier (e.g., "gpt-4-turbo").

Returns

Type	Description
ModelPricing \| None	ModelPricing if found, None otherwise.

get_all_pricing

async def get_all_pricing() -> dict[str, ModelPricing]

source

Get all available pricing data.

Returns

Type	Description
dict[str, ModelPricing]	Dictionary mapping model names to pricing.

source_name

property source_name() -> str

source

Get the name of this pricing source.

Returns

Type	Description
str	Human-readable source name.

`AnthropicClient`

source

Anthropic Claude LLM client implementation.

Conforms to: LLMClientProtocol protocol via structural typing.

Supports Claude 3 (Opus, Sonnet, Haiku) models with:

Streaming responses
Tool calling
Vision capabilities
Automatic retry and error handling

Example

from quadkit.ai import ClientConfig
config = ClientConfig(provider="anthropic", model="claude-3-sonnet-20240229")
client = AnthropicClient(config)
completion = await client.complete([
ChatMessage(role="user", content="Hello!")
])

__init__

def __init__(config: ClientConfig)

source

Initialize Anthropic client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

Raises

Exception	Description
ImportError	If anthropic package is not installed

async def close() -> None

source

Close the Anthropic client.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

`CSVOutputParser`

source

Parse LLM responses into lists of dictionaries (CSV format).

Extracts JSON array from the response and converts to list of dicts, where each dict represents a CSV row with column names as keys.

Example

parser = CSVOutputParser()
result = parser.parse('[{"name": "John", "age": 30}, {"name": "Jane", "age": 25}]')
assert len(result) == 2
assert result[0]["name"] == "John"

parse

def parse(text: str) -> list[dict[str, Any]]

source

Parse text into a list of dictionaries.

Parameters

Parameter	Type	Description
`text`	str	Raw LLM response text that may contain JSON array.

Returns

Type	Description
list[dict[str, Any]]	List of dictionaries, each representing a CSV row.

Raises

Exception	Description
ParseError	When JSON cannot be extracted or is not an array.

parse_csv_string

def parse_csv_string(text: str) -> list[dict[str, Any]]

source

Parse raw CSV text (not JSON) into list of dictionaries.

Parameters

Parameter	Type	Description
`text`	str	Raw CSV text with header row.

Returns

Type	Description
list[dict[str, Any]]	List of dictionaries, each representing a CSV row.

Raises

Exception	Description
ParseError	When CSV cannot be parsed.

get_format_instructions

def get_format_instructions() -> str

source

Return format instructions for the LLM.

Returns

Type	Description
str	Format instruction string telling the model to output a valid JSON array of objects.

`CacheEntry`

source

Cache entry with metadata.

Attributes: key: Cache key. value: Cached value. created_at: When entry was created. last_access_at: When the entry was last read (LRU eviction order). expires_at: When entry expires (Unix timestamp). hits: Number of cache hits. size_bytes: Approximate size in bytes.

`CacheMetrics`

source

Cache performance metrics.

Attributes: cache_hits: Number of cache hits cache_misses: Number of cache misses tokens_saved: Total tokens saved by caching bytes_saved: Total bytes saved by caching

hit_rate

property hit_rate() -> float

source

Calculate cache hit rate as percentage (0-100).

`CacheStats`

source

Cache statistics.

Attributes: hits: Number of cache hits. misses: Number of cache misses. evictions: Number of evictions. total_entries: Current number of entries. total_size_bytes: Total cache size in bytes.

hit_rate

property hit_rate() -> float

source

Calculate cache hit rate.

`CharEstimateCounter`

source

Character-based token count estimator (~4 chars per token).

Always available without any optional dependencies. Suitable as a safe fallback counter.

Parameters

Parameter	Type	Description
`model`		Model name (used for identification only).

__init__

def __init__(model: str = 'unknown') -> None

source

Initialize CharEstimateCounter.

Parameters

Parameter	Type	Description
`model`	str	Model name for identification.

model

property model() -> str

source

The model this counter is calibrated for.

count

def count(text: str) -> int

source

Count tokens using character estimation.

count_messages

def count_messages(messages: Sequence[ChatMessageProtocol]) -> int

source

Count tokens in a list of chat messages.

`ChatMessage`

source

A single chat message.

Implements ChatMessageProtocol with DomainModel semantics for validation.

Example

msg = ChatMessage(role="user", content="Hello, how are you?")

`ClientConfig`

source

Configuration for LLM clients.

Example

config = ClientConfig(
provider="openai",
model="gpt-4-turbo",
api_key="sk-...",
temperature=0.7,
max_tokens=2000,
)

`CohereClient`

source

Client for Cohere's enterprise NLP API.

Conforms to: LLMClientProtocol protocol via structural typing.

Supports Chat, Embeddings, and Reranking with:

RAG-optimized models (Command R/R+)
High-performance embeddings
Native reranking support

__init__

def __init__(config: ClientConfig) -> None

source

Initialize Cohere client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

api_key

property api_key() -> SecretStr

source

Get API key from config.

base_url

property base_url() -> str

source

Get base URL from config.

embed

async def embed(
    texts: list[str] | str,
    model: str = 'embed-english-v3.0',
    input_type: str = 'search_document',
    **kwargs: Any
) -> list[list[float]]

source

Generate embeddings.

Parameters

Parameter	Type	Description
`texts`	list[str] \| str	Text or list of texts to embed.
`model`	str	Model ID (default: "embed-english-v3.0").
`input_type`	str	Type of input ("search_document", "search_query", "classification", "clustering"). **kwargs: Additional parameters.

Returns

Type	Description
list[list[float]]	List of embedding vectors.

Example

# Embed documents
doc_embeddings = await client.embed(
texts=["Doc 1", "Doc 2"],
input_type="search_document"
)

# Embed query
query_embedding = await client.embed(
texts="What is AI?",
input_type="search_query"
)

rerank

async def rerank(
    query: str,
    documents: list[str] | list[dict[str, str]],
    model: str = 'rerank-english-v3.0',
    top_n: int | None = None,
    **kwargs: Any
) -> list[dict[str, Any]]

source

Rerank documents for a query.

Parameters

Parameter	Type	Description
`query`	str	Search query.
`documents`	list[str] \| list[dict[str, str]]	List of documents (strings or dicts with 'text' key).
`model`	str	Reranking model (default: "rerank-english-v3.0").
`top_n`	int \| None	Return top N results (default: all). **kwargs: Additional parameters.

Returns

Type	Description
list[dict[str, Any]]	List of ranked documents with scores.

Example

results = await client.rerank(
query="What is machine learning?",
documents=[
"ML is a subset of AI...",
"Unrelated document...",
"Deep learning uses neural networks..."
],
top_n=2
)
for result in results:
print(f"Score: {result['relevance_score']:.3f} - {result['document']['text']}")

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight health check against the provider.

async def close() -> None

source

Close the HTTP client.

`Completion`

source

LLM completion response.

Implements completion semantics with DomainModel for validation and additional fields.

Example

completion = Completion(
content="Hello! I'm doing well, thank you.",
model="gpt-4-turbo",
usage=TokenUsage(prompt_tokens=10, completion_tokens=8, total_tokens=18)
)

`ConversationConfig`

source

Configuration for conversation management.

Example

config = ConversationConfig(
max_tokens=4096,
reserve_tokens=1000,
trim_strategy="oldest"
)

`ConversationManager`

source

Manage multi-turn conversations with automatic context window management.

This class handles:

Message history management
Automatic token counting
Context window trimming
System prompt handling
Conversation statistics

Example

from quadkit.ai.llm import OpenAIClient, ConversationManager

client = OpenAIClient(api_key="sk-...", model="gpt-4")
manager = ConversationManager(
client=client,
system_prompt="You are a helpful assistant.",
max_tokens=4096
)

# Add user message and get response
response = await manager.chat("What is Python?")
print(response.content)

# Continue conversation
response = await manager.chat("Tell me more about it")
print(response.content)

# Get conversation history
history = manager.get_history()
stats = manager.get_stats()
print(f"Total messages: {stats.total_messages}")
print(f"Total tokens: {stats.total_tokens}")

__init__

def __init__(
    client: AbstractLLMClient,
    system_prompt: str | None = None,
    max_tokens: int = 4096,
    reserve_tokens: int = 1000,
    trim_strategy: str = 'oldest',
    metadata: Metadata | None = None,
    token_counter: TokenCounterProtocol | None = None
) -> None

source

Initialize conversation manager.

Parameters

Parameter	Type	Description
`client`	AbstractLLMClient	LLM client for completions
`system_prompt`	str \| None	Optional system prompt (prepended to all conversations)
`max_tokens`	int	Maximum context window size
`reserve_tokens`	int	Tokens to reserve for completion
`trim_strategy`	str	Message trimming strategy ('oldest', 'middle', 'summary')
`metadata`	Metadata \| None	Additional metadata for the conversation
`token_counter`	TokenCounterProtocol \| None	Optional TokenCounterProtocol implementation. If not provided, uses CharEstimateCounter.

chat

async def chat(
    message: str,
    role: Role = Role.USER,
    **completion_kwargs: Any
) -> Completion

source

Send a message and get a response.

Parameters

Parameter	Type	Description
`message`	str	Message content
`role`	Role	Message role (default: USER) **completion_kwargs: Additional kwargs for completion

Returns

Type	Description
Completion	Completion response from LLM

Example

response = await manager.chat("Hello!")
print(response.content)

add_message

async def add_message(
    role: Role,
    content: str,
    update_stats: bool = True
) -> None

source

Add a message to conversation history without getting a response.

Parameters

Parameter	Type	Description
`role`	Role	Message role
`content`	str	Message content
`update_stats`	bool	Whether to update statistics

Example

await manager.add_message(Role.USER, "Hello")
await manager.add_message(Role.ASSISTANT, "Hi there!")

get_history

def get_history(
    include_system: bool = True,
    limit: int | None = None
) -> list[ChatMessage]

source

Get conversation history.

Parameters

Parameter	Type	Description
`include_system`	bool	Include system message in history
`limit`	int \| None	Maximum number of messages to return (most recent)

Returns

Type	Description
list[ChatMessage]	List of chat messages

Example

history = manager.get_history(limit=10)
for msg in history:
print(f"{msg.role}: {msg.content}")

get_stats

def get_stats() -> ConversationStats

source

Get conversation statistics.

Returns

Type	Description
ConversationStats	Conversation statistics

Example

stats = manager.get_stats()
print(f"Total tokens: {stats.total_tokens}")

clear_history

def clear_history(keep_system: bool = True) -> None

source

Clear conversation history.

Parameters

Parameter	Type	Description
`keep_system`	bool	Keep system message when clearing

Example

manager.clear_history()

update_system_prompt

def update_system_prompt(system_prompt: str) -> None

source

Update the system prompt.

Parameters

Parameter	Type	Description
`system_prompt`	str	New system prompt

Example

manager.update_system_prompt("You are a Python expert.")

get_token_count

def get_token_count() -> int

source

Get current total token count.

Returns

Type	Description
int	Total tokens in conversation

Example

tokens = manager.get_token_count()
print(f"Current tokens: {tokens}")

get_available_tokens

def get_available_tokens() -> int

source

Get available tokens for completion.

Returns

Type	Description
int	Available tokens (max_tokens - current_tokens - reserve_tokens) Can be negative if context window is exceeded

Example

available = manager.get_available_tokens()
print(f"Available for completion: {available}")

export_history

def export_history() -> dict[str, Any]

source

Export conversation history to dictionary.

Returns

Type	Description
dict[str, Any]	Dictionary with conversation data (JSON-serializable)

Example

data = manager.export_history()
from quadkit import serialization as json
with open("conversation.json", "w") as f:
json.dump(data, f)

from_history

def from_history(
    cls,
    client: AbstractLLMClient,
    history_data: dict[str, Any]
) -> ConversationManager

source

Create conversation manager from exported history.

Parameters

Parameter	Type	Description
`client`	AbstractLLMClient	LLM client
`history_data`	dict[str, Any]	Exported history data

Returns

Type	Description
ConversationManager	ConversationManager instance

Example

from quadkit import serialization as json
with open("conversation.json") as f:
data = json.load(f)
manager = ConversationManager.from_history(client, data)

`ConversationStats`

source

Statistics for a conversation.

Example

stats = ConversationStats(
total_messages=10,
total_tokens=2048,
user_messages=5,
assistant_messages=5
)

`CostEstimate`

source

Cost estimation result.

Attributes: prompt_cost: Cost for prompt tokens. completion_cost: Cost for completion tokens. total_cost: Total estimated cost. currency: Currency code (default: USD). model: Model name. rate_per_1k_prompt: Rate per 1000 prompt tokens. rate_per_1k_completion: Rate per 1000 completion tokens.

`EnumOutputParser`

source

Parse LLM responses into Enum members.

Extracts JSON from the response and maps it to an Enum member. Supports both string values and integer values.

Example

from enum import Enum

class Status(Enum):
ACTIVE = "active"
INACTIVE = "inactive"

parser = EnumOutputParser(Status)
result = parser.parse('"active"')
assert result == Status.ACTIVE

__init__

def __init__(enum: type[Enum]) -> None

source

Initialize with an Enum class.

Parameters

Parameter	Type	Description
`enum`	type[Enum]	Enum subclass to parse into.

parse

def parse(text: str) -> Enum

source

Parse text into an Enum member.

Parameters

Parameter	Type	Description
`text`	str	Raw LLM response text that may contain JSON with enum value.

Returns

Type	Description
Enum	Corresponding Enum member.

Raises

Exception	Description
ParseError	When JSON cannot be extracted or enum value is invalid.

get_format_instructions

def get_format_instructions() -> str

source

Return format instructions for the LLM.

Returns

Type	Description
str	Format instruction string telling the model to output a valid enum value.

`ErrorMetrics`

source

Error tracking and rates for a provider.

Attributes: provider: Provider name model_id: Model identifier total_requests: Total number of requests attempted error_count: Total number of errors rate_limit_errors: Number of rate limit errors timeout_errors: Number of timeout errors auth_errors: Number of authentication errors other_errors: Number of other errors

error_rate

property error_rate() -> float

source

Calculate current error rate as percentage (0-100).

record_error

def record_error(error_type: str) -> None

source

Record an error occurrence.

Parameters

Parameter	Type	Description
`error_type`	str	One of 'rate_limit', 'timeout', 'auth', 'other'

record_request

def record_request() -> None

source

Record a request attempt.

`FormatFixingParser`

source

Parser that retries with LLM-assisted fixing on parse failure.

Wraps a base parser and, on parse failure, calls the LLM with a fixing prompt that includes the original output and the parse error. Retries are bounded by the retry_budget.

Example

parser = FormatFixingParser(
base_parser=JSONOutputParser(),
llm_client=llm_client,
retry_budget=3
)
result = parser.parse('not valid json')

__init__

def __init__(
    base_parser: Any,
    llm_client: Any,
    *,
    retry_budget: int = 3,
    guard_check: Callable[[str], bool] | None = None
) -> None

source

Initialize the format fixing parser.

Parameters

Parameter	Type	Description
`base_parser`	Any	The underlying parser to use for parsing.
`llm_client`	Any	LLM client to use for fixing attempts.
`retry_budget`	int	Maximum number of fix attempts (default 3).
`guard_check`	Callable[[str], bool] \| None	Optional guard function to validate malformed input before sending to LLM. Should return True if safe.

parse

async def parse(text: str) -> Any

source

Parse text, attempting fixes on failure.

Parameters

Parameter	Type	Description
`text`	str	Raw LLM response text to parse.

Returns

Type	Description
Any	Parsed output from the base parser.

Raises

Exception	Description
ParseError	When all fix attempts fail or guard check fails.

get_format_instructions

def get_format_instructions() -> str

source

Return format instructions from the base parser.

Returns

Type	Description
str	Format instructions from the wrapped parser.

`FunctionCall`

source

Function call request from LLM.

`GenerationDefaults`

source

Default generation parameters applied to every routing attempt.

Example

defaults = GenerationDefaults(temperature=0.3, max_tokens=2048)

`GroqClient`

source

Client for Groq's ultra-fast LLM inference API.

Conforms to: LLMClientProtocol protocol via structural typing.

Supports Chat, Stream, and Vision with:

Ultra-fast LPU hardware synergy
OpenAI-compatible API surface
Blazing-fast token generation

__init__

def __init__(config: ClientConfig)

source

Initialize Groq client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

api_key

property api_key() -> SecretStr

source

Get API key from config.

base_url

property base_url() -> str

source

Get base URL from config.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Run a lightweight provider health probe.

list_models

async def list_models() -> list[dict[str, Any]]

source

List models available from the Groq API.

async def close() -> None

source

Close the HTTP client.

Example

await client.close()

`HuggingFaceCounter`

source

Token counter using HuggingFace AutoTokenizer (lazy-loaded).

When constructed without a model, uses character estimation (~4 chars/token). When constructed with a model name, lazy-loads that model’s tokenizer on first use.

Parameters

Parameter	Type	Description
`model`		Optional HuggingFace model name. If None, uses char estimation fallback.

__init__

def __init__(model: str | None = None) -> None

source

Initialize HuggingFaceCounter.

Parameters

Parameter	Type	Description
`model`	str \| None	Optional HuggingFace model name for tokenizer loading.

model

property model() -> str

source

Backend identifier.

count

def count(text: str) -> int

source

Count tokens in a text string.

count_messages

def count_messages(messages: Sequence[ChatMessageProtocol]) -> int

source

Count tokens in a list of chat messages.

`ImageBase64Part`

source

An image pre-encoded as base64 in a multimodal message.

Attributes: data: Raw base64-encoded bytes (no data: prefix). media_type: MIME type, e.g. "image/jpeg". detail: OpenAI vision detail level, or "" when the source carries none (e.g. Claude-sourced images). type: Discriminator field, always "image_base64".

`ImageUrlPart`

source

An image specified by URL in a multimodal message.

The framework passes the URL through to providers that support it natively (OpenAI, Anthropic, Gemini). For providers that require base64 (Ollama, Bedrock), the client fetches and converts.

Attributes: url: Public or data-URI URL of the image. detail: OpenAI vision detail level ("auto", "low", "high"). type: Discriminator field, always "image_url".

`InstructorExtractor`

source

Structured extraction from LLM completions using instructor library.

Extracts typed Pydantic models from LLM responses by:

Building a ChatMessage list with extraction instructions
Calling llm_client.complete() to get a Completion
Parsing the completion text as JSON
Validating against the response_model
Retrying on validation/parse failures up to max_retries

Unlike direct instructor usage, this implementation uses the standard LLMClientProtocol.complete() method, avoiding coupling to provider-specific client patching mechanisms.

Example

from pydantic import BaseModel

class UserInfo(BaseModel):
    name: str
    age: int

extractor = InstructorExtractor(llm_client)
result = await extractor.extract(
    prompt="Extract user info from: 'John is 30 years old'",
    response_model=UserInfo,
)
if result.is_ok():
    user = result.unwrap()
    print(user.name, user.age)
else:
    error = result.unwrap_err()
    # handle ExtractionError

from pydantic import BaseModel

class UserInfo(BaseModel):
    name: str
    age: int

extractor = InstructorExtractor(llm_client)
result = await extractor.extract(
    prompt="Extract user info from: 'John is 30 years old'",
    response_model=UserInfo,
)
if result.is_ok():
    user = result.unwrap()
    print(user.name, user.age)
else:
    error = result.unwrap_err()
    # handle ExtractionError

__init__

def __init__(
    llm_client: LLMClientProtocol,
    mode: str = 'json',
    max_retries: int = 3
) -> None

source

Initialize InstructorExtractor.

Parameters

Parameter	Type	Description
`llm_client`	LLMClientProtocol	LLMClientProtocol instance for making LLM calls.
`mode`	str	Instructor patching mode (reserved for future provider-level integration; currently unused).
`max_retries`	int	Maximum number of retries on validation/parse failure.

extract

async def extract(
    prompt: str,
    response_model: type[T],
    context: list | None = None,
    **kwargs: Any
) -> Result[T, ExtractionError]

source

Extract a structured response_model instance from an LLM call.

Parameters

Parameter	Type	Description
`prompt`	str	User prompt for extraction.
`response_model`	type[T]	Pydantic BaseModel class to extract and validate.
`context`	list \| None	Optional list of additional ChatMessage objects for context. **kwargs: Additional parameters passed to llm_client.complete().

Returns

Type	Description
Result[T, ExtractionError]	``Ok(instance)` `on successful extraction and validation.` `Err(ExtractionError)`` on parse, validation, or max retries failure.

Raises

Exception	Description

`JSONExtractor`

source

Extract and parse JSON from LLM responses.

extract

def extract(
    text: str,
    multiple: bool = False
) -> Any

source

Extract JSON from text.

extract_array

def extract_array(text: str) -> list[Any]

source

Extract JSON array from text.

`JSONFilePricingSource`

source

Pricing source from local JSON file.

This is the fastest and most reliable source as it doesn’t require network calls and works offline.

Attributes: file_path: Path to JSON pricing file. cache: In-memory cache of loaded pricing.

Example

source = JSONFilePricingSource(Path("custom_pricing.json"))
pricing = await source.get_pricing("gpt-4-turbo")

__init__

def __init__(file_path: Path)

source

Initialize JSON file pricing source.

Parameters

Parameter	Type	Description
`file_path`	Path	Path to JSON file containing pricing data.

get_pricing

async def get_pricing(model: str) -> ModelPricing | None

source

Get pricing for a specific model.

Parameters

Parameter	Type	Description
`model`	str	Model identifier.

Returns

Type	Description
ModelPricing \| None	ModelPricing if found, None otherwise.

get_all_pricing

async def get_all_pricing() -> dict[str, ModelPricing]

source

Get all pricing data.

Returns

Type	Description
dict[str, ModelPricing]	All pricing data from JSON file.

source_name

property source_name() -> str

source

Get source name.

invalidate_cache

def invalidate_cache() -> None

source

Clear cached pricing data to force reload.

`JSONOutputParser`

source

Parse LLM responses into JSON dicts.

Handles common LLM output patterns like markdown code fences, prose before/after JSON, and malformed JSON.

Example

parser = JSONOutputParser()
result = parser.parse('{"key": "value"}')
assert result == {"key": "value"}

parse

def parse(text: str) -> dict[str, Any]

source

Parse text into a JSON dict.

Parameters

Parameter	Type	Description
`text`	str	Raw LLM response text that may contain JSON.

Returns

Type	Description
dict[str, Any]	Parsed JSON as a dict.

Raises

Exception	Description
ParseError	When JSON cannot be extracted or parsed.

get_format_instructions

def get_format_instructions() -> str

source

Return format instructions for the LLM.

Returns

Type	Description
str	Format instruction string telling the model to output valid JSON.

`LLMAuditBridge`

source

Emits one ``AuditEntry`` per LLM completion.

Wired via LLMModule.configure(audit_calls=True). Default is False to avoid noise for high-volume callers.

__init__

def __init__(
    audit_logger: AuditLoggerProtocol,
    ctx: Context
) -> None

source

on_completion

async def on_completion(completion: Completion) -> None

source

Log an audit entry for a completed LLM call.

Parameters

Parameter	Type	Description
`completion`	Completion	The completed LLM response with provenance fields.

wrap

def wrap(
    client: LLMClientProtocol,
    resolver: Any
) -> LLMClientProtocol

source

Wrap an LLM client to emit audit entries per completion.

Parameters

Parameter	Type	Description
`client`	LLMClientProtocol	The LLM client to wrap.
`resolver`	Any	Container or resolver used to lazily resolve ``LLMAuditBridge` `on first` `complete()`` call.

Returns

Type	Description
LLMClientProtocol	A wrapped client that intercepts ``complete()`` to log audit entries via this bridge.

`LLMCache`

source

In-memory cache for LLM responses with TTL.

Implements LRU eviction when max_size is reached.

Parameters

Parameter	Type	Description
`ttl`		Time-to-live in seconds (default: 1 hour).
`max_size`		Maximum number of entries (default: 1000).
`max_size_bytes`		Maximum cache size in bytes (default: 100MB).

Example

cache = LLMCache(ttl=3600, max_size=500)
result = await cache.get("key")
await cache.set("key", "value")

__init__

def __init__(
    ttl: float = 3600,
    max_size: int = 1000,
    max_size_bytes: int = 100 * 1024 * 1024
)

source

Initialize LLM cache.

Parameters

Parameter	Type	Description
`ttl`	float	Time-to-live in seconds.
`max_size`	int	Maximum number of entries.
`max_size_bytes`	int	Maximum total size in bytes.

get

async def get(key: str | dict[str, Any]) -> Any | None

source

Get value from cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key (string or dict).

Returns

Type	Description
Any \| None	Cached value or None if not found/expired.

set

async def set(
    key: str | dict[str, Any],
    value: Any,
    ttl: float | None = None
) -> None

source

Set value in cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key (string or dict).
`value`	Any	Value to cache.
`ttl`	float \| None	Optional TTL override.

get_or_compute

async def get_or_compute(
    key: str | dict[str, Any],
    compute_fn: Callable[[], Any],
    ttl: float | None = None
) -> Any

source

Get from cache or compute and cache result.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key.
`compute_fn`	Callable[[], Any]	Function to compute value if cache miss.
`ttl`	float \| None	Optional TTL override.

Returns

Type	Description
Any	Cached or computed value.

Example

result = await cache.get_or_compute(
key="greeting",
compute_fn=lambda: llm.complete("Say hello")
)

delete

async def delete(key: str | dict[str, Any]) -> bool

source

Delete entry from cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key to delete.

Returns

Type	Description
bool	True if entry was deleted.

clear

async def clear() -> None

source

Clear all cache entries.

get_stats

def get_stats() -> CacheStats

source

Get cache statistics.

Returns

Type	Description
CacheStats	CacheStats object.

`LLMCompletionEvent`

source

Emitted when an LLM completion is received.

Distinct from LLMCallStartedHook (which intercepts); this is the immutable record that a completion happened.

Consumed by: cost accounting, audit, safety review.

`LLMConfig`

source

Root configuration object for the LLM routing system.

All providers are opt-in: a provider joins the cascade only when its credential environment variable is set. Use from_env to build from QK_AI_LLM__ environment variables.

Example

config = LLMConfig(
    providers=[
        ProviderConfig(name="groq", model="llama-3.3-70b-versatile", api_key="gsk_..."),
        ProviderConfig(name="gemini", model="gemini-2.5-flash", api_key="AIza..."),
    ],
    defaults=GenerationDefaults(temperature=0.3),
)

config = LLMConfig(
    providers=[
        ProviderConfig(name="groq", model="llama-3.3-70b-versatile", api_key="gsk_..."),
        ProviderConfig(name="gemini", model="gemini-2.5-flash", api_key="AIza..."),
    ],
    defaults=GenerationDefaults(temperature=0.3),
)

Environment variables (prefix QK_AI_LLM__)

Global:

    QK_AI_LLM__STRATEGY              sequential | parallel_race |
                                           cost_optimized | latency_optimized
    QK_AI_LLM__DEFAULTS__TEMPERATURE float  (default 0.2)
    QK_AI_LLM__DEFAULTS__MAX_TOKENS  int    (default: provider default)
    QK_AI_LLM__QUOTA__BACKEND        memory | database  (default memory)
    QK_AI_LLM__LOG__BACKEND          memory | database  (default memory)
    QK_AI_LLM__LOG__MAX_ENTRIES      int    (default 1000)

Per-provider (pattern: QK_AI_LLM__PROVIDERS__{NAME}__{FIELD}):

    __{NAME}__API_KEY    str   API key -- activates key-auth providers
    __{NAME}__BASE_URL   str   Endpoint -- activates local/custom providers
    __{NAME}__MODEL      str   Model override (has per-provider defaults)
    __{NAME}__TIMEOUT    int   Request timeout in seconds (default 30)
    __{NAME}__ENABLED    bool  Explicit enable/disable (default true)

Supported provider names and their activation:

    OPENAI       API_KEY required   default model: gpt-4o
    ANTHROPIC    API_KEY required   default model: claude-3-5-sonnet-20241022
    GROQ         API_KEY required   default model: llama-3.3-70b-versatile
    GEMINI       API_KEY required   default model: gemini-2.5-flash
    MISTRAL      API_KEY required   default model: mistral-large-latest
    COHERE       API_KEY required   default model: command-r-plus
    OPENROUTER   API_KEY required   default model: openai/gpt-4o-mini
    DEEPSEEK     API_KEY required   default model: deepseek-chat
    TOGETHER     API_KEY required   default model: meta-llama/Llama-3-8b-chat-hf
    FIREWORKS    API_KEY required   default model: accounts/fireworks/models/llama-v3-70b-instruct
    OLLAMA       BASE_URL required  default model: llama3.2  (default base: http://localhost:11434)
    OPENAI_COMPATIBLE  BASE_URL + MODEL required  (generic OpenAI-compatible: LM Studio, VLLM, etc.)

Azure-specific extras (activated by AZURE__API_KEY + AZURE__BASE_URL):

    QK_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_RESOURCE
    QK_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_DEPLOYMENT
    QK_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_API_VERSION

Cloudflare-specific extras (activated by CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID):

    QK_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID  <- activates
    QK_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_API_TOKEN
    QK_AI_LLM__PROVIDERS__CLOUDFLARE__MODEL

AWS Bedrock extras (activated by BEDROCK__EXTRAS__AWS_REGION):

    QK_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_REGION        <- activates
    QK_AI_LLM__PROVIDERS__BEDROCK__MODEL
    QK_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_ACCESS_KEY_ID
    QK_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_SECRET_ACCESS_KEY

Google Vertex AI extras (activated by VERTEX__EXTRAS__VERTEX_PROJECT):

    QK_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_PROJECT      <- activates
    QK_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_LOCATION
    QK_AI_LLM__PROVIDERS__VERTEX__MODEL
    QK_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_CREDENTIALS_FILE

Global:

    QK_AI_LLM__STRATEGY              sequential | parallel_race |
                                           cost_optimized | latency_optimized
    QK_AI_LLM__DEFAULTS__TEMPERATURE float  (default 0.2)
    QK_AI_LLM__DEFAULTS__MAX_TOKENS  int    (default: provider default)
    QK_AI_LLM__QUOTA__BACKEND        memory | database  (default memory)
    QK_AI_LLM__LOG__BACKEND          memory | database  (default memory)
    QK_AI_LLM__LOG__MAX_ENTRIES      int    (default 1000)

Per-provider (pattern: QK_AI_LLM__PROVIDERS__{NAME}__{FIELD}):

    __{NAME}__API_KEY    str   API key -- activates key-auth providers
    __{NAME}__BASE_URL   str   Endpoint -- activates local/custom providers
    __{NAME}__MODEL      str   Model override (has per-provider defaults)
    __{NAME}__TIMEOUT    int   Request timeout in seconds (default 30)
    __{NAME}__ENABLED    bool  Explicit enable/disable (default true)

Supported provider names and their activation:

    OPENAI       API_KEY required   default model: gpt-4o
    ANTHROPIC    API_KEY required   default model: claude-3-5-sonnet-20241022
    GROQ         API_KEY required   default model: llama-3.3-70b-versatile
    GEMINI       API_KEY required   default model: gemini-2.5-flash
    MISTRAL      API_KEY required   default model: mistral-large-latest
    COHERE       API_KEY required   default model: command-r-plus
    OPENROUTER   API_KEY required   default model: openai/gpt-4o-mini
    DEEPSEEK     API_KEY required   default model: deepseek-chat
    TOGETHER     API_KEY required   default model: meta-llama/Llama-3-8b-chat-hf
    FIREWORKS    API_KEY required   default model: accounts/fireworks/models/llama-v3-70b-instruct
    OLLAMA       BASE_URL required  default model: llama3.2  (default base: http://localhost:11434)
    OPENAI_COMPATIBLE  BASE_URL + MODEL required  (generic OpenAI-compatible: LM Studio, VLLM, etc.)

Azure-specific extras (activated by AZURE__API_KEY + AZURE__BASE_URL):

    QK_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_RESOURCE
    QK_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_DEPLOYMENT
    QK_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_API_VERSION

Cloudflare-specific extras (activated by CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID):

    QK_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID  <- activates
    QK_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_API_TOKEN
    QK_AI_LLM__PROVIDERS__CLOUDFLARE__MODEL

AWS Bedrock extras (activated by BEDROCK__EXTRAS__AWS_REGION):

    QK_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_REGION        <- activates
    QK_AI_LLM__PROVIDERS__BEDROCK__MODEL
    QK_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_ACCESS_KEY_ID
    QK_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_SECRET_ACCESS_KEY

Google Vertex AI extras (activated by VERTEX__EXTRAS__VERTEX_PROJECT):

    QK_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_PROJECT      <- activates
    QK_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_LOCATION
    QK_AI_LLM__PROVIDERS__VERTEX__MODEL
    QK_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_CREDENTIALS_FILE

from_env

def from_env(cls) -> LLMConfig

source

Build a routing config from QK_AI_LLM__ environment variables.

Returns

Type	Description
LLMConfig	Populated LLMConfig.

`LLMModule`

source

LLM client and model-management integration.

Declared global so its exports are visible to every module in the graph without explicit import — the LLM client is cross-cutting infrastructure consumed by agents, RAG, verification, and embedding pipelines.

Call configure to register an LLMClientProtocol implementation and optional model manager for injection.

Usage

from quadkit.ai.llm.config import ClientConfig

@module(
    imports=[
        LLMModule.configure(
            ClientConfig(provider="openai", model="gpt-4o")
        )
    ]
)
class AppModule(Module):
    pass

from quadkit.ai.llm.config import ClientConfig

@module(
    imports=[
        LLMModule.configure(
            ClientConfig(provider="openai", model="gpt-4o")
        )
    ]
)
class AppModule(Module):
    pass

Multi-provider routing

from quadkit.ai.llm import LLMModule

@module(
    imports=[LLMModule.configure(routing=LLMConfig())]
)
class AppModule(Module):
    pass

from quadkit.ai.llm import LLMModule

@module(
    imports=[LLMModule.configure(routing=LLMConfig())]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: ClientConfig | Any | None = None,
    *,
    routing: LLMConfig | Any | None = None,
    enable_model_manager: bool = False,
    enable_streaming: bool = True,
    audit_calls: bool = False
) -> DynamicModule

source

Create an LLMModule with a single configured provider.

Parameters

Parameter	Type	Description
`config`	ClientConfig \| Any \| None	ClientConfig or ``None`` to read configuration from environment variables.
`routing`	LLMConfig \| Any \| None	Optional LLMConfig enabling the multi-provider routing layer instead of the single-provider client.
`enable_model_manager`	bool	Register LLMModelManager for local model lifecycle control.
`enable_streaming`	bool	Enable streaming response support. Defaults to ``True; set to` `False`` to restrict to non-streaming clients only.
`audit_calls`	bool	Emit an ``AuditEntry` `per LLM completion via LLMAuditBridge. Requires` `AuditLoggerProtocol` `in the container. Default` `False``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: ClientConfig | Any | None = None
) -> DynamicModule

source

Create an LLMModule suitable for unit and integration testing.

Uses a no-op or stub LLM client with minimal external dependencies. Streaming is disabled by default to simplify test assertions.

Parameters

Parameter	Type	Description
`config`	ClientConfig \| Any \| None	Optional ClientConfig override. Uses safe test defaults when ``None``.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

`LLMProvider`

source

Provider that registers LLM services with the Quadkit DI container.

Registers an LLMClientProtocol, optional LLM response cache, and an LLMModelManager so all three are injectable throughout the application.

Example

from quadkit.ai.llm.di.provider import LLMProvider
from quadkit.ai.llm.config import ClientConfig

app.use(LLMProvider(ClientConfig(provider="openai", model="gpt-4o")))

# LLMClientProtocol is now injectable:
class MyService:
def __init__(self, llm: LLMClientProtocol) -> None:
self.llm = llm

__init__

def __init__(
    config: ClientConfig | None = None,
    enable_model_manager: bool = False,
    enable_streaming: bool = True,
    audit_calls: bool = False,
    name: str = 'llm',
    cache_backend: CacheBackendProtocol | None = None,
    stub_mode: bool = False
) -> None

source

Initialize the LLM Provider.

Parameters

Parameter	Type	Description
`config`	ClientConfig \| None	LLM client configuration; defaults to ClientConfig() (reads env).
`enable_model_manager`	bool	Register LLMModelManager for local model control.
`enable_streaming`	bool	Enable streaming response support.
`audit_calls`	bool	Register LLMAuditBridge to emit audit entries per completion.
`name`	str	Provider name used for identification.
`cache_backend`	CacheBackendProtocol \| None	Injected cache backend for optional response caching.

llm_client

property llm_client() -> LLMClientProtocol | None

source

The constructed LLM client (optional wrapper chain).

None until register has built the client (or when the provider runs in a mode that registers no client). Consumers that need the client — e.g. the an AI extension orchestrator — should use this accessor rather than reaching into the private attribute.

async def register(container: ContainerRegistrarProtocol) -> None

source

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The Quadkit DI container registrar.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Boot the LLM provider — validates API key presence and format.

Parameters

Parameter	Type	Description
`container`	ContainerResolverProtocol	The DI container resolver.

shutdown

async def shutdown() -> None

source

Close client connections on application shutdown.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return basic health information for the registered LLM client.

`LLMProviderRegisteredHook`

source

Payload fired when an LLM provider is registered in the provider registry.

Attributes: provider: Identifier of the provider that was registered.

`LLMRequestSentHook`

source

Payload fired when an LLM request is dispatched to a provider.

Attributes: provider: Provider identifier (e.g. "openai"). model: Model name targeted by the request (e.g. "gpt-4o").

`LLMResponseReceivedHook`

source

Payload fired when a complete LLM response is received from a provider.

Attributes: provider: Provider identifier that returned the response. model: Model name that produced the response.

`LLMRoutingProvider`

source

Provider that registers the multi-provider LLM router with the DI container.

Builds the LLMRouter from a LLMConfig, chooses the appropriate quota backend and inference logger, and registers everything as singletons.

Configuration is explicit-only: LLMConfig is built from QK_AI_LLM__ environment variables via LLMConfig.from_env() and is not bound to a QuadkitConfig section, so this provider declares no config_key/ config_model attributes.

Example

from quadkit.ai.llm.module import LLMModule
from quadkit.ai.llm.routing import LLMConfig

app.use(LLMModule.configure(routing=LLMConfig.from_env()))

# LLMRouterProtocol is now injectable:
class MyService:
def __init__(self, router: LLMRouterProtocol) -> None:
self.router = router

__init__

def __init__(
    config: LLMConfig | None = None,
    database_provider: DatabaseProviderProtocol | None = None,
    model_selector: ModelSelector | None = None
) -> None

source

Initialise the LLM routing provider.

Parameters

Parameter	Type	Description
`config`	LLMConfig \| None	Routing configuration; defaults to ``LLMConfig.from_env()``.
`database_provider`	DatabaseProviderProtocol \| None	Injected DB provider used when ``quota.backend` `or` `logging.backend` `is` `database``.
`model_selector`	ModelSelector \| None	Optional model selector for capability-based routing. When provided, ``required_capabilities`` in route kwargs will filter providers whose models lack the requested capabilities.

async def register(container: ContainerRegistrarProtocol) -> None

source

Build and register the LLMRouter with the DI container.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The Quadkit DI container registrar.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Upgrade to database backends once the DB provider is resolvable.

register() runs before the database provider is instantiated, so a caller that asked for quota.backend / logging.backend == "database" but did not hand this provider a DatabaseProviderProtocol starts on in-memory backends. During boot — where resolution is allowed — resolve the DB provider and rebind the router so telemetry and quota persist. A missing provider is not fatal: the router keeps its in-memory backends.

Parameters

Parameter	Type	Description
`container`	ContainerResolverProtocol	The DI container resolver.

shutdown

async def shutdown() -> None

source

Close all routing clients on application shutdown.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return basic health information for the router.

Parameters

Parameter	Type	Description
`timeout`	float	Unused; retained for interface compatibility.

Returns

Type	Description
HealthCheckResult	A dict with ``status` `and` `providers`` keys.

`LogConfig`

source

Configuration for inference attempt logging.

Example

cfg = LogConfig(backend="database", max_entries=5000)

`MetricType`

source

Types of metrics that can be collected.

`MetricsCollector`

source

Central metrics collection and aggregation.

Collects metrics from all providers and models, provides query interfaces, and exports data for monitoring and billing.

__init__

def __init__() -> None

source

Initialize metrics collector.

get_timing_metrics

def get_timing_metrics(
    provider: str,
    model_id: str
) -> TimingMetrics | None

source

Get timing metrics for a provider/model.

Parameters

Parameter	Type	Description
`provider`	str	Provider name
`model_id`	str	Model identifier

Returns

Type	Description
TimingMetrics \| None	TimingMetrics if available, None otherwise

get_error_metrics

def get_error_metrics(
    provider: str,
    model_id: str
) -> ErrorMetrics | None

source

Get error metrics for a provider/model.

Parameters

Parameter	Type	Description
`provider`	str	Provider name
`model_id`	str	Model identifier

Returns

Type	Description
ErrorMetrics \| None	ErrorMetrics if available, None otherwise

get_cache_metrics

def get_cache_metrics() -> dict[str, CacheMetrics]

source

Get all cache metrics.

Returns

Type	Description
dict[str, CacheMetrics]	Dictionary of provider name to CacheMetrics

get_usage_metrics

def get_usage_metrics(
    provider: str,
    model_id: str
) -> UsageMetrics | None

source

Get token usage and cost metrics for a provider/model.

Parameters

Parameter	Type	Description
`provider`	str	Provider name
`model_id`	str	Model identifier

Returns

Type	Description
UsageMetrics \| None	UsageMetrics if available, None otherwise

record_request

def record_request(
    provider: str,
    model_id: str,
    latency_ms: float,
    error: bool = False,
    error_type: str | None = None,
    *,
    prompt_tokens: int = 0,
    completion_tokens: int = 0,
    cost_usd: float = 0.0
) -> None

source

Record a request and its metrics.

Parameters

Parameter	Type	Description
`provider`	str	Provider name
`model_id`	str	Model identifier
`latency_ms`	float	Request latency in milliseconds
`error`	bool	Whether the request resulted in an error
`error_type`	str \| None	Type of error (only if error=True)
`prompt_tokens`	int	Prompt tokens consumed by the request
`completion_tokens`	int	Completion tokens produced by the request
`cost_usd`	float	Estimated cost of the request in US dollars

record_cache_hit

def record_cache_hit(
    provider: str,
    tokens_saved: int = 0
) -> None

source

Record a cache hit.

Parameters

Parameter	Type	Description
`provider`	str	Provider name
`tokens_saved`	int	Number of tokens saved by the cache hit

record_cache_miss

def record_cache_miss(provider: str) -> None

source

Record a cache miss.

Parameters

Parameter	Type	Description
`provider`	str	Provider name

take_snapshot

def take_snapshot(
    provider: str,
    metric_type: MetricType
) -> None

source

Take a snapshot of current metrics.

Parameters

Parameter	Type	Description
`provider`	str	Provider name
`metric_type`	MetricType	Type of metrics to snapshot

get_snapshots

def get_snapshots(
    provider: str | None = None,
    metric_type: MetricType | None = None
) -> list[MetricsSnapshot]

source

Get recorded snapshots, optionally filtered.

Parameters

Parameter	Type	Description
`provider`	str \| None	Only snapshots for this provider (None = all).
`metric_type`	MetricType \| None	Only snapshots of this type (None = all).

Returns

Type	Description
list[MetricsSnapshot]	Copy of the matching snapshot list, oldest first.

get_summary

def get_summary() -> dict[str, Any]

source

Get a summary of all collected metrics.

Returns

Type	Description
dict[str, Any]	Dictionary with timing, error, and cache summaries

`MetricsSnapshot`

source

Snapshot of all collected metrics at a point in time.

Attributes: timestamp: When the snapshot was taken metric_type: Type of metric provider: Provider name data: The actual metric data

`MistralClient`

source

Client for Mistral AI's LLM API.

Conforms to: LLMClientProtocol protocol via structural typing.

Supports Chat, Stream, and Embeddings with:

High-performance European LLMs
GDPR compliance and data sovereignty
Function calling and JSON mode

__init__

def __init__(config: ClientConfig)

source

Initialize Mistral client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

api_key

property api_key() -> SecretStr

source

Get API key from config.

base_url

property base_url() -> str

source

Get base URL from config.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight health check against the Mistral API.

Calls the models endpoint to verify the API key is valid and the service is reachable.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for the response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

embed

async def embed(
    model: str = 'mistral-embed',
    input_texts: list[str] | str | None = None,
    **kwargs: Any
) -> list[list[float]]

source

Generate embeddings.

Parameters

Parameter	Type	Description
`model`	str	Model ID (default: "mistral-embed").
`input_texts`	list[str] \| str \| None	Text or list of texts to embed. **kwargs: Additional parameters.

Returns

Type	Description
list[list[float]]	List of embedding vectors.

Example

embeddings = await client.embed(
input_texts=["Hello world", "Bonjour monde"]
)
print(f"Embedding dimension: {len(embeddings[0])}")

async def close() -> None

source

Close the HTTP client.

Example

await client.close()

`MistralCounter`

source

Token counter using mistral-common tokenizer (lazy-loaded).

Tokenizer is loaded on first use, not at construction time.

__init__

def __init__() -> None

source

Initialize MistralCounter.

model

property model() -> str

source

Backend identifier.

count

def count(text: str) -> int

source

Count tokens in a text string.

count_messages

def count_messages(messages: Sequence[ChatMessageProtocol]) -> int

source

Count tokens in a list of chat messages.

`ModelCapabilities`

source

Model capabilities and constraints.

`ModelPinPolicy`

source

Policy for pinning LLM model revisions.

Controls how strictly the configured model_revision is enforced against the revision returned by the provider.

`ModelPricing`

source

Pricing information for a specific LLM model.

Attributes: model: Model identifier (e.g., “gpt-4-turbo”, “claude-3-opus”). prompt_per_1m: Cost per 1 million prompt tokens in USD. completion_per_1m: Cost per 1 million completion tokens in USD. provider: Provider name (e.g., “openai”, “anthropic”). last_updated: When pricing was last updated. source: Where pricing data came from (e.g., “json”, “api”, “static”).

Example

pricing = ModelPricing(
model="gpt-4-turbo",
prompt_per_1m=10.00,
completion_per_1m=30.00,
provider="openai"
)
print(f"${pricing.prompt_per_1m} per 1M prompt tokens")

serialize_model

def serialize_model(handler: Callable[[ModelPricing], dict[str, Any]]) -> dict[str, Any]

source

Custom serializer to handle datetime objects.

`ModelSelector`

source

Intelligent model selector with fallback support.

Automatically selects the best model based on prompt characteristics and provides fallback chains for reliability.

Example

selector = ModelSelector(
default_model="gpt-3.5-turbo",
strategies=[
SelectionStrategy(
name="complex",
model="gpt-4-turbo",
conditions={"min_tokens": 1000}
),
SelectionStrategy(
name="simple",
model="claude-3-haiku-20240307",
conditions={"max_tokens": 500}
)
],
fallback_chain=["gpt-4-turbo", "gpt-3.5-turbo"]
)

# Select model for a prompt
model = selector.select("Long prompt here...")
print(model)

'gpt-4-turbo'
>>>
>>> # Get next fallback on error
>>> fallback = selector.get_fallback("gpt-4-turbo")
>>> print(fallback)
'gpt-3.5-turbo'

__init__

def __init__(
    default_model: str | None = None,
    strategies: list[SelectionStrategy] | None = None,
    fallback_chain: list[str] | None = None,
    model_capabilities: dict[str, ModelCapabilities] | None = None,
    token_counter: TokenCounterProtocol | None = None
)

source

Initialize model selector.

Parameters

Parameter	Type	Description
`default_model`	str \| None	Default model to use
`strategies`	list[SelectionStrategy] \| None	List of selection strategies
`fallback_chain`	list[str] \| None	Ordered list of fallback models
`model_capabilities`	dict[str, ModelCapabilities] \| None	Custom model capabilities
`token_counter`	TokenCounterProtocol \| None	Token counter for prompt analysis

Example

selector = ModelSelector(
default_model="gpt-3.5-turbo",
fallback_chain=["gpt-4", "claude-3-sonnet-20240229"]
)

select

def select(
    prompt: str,
    context: dict[str, Any] | None = None,
    required_capabilities: list[str] | None = None
) -> str

source

Select the best model for the given prompt.

Parameters

Parameter	Type	Description
`prompt`	str	The prompt text
`context`	dict[str, Any] \| None	Additional context for selection
`required_capabilities`	list[str] \| None	Required capabilities (e.g., ["supports_functions"])

Returns

Type	Description
str	Selected model name

Example

model = selector.select(
"Analyze this image...",
required_capabilities=["supports_vision"]
)
print(model)

'gpt-4-turbo'

get_fallback

def get_fallback(failed_model: str) -> str | None

source

Get the next model in the fallback chain.

Parameters

Parameter	Type	Description
`failed_model`	str	The model that failed

Returns

Type	Description
str \| None	Next fallback model, or None if no fallback available

Example

fallback = selector.get_fallback("gpt-4-turbo")
print(fallback)

'gpt-3.5-turbo'

get_capabilities

def get_capabilities(model: str) -> ModelCapabilities | None

source

Get capabilities for a model.

Parameters

Parameter	Type	Description
`model`	str	Model name

Returns

Type	Description
ModelCapabilities \| None	Model capabilities or None if unknown

Example

caps = selector.get_capabilities("gpt-4-turbo")
print(caps.max_tokens)

estimate_cost

def estimate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int
) -> float

source

Estimate cost for a model call.

Parameters

Parameter	Type	Description
`model`	str	Model name
`input_tokens`	int	Number of input tokens
`output_tokens`	int	Number of output tokens

Returns

Type	Description
float	Estimated cost in USD

Example

cost = selector.estimate_cost("gpt-4-turbo", 1000, 500)
print(f"${cost:.4f}")

$0.0250

`OllamaClient`

source

Ollama LLM client for local models.

Conforms to: LLMClientProtocol protocol via structural typing.

Supports running LLMs locally with Ollama:

Llama 3, Mistral, Phi, and other open models
Streaming responses
Zero API costs
Full data privacy

Example

from quadkit.ai import ClientConfig
config = ClientConfig(
provider="ollama",
model="llama3:8b",
api_base="http://localhost:11434"
)
client = OllamaClient(config)
completion = await client.complete([
ChatMessage(role="user", content="Hello!")
])

__init__

def __init__(config: ClientConfig)

source

Initialize Ollama client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

Raises

Exception	Description
ImportError	If ollama package is not installed

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight health check against the Ollama daemon.

Calls list() to verify the daemon is running and reachable.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for the response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

async def close() -> None

source

Close Ollama client.

`OpenAIClient`

source

OpenAI LLM client implementation.

Conforms to: LLMClientProtocol protocol via structural typing.

Supports GPT-4, GPT-3.5-Turbo, and other OpenAI models with:

Streaming responses
Function/tool calling
Vision models
Automatic retry with exponential backoff
Error handling and rate limit management

Example

from quadkit.ai import ClientConfig
config = ClientConfig(provider="openai", model="gpt-4-turbo")
client = OpenAIClient(config)
completion = await client.complete([
ChatMessage(role="user", content="Hello!")
])

__init__

def __init__(config: ClientConfig)

source

Initialize OpenAI client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

Raises

Exception	Description
ImportError	If openai package is not installed

async def close() -> None

source

Close the OpenAI client and cleanup resources.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

`OpenRouterClient`

source

Client for OpenRouter (OpenAI-compatible) API.

Conforms to: LLMClientProtocol protocol via structural typing.

__init__

def __init__(config: ClientConfig)

source

Initialize OpenRouter client.

Parameters

Parameter	Type	Description
`config`	ClientConfig	LLM configuration

api_key

property api_key() -> SecretStr

source

Get API key from config.

base_url

property base_url() -> str

source

Get base URL from config.

model

property model() -> str

source

Get default model from config.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight health check against the OpenRouter API.

Calls the models listing endpoint to verify the API key is valid and the service is reachable.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for the response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

embeddings

async def embeddings(
    texts: list[str],
    **kwargs: Any
) -> list[list[float]]

source

async def close() -> None

source

`OpenRouterPricingSource`

source

Pricing source from the OpenRouter models API.

Fetches https://openrouter.ai/api/v1/models (no API key required) and maps each model’s per-token USD prices to ModelPricing. OpenRouter prices reflect what you actually pay for OpenAI, Anthropic, Google, Meta, Cohere, DeepSeek, xAI, and Qwen models and are updated continuously by OpenRouter.

Each model is indexed under its full slug (e.g. openai/gpt-4o) and, when unambiguous, under its bare name (e.g. gpt-4o) so callers can look up prices by either form.

Attributes: endpoint: Models API endpoint URL. timeout: Request timeout in seconds. cache: In-memory cache of fetched pricing.

Example

source = OpenRouterPricingSource()
pricing = await source.get_pricing("gpt-4o")

__init__

def __init__(
    endpoint: str | None = None,
    timeout: float = 10.0
)

source

Initialize OpenRouter pricing source.

Parameters

Parameter	Type	Description
`endpoint`	str \| None	Models API URL. Defaults to ``https://openrouter.ai/api/v1/models``.
`timeout`	float	Request timeout in seconds (default: 10).

get_pricing

async def get_pricing(model: str) -> ModelPricing | None

source

Get pricing for a specific model.

Parameters

Parameter	Type	Description
`model`	str	Model identifier (full slug or bare name).

Returns

Type	Description
ModelPricing \| None	ModelPricing if found, None otherwise.

get_all_pricing

async def get_all_pricing() -> dict[str, ModelPricing]

source

Get all pricing data.

Returns

Type	Description
dict[str, ModelPricing]	All pricing data from the OpenRouter API.

source_name

property source_name() -> str

source

Get source name.

invalidate_cache

def invalidate_cache() -> None

source

Clear cached pricing data to force refresh.

`OutputFilter`

source

Filter LLM output for sensitive information.

Prevents leaking of system prompts, internal data, etc.

filter_output

def filter_output(
    output: str,
    system_prompt: str
) -> str

source

Filter LLM output for leaks.

Parameters

Parameter	Type	Description
`output`	str	LLM output
`system_prompt`	str	System prompt (check if leaked)

Returns

Type	Description
str	Filtered output

`ParserRegistry`

source

Registry for managing output parsers by name.

Provides a central registry for looking up parsers by name, similar to LangChain’s parser registry.

Example

registry = ParserRegistry()
registry.register("json", JSONOutputParser())
parser = registry.get("json")
assert parser is not None

__init__

def __init__() -> None

source

Initialize an empty registry.

def register(
    name: str,
    parser: Any
) -> None

source

Parameters

Parameter	Type	Description
`name`	str	Unique name for the parser.
`parser`	Any	Parser instance to register.

get

def get(name: str) -> Any

source

Get a parser by name.

Parameters

Parameter	Type	Description
`name`	str	Name of the parser to retrieve.

Returns

Type	Description
Any	The registered parser.

Raises

Exception	Description
KeyError	If no parser is registered with that name.

get_or_none

def get_or_none(name: str) -> Any | None

source

Get a parser by name, returning None if not found.

Parameters

Parameter	Type	Description
`name`	str	Name of the parser to retrieve.

Returns

Type	Description
Any \| None	The registered parser, or None if not found.

list_parsers

def list_parsers() -> list[str]

source

List all registered parser names.

Returns

Type	Description
list[str]	List of registered parser names.

unregister

def unregister(name: str) -> None

source

Unregister a parser by name.

Parameters

Parameter	Type	Description
`name`	str	Name of the parser to unregister.

Raises

Exception	Description
KeyError	If no parser is registered with that name.

with_defaults

def with_defaults(cls) -> ParserRegistry

source

Create a registry with default parsers pre-registered.

`PricingConfig`

source

Pricing and cost-estimation configuration for the LLM subsystem.

When attached to ClientConfig (ai_llm.pricing section in YAML), the LLM provider registers a CostEstimatorProtocol backed by a PricingManager over the configured sources. Agents wired to the container then get real USD cost estimates per execution.

Sources are queried in the configured order; the first source that knows a model wins. When no sources are listed, defaults are used: OpenRouter (freshest prices for OpenAI/Anthropic/Google/Meta/Cohere/ DeepSeek/xAI/Qwen) then the LiteLLM model cost map (covers the long tail including Groq and Mistral).

Example

.. code-block:: yaml

ai_llm:
  pricing:
    enabled: true
    cache_ttl: 43200
    enable_fuzzy_match: true
    sources:
      - type: openrouter
      - type: litellm
      - type: json
        file_path: pricing/private.json
      - type: static
        models:
          my-internal-model:
            prompt_per_1m: 0.25
            completion_per_1m: 0.75

build_sources

def build_sources() -> list[AbstractPricingSource]

source

Build pricing source instances from this config.

Returns

Type	Description
list[AbstractPricingSource]	Configured ``AbstractPricingSource`` instances in priority order. Empty sources list yields the defaults (OpenRouter then LiteLLM).

Raises

Exception	Description
ValueError	On unknown source type or a ``json` `source without` `file_path``.

`PricingCostEstimator`

source

Estimate USD cost of LLM usage from a pricing snapshot.

Implements CostEstimatorProtocol.estimate_cost. Model lookup is exact (case-insensitive), then a substring match in both directions (mirrors PricingManager fuzzy matching) when enabled. Unknown models price at 0.0 — callers must skip cost tracking rather than fabricate estimates.

Attributes: pricing: Snapshot of model name to pricing data. enable_fuzzy_match: Allow substring model name matching.

Example

estimator = PricingCostEstimator(
{"gpt-4o": ModelPricing(model="gpt-4o", prompt_per_1m=2.5,
completion_per_1m=10.0, provider="openai")}
)
estimator.estimate_cost("gpt-4o", 1500, prompt_tokens=1000,
completion_tokens=500)

0.0075

__init__

def __init__(
    pricing: dict[str, ModelPricing],
    *,
    enable_fuzzy_match: bool = True
) -> None

source

Initialize the estimator.

Parameters

Parameter	Type	Description
`pricing`	dict[str, ModelPricing]	Preloaded snapshot of model name to pricing data.
`enable_fuzzy_match`	bool	Allow substring model name matching (default: True).

warm

async def warm(manager: PricingManager) -> None

source

Reload the pricing snapshot from a manager.

Aggregates all sources via PricingManager.preload; source failures degrade to empty entries, which price at 0.0.

Parameters

Parameter	Type	Description
`manager`	PricingManager	Pricing manager to load from.

estimate_cost

def estimate_cost(
    model: str,
    total_tokens: int,
    provider: str | None = None,
    prompt_tokens: int = 0,
    completion_tokens: int = 0
) -> float

source

Estimate cost in USD for the given token usage.

When prompt_tokens and completion_tokens are known they are priced at the input and output rates. When both are 0 the split is unknown and total_tokens is priced at the input rate — prompt tokens dominate agent turns, so this is the closest approximation available.

Parameters

Parameter	Type	Description
`model`	str	Model identifier (e.g. ``gpt-4o``).
`total_tokens`	int	Total tokens consumed (prompt + completion).
`provider`	str \| None	Provider name (e.g. ``openai``) — currently unused, pricing is keyed by model.
`prompt_tokens`	int	Input token count. ``0`` means unknown.
`completion_tokens`	int	Output token count. ``0`` means unknown.

Returns

Type	Description
float	Estimated cost in USD. ``0.0`` when the model has no pricing.

Example

estimator.estimate_cost("gpt-4o", 1500, prompt_tokens=1000,
completion_tokens=500)

0.0075

`PricingManager`

source

Manages pricing data from multiple sources with caching.

Sources are queried in order until pricing is found. Typical hierarchy:

JSON file (fastest, most reliable)
API endpoints (for updates)
Static fallback (hardcoded)

Attributes: sources: List of pricing sources in priority order. cache: Pricing cache instance. enable_fuzzy_match: Whether to enable fuzzy model name matching.

Example

# Use defaults
manager = PricingManager.from_defaults()

# Custom configuration
manager = (
PricingManager.builder()
.add_json_source("pricing.json")
.add_api_source("https://api.example.com/pricing")
.with_cache_ttl(3600)
.enable_fuzzy_matching()
.build()
)

pricing = await manager.get_pricing("gpt-4-turbo")

__init__

def __init__(
    sources: Sequence[AbstractPricingSource],
    cache_ttl: int = 86400,
    enable_fuzzy_match: bool = True
)

source

Initialize pricing manager.

Parameters

Parameter	Type	Description
`sources`	Sequence[AbstractPricingSource]	List of pricing sources in priority order.
`cache_ttl`	int	Cache TTL in seconds (default: 24 hours).
`enable_fuzzy_match`	bool	Enable fuzzy model name matching (default: True).

get_pricing

async def get_pricing(
    model: str,
    force_refresh: bool = False
) -> ModelPricing | None

source

Get pricing for a specific model.

Queries sources in order:

Cache (if not force_refresh)
Each source in priority order
Fuzzy match if enabled
None if not found

Parameters

Parameter	Type	Description
`model`	str	Model identifier (e.g., "gpt-4-turbo").
`force_refresh`	bool	Bypass cache and fetch fresh data.

Returns

Type	Description
ModelPricing \| None	ModelPricing if found, None otherwise.

list_models

async def list_models(provider: str | None = None) -> list[str]

source

List all available models.

Parameters

Parameter	Type	Description
`provider`	str \| None	Filter by provider (optional).

Returns

Type	Description
list[str]	List of model names.

clear_cache

async def clear_cache() -> None

source

Clear pricing cache.

preload

async def preload() -> dict[str, ModelPricing]

source

Load all pricing from all sources into one merged map.

Earlier sources win on duplicate model names, mirroring get_pricing priority semantics. Used to build the synchronous snapshot for cost estimators.

Returns

Type	Description
dict[str, ModelPricing]	Merged dictionary of model name to pricing.

from_defaults

def from_defaults(cls) -> PricingManager

source

Create manager with default configuration.

Uses LiteLLM API for dynamic, up-to-date pricing data. No static pricing files - always fetches current data.

Returns

Type	Description
PricingManager	PricingManager with API source.

Example

manager = PricingManager.from_defaults()
pricing = await manager.get_pricing("gpt-4")

from_json

def from_json(
    cls,
    file_path: str | Path,
    cache_ttl: int = 86400
) -> PricingManager

source

Create manager from JSON file only.

Useful for offline applications or when you want full control over pricing data.

Parameters

Parameter	Type	Description
`file_path`	str \| Path	Path to JSON pricing file.
`cache_ttl`	int	Cache TTL in seconds (default: 24 hours).

Returns

Type	Description
PricingManager	PricingManager with JSON source only.

Example

manager = PricingManager.from_json("my_pricing.json")
pricing = await manager.get_pricing("custom-model")

from_api

def from_api(
    cls,
    endpoint: str,
    cache_ttl: int = 86400
) -> PricingManager

source

Create manager from API endpoint only.

Parameters

Parameter	Type	Description
`endpoint`	str	API endpoint URL.
`cache_ttl`	int	Cache TTL in seconds (default: 24 hours).

Returns

Type	Description
PricingManager	PricingManager with API source only.

Example

manager = PricingManager.from_api("https://api.example.com/pricing")
pricing = await manager.get_pricing("gpt-4")

builder

def builder(cls) -> PricingManagerBuilder

source

Create a builder for custom configuration.

Returns

Type	Description
PricingManagerBuilder	PricingManagerBuilder instance.

Example

manager = (
PricingManager.builder()
.add_json_source("custom.json")
.add_api_source("https://api.example.com")
.with_cache_ttl(3600)
.build()
)

`PricingManagerBuilder`

source

Builder for PricingManager with validation.

Provides a fluent API for configuring pricing sources safely.

Example

manager = (
PricingManager.builder()
.add_json_source("pricing.json")
.add_api_source("https://api.example.com/pricing")
.add_fallback({"custom-model": ModelPricing(...)})
.with_cache_ttl(3600)
.enable_fuzzy_matching()
.build()
)

__init__

def __init__() -> None

source

Initialize builder.

add_json_source

def add_json_source(file_path: str | Path) -> PricingManagerBuilder

source

Add JSON file pricing source.

Parameters

Parameter	Type	Description
`file_path`	str \| Path	Path to JSON file.

Returns

Type	Description
PricingManagerBuilder	Self for chaining.

add_api_source

def add_api_source(
    endpoint: str,
    timeout: float = 10.0
) -> PricingManagerBuilder

source

Add API endpoint pricing source.

Parameters

Parameter	Type	Description
`endpoint`	str	API endpoint URL.
`timeout`	float	Request timeout in seconds (default: 10).

Returns

Type	Description
PricingManagerBuilder	Self for chaining.

add_fallback

def add_fallback(pricing_map: dict[str, ModelPricing]) -> PricingManagerBuilder

source

Add static fallback pricing.

Parameters

Parameter	Type	Description
`pricing_map`	dict[str, ModelPricing]	Dictionary of model to pricing.

Returns

Type	Description
PricingManagerBuilder	Self for chaining.

add_source

def add_source(source: AbstractPricingSource) -> PricingManagerBuilder

source

Add custom pricing source.

Parameters

Parameter	Type	Description
`source`	AbstractPricingSource	Custom AbstractPricingSource implementation.

Returns

Type	Description
PricingManagerBuilder	Self for chaining.

with_cache_ttl

def with_cache_ttl(seconds: int) -> PricingManagerBuilder

source

Set cache TTL.

Parameters

Parameter	Type	Description
`seconds`	int	Cache TTL in seconds.

Returns

Type	Description
PricingManagerBuilder	Self for chaining.

Raises

Exception	Description
ValueError	If seconds is negative.

enable_fuzzy_matching

def enable_fuzzy_matching(enabled: bool = True) -> PricingManagerBuilder

source

Enable or disable fuzzy model name matching.

Parameters

Parameter	Type	Description
`enabled`	bool	Whether to enable fuzzy matching (default: True).

Returns

Type	Description
PricingManagerBuilder	Self for chaining.

build

def build() -> PricingManager

source

Build PricingManager instance.

Returns

Type	Description
PricingManager	Configured PricingManager.

Raises

Exception	Description
ValueError	If no sources were added.

`PricingSourceConfig`

source

A single pricing data source, configurable from YAML.

Attributes: type: Source kind. One of "litellm", "openrouter", "json", or "static". endpoint: API endpoint URL. Used by litellm (defaults to the LiteLLM model cost map on GitHub) and openrouter (defaults to https://openrouter.ai/api/v1/models). file_path: Path to a local pricing JSON file (json type only). timeout: HTTP timeout in seconds for API sources. models: Inline static prices (static type only). Maps a model name to {prompt_per_1m, completion_per_1m, provider?}.

Example

.. code-block:: yaml

ai_llm:
  pricing:
    enabled: true
    sources:
      - type: openrouter
      - type: litellm
      - type: json
        file_path: pricing/custom.json
      - type: static
        models:
          internal-model:
            prompt_per_1m: 0.5
            completion_per_1m: 1.5
            provider: custom

`ProviderConfig`

source

Configuration for a single provider in the routing cascade.

Every provider in the cascade has the same shape regardless of type. Provider-specific fields (Azure deployment, Cloudflare account ID, Bedrock region, Vertex project) go in extras.

Example

cfg = ProviderConfig(
name="groq",
model="llama-3.3-70b-versatile",
api_key="gsk_...",
)

key

property key() -> str

source

Unique cascade-entry identity: provider name + model.

Multiple entries may share name (e.g. several OpenRouter models); quota, clients, and routing state must key on this, never on name.

`ProviderInfo`

source

Information about an LLM provider.

Attributes: name: Provider identifier (e.g., “openai”, “anthropic”). client_class: LLMClientProtocol implementation class. default_models: List of default/recommended models. supports_streaming: Whether streaming is supported. supports_tools: Whether function/tool calling is supported. supports_vision: Whether vision/image inputs are supported. base_url: Default base URL for API (optional). docs_url: Documentation URL (optional). pricing_url: Pricing page URL (optional). description: Human-readable description.

`ProviderRegistry`

source

Registry for LLM providers.

Instances are always empty — use with_defaults for the in-package built-in catalogue or register for custom providers.

__init__

def __init__() -> None

source

Initialize an empty provider registry.

with_defaults

def with_defaults(cls) -> ProviderRegistry

source

Return an instance populated with the built-in providers.

def register(
    name: str,
    client_class: type[object],
    default_models: list[str] | None = None,
    supports_streaming: bool = True,
    supports_tools: bool = False,
    supports_vision: bool = False,
    base_url: str | None = None,
    docs_url: str | None = None,
    pricing_url: str | None = None,
    description: str = ''
) -> ProviderInfo

source

get_provider

def get_provider(name: str) -> ProviderInfo

source

Get provider information.

list_providers

def list_providers() -> list[str]

source

List all registered provider names.

search_providers

def search_providers(
    supports_streaming: bool | None = None,
    supports_tools: bool | None = None,
    supports_vision: bool | None = None
) -> list[ProviderInfo]

source

Search providers by capabilities.

unregister

def unregister(name: str) -> None

source

Unregister a provider.

register_provider

async def register_provider(
    name: str,
    client: LLMClientProtocol,
    models: list[ModelInfo]
) -> None

source

get_client

async def get_client(provider: str) -> LLMClientProtocol | None

source

Get an initialized client for a provider.

list_models

def list_models(capabilities: set[ModelCapability] | None = None) -> list[ModelInfo]

source

List all models matching capabilities.

get_model_info

def get_model_info(model_id: str) -> ModelInfo | None

source

Get information about a specific model.

`PydanticOutputParser`

source

Parse LLM JSON responses into typed ``DomainModel`` or Pydantic models.

__init__

def __init__(model: type[TModel]) -> None

source

parse

def parse(text: str) -> TModel

source

Parse text into a validated model instance.

get_format_instructions

def get_format_instructions() -> str

source

Return format instructions derived from model_json_schema().

`QuotaConfig`

source

Configuration for the quota tracking backend.

Example

cfg = QuotaConfig(backend="database")

`RateLimiter`

source

Rate limiter for LLM requests (RPM and TPM).

Manages multiple buckets for different models and providers.

__init__

def __init__() -> None

source

Initialize rate limiter.

check

async def check(
    provider: str,
    model: str,
    tpm_limit: int | None = None,
    rpm_limit: int | None = None,
    estimated_tokens: int = 0
) -> bool

source

Check if request is allowed under current limits.

Parameters

Parameter	Type	Description
`provider`	str	AI provider name
`model`	str	Model name
`tpm_limit`	int \| None	Tokens Per Minute limit
`rpm_limit`	int \| None	Requests Per Minute limit
`estimated_tokens`	int	Estimated tokens in request

Returns

Type	Description
bool	True if allowed, False if blocked

`RedisLLMCache`

source

Redis-backed cache for distributed deployments.

Requires redis package to be installed.

Parameters

Parameter	Type	Description
`redis_url`		Redis connection URL.
`ttl`		Time-to-live in seconds.
`key_prefix`		Prefix for all cache keys.

Example

cache = RedisLLMCache(redis_url="redis://localhost:6379")
await cache.connect()
result = await cache.get("key")

__init__

def __init__(
    cache_backend: CacheBackendProtocol,
    ttl: float = 3600,
    key_prefix: str = 'llm_cache:'
)

source

Initialize Redis cache.

Parameters

Parameter	Type	Description
`cache_backend`	CacheBackendProtocol	The platform's cache backend.
`ttl`	float	Time-to-live in seconds.
`key_prefix`	str	Prefix for cache keys.

connect

async def connect() -> None

source

Compatibility method for lifecycle-managed cache.

disconnect

async def disconnect() -> None

source

Compatibility method for lifecycle-managed cache.

get

async def get(key: str | dict[str, Any]) -> Any | None

source

Get value from Redis cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key.

Returns

Type	Description
Any \| None	Cached value or None.

set

async def set(
    key: str | dict[str, Any],
    value: Any,
    ttl: float | None = None
) -> None

source

Set value in Redis cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key.
`value`	Any	Value to cache.
`ttl`	float \| None	Optional TTL override.

get_or_compute

async def get_or_compute(
    key: str | dict[str, Any],
    compute_fn: Callable[[], Any],
    ttl: float | None = None
) -> Any

source

Get from cache or compute and cache result.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key.
`compute_fn`	Callable[[], Any]	Function to compute value if cache miss.
`ttl`	float \| None	Optional TTL override.

Returns

Type	Description
Any	Cached or computed value.

delete

async def delete(key: str | dict[str, Any]) -> bool

source

Delete entry from cache.

Parameters

Parameter	Type	Description
`key`	str \| dict[str, Any]	Cache key.

Returns

Type	Description
bool	True if deleted.

clear

async def clear() -> None

source

Clear all cache entries (Warning: clears entire backend if not namespaced).

get_stats

def get_stats() -> CacheStats

source

Get cache statistics.

Returns

Type	Description
CacheStats	CacheStats object.

`ResponseFormatter`

source

Format and convert LLM responses to various types.

Example

formatter = ResponseFormatter()
completion = Completion(content="42", ...)
num = formatter.to_int(completion)
print(num)

to_json

def to_json(completion: Completion) -> JSON

source

Convert response to JSON.

Parameters

Parameter	Type	Description
`completion`	Completion	LLM completion

Returns

Type	Description
JSON	Parsed JSON

Example

data = formatter.to_json(completion)

to_string

def to_string(
    completion: Completion,
    strip: bool = True
) -> str

source

Convert response to string.

Parameters

Parameter	Type	Description
`completion`	Completion	LLM completion
`strip`	bool	Whether to strip whitespace

Returns

Type	Description
str	Response string

Example

text = formatter.to_string(completion)

to_int

def to_int(completion: Completion) -> int

source

Convert response to integer.

Parameters

Parameter	Type	Description
`completion`	Completion	LLM completion

Returns

Type	Description
int	Parsed integer

Raises

Exception	Description
ParseError	If conversion fails

Example

num = formatter.to_int(completion)

to_float

def to_float(completion: Completion) -> float

source

Convert response to float.

Parameters

Parameter	Type	Description
`completion`	Completion	LLM completion

Returns

Type	Description
float	Parsed float

Raises

Exception	Description
ParseError	If conversion fails

Example

num = formatter.to_float(completion)

to_bool

def to_bool(completion: Completion) -> bool

source

Convert response to boolean.

Parameters

Parameter	Type	Description
`completion`	Completion	LLM completion

Returns

Type	Description
bool	Parsed boolean

Example

result = formatter.to_bool(completion)

to_list

def to_list(
    completion: Completion,
    separator: str = '\n'
) -> list[str]

source

Convert response to list of strings.

Parameters

Parameter	Type	Description
`completion`	Completion	LLM completion
`separator`	str	String separator (default: newline)

Returns

Type	Description
list[str]	List of strings

Example

items = formatter.to_list(completion)

`Role`

source

Concrete chat message role constants shared across AI packages.

`RunnableBranch`

source

Route input to different runnables based on a predicate.

Like LangChain’s RunnableBranch, this evaluates predicates to select which branch to execute.

__init__

def __init__(
    branches: list[tuple[Callable[[Any], bool], RunnableProtocol]],
    default: RunnableProtocol | None = None
) -> None

source

invoke

def invoke(input: Any) -> Any

source

Synchronously route to matching branch.

Parameters

Parameter	Type	Description
`input`	Any	Input to evaluate against predicates.

Returns

Type	Description
Any	Output from the first matching branch, or default if no match.

ainvoke

async def ainvoke(input: Any) -> Any

source

Asynchronously route to matching branch.

Parameters

Parameter	Type	Description
`input`	Any	Input to evaluate against predicates.

Returns

Type	Description
Any	Output from the first matching branch, or default if no match.

`RunnableLambda`

source

Wrap a function as a runnable.

Accepts sync or async functions and wraps them to satisfy RunnableProtocol. Failures become Err(RunnableError(…)).

Parameters

Parameter	Type	Description
`func`		A sync or async function to wrap.

__init__

def __init__(func: Callable[[Any], Any]) -> None

source

invoke

def invoke(input: Any) -> Any

source

Synchronously invoke the wrapped function.

Parameters

Parameter	Type	Description
`input`	Any	Input to the function.

Returns

Type	Description
Any	Function output or Err on failure.

ainvoke

async def ainvoke(input: Any) -> Any

source

Asynchronously invoke the wrapped function.

Parameters

Parameter	Type	Description
`input`	Any	Input to the function.

Returns

Type	Description
Any	Function output or Err on failure.

`RunnableMixin`

source

Mixin that adds pipe operator to runnables.

Provides the | operator that composes runnables into RunnableSequence. Analogous to LangChain’s RunnableBinding.

invoke

def invoke(input: Any) -> Any

source

Process input synchronously. Override in subclass.

ainvoke

async def ainvoke(input: Any) -> Any

source

Process input asynchronously. Override in subclass.

`RunnableParallel`

source

Run multiple runnables concurrently.

Each runnable receives the same input and results are returned as a dict.

__init__

def __init__(**runnables: RunnableProtocol) -> None

source

invoke

def invoke(input: Any) -> dict[str, Any]

source

Synchronously invoke all runnables.

Parameters

Parameter	Type	Description
`input`	Any	Input to pass to all runnables.

Returns

Type	Description
dict[str, Any]	Dict mapping names to outputs.

ainvoke

async def ainvoke(input: Any) -> dict[str, Any]

source

Asynchronously invoke all runnables concurrently.

Parameters

Parameter	Type	Description
`input`	Any	Input to pass to all runnables.

Returns

Type	Description
dict[str, Any]	Dict mapping names to outputs.

`RunnablePassthrough`

source

Pass input through with optional key assignment.

Returns the input unchanged but can assign it to a key in the output dict. Useful for combining with RunnableParallel.

__init__

def __init__(name: str | None = None) -> None

source

invoke

def invoke(input: Any) -> Any

source

Return input, optionally wrapped in dict with named key.

Parameters

Parameter	Type	Description
`input`	Any	Input to pass through.

Returns

Type	Description
Any	Input as-is, or dict with named key if name is set.

ainvoke

async def ainvoke(input: Any) -> Any

source

Return input, optionally wrapped in dict with named key.

Parameters

Parameter	Type	Description
`input`	Any	Input to pass through.

Returns

Type	Description
Any	Input as-is, or dict with named key if name is set.

`RunnableSequence`

source

Chain multiple runnables in sequence.

The output of each runnable becomes the input to the next. Short-circuits on Err results.

__init__

def __init__(
    first: RunnableProtocol,
    second: RunnableProtocol
) -> None

source

invoke

def invoke(input: Any) -> Any

source

Synchronously invoke the chain.

Parameters

Parameter	Type	Description
`input`	Any	Input to the first runnable.

Returns

Type	Description
Any	Output from the last runnable, or Err if any step fails.

ainvoke

async def ainvoke(input: Any) -> Any

source

Asynchronously invoke the chain.

Parameters

Parameter	Type	Description
`input`	Any	Input to the first runnable.

Returns

Type	Description
Any	Output from the last runnable, or Err if any step fails.

`SecureLLMClient`

source

LLM client with injection protection and safety features.

__init__

def __init__(
    llm_provider: Annotated[LLMClientProtocol, Inject],
    system_prompt: str = 'You are a helpful assistant.',
    enable_output_filtering: bool = True,
    rate_limiter: Annotated[RateLimiter | None, Inject] = None,
    rpm_limit: int = 60
) -> None

source

Initialize secure LLM client.

Parameters

Parameter	Type	Description
`llm_provider`	Annotated[LLMClientProtocol, Inject]	Underlying LLM provider (injected)
`system_prompt`	str	System prompt template
`enable_output_filtering`	bool	Enable output filtering

chat

async def chat(
    user_input: str,
    user_id: str,
    context: Sequence[dict[str, str]] | None = None,
    strict_validation: bool = True
) -> str

source

Send chat message with safety protections.

Parameters

Parameter	Type	Description
`user_input`	str	User message
`user_id`	str	User identifier (for rate limiting)
`context`	Sequence[dict[str, str]] \| None	Previous conversation context
`strict_validation`	bool	Reject invalid input vs sanitize

Returns

Type	Description
str	LLM response

Raises

Exception	Description
ValueError	If input invalid (strict mode)

update_system_prompt

def update_system_prompt(system_prompt: str) -> None

source

Update system prompt.

Parameters

Parameter	Type	Description
`system_prompt`	str	New system prompt

`SecurePromptTemplate`

source

Structured prompt template with injection protection.

Uses clear delimiters to separate system instructions from user input. Implements multi-layered injection detection.

detect_injection

def detect_injection(prompt: str) -> tuple[bool, list[str]]

source

Multi-layered injection detection.

Parameters

Parameter	Type	Description
`prompt`	str	Input to analyze

Returns

Type	Description
tuple[bool, list[str]]	Tuple of (is_malicious, reasons)

validate_input

def validate_input(user_input: str) -> tuple[bool, str | None]

source

Validate user input for injection attempts.

Parameters

Parameter	Type	Description
`user_input`	str	User input to validate

Returns

Type	Description
tuple[bool, str \| None]	Tuple of (is_valid, error_message)

sanitize_input

def sanitize_input(user_input: str) -> str

source

Sanitize user input by removing dangerous patterns.

Parameters

Parameter	Type	Description
`user_input`	str	User input to sanitize

Returns

Type	Description
str	Sanitized input

format

def format(
    user_input: str,
    strict: bool = True
) -> str

source

Format prompt with user input.

Parameters

Parameter	Type	Description
`user_input`	str	User input
`strict`	bool	If True, reject invalid input. If False, sanitize.

Returns

Type	Description
str	Formatted prompt

Raises

Exception	Description
ValueError	If input invalid and strict=True

`SelectionCriteria`

source

Criteria for model selection.

`SelectionStrategy`

source

Strategy for selecting models based on conditions.

Example

strategy = SelectionStrategy(
name="long_context",
model="gpt-4-turbo-preview",
conditions={
"min_tokens": 2000,
"max_tokens": 100000
}
)

matches

def matches(context: dict[str, Any]) -> bool

source

Check if this strategy matches the given context.

Parameters

Parameter	Type	Description
`context`	dict[str, Any]	Context dictionary with prompt info

Returns

Type	Description
bool	True if all conditions are met

Example

context = {"token_count": 2500, "has_code": True}
strategy.matches(context)

True

`StreamChunk`

source

A chunk of streamed completion.

Implements streaming semantics with DomainModel for validation.

Example

chunk = StreamChunk(delta="Hello", model="gpt-4-turbo", finish_reason=None)

`StructuredOutputParser`

source

Schema-aware parser that validates LLM responses against a model.

Wraps extract_json_block, validate_against_model, and build_json_schema into a convenient class-based API.

Parameters

Parameter	Type	Description
`output_model`		Model class for validation.
`strict`		Whether to enforce strict validation (default ``True``).

__init__

def __init__(
    output_model: type[Any],
    *,
    strict: bool = True
) -> None

source

Initialise with model class.

parse

def parse(completion: Any) -> Any

source

Parse and validate a completion into an output_model instance.

Parameters

Parameter	Type	Description
`completion`	Any	Completion object with ``.content`` attribute, or a string.

Returns

Type	Description
Any	Validated model instance.

Raises

Exception	Description
ParseError	When JSON cannot be extracted.
SchemaValidationError	When validation fails.

parse_array

def parse_array(completion: Any) -> list[Any]

source

Parse and validate an array of output_model instances.

Parameters

Parameter	Type	Description
`completion`	Any	Completion object with ``.content`` attribute.

Returns

Type	Description
list[Any]	List of validated model instances.

Raises

Exception	Description
ParseError	When JSON is not an array.
SchemaValidationError	When validation fails.

get_json_schema

def get_json_schema() -> dict[str, Any]

source

Return JSON Schema dict for the output model.

get_schema_prompt

def get_schema_prompt() -> str

source

Return a human-readable schema prompt string.

`TextPart`

source

A plain-text content part in a multimodal message.

Attributes: text: The text content. type: Discriminator field, always "text".

`TiktokenCounter`

source

Token counter using tiktoken (OpenAI/compatible models).

Implements TokenCounterProtocol using tiktoken for precise counting. tiktoken is a required dependency for this counter.

Parameters

Parameter	Type	Description
`model`		Model name (e.g. 'gpt-4', 'gpt-3.5-turbo').
`encoding_name`		Optional tiktoken encoding name override.

__init__

def __init__(
    model: str = 'gpt-3.5-turbo',
    encoding_name: str | None = None
) -> None

source

Initialize TiktokenCounter.

Parameters

Parameter	Type	Description
`model`	str	Model name for token counting.
`encoding_name`	str \| None	Optional tiktoken encoding name override.

Raises

Exception	Description
ImportError	If tiktoken is not installed.

model

property model() -> str

source

The model this counter is calibrated for.

count

def count(text: str) -> int

source

Count tokens in a text string.

count_messages

def count_messages(messages: Sequence[ChatMessageProtocol]) -> int

source

Count tokens in a list of chat messages, including overhead.

`TimingMetrics`

source

Latency and timing metrics for a provider.

Attributes: provider: Provider name model_id: Model identifier request_count: Total number of requests total_requests: Total number of requests (alias for request_count) error_count: Total number of errors min_latency_ms: Minimum request latency in milliseconds max_latency_ms: Maximum request latency in milliseconds avg_latency_ms: Average request latency in milliseconds p95_latency_ms: 95th percentile latency p99_latency_ms: 99th percentile latency

total_requests

property total_requests() -> int

source

Total number of requests (live alias for request_count).

record_latency

def record_latency(latency_ms: float) -> None

source

Record a request latency.

Parameters

Parameter	Type	Description
`latency_ms`	float	Latency in milliseconds

`TokenCount`

source

Token count result with metadata.

Attributes: total: Total number of tokens. prompt_tokens: Number of tokens in the prompt. completion_tokens: Number of tokens in the completion (if applicable). model: Model name used for counting. timestamp: When the count was performed.

`TokenCounterRegistry`

source

Registry mapping model-name patterns to TokenCounterProtocol backends.

Uses named backend keys and regex patterns for flexible model mapping.

Usage

registry = TokenCounterRegistry.with_defaults()
counter = registry.for_model("gpt-4o")
tokens = counter.count("Hello!")

registry = TokenCounterRegistry.with_defaults()
counter = registry.for_model("gpt-4o")
tokens = counter.count("Hello!")

__init__

def __init__() -> None

source

Create an empty registry.

with_defaults

def with_defaults(cls) -> TokenCounterRegistry

source

Create registry with all available tokenizer backends.

Registers every backend declared by _default_entries and maps the model families each backend serves.

def register(
    key: str,
    counter: TokenCounterProtocol
) -> None

source

Parameters

Parameter	Type	Description
`key`	str	Backend name (e.g., 'tiktoken', 'huggingface', 'char_estimate').
`counter`	TokenCounterProtocol	Counter implementing TokenCounterProtocol.

map_models

def map_models(
    pattern: str,
    counter_key: str
) -> None

source

Map a regex pattern of model names to a backend key.

Parameters

Parameter	Type	Description
`pattern`	str	Regex pattern matching model names (case-insensitive).
`counter_key`	str	Backend key (must be registered).

for_model

def for_model(model: str) -> TokenCounterProtocol

source

Get the best counter for the given model name.

Tries exact regex match in _patterns first, falls back to ‘char_estimate’.

Parameters

Parameter	Type	Description
`model`	str	Model name.

Returns

Type	Description
TokenCounterProtocol	TokenCounterProtocol implementation.

`TokenUsage`

source

Token usage statistics.

`ToolCall`

source

Tool call request from LLM.

`UsageMetrics`

source

Token usage and cost tracking for a provider/model pair.

Attributes: provider: Provider name model_id: Model identifier prompt_tokens: Total prompt tokens completion_tokens: Total completion tokens cost_usd: Total estimated cost in US dollars

total_tokens

property total_tokens() -> int

source

Total tokens (prompt + completion).

Functions

`complete_with_json`

complete_with_json

async def complete_with_json(
    client: LLMClientProtocol,
    prompt: str,
    system_prompt: str | None = None,
    **kwargs: Any
) -> JSON

source

Complete and parse response as JSON.

Parameters

Parameter	Type	Description
`client`	LLMClientProtocol	LLM client
`prompt`	str	User prompt
`system_prompt`	str \| None	Optional system prompt **kwargs: Additional completion arguments

Returns

Type	Description
JSON	Parsed JSON

Example

data = await complete_with_json(
client,
"Generate a config with 3 fields"
)

`complete_with_schema`

complete_with_schema

async def complete_with_schema(
    client: LLMClientProtocol,
    prompt: str,
    schema: type[T],
    system_prompt: str | None = None,
    **kwargs: Any
) -> T

source

Complete with automatic schema parsing and validation.

Parameters

Parameter	Type	Description
`client`	LLMClientProtocol	LLM client
`prompt`	str	User prompt
`schema`	type[T]	Pydantic model for validation
`system_prompt`	str \| None	Optional system prompt **kwargs: Additional completion arguments

Returns

Type	Description
T	Validated schema instance

Example

from quadkit.ai.llm import OpenAIClient

client = OpenAIClient(api_key="sk-...")
person = await complete_with_schema(
client,
"Extract person from: John Doe, age 30",
schema=Person
)

`create_assistant_template`

create_assistant_template

def create_assistant_template() -> SecurePromptTemplate

source

Create template for general assistant.

Returns

Type	Description
SecurePromptTemplate	Configured template

`create_balanced_selector`

create_balanced_selector

def create_balanced_selector() -> ModelSelector

source

Create a balanced model selector.

`create_cost_optimized_selector`

create_cost_optimized_selector

def create_cost_optimized_selector(budget_per_1k_tokens: float = 2.0) -> ModelSelector

source

Create a cost-optimized model selector.

`create_data_extraction_template`

create_data_extraction_template

def create_data_extraction_template() -> SecurePromptTemplate

source

Create template for data extraction (high security).

Returns

Type	Description
SecurePromptTemplate	Configured template

`create_json_mode_messages`

create_json_mode_messages

def create_json_mode_messages(
    prompt: str,
    schema: type[DomainModel] | None = None,
    system_prompt: str | None = None
) -> list[dict[str, str]]

source

Create messages for JSON mode with optional schema.

Parameters

Parameter	Type	Description
`prompt`	str	User prompt
`schema`	type[DomainModel] \| None	Optional Pydantic model for schema
`system_prompt`	str \| None	Optional system prompt (default: JSON instruction)

Returns

Type	Description
list[dict[str, str]]	Messages list for LLM

Example

messages = create_json_mode_messages(
"Extract person info",
schema=Person
)

`create_quality_optimized_selector`

create_quality_optimized_selector

def create_quality_optimized_selector() -> ModelSelector

source

Create a quality-optimized model selector.

`create_token_counter`

create_token_counter

def create_token_counter(
    model: str = 'gpt-3.5-turbo',
    encoding_name: str | None = None
) -> TiktokenCounter

source

Factory function for creating token counters.

Parameters

Parameter	Type	Description
`model`	str	Model name.
`encoding_name`	str \| None	Optional encoding name override.

Returns

Type	Description
TiktokenCounter	TiktokenCounter instance.

Example

from quadkit.ai.llm import create_token_counter

counter = create_token_counter("gpt-4")
count = counter.count("Hello!")
print(count)

`enforce_pin_policy`

enforce_pin_policy

def enforce_pin_policy(
    expected_revision: str | None,
    actual_revision: str,
    policy: ModelPinPolicy
) -> None

source

Enforce a model pin policy.

Parameters

Parameter	Type	Description
`expected_revision`	str \| None	The configured pinned revision, or ``None``.
`actual_revision`	str	The revision returned by the provider.
`policy`	ModelPinPolicy	The pin policy to enforce.

Raises

Exception	Description
ModelRevisionMismatchError	When the policy is violated.

`normalize_thinking_text`

normalize_thinking_text

def normalize_thinking_text(text: str) -> tuple[str, str | None]

source

Extract thinking text from raw LLM output.

Tries each pattern in THINKING_PATTERNS order. Returns (clean_content, thinking_text_or_None). clean_content has thinking block removed and is stripped. thinking_text is the raw thinking content (stripped), or None if not found.

Pattern matching is by substring presence of start_marker (and end_marker after it), NOT by model name. The bare-closing-tag pattern (end_marker="", no start) matches only when start_marker is NOT found but end_marker IS found — this covers models that output …thinking…\nresponse.

Falls back: after removing a thinking block, if clean_content is empty but thinking text was found, tries to extract from the first { or [ in the original text to recover any JSON that may have been embedded.

Parameters

Parameter	Type	Description
`text`	str	Raw LLM response text, possibly containing inline thinking tags.

Returns

Type	Description
tuple[str, str \| None]	A tuple of (clean_content, thinking_text_or_None). - clean_content: The response text with thinking stripped out, stripped of whitespace. - thinking_text_or_None: The thinking/reasoning text, or None if no thinking found.

Exceptions

`ExtractionError`

source

Base class for structured extraction errors in quadkit-ai-llm.

`ExtractionMaxRetriesError`

source

Error raised when extraction max retries are exhausted.

`ExtractionParseError`

source

Error raised when extraction response cannot be parsed as JSON.

`ExtractionValidationError`

source

Error raised when parsed extraction response fails schema validation.

`InvalidRequestError`

source

Error raised when a request to an LLM provider is invalid.

`LLMAuthenticationError`

source

Invalid API key or credentials — infrastructure error, raised not wrapped.

Raised as an exception (NOT wrapped in Result). Indicates a misconfiguration the application cannot route around.

`LLMContentFilterError`

source

Content blocked by provider safety filter — recoverable via reformulation.

Returned as Err from LLMClientProtocol.complete() / stream_chat(). The caller should reformulate the prompt or inform the user.

`LLMError`

source

Base exception for all LLM-domain errors in quadkit-ai-llm.

`LLMModelNotFoundError`

source

Model unavailable or not found — recoverable via fallback routing.

Returned as Err from LLMClientProtocol.complete() / stream_chat(). The caller should route to a different model or provider.

`LLMQuotaExceededError`

source

API quota or billing limit exceeded — recoverable by routing elsewhere.

Returned as Err from LLMClientProtocol.complete() / stream_chat(). The caller should route the request to a different provider or account.

`LLMRateLimitError`

source

Rate limit exceeded — recoverable via backoff/retry.

Returned as Err from LLMClientProtocol.complete() / stream_chat(). The caller should implement exponential backoff or route to another provider.