LLM Application Development Practical Guide

The first time you touch an LLM API, "Hello World" comes together in minutes — but building an actual production service means walking past a long list of traps. Implementing streaming, preventing a runaway bill, handling rate limits, minimizing hallucinations... In this guide, we cover everything involved in real-world LLM application development on the OpenAI, Anthropic Claude, and Google Gemini APIs.

1. LLM API Ecosystem Overview

Comparing the Major API Providers

Provider	Key models	Strengths	Weaknesses
OpenAI	GPT-4o, o1, o3	Ecosystem, function calling	Cost
Anthropic	Claude 3.5, Claude 3.7	Long context, safety	Relatively high cost
Google	Gemini 1.5 Pro/Flash	Multimodal, 1M context	Korean
Mistral AI	Mistral Large/NeMo	Cost efficiency	Ecosystem
Cohere	Command R+	Enterprise, RAG	Limited features
Together AI	Open-source model hosting	Access to open-source models	Reliability

API Cost Comparison (as of 2025, per 1M tokens)

# API cost comparison (input/output tokens, USD)
api_costs = {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
    "claude-3-haiku": {"input": 0.25, "output": 1.25},
    "gemini-1.5-pro": {"input": 1.25, "output": 5.00},
    "gemini-1.5-flash": {"input": 0.075, "output": 0.30},
    "mistral-large": {"input": 2.00, "output": 6.00},
    "mistral-small": {"input": 0.20, "output": 0.60},
}

def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """Estimate the API cost"""
    if model not in api_costs:
        return 0.0
    costs = api_costs[model]
    input_cost = (input_tokens / 1_000_000) * costs["input"]
    output_cost = (output_tokens / 1_000_000) * costs["output"]
    return input_cost + output_cost

# Example: 10,000 requests a day, averaging 200 input / 500 output tokens
daily_cost_gpt4o = estimate_cost("gpt-4o", 200 * 10000, 500 * 10000)
daily_cost_mini = estimate_cost("gpt-4o-mini", 200 * 10000, 500 * 10000)
print(f"GPT-4o daily cost: ${daily_cost_gpt4o:.2f}")
print(f"GPT-4o-mini daily cost: ${daily_cost_mini:.2f}")

2. The Complete OpenAI API Guide

Basic Setup

pip install openai

from openai import OpenAI
import os

# Initialize the client
client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
    # Defaults, but customizable
    max_retries=3,
    timeout=60.0,
)

Basic Chat Completions

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": "You are a friendly and accurate Korean-language AI assistant."
        },
        {
            "role": "user",
            "content": "Tell me the pros and cons of Python."
        }
    ],
    temperature=0.7,       # 0: deterministic, 1: creative
    max_tokens=1024,       # Maximum output tokens
    top_p=0.9,             # nucleus sampling
    frequency_penalty=0.0, # Penalty for repeated words
    presence_penalty=0.0,  # Penalty for introducing new topics
)

print(response.choices[0].message.content)
print(f"Tokens used - input: {response.usage.prompt_tokens}, output: {response.usage.completion_tokens}")

Streaming Responses

def stream_chat(messages: list, model: str = "gpt-4o") -> str:
    """Handle a streaming response"""
    full_response = ""

    with client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
    ) as stream:
        for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                print(delta.content, end="", flush=True)
                full_response += delta.content

    print()  # newline
    return full_response

# Usage example
messages = [{"role": "user", "content": "Explain the different types of machine learning."}]
response = stream_chat(messages)

Function Calling (Using Tools)

import json
from openai import OpenAI

client = OpenAI()

# Tool definitions
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a given city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name (e.g. Seoul, Busan)"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for the latest information.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query"
                    },
                    "num_results": {
                        "type": "integer",
                        "description": "Number of results to return",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        }
    }
]

def get_weather(city: str, unit: str = "celsius") -> dict:
    """Call a weather API (real implementation required)"""
    # In practice, call a real weather API
    return {
        "city": city,
        "temperature": 22,
        "unit": unit,
        "condition": "Clear",
        "humidity": 45
    }

def search_web(query: str, num_results: int = 5) -> list:
    """Web search (real implementation required)"""
    return [{"title": f"Search result {i}", "url": f"https://example.com/{i}"} for i in range(num_results)]

def run_conversation(user_message: str) -> str:
    messages = [
        {"role": "system", "content": "You are an AI assistant that helps with weather and information lookup."},
        {"role": "user", "content": user_message}
    ]

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

        message = response.choices[0].message
        messages.append(message)

        # If there is no tool call, return the response
        if not message.tool_calls:
            return message.content

        # Handle the tool calls
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            if function_name == "get_weather":
                result = get_weather(**function_args)
            elif function_name == "search_web":
                result = search_web(**function_args)
            else:
                result = {"error": "Unknown function"}

            # Append the tool result
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

# Usage example
print(run_conversation("Compare the current weather in Seoul and Busan."))

Structured Outputs

from pydantic import BaseModel
from typing import List, Optional

class ProductReview(BaseModel):
    product_name: str
    overall_rating: int  # 1-5
    pros: List[str]
    cons: List[str]
    summary: str
    recommendation: bool

def analyze_review(review_text: str) -> ProductReview:
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "Analyze the product review and extract it as structured information."
            },
            {
                "role": "user",
                "content": review_text
            }
        ],
        response_format=ProductReview,
    )

    return response.choices[0].message.parsed

# Usage example
review = """
I have been using the Samsung Galaxy S24 for 2 months.
The camera quality is truly excellent, and the battery easily lasts a full day.
The AI features are practical too, and I use them a lot.
The downsides are that the price is a bit high and it sometimes runs hot.
Overall it is a satisfying flagship smartphone.
"""

result = analyze_review(review)
print(f"Product: {result.product_name}")
print(f"Rating: {result.overall_rating}/5")
print(f"Pros: {', '.join(result.pros)}")
print(f"Cons: {', '.join(result.cons)}")
print(f"Recommended: {'Yes' if result.recommendation else 'No'}")

Vision API

import base64
from pathlib import Path

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

def analyze_image(image_path: str, question: str = "Describe this image in detail.") -> str:
    # Analyze a local image
    base64_image = encode_image(image_path)
    ext = Path(image_path).suffix.lower().replace('.', '')
    media_type = f"image/{ext if ext in ['jpeg', 'png', 'gif', 'webp'] else 'jpeg'}"

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{media_type};base64,{base64_image}",
                            "detail": "high"  # "low" or "high"
                        }
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ],
        max_tokens=1024,
    )

    return response.choices[0].message.content

# Analyze an image from a URL
def analyze_image_url(url: str, question: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": url}},
                    {"type": "text", "text": question}
                ]
            }
        ],
    )
    return response.choices[0].message.content

Embeddings API

def get_embeddings(texts: list, model: str = "text-embedding-3-small") -> list:
    response = client.embeddings.create(
        input=texts,
        model=model,
    )
    return [item.embedding for item in response.data]

# Semantic search
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticSearch:
    def __init__(self):
        self.documents = []
        self.embeddings = None

    def index(self, documents: list):
        self.documents = documents
        self.embeddings = np.array(get_embeddings(documents))

    def search(self, query: str, top_k: int = 3) -> list:
        query_emb = np.array(get_embeddings([query]))
        sims = cosine_similarity(query_emb, self.embeddings)[0]
        top_idx = np.argsort(sims)[::-1][:top_k]
        return [(self.documents[i], float(sims[i])) for i in top_idx]

3. Anthropic Claude API

Basic Setup

pip install anthropic

import anthropic

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
)

Messages API

def claude_chat(
    user_message: str,
    system: str = "You are a helpful assistant.",
    model: str = "claude-3-5-sonnet-20241022",
    max_tokens: int = 1024,
) -> str:
    message = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system,
        messages=[
            {"role": "user", "content": user_message}
        ]
    )
    return message.content[0].text

# Multi-turn conversation
def claude_conversation(messages: list, system: str = None) -> str:
    kwargs = {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 2048,
        "messages": messages
    }
    if system:
        kwargs["system"] = system

    message = client.messages.create(**kwargs)
    return message.content[0].text

Streaming

def claude_stream(user_message: str, system: str = None) -> str:
    full_text = ""

    kwargs = {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 2048,
        "messages": [{"role": "user", "content": user_message}]
    }
    if system:
        kwargs["system"] = system

    with client.messages.stream(**kwargs) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_text += text

    print()
    return full_text

System Prompt Design Best Practices

# Components of a good system prompt
system_prompt_template = """
You are [role].

## Primary responsibilities
- [Role 1]
- [Role 2]

## Behavioral guidelines
1. [Guideline 1]
2. [Guideline 2]

## Constraints
- [Constraint 1]
- [Constraint 2]

## Response format
- [Format guideline]
"""

# Korean customer-support bot example
cs_system_prompt = """
You are the customer-support AI agent for "Shopping Mall", a Korean e-commerce platform.

## Primary responsibilities
- Look up and explain order status
- Explain refund and exchange policies
- Handle shipping-related inquiries
- Provide product information

## Behavioral guidelines
1. Always use a polite and friendly tone.
2. If you do not know the answer to a question, admit it honestly and offer to connect the customer with a human agent.
3. Never ask for personal data (address, card number, and the like).
4. Answer questions that compare competitors neutrally.

## Constraints
- Do not disclose internal company information or employees' personal information.
- Do not give legal advice or medical information.
- Do not promise unconfirmed promotions or discounts.

## Response format
- Answer concisely and clearly.
- When it helps, use a numbered list to walk through the steps.
- At the end, ask whether any further help is needed.
"""

Extended Thinking (Claude 3.7+)

def claude_think(question: str, budget_tokens: int = 8000) -> dict:
    """Solve a complex reasoning problem with Extended Thinking"""
    response = client.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=16000,
        thinking={
            "type": "enabled",
            "budget_tokens": budget_tokens  # Maximum tokens to spend on internal reasoning
        },
        messages=[{"role": "user", "content": question}]
    )

    result = {"thinking": "", "answer": ""}

    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking
        elif block.type == "text":
            result["answer"] = block.text

    return result

# A complex math problem
problem = """
A company's annual revenue grows 15% every year.
If current revenue is 10 billion won, calculate whether revenue will exceed 20 billion won in 5 years, and
also compute the exact revenue figure 5 years from now.
"""

result = claude_think(problem)
print("Reasoning trace (excerpt):", result["thinking"][:300])
print("\nFinal answer:", result["answer"])

Tool Use (Claude)

import anthropic
import json

client = anthropic.Anthropic()

# Tool definitions
tools = [
    {
        "name": "calculate",
        "description": "Perform a mathematical calculation.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "Expression to evaluate (e.g. 2 + 2, 15 * 30)"
                }
            },
            "required": ["expression"]
        }
    }
]

def process_tool_call(tool_name: str, tool_input: dict) -> str:
    if tool_name == "calculate":
        try:
            # In real production, use a safe calculator instead of eval
            result = eval(tool_input["expression"])
            return str(result)
        except Exception as e:
            return f"Error: {str(e)}"
    return "Unknown tool"

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

    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

        # If there is no tool call, return the final response
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # Handle the tool calls
        tool_results = []
        for content_block in response.content:
            if content_block.type == "tool_use":
                result = process_tool_call(content_block.name, content_block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": content_block.id,
                    "content": result
                })

        # Update the message history
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

Prompt Caching (Cost Savings)

# Cache long system prompts or documents to cut cost
# After the first call, an identical prompt is 90% cheaper
long_document = "..." * 1000  # A long document

def analyze_document_with_cache(question: str) -> str:
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": "Analyze the following document and answer the question:",
                "cache_control": {"type": "ephemeral"}  # Cache directive
            },
            {
                "type": "text",
                "text": long_document,
                "cache_control": {"type": "ephemeral"}
            }
        ],
        messages=[{"role": "user", "content": question}]
    )

    # Check whether the cache was used
    usage = response.usage
    print(f"Input tokens: {usage.input_tokens}")
    print(f"Cache creation tokens: {getattr(usage, 'cache_creation_input_tokens', 0)}")
    print(f"Cache read tokens: {getattr(usage, 'cache_read_input_tokens', 0)}")

    return response.content[0].text

4. Google Gemini API

Basic Setup

pip install google-generativeai

import google.generativeai as genai
import os

genai.configure(api_key=os.environ.get("GOOGLE_API_KEY"))

# Initialize the model
model = genai.GenerativeModel(
    model_name="gemini-1.5-flash",
    generation_config=genai.GenerationConfig(
        temperature=0.7,
        top_p=0.9,
        top_k=40,
        max_output_tokens=2048,
    ),
    system_instruction="You are a helpful Korean-language AI assistant."
)

Basic Usage

# Text generation
response = model.generate_content("Explain asynchronous programming in Python.")
print(response.text)

# Chat session
chat = model.start_chat(history=[])

response1 = chat.send_message("Hello! I would like to learn about machine learning.")
print(response1.text)

response2 = chat.send_message("Then how is it different from deep learning?")
print(response2.text)

# Inspect the conversation history
for turn in chat.history:
    print(f"{turn.role}: {turn.parts[0].text[:100]}...")

Multimodal (Text + Image)

import PIL.Image
import requests
from io import BytesIO

model_vision = genai.GenerativeModel("gemini-1.5-pro")

# Local image
image = PIL.Image.open("chart.png")
response = model_vision.generate_content([
    image,
    "Analyze this chart and explain the main trends in Korean."
])
print(response.text)

# Image from a URL
response_url = requests.get("https://example.com/image.jpg")
image_from_url = PIL.Image.open(BytesIO(response_url.content))
response = model_vision.generate_content([
    image_from_url,
    "Extract the text from this image."
])

# PDF handling (a Gemini 1.5 strength)
with open("report.pdf", "rb") as f:
    pdf_data = f.read()

response = model_vision.generate_content([
    {"mime_type": "application/pdf", "data": pdf_data},
    "Summarize the key content of this PDF document."
])
print(response.text)

Making Use of the 1M Context Window

# Gemini 1.5 Pro: up to a 1,000,000-token context
# Can analyze an entire codebase in one pass

def analyze_codebase(code_files: dict) -> str:
    """Analyze a codebase made up of multiple files"""
    model_pro = genai.GenerativeModel("gemini-1.5-pro")

    content_parts = ["Analyze the following codebase:\n\n"]
    for filename, code in code_files.items():
        content_parts.append(f"File: {filename}\n```\n{code}\n```\n\n")
    content_parts.append("Tell me about the architecture, potential bugs, and improvements.")

    response = model_pro.generate_content(content_parts)
    return response.text

# Long-document summarization
def summarize_long_document(document: str) -> str:
    model_flash = genai.GenerativeModel("gemini-1.5-flash")
    # The Flash model also supports 1M tokens, and is faster and cheaper
    response = model_flash.generate_content(
        f"Summarize the following document into 5 key points:\n\n{document}"
    )
    return response.text

Structured Output (JSON Mode)

import json

def extract_structured_data(text: str, schema: dict) -> dict:
    model = genai.GenerativeModel(
        "gemini-1.5-flash",
        generation_config=genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema=schema,
        )
    )

    response = model.generate_content(text)
    return json.loads(response.text)

# Example: extracting information from a resume
resume_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "email": {"type": "string"},
        "skills": {"type": "array", "items": {"type": "string"}},
        "experience_years": {"type": "number"},
    }
}

resume_text = """
Hong Gildong
Email: hong@example.com
Experience: Python 5 years, JavaScript 3 years, Docker 2 years
Total experience: 6 years
"""

result = extract_structured_data(
    f"Extract the information from this resume:\n{resume_text}",
    resume_schema
)
print(result)

5. Building a Streaming Chatbot

FastAPI Backend

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
import json
import time
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    message: str
    conversation_id: str = None
    model: str = "gpt-4o-mini"

# Store the conversation history (use Redis or a DB in practice)
conversations = {}

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    conv_id = request.conversation_id or str(time.time())

    # Fetch the conversation history
    history = conversations.get(conv_id, [])
    history.append({"role": "user", "content": request.message})

    async def generate():
        full_response = ""

        try:
            stream = await client.chat.completions.create(
                model=request.model,
                messages=[
                    {"role": "system", "content": "You are a helpful AI assistant."},
                    *history
                ],
                stream=True,
                max_tokens=2048,
            )

            async for chunk in stream:
                if chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    full_response += content

                    data = json.dumps({
                        "content": content,
                        "conversation_id": conv_id,
                        "done": False
                    }, ensure_ascii=False)
                    yield f"data: {data}\n\n"

            # Update the conversation history
            history.append({"role": "assistant", "content": full_response})
            conversations[conv_id] = history[-20:]  # Keep only the 20 most recent

            # Completion signal
            done_data = json.dumps({
                "content": "",
                "conversation_id": conv_id,
                "done": True
            })
            yield f"data: {done_data}\n\n"

        except Exception as e:
            error_data = json.dumps({"error": str(e), "done": True})
            yield f"data: {error_data}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        }
    )

@app.get("/chat/history/{conversation_id}")
async def get_history(conversation_id: str):
    history = conversations.get(conversation_id, [])
    return {"conversation_id": conversation_id, "messages": history}

WebSocket Chatbot

from fastapi import WebSocket, WebSocketDisconnect
import asyncio

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    conversation_history = []

    try:
        while True:
            # Receive a message
            data = await websocket.receive_json()
            user_message = data.get("message", "")

            if not user_message:
                continue

            conversation_history.append({
                "role": "user",
                "content": user_message
            })

            # Send the streaming response
            full_response = ""

            stream = await client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "You are a helpful AI assistant."},
                    *conversation_history[-10:]
                ],
                stream=True,
                max_tokens=1024,
            )

            async for chunk in stream:
                if chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    full_response += content
                    await websocket.send_json({
                        "type": "stream",
                        "content": content
                    })

            # Completion signal
            await websocket.send_json({"type": "done", "content": full_response})
            conversation_history.append({
                "role": "assistant",
                "content": full_response
            })

    except WebSocketDisconnect:
        print("Client disconnected")
    except Exception as e:
        await websocket.send_json({"type": "error", "message": str(e)})

6. Conversation Memory Management

Summary Memory

from openai import OpenAI
import tiktoken

client = OpenAI()

class SummaryMemory:
    def __init__(self, model: str = "gpt-4o-mini", max_tokens: int = 3000):
        self.model = model
        self.max_tokens = max_tokens
        self.summary = ""
        self.recent_messages = []
        self.encoder = tiktoken.encoding_for_model(model)

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def add_message(self, role: str, content: str):
        self.recent_messages.append({"role": role, "content": content})
        total_tokens = sum(self.count_tokens(m['content']) for m in self.recent_messages)

        if total_tokens > self.max_tokens:
            self._summarize_old_messages()

    def _summarize_old_messages(self):
        # Summarize half of the messages
        messages_to_summarize = self.recent_messages[:len(self.recent_messages)//2]
        self.recent_messages = self.recent_messages[len(self.recent_messages)//2:]

        conversation_text = "\n".join(
            f"{m['role']}: {m['content']}"
            for m in messages_to_summarize
        )

        summary_prompt = f"""
Existing summary: {self.summary}

New conversation:
{conversation_text}

Summarize the conversation above concisely. Include important information, decisions made, and user preferences.
"""

        response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": summary_prompt}],
            max_tokens=500,
        )
        self.summary = response.choices[0].message.content

    def get_messages_with_context(self) -> list:
        messages = []
        if self.summary:
            messages.append({
                "role": "system",
                "content": f"Summary of the previous conversation:\n{self.summary}"
            })
        messages.extend(self.recent_messages)
        return messages

# Usage example
memory = SummaryMemory()

def chat_with_memory(user_input: str) -> str:
    memory.add_message("user", user_input)

    messages_with_context = memory.get_messages_with_context()
    messages_with_context.insert(0, {
        "role": "system",
        "content": "You are a helpful AI assistant."
    })

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages_with_context,
        max_tokens=1024,
    )

    assistant_reply = response.choices[0].message.content
    memory.add_message("assistant", assistant_reply)
    return assistant_reply

7. Cost Optimization Strategies

Input Token Optimization

import tiktoken

def optimize_prompt(prompt: str, max_tokens: int = 1000) -> str:
    """Optimize the prompt"""
    encoder = tiktoken.encoding_for_model("gpt-4o")
    tokens = encoder.encode(prompt)

    if len(tokens) <= max_tokens:
        return prompt

    # Truncate once the token budget is exceeded
    truncated_tokens = tokens[:max_tokens]
    return encoder.decode(truncated_tokens)

# Strip unnecessary whitespace
def clean_prompt(prompt: str) -> str:
    import re
    # Collapse repeated spaces
    prompt = re.sub(r' +', ' ', prompt)
    # Minimize repeated newlines
    prompt = re.sub(r'\n{3,}', '\n\n', prompt)
    return prompt.strip()

# Summarize the document before using it
def compress_document(document: str, ratio: float = 0.3) -> str:
    """Summarize a document to cut the token count"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Use a cheap model for summarization
        messages=[
            {
                "role": "user",
                "content": f"Summarize the following document down to {ratio*100:.0f}% of the original, keeping only the essentials:\n\n{document}"
            }
        ],
        max_tokens=int(len(document.split()) * ratio * 1.5),
    )
    return response.choices[0].message.content

Model Routing Strategy

from enum import Enum

class TaskComplexity(Enum):
    SIMPLE = "simple"
    MEDIUM = "medium"
    COMPLEX = "complex"

def classify_task_complexity(query: str) -> TaskComplexity:
    """Pick a model according to query complexity"""
    # Simple rule-based classification
    simple_keywords = ["hello", "weather", "simple", "translate", "definition"]
    complex_keywords = ["analysis", "reasoning", "write code", "paper", "strategy"]

    query_lower = query.lower()

    if any(kw in query_lower for kw in simple_keywords):
        return TaskComplexity.SIMPLE
    elif any(kw in query_lower for kw in complex_keywords):
        return TaskComplexity.COMPLEX
    else:
        return TaskComplexity.MEDIUM

def get_optimal_model(complexity: TaskComplexity) -> str:
    """Pick the best model for a given complexity"""
    model_mapping = {
        TaskComplexity.SIMPLE: "gpt-4o-mini",     # Fast and cheap
        TaskComplexity.MEDIUM: "gpt-4o-mini",      # Balanced
        TaskComplexity.COMPLEX: "gpt-4o",          # High performance
    }
    return model_mapping[complexity]

def smart_chat(query: str) -> str:
    """Smart routing based on complexity"""
    complexity = classify_task_complexity(query)
    model = get_optimal_model(complexity)

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": query}],
        max_tokens=1024,
    )

    print(f"Model used: {model} (complexity: {complexity.value})")
    return response.choices[0].message.content

Making Use of the Batch API

import json

def batch_process_documents(documents: list, task: str) -> list:
    """Process many documents at once with the Batch API (50% cost savings)"""
    # Prepare the batch requests
    batch_requests = []
    for i, doc in enumerate(documents):
        batch_requests.append({
            "custom_id": f"request-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o-mini",
                "messages": [
                    {"role": "user", "content": f"{task}\n\n{doc}"}
                ],
                "max_tokens": 512
            }
        })

    # Create the JSONL file
    with open("batch_requests.jsonl", "w", encoding="utf-8") as f:
        for request in batch_requests:
            f.write(json.dumps(request, ensure_ascii=False) + "\n")

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

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

    print(f"Batch ID: {batch.id}, status: {batch.status}")
    return batch.id

8. Production Best Practices

Retry Logic

import time
import random
from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError

class RobustLLMClient:
    def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
        self.client = OpenAI()
        self.max_retries = max_retries
        self.base_delay = base_delay

    def create_with_retry(self, **kwargs) -> any:
        """Retry using exponential backoff"""
        last_exception = None

        for attempt in range(self.max_retries):
            try:
                return self.client.chat.completions.create(**kwargs)

            except RateLimitError as e:
                # Rate limit: wait longer
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limit reached. Retrying in {delay:.1f}s ({attempt+1}/{self.max_retries})")
                time.sleep(delay)
                last_exception = e

            except APITimeoutError as e:
                # Timeout: retry quickly
                delay = self.base_delay * (1.5 ** attempt)
                print(f"Timeout. Retrying in {delay:.1f}s ({attempt+1}/{self.max_retries})")
                time.sleep(delay)
                last_exception = e

            except APIConnectionError as e:
                # Connection error: wait for the network to recover
                delay = self.base_delay * (2 ** attempt) + random.uniform(1, 3)
                print(f"Connection error. Retrying in {delay:.1f}s ({attempt+1}/{self.max_retries})")
                time.sleep(delay)
                last_exception = e

            except Exception as e:
                # Errors that cannot be retried (400, 401, and so on)
                raise e

        raise last_exception

# Usage
robust_client = RobustLLMClient()
response = robust_client.create_with_retry(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
    max_tokens=100,
)

Logging and Monitoring

import logging
import time
import uuid
from functools import wraps
from dataclasses import dataclass
import json

# Structured logging setup
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class LLMCallLog:
    request_id: str
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    success: bool
    error: str = None
    estimated_cost_usd: float = 0.0

def log_llm_call(func):
    """Decorator that logs LLM API calls automatically"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        request_id = str(uuid.uuid4())[:8]
        start_time = time.time()
        log = LLMCallLog(
            request_id=request_id,
            model=kwargs.get('model', 'unknown'),
            input_tokens=0,
            output_tokens=0,
            latency_ms=0,
            success=False
        )

        try:
            result = func(*args, **kwargs)
            log.success = True
            log.latency_ms = (time.time() - start_time) * 1000

            if hasattr(result, 'usage'):
                log.input_tokens = result.usage.prompt_tokens
                log.output_tokens = result.usage.completion_tokens
                log.estimated_cost_usd = estimate_cost(
                    log.model,
                    log.input_tokens,
                    log.output_tokens
                )

            logger.info(json.dumps({
                "request_id": log.request_id,
                "model": log.model,
                "input_tokens": log.input_tokens,
                "output_tokens": log.output_tokens,
                "latency_ms": round(log.latency_ms, 2),
                "cost_usd": round(log.estimated_cost_usd, 6),
                "success": log.success
            }))

            return result

        except Exception as e:
            log.success = False
            log.error = str(e)
            log.latency_ms = (time.time() - start_time) * 1000

            logger.error(json.dumps({
                "request_id": log.request_id,
                "model": log.model,
                "error": log.error,
                "latency_ms": round(log.latency_ms, 2),
                "success": False
            }))
            raise

    return wrapper

@log_llm_call
def logged_chat(model: str, messages: list, **kwargs) -> any:
    return client.chat.completions.create(
        model=model,
        messages=messages,
        **kwargs
    )

9. Hands-On Project: A Code Review Bot

GitHub Webhook Setup

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
import hmac
import hashlib
import os
import httpx
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
GITHUB_WEBHOOK_SECRET = os.environ.get("GITHUB_WEBHOOK_SECRET")

def verify_webhook_signature(payload: bytes, signature: str) -> bool:
    """Verify the GitHub webhook signature"""
    expected = hmac.new(
        GITHUB_WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

async def get_pr_diff(owner: str, repo: str, pr_number: int) -> str:
    """Fetch the code changes in a PR"""
    async with httpx.AsyncClient() as http_client:
        response = await http_client.get(
            f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}",
            headers={
                "Authorization": f"token {GITHUB_TOKEN}",
                "Accept": "application/vnd.github.v3.diff"
            }
        )
        return response.text

async def review_code_with_llm(diff: str, pr_title: str) -> str:
    """Generate a code review with an LLM"""
    system_prompt = """You are an experienced senior developer.
Review the code changes in the PR and give constructive feedback in Korean.

Review checklist:
1. Possible bugs
2. Performance issues
3. Security vulnerabilities
4. Code readability
5. Adherence to best practices

Write the output in Markdown."""

    user_prompt = f"""PR title: {pr_title}

Code changes (Diff excerpt):
{diff[:8000]}

Please review the code changes above."""

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        max_tokens=2048,
        temperature=0.3,
    )

    return response.choices[0].message.content

async def post_review_comment(owner: str, repo: str, pr_number: int, review: str):
    """Post a review comment on the PR"""
    async with httpx.AsyncClient() as http_client:
        await http_client.post(
            f"https://api.github.com/repos/{owner}/{repo}/issues/{pr_number}/comments",
            json={"body": f"## AI Code Review\n\n{review}"},
            headers={
                "Authorization": f"token {GITHUB_TOKEN}",
                "Accept": "application/vnd.github.v3+json"
            }
        )

async def process_pr_event(payload: dict):
    """Handle a PR event"""
    action = payload.get("action")
    if action not in ["opened", "synchronize", "reopened"]:
        return

    pr = payload["pull_request"]
    owner = payload["repository"]["owner"]["login"]
    repo = payload["repository"]["name"]
    pr_number = pr["number"]
    pr_title = pr["title"]

    print(f"Starting review of PR #{pr_number}: {pr_title}")

    try:
        diff = await get_pr_diff(owner, repo, pr_number)
        review = await review_code_with_llm(diff, pr_title)
        await post_review_comment(owner, repo, pr_number, review)
        print(f"Finished review of PR #{pr_number}")
    except Exception as e:
        print(f"Review of PR #{pr_number} failed: {e}")

@app.post("/webhook/github")
async def github_webhook(
    request: Request,
    background_tasks: BackgroundTasks
):
    payload_bytes = await request.body()
    signature = request.headers.get("X-Hub-Signature-256", "")

    if not verify_webhook_signature(payload_bytes, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event_type = request.headers.get("X-GitHub-Event")
    payload = await request.json()

    if event_type == "pull_request":
        background_tasks.add_task(process_pr_event, payload)

    return {"status": "accepted"}

Advanced Code Review Features

from typing import List, Dict

class CodeReviewBot:
    def __init__(self):
        self.client = AsyncOpenAI()

    async def review_file(self, filename: str, content: str, diff: str) -> Dict:
        """Detailed per-file review"""
        response = await self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": """You are a code review expert. Provide the feedback in JSON format.
                    Include a severity (critical/major/minor) and a suggestion for each issue."""
                },
                {
                    "role": "user",
                    "content": f"""
File name: {filename}
Changes (Diff excerpt):
{diff}
Full current file (Excerpt):
{content[:3000]}
Review the code in this file and return it as JSON.
Format: {{"issues": [{{"line": 0, "severity": "critical", "message": "", "suggestion": ""}}], "summary": ""}}
"""
                }
            ],
            response_format={"type": "json_object"},
            temperature=0,
        )

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

    def format_review_comment(self, file_reviews: List[Dict]) -> str:
        """Format the review results as Markdown"""
        comment = "## Automated Code Review Results\n\n"

        critical_count = 0
        major_count = 0
        minor_count = 0

        for file_review in file_reviews:
            filename = file_review.get('filename', 'unknown')
            review = file_review.get('review', {})

            issues = review.get('issues', [])
            for issue in issues:
                severity = issue.get('severity', 'minor')
                if severity == 'critical':
                    critical_count += 1
                elif severity == 'major':
                    major_count += 1
                else:
                    minor_count += 1

        # Summary section
        comment += f"### Summary\n"
        comment += f"- Critical: {critical_count}\n"
        comment += f"- Major: {major_count}\n"
        comment += f"- Minor: {minor_count}\n\n"

        # Per-file detail
        for file_review in file_reviews:
            filename = file_review.get('filename', 'unknown')
            review = file_review.get('review', {})

            comment += f"### {filename}\n"
            comment += f"{review.get('summary', '')}\n\n"

            issues = review.get('issues', [])
            if issues:
                comment += "**Issues:**\n"
                severity_emoji = {'critical': '🔴', 'major': '🟡', 'minor': '🟢'}
                for issue in issues:
                    emoji = severity_emoji.get(issue.get('severity', 'minor'), '⚪')
                    comment += f"- {emoji} Line {issue.get('line', '?')}: {issue.get('message', '')}\n"
                    if issue.get('suggestion'):
                        comment += f"  - Suggestion: {issue['suggestion']}\n"
                comment += "\n"

        return comment

Wrapping Up

Key principles for building applications on top of LLM APIs:

Selection criteria:

General-purpose chatbot: GPT-4o-mini (cost efficiency) or Claude 3.5 Sonnet (quality)
Coding and reasoning: GPT-4o or Claude 3.7 (Extended Thinking)
Long-document processing: Gemini 1.5 Pro (1M context)
Minimizing cost: Gemini 1.5 Flash or gpt-4o-mini

Production essentials:

Retry logic (exponential backoff)
Rate-limit handling
Structured logging
Cost monitoring
Streaming to improve UX

Cutting cost:

Use small models for small tasks
Make use of the Batch API (50% savings)
Make use of the Anthropic prompt cache (90% savings)
Optimize prompts

The most important thing in LLM API development is "using the right tool for the right job". You do not need GPT-4o for every task, and with sensible model routing and a caching strategy you can cut cost dramatically.

LLM Application Development Practical Guide: Mastering ChatGPT API, Claude API, and Gemini API

LLM Application Development Practical Guide

1. LLM API Ecosystem Overview

Comparing the Major API Providers

API Cost Comparison (as of 2025, per 1M tokens)

2. The Complete OpenAI API Guide

Basic Setup

Basic Chat Completions

Streaming Responses

Function Calling (Using Tools)

Structured Outputs

Vision API

Embeddings API

3. Anthropic Claude API

Basic Setup

Messages API

Streaming

System Prompt Design Best Practices

Extended Thinking (Claude 3.7+)

Tool Use (Claude)

Prompt Caching (Cost Savings)

4. Google Gemini API

Basic Setup

Basic Usage

Multimodal (Text + Image)

Making Use of the 1M Context Window

Structured Output (JSON Mode)

5. Building a Streaming Chatbot

FastAPI Backend

WebSocket Chatbot

6. Conversation Memory Management

Summary Memory

7. Cost Optimization Strategies

Input Token Optimization

Model Routing Strategy

Making Use of the Batch API

8. Production Best Practices

Retry Logic

Logging and Monitoring

9. Hands-On Project: A Code Review Bot

GitHub Webhook Setup

Advanced Code Review Features

Wrapping Up

Comments