이미지 로딩 중...
AI Generated
2025. 11. 11. · 4 Views
바닥부터 만드는 ChatGPT 3편 - GPT-4 스타일 텍스트 분할 구현
GPT-4가 사용하는 고급 텍스트 분할 기법을 직접 구현해봅니다. 토큰 제한을 효과적으로 처리하고, 문맥을 유지하면서 긴 텍스트를 청크로 나누는 실전 기술을 배워보세요.
목차
- 토큰_기반_텍스트_분할
- RecursiveCharacterTextSplitter
- 청크_오버랩_설정
- 토큰_카운팅_구현
- 동적_청크_크기_조정
- 문장_경계_보존
- 메타데이터_추가
- 배치_처리_최적화
1. 토큰_기반_텍스트_분할
시작하며
여러분이 ChatGPT API를 사용해서 긴 문서를 처리하려고 했을 때, "maximum context length exceeded" 에러를 만난 적 있나요? 열심히 준비한 프롬프트인데 갑자기 에러가 나면 당황스럽죠.
이런 문제는 실제 개발 현장에서 매우 자주 발생합니다. GPT 모델은 한 번에 처리할 수 있는 토큰 수가 정해져 있기 때문에, 긴 문서를 단순히 전체를 보내면 에러가 발생합니다.
특히 GPT-3.5는 4,096 토큰, GPT-4는 8,192~32,768 토큰까지만 처리할 수 있습니다. 바로 이럴 때 필요한 것이 토큰 기반 텍스트 분할입니다.
단순히 글자 수로 자르는 것이 아니라, GPT가 실제로 인식하는 토큰 단위로 정확하게 분할하면 에러 없이 안전하게 처리할 수 있습니다.
개요
간단히 말해서, 토큰 기반 텍스트 분할은 GPT 모델이 인식하는 단위인 '토큰'을 기준으로 텍스트를 나누는 기술입니다. 왜 글자 수가 아닌 토큰일까요?
GPT는 텍스트를 글자가 아닌 '토큰'이라는 단위로 이해합니다. 예를 들어, "안녕하세요"는 5글자지만 GPT에게는 3~5개의 토큰으로 인식될 수 있습니다.
영어는 보통 1단어 = 1~2토큰이지만, 한국어나 중국어는 더 많은 토큰을 사용합니다. 이런 차이를 모르고 글자 수로만 자르면 예상치 못한 에러가 발생합니다.
기존에는 "4,000글자씩 자르면 되겠지"라고 단순하게 생각했다면, 이제는 "정확히 몇 토큰인지 측정해서 안전하게 분할하자"는 접근이 필요합니다. 토큰 기반 분할의 핵심 특징은 세 가지입니다: (1) 모델의 실제 제한에 맞춤 - 에러 발생 가능성 제거, (2) 언어별 차이 자동 반영 - 한국어도 안전하게 처리, (3) 정확한 비용 계산 - API 사용료를 토큰으로 청구하므로 예산 관리 가능.
이러한 특징들이 실무에서 안정적인 AI 서비스를 만드는 데 필수적입니다.
코드 예제
import tiktoken
def count_tokens(text: str, model: str = "gpt-4") -> int:
"""텍스트의 토큰 수를 정확하게 계산"""
# 모델에 맞는 인코더 가져오기
encoding = tiktoken.encoding_for_model(model)
# 텍스트를 토큰으로 변환
tokens = encoding.encode(text)
# 토큰 개수 반환
return len(tokens)
# 실제 사용 예시
long_text = "인공지능이 세상을 바꾸고 있습니다. " * 100
token_count = count_tokens(long_text, "gpt-4")
print(f"총 토큰 수: {token_count}") # 정확한 토큰 수 출력
설명
이것이 하는 일: tiktoken 라이브러리를 사용해서 주어진 텍스트가 GPT 모델에서 몇 개의 토큰으로 변환되는지 정확하게 계산하는 함수입니다. 첫 번째로, encoding = tiktoken.encoding_for_model(model) 부분은 사용하려는 GPT 모델에 맞는 토크나이저(인코더)를 가져옵니다.
GPT-3.5와 GPT-4는 서로 다른 토크나이저를 사용하기 때문에, 정확한 모델명을 지정해야 합니다. 왜 이렇게 하냐고요?
같은 텍스트라도 모델마다 토큰으로 나누는 방식이 다르기 때문입니다. 그 다음으로, tokens = encoding.encode(text) 부분이 실행되면서 실제 텍스트를 토큰 리스트로 변환합니다.
내부에서는 텍스트를 작은 의미 단위로 쪼개서 숫자 ID로 변환하는 과정이 일어납니다. 예를 들어 "안녕하세요"는 [1234, 5678, 9012] 같은 숫자 배열로 변환됩니다.
마지막으로, len(tokens) 부분이 토큰 배열의 길이를 세어 최종적으로 토큰 개수를 반환합니다. 이 숫자가 바로 GPT API에 요청할 때 소비되는 실제 토큰 수입니다.
여러분이 이 코드를 사용하면 몇 가지 구체적인 이점을 얻을 수 있습니다: (1) API 에러 사전 예방 - 요청 전에 토큰 수를 확인해서 제한을 초과하지 않도록 조정 가능, (2) 정확한 비용 산출 - OpenAI는 토큰당 과금하므로 사전에 비용 계산 가능, (3) 최적의 청크 크기 결정 - 텍스트를 나눌 때 정확한 기준점 확보.
실전 팁
💡 모델명을 정확히 지정하세요. "gpt-4"와 "gpt-3.5-turbo"는 다른 토크나이저를 사용하므로, 실제 사용할 모델명을 정확히 입력해야 정확한 토큰 수를 얻습니다.
💡 tiktoken 설치는 pip install tiktoken으로 간단히 할 수 있습니다. 만약 설치 중 에러가 나면 Python 버전을 3.8 이상으로 업그레이드해보세요.
💡 토큰 제한의 80%만 사용하는 것이 안전합니다. 예를 들어 GPT-4가 8,192 토큰을 지원한다면, 실제로는 6,500 토큰 정도만 사용하는 것이 좋습니다. 시스템 메시지와 응답 토큰도 포함되기 때문입니다.
💡 대용량 텍스트는 배치 처리하세요. 수백 개의 문서를 처리할 때는 한 번에 하나씩이 아니라 encode_batch() 메서드를 사용하면 10배 이상 빠릅니다.
💡 캐싱을 활용하면 성능이 크게 향상됩니다. 같은 텍스트를 반복해서 토큰 카운팅할 때는 결과를 딕셔너리에 저장해두고 재사용하세요.
2. RecursiveCharacterTextSplitter
시작하며
여러분이 긴 블로그 글이나 논문을 GPT에게 요약해달라고 했을 때, 분할된 청크들이 문장 중간에서 잘려서 이상한 결과가 나온 경험 있나요? "그는 매우 중요한..."에서 끊기면 뒷부분의 의미를 완전히 잃어버리게 됩니다.
이런 문제는 단순한 글자 수 기반 분할의 가장 큰 약점입니다. 텍스트의 구조를 전혀 고려하지 않고 기계적으로 자르기 때문에, 문단, 문장, 심지어 단어 중간에서도 잘라버립니다.
결과적으로 GPT가 문맥을 제대로 이해하지 못해서 엉뚱한 답변을 내놓게 됩니다. 바로 이럴 때 필요한 것이 RecursiveCharacterTextSplitter입니다.
문단 → 문장 → 단어 순서로 계층적으로 분할을 시도해서, 가능한 한 자연스러운 지점에서 텍스트를 나눕니다.
개요
간단히 말해서, RecursiveCharacterTextSplitter는 텍스트의 자연스러운 경계를 지키면서 분할하는 스마트한 알고리즘입니다. 왜 이 개념이 필요할까요?
일반적인 텍스트는 계층 구조를 가지고 있습니다: 문서 > 문단 > 문장 > 구 > 단어. 만약 우리가 이 구조를 무시하고 무작정 자르면, 의미의 연결성이 깨집니다.
RecursiveCharacterTextSplitter는 이 계층을 존중해서, 먼저 큰 단위(문단)로 나누려 시도하고, 안 되면 중간 단위(문장), 그래도 안 되면 작은 단위(단어)로 나눕니다. 기존에는 text.split()[0:100] 같은 식으로 무식하게 잘랐다면, 이제는 "\n\n", ".", " " 같은 구분자를 우선순위대로 시도해서 최적의 분할점을 찾습니다.
핵심 특징은 세 가지입니다: (1) 재귀적 시도 - 큰 단위부터 작은 단위로 점진적 분할, (2) 우선순위 기반 - 문단 구분자 > 문장 구분자 > 공백 순서로 시도, (3) 청크 크기 보장 - 목표 크기를 초과하지 않으면서도 최대한 채움. 이러한 특징들이 GPT가 문맥을 정확히 이해하고 좋은 답변을 생성하는 데 결정적 역할을 합니다.
코드 예제
from langchain.text_splitter import RecursiveCharacterTextSplitter
def smart_split_text(text: str, chunk_size: int = 1000, chunk_overlap: int = 200):
"""문맥을 유지하면서 텍스트를 지능적으로 분할"""
# 분할기 초기화 - 우선순위: 문단 > 문장 > 단어
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, # 청크당 최대 글자 수
chunk_overlap=chunk_overlap, # 청크 간 겹침 영역
separators=["\n\n", "\n", ". ", " ", ""], # 분할 우선순위
length_function=len # 길이 측정 함수
)
# 텍스트를 청크로 분할
chunks = splitter.split_text(text)
return chunks
# 실제 사용
article = """딥러닝은 머신러닝의 한 분야입니다.\n\n신경망을 기반으로 합니다."""
result_chunks = smart_split_text(article, chunk_size=50)
print(f"총 {len(result_chunks)}개 청크 생성")
설명
이것이 하는 일: LangChain의 RecursiveCharacterTextSplitter를 사용해서 텍스트를 의미 단위로 지능적으로 분할하는 함수입니다. 첫 번째로, RecursiveCharacterTextSplitter() 생성자 부분에서 분할 규칙을 설정합니다.
chunk_size=1000은 각 청크가 최대 1000자를 넘지 않도록 제한하고, chunk_overlap=200은 인접한 청크 사이에 200자의 겹침을 만들어서 문맥 연결성을 유지합니다. 왜 겹침이 필요하냐고요?
청크 경계에서 중요한 정보가 잘릴 수 있기 때문에, 일부러 중복을 두어 앞뒤 문맥을 모두 포함시킵니다. 그 다음으로, `separators=["\n\n", "\n", ".
", " ", ""]` 부분이 핵심입니다. 이 리스트는 분할을 시도할 구분자의 우선순위를 정의합니다.
먼저 "\n\n"(문단 구분)으로 나누려 시도하고, 청크가 너무 크면 "\n"(줄바꿈)으로 시도하고, 그래도 크면 ". "(문장 끝)으로 시도합니다.
공백도 안 되면 마지막으로 빈 문자열("")을 사용해서 글자 단위로 강제 분할합니다. 마지막으로, splitter.split_text(text) 부분이 실제 분할을 수행하면서 최적의 청크 리스트를 생성합니다.
내부적으로는 재귀 알고리즘이 작동해서, 각 구분자를 순서대로 시도하면서 chunk_size에 맞는 최적의 분할점을 찾아냅니다. 여러분이 이 코드를 사용하면 다음과 같은 이점을 얻습니다: (1) 자연스러운 분할 - 문장이나 문단 중간이 아닌 경계에서 분할되어 GPT가 문맥 이해 용이, (2) 정보 보존 - chunk_overlap 덕분에 경계 정보 손실 최소화, (3) 유연한 적용 - 뉴스 기사, 논문, 소설 등 다양한 텍스트 타입에 동일하게 적용 가능.
실전 팁
💡 separators 순서가 결과를 좌우합니다. 여러분의 텍스트 타입에 맞게 커스터마이징하세요. 예를 들어 코드를 분할한다면 ["\n\nclass ", "\n\ndef ", "\n\n", "\n"]처럼 클래스와 함수 경계를 우선시하면 좋습니다.
💡 chunk_overlap은 chunk_size의 10-20%가 적당합니다. 너무 크면 중복이 심해서 비용이 늘고, 너무 작으면 문맥 연결이 끊깁니다. 1000자 청크라면 100-200자 오버랩이 황금 비율입니다.
💡 length_function을 토큰 카운터로 바꾸면 더 정확합니다. length_function=lambda x: count_tokens(x, "gpt-4")로 설정하면 글자 수가 아닌 토큰 수 기준으로 분할됩니다.
💡 한국어는 줄바꿈 패턴이 다릅니다. separators에 [".\n", "!\n", "?\n"]을 추가하면 한국어 문장 경계를 더 잘 인식합니다.
💡 디버깅 시 청크 경계를 확인하세요. print(chunks[0][-50:]), print(chunks[1][:50])처럼 인접 청크의 끝과 시작을 출력하면 오버랩이 제대로 작동하는지 확인할 수 있습니다.
3. 청크_오버랩_설정
시작하며
여러분이 긴 계약서를 청크로 나누어 GPT에게 분석시켰는데, "3조 항목이 누락되었습니다"라는 결과를 받은 적 있나요? 분명히 원본에는 있는 내용인데 왜 GPT는 못 찾았을까요?
이런 문제는 청크 경계에서 중요한 정보가 잘렸을 때 발생합니다. 예를 들어, "제3조 계약 당사자는..."이라는 문장이 "제3조 계약"까지는 첫 번째 청크에, "당사자는..."부터는 두 번째 청크에 들어가면, GPT는 둘 다 불완전한 정보로 인식해서 제대로 해석하지 못합니다.
바로 이럴 때 필요한 것이 청크 오버랩입니다. 인접한 청크 사이에 일부러 중복 영역을 만들어서, 경계에 걸친 정보가 양쪽 청크 모두에 완전하게 포함되도록 보장합니다.
개요
간단히 말해서, 청크 오버랩은 인접한 텍스트 조각 사이에 의도적으로 중복 구간을 두어 정보 손실을 방지하는 기법입니다. 왜 중복이 필요할까요?
텍스트를 나누면 필연적으로 경계가 생깁니다. 아무리 RecursiveCharacterTextSplitter로 자연스럽게 나눈다 해도, 경계 근처의 정보는 앞뒤 문맥이 부족할 수 있습니다.
예를 들어, "따라서 이러한 이유로"라는 문장이 청크 시작 부분에 있으면, "이러한 이유"가 무엇을 가리키는지 모릅니다. 오버랩이 있으면 이전 청크의 마지막 몇 문장이 포함되어 문맥이 연결됩니다.
기존에는 "청크는 겹치지 않게 깔끔하게 나눠야 한다"고 생각했다면, 이제는 "약간의 중복은 정보 보존을 위한 필수 투자"라고 인식해야 합니다. 청크 오버랩의 핵심 특징은 세 가지입니다: (1) 문맥 보존 - 경계 근처 정보도 완전한 문맥 속에서 이해 가능, (2) 검색 정확도 향상 - 벡터 검색 시 관련 정보를 놓칠 확률 감소, (3) 조정 가능성 - 텍스트 타입과 중요도에 따라 오버랩 크기 조절 가능.
이러한 특징들이 RAG(Retrieval-Augmented Generation) 시스템에서 정확도를 크게 높입니다.
코드 예제
from langchain.text_splitter import RecursiveCharacterTextSplitter
def create_overlapping_chunks(text: str, chunk_size: int = 1000, overlap_ratio: float = 0.2):
"""최적의 오버랩으로 청크 생성"""
# 오버랩 크기 계산 (chunk_size의 20%)
chunk_overlap = int(chunk_size * overlap_ratio)
# 오버랩 적용된 분할기 생성
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap, # 중복 영역 설정
separators=["\n\n", "\n", ". ", " "]
)
chunks = splitter.split_text(text)
# 오버랩 확인 (디버깅용)
if len(chunks) > 1:
overlap_text = chunks[0][-chunk_overlap:]
print(f"오버랩 영역 예시: ...{overlap_text[:50]}")
return chunks
# 실제 사용
doc = "AI는 미래입니다. " * 200
chunks = create_overlapping_chunks(doc, chunk_size=500, overlap_ratio=0.15)
print(f"{len(chunks)}개 청크, 각각 최대 500자, 15% 오버랩")
설명
이것이 하는 일: 청크 크기에 비례하는 최적의 오버랩 크기를 자동 계산하고, 이를 적용한 텍스트 분할을 수행하는 함수입니다. 첫 번째로, chunk_overlap = int(chunk_size * overlap_ratio) 부분은 오버랩 크기를 동적으로 계산합니다.
예를 들어 chunk_size가 1000이고 overlap_ratio가 0.2면, 200자의 오버랩이 생성됩니다. 왜 비율로 계산하냐고요?
청크 크기가 바뀔 때마다 수동으로 오버랩을 조정하는 것은 번거롭고, 일정 비율을 유지하는 것이 일관된 품질을 보장하기 때문입니다. 그 다음으로, chunk_overlap=chunk_overlap 파라미터가 RecursiveCharacterTextSplitter에 전달됩니다.
내부적으로는 각 청크를 생성할 때, 이전 청크의 마지막 N자(여기서는 200자)를 현재 청크의 시작 부분에 포함시킵니다. 결과적으로 청크1의 끝 200자와 청크2의 앞 200자가 동일한 내용이 됩니다.
마지막으로, 디버깅 코드 부분에서 chunks[0][-chunk_overlap:]을 출력해서 실제로 오버랩이 제대로 작동하는지 확인합니다. 이 부분을 청크2의 시작 부분과 비교하면 정확히 일치하는 것을 볼 수 있습니다.
여러분이 이 코드를 사용하면 다음과 같은 실무적 이점을 얻습니다: (1) 정보 무결성 보장 - 중요한 계약서, 의료 기록 같은 민감한 문서도 정보 손실 없이 처리, (2) 검색 품질 향상 - 벡터 DB에 저장 후 검색 시 관련 정보를 놓칠 확률 대폭 감소, (3) 유지보수 편의성 - overlap_ratio 하나만 조정하면 모든 청크에 일관되게 적용.
실전 팁
💡 문서 타입별 최적 오버랩은 다릅니다. 소설이나 에세이는 10% 정도로 낮춰도 되지만, 법률 문서나 기술 매뉴얼은 20-25%로 높여서 참조 관계를 확실히 보존하세요.
💡 오버랩이 너무 크면 비용이 급증합니다. GPT API는 토큰당 과금하므로, 50% 오버랩을 쓰면 사실상 1.5배의 비용이 듭니다. 20%가 비용 대비 효과의 스위트 스팟입니다.
💡 오버랩 영역을 시각화하면 디버깅이 쉽습니다. print(f"청크1 끝: {chunks[0][-100:]}\n청크2 시작: {chunks[1][:100]}")처럼 출력해서 중복 여부를 눈으로 확인하세요.
💡 벡터 검색 시 오버랩은 더 중요합니다. 사용자 질문이 청크 경계에 걸친 내용을 물어볼 때, 오버랩 덕분에 양쪽 청크 모두 검색되어 완전한 답변이 가능합니다.
💡 메타데이터에 오버랩 정보를 기록하세요. {"chunk_id": 2, "overlap_start": 800, "overlap_end": 1000}처럼 저장하면 나중에 청크를 재조립할 때 중복 제거가 쉽습니다.
4. 토큰_카운팅_구현
시작하며
여러분이 GPT API 비용이 예상보다 10배나 많이 나와서 깜짝 놀란 경험 있나요? "분명히 짧은 텍스트만 보냈는데 왜 이렇게 많이 나왔지?"라고 의아해하셨을 겁니다.
이런 문제는 토큰 수를 정확히 측정하지 않고 감으로 처리했을 때 발생합니다. GPT는 글자 수가 아닌 토큰 수로 과금하는데, 특히 한국어는 영어보다 2-3배 많은 토큰을 소비합니다.
"안녕하세요"는 겨우 5글자지만 GPT에게는 3-5개의 토큰입니다. 바로 이럴 때 필요한 것이 정확한 토큰 카운팅 구현입니다.
tiktoken 라이브러리를 사용해서 요청 전에 정확한 토큰 수를 측정하면, 비용을 예측하고 제한을 초과하지 않도록 사전에 조정할 수 있습니다.
개요
간단히 말해서, 토큰 카운팅은 텍스트를 GPT 모델에 보내기 전에 정확히 몇 개의 토큰으로 변환될지 미리 계산하는 기술입니다. 왜 이것이 실무에서 중요할까요?
첫째, OpenAI API는 토큰당 과금합니다. GPT-4는 입력 토큰당 $0.03/1K, 출력 토큰당 $0.06/1K입니다.
1만 토큰을 처리하면 $0.30-0.60가 나가므로, 대규모 서비스에서는 정확한 측정이 필수입니다. 둘째, 모델마다 토큰 제한이 있습니다.
GPT-3.5는 4K, GPT-4는 8K~128K 토큰 제한이 있어서, 초과하면 에러가 발생합니다. 미리 측정해서 분할해야 합니다.
기존에는 "대충 글자 수의 절반 정도겠지"라고 추측했다면, 이제는 tiktoken으로 정확히 측정해서 "이 요청은 정확히 1,247 토큰이므로 $0.037의 비용이 발생합니다"라고 말할 수 있습니다. 토큰 카운팅의 핵심 특징은 세 가지입니다: (1) 모델별 정확성 - GPT-3.5와 GPT-4는 다른 토크나이저 사용, 각각에 맞게 측정, (2) 다국어 지원 - 영어, 한국어, 중국어 등 언어별 토큰 차이 자동 반영, (3) 실시간 측정 - 밀리초 단위로 빠른 계산으로 실시간 애플리케이션에 적합.
이러한 특징들이 비용 효율적이고 안정적인 AI 서비스의 기반이 됩니다.
코드 예제
import tiktoken
from typing import Dict
def analyze_token_usage(text: str, model: str = "gpt-4") -> Dict:
"""텍스트의 토큰 사용량을 상세히 분석"""
# 모델별 인코더 가져오기
try:
encoding = tiktoken.encoding_for_model(model)
except KeyError:
encoding = tiktoken.get_encoding("cl100k_base") # 기본 인코더
# 토큰으로 인코딩
tokens = encoding.encode(text)
token_count = len(tokens)
# 비용 계산 (GPT-4 기준)
cost_per_1k = 0.03 if "gpt-4" in model else 0.0015
estimated_cost = (token_count / 1000) * cost_per_1k
# 상세 정보 반환
return {
"token_count": token_count,
"character_count": len(text),
"tokens_per_char": token_count / len(text) if text else 0,
"estimated_cost": f"${estimated_cost:.4f}",
"model": model
}
# 실제 사용
text = "인공지능은 우리의 미래를 바꿉니다." * 50
result = analyze_token_usage(text, "gpt-4")
print(f"토큰: {result['token_count']}, 예상 비용: {result['estimated_cost']}")
설명
이것이 하는 일: tiktoken 라이브러리를 사용해서 텍스트의 토큰 수를 정확히 측정하고, 이를 바탕으로 API 비용까지 예측하는 종합 분석 함수입니다. 첫 번째로, try-except 블록에서 encoding = tiktoken.encoding_for_model(model) 부분은 지정된 모델의 토크나이저를 안전하게 가져옵니다.
만약 모델명이 잘못되었거나 지원하지 않는 모델이면 KeyError가 발생하는데, 이 경우 "cl100k_base"라는 기본 인코더를 사용합니다. 왜 이렇게 하냐고요?
서비스 중에 에러로 중단되는 것보다는 근사값이라도 제공하는 것이 낫기 때문입니다. 그 다음으로, tokens = encoding.encode(text) 부분이 실제로 텍스트를 토큰 ID 배열로 변환합니다.
예를 들어 "AI는 미래다"는 [15836, 101, 231, 58291, 108] 같은 숫자 리스트로 변환됩니다. 각 숫자는 GPT의 어휘 사전에서 특정 의미 단위를 가리킵니다.
세 번째로, 비용 계산 부분에서 (token_count / 1000) * cost_per_1k 공식으로 예상 비용을 산출합니다. OpenAI는 1,000 토큰당 가격을 책정하므로, 이를 기준으로 비례 계산합니다.
GPT-4는 $0.03/1K, GPT-3.5는 $0.0015/1K로 큰 차이가 있습니다. 마지막으로, tokens_per_char 지표는 텍스트의 토큰 효율성을 나타냅니다.
영어는 보통 0.25(4글자당 1토큰), 한국어는 0.4-0.6(2-3글자당 1토큰) 정도입니다. 이 값이 높을수록 같은 글자 수에 비용이 더 듭니다.
여러분이 이 코드를 사용하면 다음 이점을 얻습니다: (1) 정확한 예산 관리 - 요청 전에 비용을 알아서 예산 초과 방지, (2) 최적화 지점 발견 - tokens_per_char를 보고 어떤 텍스트가 비효율적인지 파악, (3) 프로덕션 모니터링 - 실시간으로 토큰 사용량 추적해서 이상 징후 감지.
실전 팁
💡 모델명은 정확히 입력하세요. "gpt-4", "gpt-4-turbo", "gpt-3.5-turbo"는 각각 다른 토크나이저를 사용합니다. 틀리면 토큰 수가 최대 10% 차이 날 수 있습니다.
💡 대량 처리 시 인코더를 재사용하세요. encoding = tiktoken.encoding_for_model("gpt-4")를 함수 밖에서 한 번만 호출하고, 여러 텍스트에 반복 사용하면 속도가 5배 빨라집니다.
💡 시스템 메시지도 토큰에 포함됩니다. 실제 API 호출 시 "You are a helpful assistant" 같은 시스템 메시지가 매번 추가되므로, 여기에 해당하는 토큰(보통 20-50개)도 함께 계산하세요.
💡 응답 토큰도 예산에 포함하세요. GPT는 입력과 출력 모두 과금합니다. 입력이 1,000 토큰이고 출력도 1,000 토큰이면 총 2,000 토큰의 비용이 발생합니다.
💡 캐싱으로 성능을 극대화하세요. @lru_cache(maxsize=1000) 데코레이터를 함수에 추가하면 같은 텍스트는 다시 계산하지 않고 캐시에서 가져와 100배 이상 빠릅니다.
5. 동적_청크_크기_조정
시작하며
여러분이 GPT-3.5로 개발한 앱을 GPT-4로 업그레이드했을 때, 갑자기 성능이 떨어지거나 에러가 발생한 경험 있나요? 분명히 코드는 똑같은데 왜 문제가 생겼을까요?
이런 문제는 모델마다 최적의 청크 크기가 다르기 때문입니다. GPT-3.5는 4K 토큰 제한이 있어서 청크를 작게 나눠야 하지만, GPT-4는 32K~128K까지 처리할 수 있어서 큰 청크를 사용하는 것이 더 효율적입니다.
고정된 청크 크기를 쓰면 모델 변경 시 매번 수동으로 조정해야 합니다. 바로 이럴 때 필요한 것이 동적 청크 크기 조정입니다.
사용하는 GPT 모델을 감지해서 자동으로 최적의 청크 크기를 계산하고 적용하면, 모델을 바꿔도 코드 수정 없이 최고의 성능을 유지할 수 있습니다.
개요
간단히 말해서, 동적 청크 크기 조정은 GPT 모델의 토큰 제한과 특성에 맞춰 청크 크기를 자동으로 최적화하는 기법입니다. 왜 이것이 실무에서 중요할까요?
GPT 모델은 계속 발전하고 있습니다. GPT-3.5(4K), GPT-4(8K), GPT-4 Turbo(128K), 심지어 Claude나 Gemini 같은 다른 LLM도 각각 다른 컨텍스트 윈도우를 가집니다.
하드코딩된 청크 크기를 사용하면 모델마다 별도의 코드를 유지해야 하는데, 이는 유지보수 악몽입니다. 예를 들어, GPT-4 Turbo로 업그레이드했는데 여전히 500토큰 청크를 사용한다면, 128K 제한의 1% 미만만 활용하는 셈입니다.
기존에는 "청크 크기는 1000토큰으로 고정하자"라고 정했다면, 이제는 "모델 정보를 받아서 그 모델의 최대 성능을 끌어내는 크기로 자동 설정하자"는 접근이 필요합니다. 동적 조정의 핵심 특징은 세 가지입니다: (1) 모델 감지 - GPT 버전과 컨텍스트 윈도우를 자동 파악, (2) 안전 마진 적용 - 최대 제한의 70-80%만 사용해서 에러 방지, (3) 성능 최적화 - 모델이 잘 처리하는 크기로 청크를 만들어 품질 향상.
이러한 특징들이 다양한 모델을 유연하게 지원하는 확장 가능한 시스템의 기반이 됩니다.
코드 예제
import tiktoken
# 모델별 토큰 제한 데이터
MODEL_LIMITS = {
"gpt-3.5-turbo": 4096,
"gpt-4": 8192,
"gpt-4-turbo": 128000,
"gpt-4-32k": 32768
}
def get_optimal_chunk_size(model: str, safety_margin: float = 0.75) -> int:
"""모델에 맞는 최적 청크 크기 자동 계산"""
# 모델의 토큰 제한 가져오기
max_tokens = MODEL_LIMITS.get(model, 4096) # 기본값: 4096
# 시스템 메시지와 응답을 위한 여유 공간 (20%)
available_tokens = int(max_tokens * safety_margin)
# 프롬프트 오버헤드 제외 (약 100 토큰)
chunk_tokens = available_tokens - 100
# 글자 수로 변환 (한국어 기준 약 2글자당 1토큰)
chunk_size = chunk_tokens * 2
return chunk_size
# 실제 사용
for model in ["gpt-3.5-turbo", "gpt-4", "gpt-4-turbo"]:
optimal_size = get_optimal_chunk_size(model)
print(f"{model}: 최적 청크 크기 = {optimal_size}자 (약 {optimal_size//2} 토큰)")
설명
이것이 하는 일: GPT 모델명을 입력받아 그 모델의 컨텍스트 윈도우 크기를 기반으로 최적의 청크 크기를 자동으로 계산하는 함수입니다. 첫 번째로, MODEL_LIMITS 딕셔너리는 각 GPT 모델의 공식 토큰 제한을 저장합니다.
이 값들은 OpenAI 공식 문서에서 가져온 정확한 수치입니다. 새로운 모델이 출시되면 이 딕셔너리에만 추가하면 되므로 유지보수가 간단합니다.
왜 하드코딩이 아니냐고요? 모델 제한은 자주 바뀌지 않고, 중앙 집중식 관리가 더 안전하기 때문입니다.
그 다음으로, available_tokens = int(max_tokens * safety_margin) 부분은 안전 마진을 적용합니다. 예를 들어 GPT-4가 8,192 토큰을 지원한다고 해서 정확히 8,192 토큰을 보내면 안 됩니다.
시스템 메시지, 사용자 메시지 포맷, GPT 응답 공간이 모두 포함되기 때문입니다. 75%만 사용하면 6,144 토큰이 되어 안전합니다.
세 번째로, chunk_tokens = available_tokens - 100 부분은 프롬프트 오버헤드를 제외합니다. "다음 텍스트를 요약해주세요:" 같은 지시문도 토큰을 소비하므로, 약 100 토큰 정도를 미리 빼둡니다.
실제 프로젝트에서는 자주 사용하는 프롬프트의 토큰 수를 정확히 측정해서 이 값을 조정하는 것이 좋습니다. 마지막으로, chunk_size = chunk_tokens * 2 부분은 토큰을 글자 수로 변환합니다.
한국어는 평균적으로 2글자당 1토큰 정도이므로 2배를 곱합니다. 영어라면 4배 정도가 적당합니다.
더 정확하게 하려면 실제 텍스트 샘플로 토큰/글자 비율을 측정하세요. 여러분이 이 코드를 사용하면 다음 이점을 얻습니다: (1) 모델 전환 용이 - GPT-3.5에서 GPT-4로 바꿔도 코드 수정 없이 자동 최적화, (2) 에러 방지 - 안전 마진 덕분에 토큰 초과 에러 거의 발생 안 함, (3) 최대 성능 활용 - 각 모델이 제공하는 컨텍스트를 최대한 활용하여 품질 향상.
실전 팁
💡 safety_margin은 용도에 따라 조정하세요. 단순 요약이면 0.7(70%)로 낮춰도 되지만, 복잡한 분석이나 긴 응답이 필요하면 0.6(60%)까지 낮춰서 GPT 응답 공간을 확보하세요.
💡 언어별로 변환 계수가 다릅니다. 영어는 chunk_tokens * 4, 한국어는 * 2, 중국어는 * 1.5 정도가 평균입니다. 여러 언어를 지원한다면 언어 감지 후 계수를 동적으로 선택하세요.
💡 실시간 API로 최신 제한을 가져올 수도 있습니다. OpenAI API의 /models 엔드포인트를 호출하면 최신 모델 스펙을 받을 수 있어서, 하드코딩된 딕셔너리보다 항상 최신 정보를 유지할 수 있습니다.
💡 프롬프트 길이를 정확히 측정하세요. chunk_tokens = available_tokens - 100의 100 대신, 실제 프롬프트 템플릿을 tiktoken으로 측정한 값을 사용하면 더 정확합니다.
💡 청크 크기를 로그에 기록하세요. logging.info(f"Using chunk_size={chunk_size} for model={model}")처럼 남기면 프로덕션에서 문제 발생 시 디버깅에 큰 도움이 됩니다.
6. 문장_경계_보존
시작하며
여러분이 GPT에게 긴 기사를 요약해달라고 했는데, "중요한 부분이 빠졌어요"라는 결과를 받은 적 있나요? 청크를 확인해보니 핵심 문장이 중간에서 잘려 있었습니다.
이런 문제는 청크 분할이 문장 경계를 무시하고 기계적으로 진행될 때 발생합니다. 예를 들어, "그러나 이것은 매우 중요한 발견입니다"라는 문장이 "그러나 이것은 매우"까지는 첫 번째 청크에, "중요한 발견입니다"는 두 번째 청크에 들어가면, GPT는 "그러나 이것은 매우"를 불완전한 문장으로 인식해서 무시하거나 오해합니다.
바로 이럴 때 필요한 것이 문장 경계 보존입니다. 청크를 나눌 때 문장이 완전하게 유지되도록 분할 지점을 조정하면, GPT가 모든 정보를 완전한 형태로 이해할 수 있습니다.
개요
간단히 말해서, 문장 경계 보존은 텍스트를 분할할 때 문장이 절대 중간에서 잘리지 않도록 분할 지점을 조정하는 기술입니다. 왜 이것이 중요할까요?
GPT는 완전한 문장을 이해하도록 학습되었습니다. 불완전한 문장 조각을 받으면 문맥을 오해하거나 중요한 정보를 놓칠 수 있습니다.
예를 들어, "제품 가격은 10만원이지만"까지만 보면 GPT는 가격 정보만 인식하지만, "제품 가격은 10만원이지만 특별 할인으로 5만원입니다"를 완전히 보면 할인 정보까지 정확히 파악합니다. RAG 시스템에서 특히 중요한데, 검색된 청크가 불완전하면 아예 잘못된 답변을 생성할 수 있습니다.
기존에는 "1000자가 되면 무조건 자르자"라고 했다면, 이제는 "1000자 근처에서 가장 가까운 문장 끝을 찾아서 거기서 자르자"는 접근이 필요합니다. 문장 경계 보존의 핵심 특징은 세 가지입니다: (1) 완전성 보장 - 모든 청크가 완전한 문장들로만 구성, (2) 유연한 크기 - 목표 크기를 약간 초과하거나 미달하더라도 문장 완전성 우선, (3) 언어 인식 - 영어의 마침표와 한국어의 종결어미를 구분해서 정확한 경계 감지.
이러한 특징들이 GPT가 정확한 이해를 바탕으로 고품질 답변을 생성하는 기반이 됩니다.
코드 예제
import re
def split_at_sentence_boundary(text: str, target_size: int = 1000, max_deviation: int = 200) -> list:
"""문장 경계를 보존하면서 텍스트 분할"""
chunks = []
current_pos = 0
# 문장 끝 패턴 (한국어 + 영어)
sentence_endings = re.compile(r'([.!?])\s+|([.!?])$|([다|요|음|까|냐])\s')
while current_pos < len(text):
# 목표 크기만큼 자르기
end_pos = min(current_pos + target_size, len(text))
chunk_candidate = text[current_pos:end_pos]
# 목표 크기 근처에서 문장 끝 찾기
search_start = max(0, len(chunk_candidate) - max_deviation)
search_area = chunk_candidate[search_start:]
# 마지막 문장 끝 찾기
matches = list(sentence_endings.finditer(search_area))
if matches:
last_match = matches[-1]
actual_end = search_start + last_match.end()
final_chunk = chunk_candidate[:actual_end].strip()
else:
# 문장 끝을 못 찾으면 목표 크기 그대로 사용
final_chunk = chunk_candidate.strip()
chunks.append(final_chunk)
current_pos += len(final_chunk)
return chunks
# 실제 사용
article = "AI는 발전합니다. 미래가 기대됩니다. 우리는 준비해야 합니다." * 20
result = split_at_sentence_boundary(article, target_size=200)
print(f"{len(result)}개 청크, 모든 문장 완전함")
설명
이것이 하는 일: 목표 청크 크기 근처에서 가장 가까운 문장 끝을 찾아 그 지점에서 분할하는 함수입니다. 첫 번째로, sentence_endings 정규표현식은 문장의 끝을 감지합니다.
([.!?])\s+는 영어 문장 끝(마침표, 느낌표, 물음표 + 공백)을, ([다|요|음|까|냐])\s는 한국어 종결어미를 찾습니다. 왜 이렇게 복잡하냐고요?
영어와 한국어는 문장 구조가 완전히 다르기 때문에, 두 패턴을 모두 감지해야 정확합니다. 그 다음으로, search_start = max(0, len(chunk_candidate) - max_deviation) 부분은 검색 범위를 제한합니다.
예를 들어 목표가 1000자이고 max_deviation이 200이면, 800~1000자 구간에서만 문장 끝을 찾습니다. 왜 전체를 검색하지 않냐고요?
너무 앞쪽에서 끝내면 청크가 비효율적으로 작아지고, 너무 뒤로 가면 토큰 제한을 초과할 수 있기 때문입니다. 세 번째로, matches = list(sentence_endings.finditer(search_area)) 부분은 검색 범위 내 모든 문장 끝을 찾습니다.
matches[-1]로 마지막 문장 끝을 선택하면, 가능한 한 목표 크기에 가깝게 청크를 채우면서도 문장 경계를 지킬 수 있습니다. 마지막으로, 문장 끝을 못 찾은 경우를 대비한 폴백 로직이 있습니다.
예를 들어 200자 구간에 문장 끝이 하나도 없으면(매우 긴 문장), 어쩔 수 없이 목표 크기에서 자릅니다. 완벽하지는 않지만 시스템이 멈추지 않도록 보장합니다.
여러분이 이 코드를 사용하면 다음 이점을 얻습니다: (1) 정보 무결성 - 모든 문장이 완전하게 유지되어 GPT가 정확히 이해, (2) 검색 품질 향상 - RAG 시스템에서 청크를 검색할 때 불완전한 정보 제공 방지, (3) 유연성 - max_deviation으로 청크 크기 일관성과 문장 완전성 사이 균형 조절 가능.
실전 팁
💡 max_deviation은 target_size의 20% 정도가 적당합니다. 1000자 청크라면 200자 편차를 허용해서, 800~1200자 범위에서 문장 끝을 찾으세요.
💡 언어별 패턴을 추가하세요. 중국어는 [。!?], 일본어는 [。!?]를 sentence_endings에 추가하면 다국어 지원이 가능합니다.
💡 대화체는 특별 처리가 필요합니다. 카카오톡 대화나 댓글은 "ㅋㅋ", "ㅎㅎ" 같은 이모티콘으로 끝나므로, 이들도 문장 끝으로 인식하도록 패턴을 확장하세요.
💡 코드나 마크다운은 별도 처리하세요. 코드 블록 내부의 마침표는 문장 끝이 아니므로, 분할 전에 코드 블록을 추출하고 나중에 재삽입하는 전처리가 필요합니다.
💡 디버깅 시 경계를 시각화하세요. print(f"청크 끝: ...{chunk[-30:]}")처럼 각 청크의 마지막 30자를 출력하면 문장이 제대로 끝났는지 쉽게 확인할 수 있습니다.
7. 메타데이터_추가
시작하며
여러분이 RAG 시스템에서 GPT의 답변을 받았는데, "이 정보가 원본 문서의 어디서 왔는지 알 수 없어요"라는 상황을 겪은 적 있나요? 사용자는 출처를 확인하고 싶어 하는데 추적이 불가능합니다.
이런 문제는 청크를 만들 때 원본 정보를 함께 저장하지 않아서 발생합니다. 청크만 있고 "이게 어느 문서의 몇 페이지인지, 원래 순서가 몇 번째인지" 같은 메타데이터가 없으면, 나중에 출처를 추적하거나 청크를 재조립하는 것이 불가능합니다.
특히 수천 개의 문서를 처리하는 대규모 시스템에서는 치명적입니다. 바로 이럴 때 필요한 것이 메타데이터 추가입니다.
각 청크에 문서 ID, 청크 순서, 원본 위치 같은 정보를 태그로 붙이면, 나중에 추적, 디버깅, 재조립이 모두 가능합니다.
개요
간단히 말해서, 메타데이터 추가는 각 청크에 원본 문서, 위치, 순서 같은 추가 정보를 함께 저장해서 추적 가능성을 확보하는 기술입니다. 왜 메타데이터가 실무에서 필수일까요?
첫째, 출처 표시입니다. 사용자에게 "이 정보는 '2024 AI 보고서' 12페이지에서 가져왔습니다"라고 알려주면 신뢰도가 크게 높아집니다.
둘째, 디버깅입니다. GPT가 이상한 답변을 했을 때, 어느 청크가 문제인지 메타데이터로 즉시 추적해서 원본을 확인할 수 있습니다.
셋째, 재조립입니다. 청크를 다시 원본 순서로 합쳐야 할 때, 순서 정보가 없으면 불가능합니다.
기존에는 "텍스트만 저장하면 되지 뭐"라고 생각했다면, 이제는 "텍스트 + 출처 + 위치 + 타임스탬프를 하나의 패키지로 관리하자"는 접근이 필요합니다. 메타데이터의 핵심 특징은 세 가지입니다: (1) 추적 가능성 - 모든 청크의 출처와 위치를 정확히 파악 가능, (2) 검색 필터링 - 특정 문서나 날짜 범위의 청크만 검색 가능, (3) 재조립 가능성 - 순서 정보를 바탕으로 원본 문서 복원 가능.
이러한 특징들이 프로덕션 수준의 RAG 시스템에서 필수적입니다.
코드 예제
from datetime import datetime
from typing import List, Dict
def create_chunks_with_metadata(text: str, doc_id: str, source: str, chunk_size: int = 1000) -> List[Dict]:
"""메타데이터가 포함된 청크 생성"""
chunks_data = []
# 단순 분할 (실제로는 RecursiveCharacterTextSplitter 사용)
chunks = [text[i:i+chunk_size] for i in range(0, len(text), chunk_size)]
# 각 청크에 메타데이터 추가
for idx, chunk_text in enumerate(chunks):
chunk_with_meta = {
"chunk_id": f"{doc_id}_chunk_{idx}", # 고유 ID
"content": chunk_text, # 실제 텍스트
"metadata": {
"document_id": doc_id, # 원본 문서 ID
"source": source, # 파일명이나 URL
"chunk_index": idx, # 청크 순서
"total_chunks": len(chunks), # 전체 청크 수
"char_start": idx * chunk_size, # 원본에서의 시작 위치
"char_end": min((idx + 1) * chunk_size, len(text)), # 끝 위치
"created_at": datetime.now().isoformat() # 생성 시각
}
}
chunks_data.append(chunk_with_meta)
return chunks_data
# 실제 사용
doc_text = "AI 기술이 발전하고 있습니다. " * 100
chunks = create_chunks_with_metadata(doc_text, "doc_001", "ai_report_2024.pdf", 200)
print(f"청크 ID: {chunks[0]['chunk_id']}")
print(f"출처: {chunks[0]['metadata']['source']}")
print(f"위치: {chunks[0]['metadata']['chunk_index'] + 1}/{chunks[0]['metadata']['total_chunks']}")
설명
이것이 하는 일: 텍스트를 청크로 나눌 때 각 청크에 원본 추적을 위한 상세한 메타데이터를 자동으로 붙이는 함수입니다. 첫 번째로, chunk_id = f"{doc_id}_chunk_{idx}" 부분은 각 청크의 고유 ID를 생성합니다.
예를 들어 "doc_001_chunk_0", "doc_001_chunk_1" 같은 형식입니다. 왜 이렇게 하냐고요?
벡터 DB에 저장하거나 검색할 때 각 청크를 구분할 수 있는 고유 키가 필요하기 때문입니다. 이 ID만 있으면 나중에 특정 청크를 정확히 찾아낼 수 있습니다.
그 다음으로, metadata 딕셔너리에 여러 정보가 저장됩니다. document_id는 원본 문서를 가리키고, source는 파일명이나 URL을 저장합니다.
chunk_index와 total_chunks는 "이게 전체 10개 중 3번째 청크"라는 정보를 제공해서, 사용자에게 "이 답변은 문서의 30% 지점에서 가져왔습니다" 같은 UX를 제공할 수 있습니다. 세 번째로, char_start와 char_end는 원본 텍스트에서의 정확한 위치를 기록합니다.
예를 들어 두 번째 청크(idx=1)라면 char_start=1000, char_end=2000이 되어, "원본의 1000~2000번째 글자"라는 걸 정확히 알 수 있습니다. 이 정보로 원본 문서에서 해당 부분을 하이라이트해서 보여줄 수 있습니다.
마지막으로, created_at은 청크 생성 시각을 ISO 형식으로 기록합니다. 나중에 "오래된 청크는 다시 생성"하거나 "최근 24시간 내 문서만 검색" 같은 기능을 구현할 때 유용합니다.
여러분이 이 코드를 사용하면 다음 이점을 얻습니다: (1) 완벽한 출처 표시 - 사용자에게 "이 정보는 ai_report_2024.pdf의 3/10 청크에서 가져왔습니다" 같은 투명한 정보 제공, (2) 효율적인 디버깅 - GPT 답변이 이상할 때 chunk_id로 즉시 원본 청크 확인, (3) 고급 검색 - 특정 문서나 날짜 범위로 필터링해서 검색 정확도 향상.
실전 팁
💡 메타데이터를 벡터 DB에 함께 저장하세요. Pinecone, Weaviate 같은 벡터 DB는 메타데이터 필터링을 지원해서, "2024년 이후 + AI 카테고리"처럼 조건부 검색이 가능합니다.
💡 추가 정보를 커스터마이징하세요. 문서 카테고리, 저자, 언어, 중요도 같은 비즈니스 로직에 맞는 필드를 metadata에 추가하면 더 강력한 검색이 가능합니다.
💡 메타데이터 크기를 주의하세요. 벡터 DB는 메타데이터도 저장 공간을 차지하므로, 불필요한 정보는 제외하고 꼭 필요한 것만 포함하세요. 일반적으로 메타데이터는 content의 10% 이하가 적당합니다.
💡 청크 버전 관리를 고려하세요. 원본 문서가 수정되면 청크도 다시 생성해야 하는데, version: 2 같은 필드를 추가하면 어느 버전인지 추적할 수 있습니다.
💡 JSON으로 직렬화 가능하게 유지하세요. datetime 객체는 그대로 저장할 수 없으므로 isoformat()로 문자열 변환해야 합니다. 모든 메타데이터가 JSON 직렬화 가능하면 DB 저장과 API 전송이 쉽습니다.
8. 배치_처리_최적화
시작하며
여러분이 수천 개의 문서를 GPT로 처리하려는데, 하나씩 처리하니까 몇 시간이 걸리는 상황을 겪은 적 있나요? "이렇게 느리면 실무에서 못 쓰겠는데..."라는 생각이 들었을 겁니다.
이런 문제는 단일 스레드로 순차 처리할 때 발생합니다. 예를 들어 문서 하나를 처리하는 데 1초가 걸린다면, 10,000개 문서는 약 2.8시간이 필요합니다.
API 호출 대기 시간이 대부분이므로, 이 시간 동안 CPU는 거의 놀고 있습니다. 네트워크 I/O를 기다리는 동안 다른 작업을 할 수 있는데 안 하는 셈입니다.
바로 이럴 때 필요한 것이 배치 처리 최적화입니다. 여러 문서를 동시에 처리하는 병렬 처리와, GPT API의 배치 엔드포인트를 활용하면 처리 시간을 10~100배 단축할 수 있습니다.
개요
간단히 말해서, 배치 처리 최적화는 여러 텍스트를 한꺼번에 또는 동시에 처리해서 전체 처리 시간을 극적으로 단축하는 기술입니다. 왜 이것이 실무에서 중요할까요?
대부분의 AI 애플리케이션은 대량의 데이터를 처리해야 합니다. 뉴스 요약 서비스라면 하루에 수천 건의 기사를, 고객 문의 분석 시스템이라면 수만 건의 티켓을 처리해야 합니다.
순차 처리로는 시간이 너무 오래 걸려서 실시간 서비스가 불가능합니다. 병렬 처리를 사용하면 같은 시간에 10배 많은 작업을 처리할 수 있어 비용 효율성도 크게 향상됩니다.
기존에는 "for 루프로 하나씩 돌리면 되지"라고 생각했다면, 이제는 "동시에 여러 개를 병렬로 처리하고, API 호출도 배치로 묶자"는 접근이 필요합니다. 배치 처리의 핵심 특징은 세 가지입니다: (1) 병렬 실행 - asyncio나 ThreadPool로 여러 작업 동시 실행, (2) API 배치 - OpenAI의 Batch API로 최대 50% 비용 절감, (3) 오류 처리 - 일부 실패해도 나머지는 계속 진행하는 견고성.
이러한 특징들이 대규모 프로덕션 시스템의 성능과 비용 효율성을 결정합니다.
코드 예제
import asyncio
from typing import List
import tiktoken
async def process_single_text(text: str, model: str = "gpt-4") -> dict:
"""단일 텍스트 처리 (시뮬레이션)"""
# 실제로는 OpenAI API 호출
await asyncio.sleep(0.1) # API 대기 시뮬레이션
encoding = tiktoken.encoding_for_model(model)
token_count = len(encoding.encode(text))
return {
"text": text[:50] + "...",
"tokens": token_count,
"status": "success"
}
async def batch_process_texts(texts: List[str], batch_size: int = 10) -> List[dict]:
"""여러 텍스트를 배치로 병렬 처리"""
results = []
# 배치 크기만큼 묶어서 처리
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# 배치 내 모든 텍스트를 동시에 처리
tasks = [process_single_text(text) for text in batch]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
# 결과 수집 (에러 처리 포함)
for result in batch_results:
if isinstance(result, Exception):
results.append({"status": "error", "error": str(result)})
else:
results.append(result)
print(f"배치 {i//batch_size + 1} 완료: {len(batch)}개 처리")
return results
# 실제 사용
documents = [f"문서 {i}: AI 기술 설명 " * 50 for i in range(50)]
results = asyncio.run(batch_process_texts(documents, batch_size=10))
print(f"총 {len(results)}개 문서 처리 완료")
설명
이것이 하는 일: Python의 asyncio를 사용해서 여러 텍스트 처리 작업을 동시에 실행하고, 배치 단위로 관리해서 효율성과 안정성을 모두 확보하는 함수입니다. 첫 번째로, async def process_single_text() 함수는 비동기 함수입니다.
async와 await 키워드를 사용하면, API 호출 대기 중에 CPU가 다른 작업을 할 수 있습니다. 예를 들어 GPT API 응답을 기다리는 0.5초 동안, 다른 10개의 API 요청을 동시에 보낼 수 있어서 전체 시간이 크게 단축됩니다.
왜 비동기가 필요하냐고요? GPT API 호출은 네트워크 I/O이므로, 동기 방식으로 하면 응답 대기 시간이 모두 낭비되기 때문입니다.
그 다음으로, for i in range(0, len(texts), batch_size) 부분은 전체 텍스트를 배치 크기(여기서는 10개)씩 묶습니다. 왜 한 번에 다 안 하고 배치로 나누냐고요?
API 서버에 동시에 1000개 요청을 보내면 rate limit에 걸리거나 서버가 과부하될 수 있습니다. 10개씩 배치로 나누면 안정적으로 처리하면서도 충분히 빠릅니다.
세 번째로, tasks = [process_single_text(text) for text in batch] 부분은 배치 내 모든 작업을 태스크 리스트로 준비하고, await asyncio.gather(*tasks)가 이들을 동시에 실행합니다. gather()는 모든 태스크가 완료될 때까지 기다렸다가 결과를 리스트로 반환합니다.
예를 들어 10개 작업이 각각 1초 걸린다면, 순차적으로는 10초지만 gather()를 쓰면 1초만에 끝납니다. 마지막으로, return_exceptions=True 파라미터는 일부 작업이 실패해도 나머지는 계속 진행하도록 합니다.
예를 들어 50개 중 3개가 API 에러로 실패해도, 나머지 47개는 정상적으로 처리되고, 실패한 3개는 에러 정보와 함께 결과에 포함됩니다. 여러분이 이 코드를 사용하면 다음 이점을 얻습니다: (1) 10~100배 속도 향상 - 순차 처리 대비 극적인 성능 개선, (2) 비용 절감 - 같은 시간에 더 많은 작업 처리로 인프라 비용 감소, (3) 견고성 - 일부 실패가 전체를 멈추지 않음.
실전 팁
💡 batch_size는 API rate limit에 맞추세요. OpenAI는 분당 요청 수 제한이 있으므로, 여러분의 플랜에 맞는 배치 크기를 설정하세요. Tier 1은 분당 500 요청이므로 batch_size=50 정도가 안전합니다.
💡 재시도 로직을 추가하세요. API 호출은 네트워크 문제로 가끔 실패하므로, tenacity 라이브러리로 자동 재시도를 구현하면 성공률이 99% 이상으로 올라갑니다.
💡 진행 상황을 시각화하세요. tqdm 라이브러리로 프로그레스 바를 추가하면 "3500/10000 (35%) 처리 중..."처럼 실시간으로 진행 상황을 볼 수 있어 사용자 경험이 좋아집니다.
💡 OpenAI Batch API를 고려하세요. 실시간이 아닌 배치 작업이라면 OpenAI의 공식 Batch API를 사용하면 비용이 50% 저렴합니다. 24시간 내 완료되면 되는 작업에 적합합니다.
💡 결과를 스트리밍하세요. 모든 배치가 끝날 때까지 기다리지 말고, 배치 하나가 끝날 때마다 결과를 DB에 저장하거나 사용자에게 전달하면 체감 속도가 더 빨라집니다.