qx-llms

A unified LLM client factory and model registry for use across QX applications. Provides a consistent interface for working with multiple LLM providers through both LangChain and OpenAI Agents SDK.

Installation

pip install qx-llms

Or install from source:

pip install -e .

Features

• Multi-provider support - OpenAI, Anthropic, Google Gemini, DeepSeek, and more
• Two factory interfaces - LangChain clients and OpenAI Agents SDK models
• Model registry - Centralized model definitions with per-token pricing, context window, and capabilities
• Usage tracking middleware - Automatic token + credit accounting via a context-local usage_scope; works with both LangChain and raw OpenAI clients with zero refactor of existing call sites
• Fake models for testing - Mock implementations that don't make API calls
• Graceful fallbacks - Optional silent failure with dummy models

Quick Start

LangChain Factory

from qx_llms.factories.langchain_client_factory import get_llm_client, get_embeddings_client

# Get a chat model
llm = get_llm_client(model_name="gpt-4o", provider="openai")
response = llm.invoke("Hello, world!")

# Get an embeddings model
embeddings = get_embeddings_client(model_name="text-embedding-3-small", provider="openai")
vectors = embeddings.embed_query("Hello, world!")

OpenAI Agents Factory

from qx_llms.factories.openai_client_factory import get_openai_agents_model

# Get a model for use with OpenAI Agents SDK
model = get_openai_agents_model(provider="openai", model_name="gpt-4o")

Model Registry

from qx_llms.model_registry import (
    get_llm_by_name,
    get_llm_options,
    get_llm_credit_mapping,
    get_embedding_by_name,
    get_embedding_options,
    ModelProvider,
)

# Get a specific chat model's details
model = get_llm_by_name("gpt-4o")
print(model.name, model.provider, model.credits)

# Get all chat models with specific capabilities
options = get_llm_options(
    providers=[ModelProvider.OPENAI, ModelProvider.ANTHROPIC],
    structured_output=True,
    tool_use=True,
)

# Get credit costs for billing
credits = get_llm_credit_mapping(credit_multiplier=2)

# Get a specific embedding model's details
embedding = get_embedding_by_name("text-embedding-3-large")
print(embedding.name, embedding.dimensions, embedding.multimodal)

# Get all embedding models with specific capabilities
embedding_options = get_embedding_options(
    providers=[ModelProvider.OPENAI],
    dimensions=1536,
)

Supported Providers

LangChain Factory

OpenAI Agents Factory

Testing with Fake Models

Both factories provide fake models that return static responses without making API calls:

# LangChain fake model
from qx_llms.factories.langchain_client_factory import get_llm_client

fake_llm = get_llm_client(model_name="fake", provider="fake")

# OpenAI Agents fake model
from qx_llms.factories.openai_client_factory import get_openai_agents_model

fake_model = get_openai_agents_model(provider="fake")
response = await fake_model.get_response(...)  # Returns "fake response"

Model Registry

The model registry provides a centralized definition of available models with their capabilities.

Chat Models

from qx_llms.model_registry import ChatModel, ModelProvider

# Each chat model has these attributes:
# - name: str              - Model identifier (e.g., "gpt-4o")
# - provider: ModelProvider - Provider enum
# - structured_output: bool - Supports JSON schema output
# - tool_use: bool         - Supports function calling
# - vision: bool           - Supports image input
# - accepts_temperature: bool - Supports temperature parameter
# - credits: int           - Base credit cost per request

Embedding Models

from qx_llms.model_registry import EmbeddingModel, ModelProvider

# Each embedding model has these attributes:
# - name: str              - Model identifier (e.g., "text-embedding-3-large")
# - provider: ModelProvider - Provider enum
# - credits: int           - Base credit cost per request
# - dimensions: int        - Output vector size (768, 1536, 3072, etc.)
# - multimodal: bool       - Can embed images (e.g., CLIP, OpenAI multimodal)

Filtering Chat Models

from qx_llms.model_registry import get_llm_options, ModelProvider

# Get only models that support structured output and vision
models = get_llm_options(
    structured_output=True,
    vision=True,
)

# Get models from specific providers
openai_models = get_llm_options(providers=[ModelProvider.OPENAI])

Filtering Embedding Models

from qx_llms.model_registry import get_embedding_options, ModelProvider

# Get only embedding models with specific dimensions
models = get_embedding_options(dimensions=1536)

# Get multimodal embedding models
multimodal_models = get_embedding_options(multimodal=True)

# Get embedding models from specific providers
openai_embeddings = get_embedding_options(providers=[ModelProvider.OPENAI])

Usage Tracking

Every client returned by get_openai_client and get_llm_client is monkey-patched on construction so that completed LLM calls feed token + credit data into whatever usage_scope is currently active. Outside any scope the middleware is a strict no-op — calls run untouched, but no usage is recorded.

Pricing model

Each ChatModel in the registry carries:

• context_window: int — total tokens the model can hold (prompt + completion).
• input_credits_per_mtok: float, output_credits_per_mtok: float, cached_input_credits_per_mtok: Optional[float] — pricing in credits per million tokens. The reference scale is 1000 credits = $1.00 USD (so 1 credit ≈ $0.001). Values are hand-curated in _CHAT_MODELS_LIST; see the file header for the verification date.

The helper calculate_credits(model, *, input_tokens, output_tokens, cached_input_tokens=0, multiplier=1.0) -> int is what middleware calls per event. It rounds up via math.ceil, so a sub-credit call still books at least one credit. The multiplier lets orchestration layers scale costs (e.g. an "expensive analysis" path can charge 2x).

usage_scope

import openai
from qx_llms import usage_scope
from qx_llms.factories.openai_client_factory import get_openai_client

client = get_openai_client(provider="openai")  # middleware installed here

async with usage_scope(
    source="agent_chat",          # free-form tag your host maps to a LedgerSource
    reference_id=chat_id,         # threaded through to your host's ledger entry
    credit_multiplier=1.0,
    metadata={"user_id": user_id},
) as scope:
    await client.chat.completions.create(
        model="gpt-5.6-luna",
        messages=[...],
    )
    # ...more LLM calls inside the same scope...

print(scope.credits)         # rolled-up integer credits
print(scope.input_tokens)    # totals across every UsageEvent in the scope
print(scope.events[-1])      # last UsageEvent (model, tokens, credits)

Scopes nest via a ContextVar. Only the innermost active accumulator captures each event, so wrapping a sub-task in its own scope won't double-charge an outer one. asyncio.gather'd tasks see whichever scope was active when they were scheduled — push a fresh scope inside each task to keep their usage isolated.

Context-window pressure

from qx_llms import estimate_context_pressure
from qx_llms.model_registry import get_llm_by_name

model = get_llm_by_name("gpt-5.6-luna")
ratio = estimate_context_pressure(model, scope.events[-1].input_tokens)
if ratio >= 0.7:
    # signal the user that the next turn may hit a hard context-window error
    ...

LangChain

from qx_llms import usage_scope
from qx_llms.factories.langchain_client_factory import get_llm_client

llm = get_llm_client(model_name="claude-sonnet-4", provider="anthropic")

async with usage_scope(source="agent_chat", reference_id=chat_id) as scope:
    response = await llm.ainvoke("Hello, world!")
# scope.credits is populated via the LangChain QxUsageCallback registered
# on the client by the factory; works for streaming and non-streaming calls.

Configuration

import qx_llms

# Surface middleware bugs as exceptions instead of swallowing (default: True).
qx_llms.configure(fail_open=False)

# Emit a structured INFO log per recorded UsageEvent (default: False).
# Useful in dev/staging to verify token-based charging without scraping a ledger.
qx_llms.configure(usage_log_enabled=True)
# Log line: `qx_llms.usage` with extras {"qx_llms_usage": {model, source,
# reference_id, input_tokens, output_tokens, cached_input_tokens,
# reasoning_tokens, credits, credit_multiplier}}.

Error Handling

Both factories support graceful fallbacks:

# Raises ValueError if provider is invalid or API key is missing
llm = get_llm_client(model_name="gpt-4o", provider="openai", fail_silently=False)

# Returns a dummy model instead of raising
llm = get_llm_client(model_name="gpt-4o", provider="openai", fail_silently=True)

Development

Running Tests

pip install -e ".[test]"
pytest tests/ -v

Requirements

• Python >= 3.12
• See requirements.txt for dependencies