OpenAI API Integration: Complete Python Guide for Building AI Applications

# Install the OpenAI Python SDK
# pip install openai

# Recommended: Use environment variables for API keys
# Never hardcode API keys in source code

# Option 1: Environment variable (recommended)
import os
os.environ["OPENAI_API_KEY"] = "sk-..."  # Usually set in .env file

# Option 2: .env file with python-dotenv
# pip install python-dotenv
from dotenv import load_dotenv
load_dotenv()  # Loads from .env file

# Option 3: Pass directly (only for testing)
from openai import OpenAI
client = OpenAI(api_key="sk-...")

# Recommended: let SDK read from environment
client = OpenAI()  # Reads OPENAI_API_KEY automatically

# Test your connection
response = client.models.list()
print([model.id for model in response.data[:5]])

from openai import OpenAI

client = OpenAI()

# Basic completion
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the difference between a list and a tuple in Python?"}
    ],
    temperature=0.7,       # 0 = deterministic, 2 = very creative
    max_tokens=500,        # Maximum response length
    n=1,                   # Number of responses to generate
)

# Access the response
message = response.choices[0].message.content
print(message)

# Token usage
print(f"Input tokens: {response.usage.prompt_tokens}")
print(f"Output tokens: {response.usage.completion_tokens}")
print(f"Total tokens: {response.usage.total_tokens}")

# Streaming
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Explain decorators in Python"}],
    stream=True
)

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

# JSON mode (guaranteed valid JSON output)
json_response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "user",
            "content": "Extract: name, age, email from: 'John Smith, 30, john@example.com'"
        }
    ],
    response_format={"type": "json_object"}
)

import json
data = json.loads(json_response.choices[0].message.content)
print(data)  # {"name": "John Smith", "age": 30, "email": "john@example.com"}

import json

# Define tools (functions the model can call)
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather in a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City and country, e.g. 'London, UK'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Search product database for items matching a query",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "max_results": {"type": "integer", "description": "Maximum results to return"}
                },
                "required": ["query"]
            }
        }
    }
]

# Actual function implementations
def get_current_weather(location: str, unit: str = "celsius") -> str:
    # In production: call a real weather API
    return f"The weather in {location} is 22°{unit[0].upper()}, partly cloudy."

def search_database(query: str, max_results: int = 5) -> str:
    # In production: query your database
    return f"Found {max_results} products matching '{query}': [Product 1, Product 2...]"

def execute_tool(tool_call) -> str:
    """Execute the appropriate function based on tool call."""
    function_name = tool_call.function.name
    arguments = json.loads(tool_call.function.arguments)
    
    if function_name == "get_current_weather":
        return get_current_weather(**arguments)
    elif function_name == "search_database":
        return search_database(**arguments)
    else:
        return f"Unknown function: {function_name}"

def agent_with_tools(user_message: str) -> str:
    """Simple agent loop with tool execution."""
    
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=tools,
            tool_choice="auto"  # Model decides when to use tools
        )
        
        choice = response.choices[0]
        
        if choice.finish_reason == "stop":
            # Model finished without calling a tool
            return choice.message.content
        
        elif choice.finish_reason == "tool_calls":
            # Model wants to call one or more tools
            messages.append(choice.message)  # Add assistant message with tool calls
            
            for tool_call in choice.message.tool_calls:
                result = execute_tool(tool_call)
                
                # Add tool result to messages
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
        else:
            break
    
    return messages[-1]["content"]

# Example
answer = agent_with_tools("What's the weather in Tokyo, and search for umbrellas?")
print(answer)

# Generate embeddings for semantic search
response = client.embeddings.create(
    model="text-embedding-3-small",  # 1536 dimensions, $0.02/1M tokens
    input=[
        "Machine learning is a subset of AI.",
        "Python is great for data science.",
        "The weather is sunny today."
    ]
)

embeddings = [item.embedding for item in response.data]
print(f"Embedding dimension: {len(embeddings[0])}")  # 1536

# Semantic similarity
import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# ML and data science should be more similar than weather
sim_ml_ds = cosine_similarity(embeddings[0], embeddings[1])
sim_ml_weather = cosine_similarity(embeddings[0], embeddings[2])

print(f"ML ↔ Data Science: {sim_ml_ds:.3f}")   # ~0.82
print(f"ML ↔ Weather: {sim_ml_weather:.3f}")     # ~0.12

# Dimension reduction (Matryoshka embeddings)
response_reduced = client.embeddings.create(
    model="text-embedding-3-small",
    input="Machine learning explanation",
    dimensions=256  # Reduce from 1536 to 256 (cost savings, small quality loss)
)

import base64

def analyze_image(image_path: str, prompt: str) -> str:
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{image_data}",
                            "detail": "high"  # or "low" for faster/cheaper
                        }
                    },
                    {"type": "text", "text": prompt}
                ]
            }
        ]
    )
    return response.choices[0].message.content

# Extract data from invoice
invoice_data = analyze_image(
    "invoice.jpg",
    "Extract all data as JSON: invoice number, date, items with prices, total."
)

import time
from openai import OpenAI, RateLimitError, APIConnectionError, APIStatusError

client = OpenAI(max_retries=5)  # SDK handles retries automatically

def robust_completion(messages: list, max_attempts: int = 3) -> str:
    """Completion with explicit error handling."""
    
    for attempt in range(max_attempts):
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                timeout=30.0  # 30 second timeout
            )
            return response.choices[0].message.content
        
        except RateLimitError as e:
            if attempt == max_attempts - 1:
                raise
            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
        
        except APIConnectionError:
            if attempt == max_attempts - 1:
                raise
            time.sleep(1)
        
        except APIStatusError as e:
            if e.status_code == 400:
                raise  # Bad request — don't retry
            if attempt == max_attempts - 1:
                raise
            time.sleep(1)
    
    raise RuntimeError("Max attempts reached")