hibiz-llm-wrapper 1.0.3

A comprehensive Python wrapper for Large Language Models with database integration and usage tracking

Hibiz LLM Wrapper

A comprehensive Python wrapper for Azure OpenAI services, specifically designed for Hibiz Solutions' applications. This library provides seamless integration with Azure OpenAI's Chat Completions and Embeddings APIs while offering robust token usage tracking, database logging, and error handling.

Features

• Chat Completions: Support for text and JSON response types with automatic token calculation
• Embeddings: Create embeddings for single or multiple text inputs with comprehensive logging
• Token Usage Tracking: Automatic calculation and database logging of input, output, and total tokens
• Database Integration: PostgreSQL integration for usage analytics and monitoring
• Error Handling: Comprehensive error handling with detailed logging
• Response Time Tracking: Automatic measurement and logging of API response times
• Application Tracking: Enhanced logging with app_name, module_name, and function_name for detailed usage analytics

Installation

pip install hibiz-llm-wrapper

Quick Start

from hibiz_llm_wrapper import LLMWrapper

# Initialize the wrapper
llm = LLMWrapper(
    service_url="https://your-azure-openai-service.openai.azure.com",
    api_key="your-api-key",
    deployment_name="your-chat-deployment",
    api_version="your-api-version",
    default_model="your-model-name",
    default_embedding_model="your-embedding-model-name"
)

# Send a chat completion request
response = llm.send_request(
    prompt_payload=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello, how are you?"}
    ],
    customer_id=123,
    organization_id=456,
    app_name="ChatBot",
    module_name="Conversation",
    function_name="handle_greeting"
)

print(response["output_text"])
print(f"Tokens used: {response['total_tokens']}")

API Reference

LLMWrapper Class

Constructor

LLMWrapper(
    service_url: str,
    api_key: str,
    deployment_name: str,
    api_version: str,
    default_model: str = "gpt-4",
    default_embedding_model: str = "text-embedding-ada-002",
    timeout: int = 600
)

Parameters:

• service_url: Azure OpenAI service endpoint URL
• api_key: Azure OpenAI API key
• deployment_name: Deployment name for chat completions
• api_version: API version (e.g., "2024-02-15-preview")
• default_model: Default model name for chat completions
• default_embedding_model: Default model name for embeddings
• timeout: Request timeout in seconds

Methods

send_request()

Send a chat completion request to Azure OpenAI.

send_request(
    prompt_payload: List[Dict[str, Any]],
    customer_id: int,
    organization_id: int,
    app_name: str,
    module_name: str,
    function_name: str,
    model: Optional[str] = None,
    response_type: str = "text",
    **kwargs
) -> Dict[str, Any]

Parameters:

• prompt_payload: List of message dictionaries (OpenAI chat format)
• customer_id: Customer identifier for tracking
• organization_id: Organization identifier for tracking
• app_name: Application name using the service
• module_name: Module name within the application
• function_name: Specific function name for detailed tracking
• model: Model to use (overrides default)
• response_type: "text" or "json" for response format
• **kwargs: Additional parameters for the API request

Returns:

{
    "output_text": str,
    "processed_output": Any,
    "response_type": str,
    "input_tokens": int,
    "output_tokens": int,
    "total_tokens": int,
    "response_time_ms": int,
    "model": str,
    "app_name": str,
    "module_name": str,
    "function_name": str,
    "full_response": dict,
    "original_prompt": list
}

create_embeddings()

Create embeddings for text inputs using Azure OpenAI.

create_embeddings(
    input_texts: Union[str, List[str]],
    customer_id: int,
    organization_id: int,
    app_name: str,
    module_name: str,
    function_name: str,
    model: Optional[str] = None,
    embedding_deployment_name: Optional[str] = None,
    **kwargs
) -> Dict[str, Any]

Parameters:

• input_texts: Single string or list of strings to embed
• customer_id: Customer identifier for tracking
• organization_id: Organization identifier for tracking
• app_name: Application name using the service
• module_name: Module name within the application
• function_name: Specific function name for detailed tracking
• model: Embedding model to use (overrides default)
• embedding_deployment_name: Specific deployment name for embeddings
• **kwargs: Additional parameters for the embedding API

Returns:

{
    "embeddings": Union[List[float], List[List[float]]],
    "input_tokens": int,
    "output_tokens": int,
    "total_tokens": int,
    "response_time_ms": int,
    "model": str,
    "embedding_count": int,
    "input_text_count": int,
    "app_name": str,
    "module_name": str,
    "function_name": str,
    "original_input": Union[str, List[str]]
}

get_usage_stats()

Retrieve usage statistics from the database.

get_usage_stats(
    customer_id: Optional[int] = None,
    organization_id: Optional[int] = None,
    app_name: Optional[str] = None,
    module_name: Optional[str] = None,
    function_name: Optional[str] = None,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    request_type: Optional[str] = None
) -> Dict[str, Any]

Usage Examples

Basic Chat Completion

# Simple text response
response = llm.send_request(
    prompt_payload=[
        {"role": "user", "content": "What is the capital of France?"}
    ],
    customer_id=123,
    organization_id=456,
    app_name="KnowledgeBot",
    module_name="Geography",
    function_name="get_capital"
)

print(response["output_text"])

JSON Response

# Request structured JSON response
response = llm.send_request(
    prompt_payload=[
        {"role": "user", "content": "List 3 programming languages with their use cases in JSON format"}
    ],
    customer_id=123,
    organization_id=456,
    app_name="DevAssistant",
    module_name="Programming",
    function_name="list_languages",
    response_type="json"
)

# Access parsed JSON data
json_data = response["processed_output"]
print(json_data)

Creating Embeddings

# Single text embedding
embedding_response = llm.create_embeddings(
    input_texts="This is a sample text for embedding",
    customer_id=123,
    organization_id=456,
    app_name="SearchEngine",
    module_name="DocumentProcessing",
    function_name="create_document_embedding"
)

print(f"Embedding dimension: {len(embedding_response['embeddings'])}")

# Multiple text embeddings
texts = [
    "First document text",
    "Second document text",
    "Third document text"
]

batch_response = llm.create_embeddings(
    input_texts=texts,
    customer_id=123,
    organization_id=456,
    app_name="SearchEngine",
    module_name="DocumentProcessing",
    function_name="batch_embed_documents"
)

print(f"Created {batch_response['embedding_count']} embeddings")

Using Context Manager

# Automatic resource cleanup
with LLMWrapper(
    service_url="https://your-service.openai.azure.com",
    api_key="your-key",
    deployment_name="your-deployment",
    api_version="your-api-version"
) as llm:
    response = llm.send_request(
        prompt_payload=[{"role": "user", "content": "Hello!"}],
        customer_id=123,
        organization_id=456,
        app_name="TestApp",
        module_name="Main",
        function_name="test_function"
    )
    print(response["output_text"])

Getting Usage Statistics

# Get usage stats for a specific app
stats = llm.get_usage_stats(
    customer_id=123,
    app_name="ChatBot",
    start_date="2024-01-01T00:00:00",
    end_date="2024-01-31T23:59:59"
)

print(f"Total tokens used: {stats.get('total_tokens', 0)}")

# Get embedding-specific stats
embedding_stats = llm.get_usage_stats(
    organization_id=456,
    request_type="embedding",
    module_name="DocumentProcessing"
)

Enhanced Logging and Tracking

The library automatically logs all requests with detailed information:

• Request Details: Model, parameters, response type
• Token Usage: Input, output, and total token counts
• Performance: Response time in milliseconds
• Application Context: App name, module name, function name
• Status: Success or failure with error details

This enables comprehensive analytics and monitoring of your Azure OpenAI usage across different applications and modules.

Error Handling

The library includes comprehensive error handling:

from llm_wrapper.exceptions import APIError, DatabaseError

try:
    response = llm.send_request(
        prompt_payload=[{"role": "user", "content": "Hello!"}],
        customer_id=123,
        organization_id=456,
        app_name="TestApp",
        module_name="Main",
        function_name="test_function"
    )
except APIError as e:
    print(f"API request failed: {e}")
except DatabaseError as e:
    print(f"Database logging failed: {e}")

Database Schema

The library automatically creates the necessary database tables for token usage tracking. All usage data is stored with customer, organization, and application context for detailed analytics.

Requirements

• Python 3.7+
• requests
• PostgreSQL database
• Azure OpenAI service

Support

This library is developed and maintained exclusively for Hibiz Solutions. For support and questions, please contact the internal development team.

License

Proprietary - Hibiz Solutions Internal Use Only