How to Build Your First AI Wrapper Tool using OpenAI APIs – A Complete Beginner’s Guide

				
					# ==============================
# 1. Setting Up Your Environment
# ==============================

# Install dependencies:
# pip install openai python-dotenv requests

# Create a .env file and store your API key like this:
# OPENAI_API_KEY=your_api_key_here

# ============================
# 2. Basic Wrapper Structure
# ============================

import os
import time
import logging
from dotenv import load_dotenv
import openai

# Load environment variables
load_dotenv()

class AIWrapper:
    def __init__(self, api_key=None):
        self.client = openai.OpenAI(
            api_key=api_key or os.getenv("OPENAI_API_KEY")
        )
        self.setup_logging()

    def setup_logging(self):
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

    def generate_text(self, prompt, model="gpt-3.5-turbo", max_tokens=150):
        """Generate text using OpenAI's chat completion API"""
        try:
            response = self.client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens,
                temperature=0.7
            )
            return response.choices[0].message.content.strip()

        except openai.RateLimitError:
            self.logger.warning("Rate limit exceeded, waiting...")
            time.sleep(60)
            return self.generate_text(prompt, model, max_tokens)

        except Exception as e:
            self.logger.error(f"API call failed: {str(e)}")
            return None


# Example usage
wrapper = AIWrapper()
result = wrapper.generate_text("Explain quantum computing in simple terms")
print(result)

# ======================================
# 3. Advanced Error Handling with Retry
# ======================================

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    """Decorator for retry logic with exponential backoff"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)

                except openai.RateLimitError:
                    wait_time = delay * (2 ** attempt)
                    time.sleep(wait_time)
                    if attempt == max_retries - 1:
                        raise

                except openai.APIError:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay)

            return None
        return wrapper
    return decorator


class EnhancedAIWrapper(AIWrapper):
    @retry_on_failure(max_retries=3)
    def safe_generate_text(self, prompt, **kwargs):
        return self.generate_text(prompt, **kwargs)

# ==================================
# 4. Practical Example: Q&A Chat Bot
# ==================================

class QABot(EnhancedAIWrapper):
    def __init__(self, context=""):
        super().__init__()
        self.context = context
        self.conversation_history = []

    def ask_question(self, question, include_context=True):
        """Process a question with optional context and conversation history"""
        prompt_parts = []

        if include_context and self.context:
            prompt_parts.append(f"Context: {self.context}")

        if self.conversation_history:
            prompt_parts.append("Previous conversation:")
            for qa in self.conversation_history[-3:]:  # keep last 3
                prompt_parts.append(f"Q: {qa['question']}")
                prompt_parts.append(f"A: {qa['answer']}")

        prompt_parts.append(f"Question: {question}")
        prompt_parts.append("Answer:")

        full_prompt = "\n".join(prompt_parts)

        answer = self.safe_generate_text(
            full_prompt,
            model="gpt-3.5-turbo",
            max_tokens=200
        )

        if answer:
            self.conversation_history.append({
                "question": question,
                "answer": answer
            })

        return answer


# Example usage
bot = QABot(context="You are a helpful programming assistant specializing in Python.")
response = bot.ask_question("How do I handle exceptions in Python?")
print(response)

# ======================================
# 5. Prompt Templates for Consistency
# ======================================

class PromptTemplate:
    def __init__(self, template):
        self.template = template

    def format(self, **kwargs):
        return self.template.format(**kwargs)


class TemplatedAIWrapper(EnhancedAIWrapper):
    def __init__(self):
        super().__init__()
        self.templates = {
            "summarizer": PromptTemplate(
                "Summarize the following text in {word_count} words:\n\n{text}"
            ),
            "translator": PromptTemplate(
                "Translate the following text from {source_lang} to {target_lang}:\n\n{text}"
            )
        }

    def summarize(self, text, word_count=50):
        prompt = self.templates["summarizer"].format(
            text=text,
            word_count=word_count
        )
        return self.safe_generate_text(prompt)

# ==============================
# 6. Logging and Monitoring
# ==============================

import json
from datetime import datetime

class MonitoredAIWrapper(TemplatedAIWrapper):
    def __init__(self):
        super().__init__()
        self.usage_stats = {
            "total_requests": 0,
            "total_tokens": 0,
            "errors": 0
        }

    def log_request(self, prompt, response, tokens_used=0):
        """Log API usage for monitoring and debugging"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "prompt_length": len(prompt),
            "response_length": len(response) if response else 0,
            "tokens_used": tokens_used,
            "success": response is not None
        }
        self.logger.info(f"API Request: {json.dumps(log_entry)}")

        # Update usage statistics
        self.usage_stats["total_requests"] += 1
        self.usage_stats["total_tokens"] += tokens_used
        if not response:
            self.usage_stats["errors"] += 1

    def get_usage_stats(self):
        return self.usage_stats.copy()

# ==============================
# 7. Structured JSON Responses
# ==============================

class StructuredAIWrapper(MonitoredAIWrapper):
    def generate_json_response(self, prompt, schema_description=""):
        """Generate structured JSON responses"""
        structured_prompt = f"""
        {prompt}
        {schema_description}
        Please respond with valid JSON only.
        """
        response = self.safe_generate_text(
            structured_prompt,
            model="gpt-3.5-turbo",
            max_tokens=500
        )
        if response:
            try:
                return json.loads(response)
            except json.JSONDecodeError:
                self.logger.error("Failed to parse JSON response")
                return None
        return None

# ==============================
# 8. Cost Optimization Wrapper
# ==============================

class CostOptimizedWrapper(StructuredAIWrapper):
    def __init__(self, daily_token_limit=10000):
        super().__init__()
        self.daily_token_limit = daily_token_limit
        self.daily_usage = 0
        self.usage_date = datetime.now().date()

    def check_token_limit(self, estimated_tokens):
        current_date = datetime.now().date()

        # Reset daily usage if it's a new day
        if current_date != self.usage_date:
            self.daily_usage = 0
            self.usage_date = current_date

        return self.daily_usage + estimated_tokens <= self.daily_token_limit

# ============================
# 9. Caching Layer
# ============================

import hashlib

class CachedAIWrapper(CostOptimizedWrapper):
    def __init__(self, cache_size=128):
        super().__init__()
        self.response_cache = {}
        self.cache_size = cache_size

    def _cache_key(self, prompt, model):
        """Generate cache key from prompt and model"""
        content = f"{prompt}:{model}"
        return hashlib.md5(content.encode()).hexdigest()

    def cached_generate_text(self, prompt, model="gpt-3.5-turbo", **kwargs):
        cache_key = self._cache_key(prompt, model)

        # Check cache first
        if cache_key in self.response_cache:
            self.logger.info("Cache hit")
            return self.response_cache[cache_key]

        # Generate new response
        response = self.safe_generate_text(prompt, model=model, **kwargs)

        # Store in cache (with size limit)
        if response and len(self.response_cache) < self.cache_size:
            self.response_cache[cache_key] = response

        return response

============================
10. Async Support
============================

import asyncio
import aiohttp

class AsyncAIWrapper:
    def __init__(self, api_key=None):
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
        self.base_url = "https://api.openai.com/v1"

    async def async_generate_text(self, prompt, model="gpt-3.5-turbo"):
        """Async text generation for high-throughput applications"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 150
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                data = await response.json()
                return data["choices"][0]["message"]["content"]