API Reference

Protocols

LLMCacheProtocol

source

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 lexigram.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

close

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. expires_at: When entry expires (Unix timestamp). hits: Number of cache hits. size_bytes: Approximate size in bytes.

---

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: list[ChatMessage]) -> 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.

close

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 lexigram.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 lexigram 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 lexigram 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.

---

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.

close

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: list[ChatMessage]) -> 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". 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:

1. Building a ChatMessage list with extraction instructions
2. Calling llm_client.complete() to get a Completion
3. Parsing the completion text as JSON
4. Validating against the response_model
5. 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.

---

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 LEX_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 LEX_AI_LLM__)

Global:

    LEX_AI_LLM__STRATEGY              sequential | parallel_race |
                                           cost_optimized | latency_optimized
    LEX_AI_LLM__DEFAULTS__TEMPERATURE float  (default 0.2)
    LEX_AI_LLM__DEFAULTS__MAX_TOKENS  int    (default: provider default)
    LEX_AI_LLM__QUOTA__BACKEND        memory | database  (default memory)
    LEX_AI_LLM__LOG__BACKEND          memory | database  (default memory)
    LEX_AI_LLM__LOG__MAX_ENTRIES      int    (default 1000)

Per-provider (pattern: LEX_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):

    LEX_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_RESOURCE
    LEX_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_DEPLOYMENT
    LEX_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_API_VERSION

Cloudflare-specific extras (activated by CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID):

    LEX_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID  <- activates
    LEX_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_API_TOKEN
    LEX_AI_LLM__PROVIDERS__CLOUDFLARE__MODEL

AWS Bedrock extras (activated by BEDROCK__EXTRAS__AWS_REGION):

    LEX_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_REGION        <- activates
    LEX_AI_LLM__PROVIDERS__BEDROCK__MODEL
    LEX_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_ACCESS_KEY_ID
    LEX_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_SECRET_ACCESS_KEY

Google Vertex AI extras (activated by VERTEX__EXTRAS__VERTEX_PROJECT):

    LEX_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_PROJECT      <- activates
    LEX_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_LOCATION
    LEX_AI_LLM__PROVIDERS__VERTEX__MODEL
    LEX_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_CREDENTIALS_FILE

Global:

    LEX_AI_LLM__STRATEGY              sequential | parallel_race |
                                           cost_optimized | latency_optimized
    LEX_AI_LLM__DEFAULTS__TEMPERATURE float  (default 0.2)
    LEX_AI_LLM__DEFAULTS__MAX_TOKENS  int    (default: provider default)
    LEX_AI_LLM__QUOTA__BACKEND        memory | database  (default memory)
    LEX_AI_LLM__LOG__BACKEND          memory | database  (default memory)
    LEX_AI_LLM__LOG__MAX_ENTRIES      int    (default 1000)

Per-provider (pattern: LEX_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):

    LEX_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_RESOURCE
    LEX_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_DEPLOYMENT
    LEX_AI_LLM__PROVIDERS__AZURE__EXTRAS__AZURE_API_VERSION

Cloudflare-specific extras (activated by CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID):

    LEX_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_ACCOUNT_ID  <- activates
    LEX_AI_LLM__PROVIDERS__CLOUDFLARE__EXTRAS__CF_API_TOKEN
    LEX_AI_LLM__PROVIDERS__CLOUDFLARE__MODEL

AWS Bedrock extras (activated by BEDROCK__EXTRAS__AWS_REGION):

    LEX_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_REGION        <- activates
    LEX_AI_LLM__PROVIDERS__BEDROCK__MODEL
    LEX_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_ACCESS_KEY_ID
    LEX_AI_LLM__PROVIDERS__BEDROCK__EXTRAS__AWS_SECRET_ACCESS_KEY

Google Vertex AI extras (activated by VERTEX__EXTRAS__VERTEX_PROJECT):

    LEX_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_PROJECT      <- activates
    LEX_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_LOCATION
    LEX_AI_LLM__PROVIDERS__VERTEX__MODEL
    LEX_AI_LLM__PROVIDERS__VERTEX__EXTRAS__VERTEX_CREDENTIALS_FILE

from_env

def from_env(cls) -> LLMConfig

source

Build a routing config from LEX_AI_LLM__ environment variables.

Returns

Type	Description
LLMConfig	Populated LLMConfig.

---

LLMModule

source

LLM client and model-management integration.

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

Usage

from lexigram.ai.llm.config import ClientConfig

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

from lexigram.ai.llm.config import ClientConfig

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

Multi-provider routing

from lexigram.ai.llm import LLMModule

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

from lexigram.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
) -> 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.

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 Lexigram DI container.

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

Example

from lexigram.ai.llm.di.provider import LLMProvider
from lexigram.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,
    name: str = 'llm',
    cache_backend: CacheBackendProtocol | None = None
) -> 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.
`name`	str	Provider name used for identification.
`cache_backend`	CacheBackendProtocol | None	Injected cache backend for optional response caching.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register LLM services with the DI container.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The Lexigram 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.

Example

from lexigram.ai.llm.module import LLMModule
from lexigram.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.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Build and register the LLMRouter with the DI container.

Parameters

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

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Boot phase — no-op for this provider.

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)

---

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
) -> 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])}")

close

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: list[ChatMessage]) -> int

source

Count tokens in a list of chat messages.

---

ModelCapabilities

source

Model capabilities and constraints.

---

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)

128000

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 lexigram.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.

close

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 lexigram.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

close

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

close

async def close() -> None

source

---

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.

register

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

source

Register a parser with a name.

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.

Returns

Type	Description
ParserRegistry	A new ParserRegistry with default parsers.

---

PricingManager

source

Manages pricing data from multiple sources with caching.

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

1. JSON file (fastest, most reliable)
2. API endpoints (for updates)
3. 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

source

Get pricing for a specific model.

Queries sources in order:

1. Cache (if not force_refresh)
2. Each source in priority order
3. Fuzzy match if enabled
4. Default fallback

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	ModelPricing for the model.

Raises

Exception	Description
ValueError	If model not found in any source.

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.

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.

---

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_...",
)

---

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.

Singleton registry that maintains information about all available LLM providers, both built-in and custom.

__init__

def __init__() -> None

source

Initialize provider registry.

register

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

Register a new LLM provider.

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

Register a provider following the ProviderRegistryProtocol.

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 responses into Pydantic models.

Uses the existing structured parser’s validation logic to parse and validate against a Pydantic model.

Example

from pydantic import BaseModel

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

parser = PydanticOutputParser(User)
result = parser.parse('{"name": "John", "age": 30}')
assert result.name == "John"

__init__

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

source

Initialize with a Pydantic model class.

Parameters

Parameter	Type	Description
`model`	type[BaseModel]	Pydantic BaseModel subclass to parse into.

parse

def parse(text: str) -> BaseModel

source

Parse text into a Pydantic model instance.

Parameters

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

Returns

Type	Description
BaseModel	Validated Pydantic model instance.

Raises

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

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 that matches the Pydantic model 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)

42

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

---

StaticPricingSource

source

Pricing source from static dictionary.

Hardcoded pricing data as a fallback when other sources are unavailable. Useful for custom internal models or as ultimate fallback.

Attributes: pricing_map: Dictionary of model name to pricing.

Example

source = StaticPricingSource({
"my-model": ModelPricing(
model="my-model",
prompt_per_1m=5.0,
completion_per_1m=10.0,
provider="custom"
)
})

__init__

def __init__(pricing_map: dict[str, ModelPricing])

source

Initialize static pricing source.

Parameters

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

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 static pricing data.

source_name

property source_name() -> str

source

Get source name.

---

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: list[ChatMessage]) -> int

source

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

---

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:

• char_estimate (always available, fallback)
• tiktoken (if installed, for OpenAI/Anthropic models)
• huggingface (if installed, for HuggingFace models)
• mistral (if installed, for Mistral models)

Returns

Type	Description
TokenCounterRegistry	TokenCounterRegistry pre-populated with default backends.

register

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

source

Register a counter backend under a named key.

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.

---

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 lexigram.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 lexigram.ai.llm import create_token_counter

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

---

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 lexigram-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 lexigram-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.

---

ModelNotFoundError

source

Model unavailable or not found — recoverable via fallback routing.

---

ParseError

source

Raised when response cannot be parsed.

---

ProviderConnectionError

source

Error raised when connection to an LLM provider fails.

---

SchemaValidationError

source

Raised when parsed response fails validation.

---

StreamError

source

Error raised during LLM response streaming.

---

StructuredOutputError

source

Base exception for structured output errors.

---

TokenLimitError

source

Error raised when the token limit for a request is exceeded.

---