DeepAI Lab Python SDK

Official Python SDK for the DeepAI Lab API platform. Supports OpenAI-compatible endpoints, model marketplace, and enterprise features.

🚀 Quick Start (30 seconds)

Installation

pip install deepailab
# or
poetry add deepailab
# or
pipenv install deepailab

Basic Usage

import deepailab

client = deepailab.DeepAILab(
    api_key="sk-deepailab-your-api-key-here",
    # base_url="https://api.deepailab.ai"  # Optional, defaults to production
)

# Chat completion
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Hello, world!"}
    ],
    max_tokens=100
)

print(response.choices[0].message.content)

📚 Features

• ✅ OpenAI Compatible: Drop-in replacement for OpenAI SDK
• ✅ Type Hints: Full type safety with mypy support
• ✅ Async/Await: Native asyncio support
• ✅ Streaming: Server-sent events (SSE) support
• ✅ Error Handling: Comprehensive exception types and retry logic
• ✅ Rate Limiting: Built-in exponential backoff
• ✅ Observability: Request metrics and cost tracking
• ✅ Multi-tenancy: On-behalf-of (OBO) support
• ✅ Model Marketplace: Access to user-published models
• ✅ Context Managers: Automatic resource cleanup
• ✅ Cancellation: asyncio.CancelledError support

🔧 API Reference

Chat Completions

# Synchronous
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain quantum computing"}
    ],
    max_tokens=500,
    temperature=0.7
)

# Asynchronous
import asyncio

async def main():
    async with deepailab.AsyncDeepAILab(api_key="your-key") as client:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello"}]
        )
        print(response.choices[0].message.content)

asyncio.run(main())

Streaming

# Synchronous streaming
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True
)

for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

# Asynchronous streaming
async def stream_example():
    async with deepailab.AsyncDeepAILab(api_key="your-key") as client:
        stream = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Tell me a story"}],
            stream=True
        )
        
        async for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                print(content, end="", flush=True)

Embeddings

response = client.embeddings.create(
    model="text-embedding-3-large",
    input=["Hello world", "How are you?"]
)

print(response.data[0].embedding)  # [0.1, 0.2, ...]

Models

models = client.models.list()
for model in models.data:
    print(f"{model.id}: {model.owned_by}")

Model Marketplace

# Call a user-published model
result = client.model_gateway.infer(
    user_id="user123",
    model_id="my-model",
    input={"text": "Analyze this sentiment"},
    parameters={"temperature": 0.5}
)

# Batch processing
batch = client.model_gateway.batch(
    user_id="user123",
    model_id="my-model",
    input_file_id="file-abc123",
    endpoint="/v1/inference"
)

# Check model status
status = client.model_gateway.status("user123", "my-model")
print(status.status)  # 'deployed' | 'deploying' | 'failed' | 'maintenance'

On-Behalf-Of (Multi-tenancy)

# Make requests on behalf of end users
obo_client = client.as_user(
    user="end-user-123",
    tenant="organization-456"
)

response = obo_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
# Usage will be attributed to end-user-123 and organization-456

🔒 Security Best Practices

Environment Variables

import os
import deepailab

# ✅ Use environment variables
client = deepailab.DeepAILab(
    api_key=os.getenv("DEEPAILAB_API_KEY")
)

# ✅ Or use a .env file with python-dotenv
from dotenv import load_dotenv
load_dotenv()

client = deepailab.DeepAILab(
    api_key=os.getenv("DEEPAILAB_API_KEY")
)

Session Tokens for Web Apps

# For web applications, use short-lived session tokens
def get_session_token(user_jwt: str) -> str:
    """Get a short-lived session token from your auth service."""
    # Your implementation here
    pass

client = deepailab.DeepAILab(
    api_key=get_session_token(user_jwt)  # Short-lived token
)

📊 Observability & Metrics

def metrics_callback(metrics):
    print(f"Request ID: {metrics.request_id}")
    print(f"Response Time: {metrics.response_time}ms")
    print(f"Tokens Used: {metrics.usage.total_tokens}")
    print(f"Cost: {metrics.cost.amount} {metrics.cost.currency}")

client = deepailab.DeepAILab(
    api_key="your-key",
    on_metrics=metrics_callback
)

🔄 Error Handling & Retries

import deepailab
from deepailab import (
    RateLimitError,
    AuthenticationError,
    TimeoutError,
    ValidationError
)

try:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}]
    )
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}")
except AuthenticationError:
    print("Invalid API key")
except TimeoutError:
    print("Request timed out")
except ValidationError as e:
    print(f"Validation error: {e.message}")
except deepailab.DeepAILabError as e:
    print(f"Other error: {e}")

🔧 Configuration

client = deepailab.DeepAILab(
    api_key="your-key",
    base_url="https://api.deepailab.ai",  # Custom base URL
    timeout=30.0,  # Request timeout in seconds
    max_retries=3,  # Maximum retry attempts
    default_headers={"User-Agent": "MyApp/1.0"},  # Custom headers
    debug=True  # Enable debug logging
)

🧪 Testing

# Use the test client for unit tests
from deepailab.testing import MockDeepAILab

def test_chat_completion():
    client = MockDeepAILab()
    client.mock_response("chat.completions.create", {
        "choices": [{"message": {"content": "Hello!"}}]
    })
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hi"}]
    )
    
    assert response.choices[0].message.content == "Hello!"

📖 More Examples

• Synchronous Example
• Asynchronous Example
• Streaming Example
• Model Marketplace Example
• On-Behalf-Of Example
• Django Integration
• FastAPI Integration

🤝 Contributing

We welcome contributions! Please see our Contributing Guide for details.

📄 License

MIT License - see LICENSE for details.

🆘 Support

• 📚 Documentation
• 🐛 Bug Reports
• 💬 Discord Community
• 📧 Email Support