이미지 로딩 중...
AI Generated
2025. 11. 11. · 3 Views
Python으로 AI 에이전트 만들기 2편 - OpenAI API 연동과 첫 번째 챗봇
OpenAI API를 사용하여 실제로 대화가 가능한 AI 챗봇을 만들어봅니다. API 키 설정부터 GPT 모델 호출, 대화 컨텍스트 관리, 스트리밍 응답까지 실무에서 바로 활용 가능한 완전한 챗봇 구현 방법을 단계별로 학습합니다.
목차
- OpenAI API 키 설정
- OpenAI Python 라이브러리 설치 및 초기화
- Chat Completions API 기본 호출
- 시스템 프롬프트 설정
- 대화 컨텍스트 관리
- 스트리밍 응답 구현
- 토큰 사용량 추적
- 에러 핸들링
1. OpenAI API 키 설정
시작하며
여러분이 OpenAI API를 처음 사용하려고 할 때 가장 먼저 만나는 질문이 "API 키를 어디에 어떻게 저장해야 할까?"입니다. 코드에 직접 하드코딩했다가 실수로 GitHub에 올려서 API 키가 노출되고, 며칠 뒤 수백 달러의 청구서를 받는 사례가 실제로 자주 발생합니다.
이런 보안 사고는 개인 개발자뿐만 아니라 기업에서도 심각한 문제를 일으킵니다. 노출된 API 키는 악의적인 사용자에 의해 몇 시간 만에 수천 달러의 비용을 발생시킬 수 있습니다.
바로 이럴 때 필요한 것이 환경변수를 사용한 API 키 관리입니다. 환경변수를 사용하면 코드와 민감한 정보를 완전히 분리하여 보안을 유지하면서도 편리하게 개발할 수 있습니다.
개요
간단히 말해서, 환경변수를 사용한 API 키 관리는 민감한 인증 정보를 코드 외부에 저장하는 방법입니다. 실제 프로젝트에서는 개발 환경, 스테이징 환경, 프로덕션 환경마다 다른 API 키를 사용해야 합니다.
환경변수를 사용하면 코드 변경 없이 환경에 따라 다른 키를 자동으로 사용할 수 있습니다. 예를 들어, 로컬에서는 개발용 API 키를, 서버에서는 프로덕션 키를 자동으로 사용하게 할 수 있습니다.
기존에는 설정 파일에 API 키를 직접 작성했다면, 이제는 .env 파일과 python-dotenv 라이브러리를 사용하여 안전하게 관리할 수 있습니다. 환경변수 방식의 핵심 특징은 세 가지입니다.
첫째, .gitignore에 .env 파일을 추가하여 버전 관리에서 제외됩니다. 둘째, 환경마다 다른 설정을 쉽게 적용할 수 있습니다.
셋째, 팀원들과 코드를 공유할 때 각자의 API 키를 사용할 수 있습니다. 이러한 특징들이 실무에서 안전하고 유연한 개발을 가능하게 합니다.
코드 예제
# 1. 필요한 패키지 설치 (터미널에서 실행)
# pip install python-dotenv openai
import os
from dotenv import load_dotenv
# .env 파일에서 환경변수 로드
load_dotenv()
# 환경변수에서 API 키 가져오기
api_key = os.getenv("OPENAI_API_KEY")
# API 키 존재 여부 확인
if not api_key:
raise ValueError("OPENAI_API_KEY가 설정되지 않았습니다. .env 파일을 확인하세요.")
print("API 키가 성공적으로 로드되었습니다!")
print(f"키의 앞 4자리: {api_key[:4]}...") # 보안을 위해 일부만 출력
설명
이것이 하는 일: 이 코드는 OpenAI API 키를 안전하게 로드하고 검증하는 전체 프로세스를 보여줍니다. 프로젝트 루트에 .env 파일을 만들고 그곳에 OPENAI_API_KEY=sk-... 형식으로 API 키를 저장한 후, 이 코드를 실행하면 됩니다.
첫 번째 단계는 python-dotenv 라이브러리를 사용하여 .env 파일을 찾고 읽는 것입니다. load_dotenv() 함수는 현재 디렉토리와 상위 디렉토리에서 .env 파일을 자동으로 찾아 그 안의 변수들을 시스템 환경변수로 등록합니다.
이렇게 하는 이유는 코드가 운영체제의 환경변수를 읽는 것처럼 자연스럽게 .env 파일의 값을 사용할 수 있게 하기 위함입니다. 그 다음으로 os.getenv() 함수를 사용하여 실제 API 키 값을 가져옵니다.
이 함수는 환경변수가 존재하지 않을 경우 None을 반환하므로, 항상 존재 여부를 확인해야 합니다. 만약 API 키가 없다면 프로그램을 계속 실행하는 것은 의미가 없으므로 명확한 에러 메시지와 함께 예외를 발생시킵니다.
마지막으로 보안을 위해 API 키 전체를 출력하지 않고 앞 4자리만 출력하여 키가 올바르게 로드되었는지 확인합니다. 이는 로그 파일이나 콘솔 출력에 전체 API 키가 노출되는 것을 방지합니다.
여러분이 이 코드를 사용하면 GitHub에 코드를 푸시할 때 .env 파일만 .gitignore에 추가하면 되므로 API 키 노출 걱정 없이 안전하게 협업할 수 있습니다. 팀원들은 각자의 .env 파일을 만들어 자신의 API 키를 사용하면 되고, 프로덕션 서버에서는 환경변수를 직접 설정하여 .env 파일 없이도 동작하게 할 수 있습니다.
실전 팁
💡 .env 파일을 생성한 후 반드시 .gitignore에 .env를 추가하세요. 이미 커밋한 경우 git rm --cached .env로 Git 추적에서 제거하고, GitHub에 푸시했다면 즉시 OpenAI 대시보드에서 해당 API 키를 삭제하고 새로 발급받아야 합니다.
💡 .env.example 파일을 만들어 OPENAI_API_KEY=your_api_key_here 같은 템플릿을 제공하면 팀원들이 쉽게 설정할 수 있습니다. 이 파일은 실제 값이 없으므로 Git에 커밋해도 안전합니다.
💡 개발 환경과 프로덕션 환경에서 다른 API 키를 사용하세요. OpenAI 대시보드에서 용도별로 키를 분리하면 비용 추적과 보안 관리가 훨씬 쉬워집니다.
💡 load_dotenv()는 기본적으로 현재 디렉토리의 .env 파일을 찾지만, load_dotenv("/path/to/.env") 형식으로 경로를 지정할 수도 있습니다. 복잡한 프로젝트 구조에서 유용합니다.
💡 Docker를 사용한다면 .env 파일 대신 docker-compose.yml의 environment 섹션이나 --env-file 옵션을 활용하여 환경변수를 주입하는 것이 더 적합합니다.
2. OpenAI Python 라이브러리 설치 및 초기화
시작하며
여러분이 OpenAI API를 사용하려고 할 때 직접 HTTP 요청을 만들어 보내는 방법도 있지만, 이는 매우 번거롭고 에러가 발생하기 쉽습니다. 요청 헤더 설정, JSON 파싱, 에러 처리 등을 모두 직접 구현해야 하기 때문입니다.
실제로 많은 초보 개발자들이 requests 라이브러리로 API를 직접 호출하다가 인증 오류, 타임아웃, 응답 파싱 문제 등으로 어려움을 겪습니다. 특히 스트리밍 응답이나 함수 호출 같은 고급 기능을 사용할 때는 복잡도가 기하급수적으로 증가합니다.
바로 이럴 때 필요한 것이 공식 OpenAI Python 라이브러리입니다. 이 라이브러리는 모든 복잡한 처리를 내부적으로 해결해주고, 간단한 Python 코드만으로 강력한 AI 기능을 사용할 수 있게 해줍니다.
개요
간단히 말해서, OpenAI Python 라이브러리는 OpenAI API를 Python에서 쉽게 사용할 수 있도록 만든 공식 SDK(Software Development Kit)입니다. 이 라이브러리가 필요한 이유는 API 통신의 모든 세부사항을 추상화하여 개발자가 비즈니스 로직에만 집중할 수 있게 하기 때문입니다.
인증, 재시도 로직, 에러 처리, 타입 힌트, 자동완성 등이 모두 포함되어 있어 개발 생산성이 크게 향상됩니다. 예를 들어, 네트워크 오류가 발생했을 때 자동으로 재시도하는 기능이 내장되어 있어 안정적인 서비스를 쉽게 만들 수 있습니다.
기존에는 curl이나 requests로 직접 HTTP 요청을 만들어야 했다면, 이제는 client.chat.completions.create() 같은 직관적인 메서드 호출만으로 동일한 작업을 수행할 수 있습니다. 이 라이브러리의 핵심 특징은 세 가지입니다.
첫째, 타입 안전성을 제공하여 IDE에서 자동완성과 타입 체크가 가능합니다. 둘째, 동기/비동기 API를 모두 지원하여 다양한 아키텍처에 적용할 수 있습니다.
셋째, 공식 라이브러리이므로 새로운 API 기능이 추가되면 즉시 업데이트됩니다. 이러한 특징들이 안정적이고 유지보수하기 쉬운 AI 애플리케이션 개발을 가능하게 합니다.
코드 예제
# 터미널에서 실행: pip install openai
from openai import OpenAI
import os
from dotenv import load_dotenv
# 환경변수 로드
load_dotenv()
# OpenAI 클라이언트 초기화
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
timeout=30.0, # 30초 타임아웃 설정
max_retries=2 # 실패 시 최대 2번 재시도
)
# 연결 테스트: 사용 가능한 모델 목록 가져오기
models = client.models.list()
print(f"사용 가능한 모델 수: {len(models.data)}")
print(f"첫 번째 모델: {models.data[0].id}")
설명
이것이 하는 일: 이 코드는 OpenAI API와 통신할 수 있는 클라이언트 객체를 생성하고 설정하는 과정을 보여줍니다. 클라이언트 객체는 이후 모든 API 호출에서 재사용되므로 애플리케이션 시작 시 한 번만 생성하면 됩니다.
첫 번째로, OpenAI 클래스를 임포트하고 인스턴스를 생성합니다. 생성자에는 여러 옵션을 전달할 수 있는데, 가장 중요한 것은 api_key입니다.
이전에 설정한 환경변수에서 API 키를 가져와 전달합니다. timeout 매개변수는 API 응답을 기다리는 최대 시간을 초 단위로 설정하는데, GPT-4 같은 모델은 응답이 느릴 수 있으므로 30초 정도가 적당합니다.
그 다음으로 max_retries 옵션을 설정합니다. 이는 네트워크 오류나 일시적인 서버 문제가 발생했을 때 자동으로 재시도하는 횟수를 지정합니다.
2번 정도로 설정하면 일시적인 문제는 자동으로 해결되면서도 무한 대기를 방지할 수 있습니다. 라이브러리는 지수 백오프(exponential backoff) 전략을 사용하여 재시도 사이의 대기 시간을 점진적으로 늘립니다.
마지막으로 연결을 테스트하기 위해 간단한 API 호출을 수행합니다. client.models.list()는 사용 가능한 모델 목록을 가져오는 가벼운 API로, 인증이 올바르게 되었는지 확인하는 데 적합합니다.
이 호출이 성공하면 클라이언트가 제대로 설정된 것입니다. 여러분이 이 코드를 사용하면 한 번 생성한 client 객체를 애플리케이션 전체에서 재사용할 수 있습니다.
FastAPI나 Flask 같은 웹 프레임워크에서는 애플리케이션 시작 시 전역 변수로 클라이언트를 생성하고, 각 요청 핸들러에서 이를 사용하는 패턴이 일반적입니다. 이렇게 하면 매 요청마다 클라이언트를 새로 만드는 오버헤드를 피할 수 있습니다.
실전 팁
💡 클라이언트 객체는 스레드 안전(thread-safe)하므로 멀티스레드 환경에서 하나의 인스턴스를 공유해도 안전합니다. 애플리케이션 시작 시 한 번만 생성하여 전역적으로 사용하세요.
💡 비동기 코드를 작성한다면 from openai import AsyncOpenAI를 사용하세요. FastAPI나 asyncio 기반 애플리케이션에서는 AsyncOpenAI가 훨씬 효율적입니다.
💡 timeout 값은 사용하는 모델과 요청 크기에 따라 조절해야 합니다. GPT-4로 긴 문서를 처리할 때는 60초 이상이 필요할 수 있고, 간단한 챗봇은 10초로도 충분합니다.
💡 개발 중에는 환경변수 대신 OpenAI(api_key="sk-...")로 직접 전달해도 되지만, 프로덕션 코드에서는 절대 하드코딩하지 마세요. 코드 리뷰에서 가장 많이 지적되는 보안 이슈입니다.
💡 client.models.retrieve("gpt-4") 같은 메서드로 특정 모델의 상세 정보를 확인할 수 있습니다. 새로운 모델이 출시되었는지, 어떤 기능을 지원하는지 프로그래밍 방식으로 확인할 때 유용합니다.
3. Chat Completions API 기본 호출
시작하며
여러분이 처음으로 AI에게 질문을 하고 답변을 받고 싶을 때, 어떤 메서드를 어떻게 호출해야 할지 막막할 수 있습니다. OpenAI API는 여러 종류가 있는데, 챗봇을 만들려면 어떤 API를 사용해야 할까요?
많은 초보자들이 구버전 Completions API를 사용하거나, 매개변수를 잘못 설정하여 예상과 다른 결과를 받곤 합니다. 예를 들어 messages 형식을 잘못 구성하거나, 모델 이름을 잘못 입력하여 에러를 만나는 경우가 흔합니다.
바로 이럴 때 필요한 것이 Chat Completions API의 기본 사용법입니다. 이 API는 현재 가장 권장되는 방식으로, GPT-3.5부터 GPT-4까지 모든 최신 모델을 지원하며 대화형 AI를 구현하는 표준 방법입니다.
개요
간단히 말해서, Chat Completions API는 대화형 메시지를 주고받는 형식으로 AI와 상호작용하는 API입니다. 이 API가 필요한 이유는 단순한 텍스트 완성을 넘어 맥락을 이해하는 대화를 만들 수 있기 때문입니다.
각 메시지에 역할(role)을 지정할 수 있어 시스템 지시사항, 사용자 입력, AI 응답을 명확히 구분할 수 있습니다. 예를 들어, 고객 상담 챗봇을 만들 때 시스템 역할로 "당신은 친절한 고객 상담원입니다"라고 지정하면 AI가 그 역할에 맞게 응답합니다.
기존 Completions API에서는 단순히 텍스트를 이어서 생성했다면, Chat Completions API는 대화의 구조를 이해하고 이전 메시지를 참조하여 일관성 있는 답변을 생성합니다. 이 API의 핵심 특징은 세 가지입니다.
첫째, 메시지 배열 형식으로 대화 히스토리를 관리합니다. 둘째, system, user, assistant 세 가지 역할로 맥락을 제공합니다.
셋째, temperature, max_tokens 같은 매개변수로 응답 스타일을 세밀하게 조절할 수 있습니다. 이러한 특징들이 실무에서 다양한 용도의 챗봇을 만들 수 있게 해줍니다.
코드 예제
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 간단한 질문-답변
response = client.chat.completions.create(
model="gpt-3.5-turbo", # 또는 "gpt-4"
messages=[
{"role": "user", "content": "Python에서 리스트와 튜플의 차이점을 간단히 설명해줘"}
],
temperature=0.7, # 0.0(결정적) ~ 2.0(창의적)
max_tokens=150 # 최대 응답 길이
)
# 응답 내용 추출
answer = response.choices[0].message.content
print(f"AI 응답: {answer}")
print(f"\n사용한 토큰: {response.usage.total_tokens}")
설명
이것이 하는 일: 이 코드는 OpenAI의 GPT 모델에게 질문을 보내고 답변을 받는 가장 기본적인 과정을 보여줍니다. 한 번의 API 호출로 질문과 답변이 완성되는 구조입니다.
첫 번째로, client.chat.completions.create() 메서드를 호출합니다. 이 메서드는 여러 매개변수를 받는데, 가장 중요한 것은 model과 messages입니다.
model은 사용할 GPT 모델을 지정하는데, "gpt-3.5-turbo"는 빠르고 저렴하며, "gpt-4"는 더 정확하지만 비용이 높습니다. 프로토타입 단계에서는 3.5-turbo로 시작하는 것이 좋습니다.
그 다음으로 messages 배열을 구성합니다. 각 메시지는 딕셔너리 형태로 role과 content를 가집니다.
role은 "system"(AI의 행동 지시), "user"(사용자 입력), "assistant"(AI 응답) 중 하나입니다. 이 예제에서는 가장 간단한 형태로 user 역할의 메시지 하나만 전달했습니다.
temperature는 응답의 무작위성을 조절하는데, 0에 가까우면 일관되고 예측 가능한 답변을, 2에 가까우면 창의적이고 다양한 답변을 생성합니다. 세 번째로 max_tokens 매개변수로 응답의 최대 길이를 제한합니다.
토큰은 대략 단어의 일부 정도로, 영어는 단어당 1-2토큰, 한국어는 단어당 2-3토큰 정도입니다. 150 토큰이면 한국어로 약 50-75단어 정도의 간단한 답변을 받을 수 있습니다.
이를 설정하지 않으면 모델의 최대 컨텍스트 길이까지 생성할 수 있어 비용이 크게 증가할 수 있습니다. 마지막으로 응답 객체에서 실제 답변 텍스트를 추출합니다.
response.choices[0].message.content는 AI가 생성한 텍스트이고, response.usage.total_tokens는 이번 호출에서 사용한 총 토큰 수(입력 + 출력)를 보여줍니다. 이 정보는 비용 계산과 최적화에 필수적입니다.
여러분이 이 코드를 사용하면 몇 줄의 코드만으로 강력한 AI 기능을 애플리케이션에 추가할 수 있습니다. 웹 애플리케이션의 FAQ 봇, 문서 요약 기능, 코드 설명 도구 등 다양한 곳에 활용할 수 있습니다.
실전 팁
💡 model 선택은 용도에 따라 달라집니다. 간단한 분류나 요약은 "gpt-3.5-turbo", 복잡한 추론이나 코드 생성은 "gpt-4", 최신 기능이 필요하면 "gpt-4-turbo"를 사용하세요. 가격은 3.5 < 4 < 4-turbo 순입니다.
💡 temperature는 작업 유형에 따라 조절하세요. 수학 문제나 코드 생성은 0.0-0.3(정확성), 창작 글쓰기나 아이디어 생성은 0.7-1.0(창의성), 시나 소설은 1.0-2.0(다양성)이 적합합니다.
💡 max_tokens를 너무 낮게 설정하면 답변이 중간에 잘릴 수 있습니다. response.choices[0].finish_reason이 "length"이면 토큰 제한으로 잘린 것이므로 max_tokens를 늘려야 합니다. "stop"이면 정상적으로 완료된 것입니다.
💡 API 호출은 비용이 발생하므로 개발 중에는 응답을 캐싱하는 것이 좋습니다. 같은 질문에 대해 매번 API를 호출하지 말고, 딕셔너리나 Redis 같은 캐시에 저장하여 재사용하세요.
💡 에러 처리를 반드시 추가하세요. try-except 블록으로 openai.APIError, openai.RateLimitError, openai.APIConnectionError를 처리하면 네트워크 문제나 할당량 초과 시에도 애플리케이션이 안정적으로 동작합니다.
4. 시스템 프롬프트 설정
시작하며
여러분이 챗봇을 만들 때 AI가 항상 일관된 태도와 톤으로 대답하길 원한다면, 매번 사용자 메시지에 "친절하게 대답해줘"라고 붙이는 것은 비효율적입니다. 또한 사용자가 AI의 역할을 바꾸려고 시도할 때 이를 제어하기 어렵습니다.
실제 프로덕션 환경에서는 AI가 특정 전문가 역할(예: 의료 상담, 법률 자문, 프로그래밍 튜터)을 수행하도록 해야 하는데, 이를 사용자에게 맡기면 일관성이 떨어지고 품질을 보장할 수 없습니다. 바로 이럴 때 필요한 것이 시스템 프롬프트(system prompt)입니다.
시스템 메시지로 AI의 성격, 역할, 제약사항을 미리 정의하면 모든 대화에서 일관된 품질을 유지할 수 있습니다.
개요
간단히 말해서, 시스템 프롬프트는 AI에게 "당신은 누구이고 어떻게 행동해야 하는지"를 알려주는 특별한 지시사항입니다. 시스템 프롬프트가 중요한 이유는 사용자의 개별 메시지보다 더 강한 영향력을 가지며, 대화 전체의 맥락을 설정하기 때문입니다.
OpenAI 모델은 시스템 메시지를 특별히 취급하여 이후 모든 응답의 기준으로 삼습니다. 예를 들어, "당신은 초등학생에게 설명하는 과학 선생님입니다"라고 설정하면 복잡한 개념도 쉬운 말로 풀어서 설명합니다.
기존에는 사용자 메시지에 역할 설명을 포함시켰다면, 이제는 시스템 메시지로 분리하여 사용자가 보지 못하게 하면서도 AI의 행동을 제어할 수 있습니다. 시스템 프롬프트의 핵심 특징은 네 가지입니다.
첫째, 대화 전체에 걸쳐 지속적으로 영향을 미칩니다. 둘째, 사용자가 직접 보거나 수정할 수 없어 보안성이 높습니다.
셋째, 톤, 스타일, 지식 범위, 응답 형식 등을 세밀하게 조절할 수 있습니다. 넷째, 잘 작성된 시스템 프롬프트는 few-shot 예제 없이도 높은 품질의 응답을 보장합니다.
이러한 특징들이 전문적이고 일관된 AI 서비스를 만드는 핵심 요소입니다.
코드 예제
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 전문적인 Python 튜터 챗봇 만들기
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "system",
"content": """당신은 10년 경력의 Python 전문 프로그래밍 강사입니다.
초보자도 이해할 수 있도록 쉽게 설명하되, 정확한 기술 용어를 사용합니다.
코드 예제를 항상 포함하고, 실무 관점의 조언을 제공합니다.
답변은 200단어 이내로 간결하게 작성합니다."""
},
{
"role": "user",
"content": "리스트 컴프리헨션이 뭐야?"
}
],
temperature=0.7,
max_tokens=300
)
print(response.choices[0].message.content)
설명
이것이 하는 일: 이 코드는 시스템 프롬프트를 사용하여 AI를 특정 역할을 가진 전문가로 만드는 과정을 보여줍니다. 시스템 메시지가 있을 때와 없을 때의 응답 품질 차이는 매우 큽니다.
첫 번째로, messages 배열의 첫 번째 요소로 role이 "system"인 메시지를 추가합니다. 이 메시지는 반드시 대화의 맨 처음에 와야 가장 효과적입니다.
시스템 메시지의 content에는 AI의 정체성(누구인지), 전문성(무엇을 잘하는지), 응답 스타일(어떻게 대답할지), 제약사항(무엇을 하지 말아야 하는지)을 명확히 작성합니다. 이 예제에서는 "10년 경력의 Python 강사"라는 정체성을 부여했습니다.
그 다음으로 시스템 프롬프트 내에서 구체적인 지침을 여러 개 제공합니다. "초보자도 이해할 수 있도록"은 설명 수준을, "정확한 기술 용어 사용"은 전문성을, "코드 예제 포함"은 응답 형식을, "200단어 이내"는 길이 제약을 지정합니다.
이렇게 세부적으로 지정할수록 원하는 형태의 답변을 얻을 가능성이 높아집니다. 여러 줄로 작성할 때는 삼중 따옴표(""")를 사용하면 가독성이 좋습니다.
세 번째로 시스템 메시지 다음에 실제 사용자 질문을 user role로 추가합니다. 사용자는 "리스트 컴프리헨션이 뭐야?"라는 간단한 질문만 했지만, AI는 시스템 프롬프트의 지침에 따라 코드 예제를 포함하고 초보자 수준에 맞춰 설명하며 200단어 이내로 답변합니다.
마지막으로 temperature를 0.7로 설정하여 교육적이면서도 약간의 창의성을 가진 설명을 유도합니다. 너무 낮으면 기계적인 답변이 되고, 너무 높으면 산만해질 수 있습니다.
교육용 챗봇에는 0.7 정도가 적절합니다. 여러분이 이 코드를 사용하면 고객 상담 봇, 기술 지원 봇, 교육 튜터 등 다양한 도메인별 전문 챗봇을 만들 수 있습니다.
시스템 프롬프트만 바꾸면 같은 코드로 전혀 다른 성격의 봇을 만들 수 있습니다. 예를 들어 "당신은 친절한 고객 상담원입니다"로 바꾸면 상담 봇이 되고, "당신은 코드 리뷰어입니다"로 바꾸면 코드 리뷰 봇이 됩니다.
실전 팁
💡 시스템 프롬프트에는 출력 형식도 지정할 수 있습니다. "답변은 항상 JSON 형식으로 {answer: ..., confidence: ...} 구조를 따릅니다"라고 하면 파싱하기 쉬운 구조화된 응답을 받을 수 있습니다.
💡 제약사항을 명확히 하세요. "정치, 종교, 의료 조언은 제공하지 않습니다"처럼 AI가 답하지 말아야 할 주제를 지정하면 법적 리스크를 줄일 수 있습니다.
💡 시스템 프롬프트도 토큰을 소비하므로 너무 길면 비용이 증가합니다. 핵심만 간결하게 작성하되, 100-300 토큰(한국어 약 50-150단어) 정도가 적당합니다. 필요하다면 외부 문서를 참조하게 하는 것이 더 효율적입니다.
💡 A/B 테스팅으로 시스템 프롬프트를 최적화하세요. 같은 질문에 대해 다른 프롬프트로 응답을 생성하고 품질을 비교하면 어떤 표현이 더 효과적인지 알 수 있습니다.
💡 버전 관리를 하세요. 시스템 프롬프트를 코드에 하드코딩하지 말고 별도 파일이나 데이터베이스에 저장하면 배포 없이 프롬프트를 업데이트할 수 있고, 여러 버전을 테스트하기도 쉽습니다.
5. 대화 컨텍스트 관리
시작하며
여러분이 챗봇과 여러 차례 대화를 주고받을 때, AI가 이전에 무슨 말을 했는지 기억하지 못한다면 대화가 전혀 자연스럽지 않을 것입니다. "내 이름은 철수야"라고 했는데 다음 메시지에서 "너 이름이 뭐였지?"라고 물으면 사용자 경험이 최악이 됩니다.
많은 초보자들이 매번 새로운 API 호출을 할 때 이전 대화를 포함시키지 않아 AI가 기억상실증에 걸린 것처럼 동작하는 문제를 겪습니다. OpenAI API는 기본적으로 상태를 유지하지 않으므로(stateless) 개발자가 직접 대화 히스토리를 관리해야 합니다.
바로 이럴 때 필요한 것이 대화 컨텍스트 관리입니다. 이전 메시지들을 messages 배열에 계속 추가하여 AI가 대화의 흐름을 이해하고 일관된 응답을 할 수 있게 만듭니다.
개요
간단히 말해서, 대화 컨텍스트 관리는 사용자와 AI의 이전 메시지들을 모두 기록하고 다음 API 호출 시 함께 전달하는 기법입니다. 이 기법이 필요한 이유는 OpenAI API가 각 호출을 독립적으로 처리하기 때문입니다.
웹 애플리케이션의 세션처럼 서버가 상태를 기억해주지 않으므로, 클라이언트(우리 코드)가 전체 대화를 저장하고 매번 함께 보내야 합니다. 예를 들어, 고객이 "주문을 취소하고 싶어요"라고 했을 때 AI가 "어떤 주문을 취소하시겠어요?"라고 물었다면, 다음 "어제 주문한 것"이라는 답변을 처리할 때 이 전체 흐름이 필요합니다.
기존에는 매번 새로운 단일 질문만 보냈다면, 이제는 system 메시지 + 전체 대화 히스토리 + 새 질문을 모두 함께 보내 맥락을 유지합니다. 대화 컨텍스트 관리의 핵심 특징은 네 가지입니다.
첫째, messages 배열에 user와 assistant 메시지를 번갈아 추가하여 대화 흐름을 재현합니다. 둘째, 대화가 길어질수록 토큰 사용량이 증가하므로 비용 관리가 필요합니다.
셋째, 컨텍스트 길이 제한(예: GPT-3.5는 4096 토큰)을 초과하지 않도록 오래된 메시지를 제거해야 할 수 있습니다. 넷째, 메모리에 대화를 저장하거나 데이터베이스에 영구 저장할 수 있습니다.
이러한 특징들이 실시간 챗봇 서비스의 핵심 로직을 구성합니다.
코드 예제
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 대화 히스토리를 저장할 리스트
conversation_history = [
{
"role": "system",
"content": "당신은 도움이 되는 Python 프로그래밍 조수입니다."
}
]
def chat(user_message):
# 사용자 메시지 추가
conversation_history.append({
"role": "user",
"content": user_message
})
# API 호출 (전체 대화 히스토리 포함)
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=conversation_history,
temperature=0.7
)
# AI 응답 추출 및 히스토리에 추가
assistant_message = response.choices[0].message.content
conversation_history.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
# 연속 대화 테스트
print("Bot:", chat("내 이름은 철수야"))
print("\nBot:", chat("내 이름이 뭐였지?")) # AI가 "철수"라고 기억함
print("\n대화 길이:", len(conversation_history), "메시지")
설명
이것이 하는 일: 이 코드는 상태가 없는(stateless) API를 상태가 있는(stateful) 대화로 만드는 핵심 패턴을 보여줍니다. 웹 애플리케이션의 세션 관리와 비슷한 개념입니다.
첫 번째로, 전역 변수 conversation_history를 만들어 대화의 모든 메시지를 저장합니다. 이 리스트는 시스템 메시지로 시작하며, 이후 user와 assistant 메시지가 번갈아 추가됩니다.
리스트를 사용하는 이유는 순서가 중요하기 때문입니다. AI는 메시지의 순서대로 대화를 해석합니다.
그 다음으로 chat() 함수를 만들어 대화 로직을 캡슐화합니다. 이 함수는 사용자 메시지를 받아 히스토리에 추가하고, 전체 히스토리를 API에 전달한 후, AI 응답을 다시 히스토리에 추가합니다.
이 패턴이 핵심입니다: append user message → API call with full history → append assistant message. 이 세 단계를 반복하면 무한히 긴 대화를 이어갈 수 있습니다.
세 번째로 API 호출 시 messages=conversation_history로 전체 대화를 전달합니다. 첫 번째 호출에는 system + user 메시지 2개만 있지만, 두 번째 호출에는 system + user + assistant + user 메시지 4개가, 세 번째 호출에는 6개가 전달됩니다.
이렇게 매번 증가하는 것이 정상입니다. 마지막으로 응답을 받은 후 conversation_history.append()로 assistant의 답변을 히스토리에 추가합니다.
이 단계를 빠뜨리면 AI가 자기 자신이 한 말을 기억하지 못해 대화가 어색해집니다. 예를 들어 AI가 "A 방법과 B 방법이 있습니다"라고 했는데 사용자가 "A 방법 자세히 알려줘"라고 하면, AI의 이전 답변이 히스토리에 있어야 "A 방법"이 무엇인지 알 수 있습니다.
여러분이 이 코드를 사용하면 간단한 챗봇 애플리케이션을 만들 수 있습니다. Flask나 FastAPI로 웹 서버를 만들 때는 사용자별로 별도의 conversation_history를 관리해야 합니다.
보통 세션 ID나 사용자 ID를 키로 하는 딕셔너리에 각 사용자의 대화를 저장합니다. Redis 같은 캐시나 PostgreSQL 같은 데이터베이스에 저장하면 서버 재시작 후에도 대화를 이어갈 수 있습니다.
실전 팁
💡 대화가 길어지면 토큰 제한을 초과할 수 있습니다. GPT-3.5-turbo는 4096 토큰 제한이 있으므로, 대화가 일정 길이를 넘으면 오래된 메시지를 제거하세요. 단, system 메시지는 항상 유지해야 합니다: conversation_history = [conversation_history[0]] + conversation_history[-10:] 형식으로 최근 10개만 남깁니다.
💡 토큰 수를 추적하려면 tiktoken 라이브러리를 사용하세요. import tiktoken; enc = tiktoken.encoding_for_model("gpt-3.5-turbo"); num_tokens = len(enc.encode(text))로 메시지의 토큰 수를 미리 계산할 수 있습니다.
💡 웹 애플리케이션에서는 사용자별 대화를 분리하세요. conversations = {} 딕셔너리에 conversations[user_id] = [...] 형식으로 저장하면 여러 사용자가 동시에 사용해도 대화가 섞이지 않습니다.
💡 대화 요약 기능을 구현하면 토큰을 절약할 수 있습니다. 20개 이상의 메시지가 쌓이면 오래된 10개를 GPT로 요약하고, 요약본 1개 + 최근 메시지 10개만 유지하는 방식입니다.
💡 중요한 정보는 system 메시지에 추가하세요. 사용자가 "내 이름은 철수야"라고 했다면, 이를 감지하여 system 메시지를 "사용자 이름: 철수"로 업데이트하면 토큰을 절약하면서도 정보를 유지할 수 있습니다.
6. 스트리밍 응답 구현
시작하며
여러분이 긴 답변을 기다릴 때 화면이 멈춘 것처럼 아무 반응이 없다가 몇 초 후에 갑자기 전체 텍스트가 나타나면 답답하고 불안합니다. 사용자는 "지금 제대로 작동하는 건가?"라고 의심하게 됩니다.
실제로 많은 사용자 조사에서 응답 지연이 길 때 중간 피드백이 없으면 이탈률이 크게 증가한다는 결과가 나왔습니다. ChatGPT나 Claude 같은 최신 AI 서비스들이 모두 타이핑하듯이 한 단어씩 출력하는 이유가 바로 이것입니다.
바로 이럴 때 필요한 것이 스트리밍 응답(streaming response)입니다. API 호출 시 stream=True 옵션을 사용하면 전체 응답을 한 번에 받는 대신, 생성되는 즉시 작은 조각들을 연속적으로 받아 실시간으로 표시할 수 있습니다.
개요
간단히 말해서, 스트리밍 응답은 AI가 텍스트를 생성하는 동안 완성된 부분을 즉시 전달받아 표시하는 방식입니다. 이 방식이 중요한 이유는 사용자 경험을 크게 개선하기 때문입니다.
긴 에세이나 코드를 생성할 때 전체 완성까지 10-20초가 걸릴 수 있는데, 스트리밍을 사용하면 0.5초 만에 첫 단어가 나타나기 시작합니다. 예를 들어, 고객 상담 봇에서 복잡한 질문에 대한 긴 답변을 줄 때 스트리밍을 사용하면 고객이 답변을 읽기 시작하는 동안 나머지 부분이 생성되어 실제 대기 시간이 크게 줄어듭니다.
기존에는 response = client.chat.completions.create(...) 형식으로 전체 응답을 한 번에 받았다면, 이제는 for chunk in client.chat.completions.create(..., stream=True): 형식으로 조각들을 순차적으로 받습니다. 스트리밍 응답의 핵심 특징은 네 가지입니다.
첫째, 첫 번째 토큰까지의 시간(Time To First Token, TTFT)이 매우 짧아 즉각적인 피드백을 줍니다. 둘째, 전체 응답 시간은 동일하지만 체감 속도가 훨씬 빠릅니다.
셋째, 각 청크(chunk)는 델타(delta) 형태로 도착하며 이를 누적해야 전체 메시지가 됩니다. 넷째, 웹소켓이나 Server-Sent Events(SSE)와 조합하면 실시간 웹 챗봇을 만들 수 있습니다.
이러한 특징들이 현대적인 대화형 AI 서비스의 필수 요소입니다.
코드 예제
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def chat_with_streaming(user_message):
# stream=True 옵션으로 스트리밍 활성화
stream = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "당신은 친절한 AI 조수입니다."},
{"role": "user", "content": user_message}
],
stream=True, # 핵심: 스트리밍 활성화
temperature=0.7
)
print("AI: ", end="", flush=True) # 줄바꿈 없이 출력 시작
full_response = ""
# 청크를 하나씩 받아 처리
for chunk in stream:
# 델타에 content가 있으면 출력
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
print(content, end="", flush=True) # 실시간 출력
full_response += content # 전체 응답 누적
print() # 마지막 줄바꿈
return full_response
# 테스트
response = chat_with_streaming("Python의 장점 5가지를 설명해줘")
print(f"\n\n전체 응답 길이: {len(response)} 글자")
설명
이것이 하는 일: 이 코드는 AI의 응답을 마치 사람이 타이핑하는 것처럼 실시간으로 표시하는 완전한 구현을 보여줍니다. ChatGPT 웹 인터페이스와 동일한 경험을 제공합니다.
첫 번째로, API 호출 시 stream=True 매개변수를 추가합니다. 이 옵션을 켜면 반환 타입이 완전한 응답 객체가 아니라 청크들의 스트림(iterator)이 됩니다.
각 청크는 생성되는 즉시 네트워크를 통해 전달되므로 첫 번째 청크를 받는 시간이 매우 짧습니다. 일반적으로 100-300ms 정도면 첫 단어가 도착하기 시작합니다.
그 다음으로 for chunk in stream: 루프로 청크들을 순차적으로 처리합니다. 각 청크는 ChatCompletionChunk 객체인데, 일반 응답과 다르게 message 대신 delta 속성을 가집니다.
delta는 "변화량"이라는 뜻으로, 이전 청크 이후 새로 생성된 텍스트 조각만 포함합니다. chunk.choices[0].delta.content를 확인하여 실제 텍스트가 있는지 검사해야 하는데, 첫 번째와 마지막 청크는 메타데이터만 있고 content가 None일 수 있습니다.
세 번째로 print(..., end="", flush=True)를 사용하여 실시간 출력을 구현합니다. end=""는 줄바꿈을 하지 않고 같은 줄에 계속 추가하라는 의미이고, flush=True는 버퍼를 즉시 비워 화면에 표시하라는 의미입니다.
이 두 옵션이 없으면 청크를 받아도 화면에 바로 나타나지 않습니다. 마지막으로 full_response += content로 모든 청크를 누적하여 전체 응답을 저장합니다.
스트리밍은 표시 방식일 뿐이고, 나중에 대화 히스토리에 저장하거나 처리하려면 완전한 텍스트가 필요합니다. 스트리밍이 끝난 후 full_response를 conversation_history에 추가하면 됩니다.
여러분이 이 코드를 사용하면 터미널 기반 챗봇을 만들 수 있고, 웹 애플리케이션에서는 FastAPI의 StreamingResponse나 Flask의 stream_with_context와 조합하여 브라우저에서 실시간으로 텍스트가 나타나게 할 수 있습니다. WebSocket을 사용하면 더욱 실시간성이 높은 채팅 경험을 제공할 수 있습니다.
실전 팁
💡 웹 애플리케이션에서 스트리밍을 구현하려면 Server-Sent Events(SSE)를 사용하세요. FastAPI에서는 StreamingResponse를 반환하고, 프론트엔드에서는 EventSource API로 받습니다. 예: return StreamingResponse(generate(), media_type="text/event-stream")
💡 스트리밍 중에는 finish_reason을 확인할 수 없으므로 마지막 청크에서 확인해야 합니다. if chunk.choices[0].finish_reason == "stop"으로 정상 완료를 검증하세요.
💡 에러 처리가 중요합니다. 스트리밍 중 네트워크 오류가 발생하면 일부 청크만 받고 중단될 수 있습니다. try-except 블록으로 감싸고, 중단 시 "응답이 중단되었습니다"같은 메시지를 표시하세요.
💡 토큰 사용량은 마지막 청크에서만 제공됩니다. 스트리밍 중에는 usage 정보가 없으므로, 비용 추적이 필요하면 마지막 청크를 별도로 처리하거나 tiktoken으로 사후 계산해야 합니다.
💡 함수 호출(function calling)과 스트리밍을 함께 사용할 때는 delta.function_call을 확인하세요. 함수 인자가 JSON 형식으로 조각조각 도착하므로 완전한 JSON이 될 때까지 누적한 후 파싱해야 합니다.
7. 토큰 사용량 추적
시작하며
여러분이 OpenAI API를 사용하다가 월말에 청구서를 보고 깜짝 놀란 경험이 있나요? 개발 중에 무심코 긴 프롬프트를 수천 번 호출했거나, GPT-4를 테스트용으로 써서 예상보다 10배 높은 비용이 나올 수 있습니다.
실제로 많은 스타트업과 개발자들이 비용 모니터링 없이 서비스를 런칭했다가 사용자가 급증하면서 하루에 수백 달러의 API 비용이 발생하여 당황하는 사례가 있습니다. 특히 악의적인 사용자가 과도하게 API를 호출하는 경우도 있습니다.
바로 이럴 때 필요한 것이 토큰 사용량 추적입니다. 매 API 호출마다 사용된 토큰 수를 기록하고 분석하면 비용을 예측하고, 최적화가 필요한 부분을 찾아내며, 사용자별 사용량을 제한할 수 있습니다.
개요
간단히 말해서, 토큰 사용량 추적은 API 응답에 포함된 usage 정보를 수집하고 분석하여 비용을 관리하는 기법입니다. 토큰 추적이 중요한 이유는 OpenAI 비용이 토큰 수에 비례하기 때문입니다.
GPT-3.5-turbo는 1000 토큰당 약 $0.002, GPT-4는 입력 $0.03/출력 $0.06로 큰 차이가 있습니다. 예를 들어, 1000명의 사용자가 하루 평균 10번씩 대화하고 매번 500 토큰을 사용한다면 GPT-4 기준 하루 $250, 한 달에 $7,500의 비용이 발생합니다.
이를 미리 알고 대응하지 않으면 서비스 운영이 불가능해집니다. 기존에는 그냥 API를 호출하고 응답만 사용했다면, 이제는 response.usage 객체에서 토큰 정보를 추출하여 로깅, 데이터베이스 저장, 알림 등의 처리를 합니다.
토큰 사용량 추적의 핵심 특징은 네 가지입니다. 첫째, prompt_tokens(입력), completion_tokens(출력), total_tokens(합계)를 각각 추적할 수 있습니다.
둘째, 사용자별, 기능별, 시간대별로 집계하여 패턴을 분석할 수 있습니다. 셋째, 일일/월간 한도를 설정하여 예산 초과를 방지할 수 있습니다.
넷째, 토큰 수를 미리 예측하여 비용을 사전에 추정할 수 있습니다. 이러한 특징들이 지속 가능한 AI 서비스 운영의 기반이 됩니다.
코드 예제
from openai import OpenAI
import os
from dotenv import load_dotenv
from datetime import datetime
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class TokenTracker:
def __init__(self):
self.total_tokens = 0
self.total_cost = 0.0
self.requests = []
# GPT-3.5-turbo 가격 (1000토큰당)
self.price_per_1k_tokens = 0.002
def track(self, response, model="gpt-3.5-turbo"):
usage = response.usage
tokens = usage.total_tokens
cost = (tokens / 1000) * self.price_per_1k_tokens
self.total_tokens += tokens
self.total_cost += cost
# 요청 기록 저장
self.requests.append({
"timestamp": datetime.now(),
"model": model,
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": tokens,
"cost": cost
})
print(f"[토큰] 입력: {usage.prompt_tokens}, "
f"출력: {usage.completion_tokens}, "
f"합계: {tokens}, 비용: ${cost:.4f}")
def summary(self):
print(f"\n=== 사용량 요약 ===")
print(f"총 요청 수: {len(self.requests)}")
print(f"총 토큰: {self.total_tokens:,}")
print(f"총 비용: ${self.total_cost:.4f}")
if self.requests:
avg_tokens = self.total_tokens / len(self.requests)
print(f"평균 토큰/요청: {avg_tokens:.0f}")
# 사용 예시
tracker = TokenTracker()
response1 = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello!"}]
)
tracker.track(response1)
response2 = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Python의 장점 5가지"}]
)
tracker.track(response2)
tracker.summary()
설명
이것이 하는 일: 이 코드는 모든 API 호출의 토큰 사용량을 자동으로 추적하고 비용을 계산하는 완전한 시스템을 보여줍니다. 프로덕션 환경에서 필수적인 기능입니다.
첫 번째로, TokenTracker 클래스를 만들어 추적 로직을 캡슐화합니다. __init__ 메서드에서 누적 토큰 수, 누적 비용, 개별 요청 기록을 저장할 변수를 초기화합니다.
price_per_1k_tokens에는 사용하는 모델의 가격을 설정하는데, GPT-3.5-turbo는 0.002, GPT-4는 입력 0.03/출력 0.06으로 다르므로 모델에 따라 조정해야 합니다. 그 다음으로 track() 메서드에서 API 응답을 받아 토큰 정보를 추출합니다.
response.usage 객체는 세 가지 속성을 가집니다: prompt_tokens(우리가 보낸 메시지의 토큰 수), completion_tokens(AI가 생성한 응답의 토큰 수), total_tokens(둘의 합). 비용은 총 토큰을 1000으로 나누고 단가를 곱하여 계산합니다.
이를 누적 변수에 더하고 개별 기록도 리스트에 저장합니다. 세 번째로 각 요청 기록에 타임스탬프, 모델명, 토큰 세부 정보를 함께 저장합니다.
이렇게 하면 나중에 "어떤 시간대에 사용량이 많은지", "어떤 기능이 토큰을 많이 쓰는지" 분석할 수 있습니다. 실제 서비스에서는 이 데이터를 PostgreSQL이나 MongoDB 같은 데이터베이스에 저장하여 영구 보관하고 대시보드로 시각화합니다.
마지막으로 summary() 메서드로 전체 통계를 출력합니다. 총 요청 수, 총 토큰, 총 비용, 평균 토큰 등을 보여주어 사용 패턴을 파악할 수 있습니다.
예를 들어 평균 토큰이 1000을 넘는다면 프롬프트가 너무 길거나 응답 길이 제한이 없는 것이므로 최적화가 필요합니다. 여러분이 이 코드를 사용하면 개발 중에는 비용을 실시간으로 모니터링하고, 프로덕션에서는 사용자별 한도를 설정하거나 일일 예산을 초과하면 알림을 보내는 시스템을 만들 수 있습니다.
예를 들어 if tracker.total_cost > 10.0: send_alert("일일 예산 $10 초과!")처럼 임계값 알림을 추가할 수 있습니다.
실전 팁
💡 GPT-4는 입력과 출력 가격이 다르므로 별도로 계산하세요. cost = (usage.prompt_tokens / 1000 * 0.03) + (usage.completion_tokens / 1000 * 0.06) 형식으로 정확한 비용을 계산할 수 있습니다.
💡 사용자별 사용량을 추적하려면 딕셔너리로 관리하세요. user_trackers = {}; user_trackers[user_id] = TokenTracker() 형식으로 각 사용자의 추적기를 분리하면 남용을 감지하고 한도를 적용할 수 있습니다.
💡 tiktoken 라이브러리로 API 호출 전에 토큰을 예측하세요. import tiktoken; enc = tiktoken.encoding_for_model("gpt-3.5-turbo"); num_tokens = len(enc.encode(text))로 사전 계산하면 비용 초과를 방지할 수 있습니다.
💡 로그를 파일이나 데이터베이스에 저장하세요. with open("token_usage.log", "a") as f: f.write(json.dumps(record) + "\n") 형식으로 저장하면 나중에 분석하거나 청구서와 대조할 수 있습니다.
💡 시간대별 분석을 위해 pandas를 사용하세요. df = pd.DataFrame(tracker.requests); daily = df.groupby(df.timestamp.dt.date).agg({"total_tokens": "sum", "cost": "sum"}) 형식으로 일별 집계를 쉽게 만들 수 있습니다.
8. 에러 핸들링
시작하며
여러분이 API를 호출했는데 네트워크가 불안정하거나, API 키가 만료되었거나, 할당량을 초과했을 때 어떻게 대응하시나요? 에러 처리 없이 그냥 프로그램이 멈추면 사용자는 무슨 일이 일어났는지 알 수 없고, 개발자도 디버깅이 어렵습니다.
실제 프로덕션 환경에서는 다양한 종류의 오류가 발생합니다. OpenAI 서버의 일시적 장애, 사용자의 인터넷 연결 끊김, 잘못된 요청 형식, 할당량 초과, API 키 문제 등입니다.
각각 다른 방식으로 처리해야 하는데 많은 초보자들이 모든 에러를 동일하게 처리하거나 아예 무시합니다. 바로 이럴 때 필요한 것이 체계적인 에러 핸들링입니다.
OpenAI 라이브러리는 다양한 예외 타입을 제공하므로, 각 상황에 맞는 적절한 대응(재시도, 사용자 안내, 로깅 등)을 할 수 있습니다.
개요
간단히 말해서, 에러 핸들링은 API 호출 중 발생할 수 있는 다양한 예외 상황을 감지하고 적절히 대응하는 기법입니다. 에러 핸들링이 중요한 이유는 안정적인 서비스 운영을 위해 필수적이기 때문입니다.
사용자가 1000명이고 하루 평균 API 호출이 10000번이라면, 0.1%의 오류율만 해도 하루 10번의 에러가 발생합니다. 이를 제대로 처리하지 않으면 사용자는 에러 메시지도 보지 못한 채 답답해하고, 개발자는 왜 실패했는지 알 수 없습니다.
예를 들어, 네트워크 지연으로 타임아웃이 발생했다면 자동 재시도가 적절하지만, API 키가 잘못되었다면 재시도는 의미가 없고 관리자에게 알림을 보내야 합니다. 기존에는 단순히 try-except Exception으로 모든 에러를 잡았다면, 이제는 APIError, RateLimitError, APIConnectionError 등 구체적인 예외 타입별로 다른 처리를 합니다.
에러 핸들링의 핵심 특징은 네 가지입니다. 첫째, OpenAI 라이브러리는 상황별로 다른 예외 클래스를 제공합니다.
둘째, 재시도 가능한 에러(네트워크 문제)와 치명적인 에러(인증 실패)를 구분해야 합니다. 셋째, 사용자에게는 친절한 메시지를, 로그에는 상세한 기술 정보를 기록합니다.
넷째, 지수 백오프(exponential backoff) 전략으로 재시도 간격을 점진적으로 늘립니다. 이러한 특징들이 안정적이고 유지보수하기 쉬운 AI 애플리케이션의 기반입니다.
코드 예제
from openai import OpenAI, APIError, RateLimitError, APIConnectionError, AuthenticationError
import os
from dotenv import load_dotenv
import time
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def chat_with_retry(user_message, max_retries=3):
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": user_message}],
timeout=30.0
)
return response.choices[0].message.content
except AuthenticationError as e:
# API 키 문제 - 재시도 불가
print(f"인증 오류: {e}")
print("API 키를 확인하세요.")
return None
except RateLimitError as e:
# 할당량 초과 - 재시도 가능
wait_time = 2 ** attempt # 지수 백오프: 1초, 2초, 4초
print(f"할당량 초과. {wait_time}초 후 재시도... (시도 {attempt + 1}/{max_retries})")
if attempt < max_retries - 1:
time.sleep(wait_time)
else:
print("최대 재시도 횟수 초과")
return None
except APIConnectionError as e:
# 네트워크 오류 - 재시도 가능
print(f"네트워크 오류: {e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
else:
print("네트워크 연결 실패")
return None
except APIError as e:
# 기타 API 오류
print(f"API 오류: {e.status_code} - {e.message}")
return None
except Exception as e:
# 예상치 못한 오류
print(f"예상치 못한 오류: {type(e).__name__} - {e}")
return None
return None
# 테스트
result = chat_with_retry("안녕하세요!")
if result:
print(f"성공: {result}")
else:
print("실패: 응답을 받지 못했습니다")
설명
이것이 하는 일: 이 코드는 API 호출 중 발생할 수 있는 모든 주요 예외를 포착하고 각각에 맞는 전략으로 대응하는 완전한 에러 핸들링 시스템을 보여줍니다. 프로덕션 환경에서 필수적인 패턴입니다.
첫 번째로, chat_with_retry() 함수는 최대 재시도 횟수를 받아 여러 번 시도할 수 있게 합니다. for attempt in range(max_retries): 루프 안에 전체 API 호출 로직을 넣어, 실패하면 다음 시도로 넘어가고 성공하면 즉시 반환합니다.
이 패턴은 일시적인 네트워크 문제를 자동으로 해결해줍니다. 그 다음으로 예외를 구체적인 타입별로 처리합니다.
AuthenticationError는 API 키가 잘못되었거나 만료된 경우로, 재시도해도 소용없으므로 즉시 None을 반환하고 사용자에게 키 확인을 안내합니다. RateLimitError는 분당/일일 할당량을 초과한 경우로, 잠시 기다리면 해결되므로 지수 백오프 전략으로 재시도합니다.
1번째 시도는 1초, 2번째는 2초, 3번째는 4초 대기하여 서버 부하를 줄입니다. 세 번째로 APIConnectionError는 네트워크 연결 문제로, 사용자의 인터넷이 끊겼거나 OpenAI 서버에 연결할 수 없는 경우입니다.
이것도 일시적일 수 있으므로 재시도하지만, 3번 모두 실패하면 명확한 메시지와 함께 중단합니다. APIError는 잘못된 요청 형식, 모델명 오류, 토큰 초과 등 다양한 API 수준 오류를 포괄하며, status_code와 message로 상세 정보를 확인할 수 있습니다.
마지막으로 except Exception으로 예상치 못한 모든 오류를 처리합니다. 이는 라이브러리 버그나 Python 자체 오류를 잡는 안전망입니다.
type(e).__name__으로 예외 클래스 이름을 출력하여 디버깅에 도움을 줍니다. 실제 프로덕션에서는 여기서 Sentry 같은 오류 추적 서비스로 보고하거나, 로그 파일에 스택 트레이스를 기록해야 합니다.
여러분이 이 코드를 사용하면 불안정한 네트워크 환경에서도 안정적으로 동작하는 애플리케이션을 만들 수 있습니다. 웹 서비스에서는 각 에러 타입에 맞는 HTTP 상태 코드를 반환하고(예: RateLimitError → 429 Too Many Requests, AuthenticationError → 401 Unauthorized), 모바일 앱에서는 사용자에게 적절한 안내 메시지를 표시할 수 있습니다.
실전 팁
💡 재시도 로직을 간단히 하려면 tenacity 라이브러리를 사용하세요. @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)) 데코레이터로 복잡한 재시도 로직을 한 줄로 구현할 수 있습니다.
💡 타임아웃을 반드시 설정하세요. OpenAI(timeout=30.0)으로 전역 타임아웃을 설정하거나, 각 API 호출마다 설정할 수 있습니다. 설정하지 않으면 네트워크 문제 시 무한 대기할 수 있습니다.
💡 오류를 로깅할 때는 민감한 정보를 제외하세요. API 키나 사용자 메시지 전체를 로그에 남기면 보안 문제가 됩니다. 대신 요청 ID, 타임스탬프, 에러 타입만 기록하세요.
💡 Circuit Breaker 패턴을 구현하면 연속적인 실패 시 일정 시간 동안 API 호출을 중단하여 불필요한 비용을 절약할 수 있습니다. pybreaker 라이브러리가 유용합니다.
💡 사용자에게 보여주는 메시지와 로그 메시지를 분리하세요. 사용자에게는 "일시적인 오류가 발생했습니다. 잠시 후 다시 시도해주세요" 같은 친절한 메시지를, 로그에는 f"RateLimitError: {e.response.headers.get('retry-after')} seconds" 같은 상세 정보를 기록하세요.