Lab 01: OpenAI SDK Fundamentals

Goal

Master the OpenAI SDK: completions, streaming, structured output, function/tool calling, embeddings, and cost tracking.

Prerequisites

  • Python 3.11+
  • pip install openai tiktoken
  • OPENAI_API_KEY in environment

Lab Structure

  1. Basic completion
  2. Streaming
  3. Structured JSON output
  4. Tool calling
  5. Embeddings
  6. Cost tracking

Exercise 1: Basic Completion

# lab01/01_basic.py
from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a concise technical assistant."},
        {"role": "user", "content": "What is the KV cache in LLMs? Explain in 3 sentences."}
    ],
    max_tokens=200,
    temperature=0
)

print(response.choices[0].message.content)
print(f"\nTokens used: {response.usage.prompt_tokens} in / {response.usage.completion_tokens} out")

Task: Run this. Now change the system prompt to make the model respond in bullet points. Notice how the output format changes.


Exercise 2: Streaming

# lab01/02_streaming.py
from openai import OpenAI

client = OpenAI()

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "user", "content": "Count from 1 to 10 slowly, one number per line."}
    ],
    stream=True
)

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

print()  # Final newline

Task: Measure the time to first token vs. time to complete. Use time.time() around the first chunk received vs. the full stream.


Exercise 3: Structured Output

# lab01/03_structured.py
from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class ProductInfo(BaseModel):
    product_name: str
    price_usd: float
    category: str
    in_stock: bool
    key_features: list[str]

response = client.beta.chat.completions.parse(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "Extract product info from: The Sony WH-1000XM5 headphones feature 30-hour battery life, industry-leading noise cancellation, and are priced at $349. Currently available."
    }],
    response_format=ProductInfo
)

product = response.choices[0].message.parsed
print(product)
print(f"Price: ${product.price_usd}")
print(f"Features: {', '.join(product.key_features)}")

Task: Add more fields to ProductInfo. What happens if the source text doesn't mention a required field?


Exercise 4: Tool Calling

# lab01/04_tools.py
from openai import OpenAI
import json, math

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Evaluate a mathematical expression. Safe — only arithmetic.",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "A mathematical expression like '2 * (3 + 4)' or 'sqrt(16)'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

def safe_calculate(expression: str) -> float:
    """Evaluate a math expression safely."""
    allowed = {
        'sqrt': math.sqrt, 'sin': math.sin, 'cos': math.cos,
        'pi': math.pi, 'e': math.e, 'abs': abs, 'round': round
    }
    return eval(expression, {"__builtins__": {}}, allowed)

messages = [
    {"role": "user", "content": "What is the square root of 144, plus 7 squared?"}
]

# First call — model decides to use a tool
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

message = response.choices[0].message

if message.tool_calls:
    messages.append(message)
    
    for tool_call in message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = safe_calculate(args["expression"])
        
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": str(result)
        })
    
    # Second call — model has the tool result
    final_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools
    )
    
    print(final_response.choices[0].message.content)
else:
    print(message.content)

Task: Add a second tool: get_current_date() that returns today's date. Ask "What's the date today plus 7 days?"


Exercise 5: Embeddings

# lab01/05_embeddings.py
from openai import OpenAI
import numpy as np

client = OpenAI()

texts = [
    "The transformer architecture uses self-attention",
    "Attention mechanisms allow the model to focus on relevant tokens",
    "Paris is the capital of France",
    "The Eiffel Tower is in Paris",
    "Python is a programming language"
]

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

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

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

# Find most similar pairs
print("Similarity scores:")
for i in range(len(texts)):
    for j in range(i+1, len(texts)):
        sim = cosine_similarity(embeddings[i], embeddings[j])
        print(f"  {sim:.3f}: '{texts[i][:40]}' vs '{texts[j][:40]}'")

Task: Which pairs have the highest similarity? Does it match your intuition? Try adding your own texts and testing semantic similarity.


Exercise 6: Cost Tracker

# lab01/06_cost_tracker.py
from openai import OpenAI
from dataclasses import dataclass, field

PRICING = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
}

@dataclass
class CostTracker:
    session_costs: list[float] = field(default_factory=list)
    
    def track(self, model: str, usage) -> float:
        prices = PRICING.get(model, {"input": 1.0, "output": 4.0})
        cost = (
            usage.prompt_tokens / 1_000_000 * prices["input"] +
            usage.completion_tokens / 1_000_000 * prices["output"]
        )
        self.session_costs.append(cost)
        return cost
    
    @property
    def total(self) -> float:
        return sum(self.session_costs)

tracker = CostTracker()
client = OpenAI()

questions = [
    "What is attention in transformers?",
    "What is the KV cache?",
    "What is quantization?",
]

for q in questions:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": q}],
        max_tokens=100
    )
    cost = tracker.track("gpt-4o-mini", response.usage)
    print(f"Q: {q[:40]}... | Cost: ${cost:.6f}")

print(f"\nTotal session cost: ${tracker.total:.6f}")
print(f"Estimated monthly at 10k/day: ${tracker.total / len(questions) * 10_000 * 30:.2f}")

Deliverables

  • All 6 exercises run successfully
  • Exercise 2: TTFT and full latency measured
  • Exercise 3: ProductInfo schema extended with one new field
  • Exercise 4: Second tool (get_current_date) added and tested
  • Exercise 5: Semantic similarity analysis complete — note which pairs are most/least similar
  • Exercise 6: Cost estimate for your use case calculated

Reflection Questions

  1. What is the minimum information needed in a chat message?
  2. What's the difference between stream=True and non-streaming, from a latency perspective?
  3. Why does structured output (.parse()) use a different API endpoint?
  4. What would happen if safe_calculate used Python's built-in eval() without restrictions?
  5. Why does text-embedding-3-small output 1536 floats per text?