openai-api

name	openai-api
description	Build with OpenAI APIs including GPT-4, GPT-4o, function calling, embeddings, vision, and Assistants. Covers chat completions, structured outputs, streaming, and token optimization. Use when integrating OpenAI models into applications.

OpenAI API Skill

Build production-ready applications with OpenAI's GPT-4, GPT-4o, embeddings, vision, and Assistants API.

Quick Reference

Feature	Model	Use Case
Chat Completions	gpt-4o, gpt-4-turbo	Conversations, reasoning
Structured Outputs	gpt-4o-2024-08-06+	JSON schemas, typed responses
Function Calling	gpt-4o, gpt-4-turbo	Tool use, API integration
Vision	gpt-4o, gpt-4-vision-preview	Image analysis
Embeddings	text-embedding-3-small/large	Semantic search, RAG
Assistants	gpt-4o, gpt-4-turbo	Stateful agents, file handling

---

Installation

# Python
pip install openai

# Node.js / TypeScript
npm install openai

Client Setup

Python:

from openai import OpenAI

# Uses OPENAI_API_KEY env var by default
client = OpenAI()

# Or explicit key
client = OpenAI(api_key="sk-...")

TypeScript:

import OpenAI from "openai";

const openai = new OpenAI(); // Uses OPENAI_API_KEY env var
// Or: new OpenAI({ apiKey: 'sk-...' })

---

Chat Completions

Basic Chat

Python:

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

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

TypeScript:

const response = await openai.chat.completions.create({
  model: "gpt-4o",
  messages: [
    { role: "system", content: "You are a helpful assistant." },
    { role: "user", content: "Explain async/await in TypeScript" },
  ],
  temperature: 0.7,
  max_tokens: 500,
});

console.log(response.choices[0].message.content);

Multi-Turn Conversation

conversation = [
    {"role": "system", "content": "You are a Python tutor."}
]

def chat(user_message: str) -> str:
    conversation.append({"role": "user", "content": user_message})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=conversation
    )

    assistant_message = response.choices[0].message.content
    conversation.append({"role": "assistant", "content": assistant_message})

    return assistant_message

Model Selection Guide

Model	Context	Best For	Cost
gpt-4o	128K	General purpose, vision	$$
gpt-4o-mini	128K	Fast, cost-effective	$
gpt-4-turbo	128K	Complex reasoning	$$$
gpt-3.5-turbo	16K	Simple tasks	$

---

Structured Outputs / JSON Mode

JSON Mode (Legacy)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Output valid JSON only."},
        {"role": "user", "content": "List 3 programming languages with their use cases"}
    ],
    response_format={"type": "json_object"}
)

import json
data = json.loads(response.choices[0].message.content)

Structured Outputs with JSON Schema (Recommended)

Python with Pydantic:

from pydantic import BaseModel
from typing import List

class Language(BaseModel):
    name: str
    use_cases: List[str]
    difficulty: str

class LanguageList(BaseModel):
    languages: List[Language]

response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "user", "content": "List 3 programming languages"}
    ],
    response_format=LanguageList
)

# Typed response
languages: LanguageList = response.choices[0].message.parsed
for lang in languages.languages:
    print(f"{lang.name}: {lang.difficulty}")

TypeScript with Zod:

import { z } from "zod";
import { zodResponseFormat } from "openai/helpers/zod";

const LanguageSchema = z.object({
  languages: z.array(
    z.object({
      name: z.string(),
      use_cases: z.array(z.string()),
      difficulty: z.enum(["beginner", "intermediate", "advanced"]),
    }),
  ),
});

const response = await openai.beta.chat.completions.parse({
  model: "gpt-4o-2024-08-06",
  messages: [{ role: "user", content: "List 3 programming languages" }],
  response_format: zodResponseFormat(LanguageSchema, "languages"),
});

const languages = response.choices[0].message.parsed;

---

Function Calling / Tools

Define Tools

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City and state, e.g., San Francisco, CA"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "default": "fahrenheit"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Search products in database",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "limit": {"type": "integer", "default": }
                },
                : []
            }
        }
    }
]

Execute Tool Calls

import json

def execute_tool(name: str, args: dict) -> str:
    """Execute tool and return result as string."""
    if name == "get_weather":
        # Call actual weather API
        return json.dumps({"temp": 72, "condition": "sunny"})
    elif name == "search_database":
        # Query database
        return json.dumps({"results": ["Product A", "Product B"]})
    return json.dumps({"error": "Unknown tool"})

def chat_with_tools(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        tool_choice="auto"  # or "required" to force tool use
    )

    message = response.choices[0].message

    # Check if model wants to call tools
    if message.tool_calls:
        messages.append(message)  # Add assistant message with tool calls

        # Execute each tool call
        for tool_call in message.tool_calls:
            result = execute_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments)
            )
            messages.append({
                : ,
                : tool_call.,
                : result
            })

        
        final_response = client.chat.completions.create(
            model=,
            messages=messages,
            tools=tools
        )
         final_response.choices[].message.content

     message.content

Parallel Tool Calls

# Model can call multiple tools in parallel
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What's the weather in NYC and LA?"}],
    tools=tools,
    parallel_tool_calls=True  # Default is True
)

# Process all tool calls
for tool_call in response.choices[0].message.tool_calls:
    print(f"Tool: {tool_call.function.name}")
    print(f"Args: {tool_call.function.arguments}")

---

Vision (Image Inputs)

Analyze Images

Python:

import base64

def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.standard_b64encode(f.read()).decode("utf-8")

# From URL
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "What's in this image?"},
            {
                "type": "image_url",
                "image_url": {"url": "https://example.com/image.jpg"}
            }
        ]
    }]
)

# From base64
image_data = encode_image("screenshot.png")
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Describe this UI screenshot"},
            {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/png;base64,{image_data}",
                    "detail": "high"  # "low", "high", or "auto"
                }
            }
        ]
    }]
)

Multiple Images

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Compare these two images"},
            {"type": "image_url", "image_url": {"url": "https://example.com/before.jpg"}},
            {"type": "image_url", "image_url": {"url": "https://example.com/after.jpg"}}
        ]
    }]
)

---

Embeddings

Generate Embeddings

Python:

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    response = client.embeddings.create(
        input=text,
        model=model
    )
    return response.data[0].embedding

# Single text
embedding = get_embedding("OpenAI makes great APIs")
print(f"Dimensions: {len(embedding)}")  # 1536 for small, 3072 for large

# Batch embeddings (more efficient)
texts = ["First document", "Second document", "Third document"]
response = client.embeddings.create(
    input=texts,
    model="text-embedding-3-small"
)
embeddings = [item.embedding for item in response.data]

Reduced Dimensions

# Use dimensions parameter for smaller embeddings
response = client.embeddings.create(
    input="Sample text",
    model="text-embedding-3-large",
    dimensions=256  # Reduce from 3072 to 256
)

Semantic Search Example

import numpy as np
from typing import List, Tuple

def cosine_similarity(a: List[float], b: List[float]) -> float:
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def semantic_search(query: str, documents: List[str], top_k: int = 3) -> List[Tuple[str, float]]:
    query_embedding = get_embedding(query)
    doc_embeddings = [get_embedding(doc) for doc in documents]

    similarities = [
        (doc, cosine_similarity(query_embedding, emb))
        for doc, emb in zip(documents, doc_embeddings)
    ]

    return sorted(similarities, key=lambda x: x[1], reverse=True)[:top_k]

---

Streaming Responses

Basic Streaming

Python:

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

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

TypeScript:

const stream = await openai.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Write a short story" }],
  stream: true,
});

for await (const chunk of stream) {
  const content = chunk.choices[0]?.delta?.content;
  if (content) process.stdout.write(content);
}

Streaming with Tools

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What's the weather?"}],
    tools=tools,
    stream=True
)

tool_calls = []
current_tool = None

for chunk in stream:
    delta = chunk.choices[0].delta

    if delta.tool_calls:
        for tc in delta.tool_calls:
            if tc.index >= len(tool_calls):
                tool_calls.append({
                    "id": tc.id,
                    "function": {"name": tc.function.name, "arguments": ""}
                })
            if tc.function.arguments:
                tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments

    if delta.content:
        print(delta.content, end="", flush=True)

---

Assistants API

Create an Assistant

assistant = client.beta.assistants.create(
    name="Data Analyst",
    instructions="You are a data analyst. Analyze data and create visualizations.",
    model="gpt-4o",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"}
    ]
)

Run a Conversation

# Create thread
thread = client.beta.threads.create()

# Add message
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Analyze this CSV and create a chart"
)

# Run assistant
run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id
)

if run.status == "completed":
    messages = client.beta.threads.messages.list(thread_id=thread.id)
    for msg in messages.data:
        if msg.role == "assistant":
            print(msg.content[0].text.value)

File Handling

# Upload file
file = client.files.create(
    file=open("data.csv", "rb"),
    purpose="assistants"
)

# Create vector store for file search
vector_store = client.beta.vector_stores.create(name="Knowledge Base")
client.beta.vector_stores.files.create(
    vector_store_id=vector_store.id,
    file_id=file.id
)

# Attach to assistant
client.beta.assistants.update(
    assistant_id=assistant.id,
    tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}}
)

---

Token Management & Cost Optimization

Count Tokens

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def count_message_tokens(messages: list, model: str = "gpt-4o") -> int:
    encoding = tiktoken.encoding_for_model(model)
    tokens = 0
    for msg in messages:
        tokens += 4  # Message overhead
        tokens += len(encoding.encode(msg["content"]))
    tokens += 2  # Reply priming
    return tokens

Optimize Context

def truncate_to_token_limit(text: str, max_tokens: int, model: str = "gpt-4o") -> str:
    encoding = tiktoken.encoding_for_model(model)
    tokens = encoding.encode(text)
    if len(tokens) <= max_tokens:
        return text
    return encoding.decode(tokens[:max_tokens])

def manage_conversation_context(
    messages: list,
    max_context_tokens: int = 8000,
    model: str = "gpt-4o"
) -> list:
    """Keep conversation under token limit by removing old messages."""
    while count_message_tokens(messages, model) > max_context_tokens:
        # Keep system message, remove oldest user/assistant pair
        if len(messages) > 3:
            messages.pop(1)
            messages.pop(1)
        else:
            break
    return messages

Cost Tracking

# Pricing per 1M tokens (as of 2024)
PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4-turbo": {"input": 10.00, "output": 30.00},
    "text-embedding-3-small": {"input": 0.02},
    "text-embedding-3-large": {"input": 0.13}
}

def calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int = 0
) -> float:
    pricing = PRICING.get(model, {})
    input_cost = (input_tokens / 1_000_000) * pricing.get("input", 0)
    output_cost = (output_tokens / 1_000_000) * pricing.get("output", 0)
    return input_cost + output_cost

# Track usage from response
response = client.chat.completions.create(...)
usage = response.usage
cost = calculate_cost(
    "gpt-4o",
    usage.prompt_tokens,
    usage.completion_tokens
)
print(f"Cost: ${cost:.4f}")

---

Error Handling & Retries

Retry with Exponential Backoff

import time
from openai import RateLimitError, APITimeoutError, APIConnectionError

def chat_with_retry(
    messages: list,
    model: str = "gpt-4o",
    max_retries: int = 3,
    base_delay: float = 1.0
) -> str:
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages
            )
            return response.choices[0].message.content

        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"Rate limited. Retrying in {delay}s...")
            time.sleep(delay)

        except (APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"Connection error. Retrying in {delay}s...")
            time.sleep(delay)

    raise Exception("Max retries exceeded")

Using Tenacity

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from openai import RateLimitError, APITimeoutError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type((RateLimitError, APITimeoutError))
)
def robust_chat(messages: list, model: str = "gpt-4o") -> str:
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    return response.choices[0].message.content

Handle All Errors

from openai import (
    OpenAIError,
    APIError,
    AuthenticationError,
    BadRequestError,
    RateLimitError
)

try:
    response = client.chat.completions.create(...)
except AuthenticationError:
    print("Invalid API key")
except BadRequestError as e:
    print(f"Invalid request: {e.message}")
except RateLimitError:
    print("Rate limited - implement backoff")
except APIError as e:
    print(f"API error: {e.status_code} - {e.message}")
except OpenAIError as e:
    print(f"OpenAI error: {e}")

---

Best Practices

System Prompt Design

SYSTEM_PROMPT = """You are a helpful assistant for {company_name}.

## Your Role
- Answer questions about our products
- Help troubleshoot issues
- Escalate complex problems to human support

## Guidelines
- Be concise and direct
- Use bullet points for lists
- If unsure, say so honestly
- Never make up information

## Tone
- Professional but friendly
- Patient with confused users
- Empathetic to frustrations
"""

Prompt Templates

from string import Template

ANALYSIS_TEMPLATE = Template("""
Analyze the following $content_type:

---
$content
---

Provide:
1. Summary (2-3 sentences)
2. Key points (bullet list)
3. Recommendations (if applicable)
""")

prompt = ANALYSIS_TEMPLATE.substitute(
    content_type="customer feedback",
    content="The product is great but shipping was slow..."
)

Production Checklist

Category	Recommendation
Security	Never expose API keys; use env vars
Rate Limits	Implement exponential backoff
Costs	Set usage limits; monitor daily spend
Latency	Use streaming for long responses
Reliability	Add fallback models (4o → 4o-mini)
Logging	Log prompts, responses, tokens, costs
Testing	Test with edge cases; mock in tests

Fallback Pattern

MODELS = ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"]

def chat_with_fallback(messages: list) -> str:
    for model in MODELS:
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                timeout=30
            )
            return response.choices[0].message.content
        except Exception as e:
            print(f"{model} failed: {e}")
            continue
    raise Exception("All models failed")

---

Quick Patterns

Simple Q&A

def ask(question: str) -> str:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}]
    ).choices[0].message.content

Summarize Text

def summarize(text: str, max_words: int = 100) -> str:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Summarize in {max_words} words:\n\n{text}"
        }]
    ).choices[0].message.content

Extract Structured Data

from pydantic import BaseModel

class Contact(BaseModel):
    name: str
    email: str | None
    phone: str | None

def extract_contact(text: str) -> Contact:
    response = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[{
            "role": "user",
            "content": f"Extract contact info:\n\n{text}"
        }],
        response_format=Contact
    )
    return response.choices[0].message.parsed

---

Resources

• OpenAI API Reference
• OpenAI Cookbook
• Pricing
• Rate Limits
• Tiktoken Tokenizer