OpenAI API 최신 가이드

개요

이 가이드는 최신 OpenAI Python SDK (v1.68.0)를 기반으로 GPT-4o의 멀티모달 기능을 활용하는 방법을 설명합니다. 텍스트, 이미지, 스트리밍, 구조화된 출력 등 다양한 기능을 포괄합니다.

설치 및 환경 설정

1. 패키지 설치

pip install openai python-dotenv

2. 환경 변수 설정

# .env 파일에 API 키 설정
OPENAI_API_KEY=your_api_key_here

# 환경 변수 로드
from dotenv import load_dotenv
import os

load_dotenv()

3. 클라이언트 초기화

from openai import OpenAI, AsyncOpenAI

# 동기 클라이언트
client = OpenAI()

# 비동기 클라이언트  
async_client = AsyncOpenAI()

Chat Completions API

기본 사용법

from openai import OpenAI

client = OpenAI()

completion = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "developer", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
        {"role": "user", "content": "대한민국의 수도는 어디인가요?"}
    ],
    temperature=0.1,
    max_tokens=2048
)

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

응답 객체 구조

# 전체 응답 객체
print(completion)

# 메시지 내용만 추출
print(completion.choices[0].message.content)

# 메타데이터 확인
print(completion.usage)  # 토큰 사용량
print(completion.model)  # 사용된 모델

LogProbs 활성화

completion = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "대한민국의 수도는?"}],
    logprobs=True,
    top_logprobs=3  # 상위 3개 토큰 확률 표시
)

# LogProbs 정보 확인
for choice in completion.choices:
    if choice.logprobs:
        for content in choice.logprobs.content:
            print(f"토큰: {content.token}")
            print(f"확률: {content.logprob}")
            print(f"상위 대안들: {content.top_logprobs}")

스트리밍 응답

동기 스트리밍

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "대한민국의 아름다운 관광지 10곳을 소개해주세요."}],
    stream=True
)

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

비동기 스트리밍

import asyncio
from openai import AsyncOpenAI

async def stream_chat():
    client = AsyncOpenAI()
    
    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "파이썬 기초 개념을 설명해주세요."}],
        stream=True
    )
    
    async for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            print(chunk.choices[0].delta.content, end="", flush=True)

asyncio.run(stream_chat())

고급 스트리밍 (이벤트 기반)

async with client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "AI의 미래에 대해 설명해주세요."}]
) as stream:
    async for event in stream:
        if event.type == 'content.delta':
            print(event.delta, end="", flush=True)
        elif event.type == 'content.done':
            print(f"\n\n전체 내용: {event.content}")

구조화된 출력 (Structured Outputs)

Pydantic 모델 활용

from pydantic import BaseModel
from typing import List

class TourismSpot(BaseModel):
    name: str
    location: str
    description: str
    rating: float

class TourismResponse(BaseModel):
    spots: List[TourismSpot]
    total_count: int
    recommendation: str

# 구조화된 응답 생성
completion = client.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "당신은 여행 가이드입니다."},
        {"role": "user", "content": "서울의 유명한 관광지 5곳을 추천해주세요."}
    ],
    response_format=TourismResponse
)

# 파싱된 데이터 사용
if completion.choices[0].message.parsed:
    tourism_data = completion.choices[0].message.parsed
    for spot in tourism_data.spots:
        print(f"관광지: {spot.name}")
        print(f"위치: {spot.location}")
        print(f"평점: {spot.rating}/5.0")
        print(f"설명: {spot.description}\n")
else:
    print(f"파싱 실패: {completion.choices[0].message.refusal}")

JSON 모드

completion = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "JSON 형태로만 응답하세요."},
        {"role": "user", "content": "한국의 주요 도시 3곳의 정보를 제공해주세요."}
    ],
    response_format={"type": "json_object"}
)

import json
result = json.loads(completion.choices[0].message.content)
print(json.dumps(result, indent=2, ensure_ascii=False))

멀티모달 기능 (Vision)

URL 이미지 분석

def analyze_image_from_url(image_url: str, prompt: str):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url}
                    }
                ]
            }
        ],
        max_tokens=1000
    )
    return response.choices[0].message.content

# 사용 예시
image_url = "https://example.com/sample-image.jpg"
result = analyze_image_from_url(
    image_url, 
    "이 이미지에서 무엇을 볼 수 있나요? 자세히 설명해주세요."
)
print(result)

로컬 이미지 분석

import base64

def analyze_local_image(image_path: str, prompt: str):
    # 이미지를 base64로 인코딩
    with open(image_path, "rb") as image_file:
        base64_image = base64.b64encode(image_file.read()).decode('utf-8')
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user", 
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}"
                        }
                    }
                ]
            }
        ],
        max_tokens=1000
    )
    return response.choices[0].message.content

# 사용 예시
result = analyze_local_image(
    "./sample-image.jpg",
    "이 이미지의 내용을 분석하고 주요 특징을 설명해주세요."
)
print(result)

재무제표 분석 특화

class FinancialAnalyzer:
    def __init__(self, client: OpenAI):
        self.client = client
        self.system_prompt = """
        당신은 재무제표 분석 전문가입니다. 
        주어진 재무제표를 분석하여 다음 사항을 제공하세요:
        1. 주요 재무 지표의 변화
        2. 흥미로운 인사이트
        3. 잠재적 위험 요소
        4. 개선 사항 제안
        """
    
    def analyze_financial_statement(self, image_path: str):
        with open(image_path, "rb") as image_file:
            base64_image = base64.b64encode(image_file.read()).decode('utf-8')
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text", 
                            "text": "첨부된 재무제표를 분석해주세요."
                        },
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{base64_image}"
                            }
                        }
                    ]
                }
            ],
            max_tokens=2000
        )
        return response.choices[0].message.content

# 사용 예시
analyzer = FinancialAnalyzer(client)
analysis = analyzer.analyze_financial_statement("./financial-report.png")
print(analysis)

Responses API (차세대 API)

기본 사용법

# Responses API 사용 (최신 기능)
response = client.responses.create(
    model="gpt-4o",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "이 이미지에 대해 설명해주세요."},
                {"type": "input_image", "image_url": "https://example.com/image.jpg"}
            ]
        }
    ]
)

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

스트리밍과 함께 사용

import asyncio

async def stream_responses():
    stream = await client.responses.create(
        model="gpt-4o",
        input="AI의 미래에 대해 한 문장으로 이야기해주세요.",
        stream=True
    )
    
    async for event in stream:
        if hasattr(event, 'content'):
            print(event.content, end="", flush=True)

asyncio.run(stream_responses())

고급 기능

함수 호출 (Function Calling)

import json

# 함수 정의
def get_weather(location: str, unit: str = "celsius"):
    """날씨 정보를 가져옵니다."""
    # 실제 API 호출 대신 예시 데이터
    return {
        "location": location,
        "temperature": 22,
        "condition": "맑음",
        "unit": unit
    }

# 함수 스키마 정의
functions = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "지정된 위치의 현재 날씨를 가져옵니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "도시 이름, 예: 서울, 부산"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "온도 단위"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

# 함수 호출이 가능한 채팅
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "서울의 현재 날씨는 어때요?"}
    ],
    tools=functions,
    tool_choice="auto"
)

# 함수 호출 처리
if response.choices[0].message.tool_calls:
    for tool_call in response.choices[0].message.tool_calls:
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)
        
        if function_name == "get_weather":
            function_result = get_weather(**function_args)
            
            # 함수 결과를 다시 모델에 전달
            messages = [
                {"role": "user", "content": "서울의 현재 날씨는 어때요?"},
                response.choices[0].message,
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(function_result, ensure_ascii=False)
                }
            ]
            
            final_response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            
            print(final_response.choices[0].message.content)

배치 처리

# 여러 요청을 배치로 처리
batch_requests = [
    {
        "model": "gpt-4o",
        "messages": [{"role": "user", "content": f"숫자 {i}에 대해 설명해주세요."}],
        "max_tokens": 100
    } for i in range(1, 6)
]

# 동시 처리를 위한 비동기 함수
import asyncio

async def process_batch():
    client = AsyncOpenAI()
    
    tasks = [
        client.chat.completions.create(**request)
        for request in batch_requests
    ]
    
    results = await asyncio.gather(*tasks)
    
    for i, result in enumerate(results):
        print(f"요청 {i+1}: {result.choices[0].message.content}\n")

asyncio.run(process_batch())

커스텀 HTTP 클라이언트

import httpx
from openai import OpenAI

# 커스텀 HTTP 설정
custom_client = httpx.Client(
    timeout=30.0,
    limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
)

client = OpenAI(
    http_client=custom_client,
    max_retries=3,
    timeout=60.0
)

실용적인 활용 예제

1. 문서 요약 시스템

class DocumentSummarizer:
    def __init__(self, client: OpenAI):
        self.client = client
    
    def summarize_document(self, text: str, summary_type: str = "bullet_points"):
        prompts = {
            "bullet_points": "다음 문서를 주요 포인트별로 요약해주세요:",
            "executive": "다음 문서의 경영진 요약을 작성해주세요:",
            "technical": "다음 문서의 기술적 핵심 내용을 요약해주세요:"
        }
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": prompts.get(summary_type, prompts["bullet_points"])},
                {"role": "user", "content": text}
            ],
            max_tokens=1000,
            temperature=0.3
        )
        
        return response.choices[0].message.content

# 사용 예시
summarizer = DocumentSummarizer(client)
summary = summarizer.summarize_document(
    "긴 문서 텍스트...",
    summary_type="executive"
)

2. 다국어 번역기

class MultilingualTranslator:
    def __init__(self, client: OpenAI):
        self.client = client
    
    async def translate_batch(self, texts: List[str], target_language: str):
        tasks = [
            self.translate_single(text, target_language)
            for text in texts
        ]
        return await asyncio.gather(*tasks)
    
    async def translate_single(self, text: str, target_language: str):
        response = await AsyncOpenAI().chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": f"다음 텍스트를 {target_language}로 정확하게 번역해주세요. 번역 결과만 출력하세요."
                },
                {"role": "user", "content": text}
            ],
            temperature=0.1
        )
        return response.choices[0].message.content

# 사용 예시
translator = MultilingualTranslator(client)
texts = ["안녕하세요", "좋은 하루 되세요", "감사합니다"]
translations = await translator.translate_batch(texts, "영어")

3. 코드 리뷰 도우미

class CodeReviewer:
    def __init__(self, client: OpenAI):
        self.client = client
        self.system_prompt = """
        당신은 숙련된 소프트웨어 엔지니어입니다.
        제공된 코드를 리뷰하여 다음을 제공하세요:
        1. 코드 품질 평가 (1-10점)
        2. 개선 제안사항
        3. 보안 취약점 (있다면)
        4. 성능 최적화 방안
        5. 코드 스타일 및 가독성 피드백
        """
    
    def review_code(self, code: str, language: str):
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {
                    "role": "user", 
                    "content": f"다음 {language} 코드를 리뷰해주세요:\n\n```{language}\n{code}\n```"
                }
            ],
            max_tokens=2000,
            temperature=0.2
        )
        return response.choices[0].message.content

# 사용 예시
reviewer = CodeReviewer(client)
python_code = """
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
"""

review = reviewer.review_code(python_code, "python")
print(review)

모니터링 및 디버깅

사용량 추적

class UsageTracker:
    def __init__(self):
        self.total_tokens = 0
        self.total_cost = 0
        self.request_count = 0
    
    def track_request(self, response):
        if hasattr(response, 'usage'):
            self.total_tokens += response.usage.total_tokens
            self.request_count += 1
            
            # GPT-4o 가격 계산 (예시)
            input_cost = response.usage.prompt_tokens * 0.005 / 1000
            output_cost = response.usage.completion_tokens * 0.015 / 1000
            self.total_cost += input_cost + output_cost
    
    def get_stats(self):
        return {
            "total_requests": self.request_count,
            "total_tokens": self.total_tokens,
            "estimated_cost": f"${self.total_cost:.4f}"
        }

# 사용 예시
tracker = UsageTracker()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "안녕하세요!"}]
)

tracker.track_request(response)
print(tracker.get_stats())

오류 처리

from openai import OpenAI, APIError, RateLimitError, APIConnectionError
import time

class RobustOpenAIClient:
    def __init__(self, max_retries=3, backoff_factor=2):
        self.client = OpenAI()
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
    
    def robust_completion(self, **kwargs):
        for attempt in range(self.max_retries):
            try:
                return self.client.chat.completions.create(**kwargs)
                
            except RateLimitError as e:
                print(f"Rate limit exceeded. Waiting...")
                time.sleep(60)  # 1분 대기
                
            except APIConnectionError as e:
                print(f"Connection error: {e}")
                wait_time = self.backoff_factor ** attempt
                time.sleep(wait_time)
                
            except APIError as e:
                print(f"API error: {e}")
                if attempt == self.max_retries - 1:
                    raise
                    
        raise Exception("Max retries exceeded")

# 사용 예시
robust_client = RobustOpenAIClient()
response = robust_client.robust_completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "안녕하세요!"}]
)

모범 사례

1. 프롬프트 엔지니어링

class PromptTemplate:
    @staticmethod
    def few_shot_learning(task: str, examples: List[dict], query: str):
        system_prompt = f"다음 작업을 수행하세요: {task}\n\n예시들:\n"
        
        for example in examples:
            system_prompt += f"입력: {example['input']}\n출력: {example['output']}\n\n"
        
        return [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"입력: {query}\n출력:"}
        ]
    
    @staticmethod
    def chain_of_thought(question: str):
        return [
            {
                "role": "system", 
                "content": "문제를 단계별로 분석하여 해결하세요. 각 단계의 사고 과정을 명확히 설명하고 최종 답을 제시하세요."
            },
            {"role": "user", "content": question}
        ]

# 사용 예시
examples = [
    {"input": "기쁘다", "output": "긍정"},
    {"input": "슬프다", "output": "부정"},
    {"input": "화나다", "output": "부정"}
]

messages = PromptTemplate.few_shot_learning(
    "감정 분류", 
    examples, 
    "즐겁다"
)

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

2. 환경별 설정

import os
from enum import Enum

class Environment(Enum):
    DEVELOPMENT = "dev"
    PRODUCTION = "prod"

class OpenAIConfig:
    def __init__(self, environment: Environment):
        self.environment = environment
        self.setup_config()
    
    def setup_config(self):
        if self.environment == Environment.DEVELOPMENT:
            self.model = "gpt-4o-mini"  # 개발용 저비용 모델
            self.max_tokens = 1000
            self.temperature = 0.7
        else:
            self.model = "gpt-4o"
            self.max_tokens = 4000
            self.temperature = 0.1
    
    def get_client(self):
        return OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            timeout=60.0,
            max_retries=3
        )

# 사용 예시
config = OpenAIConfig(Environment.DEVELOPMENT)
client = config.get_client()

결론

이 가이드는 최신 OpenAI Python SDK의 핵심 기능들을 포괄적으로 다룹니다:

• Chat Completions API: 기본 텍스트 생성부터 고급 기능까지
• 스트리밍: 실시간 응답 처리
• 구조화된 출력: Pydantic 모델과 JSON 스키마 활용
• 멀티모달: 이미지 분석과 텍스트의 결합
• Responses API: 차세대 통합 인터페이스
• 함수 호출: 외부 도구와의 연동
• 모범 사례: 프로덕션 환경을 위한 실용적 패턴

OpenAI의 강력한 AI 모델들을 효과적으로 활용하여 다양한 애플리케이션을 구축할 수 있습니다. 지속적으로 업데이트되는 SDK의 새로운 기능들을 활용하여 더욱 혁신적인 솔루션을 개발해보세요.

---