OpenAI API Deep Dive: Building AI Applications in Australia

Getting Started (Python)

from openai import OpenAI

# Initialize client with API key
client = OpenAI(api_key="sk-...")  # Or set OPENAI_API_KEY env var

# Basic completion
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"}
    ]
)

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

Getting Started (JavaScript/TypeScript)

import OpenAI from 'openai';

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

const response = await openai.chat.completions.create({
  model: "gpt-4o-mini",
  messages: [
    { role: "system", content: "You are a helpful assistant." },
    { role: "user", content: "Hello!" }
  ]
});

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

Function Calling Example (Python)

import json
from openai import OpenAI

client = OpenAI()

# Define available functions
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_customer_info",
            "description": "Retrieve customer information from CRM",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {
                        "type": "string",
                        "description": "The customer's unique identifier"
                    },
                    "fields": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Fields to retrieve: name, email, orders, balance"
                    }
                },
                "required": ["customer_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_support_ticket",
            "description": "Create a support ticket in the helpdesk system",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "subject": {"type": "string"},
                    "priority": {"type": "string", "enum": ["low", "medium", "high"]},
                    "description": {"type": "string"}
                },
                "required": ["customer_id", "subject", "description"]
            }
        }
    }
]

# Make API call with function definitions
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a customer service assistant. Use the available tools to help customers."},
        {"role": "user", "content": "Can you check the order history for customer ABC123 and create a ticket about their delayed shipment?"}
    ],
    tools=tools,
    tool_choice="auto"  # Let model decide when to use tools
)

# Handle function calls
message = response.choices[0].message

if message.tool_calls:
    for tool_call in message.tool_calls:
        function_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)

        # Execute the actual function
        if function_name == "get_customer_info":
            result = get_customer_info(**arguments)  # Your implementation
        elif function_name == "create_support_ticket":
            result = create_support_ticket(**arguments)  # Your implementation

        # Continue conversation with function result
        # ... (send result back to model)

Creating Embeddings (Python)

from openai import OpenAI

client = OpenAI()

# Single text embedding
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="How do I automate invoice processing with AI?"
)

embedding = response.data[0].embedding  # 1536-dimensional vector
print(f"Embedding dimension: {len(embedding)}")

# Batch embeddings for efficiency
texts = [
    "Invoice processing automation",
    "Customer support chatbot",
    "Sales pipeline analysis"
]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts
)

embeddings = [item.embedding for item in response.data]

RAG Implementation Pattern

import numpy as np
from openai import OpenAI

client = OpenAI()

# 1. INDEXING: Embed and store your documents
def embed_documents(documents):
    """Embed a list of documents and return vectors."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=documents
    )
    return [item.embedding for item in response.data]

# 2. RETRIEVAL: Find relevant documents for a query
def find_relevant_docs(query, doc_embeddings, documents, top_k=3):
    """Find most relevant documents using cosine similarity."""
    # Embed the query
    query_response = client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    query_embedding = query_response.data[0].embedding

    # Calculate similarities
    similarities = []
    for i, doc_emb in enumerate(doc_embeddings):
        similarity = np.dot(query_embedding, doc_emb) / (
            np.linalg.norm(query_embedding) * np.linalg.norm(doc_emb)
        )
        similarities.append((similarity, documents[i]))

    # Return top-k most similar
    similarities.sort(reverse=True)
    return [doc for _, doc in similarities[:top_k]]

# 3. GENERATION: Answer using retrieved context
def answer_with_context(question, context_docs):
    """Generate answer using retrieved documents as context."""
    context = "\n\n".join(context_docs)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"""Answer questions based on the following context.
If the context doesn't contain relevant information, say so.

Context:
{context}"""
            },
            {"role": "user", "content": question}
        ]
    )

    return response.choices[0].message.content

# Usage
documents = ["...your documents..."]
doc_embeddings = embed_documents(documents)

question = "How do I set up automated invoicing?"
relevant = find_relevant_docs(question, doc_embeddings, documents)
answer = answer_with_context(question, relevant)
print(answer)

Streaming in Python

from openai import OpenAI

client = OpenAI()

# Stream response
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Explain AI in 200 words"}],
    stream=True
)

# Print tokens as they arrive
for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()  # Newline at end

Streaming in TypeScript (Next.js)

// app/api/chat/route.ts
import OpenAI from 'openai';

const openai = new OpenAI();

export async function POST(req: Request) {
  const { messages } = await req.json();

  const stream = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    messages,
    stream: true,
  });

  // Return streaming response
  const encoder = new TextEncoder();
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content || '';
        controller.enqueue(encoder.encode(text));
      }
      controller.close();
    },
  });

  return new Response(readable, {
    headers: { 'Content-Type': 'text/plain; charset=utf-8' },
  });
}

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    tools=tools,
    stream=True
)

tool_calls = []
current_tool_call = None

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

    # Handle tool calls in stream
    if delta.tool_calls:
        for tc_chunk in delta.tool_calls:
            if tc_chunk.index is not None:
                if tc_chunk.id:  # New tool call
                    current_tool_call = {
                        "id": tc_chunk.id,
                        "type": "function",
                        "function": {"name": "", "arguments": ""}
                    }
                    tool_calls.append(current_tool_call)

                if tc_chunk.function:
                    if tc_chunk.function.name:
                        current_tool_call["function"]["name"] = tc_chunk.function.name
                    if tc_chunk.function.arguments:
                        current_tool_call["function"]["arguments"] += tc_chunk.function.arguments

    # Handle regular content
    elif delta.content:
        print(delta.content, end="", flush=True)

Track Usage Per Request

# Track tokens and estimate costs
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
)

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

# Estimate cost (GPT-4o-mini pricing)
input_cost = usage.prompt_tokens * 0.00000015  # $0.15/1M
output_cost = usage.completion_tokens * 0.0000006  # $0.60/1M
total_cost = input_cost + output_cost

print(f"Estimated cost: ${total_cost:.6f} USD")

Python with Tenacity Retry

from openai import OpenAI, RateLimitError, APIError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

client = OpenAI()

@retry(
    retry=retry_if_exception_type((RateLimitError, APIError)),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    stop=stop_after_attempt(5)
)
def call_openai_with_retry(messages, model="gpt-4o-mini"):
    """Make OpenAI API call with automatic retry on transient errors."""
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            timeout=30  # Set reasonable timeout
        )
        return response.choices[0].message.content

    except RateLimitError:
        # Log rate limit hit, will retry automatically
        print("Rate limited, retrying...")
        raise

    except APIError as e:
        if e.status_code >= 500:
            # Server error, retry
            print(f"Server error {e.status_code}, retrying...")
            raise
        else:
            # Client error, don't retry
            raise ValueError(f"API error: {e}")

    except Exception as e:
        # Unexpected error
        print(f"Unexpected error: {e}")
        raise

# Usage
try:
    result = call_openai_with_retry([
        {"role": "user", "content": "Hello!"}
    ])
    print(result)
except Exception as e:
    print(f"Failed after retries: {e}")
    # Fallback logic here

Truncate to Fit Context Window

import tiktoken

def count_tokens(text, model="gpt-4o-mini"):
    """Count tokens in text for a given model."""
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

def truncate_to_fit(messages, max_tokens=100000, model="gpt-4o-mini"):
    """Truncate conversation to fit within token limit."""
    encoding = tiktoken.encoding_for_model(model)

    # Keep system message, truncate from oldest user/assistant
    total_tokens = 0
    result = []

    # Always include system message first
    for msg in messages:
        if msg["role"] == "system":
            result.append(msg)
            total_tokens += count_tokens(msg["content"], model)
            break

    # Add messages from most recent, stop when limit reached
    for msg in reversed(messages):
        if msg["role"] == "system":
            continue

        msg_tokens = count_tokens(msg["content"], model)
        if total_tokens + msg_tokens > max_tokens:
            break

        result.insert(1, msg)  # Insert after system message
        total_tokens += msg_tokens

    return result