Skip to main content

Python Integration

Use Python with the OpenAI SDK to connect to Enclava.

OpenAI SDK​

Installation​

pip install openai

Basic Configuration​

import openai

# Configure with Enclava endpoint
openai.api_base = "https://your-enclava-instance/api/v1"
openai.api_key = "YOUR_API_KEY"

# Make a request
response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello, Enclava!"}]
)

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

Complete Example​

import openai

class EnclavaClient:
    def __init__(self, api_key, base_url="https://localhost/api/v1"):
        openai.api_base = base_url
        openai.api_key = api_key

    def chat(self, messages, model="gpt-3.5-turbo", **kwargs):
        """Send chat completion request"""
        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            **kwargs
        )
        return response.choices[0].message.content

    def embed(self, text):
        """Generate embeddings"""
        response = openai.Embedding.create(
            model="text-embedding-ada-002",
            input=text
        )
        return response["data"][0]["embedding"]

    def get_models(self):
        """List available models"""
        response = openai.Model.list()
        return response["data"]

# Usage
client = EnclavaClient(api_key="YOUR_API_KEY")

# Chat
response = client.chat([{"role": "user", "content": "Tell me a joke"}])
print(response)

# Embed
embedding = client.embed("Hello, world!")
print(f"Embedding dimensions: {len(embedding)}")

# List models
models = client.get_models()
for model in models:
    print(f"Model: {model['id']}")

Streaming Example​

import openai

openai.api_base = "https://your-enclava-instance/api/v1"
openai.api_key = "YOUR_API_KEY"

# Stream responses
stream = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Write a story about AI"}],
    stream=True
)

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

Async Example​

import openai
import asyncio

openai.api_base = "https://your-enclava-instance/api/v1"
openai.api_key = "YOUR_API_KEY"

async def chat_async(messages):
    """Async chat completion"""
    response = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=messages
    )
    return response.choices[0].message.content

async def main():
    messages = [{"role": "user", "content": "What is AI?"}]
    response = await chat_async(messages)
    print(response)

asyncio.run(main())

Error Handling​

import openai
import time

openai.api_base = "https://your-enclava-instance/api/v1"
openai.api_key = "YOUR_API_KEY"

def chat_with_retry(messages, max_retries=3):
    """Chat with retry logic"""
    for attempt in range(max_retries):
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages
            )
            return response.choices[0].message.content
        except openai.error.RateLimitError:
            print(f"Rate limited. Attempt {attempt + 1}/{max_retries}")
            wait_time = 2 ** attempt
            time.sleep(wait_time)
        except openai.error.APIError as e:
            print(f"API error: {e}")
            raise
    raise Exception("Max retries exceeded")

# Usage
response = chat_with_retry([{"role": "user", "content": "Hello"}])
print(response)

Requests Library​

If you prefer direct HTTP requests:

import requests

class EnclavaClient:
    def __init__(self, api_key, base_url="https://localhost/api/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def chat(self, messages, **kwargs):
        """Chat completion"""
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json={
                "model": kwargs.get("model", "gpt-3.5-turbo"),
                "messages": messages,
                **kwargs
            }
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]

    def embed(self, text):
        """Generate embeddings"""
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers=self.headers,
            json={
                "model": "text-embedding-ada-002",
                "input": text
            }
        )
        response.raise_for_status()
        return response.json()["data"][0]["embedding"]

# Usage
client = EnclavaClient(api_key="YOUR_API_KEY")

response = client.chat([{"role": "user", "content": "Hello"}])
print(response)

embedding = client.embed("Hello, world!")
print(f"Embedding: {len(embedding)} dimensions")

Best Practices​

  1. Environment Variables: Store API keys in environment, not code
  2. Retry Logic: Implement exponential backoff for rate limits
  3. Error Handling: Catch and handle specific errors
  4. Streaming: Use streaming for long responses
  5. Async: Use async for concurrent requests

Next Steps​