Openai Api

OpenAI API

Comprehensive guide to using the OpenAI API in production: Completions with structured outputs, Embeddings, function calling, fine-tuning, Batch API, streaming, Assistants API, cost optimization, error handling with retries, safety guardrails, and production architecture patterns.

When to Use

• Building applications that call LLMs for chat completion, text generation, or structured data extraction
• Generating and using embeddings for semantic search, clustering, or similarity matching
• Integrating function/tool calling for LLM-powered agent workflows
• Fine-tuning models for domain-specific tasks with higher accuracy
• Processing large document batches asynchronously with the Batch API
• Building conversational agents with the Assistants API (threads, code interpreter, file search)
• Optimizing API costs, handling rate limits, and implementing production-grade error handling
• Adding safety guardrails with moderation and output validation

Core Concepts

1. Completions API

Chat Completions with System Prompts:

from openai import OpenAI
import os

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def chat_completion(messages: list[dict], model: str = "gpt-4o-mini") -> str:
    """Generate a chat completion with proper error handling."""
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0.3,
        max_tokens=1024,
        top_p=0.95,
        frequency_penalty=0.0,
        presence_penalty=0.0,
    )
    return response.choices[0].message.content

Structured Outputs via JSON Mode:

import json
from pydantic import BaseModel, Field
from typing import Literal

class ExtractionResult(BaseModel):
    """Structured extraction from unstructured text."""
    entities: list[dict[str, str]] = Field(description="Named entities found")
    sentiment: Literal["positive", "negative", "neutral"] = Field(description="Overall sentiment")
    summary: str = Field(description="One-sentence summary")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence in extraction")

def extract_structured(text: str) -> ExtractionResult:
    """Extract structured data using JSON mode with schema enforcement."""
    schema = ExtractionResult.model_json_schema()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Extract structured data from the text. Return valid JSON only."},
            {"role": "user", "content": text},
        ],
        response_format={"type": "json_schema", "json_schema": {"name": "extraction", "schema": schema}},
        temperature=0.0,  # Deterministic for extraction
    )

    result = json.loads(response.choices[0].message.content)
    return ExtractionResult(**result)

Tool Use / Function Calling:

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "Search the internal knowledge base for relevant documents",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "top_k": {"type": "integer", "description": "Number of results", "default": 5},
                    "category": {"type": "string", "enum": ["engineering", "support", "sales"]},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_user_profile",
            "description": "Retrieve user profile by ID",
            "parameters": {
                "type": "object",
                "properties": {
                    "user_id": {"type": "string", "description": "User ID"},
                },
                "required": ["user_id"],
            },
        },
    },
]

def chat_with_tools(user_message: str) -> str:
    """Chat with tool calling — LLM decides which tools to call."""
    messages = [{"role": "user", "content": user_message}]

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

    assistant_msg = response.choices[0].message

    # Execute tool calls
    if assistant_msg.tool_calls:
        messages.append(assistant_msg)

        for tool_call in assistant_msg.tool_calls:
            fn_name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)

            if fn_name == "search_knowledge_base":
                result = search_knowledge_base(args["query"], args.get("top_k", 5))
            elif fn_name == "get_user_profile":
                result = get_user_profile(args["user_id"])
            else:
                result = {"error": f"Unknown function: {fn_name}"}

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result),
            })

        # Get final response with tool results
        response = client.chat.completions.create(model="gpt-4o", messages=messages)

    return response.choices[0].message.content

Parallel Function Calling — multiple tools in a single turn:

# LLM can call multiple tools simultaneously
# The response will contain multiple tool_calls in assistant_msg.tool_calls
# Execute all tool calls in parallel for better latency

import asyncio

async def execute_tool_calls_parallel(tool_calls):
    """Execute multiple tool calls concurrently."""
    async def execute_one(tc):
        args = json.loads(tc.function.arguments)
        result = await TOOL_FUNCTIONS[tc.function.name](**args)
        return {"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)}

    return await asyncio.gather(*(execute_one(tc) for tc in tool_calls))

2. Embeddings API

def generate_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """Generate embeddings for a batch of texts."""
    response = client.embeddings.create(
        input=texts,
        model=model,
        dimensions=1536,  # Optional: reduce dimensions for cost savings
        encoding_format="float",
    )
    return [e.embedding for e in response.data]

def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Compute cosine similarity between two embedding vectors."""
    import numpy as np
    a_np = np.array(a)
    b_np = np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

# Clustering with embeddings
from sklearn.cluster import KMeans

def cluster_documents(texts: list[str], n_clusters: int = 5) -> dict:
    """Cluster documents by embedding similarity."""
    embeddings = generate_embeddings(texts)
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    labels = kmeans.fit_predict(embeddings)

    clusters = {}
    for i, label in enumerate(labels):
        clusters.setdefault(int(label), []).append(texts[i])
    return clusters

3. Fine-Tuning

Dataset Preparation:

// training.jsonl — one JSON object per line
{"messages": [{"role": "system", "content": "You are a helpful legal assistant."}, {"role": "user", "content": "What is the statute of limitations for breach of contract in California?"}, {"role": "assistant", "content": "In California, the statute of limitations for written contracts is 4 years from the date of breach (CCP § 337). For oral contracts, it is 2 years (CCP § 339)."}]}
{"messages": [{"role": "system", "content": "You are a helpful legal assistant."}, {"role": "user", "content": "Can I sue for emotional distress in a wrongful termination case?"}, {"role": "assistant", "content": "Yes, emotional distress damages may be available in wrongful termination claims, particularly under FEHA. However, the availability depends on the specific circumstances and legal theory pursued."}]}

Training Pipeline:

# Step 1: Validate dataset
import tiktoken

enc = tiktoken.get_encoding("cl100k_base")
total_tokens = 0
with open("training.jsonl") as f:
    for line in f:
        messages = json.loads(line)["messages"]
        tokens = sum(len(enc.encode(m["content"])) for m in messages)
        total_tokens += tokens

print(f"Total tokens: {total_tokens}, Estimated cost: ${total_tokens / 1000 * 0.008:.2f}")

# Step 2: Upload and start training
with open("training.jsonl", "rb") as f:
    file_resp = client.files.create(file=f, purpose="fine-tune")

job = client.fine_tuning.jobs.create(
    training_file=file_resp.id,
    model="gpt-4o-mini-2024-07-18",
    suffix="legal-assistant-v1",
    hyperparameters={"n_epochs": 3, "learning_rate_multiplier": 0.1},
)

# Step 3: Monitor training
while True:
    job = client.fine_tuning.jobs.retrieve(job.id)
    print(f"Status: {job.status}")
    if job.status in ["succeeded", "failed"]:
        break
    time.sleep(30)

fine_tuned_model = job.fine_tuned_model

4. Batch API

# Async batch processing — cheaper, slower
import json

def submit_batch_request(requests: list[dict], output_file: str = "batch_results.jsonl"):
    """Submit a batch of API requests for async processing."""
    # Write requests to JSONL
    batch_input_path = "batch_input.jsonl"
    with open(batch_input_path, "w") as f:
        for i, req in enumerate(requests):
            f.write(json.dumps({
                "custom_id": f"req-{i}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": req,
            }) + "\n")

    # Upload file
    with open(batch_input_path, "rb") as f:
        batch_file = client.files.create(file=f, purpose="batch")

    # Create batch job
    batch_job = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
    )

    print(f"Batch job created: {batch_job.id}")
    return batch_job.id

def retrieve_batch_results(batch_id: str):
    """Retrieve completed batch results."""
    job = client.batches.retrieve(batch_id)
    if job.status != "completed":
        raise ValueError(f"Batch not completed: {job.status}")

    results_file = client.files.retrieve_content(job.output_file_id)
    results = []
    for line in results_file.text.strip().split("\n"):
        results.append(json.loads(line))
    return results

5. Streaming

def stream_completion(messages: list[dict], model: str = "gpt-4o"):
    """Stream tokens incrementally for responsive UX."""
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        stream_options={"include_usage": True},
    )

    full_response = ""
    total_tokens = 0

    for chunk in stream:
        # Accumulate content
        if chunk.choices and chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            full_response += content
            yield content  # Yield to caller (e.g., SSE response)

        # Track token usage (only in final chunk)
        if chunk.usage:
            total_tokens = chunk.usage.total_tokens

    return full_response, total_tokens

# FastAPI SSE endpoint
from fastapi.responses import StreamingResponse

@app.post("/v1/chat/stream")
async def stream_chat(request: ChatRequest):
    async def event_generator():
        messages = [{"role": "user", "content": request.query}]
        for token in stream_completion(messages):
            yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

6. Assistants API

# Create an assistant with file search and code interpreter
assistant = client.beta.assistants.create(
    name="Research Assistant",
    model="gpt-4o",
    instructions="You are a research assistant. Use file search to find relevant information in the knowledge base.",
    tools=[
        {"type": "file_search"},
        {"type": "code_interpreter"},
    ],
)

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

# Add a message
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Analyze the sales data from Q3 and summarize the key trends.",
)

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

# Poll for completion
while run.status != "completed":
    run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
    time.sleep(1)

# Get response messages
messages = client.beta.threads.messages.list(thread_id=thread.id)
for msg in messages.data:
    print(f"{msg.role}: {msg.content[0].text.value}")

7. Cost Optimization

| Strategy | Savings | Example | |----------|---------|---------| | Model selection | 10-100x | Use gpt-4o-mini for simple tasks, gpt-4o for complex reasoning | | Prompt compression | 20-40% | Remove redundant context, use concise instructions | | Semantic caching | 10-30% | Cache responses for similar queries (see mlops-rag skill) | | Batch API | 50% | Async processing for non-urgent requests | | Dimensionality reduction | 30-50% | Use 256-dim embeddings instead of 1536 when acceptable | | Fine-tuning | 20-40% | Smaller model fine-tuned can match larger model performance |

def smart_model_selection(task: str, complexity: str) -> str:
    """Select model based on task complexity."""
    if complexity == "high":
        return "gpt-4o"
    elif complexity == "medium":
        return "gpt-4o-mini"
    else:
        return "gpt-4o-mini"  # Default: cheapest capable model

8. Error Handling

import time
from openai import (
    OpenAI,
    RateLimitError,
    APIConnectionError,
    APIStatusError,
    APITimeoutError,
)

class CircuitBreaker:
    """Circuit breaker for API calls — fail fast when downstream is unhealthy."""
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise RuntimeError("Circuit breaker is open")

        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
            self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise

client = OpenAI()
circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)

def call_openai_with_retry(messages: list[dict], model: str, max_retries: int = 5) -> str:
    """Call OpenAI with exponential backoff, fallback models, and circuit breaker."""
    models_to_try = [model, "gpt-4o-mini", "gpt-3.5-turbo"]
    base_delay = 1.0

    for attempt in range(max_retries):
        for model_choice in models_to_try:
            try:
                return circuit_breaker.call(
                    client.chat.completions.create,
                    model=model_choice,
                    messages=messages,
                    timeout=30.0,
                ).choices[0].message.content
            except RateLimitError:
                delay = base_delay * (2 ** attempt) + (hash(str(time.time())) % 1)  # jitter
                time.sleep(delay)
                continue

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [noah-sheldon](https://github.com/noah-sheldon)
- **Source:** [noah-sheldon/ai-dev-kit](https://github.com/noah-sheldon/ai-dev-kit)
- **License:** MIT
- **Homepage:** https://noahsheldon.dev

Install and usage instructions live in the source repository linked above.