A comprehensive Python wrapper for Large Language Models with database integration and usage tracking
A comprehensive, modular Python library for seamless integration with multiple Large Language Model providers. Built with enterprise-grade features including token tracking, usage analytics, and robust error handling.
pip install hibiz-any-llm
llm_wrapper/
├── core/ # Core functionality and factory
├── providers/ # LLM provider implementations
├── models/ # Data models and schemas
├── database/ # Database management
└──utils/ # Token Calculation and validators
from hibiz_any_llm import LLMWrapper, LLMProvider
# Configure your provider
azure_config = {
'service_url': 'https://your-resource.openai.azure.com',
'api_key': 'your-api-key',
'deployment_name': 'gpt-4',
'api_version': '2023-12-01-preview'
}
# Database configuration
db_config = {
'host': 'localhost',
'port': 5432,
'dbname': 'llm_usage',
'user': 'username',
'password': 'password'
}
# Initialize wrapper
wrapper = LLMWrapper(
provider_type=LLMProvider.AZURE_OPENAI,
provider_config=azure_config,
db_config=db_config,
enable_logging=True
)
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is machine learning?"}
]
response = wrapper.send_request(
prompt_payload=messages,
customer_id="customer_123",
organization_id="org_456",
app_name="chatbot",
module_name="education",
function_name="explain_concepts",
model="gpt-4",
temperature=0.7,
max_tokens=500
)
print(f"Response: {response['output_text']}")
print(f"Tokens used: {response['total_tokens']}")
embedding_response = wrapper.create_embeddings(
input_texts=["Machine learning is amazing", "AI will change the world"],
customer_id="customer_123",
organization_id="org_456",
app_name="search_engine",
module_name="vectorization",
function_name="create_embeddings",
model="text-embedding-3-small"
)
print(f"Embeddings: {embedding_response['embeddings']}")
azure_config = {
'service_url': 'https://your-resource.openai.azure.com',
'api_key': 'your-api-key',
'deployment_name': 'gpt-4',
'api_version': '2023-12-01-preview',
'timeout': 600
}
wrapper = LLMWrapper(LLMProvider.AZURE_OPENAI, azure_config, db_config)
openai_config = {
'api_key': 'sk-your-openai-api-key',
'organization_id': 'org-your-org-id', # Optional
'timeout': 600
}
wrapper = LLMWrapper(LLMProvider.OPENAI, openai_config, db_config)
anthropic_config = {
'api_key': 'sk-ant-your-anthropic-api-key',
'default_model': 'claude-opus-4-20250514',
'timeout': 300
}
wrapper = LLMWrapper(LLMProvider.ANTHROPIC, anthropic_config, db_config)
google_config = {
'api_key': 'your-gemini-api-key',
'default_model': 'gemini-2.0-flash',
'timeout': 300
}
wrapper = LLMWrapper(LLMProvider.GOOGLE, google_config, db_config)
grok_config = {
'api_key': 'your-grok-api-key',
'default_model': 'grok-4',
'timeout': 300
}
wrapper = LLMWrapper(LLMProvider.GROK, grok_config, db_config)
qwen_config = {
'api_key': 'your-qwen-api-key',
'default_model': 'qwen-plus',
'timeout': 300
}
wrapper = LLMWrapper(LLMProvider.QWEN, qwen_config, db_config)
deepseek_config = {
'api_key': 'your-deepseek-api-key',
'default_model': 'deepseek-chat',
'timeout': 300
}
wrapper = LLMWrapper(LLMProvider.DEEP_SEEK, deepseek_config, db_config)
stats = wrapper.get_usage_stats(
customer_id="customer_123",
start_date="2024-01-01T00:00:00",
end_date="2024-12-31T23:59:59",
app_name="chatbot"
)
print(f"Total requests: {stats['summary']['total_requests']}")
print(f"Total tokens: {stats['summary']['total_tokens']}")
print(f"Success rate: {stats['summary']['success_rate']}")
# By application
app_stats = wrapper.get_usage_stats(app_name="chatbot")
# By model
model_stats = wrapper.get_usage_stats(filters={"model_name": "gpt-4"})
# By request type
embedding_stats = wrapper.get_usage_stats(request_type="embedding")
response = wrapper.send_request(
prompt_payload=[
{"role": "user", "content": "List 3 benefits of exercise in JSON format"}
],
customer_id="customer_789",
organization_id="org_456",
app_name="health_app",
module_name="exercise",
function_name="get_benefits",
response_type="json" # Automatically ensures JSON output
)
# Access parsed JSON
json_data = response['processed_output']
with LLMWrapper(LLMProvider.AZURE_OPENAI, azure_config, db_config) as wrapper:
response = wrapper.send_request(
prompt_payload=messages,
customer_id="customer_123",
organization_id="org_456",
app_name="temp_app",
module_name="test",
function_name="context_test"
)
# Automatic cleanup on exit
providers = {
LLMProvider.AZURE_OPENAI: azure_config,
LLMProvider.OPENAI: openai_config,
LLMProvider.ANTHROPIC: anthropic_config
}
for provider_type, config in providers.items():
wrapper = LLMWrapper(provider_type, config, db_config)
response = wrapper.send_request(
prompt_payload=[{"role": "user", "content": "Hello!"}],
customer_id="multi_test",
organization_id="org_test",
app_name="provider_comparison",
module_name="testing",
function_name="hello_test"
)
print(f"{provider_type.value}: {response['output_text']}")
wrapper.close()
The library automatically creates the following PostgreSQL table:
CREATE TABLE token_usage_log (
id SERIAL PRIMARY KEY,
customer_id VARCHAR(255) NOT NULL,
organization_id VARCHAR(255) NOT NULL,
provider VARCHAR(100) NOT NULL,
model_name VARCHAR(255) NOT NULL,
app_name VARCHAR(255),
module_name VARCHAR(255),
function_name VARCHAR(255),
request_type VARCHAR(50) NOT NULL,
request_params JSONB,
response_params JSONB,
input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0,
total_tokens INTEGER DEFAULT 0,
request_timestamp TIMESTAMP DEFAULT NOW(),
response_time_ms INTEGER DEFAULT 0,
status VARCHAR(50) DEFAULT 'success',
request_id VARCHAR(255),
cost FLOAT
);
idx_customer_date on (customer_id, request_timestamp)idx_org_model on (organization_id, model_name)idx_app_module on (app_name, module_name)from llm_wrapper import APIError, DatabaseError, ConfigurationError
try:
response = wrapper.send_request(...)
except APIError as e:
print(f"API Error: {e}")
except DatabaseError as e:
print(f"Database Error: {e}")
except ConfigurationError as e:
print(f"Configuration Error: {e}")
import os
config = {
'api_key': os.getenv('AZURE_OPENAI_API_KEY'),
'service_url': os.getenv('AZURE_OPENAI_SERVICE_URL'),
# ...
}
db_config = {
'host': os.getenv('DB_HOST'),
'user': os.getenv('DB_USER'),
'password': os.getenv('DB_PASSWORD'),
'pool_size': 10,
'max_overflow': 20
}
The library uses SQLAlchemy's connection pooling:
db_config = {
'host': 'localhost',
'port': 5432,
'dbname': 'llm_usage',
'user': 'username',
'password': 'password',
'pool_size': 10, # Number of persistent connections
'max_overflow': 20, # Additional connections when needed
'pool_pre_ping': True # Validate connections before use
}
For multiple requests, use connection reuse:
with LLMWrapper(provider_type, config, db_config) as wrapper:
for request_data in batch_requests:
response = wrapper.send_request(**request_data)
# Process response
Configure logging for production:
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('llm_wrapper.log'),
logging.StreamHandler()
]
)
# Old way
wrapper = LLMWrapper(service_url, api_key, deployment_name, api_version, db_config)
response = wrapper.send_request(messages, customer_id, ...)
# New way
wrapper = LLMWrapper(LLMProvider.AZURE_OPENAI, provider_config, db_config)
response = wrapper.send_request(messages, customer_id, ...)
The response format remains the same for backward compatibility.
git checkout -b feature/new-providerpytestBaseLLMProvidersend_chat_completion, create_embeddings, validate_configLLMProviderFactoryThis project is licensed under the MIT License - see the LICENSE file for details.
Made with ❤️ by Hibiz Solutions
FAQs
A comprehensive Python wrapper for Large Language Models with database integration and usage tracking
We found that hibiz-any-llm demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
/Security News
Socket identified 80 fake candidates targeting engineering roles, including suspected North Korean operators, exposing the new reality of hiring as a security function.
Application Security
/Research
/Security News
Socket detected multiple compromised CrowdStrike npm packages, continuing the "Shai-Hulud" supply chain attack that has now impacted nearly 500 packages.
Research
/Security News
Malicious update to @ctrl/tinycolor on npm is part of a supply-chain attack hitting 40+ packages across maintainers