Demo

Features

• ⚡ Asynchronous request handling for optimal performance
• 🔄 Built-in rate limiting for both requests and tokens
• 🎛️ Configurable concurrency control
• 🔁 Automatic retry mechanism with backoff
• 📊 Progress tracking for batch requests
• 🎯 Token counting and management
• 📝 Comprehensive logging

Installation

pip install speedy-openai

Quick Start

import asyncio
from speedy_openai import OpenAIClient

async def main():
    # Initialize the client
    client = OpenAIClient(
        api_key="your-api-key",
        max_requests_per_min=5000,  # Optional: default 5000
        max_tokens_per_min=15000000,  # Optional: default 15M
        max_concurrent_requests=250  # Optional: default 250
    )

    # Single request
    request = {
        "custom_id": "req1",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-3.5-turbo",
            "messages": [{"role": "user", "content": "Hello!"}]
        }
    }
    
    response = await client.process_request(request)

    # Batch requests
    requests = [request, request]  # List of requests
    responses = await client.process_batch(requests)

if __name__ == "__main__":
    asyncio.run(main())

Configuration Options

Parameter	Default	Description
api_key	Required	Your OpenAI API key
max_requests_per_min	5000	Maximum API requests per minute
max_tokens_per_min	15000000	Maximum tokens per minute
max_concurrent_requests	250	Maximum concurrent requests
max_retries	5	Maximum retry attempts
max_sleep_time	60	Maximum sleep time between retries (seconds)

Features in Detail

Rate Limiting

The client includes a sophisticated rate limiter that manages both request frequency and token usage:

• Automatically tracks remaining requests and tokens
• Updates limits from API response headers
• Implements waiting periods when limits are reached
• Supports dynamic limit adjustments

Concurrency Control

• Manages concurrent requests using asyncio semaphores
• Prevents overwhelming the API with too many simultaneous requests
• Configurable maximum concurrent requests

Retry Mechanism

Built-in retry logic for handling common API errors:

• Automatic retries with fixed wait times
• Configurable maximum retry attempts
• Specific exception handling for API-related errors

Progress Tracking

Batch requests include:

• Progress bar visualization using tqdm
• Processing time logging
• Detailed success/failure reporting

Error Handling

The client includes comprehensive error handling:

• API response validation
• Rate limit handling
• Network error recovery
• Invalid request detection

Requirements

• Python 3.7+
• aiohttp
• tiktoken
• tenacity
• tqdm
• loguru
• pydantic

Common Use Cases

1. Chat Completion with GPT-4

import asyncio
from speedy_openai import OpenAIClient

async def chat_with_gpt4():
    client = OpenAIClient(api_key="your-api-key")
    
    request = {
        "custom_id": "chat-1",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4",
            "messages": [
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": "Explain quantum computing in simple terms."}
            ],
            "temperature": 0.7
        }
    }
    
    response = await client.process_request(request)
    print(response["response"]["choices"][0]["message"]["content"])

asyncio.run(chat_with_gpt4())

2. Batch Processing Multiple Conversations

async def process_multiple_conversations():
    client = OpenAIClient(api_key="your-api-key")
    
    conversations = [
        {"role": "user", "content": "What is AI?"},
        {"role": "user", "content": "Explain machine learning."},
        {"role": "user", "content": "What is deep learning?"}
    ]
    
    requests = [
        {
            "custom_id": f"batch-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-3.5-turbo",
                "messages": [conv],
                "temperature": 0.7
            }
        }
        for i, conv in enumerate(conversations)
    ]
    
    responses = await client.process_batch(requests)
    return responses

Testing

The project uses pytest for testing. To run the tests:

1. Clone the repository:

git clone https://github.com/yourusername/speedy-openai.git
cd speedy-openai

2. Install development dependencies:

poetry install

3. Run tests:

poetry run pytest

Test Structure

The test suite includes:

• Unit tests for core functionality
• Integration tests for API interactions
• Rate limiting tests
• Concurrency tests
• Error handling tests

Writing Tests

When contributing new features, please ensure:

• All new features have corresponding tests
• Test coverage remains above 80%
• Tests are properly documented
• Both success and failure cases are covered

License

This project is licensed under the MIT License - see the LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

For issues, questions, or contributions, please create an issue in the GitHub repository.

Credits

Blogpost by Villoro