Elsai Model v1.0.0#
The Elsai Model Package provides a unified interface to connect with multiple LLM providers with enhanced capabilities for v1.0.0:
OpenAI Connector (Synchronous & Asynchronous)
Azure OpenAI Connector (Synchronous & Asynchronous)
AWS Bedrock Connector (Native boto3 implementation)
Gemini Connector (New)
LITELLM Integration (New)
Note
Streaming Functionality: In version 1.0.1, streaming capabilities have been added to both OpenAI and Azure OpenAI connectors (synchronous and asynchronous). This allows for real-time streaming of responses as they are generated.
Prerequisites#
Python >= 3.9
.env file with appropriate API keys and configuration variables
Installation#
To install the elsai-model package:
pip install --index-url https://elsai-core-package.optisolbusiness.com/root/elsai-model/ elsai-model==1.0.0
Components#
1. OpenAIConnector#
OpenAIConnector is a class designed to establish secure connections with the OpenAI API. It retrieves API credentials from environment variables and provides a method to initialize the connection with a specified language model.
from elsai_model.openai import OpenAIConnector
llm = OpenAIConnector(
openai_api_key="your_openai_api_key", # or set OPENAI_API_KEY env var
model_name="gpt-4o-mini", # or set OPENAI_MODEL_NAME env var
temperature=0.1
)
messages = [
{"role": "user", "content": "Hello! Can you tell me a short joke?"}
]
response = llm.invoke(messages=messages)
print(response)
# Streaming functionality (available in v1.0.1)
for response in llm.stream(messages=messages):
print(response, end='', flush=True)
Required Environment Variables:
OPENAI_API_KEY– Your OpenAI API key for authentication.OPENAI_MODEL_NAME– The name of the model to use (e.g., “gpt-4o-mini”).OPENAI_TEMPERATURE– Temperature value to control the randomness of model outputs.
2. AzureOpenAIConnector#
AzureOpenAIConnector is a class that facilitates connecting to the Azure-hosted OpenAI service. It allows configuration through direct parameters or environment variables, and supports deployment-specific model initialization.
from elsai_model.azure_openai import AzureOpenAIConnector
llm = AzureOpenAIConnector(
azure_endpoint="https://your-azure-openai-endpoint.openai.azure.com/",
openai_api_key="your-azure-openai-api-key",
openai_api_version="2023-05-15",
deployment_name="gpt-4o-mini",
temperature=0.1
)
messages = [
{"role": "user", "content": "Hello! Can you tell me a short joke?"}
]
response = llm.invoke(messages=messages)
print(response)
# Streaming functionality (available in v1.0.1)
for msg in llm.stream(messages=messages):
print(msg, end='', flush=True)
Required Environment Variables:
AZURE_OPENAI_API_KEY– API key used to authenticate with the Azure-hosted OpenAI service.AZURE_OPENAI_ENDPOINT– Endpoint URL of your Azure OpenAI resource.OPENAI_API_VERSION– API version to use when connecting to Azure OpenAI (e.g., 2023-05-15).AZURE_OPENAI_TEMPERATURE– Temperature value to control the randomness of model outputs (e.g., 0.0 for deterministic results).AZURE_OPENAI_DEPLOYMENT_NAME– Name of the specific deployment of the OpenAI model you want to use.
3. BedrockConnector#
BedrockConnector has been migrated from LangChain to a native boto3-based implementation, providing lower latency, fewer dependencies, and more fine-grained control over request/response handling.
from elsai_model.bedrock import BedrockConnector
messages = [{
"role": "user",
"content": "What is the capital of France?"
}]
llm = BedrockConnector(
aws_access_key="your_aws_access_key", # or set AWS_ACCESS_KEY_ID env var
aws_secret_key="your_aws_secret_key", # or set AWS_SECRET_ACCESS_KEY env var
aws_session_token="your_session_token", # or set AWS_SESSION_TOKEN env var
aws_region="us-east-1", # or set AWS_REGION env var
max_tokens=500, # Default is 500
temperature=0.1,
model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0", # or set BEDROCK_MODEL_ID env var
config=None
)
result = llm.invoke(messages)
print("LLM Response:", result)
Required Environment Variables:
AWS_ACCESS_KEY_ID– Your AWS access key ID for authenticating with AWS services.AWS_SECRET_ACCESS_KEY– Your AWS secret access key for secure authentication.AWS_SESSION_TOKEN– Temporary session token for secure AWS authentication (used with IAM roles or temporary credentials).AWS_REGION– AWS region (e.g., us-east-1) where the Bedrock service is hosted.BEDROCK_MODEL_ID– The model ID to use (e.g., us.anthropic.claude-3-7-sonnet-20250219-v1:0).BEDROCK_TEMPERATURE– Controls the randomness of the output from the model (optional; default can be set in code).
4. AsyncOpenAIConnector#
AsyncOpenAIConnector is an asynchronous version of the OpenAI connector that provides non-blocking API calls. It’s designed for high-performance applications that need to handle multiple concurrent requests efficiently.
from elsai_model.openai import AsyncOpenAIConnector
async def example_openai_usage():
"""Example usage of AsyncOpenAI connector."""
print("=== AsyncOpenAI Example ===")
# Initialize the async connector
# You can set these via environment variables or pass them directly
connector = AsyncOpenAIConnector(
openai_api_key="your_openai_api_key", # or set OPENAI_API_KEY env var
model_name="gpt-4o-mini", # or set OPENAI_MODEL_NAME env var
temperature=0.1
)
messages = [
{"role": "user", "content": "Hello! Can you tell me a short joke?"}
]
try:
# Make the async call
response = await connector.invoke(messages)
print(f"Response: {response.choices[0].message.content}")
# Streaming functionality (available in v1.0.1)
async for msg in connector.stream(messages=messages):
print(msg, end='', flush=True)
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
import asyncio
# Run the async example
asyncio.run(example_openai_usage())
Required Environment Variables:
OPENAI_API_KEY– Your OpenAI API key for authentication.OPENAI_MODEL_NAME– The name of the model to use (e.g., “gpt-4o-mini”).OPENAI_TEMPERATURE– Temperature value to control the randomness of model outputs.
Key Features:
Asynchronous Operations: All API calls are non-blocking and use Python’s async/await syntax
High Performance: Designed for handling multiple concurrent requests efficiently
Error Handling: Comprehensive exception handling for API errors
Flexible Configuration: Supports both environment variables and direct parameter passing
5. AsyncAzureOpenAIConnector#
AsyncAzureOpenAIConnector provides asynchronous capabilities for Azure OpenAI, enabling non-blocking operations and better scalability for high-throughput workloads.
from elsai_model.azure_openai import AsyncAzureOpenAIConnector
async def example_azure_openai_usage():
"""Example usage of AsyncAzureOpenAI connector."""
print("\n=== AsyncAzureOpenAI Example ===")
# Initialize the async connector
# You can set these via environment variables or pass them directly
llm = AsyncAzureOpenAIConnector(
azure_endpoint="https://your-azure-openai-endpoint.openai.azure.com/", # or set AZURE_OPENAI_ENDPOINT env var
openai_api_key="your-azure-openai-api-key", # or set AZURE_OPENAI_API_KEY env var
openai_api_version="2023-05-15", # or set AZURE_OPENAI_API_VERSION env var
deployment_name="gpt-4o-mini", # or set AZURE_OPENAI_DEPLOYMENT_NAME env var
temperature=0.1
)
messages = [
{"role": "user", "content": "Hello! Can you explain what async programming is?"}
]
try:
# Make the async call
response = await llm(messages)
print(f"Response: {response.choices[0].message.content}")
# Streaming functionality (available in v1.0.1)
async for msg in llm.stream(messages=messages):
print(msg, end='', flush=True)
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
import asyncio
# Run the async example
asyncio.run(example_azure_openai_usage())
Required Environment Variables:
AZURE_OPENAI_API_KEY– API key used to authenticate with the Azure-hosted OpenAI service.AZURE_OPENAI_ENDPOINT– Endpoint URL of your Azure OpenAI resource.OPENAI_API_VERSION– API version to use when connecting to Azure OpenAI (e.g., 2023-05-15).AZURE_OPENAI_TEMPERATURE– Temperature value to control the randomness of model outputs (e.g., 0.0 for deterministic results).AZURE_OPENAI_DEPLOYMENT_NAME– Name of the specific deployment of the OpenAI model you want to use.
6. GeminiConnector (New Feature)#
A new Gemini model connector has been added to the package with comprehensive capabilities. The GeminiService class provides multiple ways to interact with Google’s Gemini models.
import os
from elsai_model.gemini import GeminiService
def test_gemini_service():
# Initialize the service with your API key and model
service = GeminiService(
api_key="your_gemini_api_key", # or set GEMINI_API_KEY env var
model="gemini-2.5-flash"
)
# Single-shot text generation
print("=== Test: generate_text ===")
result = service.generate_text("What is Artificial Intelligence?")
print("Response:", result)
# Streaming text generation
print("=== Test: stream_text ===")
for chunk in service.stream_text("Tell me a short story about a cat"):
print(chunk, end="", flush=True)
# Multi-turn chat
print("=== Test: create_chat ===")
chat = service.create_chat()
first_response = chat.send_message("I have 2 dogs in my house.")
print("Chat response 1:", first_response)
second_response = chat.send_message("How many paws are in my house?")
print("Chat response 2:", second_response)
# View chat history
print("=== Chat History ===")
for msg in chat.get_history():
print(f"{msg['role']}: {msg['text']}")
# Multimodal generation (image + text)
print("=== Test: generate_with_image ===")
try:
image_path = "image.png"
from pathlib import Path
if Path(image_path).exists():
img_response = service.generate_with_image(
image_path=image_path,
prompt="Describe this image in detail."
)
print("Image Response:", img_response)
else:
print(f"[SKIPPED] Image not found at path: {image_path}")
except Exception as e:
print(f"Error during generate_with_image test: {e}")
# Streaming chat messages
print("=== Test: chat.send_message_stream ===")
try:
chat_stream = service.create_chat()
print("Streaming response 1:")
for chunk in chat_stream.send_message_stream("Tell me a fun fact about space."):
print(chunk, end="", flush=True)
print("\n")
print("Streaming response 2:")
for chunk in chat_stream.send_message_stream("And another fun fact please."):
print(chunk, end="", flush=True)
print("\n")
print("=== Chat History (streaming chat) ===")
for msg in chat_stream.get_history():
print(f"{msg['role']}: {msg['text']}")
except Exception as e:
print(f"Error during chat.send_message_stream test: {e}")
if __name__ == "__main__":
test_gemini_service()
Required Environment Variables:
GEMINI_API_KEY– Your Gemini API key for authentication.
Key Features:
Single-shot Generation: Send a single prompt and receive a complete response
Streaming Generation: Receive generated text in real-time streaming manner
Multi-turn Chat: Create chat sessions with context retention
Multimodal Support: Process images along with text prompts
Streaming Chat: Stream responses in chat sessions for real-time interaction
7. LITELLM Integration#
LITELLM integration provides a unified interface for multiple LLM providers with consistent API patterns. The LiteLLMConnector class allows you to use different LLM providers with a single, consistent interface.
from elsai_model.litellm import LiteLLMConnector
import os
# Set your API key (you can also use environment variables)
os.environ["OPENAI_API_KEY"] = "your_openai_api_key"
# Initialize the connector with model name and parameters
connector = LiteLLMConnector(
model_name="openai/gpt-4o-mini", # Format: provider/model-name
temperature=0.1
)
# Use the consistent API across different providers
response = connector.invoke(messages=[{
"content": "Hello, how are you?",
"role": "user"
}])
print(response)
Model Name Formats:
OpenAI: openai/gpt-4o-mini, openai/gpt-4, openai/gpt-3.5-turbo
Azure: azure/your-deployment-name
Bedrock: bedrock/your-deployment-name
Other Providers: Refer to the LiteLLM documentation for supported providers