이미지 로딩 중...
AI Generated
2025. 11. 12. · 2 Views
바닥부터 만드는 ChatGPT 명령줄 인터페이스 구축 가이드
ChatGPT를 웹 브라우저 없이 터미널에서 직접 사용할 수 있는 CLI 도구를 처음부터 만들어보는 실전 가이드입니다. Python으로 API 통신부터 스트리밍 응답 처리까지 단계별로 구현합니다.
목차
- CLI 프로젝트 기본 구조 설정 - 체계적인 시작이 성공의 반
- 환경 변수와 API 키 안전하게 관리하기 - 보안의 첫걸음
- OpenAI API 클라이언트 초기화 - 통신의 시작점
- 기본 Chat Completions API 호출 - 첫 번째 대화
- 스트리밍 응답 구현 - 실시간 타이핑 효과
- 커맨드라인 인터페이스 기본 구조 - 사용자와의 대화창
- 스트리밍 응답을 CLI에 통합하기 - 실시간 출력 연결
- 대화 히스토리 관리 - 맥락 유지의 핵심
- 특수 명령어 구현 - CLI를 더 강력하게
- 에러 처리와 재시도 로직 - 안정성 확보
- 시스템 메시지로 AI 행동 제어하기 - 맞춤형 어시스턴트
- 대화 저장 및 불러오기 - 세션 영속성
1. CLI 프로젝트 기본 구조 설정 - 체계적인 시작이 성공의 반
시작하며
여러분이 새로운 CLI 도구를 만들 때 파일을 어떻게 구성해야 할지 막연하게 느껴진 적 있나요? 특히 나중에 기능을 추가하거나 다른 사람과 협업할 때를 고려하면 처음부터 제대로 된 구조를 잡는 것이 정말 중요합니다.
많은 초보 개발자들이 모든 코드를 한 파일에 때려박고 시작하는데, 이렇게 하면 나중에 유지보수가 정말 힘들어집니다. 코드가 200줄만 넘어가도 어디서 뭘 고쳐야 할지 찾기 어렵죠.
바로 이럴 때 필요한 것이 모듈화된 프로젝트 구조입니다. API 통신, 설정 관리, 사용자 인터페이스를 명확히 분리하면 나중에 얼마나 편한지 실감하게 될 겁니다.
개요
간단히 말해서, CLI 프로젝트 구조는 코드를 역할별로 나누어 관리하는 방법입니다. 실무에서는 작은 프로젝트라도 처음부터 확장 가능한 구조로 시작하는 것이 중요합니다.
예를 들어, 나중에 GPT-4에서 GPT-5로 모델을 변경하거나 새로운 기능을 추가할 때, 잘 정돈된 구조라면 어느 파일을 수정해야 하는지 즉시 알 수 있습니다. 기존에는 main.py 하나에 모든 것을 작성했다면, 이제는 config.py(설정), api.py(API 통신), cli.py(사용자 인터페이스)로 분리할 수 있습니다.
핵심 특징은 관심사의 분리(Separation of Concerns), 재사용성, 테스트 용이성입니다. 이러한 특징들이 결국 개발 속도와 코드 품질을 결정짓습니다.
코드 예제
# 프로젝트 구조
chatgpt-cli/
├── src/
│ ├── __init__.py # 패키지 초기화
│ ├── config.py # API 키, 설정 관리
│ ├── api.py # OpenAI API 통신
│ ├── cli.py # 사용자 인터페이스
│ └── utils.py # 유틸리티 함수
├── tests/ # 테스트 코드
├── requirements.txt # 의존성 목록
└── main.py # 진입점
# requirements.txt
openai>=1.0.0
python-dotenv>=1.0.0
rich>=13.0.0 # 터미널 UI 라이브러리
설명
이것이 하는 일: 프로젝트를 논리적인 모듈로 나누어 각 파일이 명확한 책임을 갖도록 합니다. 첫 번째로, src/ 디렉토리 안에 모든 소스 코드를 모읍니다.
이렇게 하는 이유는 프로젝트 루트를 깔끔하게 유지하고, 실제 코드와 설정 파일을 구분하기 위함입니다. init.py 파일은 Python에게 이 디렉토리가 패키지라고 알려주는 역할을 합니다.
그 다음으로, 각 모듈이 담당하는 역할을 명확히 합니다. config.py는 환경 변수와 API 키를 관리하고, api.py는 OpenAI와의 모든 통신을 담당하며, cli.py는 사용자와의 상호작용을 처리합니다.
이렇게 분리하면 예를 들어 API 키 관리 방식을 바꾸고 싶을 때 config.py만 수정하면 됩니다. 마지막으로, requirements.txt에 필요한 라이브러리를 명시합니다.
openai는 API 통신용, python-dotenv는 환경 변수 관리용, rich는 터미널에서 예쁜 출력을 위한 것입니다. 이 파일이 있으면 다른 개발자가 pip install -r requirements.txt 한 줄로 모든 의존성을 설치할 수 있습니다.
여러분이 이 구조를 사용하면 코드 검색이 빨라지고, 팀원이 코드를 이해하는 시간이 줄어들며, 버그가 발생했을 때 어디를 봐야 할지 명확해집니다. 특히 나중에 Slack 봇이나 웹 API로 확장할 때도 기존 api.py를 그대로 재사용할 수 있습니다.
실전 팁
💡 src/ 디렉토리를 사용하면 테스트 코드에서 from src.api import ChatGPT처럼 명확하게 import할 수 있어 네임스페이스 충돌을 방지할 수 있습니다.
💡 .env 파일에 API 키를 저장하되 반드시 .gitignore에 추가하세요. 실수로 GitHub에 올리면 몇 분 안에 봇이 크롤링해서 악용할 수 있습니다.
💡 각 모듈의 맨 위에 docstring으로 그 파일의 역할을 명시해두면 나중에 코드를 다시 볼 때 맥락을 빠르게 파악할 수 있습니다.
💡 utils.py는 처음부터 만들지 말고, 같은 코드가 2-3번 반복될 때 그때 만드세요. 너무 일찍 추상화하면 오히려 복잡해집니다.
2. 환경 변수와 API 키 안전하게 관리하기 - 보안의 첫걸음
시작하며
여러분이 코드를 GitHub에 올렸는데 몇 시간 뒤 OpenAI로부터 "비정상적인 API 사용이 감지되었습니다"라는 이메일을 받은 적이 있나요? 실제로 많은 개발자들이 실수로 API 키를 소스 코드에 하드코딩해서 올리고, 그 결과 수백 달러의 요금 폭탄을 맞습니다.
이런 문제는 코드와 설정을 분리하지 않아서 발생합니다. API 키는 민감한 정보인데, 코드와 함께 버전 관리 시스템에 들어가면 영원히 git 히스토리에 남게 되죠.
심지어 나중에 커밋을 삭제해도 이미 크롤링된 상태일 수 있습니다. 바로 이럴 때 필요한 것이 환경 변수 관리입니다.
.env 파일과 python-dotenv를 사용하면 민감한 정보를 안전하게 분리할 수 있고, 개발/스테이징/프로덕션 환경마다 다른 설정을 쉽게 적용할 수 있습니다.
개요
간단히 말해서, 환경 변수 관리는 API 키 같은 민감한 정보를 코드 밖으로 빼내어 안전하게 보관하는 방법입니다. 실무에서는 절대로 API 키를 코드에 직접 작성하지 않습니다.
예를 들어, 팀 프로젝트에서 각자 다른 OpenAI 계정을 사용하거나, 로컬 개발과 서버 배포 시 다른 설정을 쓰는 경우가 매우 흔합니다. 환경 변수를 사용하면 코드는 하나인데 환경에 따라 동작을 바꿀 수 있습니다.
기존에는 api_key = "sk-..."처럼 하드코딩했다면, 이제는 api_key = os.getenv("OPENAI_API_KEY")로 외부에서 주입받을 수 있습니다. 핵심 특징은 보안성, 환경 독립성, 설정 중앙화입니다.
특히 12-Factor App 방법론에서도 설정을 환경 변수로 관리하는 것을 강력히 권장합니다.
코드 예제
# config.py
import os
from dotenv import load_dotenv
# .env 파일에서 환경 변수 로드
load_dotenv()
class Config:
"""애플리케이션 설정을 관리하는 클래스"""
# API 키 가져오기 (없으면 에러 발생)
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY가 설정되지 않았습니다")
# 기본 설정값들
MODEL = os.getenv("MODEL", "gpt-3.5-turbo") # 기본값 제공
MAX_TOKENS = int(os.getenv("MAX_TOKENS", "1000"))
TEMPERATURE = float(os.getenv("TEMPERATURE", "0.7"))
설명
이것이 하는 일: .env 파일에서 설정값을 읽어와 애플리케이션 전체에서 사용할 수 있도록 Config 클래스로 제공합니다. 첫 번째로, load_dotenv()가 실행되면서 프로젝트 루트의 .env 파일을 찾아 모든 변수를 환경 변수로 로드합니다.
이 과정은 자동으로 이루어지며, .env 파일이 없어도 에러가 나지 않습니다(이미 시스템 환경 변수가 있을 수 있으니까요). 그 다음으로, os.getenv()로 각 환경 변수를 가져옵니다.
중요한 것은 OPENAI_API_KEY처럼 필수 값은 없을 때 즉시 에러를 발생시키고, MODEL처럼 선택적인 값은 기본값을 제공한다는 점입니다. 이렇게 하면 애플리케이션이 시작하자마자 설정 문제를 발견할 수 있습니다.
마지막으로, Config 클래스로 감싸서 타입 안전성을 확보하고 중앙화된 설정 관리를 구현합니다. 코드 어디서든 from config import Config로 import해서 Config.OPENAI_API_KEY처럼 접근할 수 있습니다.
나중에 설정을 바꾸고 싶으면 .env 파일만 수정하면 되고, 코드는 전혀 건드릴 필요가 없습니다. 여러분이 이 패턴을 사용하면 팀원마다 각자의 .env 파일을 가질 수 있고, 실수로 API 키를 커밋할 위험이 사라지며, 배포 환경에서는 시스템 환경 변수를 사용하고 로컬에서는 .env 파일을 사용하는 유연성을 얻게 됩니다.
실전 팁
💡 .env.example 파일을 만들어서 필요한 환경 변수 목록을 공유하세요(실제 값 없이 키 이름만). 새로운 팀원이 어떤 설정이 필요한지 바로 알 수 있습니다.
💡 로컬 개발용 .env.local, 테스트용 .env.test처럼 환경별 파일을 만들면 load_dotenv('.env.test')로 명시적으로 로드할 수 있습니다.
💡 Docker를 사용한다면 docker-compose.yml의 environment 섹션에 환경 변수를 정의하면 .env 파일 없이도 동작합니다.
💡 배포 시에는 Heroku의 Config Vars, AWS의 Systems Manager Parameter Store 같은 플랫폼 기능을 사용하는 것이 더 안전합니다.
3. OpenAI API 클라이언트 초기화 - 통신의 시작점
시작하며
여러분이 OpenAI API를 호출하려고 할 때 매번 인증 헤더를 설정하고 엔드포인트 URL을 작성하는 것이 번거로웠던 적 있나요? 특히 여러 곳에서 API를 호출해야 하면 같은 설정 코드가 계속 반복되죠.
이런 문제는 API 클라이언트를 제대로 초기화하지 않아서 발생합니다. 매번 requests.post()를 직접 호출하면 인증 로직이 흩어지고, 에러 처리도 일관성이 없어지며, 나중에 API 버전을 바꾸려면 모든 호출 코드를 다 수정해야 합니다.
바로 이럴 때 필요한 것이 OpenAI 공식 라이브러리를 사용한 클라이언트 초기화입니다. 한 번만 설정하면 나머지는 라이브러리가 알아서 처리해주고, 타입 힌트와 자동 완성까지 제공됩니다.
개요
간단히 말해서, OpenAI 클라이언트 초기화는 API 통신을 위한 준비 작업으로, 인증과 기본 설정을 한 곳에서 처리하는 것입니다. 실무에서는 OpenAI 공식 라이브러리를 사용하는 것이 필수입니다.
이 라이브러리는 인증, 재시도 로직, 타임아웃, 에러 처리를 모두 내장하고 있어서 직접 HTTP 요청을 만드는 것보다 훨씬 안전하고 편리합니다. 예를 들어, 네트워크가 일시적으로 끊겼을 때 자동으로 재시도하는 기능이 기본 탑재되어 있습니다.
기존에는 requests 라이브러리로 직접 HTTP POST를 날렸다면, 이제는 client.chat.completions.create()처럼 의미 있는 메서드를 호출할 수 있습니다. 핵심 특징은 간편한 인증, 자동 재시도, 타입 안전성입니다.
이러한 특징들이 개발 생산성을 크게 높이고 런타임 에러를 줄여줍니다.
코드 예제
# api.py
from openai import OpenAI
from config import Config
class ChatGPTClient:
"""ChatGPT API와 통신하는 클라이언트 클래스"""
def __init__(self):
# OpenAI 클라이언트 초기화 (API 키 자동 인증)
self.client = OpenAI(api_key=Config.OPENAI_API_KEY)
self.model = Config.MODEL
self.max_tokens = Config.MAX_TOKENS
self.temperature = Config.TEMPERATURE
def get_client(self):
"""초기화된 클라이언트 반환"""
return self.client
설명
이것이 하는 일: OpenAI 공식 라이브러리를 사용해서 API 통신을 위한 클라이언트 객체를 생성하고, 앱 전체에서 재사용할 수 있도록 준비합니다. 첫 번째로, ChatGPTClient 클래스의 init 메서드에서 OpenAI 객체를 생성합니다.
이때 api_key 매개변수에 Config.OPENAI_API_KEY를 전달하면 모든 후속 요청에 자동으로 인증 헤더가 포함됩니다. 내부적으로 "Authorization: Bearer sk-..." 형태의 헤더가 추가되는 것이죠.
그 다음으로, model, max_tokens, temperature 같은 기본 설정값들을 인스턴스 변수로 저장합니다. 이렇게 하면 나중에 API를 호출할 때마다 이 값들을 전달할 필요 없이 한 번만 설정하면 됩니다.
물론 필요하면 특정 호출에서 이 값들을 오버라이드할 수도 있습니다. 마지막으로, 클래스로 감싸는 이유는 단순히 클라이언트만 제공하는 것이 아니라 나중에 메시지 히스토리 관리, 토큰 카운팅, 비용 추적 같은 추가 기능을 붙이기 위함입니다.
객체 지향 설계의 장점을 살려 확장 가능한 구조를 만드는 것이죠. 여러분이 이 패턴을 사용하면 API 키를 여러 곳에 흩어놓지 않아도 되고, 설정을 한 곳에서 관리할 수 있으며, 나중에 Azure OpenAI나 다른 LLM 서비스로 교체할 때도 이 클래스만 수정하면 됩니다.
실제로 많은 기업들이 이렇게 추상화 계층을 만들어서 벤더 종속성을 줄입니다.
실전 팁
💡 클라이언트 초기화는 비용이 거의 없으니 매번 새로 만들어도 되지만, 싱글톤 패턴으로 하나만 만들어 재사용하면 조금 더 효율적입니다.
💡 timeout 매개변수를 설정하면 네트워크 문제로 무한정 기다리는 것을 방지할 수 있습니다: OpenAI(timeout=30.0).
💡 max_retries 매개변수로 재시도 횟수를 조절할 수 있습니다. 기본값은 2회인데, 안정성이 중요하면 OpenAI(max_retries=3)으로 늘리세요.
💡 프록시를 사용해야 하는 회사 환경이라면 OpenAI(http_client=httpx.Client(proxies={...}))로 커스텀 HTTP 클라이언트를 주입할 수 있습니다.
4. 기본 Chat Completions API 호출 - 첫 번째 대화
시작하며
여러분이 처음으로 ChatGPT에게 질문을 보내고 답변을 받아보려고 하는데, API 문서를 봐도 어떤 매개변수를 써야 할지 헷갈린 적 있나요? 특히 messages 배열 구조가 낯설어서 첫 호출에서 막히는 경우가 많습니다.
이런 문제는 ChatGPT API의 대화형 구조를 이해하지 못해서 발생합니다. 일반적인 REST API처럼 단순히 텍스트를 보내는 게 아니라, 역할(role)과 내용(content)을 구분해서 보내야 하는데 처음에는 이게 복잡하게 느껴지죠.
바로 이럴 때 필요한 것이 Chat Completions API의 기본 사용법입니다. messages 배열의 구조만 이해하면 단순한 질의응답부터 복잡한 멀티턴 대화까지 모두 처리할 수 있습니다.
개요
간단히 말해서, Chat Completions API는 사용자 메시지를 보내고 AI의 응답을 받는 가장 기본적인 대화 인터페이스입니다. 실무에서는 이 API가 ChatGPT의 핵심입니다.
웹사이트에서 보는 ChatGPT도 내부적으로 이 API를 사용합니다. 메시지는 "system"(AI의 행동 지침), "user"(사용자 입력), "assistant"(AI 응답) 세 가지 역할로 구성되며, 대화 맥락을 유지하려면 이전 메시지들을 모두 포함해서 보내야 합니다.
기존에는 단순한 text-in, text-out API를 상상했다면, 이제는 대화 히스토리를 관리하는 스테이트풀한 통신임을 이해해야 합니다. 핵심 특징은 역할 기반 메시지 구조, 컨텍스트 유지를 위한 히스토리 관리, 스트리밍 지원입니다.
이러한 특징들이 자연스러운 대화 경험을 가능하게 합니다.
코드 예제
# api.py (계속)
def send_message(self, user_message: str, conversation_history: list = None):
"""사용자 메시지를 보내고 응답을 받습니다"""
# 대화 히스토리 초기화 (없으면 새로 시작)
if conversation_history is None:
conversation_history = []
# 사용자 메시지 추가
conversation_history.append({
"role": "user",
"content": user_message
})
# API 호출
response = self.client.chat.completions.create(
model=self.model,
messages=conversation_history,
max_tokens=self.max_tokens,
temperature=self.temperature
)
# 응답 추출 및 히스토리에 추가
assistant_message = response.choices[0].message.content
conversation_history.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message, conversation_history
설명
이것이 하는 일: 사용자 메시지를 적절한 형식으로 포장해서 OpenAI API에 전송하고, 응답을 받아서 대화 히스토리에 추가한 뒤 반환합니다. 첫 번째로, conversation_history 리스트를 확인합니다.
이게 None이면 새로운 대화의 시작이므로 빈 리스트로 초기화합니다. 그렇지 않으면 기존 대화를 이어가는 것이므로 전달받은 히스토리를 그대로 사용합니다.
이 히스토리에는 이전 user와 assistant 메시지들이 모두 들어있어서 AI가 맥락을 이해할 수 있습니다. 그 다음으로, 새로운 사용자 메시지를 딕셔너리 형태로 만들어 히스토리에 추가합니다.
"role": "user"는 이게 사용자가 한 말이라는 것을 표시하고, "content"에 실제 메시지 내용이 들어갑니다. 그리고 client.chat.completions.create()를 호출하면서 전체 히스토리를 messages 매개변수로 전달합니다.
마지막으로, API 응답에서 텍스트를 추출합니다. response.choices[0]은 여러 응답 중 첫 번째(보통 하나만 생성됨)를 선택하고, message.content가 실제 AI가 생성한 텍스트입니다.
이것도 히스토리에 "role": "assistant"로 추가해서 반환하면, 다음 호출 시 이 대화가 이어집니다. 여러분이 이 함수를 사용하면 대화 맥락이 자동으로 유지되고, 같은 함수를 반복 호출하기만 하면 멀티턴 대화가 구현되며, 나중에 대화를 파일로 저장하거나 데이터베이스에 넣을 때도 conversation_history를 그대로 사용할 수 있습니다.
실제 프로덕션 챗봇들도 이 패턴을 기반으로 만들어집니다.
실전 팁
💡 시스템 메시지를 맨 앞에 추가하면 AI의 행동을 제어할 수 있습니다: [{"role": "system", "content": "너는 친절한 코딩 선생님이야"}].
💡 토큰 제한(GPT-3.5는 4K, GPT-4는 8K/32K)을 넘지 않도록 오래된 대화는 잘라내야 합니다. 최근 N개 메시지만 유지하는 슬라이딩 윈도우 방식을 쓰세요.
💡 response.usage 속성을 보면 prompt_tokens, completion_tokens 정보가 있어서 비용 계산과 디버깅에 유용합니다.
💡 에러 처리를 위해 try-except로 openai.APIError를 잡아서 네트워크 문제나 요금 한도 초과 같은 상황을 사용자에게 친절하게 안내하세요.
5. 스트리밍 응답 구현 - 실시간 타이핑 효과
시작하며
여러분이 ChatGPT 웹사이트를 사용할 때 답변이 한 글자씩 타이핑되는 것을 본 적 있죠? 그런데 API로 구현하면 모든 텍스트가 한 번에 나타나서 사용자가 긴 답변을 기다리는 동안 먹통처럼 보이는 문제를 겪은 적 있나요?
이런 문제는 기본 API 호출이 동기식이어서 모든 생성이 끝날 때까지 기다려야 하기 때문입니다. GPT-4로 긴 답변을 생성하면 30초 이상 걸릴 수도 있는데, 그동안 사용자는 아무것도 볼 수 없으니 답답하죠.
바로 이럴 때 필요한 것이 스트리밍 응답입니다. stream=True 옵션을 켜면 생성되는 대로 토큰이 즉시 전달되어, 웹사이트처럼 실시간으로 텍스트가 나타나는 효과를 만들 수 있습니다.
개요
간단히 말해서, 스트리밍 응답은 AI가 텍스트를 생성하는 즉시 조각조각 받아오는 방식으로, 사용자 경험을 크게 개선합니다. 실무에서는 특히 긴 답변이나 느린 모델(GPT-4)을 사용할 때 스트리밍이 필수입니다.
사용자는 첫 단어가 나타나는 순간부터 "작동하고 있구나"라고 느끼고, 전체 답변을 기다리지 않아도 읽기 시작할 수 있습니다. 예를 들어, 코드 생성이나 긴 설명을 요청할 때 스트리밍이 없으면 사용자는 버그인 줄 알고 창을 닫아버릴 수도 있습니다.
기존에는 모든 응답이 끝날 때까지 블로킹되었다면, 이제는 Server-Sent Events 방식으로 실시간 데이터를 받을 수 있습니다. 핵심 특징은 즉각적인 피드백, 낮은 체감 지연시간, 부분적 결과 활용 가능성입니다.
이러한 특징들이 사용자가 기다림을 참을 수 있게 만들어줍니다.
코드 예제
# api.py (계속)
def send_message_stream(self, user_message: str, conversation_history: list = None):
"""스트리밍 방식으로 메시지를 보내고 응답을 받습니다"""
if conversation_history is None:
conversation_history = []
conversation_history.append({"role": "user", "content": user_message})
# stream=True 옵션으로 스트리밍 활성화
stream = self.client.chat.completions.create(
model=self.model,
messages=conversation_history,
max_tokens=self.max_tokens,
temperature=self.temperature,
stream=True # 핵심: 스트리밍 활성화
)
# 응답 조각들을 모으기 위한 변수
full_response = ""
# 각 청크를 순회하면서 즉시 yield
for chunk in stream:
# delta에 새로운 내용이 있으면 추출
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response += content
yield content # 즉시 반환 (generator)
# 완성된 응답을 히스토리에 추가
conversation_history.append({"role": "assistant", "content": full_response})
설명
이것이 하는 일: API 요청에 stream=True를 추가해서 응답을 청크 단위로 받고, Python generator를 사용해 각 청크를 즉시 호출자에게 전달합니다. 첫 번째로, stream=True 매개변수를 추가하면 API 응답이 한 번에 오는 게 아니라 이터레이터 객체로 반환됩니다.
이것은 내부적으로 HTTP의 Server-Sent Events를 사용해서 서버가 생성하는 대로 데이터를 푸시하는 방식입니다. for 루프로 이 스트림을 순회하면 새로운 청크가 올 때마다 반복됩니다.
그 다음으로, 각 chunk 객체를 검사합니다. chunk.choices[0].delta가 증분 업데이트를 나타내고, 여기에 content 속성이 있으면 새로운 텍스트 조각이라는 뜻입니다.
가끔 content가 None인 경우도 있어서(스트림 시작/종료 신호) if로 체크하는 게 중요합니다. 새 내용이 있으면 full_response에 누적해서 나중에 히스토리에 추가합니다.
마지막으로, yield 키워드를 사용합니다. 이게 핵심인데, yield는 함수를 generator로 만들어서 값을 즉시 반환하면서도 함수 실행을 중단하지 않습니다.
호출자가 next()를 부르거나 for 루프를 돌면 여기서 멈췄다가 다음 청크가 올 때 계속 실행됩니다. 이 메커니즘이 실시간 스트리밍을 가능하게 합니다.
여러분이 이 패턴을 사용하면 CLI에서 print()로 받는 즉시 출력할 수 있고, 웹 프레임워크에서는 WebSocket이나 SSE로 브라우저에 실시간 전송할 수 있으며, 사용자는 긴 답변도 지루하지 않게 기다릴 수 있습니다. 실제로 모든 프로덕션 ChatGPT 클론들이 이 방식을 사용합니다.
실전 팁
💡 스트리밍 중에 네트워크 오류가 나면 부분 응답이 손실될 수 있으니 full_response를 주기적으로 저장하거나 에러 발생 시 지금까지 받은 내용을 사용자에게 보여주세요.
💡 CLI에서는 sys.stdout.write(content); sys.stdout.flush()로 버퍼링 없이 즉시 출력하면 더 부드러운 타이핑 효과를 낼 수 있습니다.
💡 스트리밍 응답에는 finish_reason이 마지막 청크에만 들어있으니 chunk.choices[0].finish_reason == "stop"으로 정상 종료를 확인하세요.
💡 토큰 사용량은 스트리밍 중간에는 알 수 없고 마지막 청크의 usage 필드에 나옵니다. 비용 추적이 중요하면 이걸 따로 저장하세요.
6. 커맨드라인 인터페이스 기본 구조 - 사용자와의 대화창
시작하며
여러분이 터미널에서 사용자 입력을 받고 출력하는 반복 루프를 만들려고 하는데, 어떻게 깔끔하게 구조화해야 할지 고민된 적 있나요? 특히 종료 조건 처리나 예외 상황 관리가 복잡해 보입니다.
이런 문제는 CLI의 기본 패턴인 REPL(Read-Eval-Print Loop)을 이해하지 못해서 발생합니다. 단순히 input()과 print()를 쓰는 것은 쉽지만, 제대로 된 CLI 도구는 명령어 파싱, 히스토리 관리, 우아한 종료 등을 모두 처리해야 합니다.
바로 이럴 때 필요한 것이 잘 구조화된 CLI 메인 루프입니다. 무한 루프 안에서 입력을 받고, 처리하고, 출력하고, 예외를 핸들링하는 패턴을 익히면 어떤 CLI 도구도 만들 수 있습니다.
개요
간단히 말해서, CLI 기본 구조는 사용자 입력을 반복적으로 받아서 처리하고 결과를 보여주는 무한 루프입니다. 실무에서는 CLI가 단순해 보여도 제대로 만들려면 신경 쓸 게 많습니다.
Ctrl+C를 눌렀을 때 우아하게 종료해야 하고, 빈 입력이나 특수 명령어(/exit, /help 등)를 처리해야 하며, 에러가 나도 프로그램이 죽지 않고 계속 실행되어야 합니다. 예를 들어, 네트워크 오류로 API 호출이 실패해도 사용자는 다시 시도할 수 있어야 하죠.
기존에는 input()을 한두 번 쓰고 끝냈다면, 이제는 while True 루프 안에서 모든 것을 관리하는 견고한 구조를 만듭니다. 핵심 특징은 무한 루프 기반 REPL, 명령어 파싱, 예외 처리입니다.
이러한 특징들이 안정적이고 사용자 친화적인 CLI를 만들어냅니다.
코드 예제
# cli.py
import sys
from api import ChatGPTClient
from rich.console import Console
class ChatCLI:
"""커맨드라인 채팅 인터페이스"""
def __init__(self):
self.client = ChatGPTClient()
self.console = Console() # rich 라이브러리로 예쁜 출력
self.conversation_history = []
def run(self):
"""메인 루프 실행"""
self.console.print("[bold green]ChatGPT CLI 시작![/bold green]")
self.console.print("종료하려면 'exit' 또는 Ctrl+C를 누르세요.\n")
while True:
try:
# 사용자 입력 받기
user_input = input("You: ").strip()
# 종료 명령어 체크
if user_input.lower() in ['exit', 'quit', 'q']:
self.console.print("[yellow]대화를 종료합니다.[/yellow]")
break
# 빈 입력 무시
if not user_input:
continue
# API 호출 및 응답 출력 (다음 카드에서 구현)
self.handle_message(user_input)
except KeyboardInterrupt:
# Ctrl+C 처리
self.console.print("\n[yellow]대화를 종료합니다.[/yellow]")
break
except Exception as e:
# 예상치 못한 에러 처리
self.console.print(f"[red]에러 발생: {e}[/red]")
설명
이것이 하는 일: 터미널에서 계속해서 사용자 입력을 받고, 특수 명령어나 예외 상황을 적절히 처리하면서 ChatGPT와의 대화를 진행합니다. 첫 번째로, __init__에서 필요한 객체들을 초기화합니다.
ChatGPTClient는 API 통신을 담당하고, Console은 rich 라이브러리의 객체로 터미널에 색깔과 포맷팅이 적용된 예쁜 텍스트를 출력할 수 있게 해줍니다. conversation_history는 대화 맥락을 계속 유지하기 위한 리스트입니다.
그 다음으로, while True 무한 루프 안에서 input()으로 사용자 입력을 기다립니다. .strip()으로 앞뒤 공백을 제거한 뒤, 'exit', 'quit', 'q' 같은 종료 명령어인지 확인합니다.
만약 맞다면 break로 루프를 빠져나가고, 빈 문자열이면 continue로 다음 반복으로 넘어갑니다. 이렇게 하면 사용자가 실수로 엔터만 눌렀을 때 불필요한 API 호출을 방지할 수 있습니다.
마지막으로, 두 가지 예외를 처리합니다. KeyboardInterrupt는 사용자가 Ctrl+C를 눌렀을 때 발생하는데, 이걸 잡아서 우아하게 종료 메시지를 보여주고 break합니다.
일반적인 Exception은 API 오류나 예상치 못한 버그를 잡아서 에러 메시지만 출력하고 프로그램은 계속 실행되게 합니다. 이게 중요한 이유는 한 번의 에러로 전체 프로그램이 죽으면 사용자가 그동안의 대화 맥락을 다 잃어버리기 때문입니다.
여러분이 이 패턴을 사용하면 견고한 CLI 도구를 만들 수 있고, 사용자는 언제든 원하는 방식으로 종료할 수 있으며, 일시적인 네트워크 문제나 버그가 있어도 프로그램이 계속 동작합니다. 실제 유명한 CLI 도구들(Redis CLI, PostgreSQL psql 등)도 모두 이런 구조를 기반으로 합니다.
실전 팁
💡 rich 라이브러리를 사용하면 [bold red]텍스트[/bold red] 같은 마크업으로 색깔과 스타일을 쉽게 적용할 수 있어 사용자 경험이 크게 개선됩니다.
💡 입력이 여러 줄일 때를 대비해 특수 명령어(예: /multiline)를 만들어서 빈 줄이 나올 때까지 계속 입력받는 기능을 추가할 수 있습니다.
💡 readline 모듈을 import하면 화살표 키로 이전 명령어를 불러오는 히스토리 기능이 자동으로 활성화됩니다.
💡 시그널 핸들러를 등록하면 Ctrl+C 외에 Ctrl+D(EOF) 같은 다른 종료 시그널도 우아하게 처리할 수 있습니다.
7. 스트리밍 응답을 CLI에 통합하기 - 실시간 출력 연결
시작하며
여러분이 앞서 만든 스트리밍 API 함수와 CLI 메인 루프를 연결하려고 하는데, generator를 어떻게 처리해야 할지 막막했던 적 있나요? 특히 yield로 나오는 값을 하나씩 받아서 즉시 터미널에 출력하는 부분이 낯섭니다.
이런 문제는 Python의 generator 개념과 스트리밍 출력 방식을 제대로 이해하지 못해서 발생합니다. generator는 값을 하나씩 순회할 수 있는 이터레이터인데, CLI에서 for 루프로 돌면서 각 청크를 즉시 print하면 됩니다.
바로 이럴 때 필요한 것이 generator를 활용한 스트리밍 출력 패턴입니다. API의 generator를 for 루프로 순회하면서 sys.stdout.write()로 버퍼링 없이 즉시 출력하면 웹사이트처럼 타이핑 효과를 구현할 수 있습니다.
개요
간단히 말해서, 스트리밍 CLI 통합은 API의 generator 함수를 for 루프로 순회하면서 각 텍스트 조각을 즉시 터미널에 출력하는 것입니다. 실무에서는 사용자 경험을 위해 스트리밍이 거의 필수입니다.
특히 GPT-4처럼 느린 모델이나 긴 답변을 생성할 때 스트리밍 없이는 사용자가 30초 이상 빈 화면을 보게 됩니다. 예를 들어, "Python으로 웹 스크래퍼 만드는 방법 자세히 알려줘" 같은 요청은 응답이 길어서 스트리밍이 없으면 사용자가 창을 닫아버릴 수도 있습니다.
기존에는 모든 응답을 받은 뒤 한 번에 출력했다면, 이제는 생성되는 즉시 실시간으로 보여줍니다. 핵심 특징은 generator 순회, 버퍼링 없는 즉시 출력, 부드러운 타이핑 효과입니다.
이러한 특징들이 사용자가 기다림을 견딜 수 있게 만들어줍니다.
코드 예제
# cli.py (계속)
import sys
def handle_message(self, user_input: str):
"""사용자 메시지를 처리하고 스트리밍 응답을 출력합니다"""
self.console.print("\n[bold blue]Assistant:[/bold blue] ", end="")
try:
# 스트리밍 generator 받기
stream = self.client.send_message_stream(
user_input,
self.conversation_history
)
# 각 청크를 즉시 출력
for chunk in stream:
# 버퍼링 없이 즉시 출력
sys.stdout.write(chunk)
sys.stdout.flush() # 버퍼를 강제로 비움
print("\n") # 응답 끝나면 줄바꿈
except Exception as e:
self.console.print(f"\n[red]응답 중 에러: {e}[/red]")
설명
이것이 하는 일: API의 스트리밍 generator를 순회하면서 생성되는 텍스트 조각을 버퍼링 없이 터미널에 즉시 출력하여 실시간 타이핑 효과를 구현합니다. 첫 번째로, "Assistant: " 프롬프트를 먼저 출력합니다.
end="" 매개변수는 기본 줄바꿈을 방지해서 이어서 응답이 같은 줄에 나타나게 만듭니다. 그 다음 send_message_stream()을 호출하면 generator 객체가 반환되는데, 이것은 아직 실행되지 않은 상태입니다.
API 호출은 for 루프가 시작될 때 실제로 이루어집니다. 그 다음으로, for chunk in stream으로 generator를 순회합니다.
첫 반복에서 API 요청이 시작되고, 서버에서 첫 청크가 도착하면 chunk 변수에 담깁니다. 여기서 중요한 것은 print() 대신 sys.stdout.write()를 사용한다는 점입니다.
print()는 기본적으로 줄바꿈을 추가하고 내부 버퍼가 있어서 즉시 출력되지 않을 수 있습니다. 마지막으로, sys.stdout.flush()를 매번 호출합니다.
이게 핵심인데, Python의 stdout은 기본적으로 버퍼링되어서 일정량이 쌓이거나 줄바꿈이 나올 때까지 실제 화면에 출력되지 않습니다. flush()를 호출하면 버퍼에 있는 내용을 강제로 즉시 출력시켜서 한 글자 한 글자 타이핑되는 효과를 만듭니다.
루프가 끝나면 print("\n")로 줄바꿈을 추가해서 다음 입력 프롬프트가 깔끔하게 나타나게 합니다. 여러분이 이 패턴을 사용하면 ChatGPT 웹사이트와 똑같은 사용자 경험을 CLI에서도 제공할 수 있고, 긴 응답도 지루하지 않게 기다릴 수 있으며, 코드는 깔끔하고 이해하기 쉽게 유지됩니다.
실제로 GitHub Copilot CLI 같은 도구들도 이 방식을 사용합니다.
실전 팁
💡 Windows에서는 colorama 라이브러리를 import하면 ANSI 색상 코드가 제대로 표시됩니다: import colorama; colorama.init().
💡 매우 빠른 스트리밍 출력이 오히려 읽기 어려울 수 있으니 time.sleep(0.01)로 약간의 지연을 추가하면 더 자연스러울 수 있습니다.
💡 터미널 너비를 초과하는 긴 줄은 자동으로 줄바꿈되는데, 이를 제어하려면 textwrap 모듈로 수동으로 줄바꿈을 삽입할 수 있습니다.
💡 rich 라이브러리의 Live 클래스를 사용하면 더 정교한 스트리밍 출력과 스피너 애니메이션을 동시에 보여줄 수 있습니다.
8. 대화 히스토리 관리 - 맥락 유지의 핵심
시작하며
여러분이 ChatGPT에게 "그게 뭐야?"라고 물었을 때 이전 대화 맥락을 기억하지 못하고 엉뚱한 답변을 하는 문제를 겪어본 적 있나요? 이는 대화 히스토리를 제대로 관리하지 않아서 발생합니다.
이런 문제는 API가 기본적으로 상태를 저장하지 않기(stateless) 때문입니다. 매 요청마다 전체 대화 맥락을 다시 보내줘야 하는데, 이걸 깜빡하면 AI는 이전 대화를 전혀 모르는 상태가 되죠.
웹 ChatGPT처럼 자연스러운 대화를 하려면 반드시 히스토리 관리가 필요합니다. 바로 이럴 때 필요한 것이 conversation_history 리스트입니다.
모든 user와 assistant 메시지를 순서대로 저장하고, 매 요청마다 전체 히스토리를 함께 전송하면 완벽한 맥락 유지가 가능합니다.
개요
간단히 말해서, 대화 히스토리 관리는 모든 주고받은 메시지를 리스트에 저장하고 매 API 호출 시 함께 전송하는 것입니다. 실무에서는 단순히 히스토리를 쌓는 것뿐만 아니라 토큰 제한 관리도 중요합니다.
GPT-3.5-turbo는 4,096 토큰, GPT-4는 8,192 토큰(또는 32K 버전)의 컨텍스트 윈도우가 있어서, 대화가 길어지면 오래된 메시지를 잘라내야 합니다. 예를 들어, 코드 리뷰 봇처럼 긴 대화를 하는 경우 최근 10개 메시지만 유지하는 슬라이딩 윈도우 전략을 쓸 수 있습니다.
기존에는 각 요청이 독립적이었다면, 이제는 전체 대화를 하나의 세션으로 관리합니다. 핵심 특징은 순차적 메시지 저장, 전체 컨텍스트 전송, 토큰 제한 관리입니다.
이러한 특징들이 자연스러운 멀티턴 대화를 가능하게 합니다.
코드 예제
# cli.py (계속)
def handle_message(self, user_input: str):
"""사용자 메시지 처리 (히스토리 관리 포함)"""
# 사용자 메시지를 히스토리에 추가
self.conversation_history.append({
"role": "user",
"content": user_input
})
self.console.print("\n[bold blue]Assistant:[/bold blue] ", end="")
# 전체 응답을 모으기 위한 변수
full_response = ""
try:
# 스트리밍으로 응답 받기
stream = self.client.send_message_stream(
user_input,
self.conversation_history[:-1] # 방금 추가한 user 메시지 제외
)
for chunk in stream:
sys.stdout.write(chunk)
sys.stdout.flush()
full_response += chunk # 응답 누적
# 완성된 응답을 히스토리에 추가
self.conversation_history.append({
"role": "assistant",
"content": full_response
})
print("\n")
# 토큰 제한 관리: 히스토리가 너무 길면 잘라내기
self.trim_history()
except Exception as e:
self.console.print(f"\n[red]에러: {e}[/red]")
def trim_history(self, max_messages: int = 20):
"""히스토리가 너무 길면 오래된 메시지 제거"""
if len(self.conversation_history) > max_messages:
# 최근 N개 메시지만 유지
self.conversation_history = self.conversation_history[-max_messages:]
설명
이것이 하는 일: 대화의 모든 메시지를 conversation_history 리스트에 저장하고, API 호출 시 이 리스트를 전달하여 맥락을 유지하며, 너무 길어지면 자동으로 잘라냅니다. 첫 번째로, 사용자가 메시지를 입력하면 즉시 conversation_history에 추가합니다.
이때 {"role": "user", "content": user_input} 형식을 지켜야 OpenAI API가 이해할 수 있습니다. 그리고 API를 호출할 때 self.conversation_history[:-1]을 전달하는 것에 주목하세요.
이는 방금 추가한 user 메시지를 제외하는데, send_message_stream 함수 내부에서 다시 추가하기 때문입니다(중복 방지). 그 다음으로, 스트리밍 응답을 받으면서 full_response 변수에 모든 청크를 누적합니다.
화면에 출력하는 것과 별개로 완성된 응답을 저장해야 나중에 히스토리에 추가할 수 있습니다. 루프가 끝나면 이 full_response를 {"role": "assistant", "content": full_response} 형태로 히스토리에 append합니다.
마지막으로, trim_history()를 호출해서 히스토리가 너무 길어지는 것을 방지합니다. 예를 들어 max_messages=20이면 user와 assistant 메시지를 합쳐서 최근 20개만 유지하고 나머지는 버립니다.
리스트 슬라이싱 [-20:]은 마지막 20개 요소를 의미합니다. 이렇게 하면 토큰 제한에 걸리지 않으면서도 적절한 맥락을 유지할 수 있습니다.
여러분이 이 패턴을 사용하면 "그거", "그 방법" 같은 대명사를 써도 AI가 이해하고, 여러 번에 걸친 복잡한 요청도 처리할 수 있으며, 메모리와 비용을 통제하면서도 충분한 맥락을 유지할 수 있습니다. 실제 프로덕션 챗봇들은 더 정교한 전략(요약 생성, 중요한 메시지만 보존 등)을 쓰지만 이것이 기본입니다.
실전 팁
💡 시스템 메시지는 히스토리의 맨 앞에 고정으로 유지하고 절대 제거하지 마세요. AI의 행동 지침이기 때문에 항상 필요합니다.
💡 tiktoken 라이브러리로 실제 토큰 수를 계산하면 메시지 개수 대신 정확한 토큰 수로 제한할 수 있습니다: import tiktoken; enc = tiktoken.encoding_for_model("gpt-3.5-turbo").
💡 중요한 대화는 파일로 저장해두면 나중에 이어서 대화하거나 분석할 수 있습니다: json.dump(self.conversation_history, open("chat.json", "w")).
💡 토큰 제한에 자주 걸린다면 중간중간 대화를 요약해서 압축하는 전략도 있습니다(OpenAI API로 "지금까지 대화를 요약해줘" 요청 후 히스토리 교체).
9. 특수 명령어 구현 - CLI를 더 강력하게
시작하며
여러분이 CLI 도구를 사용하다가 대화를 초기화하거나 설정을 바꾸고 싶을 때 프로그램을 재시작하는 것이 불편했던 적 있나요? 모던한 CLI 도구들은 /help, /clear, /model 같은 특수 명령어로 다양한 기능을 제공합니다.
이런 문제는 모든 입력을 API로 바로 보내버려서 발생합니다. 사용자 입력을 먼저 파싱해서 특수 명령어인지 확인하고, 그에 따라 다르게 처리하는 로직이 필요합니다.
Redis CLI나 PostgreSQL psql 같은 도구들도 모두 이런 명령어 시스템을 갖추고 있습니다. 바로 이럴 때 필요한 것이 명령어 디스패처 패턴입니다.
'/'로 시작하는 입력을 감지해서 명령어 처리 함수로 라우팅하고, 그렇지 않으면 일반 메시지로 API에 보내는 것이죠.
개요
간단히 말해서, 특수 명령어는 '/'로 시작하는 입력을 가로채서 프로그램 자체의 기능을 실행하는 메타 명령입니다. 실무에서는 사용자 편의성을 위해 특수 명령어가 거의 필수입니다.
/clear로 대화 초기화, /model gpt-4로 모델 변경, /save로 대화 저장, /help로 도움말 표시 등의 기능이 있으면 사용자 경험이 크게 개선됩니다. 예를 들어, 새로운 주제로 대화를 시작하고 싶을 때 프로그램을 껐다 켜는 것보다 /clear가 훨씬 편리합니다.
기존에는 모든 입력이 API로 갔다면, 이제는 입력을 분류해서 명령어는 로컬에서 처리하고 일반 메시지만 API로 보냅니다. 핵심 특징은 명령어 파싱, 라우팅 로직, 확장 가능한 구조입니다.
이러한 특징들이 강력하고 유연한 CLI 도구를 만들어줍니다.
코드 예제
# cli.py (계속)
def run(self):
"""메인 루프 (명령어 처리 추가)"""
self.console.print("[bold green]ChatGPT CLI 시작![/bold green]")
self.console.print("명령어: /help, /clear, /model <name>, /exit\n")
while True:
try:
user_input = input("You: ").strip()
if not user_input:
continue
# 명령어 처리
if user_input.startswith('/'):
if self.handle_command(user_input):
continue # 명령어 처리 완료, 다음 루프
else:
break # exit 명령어
# 일반 메시지는 API로
self.handle_message(user_input)
except KeyboardInterrupt:
self.console.print("\n[yellow]대화를 종료합니다.[/yellow]")
break
def handle_command(self, command: str) -> bool:
"""특수 명령어 처리 (False 반환 시 프로그램 종료)"""
cmd_parts = command.split()
cmd_name = cmd_parts[0].lower()
if cmd_name in ['/exit', '/quit']:
self.console.print("[yellow]대화를 종료합니다.[/yellow]")
return False
elif cmd_name == '/clear':
self.conversation_history.clear()
self.console.print("[green]대화 히스토리를 초기화했습니다.[/green]")
elif cmd_name == '/model':
if len(cmd_parts) > 1:
self.client.model = cmd_parts[1]
self.console.print(f"[green]모델을 {cmd_parts[1]}(으)로 변경했습니다.[/green]")
else:
self.console.print(f"[blue]현재 모델: {self.client.model}[/blue]")
elif cmd_name == '/help':
self.show_help()
else:
self.console.print(f"[red]알 수 없는 명령어: {cmd_name}[/red]")
return True
def show_help(self):
"""도움말 표시"""
help_text = """
[bold]사용 가능한 명령어:[/bold]
/help - 이 도움말 표시
/clear - 대화 히스토리 초기화
/model [name] - 사용 중인 모델 확인 또는 변경
/exit - 프로그램 종료
"""
self.console.print(help_text)
설명
이것이 하는 일: 사용자 입력이 '/'로 시작하는지 확인해서 특수 명령어면 로컬에서 처리하고, 일반 메시지면 API로 보내는 라우팅 로직을 구현합니다. 첫 번째로, 메인 루프에서 user_input.startswith('/')로 명령어 여부를 판단합니다.
이게 True면 handle_command()로 보내고, False면 기존처럼 handle_message()로 보냅니다. handle_command()는 처리 성공 시 True를 반환해서 루프가 계속되고, 종료 명령어면 False를 반환해서 break됩니다.
그 다음으로, handle_command() 안에서 명령어를 파싱합니다. command.split()으로 공백으로 나누면 ['/model', 'gpt-4']처럼 리스트가 되고, 첫 번째 요소가 명령어 이름입니다.
if-elif 체인으로 각 명령어를 매칭하는데, 여기서 중요한 것은 확장 가능한 구조입니다. 새로운 명령어를 추가하려면 elif 블록만 추가하면 됩니다.
마지막으로, 각 명령어가 하는 일을 구현합니다. /clear는 conversation_history.clear()로 리스트를 비우고, /model은 client.model 속성을 변경하며, /help는 도움말 텍스트를 출력합니다.
/exit는 False를 반환해서 메인 루프를 종료시킵니다. 이렇게 하면 사용자는 프로그램 내에서 모든 것을 제어할 수 있습니다.
여러분이 이 패턴을 사용하면 사용자 편의성이 크게 향상되고, 새로운 기능 추가가 쉬워지며, 프로그램이 더 전문적이고 완성도 있게 보입니다. 실제로 성공한 CLI 도구들은 모두 풍부한 명령어 시스템을 갖추고 있습니다.
실전 팁
💡 명령어 이름은 대소문자를 구분하지 않도록 .lower()를 사용하세요. 사용자가 /HELP든 /help든 /Help든 상관없이 작동하게 만듭니다.
💡 argparse나 click 라이브러리를 사용하면 명령어에 플래그와 옵션을 추가할 수 있습니다: /save --format json --output chat.json.
💡 명령어를 딕셔너리로 매핑하면 더 깔끔합니다: commands = {'/clear': self.clear_history, '/help': self.show_help}; commands[cmd_name]().
💡 /history 명령어로 최근 대화를 표시하거나 /undo로 마지막 메시지를 취소하는 기능도 유용합니다.
10. 에러 처리와 재시도 로직 - 안정성 확보
시작하며
여러분이 네트워크가 불안정한 환경에서 CLI를 사용하다가 "Connection timeout" 에러 하나 때문에 프로그램이 죽고 그동안의 대화가 다 날아간 경험이 있나요? 실제 사용 환경은 완벽하지 않아서 이런 일시적인 오류가 자주 발생합니다.
이런 문제는 에러 처리를 제대로 하지 않아서 발생합니다. 단순히 try-except로 잡기만 하는 게 아니라, 어떤 에러는 재시도해야 하고(네트워크 오류), 어떤 에러는 사용자에게 알려야 하며(API 키 오류), 어떤 에러는 우아하게 복구해야 합니다(토큰 초과).
바로 이럴 때 필요한 것이 정교한 에러 처리와 재시도 로직입니다. OpenAI 라이브러리의 에러 타입을 구분해서 각각에 맞는 처리를 하고, 일시적인 오류는 지수 백오프로 재시도하는 것이죠.
개요
간단히 말해서, 에러 처리는 발생 가능한 다양한 오류를 분류해서 각각에 맞는 복구 전략을 적용하는 것입니다. 실무에서는 에러 처리가 사용자 경험의 핵심입니다.
API 호출은 네트워크 오류, 서버 과부하, 요금 한도 초과, 잘못된 요청 등 수많은 이유로 실패할 수 있습니다. 예를 들어, 500 Internal Server Error는 자동으로 재시도하면 성공할 수 있지만, 401 Unauthorized는 API 키 문제라서 재시도해도 소용없습니다.
기존에는 모든 에러를 똑같이 취급했다면, 이제는 에러 타입에 따라 다르게 반응합니다. 핵심 특징은 에러 분류, 선택적 재시도, 사용자 친화적 메시지입니다.
이러한 특징들이 견고하고 신뢰할 수 있는 애플리케이션을 만들어줍니다.
코드 예제
# api.py (계속)
import time
from openai import APIError, RateLimitError, APITimeoutError
def send_message_stream_with_retry(self, user_message: str,
conversation_history: list = None,
max_retries: int = 3):
"""재시도 로직이 포함된 스트리밍 메시지 전송"""
if conversation_history is None:
conversation_history = []
conversation_history.append({"role": "user", "content": user_message})
# 재시도 루프
for attempt in range(max_retries):
try:
stream = self.client.chat.completions.create(
model=self.model,
messages=conversation_history,
max_tokens=self.max_tokens,
temperature=self.temperature,
stream=True
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response += content
yield content
conversation_history.append({
"role": "assistant",
"content": full_response
})
return # 성공하면 함수 종료
except RateLimitError:
# 요금 한도 초과 - 재시도하지 않음
raise Exception("API 요금 한도를 초과했습니다. 잠시 후 다시 시도하세요.")
except APITimeoutError:
# 타임아웃 - 재시도
if attempt < max_retries - 1:
wait_time = 2 ** attempt # 지수 백오프: 1초, 2초, 4초
yield f"\n[네트워크 오류, {wait_time}초 후 재시도...]\n"
time.sleep(wait_time)
else:
raise Exception("네트워크 타임아웃이 반복됩니다.")
except APIError as e:
# 서버 오류 - 재시도
if attempt < max_retries - 1:
wait_time = 2 ** attempt
yield f"\n[서버 오류, {wait_time}초 후 재시도...]\n"
time.sleep(wait_time)
else:
raise Exception(f"API 오류: {str(e)}")
설명
이것이 하는 일: API 호출 중 발생할 수 있는 다양한 에러를 구체적인 타입으로 잡아내고, 일시적인 오류는 점진적으로 대기 시간을 늘려가며 재시도하고, 치명적인 오류는 즉시 사용자에게 알립니다. 첫 번째로, for attempt in range(max_retries) 루프로 재시도 로직을 감쌉니다.
기본적으로 3번까지 시도하는데, 성공하면 return으로 함수를 즉시 종료하고, 실패하면 except 블록에서 재시도 여부를 결정합니다. 이 구조가 중요한 이유는 재시도 횟수를 명시적으로 제한해서 무한 루프를 방지하기 때문입니다.
그 다음으로, 각 예외 타입을 구분해서 처리합니다. RateLimitError는 API 사용량 한도를 초과한 것으로, 재시도해도 즉시 해결되지 않으니 바로 에러를 발생시킵니다.
APITimeoutError는 네트워크 지연이나 일시적인 문제일 가능성이 높아서 재시도하고, 일반적인 APIError도 서버 과부하일 수 있으니 재시도합니다. 마지막으로, 지수 백오프(exponential backoff) 전략을 사용합니다.
wait_time = 2 ** attempt는 첫 번째 재시도는 1초, 두 번째는 2초, 세 번째는 4초 대기하는 것입니다. 이렇게 하는 이유는 서버가 과부하 상태일 때 모든 클라이언트가 동시에 재시도하면 상황이 더 악화되기 때문입니다.
yield로 재시도 메시지를 사용자에게 보여주면 무슨 일이 일어나는지 투명하게 알 수 있습니다. 여러분이 이 패턴을 사용하면 일시적인 네트워크 문제로 프로그램이 죽지 않고, 사용자는 재시도 과정을 볼 수 있어 안심하며, 서버에도 과도한 부하를 주지 않는 예의 바른 클라이언트가 됩니다.
AWS나 Google Cloud 같은 모든 주요 클라우드 서비스들도 이 재시도 전략을 권장합니다.
실전 팁
💡 openai.AuthenticationError를 따로 잡아서 "API 키를 확인하세요"라는 명확한 메시지를 주면 디버깅이 쉬워집니다.
💡 tenacity 라이브러리를 사용하면 데코레이터로 간단히 재시도 로직을 추가할 수 있습니다: @retry(stop=stop_after_attempt(3)).
💡 재시도 시 jitter(무작위 지연)를 추가하면 여러 클라이언트의 재시도가 동시에 일어나는 thundering herd 문제를 방지할 수 있습니다: time.sleep(wait_time + random.uniform(0, 1)).
💡 Sentry 같은 에러 추적 서비스와 통합하면 프로덕션에서 어떤 에러가 얼마나 자주 발생하는지 모니터링할 수 있습니다.
11. 시스템 메시지로 AI 행동 제어하기 - 맞춤형 어시스턴트
시작하며
여러분이 ChatGPT를 코딩 선생님처럼 사용하고 싶은데, 매번 "너는 친절한 프로그래밍 튜터야"라고 설명하는 게 번거로웠던 적 있나요? 또는 특정 스타일(간결함, 자세함, 유머 등)로 답변하게 만들고 싶을 때 어떻게 해야 할지 막막했나요?
이런 문제는 시스템 메시지를 활용하지 않아서 발생합니다. ChatGPT API는 "system" 역할의 메시지로 AI의 전체 행동 지침을 설정할 수 있는데, 이걸 대화의 맨 앞에 넣으면 모든 응답에 일관되게 적용됩니다.
바로 이럴 때 필요한 것이 시스템 메시지 활용입니다. 한 번만 설정하면 AI가 특정 캐릭터나 전문가처럼 행동하게 만들 수 있고, 응답 형식이나 길이도 제어할 수 있습니다.
개요
간단히 말해서, 시스템 메시지는 대화의 맨 앞에 놓이는 특별한 메시지로, AI에게 "너는 이런 역할이야"라고 지시하는 것입니다. 실무에서는 시스템 메시지가 AI 제품의 정체성을 결정합니다.
GitHub Copilot은 "너는 코드 완성 도우미야", Customer Service Bot은 "너는 친절한 고객 지원 직원이야" 같은 시스템 메시지를 사용합니다. 예를 들어, 의료 챗봇이라면 "너는 의학 용어를 쉽게 설명하는 건강 상담사야.
절대 진단을 내리지 마"같은 지침을 줄 수 있습니다. 기존에는 AI의 기본 행동에 의존했다면, 이제는 우리가 원하는 대로 맞춤 설정할 수 있습니다.
핵심 특징은 전역적 행동 제어, 일관된 톤과 스타일, 안전 가드레일 설정입니다. 이러한 특징들이 전문적이고 목적에 맞는 AI 어시스턴트를 만들어줍니다.
코드 예제
# cli.py (계속)
def __init__(self):
self.client = ChatGPTClient()
self.console = Console()
# 시스템 메시지로 AI 행동 정의
self.system_message = {
"role": "system",
"content": """당신은 친절하고 전문적인 프로그래밍 튜터입니다.
지침:
- 초보자도 이해할 수 있도록 쉽게 설명하세요
- 코드 예제를 제공할 때는 주석을 상세히 다세요
- 단계별로 나누어 설명하세요
- 질문이 불명확하면 명확히 하도록 물어보세요
- 긍정적이고 격려하는 톤을 유지하세요
"""
}
# 대화 히스토리 초기화 (시스템 메시지 포함)
self.conversation_history = [self.system_message]
def handle_command(self, command: str) -> bool:
"""명령어 처리 (시스템 메시지 변경 추가)"""
cmd_parts = command.split(maxsplit=1) # 최대 2개로 분리
cmd_name = cmd_parts[0].lower()
# ... 기존 명령어들 ...
elif cmd_name == '/system':
if len(cmd_parts) > 1:
# 새로운 시스템 메시지 설정
self.system_message["content"] = cmd_parts[1]
# 히스토리 재구성 (시스템 메시지만 새 것으로 교체)
self.conversation_history = [
self.system_message
] + [m for m in self.conversation_history[1:]]
self.console.print("[green]시스템 메시지를 변경했습니다.[/green]")
else:
self.console.print(f"[blue]현재 시스템 메시지:[/blue]\n{self.system_message['content']}")
return True
설명
이것이 하는 일: 대화가 시작되기 전에 시스템 메시지를 히스토리의 첫 번째 요소로 삽입해서, 모든 응답이 이 지침을 따르도록 만들고, 필요시 /system 명령어로 동적으로 변경할 수 있게 합니다. 첫 번째로, __init__에서 self.system_message 딕셔너리를 정의합니다.
"role": "system"이 핵심인데, 이것은 OpenAI에게 "이것은 사용자 메시지가 아니라 AI 자체에 대한 메타 지시야"라고 알려줍니다. content에는 구체적이고 명확한 지침을 작성하는데, 여러 줄 문자열(""")로 구조화하면 가독성이 좋습니다.
그 다음으로, conversation_history를 초기화할 때 시스템 메시지를 첫 번째 요소로 넣습니다. [self.system_message]로 시작하면 이후 추가되는 모든 user와 assistant 메시지 앞에 항상 이 시스템 메시지가 있게 됩니다.
OpenAI API는 매 요청마다 전체 히스토리를 받으므로, 이 시스템 메시지도 매번 전송되어 일관된 행동을 보장합니다. 마지막으로, /system 명령어로 런타임에 시스템 메시지를 바꿀 수 있게 합니다.
이게 강력한 이유는 대화 중간에 AI의 역할을 전환할 수 있기 때문입니다. 예를 들어 코드 리뷰 모드에서 디버깅 모드로 바꾸거나, 자세한 설명에서 간결한 답변으로 전환할 수 있습니다.
히스토리를 재구성할 때 [1:]로 슬라이싱해서 기존 시스템 메시지를 제외한 나머지를 유지합니다. 여러분이 이 패턴을 사용하면 AI의 정체성을 명확히 정의할 수 있고, 부적절한 응답을 필터링하는 가드레일을 설정할 수 있으며, 다양한 사용 케이스에 맞는 전문화된 어시스턴트를 만들 수 있습니다.
실제로 모든 프로덕션 ChatGPT 제품들이 정교한 시스템 메시지를 사용합니다.
실전 팁
💡 시스템 메시지에 "절대 ~하지 마"같은 금지 사항을 명시하면 안전성을 높일 수 있습니다: "절대 의학적 진단을 내리지 마세요".
💡 응답 형식을 제어할 수 있습니다: "항상 JSON 형식으로만 답변하세요" 또는 "코드 블록에는 항상 언어 태그를 붙이세요".
💡 여러 페르소나를 프리셋으로 만들어두면 편합니다: presets = {"tutor": "...", "reviewer": "...", "debugger": "..."}.
💡 시스템 메시지도 토큰을 소비하므로 너무 길면 비용이 증가합니다. 핵심만 간결하게 작성하세요(100-200 토큰 정도).
12. 대화 저장 및 불러오기 - 세션 영속성
시작하며
여러분이 긴 대화를 하다가 프로그램을 종료했는데, 다시 시작하면 모든 맥락이 사라져서 처음부터 다시 설명해야 했던 경험이 있나요? 특히 복잡한 코드를 함께 작성하거나 긴 브레인스토밍을 하는 경우 세션을 저장하지 못하면 정말 답답합니다.
이런 문제는 대화 데이터를 메모리에만 저장해서 발생합니다. conversation_history는 프로그램이 종료되면 사라지는 휘발성 데이터인데, 이걸 파일이나 데이터베이스에 저장하면 나중에 이어서 할 수 있습니다.
바로 이럴 때 필요한 것이 대화 저장/불러오기 기능입니다. JSON 형식으로 히스토리를 직렬화해서 파일에 쓰고, 다시 읽어서 복원하면 프로그램을 재시작해도 정확히 이어갈 수 있습니다.
개요
간단히 말해서, 대화 저장은 conversation_history를 JSON 파일로 직렬화하고, 불러오기는 그 파일을 읽어 히스토리로 복원하는 것입니다. 실무에서는 세션 영속성이 사용자 경험에 큰 영향을 줍니다.
사용자는 며칠에 걸쳐 프로젝트를 작업하거나, 중간에 컴퓨터를 재시작하거나, 나중에 참고하기 위해 대화를 보관하고 싶어 합니다. 예를 들어, 코드 리뷰 세션을 저장해두면 나중에 같은 코드를 수정할 때 이전 피드백을 참고할 수 있습니다.
기존에는 대화가 메모리에만 존재했다면, 이제는 영구 저장소에 보관하고 필요할 때 불러올 수 있습니다. 핵심 특징은 JSON 직렬화, 파일 시스템 저장, 세션 복원입니다.
이러한 특징들이 장기적인 사용성과 데이터 분석 가능성을 제공합니다.
코드 예제
# cli.py (계속)
import json
from datetime import datetime
from pathlib import Path
def __init__(self):
# ... 기존 초기화 ...
self.save_dir = Path("chat_history") # 저장 디렉토리
self.save_dir.mkdir(exist_ok=True) # 없으면 생성
def handle_command(self, command: str) -> bool:
"""명령어 처리 (저장/불러오기 추가)"""
cmd_parts = command.split(maxsplit=1)
cmd_name = cmd_parts[0].lower()
# ... 기존 명령어들 ...
elif cmd_name == '/save':
filename = cmd_parts[1] if len(cmd_parts) > 1 else None
self.save_conversation(filename)
elif cmd_name == '/load':
if len(cmd_parts) > 1:
self.load_conversation(cmd_parts[1])
else:
self.list_saved_conversations()
return True
def save_conversation(self, filename: str = None):
"""대화 히스토리를 JSON 파일로 저장"""
if filename is None:
# 타임스탬프로 자동 파일명 생성
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"chat_{timestamp}.json"
if not filename.endswith('.json'):
filename += '.json'
filepath = self.save_dir / filename
# JSON으로 저장
with open(filepath, 'w', encoding='utf-8') as f:
json.dump({
'timestamp': datetime.now().isoformat(),
'model': self.client.model,
'history': self.conversation_history
}, f, ensure_ascii=False, indent=2)
self.console.print(f"[green]대화를 {filepath}에 저장했습니다.[/green]")
def load_conversation(self, filename: str):
"""저장된 대화 히스토리 불러오기"""
if not filename.endswith('.json'):
filename += '.json'
filepath = self.save_dir / filename
if not filepath.exists():
self.console.print(f"[red]파일을 찾을 수 없습니다: {filepath}[/red]")
return
# JSON에서 읽기
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
# 히스토리 복원
self.conversation_history = data['history']
self.client.model = data.get('model', self.client.model)
self.console.print(f"[green]대화를 불러왔습니다. (저장 시각: {data['timestamp']})[/green]")
def list_saved_conversations(self):
"""저장된 대화 목록 표시"""
files = sorted(self.save_dir.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
if not files:
self.console.print("[yellow]저장된 대화가 없습니다.[/yellow]")
return
self.console.print("\n[bold]저장된 대화 목록:[/bold]")
for file in files[:10]: # 최근 10개만 표시
mtime = datetime.fromtimestamp(file.stat().st_mtime)
self.console.print(f" - {file.name} ({mtime.strftime('%Y-%m-%d %H:%M')})")
설명
이것이 하는 일: 대화 히스토리를 JSON 형식으로 변환하여 파일 시스템에 저장하고, 저장된 파일을 다시 읽어 정확히 같은 상태로 복원하며, 저장된 대화 목록을 보여줍니다. 첫 번째로, save_conversation()에서 히스토리를 JSON으로 직렬화합니다.
conversation_history는 딕셔너리 리스트이므로 json.dump()로 바로 저장할 수 있습니다. 추가로 timestamp와 model 정보도 함께 저장해서 나중에 언제, 어떤 모델로 했던 대화인지 알 수 있게 합니다.
ensure_ascii=False는 한글이 깨지지 않게, indent=2는 사람이 읽을 수 있게 포맷팅하는 옵션입니다. 그 다음으로, load_conversation()에서 파일을 읽어 히스토리를 복원합니다.
json.load()로 딕셔너리를 얻은 뒤 data['history']를 self.conversation_history에 할당하면 끝입니다. 이때 시스템 메시지부터 모든 user/assistant 메시지까지 정확히 같은 순서로 복원되므로, 다음 메시지는 이전 맥락을 완벽하게 이어받습니다.
마지막으로, list_saved_conversations()는 chat_history 디렉토리의 모든 JSON 파일을 수정 시간 순으로 정렬해서 보여줍니다. .glob("*.json")으로 파일 목록을 얻고, sorted()에 key=lambda로 수정 시간(st_mtime)을 기준으로 정렬하며, reverse=True로 최신 파일이 먼저 나오게 합니다.
파일명에 타임스탬프가 포함되어 있어 어떤 대화인지 구분하기 쉽습니다. 여러분이 이 기능을 사용하면 긴 프로젝트를 여러 세션에 걸쳐 진행할 수 있고, 유용한 대화를 보관해서 나중에 참고할 수 있으며, 대화 데이터를 분석하거나 파인튜닝에 활용할 수도 있습니다.
실제 프로덕션 챗봇들은 데이터베이스에 저장하지만, 개인 도구는 JSON 파일로도 충분합니다.
실전 팁
💡 대화에 제목을 붙이는 기능을 추가하면 나중에 찾기 쉽습니다: /save "API 통합 논의".
💡 자동 저장 기능을 추가하면 프로그램이 비정상 종료되어도 데이터를 잃지 않습니다: 매 N개 메시지마다 임시 파일에 저장.
💡 클라우드 스토리지(AWS S3, Google Drive)와 연동하면 여러 기기에서 대화를 동기화할 수 있습니다.
💡 SQLite 데이터베이스를 사용하면 대화를 키워드로 검색하거나 날짜 범위로 필터링하는 고급 기능을 구현할 수 있습니다.