LLMBridgeKit

Version: 1.0.0
Author: sreeyenan
License: MIT
Python: 3.10+

Universal Python LLM Provider Abstraction Library

A provider-agnostic Python library that lets any AI feature call multiple LLM providers through one consistent interface.

Features

🌐 Multi-Provider Support

• Cloud Providers: OpenAI, Azure OpenAI, Google Gemini, Anthropic Claude, Mistral AI, Cohere
• Fast Inference: Groq, Together AI
• Model Routers: OpenRouter, Hugging Face Inference Providers
• Local Models: Ollama, LM Studio
• Custom: Your own HTTP endpoint

🎯 Core Capabilities

• Unified API: Single interface for all providers
• Text Generation: Simple .generate() method
• Chat: Message-based .chat() method
• Structured JSON: .generate_json() with schema validation
• Tool Calling: .generate_with_tools() for function calling
• Streaming: Real-time token streaming
• Async Support: Full async/await support

🛡️ Reliability Features

• Fallback Chains: Automatic provider failover
• Retry Logic: Exponential backoff with jitter
• Timeout Control: Per-request and per-provider timeouts
• Error Normalization: Consistent error handling across providers
• Rate Limiting: Respect provider rate limits

📊 Observability

• Usage Tracking: Token counts and model usage
• Cost Estimation: Calculate API costs
• Latency Metrics: Track request duration
• Audit Logging: Request/response logging with redaction
• Request IDs: Distributed tracing support

🔒 Security

• PII Redaction: Mask sensitive data before sending
• Secret Management: Environment variable support
• API Key Rotation: Hot-reload credentials

Installation

# Core installation
pip install llmbridgekit

# With specific providers
pip install llmbridgekit[openai]
pip install llmbridgekit[gemini]
pip install llmbridgekit[anthropic]

# With multiple providers
pip install llmbridgekit[openai,gemini,groq]

# Local model support
pip install llmbridgekit[local]

# Everything
pip install llmbridgekit[all]

Quick Start

Basic Text Generation

from llmbridgekit import LLMClient

client = LLMClient(provider="gemini")

response = client.generate(
    prompt="Explain what a materialized view is in ClickHouse."
)

print(response.text)
print(f"Tokens: {response.usage['total_tokens']}")

Chat with Messages

response = client.chat(
    messages=[
        {"role": "system", "content": "You are a data analytics assistant."},
        {"role": "user", "content": "Show revenue by region."}
    ],
    provider="openai",
    model="gpt-4o"
)

print(response.text)

Structured JSON Output

schema = {
    "type": "object",
    "properties": {
        "intent": {
            "type": "string",
            "enum": ["query", "chart", "dashboard", "report"]
        },
        "confidence": {"type": "number", "minimum": 0, "maximum": 1},
        "entities": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["intent", "confidence"]
}

response = client.generate_json(
    prompt="User asks: 'Show me sales by region for last quarter'",
    schema=schema
)

print(response.json_data)
# {'intent': 'query', 'confidence': 0.95, 'entities': ['sales', 'region', 'last quarter']}

Streaming

for chunk in client.stream(
    prompt="Write a detailed analysis of Q4 performance.",
    provider="anthropic"
):
    print(chunk.text, end="", flush=True)

Fallback Chain

response = client.generate(
    prompt="Generate a SQL query for revenue by category.",
    fallback_chain=[
        {"provider": "gemini", "model": "gemini-2.0-flash-exp"},
        {"provider": "groq", "model": "llama-3.3-70b-versatile"},
        {"provider": "ollama", "model": "qwen2.5:3b"}
    ]
)

Tool Calling

tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string"}
            },
            "required": ["location"]
        }
    }
]

response = client.generate_with_tools(
    prompt="What's the weather in San Francisco?",
    tools=tools,
    provider="openai"
)

if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"Tool: {tool_call['name']}")
        print(f"Args: {tool_call['arguments']}")

Configuration

YAML Configuration

# config.yaml
default_provider: gemini
default_timeout: 60
default_retries: 2

providers:
  openai:
    api_key_env: OPENAI_API_KEY
    model: gpt-4o
    temperature: 0
    max_tokens: 1000

  gemini:
    api_key_env: GEMINI_API_KEY
    model: gemini-2.0-flash-exp
    temperature: 0

  ollama:
    base_url: http://localhost:11434
    model: qwen2.5:3b
    temperature: 0

fallback_chains:
  default:
    - provider: gemini
      model: gemini-2.0-flash-exp
    - provider: groq
      model: llama-3.3-70b-versatile

tasks:
  classification:
    provider: groq
    model: llama-3.3-70b-versatile
    temperature: 0
    max_tokens: 500

  text_to_sql:
    provider: gemini
    model: gemini-2.0-flash-exp
    temperature: 0
    fallback_chain: default

Load configuration:

from llmbridgekit import LLMClient

client = LLMClient.from_yaml("config.yaml")

# Use task-based routing
response = client.run_task(
    task="text_to_sql",
    prompt="Show me revenue by product category"
)

Supported Providers

Use Cases

NLQ (Natural Language Query)

# Context resolution
response = client.generate_json(
    prompt=context_resolver_prompt,
    schema=context_resolution_schema,
    provider="ollama",  # Local fallback
    model="qwen2.5:3b"
)

# Text-to-SQL generation
response = client.generate(
    prompt=sql_generation_prompt,
    provider="gemini",
    fallback_chain="default"
)

RAG (Retrieval Augmented Generation)

# Generate embeddings
embeddings = client.embed(
    texts=["chunk1", "chunk2", "chunk3"],
    provider="openai"
)

# Generate answer with context
response = client.chat(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": f"Context: {context}\n\nQuestion: {question}"}
    ]
)

Chart/Dashboard Generation

response = client.generate_json(
    prompt=f"Generate chart config for: {user_request}",
    schema=chart_config_schema,
    provider="openai",
    model="gpt-4o"
)

chart_config = response.json_data

Document Q&A

response = client.chat(
    messages=[
        {"role": "system", "content": "Answer based on the document."},
        {"role": "user", "content": f"Document: {document}\n\nQuestion: {question}"}
    ],
    provider="anthropic",
    model="claude-opus-4"
)

Advanced Features

Async Support

import asyncio
from llmbridgekit import LLMClient

async def main():
    client = LLMClient(provider="gemini")
    
    response = await client.agenerate(
        prompt="Explain async/await in Python"
    )
    
    print(response.text)

asyncio.run(main())

Cost Tracking

response = client.generate(
    prompt="Analyze sales data",
    provider="openai",
    model="gpt-4o"
)

print(f"Cost: ${response.cost['total_cost']:.4f}")
print(f"Input tokens: {response.usage['prompt_tokens']}")
print(f"Output tokens: {response.usage['completion_tokens']}")

Redaction

from llmbridgekit import LLMClient, RedactionConfig

client = LLMClient(
    provider="openai",
    redaction=RedactionConfig(
        mask_email=True,
        mask_phone=True,
        mask_api_keys=True,
        custom_patterns=[
            r'\b\d{3}-\d{2}-\d{4}\b'  # SSN
        ]
    )
)

response = client.generate(
    prompt="User email: john@example.com, phone: 555-1234"
)
# Prompt sent: "User email: [EMAIL_REDACTED], phone: [PHONE_REDACTED]"

Custom Provider

from llmbridgekit.providers import BaseProvider, ProviderFeatures
from llmbridgekit import provider_registry

class MyCustomProvider(BaseProvider):
    name = "my_custom"
    features = ProviderFeatures(
        chat=True,
        structured_outputs=True,
        streaming=True
    )
    
    def generate(self, prompt, config):
        # Your implementation
        response = self._call_api(prompt, config)
        return self._parse_response(response)

# Register
provider_registry.register("my_custom", MyCustomProvider)

# Use
client = LLMClient(provider="my_custom")
response = client.generate("Hello!")

Testing

# Install dev dependencies
pip install llm-gateway[dev]

# Run tests
pytest

# Run with coverage
pytest --cov=llmbridgekit --cov-report=html

# Run specific provider tests
pytest tests/providers/test_openai_provider.py -v

Documentation

• User Manual - Complete usage guide
• Changelog - Version history
• Environment Variables - Configuration reference
• Examples - Code examples for common use cases

Architecture

Application Layer
    ↓
LLMClient (llmbridgekit/client.py)
    ↓
Task Router / Fallback Chain
    ↓
Provider Registry
    ↓
BaseProvider
    ↓
OpenAI | Gemini | Anthropic | Groq | ... | Custom HTTP

License

MIT License - see LICENSE file for details.

Support

• Issues: https://github.com/sreeyenan/llmbridgekit/issues
• Discussions: https://github.com/analytic-ai/llm-gateway/discussions
• Documentation: https://llm-gateway.readthedocs.io