cxai-client 3.0.0

CXAI — Full-featured Python client for the Ollama REST API

CXAI Ollama API Client v3

CXAI — Production-grade Python client for the Ollama REST API.

What's New in v3

Feature	Description
Response Caching	In-memory + disk caching with TTL, LRU eviction, and cache middleware
Circuit Breaker	CLOSED→OPEN→HALF_OPEN state machine with per-endpoint breakers
Rate Limiter	Token-bucket rate limiter with per-model limits and backpressure
Streaming Utilities	StreamCollector, CallbackStream, TokenCountingStream, stream_to_string
Prompt Templates	Variable interpolation, few-shot examples, YAML/JSON loading, TemplateRegistry
Structured Output	Pydantic model validation, JSON Schema generation, auto-retry on bad output
Agent Loop	CXAIAgent with tool-calling lifecycle, parallel tool exec, max_steps, step_callback
Model Router	Rule-based routing, FallbackChain, RoutedClient, LoadBalancedClient
Conversation Persistence	FileConversationStore, SQLiteConversationStore, branch(), auto-save/load
OpenTelemetry	Span-per-request middleware with attribute injection, lazy import

Installation

pip install cxai-client

# Optional dependencies
pip install cxai-client[structured]  # Pydantic for structured output
pip install cxai-client[templates]    # PyYAML for YAML template loading
pip install cxai-client[otel]         # OpenTelemetry integration
pip install cxai-client[all]          # All optional dependencies

Configuration

export CXAI_API_KEY=your-api-key-here
export CXAI_BASE_URL=https://api.ollama.com  # optional

Or pass a CXAIConfig object:

from cxai_client import CXAIConfig

config = CXAIConfig(
    api_key="your-key",
    timeout=60.0,
    max_retries=3,
    enable_observability=True,
    enable_cache=True,
    cache_ttl=300.0,
    cache_max_size=512,
    enable_circuit_breaker=True,
    circuit_breaker_threshold=5,
    enable_rate_limit=True,
    rate_limit_rpm=60,
)

Quick Start

Generate

from cxai_client import CXAIOllamaClient, GenerateRequest

with CXAIOllamaClient() as client:
    resp = client.generate(GenerateRequest(model="llama3.2", prompt="Why is the sky blue?"))
    print(resp.response)

Chat

from cxai_client import CXAIOllamaClient, ChatRequest, ChatMessage

with CXAIOllamaClient() as client:
    resp = client.chat(ChatRequest(
        model="llama3.2",
        messages=[ChatMessage(role="user", content="What is 2+2?")],
    ))
    print(resp.message.content)

Caching

from cxai_client import CXAIConfig, CXAIOllamaClient, InMemoryCache
from cxai_client.cache import CacheMiddleware

cache = InMemoryCache(max_size=256, default_ttl=300.0)
config = CXAIConfig(enable_cache=True)
with CXAIOllamaClient(config=config) as client:
    # First call hits the server, second call hits the cache
    client.generate(GenerateRequest(model="llama3.2", prompt="Hello"))
    client.generate(GenerateRequest(model="llama3.2", prompt="Hello"))  # cached

# Or use DiskCache for persistent caching:
from cxai_client.cache import DiskCache
disk_cache = DiskCache("/tmp/cxai_cache", default_ttl=600.0)

Circuit Breaker

from cxai_client.circuit_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitBreakerRegistry
from cxai_client import CXAICircuitOpenError

config = CXAIConfig(enable_circuit_breaker=True, circuit_breaker_threshold=3)
with CXAIOllamaClient(config=config) as client:
    try:
        client.generate(GenerateRequest(model="llama3.2", prompt="Hi"))
    except CXAICircuitOpenError as e:
        print(f"Circuit open: {e.state}, failures={e.failure_count}")

Rate Limiting

from cxai_client import CXAIConfig

config = CXAIConfig(enable_rate_limit=True, rate_limit_rpm=30)
with CXAIOllamaClient(config=config) as client:
    for i in range(50):
        client.generate(GenerateRequest(model="llama3.2", prompt=f"Count {i}"))

Agent (Tool Calling)

from cxai_client import CXAIAgent, CXAIOllamaClient, ChatMessage

def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

with CXAIOllamaClient() as client:
    agent = CXAIAgent(
        client=client,
        model="llama3.2",
        tools={"add": add, "multiply": multiply},
        max_steps=5,
    )
    result = agent.run([ChatMessage(role="user", content="What is 3 + 4?")])
    print(result.message.content)

Structured Output

from pydantic import BaseModel
from cxai_client import CXAIOllamaClient
from cxai_client.structured import extract_structured

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

with CXAIOllamaClient() as client:
    person = extract_structured(client, "llama3.2", "Tell me about a person named Alice age 30", Person)
    print(f"{person.name} is {person.age} years old")

Streaming Utilities

from cxai_client.streaming import StreamCollector, CallbackStream, stream_to_string

with CXAIOllamaClient() as client:
    # Collect stream into a single response
    stream = client.generate_stream(GenerateRequest(model="llama3.2", prompt="Tell me a story"))
    result = StreamCollector.collect(stream)
    print(result.response)

    # Stream with callbacks
    def on_token(chunk):
        print(chunk.response, end="", flush=True)

    stream = client.generate_stream(GenerateRequest(model="llama3.2", prompt="Hi"))
    for chunk in CallbackStream(stream, on_token=on_token):
        pass  # on_token handles it

Prompt Templates

from cxai_client.templates import PromptTemplate, TemplateRegistry

registry = TemplateRegistry()
registry.register(PromptTemplate(
    name="summarize",
    template="Summarize the following text in {style} style:\n\n{text}",
    variables=["text"],
    defaults={"style": "concise"},
    system_prompt="You are a summarizer.",
))
result = registry.render("summarize", text="Long text here...", style="academic")

Model Router

from cxai_client.router import ModelRouter, ModelNameRule, FallbackChain, FallbackEntry, RoutedClient

# Route specific models to alternatives
router = ModelRouter([
    ModelNameRule("expensive-model", "cheap-model"),
])

client = CXAIOllamaClient()
routed = RoutedClient(client, router)
result = routed.generate(GenerateRequest(model="expensive-model", prompt="Hi"))
# Actually uses "cheap-model"

# Fallback chain
chain = FallbackChain([
    FallbackEntry(model="primary-model"),
    FallbackEntry(model="fallback-model"),
])
result = chain.execute_generate(client, GenerateRequest(model="primary-model", prompt="Hi"))

Conversation Persistence

from cxai_client import CXAIOllamaClient, Conversation, ConversationConfig
from cxai_client.persistence import FileConversationStore, SQLiteConversationStore

store = FileConversationStore("/tmp/conversations")
# Or: store = SQLiteConversationStore("/tmp/conversations.db")

with CXAIOllamaClient() as client:
    conv = Conversation(
        client=client,
        config=ConversationConfig(model="llama3.2"),
        conversation_id="my-conversation-1",
        store=store,
    )
    resp = conv.send("Hello!")
    # Auto-saved to store after each turn

    # Load later
    conv2 = Conversation(client, ConversationConfig(model="llama3.2"),
                          conversation_id="my-conversation-1", store=store)
    print(conv2.history)  # Restored from store

    # Branch from a point
    branched = conv.branch(turn_index=1)

Observability

config = CXAIConfig(enable_observability=True)
with CXAIOllamaClient(config=config) as client:
    client.generate(GenerateRequest(model="llama3.2", prompt="Hi"))
    agg = client.metrics.aggregate()
    print(f"Requests: {agg.total_requests}, P50: {agg.p50_latency_ms:.1f}ms")

OpenTelemetry

from cxai_client.otel import CXAIOTelMiddleware, CXAIAsyncOTelMiddleware

# Add to middleware pipeline manually or use setup_otel()
# Requires: pip install cxai-client[otel]

Error Handling

from cxai_client.exceptions import (
    CXAIError, CXAIConnectionError, CXAITimeoutError, CXAIStreamError,
    CXAIRetryExhaustedError, AuthenticationError, ModelNotFoundError,
    InvalidRequestError, RateLimitError, ServerError,
    CXAICircuitOpenError, CXAIRateLimitExceededError,
    CXAIStructuredOutputError, CXAIAgentMaxStepsError,
)

try:
    resp = client.generate(GenerateRequest(model="nonexistent", prompt="hi"))
except ModelNotFoundError:
    print("Model not found!")
except CXAICircuitOpenError as e:
    print(f"Circuit open: {e.state}, failures={e.failure_count}")
except CXAIRateLimitExceededError as e:
    print(f"Rate limited: queue={e.queue_depth}/{e.max_queue_depth}")
except CXAIAgentMaxStepsError as e:
    print(f"Agent exceeded {e.max_steps} steps after {e.steps_taken}")

CLI

cxai generate llama3.2 "Why is the sky blue?"
cxai chat llama3.2 --system "You are a CXAI assistant."
cxai models [--running] [--json]
cxai pull llama3.2
cxai show llama3.2 --modelfile --system
cxai ping
cxai version

Running Tests

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

License

MIT — Built by CXAI