이미지 로딩 중...
AI Generated
2025. 11. 16. · 3 Views
실무 파인튜닝 데이터셋 설계 완벽 가이드
AI 모델을 우리 서비스에 맞게 최적화하는 파인튜닝 데이터셋 설계 전략을 알아봅니다. 데이터 수집부터 품질 관리, 데이터 증강, 실전 검증까지 실무에서 바로 적용할 수 있는 구체적인 방법을 다룹니다.
목차
- 파인튜닝 데이터셋 기본 구조 설계 - 성공적인 학습을 위한 첫걸음
- 고품질 데이터 수집 전략 - 쓰레기를 넣으면 쓰레기가 나온다
- 데이터 증강 기법 - 적은 데이터로 더 많이 학습하기
- 프롬프트-응답 쌍 최적화 - 모델이 학습하기 좋은 형태 만들기
- 밸런스 있는 데이터셋 구성 - 편향 없는 학습을 위한 균형 맞추기
- 검증 데이터셋 분리 - 과적합 방지와 성능 측정
- 메타데이터 활용 - 데이터에 문맥 정보 추가하기
- 품질 검증 파이프라인 - 자동화된 품질 관리 시스템
- 실전 테스트와 반복 개선 - 데이터셋 품질 검증하기
1. 파인튜닝 데이터셋 기본 구조 설계 - 성공적인 학습을 위한 첫걸음
시작하며
여러분이 AI 챗봇을 만들었는데 일반적인 대화는 잘하지만 우리 회사의 전문 용어나 업무 프로세스는 전혀 이해하지 못하는 상황을 겪어본 적 있나요? 예를 들어, "PO 승인 후 스프린트에 추가해주세요"라고 하면 엉뚱한 답변을 하는 경우가 있습니다.
이런 문제는 기본 모델이 특정 도메인의 데이터로 학습되지 않았기 때문에 발생합니다. 아무리 강력한 GPT-4나 Claude라도 여러분의 회사 업무나 서비스 특성을 처음부터 알 수는 없습니다.
바로 이럴 때 필요한 것이 파인튜닝 데이터셋입니다. 올바른 구조로 설계된 데이터셋은 모델이 여러분의 도메인을 이해하고 정확한 응답을 생성하도록 만들어줍니다.
개요
간단히 말해서, 파인튜닝 데이터셋은 AI 모델을 우리 서비스에 맞게 재학습시키기 위한 예제 데이터의 집합입니다. 일반적으로 입력(prompt)과 출력(completion) 쌍으로 구성되며, 모델이 "이런 질문에는 이렇게 답해야 한다"는 패턴을 학습하게 됩니다.
예를 들어, 고객 서비스 챗봇을 만든다면 실제 고객 문의와 상담사의 답변을 데이터로 사용할 수 있습니다. 기존에는 모든 상황을 프롬프트 엔지니어링으로 해결하려 했다면, 이제는 파인튜닝으로 모델 자체를 특화시킬 수 있습니다.
프롬프트가 길어지면 비용도 증가하고 응답 속도도 느려지지만, 파인튜닝된 모델은 짧은 입력으로도 정확한 결과를 냅니다. 핵심 특징은 첫째, 일관된 포맷으로 구조화되어야 한다는 점, 둘째, 충분한 양의 품질 높은 예제가 필요하다는 점, 셋째, 모델의 타겟 작업을 명확히 반영해야 한다는 점입니다.
이러한 특징들이 모델의 성능을 결정하는 핵심 요소가 됩니다.
코드 예제
# 파인튜닝 데이터셋의 기본 구조 (JSONL 포맷)
import json
# 단일 학습 예제 생성
def create_training_example(user_message, assistant_response, system_prompt=None):
example = {
"messages": [
{"role": "user", "content": user_message},
{"role": "assistant", "content": assistant_response}
]
}
# 시스템 프롬프트가 있으면 맨 앞에 추가
if system_prompt:
example["messages"].insert(0, {"role": "system", "content": system_prompt})
return example
# 데이터셋 생성 및 저장
system_prompt = "당신은 전문 개발자 멘토입니다. 초급 개발자에게 친절하게 설명합니다."
examples = [
create_training_example(
"파이썬 리스트 컴프리헨션이 뭔가요?",
"리스트 컴프리헨션은 반복문을 한 줄로 간결하게 작성하는 파이썬 문법입니다. 예: [x*2 for x in range(5)]",
system_prompt
),
create_training_example(
"API 응답이 느려요",
"먼저 데이터베이스 쿼리를 확인해보세요. 인덱스가 없거나 N+1 문제가 있을 수 있습니다.",
system_prompt
)
]
# JSONL 파일로 저장 (한 줄에 하나의 JSON 객체)
with open('training_data.jsonl', 'w', encoding='utf-8') as f:
for example in examples:
f.write(json.dumps(example, ensure_ascii=False) + '\n')
설명
이것이 하는 일: 이 코드는 OpenAI나 다른 LLM 플랫폼에서 사용할 수 있는 표준 파인튜닝 데이터셋을 생성합니다. 첫 번째로, create_training_example 함수는 사용자 메시지와 AI 응답을 받아 표준 대화 포맷으로 변환합니다.
이 포맷은 role(역할)과 content(내용)로 구성되는데, OpenAI API와 대부분의 LLM이 이 구조를 사용하기 때문에 호환성이 높습니다. system 메시지를 선택적으로 추가할 수 있어서 모델의 전반적인 행동 방식을 정의할 수 있습니다.
두 번째로, 실제 학습 예제들을 리스트로 만듭니다. 각 예제는 실제 사용자가 물어볼 만한 질문과 모델이 해야 할 답변으로 구성됩니다.
예를 들어 개발자 멘토 역할을 하는 모델을 만든다면, 실제 초급 개발자들이 자주 하는 질문과 전문가의 답변을 데이터로 사용합니다. 세 번째로, JSONL(JSON Lines) 포맷으로 파일을 저장합니다.
JSONL은 한 줄에 하나의 JSON 객체를 저장하는 포맷인데, 대용량 데이터를 스트리밍 방식으로 처리할 수 있어서 메모리 효율적입니다. 각 줄이 독립적이므로 데이터를 추가하거나 샘플링하기도 쉽습니다.
여러분이 이 코드를 사용하면 수십, 수백 개의 학습 예제를 일관된 포맷으로 쉽게 생성할 수 있습니다. 실무에서는 실제 고객 대화 로그, FAQ, 상담 기록 등을 이 포맷으로 변환하여 사용하면 됩니다.
데이터가 많을수록, 품질이 좋을수록 모델의 성능이 향상됩니다.
실전 팁
💡 최소 50-100개의 고품질 예제부터 시작하세요. 적은 데이터로도 특정 작업에서는 효과를 볼 수 있지만, 일반적으로 수백 개 이상이 권장됩니다.
💡 시스템 프롬프트는 모든 예제에 일관되게 사용하세요. 모델의 전반적인 톤과 스타일을 정의하는 중요한 요소입니다.
💡 JSONL 파일을 생성한 후 반드시 몇 개 줄을 직접 열어서 포맷이 올바른지 확인하세요. 잘못된 JSON 구조가 하나라도 있으면 전체 학습이 실패할 수 있습니다.
💡 UTF-8 인코딩을 사용하여 한글이나 특수문자가 깨지지 않도록 주의하세요. ensure_ascii=False 옵션이 중요합니다.
💡 각 예제의 길이(토큰 수)를 비슷하게 유지하면 학습이 더 안정적입니다. 너무 짧거나 긴 예제가 섞이면 모델이 혼란스러워할 수 있습니다.
2. 고품질 데이터 수집 전략 - 쓰레기를 넣으면 쓰레기가 나온다
시작하며
여러분이 열심히 데이터를 모아서 파인튜닝을 했는데 모델이 이상한 답변을 하거나 편향된 결과를 내는 경험을 해본 적 있나요? 예를 들어, 웹에서 크롤링한 데이터로 학습시켰더니 비속어나 부정확한 정보를 그대로 재현하는 경우가 있습니다.
이런 문제는 "Garbage In, Garbage Out"이라는 머신러닝의 기본 원칙에서 비롯됩니다. 아무리 좋은 모델과 학습 알고리즘을 사용해도 데이터 품질이 낮으면 결과도 나쁠 수밖에 없습니다.
바로 이럴 때 필요한 것이 체계적인 데이터 수집 전략입니다. 어디서, 어떻게 데이터를 모을지, 어떤 기준으로 필터링할지를 미리 계획하면 학습 효율이 크게 향상됩니다.
개요
간단히 말해서, 고품질 데이터 수집은 모델이 학습할 가치가 있는 정확하고 관련성 높은 데이터를 선별하는 과정입니다. 데이터 출처는 크게 세 가지로 나뉩니다: 실제 사용자 로그(가장 좋음), 전문가가 작성한 콘텐츠(품질 높음), 자동 생성 또는 크롤링(양은 많지만 품질 관리 필요).
예를 들어, 고객 서비스 챗봇을 만든다면 실제 상담 기록이 가장 좋은 데이터 소스입니다. 기존에는 무작정 많은 데이터를 모으는 것이 중요하다고 생각했다면, 이제는 적더라도 품질이 높은 데이터를 선별하는 것이 더 효과적이라는 것이 밝혀졌습니다.
100개의 완벽한 예제가 1000개의 노이즈 섞인 데이터보다 나을 수 있습니다. 핵심 기준은 첫째, 정확성(사실에 기반한 정보), 둘째, 관련성(타겟 작업과 직접적 연관), 셋째, 다양성(다양한 상황과 표현 포함), 넷째, 일관성(스타일과 포맷의 통일)입니다.
이 네 가지를 모두 만족하는 데이터가 이상적입니다.
코드 예제
# 데이터 품질 검증 및 필터링 시스템
import re
from typing import List, Dict
import json
class DataQualityChecker:
def __init__(self, min_length=10, max_length=2000,
forbidden_words=None):
self.min_length = min_length # 최소 글자 수
self.max_length = max_length # 최대 글자 수
# 금지어 리스트 (비속어, 민감 정보 패턴 등)
self.forbidden_words = forbidden_words or ['비속어1', '비속어2']
def check_length(self, text: str) -> bool:
"""텍스트 길이가 적절한지 확인"""
return self.min_length <= len(text) <= self.max_length
def check_forbidden_words(self, text: str) -> bool:
"""금지어 포함 여부 확인"""
text_lower = text.lower()
return not any(word in text_lower for word in self.forbidden_words)
def check_personal_info(self, text: str) -> bool:
"""개인정보 패턴 확인 (이메일, 전화번호 등)"""
# 이메일 패턴
email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
# 전화번호 패턴 (한국)
phone_pattern = r'\b01[016789]-?\d{3,4}-?\d{4}\b'
has_email = bool(re.search(email_pattern, text))
has_phone = bool(re.search(phone_pattern, text))
return not (has_email or has_phone)
def validate_example(self, example: Dict) -> tuple[bool, str]:
"""전체 검증 수행"""
messages = example.get('messages', [])
if not messages:
return False, "메시지가 비어있음"
for msg in messages:
content = msg.get('content', '')
if not self.check_length(content):
return False, f"부적절한 길이: {len(content)} 글자"
if not self.check_forbidden_words(content):
return False, "금지어 포함"
if not self.check_personal_info(content):
return False, "개인정보 포함 가능성"
return True, "통과"
# 사용 예시
checker = DataQualityChecker(min_length=5, max_length=500)
# 데이터셋 필터링
raw_data = [
{"messages": [{"role": "user", "content": "안녕하세요"},
{"role": "assistant", "content": "안녕하세요! 무엇을 도와드릴까요?"}]},
{"messages": [{"role": "user", "content": "제 전화번호는 010-1234-5678입니다"},
{"role": "assistant", "content": "확인했습니다"}]},
]
validated_data = []
for example in raw_data:
is_valid, reason = checker.validate_example(example)
if is_valid:
validated_data.append(example)
else:
print(f"제외됨: {reason}")
print(f"총 {len(raw_data)}개 중 {len(validated_data)}개 통과")
설명
이것이 하는 일: 이 코드는 수집한 원시 데이터에서 품질이 낮거나 문제가 있는 데이터를 자동으로 걸러내는 검증 시스템을 구현합니다. 첫 번째로, DataQualityChecker 클래스는 여러 가지 품질 기준을 설정합니다.
너무 짧거나 긴 텍스트는 학습에 방해가 되므로 최소/최대 길이를 설정하고, 비속어나 부적절한 표현이 포함된 데이터는 금지어 리스트로 걸러냅니다. 실무에서는 이 금지어 리스트를 도메인에 맞게 확장해야 합니다.
두 번째로, 정규표현식을 사용하여 개인정보를 탐지합니다. 이메일 주소나 전화번호 같은 민감 정보가 학습 데이터에 포함되면 프라이버시 문제가 발생할 수 있습니다.
GDPR이나 개인정보보호법을 준수하기 위해서는 이런 정보를 반드시 제거해야 합니다. 정규표현식 패턴은 필요에 따라 주민등록번호, 신용카드 번호 등으로 확장할 수 있습니다.
세 번째로, validate_example 메서드는 모든 검증을 통합하여 수행합니다. 각 메시지를 순회하면서 모든 조건을 체크하고, 하나라도 실패하면 해당 예제를 제외합니다.
어떤 이유로 제외되었는지 명확히 알려주기 때문에 데이터 개선에도 도움이 됩니다. 네 번째로, 실제 데이터셋을 필터링하는 과정을 보여줍니다.
원시 데이터에서 검증을 통과한 데이터만 선별하여 새로운 리스트를 만듭니다. 제외된 데이터는 로그로 남겨서 나중에 수동으로 검토하거나 개선할 수 있습니다.
여러분이 이 코드를 사용하면 수천, 수만 개의 데이터를 자동으로 검증하여 고품질 데이터셋만 선별할 수 있습니다. 실무에서는 여기에 더 많은 규칙을 추가할 수 있습니다: 중복 제거, 언어 감지, 감정 분석, 도메인 특화 검증 등.
데이터 품질이 향상되면 모델 성능도 자연스럽게 좋아집니다.
실전 팁
💡 실제 사용자 로그를 사용할 때는 반드시 개인정보를 익명화하세요. 정규표현식 외에도 NER(개체명 인식) 모델을 사용하면 더 정확합니다.
💡 데이터 검증 규칙을 너무 엄격하게 하면 좋은 데이터까지 버릴 수 있습니다. 처음에는 느슨하게 시작해서 점진적으로 강화하세요.
💡 제외된 데이터는 별도로 저장해두세요. 나중에 검토하면 검증 규칙을 개선할 인사이트를 얻을 수 있습니다.
💡 도메인 전문가에게 샘플 데이터를 검토받으세요. 자동화로는 찾기 어려운 미묘한 오류나 부적절한 표현을 발견할 수 있습니다.
💡 A/B 테스트로 품질 기준의 효과를 검증하세요. 엄격한 필터링 vs 느슨한 필터링으로 학습한 모델을 비교해보면 최적의 기준을 찾을 수 있습니다.
3. 데이터 증강 기법 - 적은 데이터로 더 많이 학습하기
시작하며
여러분이 파인튜닝을 하고 싶은데 실제 데이터가 너무 적어서 고민하는 상황을 겪어본 적 있나요? 예를 들어, 법률 자문 챗봇을 만들고 싶은데 실제 상담 사례가 50개밖에 없어서 학습이 부족한 경우가 있습니다.
이런 문제는 특히 전문 도메인에서 자주 발생합니다. 의료, 법률, 금융 등의 분야는 데이터를 구하기 어렵고, 전문가의 검증이 필요해서 비용도 많이 듭니다.
적은 데이터로는 과적합(overfitting)이 발생하여 학습 데이터는 잘 맞추지만 새로운 상황에서는 실패합니다. 바로 이럴 때 필요한 것이 데이터 증강(Data Augmentation)입니다.
기존 데이터를 변형하거나 LLM을 활용하여 새로운 예제를 생성함으로써 데이터셋의 크기와 다양성을 크게 늘릴 수 있습니다.
개요
간단히 말해서, 데이터 증강은 기존 데이터를 기반으로 새롭고 다양한 학습 예제를 인위적으로 생성하는 기법입니다. 이미지 분야에서는 회전, 자르기, 색상 변경 등을 사용하는데, 텍스트에서는 단어 치환, 문장 재구성, 역번역(back-translation), LLM 기반 생성 등을 사용합니다.
예를 들어, "API 응답이 느립니다"라는 질문을 "API 속도가 느린데 어떻게 하나요?", "서버 응답 시간이 오래 걸려요" 등으로 변형할 수 있습니다. 기존에는 수작업으로 다양한 표현을 일일이 만들어야 했다면, 이제는 GPT-4나 Claude 같은 강력한 LLM을 활용하여 자동으로 생성할 수 있습니다.
하나의 예제에서 10개, 20개의 변형을 쉽게 만들 수 있습니다. 핵심 원칙은 첫째, 의미는 유지하면서 표현만 변경해야 한다는 점, 둘째, 실제로 발생할 법한 자연스러운 변형이어야 한다는 점, 셋째, 과도한 증강은 오히려 노이즈가 될 수 있다는 점입니다.
적절한 균형이 중요합니다.
코드 예제
# LLM을 활용한 데이터 증강
import anthropic
import json
class DataAugmenter:
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
def augment_example(self, user_message: str, assistant_response: str,
num_variations: int = 3) -> list:
"""하나의 예제로부터 여러 변형 생성"""
prompt = f"""다음 대화 예제의 변형을 {num_variations}개 생성해주세요.
원본 의미와 답변 품질은 유지하되, 표현을 다양하게 바꿔주세요.
원본 질문: {user_message}
원본 답변: {assistant_response}
각 변형은 JSON 배열로 반환해주세요:
[
{{"question": "변형된 질문1", "answer": "변형된 답변1"}},
...
]"""
# Claude API 호출
message = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2000,
messages=[{"role": "user", "content": prompt}]
)
# 응답 파싱
response_text = message.content[0].text
# JSON 부분만 추출 (마크다운 코드 블록 제거)
json_text = response_text.strip()
if json_text.startswith('```'):
json_text = json_text.split('```')[1]
if json_text.startswith('json'):
json_text = json_text[4:]
variations = json.loads(json_text.strip())
return variations
def augment_dataset(self, original_examples: list,
variations_per_example: int = 3) -> list:
"""전체 데이터셋 증강"""
augmented = []
for example in original_examples:
# 원본 추가
augmented.append(example)
# 변형 생성 및 추가
user_msg = example['messages'][0]['content']
asst_msg = example['messages'][1]['content']
variations = self.augment_example(
user_msg, asst_msg, variations_per_example
)
for var in variations:
new_example = {
"messages": [
{"role": "user", "content": var['question']},
{"role": "assistant", "content": var['answer']}
]
}
augmented.append(new_example)
return augmented
# 사용 예시
# augmenter = DataAugmenter(api_key="your-api-key")
# original_data = [{"messages": [...]}]
# augmented_data = augmenter.augment_dataset(original_data, variations_per_example=3)
# print(f"원본 {len(original_data)}개 -> 증강 후 {len(augmented_data)}개")
설명
이것이 하는 일: 이 코드는 Claude API를 활용하여 하나의 대화 예제로부터 여러 개의 자연스러운 변형을 자동으로 생성하는 데이터 증강 시스템입니다. 첫 번째로, DataAugmenter 클래스는 Anthropic의 Claude API를 초기화합니다.
Claude 3.5 Sonnet은 자연스러운 텍스트 생성에 뛰어나기 때문에 데이터 증강에 매우 적합합니다. GPT-4를 사용해도 되지만, Claude가 프롬프트 준수와 JSON 생성에서 더 안정적인 경우가 많습니다.
두 번째로, augment_example 메서드는 원본 질문과 답변을 받아서 Claude에게 변형을 요청합니다. 프롬프트에서 핵심은 "의미와 품질은 유지하되 표현만 바꾸라"는 지시입니다.
이렇게 하면 "API가 느려요"를 "API 응답 속도가 느린데 해결 방법이 있나요?", "서버 API 성능이 안 좋아요" 등 다양한 표현으로 변환합니다. JSON 형식으로 응답을 받기 때문에 파싱이 쉽고 안정적입니다.
세 번째로, 응답을 파싱하는 과정에서 마크다운 코드 블록을 제거합니다. Claude가 가끔 ```json으로 감싸서 응답하기 때문에 이를 처리해야 합니다.
실무에서는 여기에 에러 처리를 추가하여 파싱 실패 시 재시도하거나 로그를 남기는 것이 좋습니다. 네 번째로, augment_dataset 메서드는 전체 데이터셋을 순회하면서 각 예제를 증강합니다.
원본 데이터는 그대로 유지하고 변형들을 추가하는 방식입니다. 예를 들어 원본 50개에 각각 3개씩 변형을 만들면 총 200개(50 원본 + 150 변형)의 데이터셋이 만들어집니다.
여러분이 이 코드를 사용하면 적은 양의 고품질 데이터로부터 대규모 데이터셋을 만들 수 있습니다. 실무에서는 생성된 변형을 전문가가 샘플링하여 검토하는 것이 좋습니다.
때로는 미묘하게 의미가 바뀌거나 부자연스러운 표현이 생성될 수 있기 때문입니다. 하지만 전체를 수작업으로 만드는 것보다 훨씬 효율적이고, 품질도 대부분 우수합니다.
실전 팁
💡 변형 개수는 원본 데이터 품질에 따라 조절하세요. 고품질 데이터는 5-10개, 불확실한 데이터는 1-2개 정도가 적당합니다.
💡 생성된 변형 중 10-20% 정도는 샘플링하여 수동 검토하세요. 품질 문제를 조기에 발견하고 프롬프트를 개선할 수 있습니다.
💡 역번역(영어로 번역 후 다시 한국어로 번역)도 효과적인 증강 기법입니다. 표현이 자연스럽게 바뀌면서 의미는 유지됩니다.
💡 도메인 특화 용어는 변경되지 않도록 프롬프트에 명시하세요. 예: "API, REST, JSON 같은 기술 용어는 그대로 유지"
💡 API 비용을 고려하여 배치 처리하세요. 한 번에 여러 예제를 묶어서 요청하면 토큰 효율이 높아집니다.
4. 프롬프트-응답 쌍 최적화 - 모델이 학습하기 좋은 형태 만들기
시작하며
여러분이 파인튜닝을 했는데 모델이 너무 장황하게 답하거나 질문에 제대로 답하지 않는 상황을 겪어본 적 있나요? 예를 들어, 간단한 Yes/No 질문에 3단락으로 설명하거나, 핵심 정보는 빼먹고 불필요한 내용만 길게 쓰는 경우가 있습니다.
이런 문제는 학습 데이터의 프롬프트와 응답이 일관된 패턴을 따르지 않거나, 모델이 학습하기 어려운 형태로 구성되어 있기 때문에 발생합니다. 답변의 길이, 스타일, 구조가 들쭉날쭉하면 모델이 어떤 패턴을 학습해야 할지 혼란스러워합니다.
바로 이럴 때 필요한 것이 프롬프트-응답 쌍 최적화입니다. 명확한 가이드라인을 세우고 일관된 형태로 데이터를 정제하면 모델이 훨씬 안정적이고 예측 가능하게 학습합니다.
개요
간단히 말해서, 프롬프트-응답 쌍 최적화는 학습 데이터의 입력과 출력을 모델이 효과적으로 학습할 수 있도록 일관되고 명확한 패턴으로 구조화하는 작업입니다. 좋은 프롬프트는 명확한 지시와 충분한 컨텍스트를 포함하며, 좋은 응답은 프롬프트의 요구에 정확히 부합하고 일관된 스타일을 유지합니다.
예를 들어, 코드 설명을 요청하는 프롬프트라면 모든 응답이 "1. 전체 동작 설명 2.
단계별 분석 3. 활용 팁" 같은 일관된 구조를 따르는 것이 좋습니다.
기존에는 자연스럽게 발생한 대화를 그대로 사용했다면, 이제는 학습 목적에 맞게 재구성하고 정규화해야 합니다. 실제 대화는 오타, 불완전한 문장, 문맥 의존성 등이 많아서 그대로 사용하면 모델 성능이 저하됩니다.
핵심 원칙은 첫째, 프롬프트는 구체적이고 명확해야 함, 둘째, 응답은 프롬프트에 직접 대응해야 함, 셋째, 길이와 스타일의 일관성, 넷째, 불필요한 정보 제거입니다. 이 네 가지를 지키면 학습 효율이 크게 향상됩니다.
코드 예제
# 프롬프트-응답 쌍 최적화 및 정규화
from typing import Dict, List
import re
class PromptResponseOptimizer:
def __init__(self, target_style="professional", max_response_length=500):
self.target_style = target_style
self.max_response_length = max_response_length
def normalize_prompt(self, prompt: str) -> str:
"""프롬프트를 명확하고 구체적으로 정규화"""
# 공백 정규화
prompt = re.sub(r'\s+', ' ', prompt).strip()
# 불완전한 문장 보완 (질문 형태로)
if not prompt.endswith(('?', '.', '!', '요', '다', '까')):
prompt += '?'
# 너무 짧은 프롬프트는 컨텍스트 추가
if len(prompt) < 10:
prompt = f"다음에 대해 설명해주세요: {prompt}"
return prompt
def optimize_response(self, response: str, prompt: str) -> str:
"""응답을 최적화 (일관성, 간결성)"""
# 공백 정규화
response = re.sub(r'\s+', ' ', response).strip()
# 중복 표현 제거
response = re.sub(r'(\S+)\s+\1', r'\1', response)
# 과도하게 긴 응답 자르기 (의미 단위로)
if len(response) > self.max_response_length:
# 마지막 문장 기준으로 자르기
sentences = response.split('. ')
truncated = []
current_length = 0
for sent in sentences:
if current_length + len(sent) < self.max_response_length:
truncated.append(sent)
current_length += len(sent)
else:
break
response = '. '.join(truncated)
if not response.endswith('.'):
response += '.'
# 스타일 통일 (전문적 vs 친근함)
if self.target_style == "professional":
# 이모티콘 제거
response = re.sub(r'[😀-🙏💀-🙏]', '', response)
# 반말을 존댓말로 (간단한 예시)
response = response.replace('해', '합니다').replace('야', '입니다')
return response
def create_structured_response(self, content: str,
include_summary=True) -> str:
"""구조화된 응답 형식 생성"""
parts = []
if include_summary:
# 요약을 맨 앞에 배치
summary = content.split('.')[0] + '.'
parts.append(f"요약: {summary}\n")
parts.append(f"상세 설명:\n{content}")
return '\n'.join(parts)
def optimize_pair(self, example: Dict) -> Dict:
"""전체 예제 쌍 최적화"""
messages = example['messages']
optimized_messages = []
for msg in messages:
role = msg['role']
content = msg['content']
if role == 'user':
# 프롬프트 최적화
content = self.normalize_prompt(content)
elif role == 'assistant':
# 응답 최적화 (해당 사용자 메시지 참조)
user_prompt = messages[messages.index(msg) - 1]['content']
content = self.optimize_response(content, user_prompt)
optimized_messages.append({'role': role, 'content': content})
return {'messages': optimized_messages}
# 사용 예시
optimizer = PromptResponseOptimizer(
target_style="professional",
max_response_length=300
)
raw_example = {
"messages": [
{"role": "user", "content": "API 느림"},
{"role": "assistant", "content": "아 그거 그거 데이터베이스 쿼리 확인해봐 확인해봐. 인덱스 없으면 느려 느려😅 캐싱도 해봐"}
]
}
optimized = optimizer.optimize_pair(raw_example)
print("최적화 후:")
print(f"질문: {optimized['messages'][0]['content']}")
print(f"답변: {optimized['messages'][1]['content']}")
설명
이것이 하는 일: 이 코드는 원시 대화 데이터를 모델이 효과적으로 학습할 수 있는 형태로 자동 정제하고 최적화합니다. 첫 번째로, normalize_prompt 메서드는 사용자 질문을 명확하게 만듭니다.
실제 사용자는 "API 느림" 같은 불완전한 문장을 자주 입력하는데, 이를 "API 느림?"이나 "다음에 대해 설명해주세요: API 느림"처럼 완전한 형태로 보완합니다. 공백도 정규화하여 중복 공백을 제거하고, 일관된 구조를 만듭니다.
이렇게 하면 모델이 다양한 형태의 질문에 대응하는 법을 더 잘 학습합니다. 두 번째로, optimize_response 메서드는 답변을 정제합니다.
실제 상담사나 사용자가 작성한 답변은 "확인해봐 확인해봐" 같은 중복이나 이모티콘이 포함되어 있을 수 있습니다. 중복 표현을 제거하고, 전문적인 톤으로 통일하며, 과도하게 긴 답변은 의미 단위(문장)로 자릅니다.
예를 들어 500자를 넘으면 마지막 완전한 문장까지만 포함하여 자연스럽게 만듭니다. 세 번째로, 스타일 통일이 중요합니다.
"professional" 모드에서는 이모티콘을 제거하고 반말을 존댓말로 바꿉니다. 실제 프로덕션에서는 더 정교한 자연어 처리를 사용할 수 있지만, 간단한 정규표현식과 문자열 치환으로도 상당한 효과를 볼 수 있습니다.
모든 답변이 일관된 톤을 유지하면 모델도 그 스타일을 학습하여 일관되게 응답합니다. 네 번째로, optimize_pair 메서드는 전체 대화 쌍을 최적화합니다.
각 메시지를 역할에 따라 적절한 최적화 함수로 처리하고, 새로운 예제를 생성합니다. 실무에서는 수천, 수만 개의 예제를 배치로 처리하여 전체 데이터셋을 일괄 최적화할 수 있습니다.
여러분이 이 코드를 사용하면 지저분하고 일관성 없는 원시 데이터를 깨끗하고 학습하기 좋은 형태로 변환할 수 있습니다. 실무에서는 도메인에 맞게 커스터마이징하세요: 의료 분야라면 전문 용어를 보존하는 규칙을 추가하고, 고객 서비스라면 감정적 공감 표현을 유지하는 등의 조정이 필요합니다.
데이터의 품질과 일관성이 높아지면 모델의 성능과 안정성이 크게 향상됩니다.
실전 팁
💡 원본 데이터는 항상 백업하세요. 최적화 과정에서 중요한 정보가 손실될 수 있으므로 언제든 롤백할 수 있어야 합니다.
💡 최적화 전후를 샘플로 비교하여 의도한 대로 변환되는지 확인하세요. 자동화된 정제가 오히려 데이터를 망칠 수도 있습니다.
💡 응답 길이는 작업 특성에 맞게 조절하세요. 간단한 FAQ는 100자 이내, 기술 문서 설명은 500자 이상이 적절할 수 있습니다.
💡 정규표현식은 언어와 도메인에 따라 크게 달라집니다. 한국어 특성(종결어미, 존댓말 등)을 고려한 패턴을 사용하세요.
💡 스타일 통일은 너무 획일적으로 하지 마세요. 약간의 변화는 모델의 일반화 능력을 높이는 데 도움이 됩니다.
5. 밸런스 있는 데이터셋 구성 - 편향 없는 학습을 위한 균형 맞추기
시작하며
여러분이 파인튜닝한 모델이 특정 유형의 질문에만 잘 답하고 다른 질문에는 형편없는 답변을 하는 경험을 해본 적 있나요? 예를 들어, 기술 문서 질문은 완벽하게 답하는데 트러블슈팅 질문에는 "잘 모르겠습니다"라고만 답하는 경우가 있습니다.
이런 문제는 데이터셋이 불균형하게 구성되어 있기 때문에 발생합니다. 전체 1000개 데이터 중 800개가 기술 문서 관련이고 트러블슈팅은 50개뿐이라면, 모델은 당연히 기술 문서에만 특화됩니다.
이를 클래스 불균형(class imbalance) 문제라고 합니다. 바로 이럴 때 필요한 것이 밸런스 있는 데이터셋 구성입니다.
각 카테고리, 난이도, 작업 유형이 적절한 비율로 분포되어 있어야 모델이 모든 상황에서 골고루 잘 작동합니다.
개요
간단히 말해서, 밸런스 있는 데이터셋은 다양한 카테고리, 난이도, 작업 유형이 적절한 비율로 고르게 분포된 학습 데이터를 의미합니다. 균형의 기준은 여러 가지입니다: 주제별 균형(각 주제가 비슷한 비율), 난이도별 균형(쉬운어려운 질문 골고루), 작업 유형별 균형(분류, 생성, 요약 등), 길이별 균형(짧은긴 텍스트) 등.
예를 들어, 고객 서비스 챗봇이라면 제품 문의, 결제 문의, 배송 문의, 반품 문의가 각각 20-30% 정도씩 균형있게 포함되어야 합니다. 기존에는 수집된 데이터를 그대로 사용했다면, 이제는 의도적으로 언더샘플링(많은 클래스 줄이기)이나 오버샘플링(적은 클래스 늘리기)을 통해 균형을 맞춥니다.
실제 사용 빈도와 학습 비율이 다를 수 있다는 점을 이해해야 합니다. 핵심 전략은 첫째, 데이터 분포 분석으로 불균형 파악, 둘째, 과소 대표되는 클래스는 증강으로 늘리기, 셋째, 과대 대표되는 클래스는 샘플링으로 줄이기, 넷째, 중요도에 따른 가중치 부여입니다.
완벽한 50:50 균형보다는 실무적으로 합리적인 균형이 목표입니다.
코드 예제
# 데이터셋 밸런싱 및 분포 분석
from collections import Counter, defaultdict
import random
from typing import List, Dict
import math
class DatasetBalancer:
def __init__(self, examples: List[Dict]):
self.examples = examples
self.category_distribution = defaultdict(list)
self._analyze_distribution()
def _analyze_distribution(self):
"""데이터 분포 분석 (카테고리별 분류)"""
for idx, example in enumerate(self.examples):
# 카테고리 추출 (메타데이터에 있다고 가정)
category = example.get('category', 'unknown')
self.category_distribution[category].append(idx)
def get_distribution_stats(self) -> Dict:
"""분포 통계 반환"""
stats = {}
total = len(self.examples)
for category, indices in self.category_distribution.items():
count = len(indices)
percentage = (count / total) * 100
stats[category] = {
'count': count,
'percentage': round(percentage, 2)
}
return stats
def balance_by_undersampling(self, target_count: int = None) -> List[Dict]:
"""언더샘플링: 많은 카테고리를 줄여서 균형 맞추기"""
if target_count is None:
# 가장 적은 카테고리의 개수를 기준으로
target_count = min(len(indices)
for indices in self.category_distribution.values())
balanced = []
for category, indices in self.category_distribution.items():
# 무작위 샘플링
sampled_indices = random.sample(
indices,
min(target_count, len(indices))
)
for idx in sampled_indices:
balanced.append(self.examples[idx])
return balanced
def balance_by_oversampling(self, target_count: int = None) -> List[Dict]:
"""오버샘플링: 적은 카테고리를 복제하여 균형 맞추기"""
if target_count is None:
# 가장 많은 카테고리의 개수를 기준으로
target_count = max(len(indices)
for indices in self.category_distribution.values())
balanced = []
for category, indices in self.category_distribution.items():
current_count = len(indices)
# 원본 데이터 모두 추가
for idx in indices:
balanced.append(self.examples[idx])
# 부족한 만큼 복제 (무작위 선택)
shortage = target_count - current_count
if shortage > 0:
# 복제할 인덱스 선택 (중복 허용)
duplicated_indices = random.choices(indices, k=shortage)
for idx in duplicated_indices:
balanced.append(self.examples[idx])
return balanced
def balance_by_hybrid(self, target_ratio: float = 0.7) -> List[Dict]:
"""하이브리드: 평균에 가깝게 조정"""
avg_count = int(sum(len(indices)
for indices in self.category_distribution.values())
/ len(self.category_distribution))
target_count = int(avg_count * target_ratio)
balanced = []
for category, indices in self.category_distribution.items():
current_count = len(indices)
if current_count > target_count:
# 언더샘플링
sampled = random.sample(indices, target_count)
else:
# 오버샘플링
sampled = indices + random.choices(
indices,
k=target_count - current_count
)
for idx in sampled:
balanced.append(self.examples[idx])
return balanced
# 사용 예시
examples = [
{"category": "기술문서", "messages": [...]},
{"category": "기술문서", "messages": [...]},
{"category": "기술문서", "messages": [...]},
{"category": "트러블슈팅", "messages": [...]},
{"category": "API사용법", "messages": [...]},
]
balancer = DatasetBalancer(examples)
# 분포 확인
stats = balancer.get_distribution_stats()
print("원본 분포:", stats)
# 균형 맞추기
balanced_data = balancer.balance_by_hybrid(target_ratio=0.8)
balanced_balancer = DatasetBalancer(balanced_data)
print("균형 후 분포:", balanced_balancer.get_distribution_stats())
설명
이것이 하는 일: 이 코드는 데이터셋의 카테고리별 분포를 분석하고, 다양한 샘플링 전략을 통해 균형있는 데이터셋을 생성합니다. 첫 번째로, _analyze_distribution 메서드는 전체 데이터를 카테고리별로 분류합니다.
각 예제의 메타데이터에서 카테고리를 추출하여 딕셔너리에 인덱스를 저장합니다. 이렇게 하면 "기술문서 카테고리에는 인덱스 0, 1, 2의 예제가 있다"는 식으로 빠르게 파악할 수 있습니다.
실무에서는 카테고리 외에도 난이도, 작업 유형, 언어 등 다양한 기준으로 분석할 수 있습니다. 두 번째로, get_distribution_stats 메서드는 분포를 시각화합니다.
각 카테고리가 전체에서 차지하는 비율을 계산하여 불균형을 쉽게 파악할 수 있습니다. 예를 들어 "기술문서 60%, 트러블슈팅 10%, API사용법 30%"라면 트러블슈팅이 과소 대표되고 있다는 것을 알 수 있습니다.
이를 바탕으로 어떤 밸런싱 전략을 사용할지 결정합니다. 세 번째로, 세 가지 밸런싱 전략을 제공합니다.
balance_by_undersampling은 많은 카테고리의 데이터를 줄여서 가장 적은 카테고리 수준으로 맞춥니다. 데이터가 버려지지만 과적합 위험이 줄어듭니다.
balance_by_oversampling은 적은 카테고리의 데이터를 복제하여 가장 많은 카테고리 수준으로 늘립니다. 데이터가 늘어나지만 복제로 인한 과적합 위험이 있습니다.
balance_by_hybrid는 중간 지점을 찾아 일부는 줄이고 일부는 늘려서 균형을 맞춥니다. 네 번째로, 무작위 샘플링을 사용하여 편향을 최소화합니다.
random.sample은 중복 없이 선택하고, random.choices는 중복을 허용하여 선택합니다. 오버샘플링 시 단순 복제보다는 앞서 배운 데이터 증강 기법과 결합하면 더 좋습니다.
예를 들어 부족한 카테고리는 복제 대신 LLM으로 변형을 생성하는 방식입니다. 여러분이 이 코드를 사용하면 불균형한 데이터셋을 분석하고 자동으로 균형을 맞출 수 있습니다.
실무에서는 완벽한 균형보다는 실용적 균형을 추구하세요. 예를 들어 실제 사용 빈도가 "제품 문의 70%, 반품 문의 10%"라면 학습 데이터도 "60% vs 20%" 정도로 조정하여 반품 문의 성능을 높이되 제품 문의도 충분히 학습하게 만드는 것이 현명합니다.
실전 팁
💡 완벽한 50:50 균형이 항상 좋은 것은 아닙니다. 실제 사용 빈도와 중요도를 고려하여 60:40, 70:30 같은 비율도 합리적입니다.
💡 오버샘플링 시 데이터 증강을 함께 사용하세요. 단순 복제는 과적합을 유발하므로 변형을 생성하는 것이 좋습니다.
💡 테스트 데이터셋은 밸런싱하지 마세요. 실제 분포를 반영해야 현실적인 성능 평가가 가능합니다.
💡 희귀하지만 중요한 케이스(예: 긴급 장애 처리)는 비율을 높이는 것이 좋습니다. 빈도는 낮지만 성능이 중요하기 때문입니다.
💡 밸런싱 전후의 모델 성능을 비교하세요. 때로는 불균형한 데이터가 실제 사용 패턴을 더 잘 반영할 수도 있습니다.
6. 검증 데이터셋 분리 - 과적합 방지와 성능 측정
시작하며
여러분이 파인튜닝한 모델이 학습 데이터에서는 완벽한 성능을 보이는데 실제 서비스에 배포하니 형편없는 결과를 내는 경험을 해본 적 있나요? 예를 들어, 학습 데이터의 질문에는 100% 정확하게 답하지만 조금만 다른 표현으로 질문하면 엉뚱한 답변을 하는 경우가 있습니다.
이런 문제는 과적합(overfitting) 때문에 발생합니다. 모델이 패턴을 학습한 것이 아니라 학습 데이터를 그대로 암기해버린 것입니다.
시험 문제와 답을 외워서 같은 시험에서는 만점을 받지만 조금만 변형된 문제에는 답하지 못하는 것과 같습니다. 바로 이럴 때 필요한 것이 검증 데이터셋 분리입니다.
학습에 사용하지 않은 별도의 데이터로 모델을 평가하여 진짜 성능을 측정하고, 과적합을 조기에 발견할 수 있습니다.
개요
간단히 말해서, 검증 데이터셋은 학습에는 사용하지 않고 오직 모델 성능 평가와 하이퍼파라미터 튜닝에만 사용하는 별도의 데이터 집합입니다. 일반적으로 전체 데이터를 학습(Train) 70-80%, 검증(Validation) 10-20%, 테스트(Test) 10-20%로 나눕니다.
학습 데이터로 모델을 학습시키고, 검증 데이터로 학습 중간중간 성능을 체크하며, 테스트 데이터로 최종 성능을 평가합니다. 예를 들어, 1000개 데이터가 있다면 700개 학습, 150개 검증, 150개 테스트로 나눕니다.
기존에는 모든 데이터로 학습하고 같은 데이터로 평가했다면, 이제는 반드시 분리해야 합니다. 같은 데이터로 학습하고 평가하면 성능이 좋아 보이지만 실제로는 아무것도 배우지 못한 것일 수 있습니다.
핵심 원칙은 첫째, 검증/테스트 데이터는 학습에 절대 사용 금지, 둘째, 무작위 분할로 편향 방지, 셋째, 층화 샘플링으로 각 분할이 전체 분포 반영, 넷째, 시계열 데이터는 시간 순서 고려입니다. 올바른 분리가 신뢰할 수 있는 평가의 기본입니다.
코드 예제
# 데이터셋을 학습/검증/테스트로 분리
import random
from typing import List, Dict, Tuple
from collections import defaultdict
class DatasetSplitter:
def __init__(self, examples: List[Dict],
train_ratio=0.7, val_ratio=0.15, test_ratio=0.15,
random_seed=42):
assert abs(train_ratio + val_ratio + test_ratio - 1.0) < 1e-6, \
"비율의 합이 1이어야 합니다"
self.examples = examples
self.train_ratio = train_ratio
self.val_ratio = val_ratio
self.test_ratio = test_ratio
# 재현 가능성을 위한 시드 설정
random.seed(random_seed)
def random_split(self) -> Tuple[List[Dict], List[Dict], List[Dict]]:
"""무작위 분할"""
# 데이터 섞기
shuffled = self.examples.copy()
random.shuffle(shuffled)
total = len(shuffled)
train_end = int(total * self.train_ratio)
val_end = train_end + int(total * self.val_ratio)
train_set = shuffled[:train_end]
val_set = shuffled[train_end:val_end]
test_set = shuffled[val_end:]
return train_set, val_set, test_set
def stratified_split(self, category_key='category') -> Tuple:
"""층화 샘플링 분할 (카테고리 비율 유지)"""
# 카테고리별로 그룹화
category_groups = defaultdict(list)
for example in self.examples:
category = example.get(category_key, 'unknown')
category_groups[category].append(example)
train_set, val_set, test_set = [], [], []
# 각 카테고리별로 비율에 맞게 분할
for category, items in category_groups.items():
random.shuffle(items)
total = len(items)
train_end = int(total * self.train_ratio)
val_end = train_end + int(total * self.val_ratio)
train_set.extend(items[:train_end])
val_set.extend(items[train_end:val_end])
test_set.extend(items[val_end:])
# 최종적으로 섞기
random.shuffle(train_set)
random.shuffle(val_set)
random.shuffle(test_set)
return train_set, val_set, test_set
def save_splits(self, train_set, val_set, test_set,
output_dir='./data'):
"""분할된 데이터를 파일로 저장"""
import json
import os
os.makedirs(output_dir, exist_ok=True)
datasets = {
'train.jsonl': train_set,
'validation.jsonl': val_set,
'test.jsonl': test_set
}
for filename, dataset in datasets.items():
filepath = os.path.join(output_dir, filename)
with open(filepath, 'w', encoding='utf-8') as f:
for example in dataset:
f.write(json.dumps(example, ensure_ascii=False) + '\n')
print(f"{filename}: {len(dataset)}개 저장")
# 사용 예시
examples = [
{"category": "기술문서", "messages": [...]},
{"category": "트러블슈팅", "messages": [...]},
# ... 더 많은 예제
] * 100 # 예시로 100배 복제
splitter = DatasetSplitter(
examples,
train_ratio=0.7,
val_ratio=0.15,
test_ratio=0.15,
random_seed=42 # 재현 가능성
)
# 층화 샘플링으로 분할 (카테고리 비율 유지)
train, val, test = splitter.stratified_split(category_key='category')
print(f"학습: {len(train)}개 ({len(train)/len(examples)*100:.1f}%)")
print(f"검증: {len(val)}개 ({len(val)/len(examples)*100:.1f}%)")
print(f"테스트: {len(test)}개 ({len(test)/len(examples)*100:.1f}%)")
# 파일로 저장
splitter.save_splits(train, val, test, output_dir='./split_data')
설명
이것이 하는 일: 이 코드는 전체 데이터셋을 학습, 검증, 테스트 세트로 과학적으로 분할하고, 각 분할이 원본 데이터의 특성을 잘 반영하도록 만듭니다. 첫 번째로, DatasetSplitter 클래스는 분할 비율을 설정합니다.
일반적으로 70:15:15 또는 80:10:10을 사용하지만, 데이터 양에 따라 조절할 수 있습니다. 데이터가 많으면(10만 개 이상) 95:2.5:2.5처럼 학습 데이터 비율을 높여도 되고, 데이터가 적으면(1000개 이하) 60:20:20처럼 검증/테스트 비율을 높이는 것이 좋습니다.
random_seed를 설정하여 같은 분할을 재현할 수 있게 합니다. 두 번째로, random_split 메서드는 가장 간단한 무작위 분할을 수행합니다.
데이터를 섞은 후 비율에 맞게 잘라냅니다. 이 방법은 간단하지만 카테고리 분포가 불균형한 경우 문제가 될 수 있습니다.
예를 들어 희귀 카테고리가 우연히 모두 학습 세트에 들어가면 검증/테스트에서 평가할 수 없습니다. 세 번째로, stratified_split 메서드는 층화 샘플링을 사용합니다.
각 카테고리별로 같은 비율로 분할하여 모든 세트가 전체 데이터의 분포를 반영하도록 만듭니다. 예를 들어 전체 데이터에서 "기술문서 60%, 트러블슈팅 40%"라면, 학습/검증/테스트 모두에서 이 비율이 유지됩니다.
이렇게 하면 검증 성능이 실제 성능을 더 정확하게 예측합니다. 네 번째로, save_splits 메서드는 분할된 데이터를 JSONL 파일로 저장합니다.
OpenAI나 다른 플랫폼에 업로드할 때 이 파일들을 사용합니다. 각 파일에 몇 개의 예제가 들어갔는지 출력하여 분할이 올바르게 되었는지 확인할 수 있습니다.
여러분이 이 코드를 사용하면 과학적으로 신뢰할 수 있는 데이터 분할을 쉽게 수행할 수 있습니다. 실무에서 중요한 점은 검증 데이터를 절대 학습에 사용하지 않는 것입니다.
검증 성능을 보고 하이퍼파라미터를 조정하는 과정을 여러 번 반복하다 보면 간접적으로 검증 데이터에 과적합될 수 있으므로, 최종 평가는 반드시 한 번도 본 적 없는 테스트 데이터로 해야 합니다.
실전 팁
💡 random_seed를 고정하여 실험의 재현 가능성을 보장하세요. 같은 코드를 다시 실행해도 같은 분할을 얻을 수 있습니다.
💡 시계열 데이터(날짜 순서가 중요한 데이터)는 무작위 분할하지 마세요. 과거 데이터로 학습하고 미래 데이터로 검증해야 현실적입니다.
💡 검증 데이터로 여러 번 튜닝한 후에는 마지막에 테스트 데이터로 단 한 번만 평가하세요. 테스트 성능을 보고 다시 조정하면 의미가 없습니다.
💡 데이터가 매우 적다면(100개 이하) K-Fold Cross-Validation을 고려하세요. 데이터를 K개 부분으로 나누고 돌아가며 검증하여 더 안정적인 평가를 얻을 수 있습니다.
💡 분할 후 각 세트의 통계를 비교하세요. 평균 길이, 카테고리 분포 등이 비슷하면 올바르게 분할된 것입니다.
7. 메타데이터 활용 - 데이터에 문맥 정보 추가하기
시작하며
여러분이 파인튜닝한 모델이 똑같은 질문에도 상황에 따라 다르게 답해야 하는데 항상 같은 답변만 하는 경험을 해본 적 있나요? 예를 들어, "이 기능 어떻게 써요?"라는 질문에 초급자에게는 상세히, 고급자에게는 간단히 답해야 하는데 구분하지 못하는 경우가 있습니다.
이런 문제는 모델이 질문의 문맥(context)을 이해하지 못하기 때문에 발생합니다. 누가 질문했는지, 어떤 상황인지, 이전에 무슨 대화를 했는지 같은 정보 없이는 적절한 답변을 만들기 어렵습니다.
바로 이럴 때 필요한 것이 메타데이터 활용입니다. 사용자 레벨, 사용 언어, 이전 대화 내역, 시간대 등의 추가 정보를 학습 데이터에 포함시키면 모델이 문맥을 고려한 답변을 생성할 수 있습니다.
개요
간단히 말해서, 메타데이터는 질문과 답변 외에 추가로 제공되는 문맥 정보로, 모델이 더 적절하고 개인화된 응답을 생성하도록 돕습니다. 메타데이터의 종류는 다양합니다: 사용자 속성(레벨, 언어, 선호도), 시간 정보(시간대, 요일, 긴급도), 세션 정보(이전 대화, 현재 작업), 환경 정보(디바이스, 위치, 버전) 등.
예를 들어, 고객 서비스 챗봇이라면 "VIP 고객, 한국어, 세 번째 문의, 결제 실패 직후"같은 메타데이터가 답변 품질을 크게 높일 수 있습니다. 기존에는 질문과 답변만 제공했다면, 이제는 시스템 프롬프트나 특수 토큰으로 메타데이터를 주입합니다.
"당신은 초급 개발자를 돕는 멘토입니다"같은 역할 정의가 시스템 프롬프트의 예입니다. 핵심 전략은 첫째, 관련성 높은 메타데이터만 선택(너무 많으면 혼란), 둘째, 일관된 포맷으로 제공, 셋째, 메타데이터에 따른 답변 변화를 학습 데이터에 명확히 반영, 넷째, 프라이버시 민감 정보는 제외입니다.
적절한 메타데이터는 모델의 지능을 크게 향상시킵니다.
코드 예제
# 메타데이터를 활용한 컨텍스트 인식 데이터셋 생성
from typing import Dict, List, Optional
from datetime import datetime
class ContextualDatasetBuilder:
def __init__(self):
self.user_levels = {
'beginner': '초급 개발자로, 기본 개념부터 친절하게 설명이 필요합니다.',
'intermediate': '중급 개발자로, 핵심 내용 위주로 간결한 설명을 선호합니다.',
'advanced': '고급 개발자로, 깊이 있는 기술적 세부사항과 최적화에 관심이 있습니다.'
}
def create_system_prompt(self, metadata: Dict) -> str:
"""메타데이터 기반 시스템 프롬프트 생성"""
prompts = ["당신은 전문 기술 지원 AI입니다."]
# 사용자 레벨 추가
if 'user_level' in metadata:
level_desc = self.user_levels.get(
metadata['user_level'],
'사용자의 수준에 맞게 답변합니다.'
)
prompts.append(f"사용자는 {level_desc}")
# 언어 선호도
if 'language' in metadata:
prompts.append(f"{metadata['language']}로 답변하세요.")
# 답변 스타일
if 'style' in metadata:
style_map = {
'concise': '간결하고 핵심만 전달하세요.',
'detailed': '상세하고 예제를 포함하여 설명하세요.',
'friendly': '친근하고 격려하는 톤을 사용하세요.'
}
prompts.append(style_map.get(metadata['style'], ''))
# 긴급도
if metadata.get('urgent', False):
prompts.append("긴급한 상황이므로 즉시 해결 가능한 방법을 우선 제시하세요.")
return ' '.join(prompts)
def add_conversation_history(self, messages: List[Dict],
history: List[Dict]) -> List[Dict]:
"""이전 대화 내역을 컨텍스트에 추가"""
if not history:
return messages
# 최근 N개의 대화만 포함 (토큰 제한 고려)
recent_history = history[-3:] # 최근 3개 대화
# 히스토리를 시스템 메시지로 요약
history_summary = "이전 대화 요약: "
for h in recent_history:
history_summary += f"사용자가 '{h['question']}'에 대해 물었고, "
# 첫 번째 메시지로 추가
context_message = {
"role": "system",
"content": history_summary
}
return [context_message] + messages
def create_contextual_example(
self,
user_message: str,
assistant_response: str,
metadata: Optional[Dict] = None,
conversation_history: Optional[List[Dict]] = None
) -> Dict:
"""메타데이터와 히스토리를 포함한 완전한 예제 생성"""
metadata = metadata or {}
conversation_history = conversation_history or []
# 기본 메시지 구조
messages = [
{"role": "user", "content": user_message},
{"role": "assistant", "content": assistant_response}
]
# 시스템 프롬프트 생성 및 추가
system_prompt = self.create_system_prompt(metadata)
messages.insert(0, {"role": "system", "content": system_prompt})
# 대화 히스토리 추가
if conversation_history:
messages = self.add_conversation_history(messages, conversation_history)
# 메타데이터도 별도로 저장 (분석용)
example = {
"messages": messages,
"metadata": {
**metadata,
"created_at": datetime.now().isoformat()
}
}
return example
# 사용 예시
builder = ContextualDatasetBuilder()
# 초급자를 위한 예제
beginner_example = builder.create_contextual_example(
user_message="리스트 컴프리헨션이 뭔가요?",
assistant_response="리스트 컴프리헨션은 파이썬에서 리스트를 간결하게 만드는 방법입니다. "
"예를 들어, [x*2 for x in range(5)]는 [0, 2, 4, 6, 8]을 만듭니다. "
"기본 for 문보다 짧고 읽기 쉬워서 파이썬 개발자들이 자주 사용합니다.",
metadata={
"user_level": "beginner",
"language": "한국어",
"style": "detailed"
}
)
# 고급자를 위한 같은 질문
advanced_example = builder.create_contextual_example(
user_message="리스트 컴프리헨션 최적화 팁 있나요?",
assistant_response="제너레이터 표현식 (x*2 for x in data)을 사용하면 메모리 효율적입니다. "
"대용량 데이터는 itertools와 조합하고, 조건문은 앞쪽에 배치하세요.",
metadata={
"user_level": "advanced",
"language": "한국어",
"style": "concise"
},
conversation_history=[
{"question": "파이썬 성능 최적화 방법?", "answer": "프로파일링부터 시작하세요"}
]
)
print("초급자 예제:")
print(beginner_example['messages'][0]['content'][:100] + "...")
print("\n고급자 예제:")
print(advanced_example['messages'][0]['content'][:100] + "...")
설명
이것이 하는 일: 이 코드는 메타데이터와 대화 히스토리를 활용하여 문맥을 인식하는 고급 학습 데이터를 생성합니다. 첫 번째로, create_system_prompt 메서드는 메타데이터를 자연어 지시문으로 변환합니다.
"user_level: beginner"같은 구조화된 데이터를 "사용자는 초급 개발자로, 기본 개념부터 친절하게 설명이 필요합니다"같은 명확한 지시로 만듭니다. 이렇게 하면 모델이 메타데이터의 의미를 더 잘 이해하고, 답변 스타일을 그에 맞게 조정합니다.
각 메타데이터 필드마다 적절한 지시문을 추가하여 종합적인 시스템 프롬프트를 만듭니다. 두 번째로, 사용자 레벨별로 다른 응답 스타일을 정의합니다.
초급자에게는 "기본 개념부터 친절하게", 중급자에게는 "핵심 내용 위주로 간결하게", 고급자에게는 "깊이 있는 기술적 세부사항" 식으로 구분합니다. 이렇게 하면 같은 질문이라도 사용자 수준에 맞는 답변을 학습할 수 있습니다.
실무에서는 수십 개의 메타데이터 조합으로 다양한 시나리오를 커버할 수 있습니다. 세 번째로, add_conversation_history 메서드는 이전 대화를 컨텍스트로 추가합니다.
챗봇이 연속된 대화를 나눌 때 이전 내용을 기억하고 참조할 수 있게 만듭니다. 예를 들어 "그거 어떻게 써요?"같은 대명사를 사용한 질문도 이전 대화를 보고 이해할 수 있습니다.
최근 3개 대화만 포함하여 토큰 제한을 고려합니다. 실제로는 요약 기법이나 벡터 검색으로 더 긴 히스토리를 효율적으로 처리할 수 있습니다.
네 번째로, create_contextual_example 메서드는 모든 요소를 통합합니다. 기본 메시지에 시스템 프롬프트를 추가하고, 히스토리를 결합하며, 메타데이터를 별도로 저장합니다.
created_at 같은 타임스탬프도 추가하여 나중에 데이터 분석에 활용할 수 있습니다. 결과적으로 풍부한 컨텍스트를 가진 완전한 학습 예제가 만들어집니다.
여러분이 이 코드를 사용하면 단순한 질문-답변 쌍을 넘어서 실제 서비스에 가까운 문맥 인식 데이터셋을 만들 수 있습니다. 실무에서는 A/B 테스트로 메타데이터 활용의 효과를 측정하세요.
메타데이터 있는 모델 vs 없는 모델을 비교하면 개인화가 얼마나 성능을 향상시키는지 알 수 있습니다. 다만 너무 많은 메타데이터는 오히려 혼란을 줄 수 있으므로, 정말 중요한 것만 선택적으로 사용하세요.
실전 팁
💡 메타데이터는 3-5개 이내로 제한하세요. 너무 많으면 시스템 프롬프트가 길어져 토큰 비용이 증가하고 모델이 혼란스러워집니다.
💡 메타데이터 값은 미리 정의된 카테고리를 사용하세요(예: beginner/intermediate/advanced). 자유 형식 텍스트는 일관성이 떨어집니다.
💡 대화 히스토리는 요약하여 제공하세요. 전체 내용을 다 넣으면 토큰이 낭비되고, 핵심만 추출하면 효율적입니다.
💡 메타데이터 효과를 측정하세요. 각 메타데이터 필드가 실제로 성능 향상에 기여하는지 확인하고, 효과 없는 것은 제거하세요.
💡 프라이버시에 주의하세요. 사용자 이름, 이메일, 위치 같은 민감 정보는 메타데이터에 포함하지 마세요. 익명화된 ID나 카테고리만 사용하세요.
8. 품질 검증 파이프라인 - 자동화된 품질 관리 시스템
시작하며
여러분이 수천 개의 학습 데이터를 준비했는데 나중에 보니 오타, 중복, 잘못된 답변이 섞여 있어서 다시 정리해야 하는 경험을 해본 적 있나요? 예를 들어, 학습이 끝난 후에야 100개의 중복 데이터나 개인정보가 포함된 예제를 발견하는 경우가 있습니다.
이런 문제는 품질 검증이 수동으로, 산발적으로 이루어지기 때문에 발생합니다. 사람이 일일이 확인하다 보면 실수가 생기고, 데이터가 많아지면 불가능해집니다.
학습이 끝난 후 문제를 발견하면 시간과 비용이 엄청나게 낭비됩니다. 바로 이럴 때 필요한 것이 자동화된 품질 검증 파이프라인입니다.
데이터가 추가될 때마다 자동으로 여러 검증 단계를 거쳐서 문제를 조기에 발견하고 차단할 수 있습니다.
개요
간단히 말해서, 품질 검증 파이프라인은 데이터가 최종 학습 데이터셋에 포함되기 전에 여러 단계의 자동 검증을 수행하는 시스템입니다. 일반적인 검증 단계는: 1) 포맷 검증(JSON 구조 확인), 2) 필수 필드 검증(role, content 존재 확인), 3) 데이터 품질 검증(길이, 금지어, 개인정보), 4) 중복 제거, 5) 의미적 검증(답변이 질문과 관련 있는지), 6) 최종 승인입니다.
예를 들어, CI/CD 파이프라인처럼 각 단계를 통과해야만 다음 단계로 진행됩니다. 기존에는 데이터를 모은 후 한 번에 검증했다면, 이제는 실시간으로 각 데이터가 추가될 때마다 검증합니다.
문제가 있는 데이터는 즉시 거부되고, 통과한 데이터만 누적됩니다. 핵심 구성 요소는 첫째, 다단계 검증 체인(각 단계가 독립적), 둘째, 자동 리포팅(어떤 데이터가 왜 실패했는지), 셋째, 수동 검토 큐(자동 판단 어려운 경계 케이스), 넷째, 품질 메트릭 대시보드(통과율, 실패 유형 통계)입니다.
자동화와 사람의 검토를 적절히 조합하는 것이 핵심입니다.
코드 예제
# 자동화된 품질 검증 파이프라인
from typing import List, Dict, Tuple, Callable
from dataclasses import dataclass
import hashlib
import json
@dataclass
class ValidationResult:
"""검증 결과"""
passed: bool
stage: str
message: str
severity: str = "error" # error, warning, info
class QualityPipeline:
def __init__(self):
self.validators = [] # 검증 함수들
self.seen_hashes = set() # 중복 체크용
self.validation_report = [] # 검증 리포트
def add_validator(self, name: str, validator_func: Callable):
"""검증 단계 추가"""
self.validators.append((name, validator_func))
def _hash_example(self, example: Dict) -> str:
"""예제의 해시값 생성 (중복 체크용)"""
# 메시지 내용만으로 해시 생성
content = json.dumps(example['messages'], sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()
def validate_format(self, example: Dict) -> ValidationResult:
"""포맷 검증"""
if 'messages' not in example:
return ValidationResult(
False, "format",
"messages 필드가 없습니다", "error"
)
if not isinstance(example['messages'], list):
return ValidationResult(
False, "format",
"messages는 리스트여야 합니다", "error"
)
for msg in example['messages']:
if 'role' not in msg or 'content' not in msg:
return ValidationResult(
False, "format",
"role 또는 content 필드가 없습니다", "error"
)
return ValidationResult(True, "format", "포맷 검증 통과", "info")
def validate_quality(self, example: Dict) -> ValidationResult:
"""품질 검증 (길이, 내용)"""
for msg in example['messages']:
content = msg['content']
# 너무 짧거나 긴 경우
if len(content) < 5:
return ValidationResult(
False, "quality",
f"내용이 너무 짧습니다: {len(content)}자", "error"
)
if len(content) > 5000:
return ValidationResult(
False, "quality",
f"내용이 너무 깁니다: {len(content)}자", "warning"
)
# 금지어 체크 (간단한 예시)
forbidden = ['비속어', '욕설']
if any(word in content for word in forbidden):
return ValidationResult(
False, "quality",
"금지어가 포함되어 있습니다", "error"
)
return ValidationResult(True, "quality", "품질 검증 통과", "info")
def validate_duplication(self, example: Dict) -> ValidationResult:
"""중복 검증"""
example_hash = self._hash_example(example)
if example_hash in self.seen_hashes:
return ValidationResult(
False, "duplication",
"중복된 데이터입니다", "warning"
)
self.seen_hashes.add(example_hash)
return ValidationResult(True, "duplication", "중복 검증 통과", "info")
def validate_example(self, example: Dict) -> Tuple[bool, List[ValidationResult]]:
"""전체 파이프라인 실행"""
results = []
# 기본 검증 단계들
default_validators = [
("format", self.validate_format),
("quality", self.validate_quality),
("duplication", self.validate_duplication)
]
# 모든 검증 실행
all_validators = default_validators + self.validators
for name, validator in all_validators:
result = validator(example)
results.append(result)
# error 발생 시 즉시 중단
if not result.passed and result.severity == "error":
return False, results
# 모든 검증 통과
return True, results
def process_batch(self, examples: List[Dict]) -> Dict:
"""배치 처리 및 리포팅"""
passed_examples = []
failed_examples = []
warnings = []
for idx, example in enumerate(examples):
is_valid, results = self.validate_example(example)
if is_valid:
passed_examples.append(example)
else:
failed_examples.append({
'index': idx,
'example': example,
'results': results
})
# 경고 수집
for result in results:
if result.severity == "warning":
warnings.append({
'index': idx,
'message': result.message
})
# 리포트 생성
report = {
'total': len(examples),
'passed': len(passed_examples),
'failed': len(failed_examples),
'warnings': len(warnings),
'pass_rate': len(passed_examples) / len(examples) * 100,
'failed_details': failed_examples[:10], # 처음 10개만
'warning_details': warnings[:10]
}
return {
'passed': passed_examples,
'report': report
}
# 사용 예시
pipeline = QualityPipeline()
# 커스텀 검증 추가
def validate_code_examples(example: Dict) -> ValidationResult:
"""코드 예제가 있는지 확인"""
for msg in example['messages']:
if msg['role'] == 'assistant' and '```' in msg['content']:
return ValidationResult(True, "code", "코드 예제 포함", "info")
return ValidationResult(True, "code", "코드 예제 없음", "info")
pipeline.add_validator("code_check", validate_code_examples)
# 테스트 데이터
test_data = [
{"messages": [{"role": "user", "content": "안녕하세요"},
{"role": "assistant", "content": "안녕하세요!"}]},
{"messages": [{"role": "user", "content": "짧음"},
{"role": "assistant", "content": "ㅇㅋ"}]}, # 품질 실패
{"messages": [{"role": "user", "content": "안녕하세요"},
{"role": "assistant", "content": "안녕하세요!"}]}, # 중복
]
result = pipeline.process_batch(test_data)
print(f"통과: {result['report']['passed']}/{result['report']['total']}")
print(f"통과율: {result['report']['pass_rate']:.1f}%")
print(f"실패 상세:\n{json.dumps(result['report']['failed_details'], indent=2, ensure_ascii=False)[:500]}")
설명
이것이 하는 일: 이 코드는 데이터가 학습에 사용되기 전에 여러 단계의 자동 검증을 수행하고, 상세한 리포트를 생성하는 품질 관리 시스템입니다. 첫 번째로, ValidationResult 데이터클래스는 각 검증 단계의 결과를 구조화합니다.
통과 여부뿐만 아니라 어느 단계에서 실패했는지, 무엇이 문제인지, 심각도는 어느 정도인지를 명확히 기록합니다. severity를 "error", "warning", "info"로 구분하여 치명적인 문제와 경미한 문제를 구별합니다.
error는 즉시 거부, warning은 통과하되 로그 남김, info는 단순 정보 기록입니다. 두 번째로, 세 가지 기본 검증을 수행합니다.
validate_format은 JSON 구조가 올바른지, 필수 필드가 있는지 확인합니다. 잘못된 포맷은 학습 자체를 실패시키므로 가장 먼저 체크합니다.
validate_quality는 내용의 품질을 확인합니다. 너무 짧은 답변("ㅇㅋ", "ㄱㅅ")이나 너무 긴 답변, 금지어 포함 등을 걸러냅니다.
validate_duplication은 MD5 해시로 중복을 탐지합니다. 같은 데이터가 여러 번 학습되면 과적합이 발생하므로 중복 제거가 중요합니다.
세 번째로, validate_example 메서드는 모든 검증을 순차적으로 실행합니다. error가 발생하면 즉시 중단하여 불필요한 연산을 방지합니다.
예를 들어 포맷이 잘못되면 품질 검증을 할 필요 없이 바로 거부합니다. warning은 기록하되 다음 단계로 진행합니다.
모든 검증 결과를 리스트로 반환하여 나중에 분석할 수 있습니다. 네 번째로, process_batch 메서드는 전체 데이터셋을 처리하고 리포트를 생성합니다.
통과한 데이터, 실패한 데이터, 경고를 각각 분류하고, 통과율 같은 통계를 계산합니다. 실패한 데이터의 처음 10개만 리포트에 포함하여 대용량 데이터에서도 리포트가 너무 커지지 않게 합니다.
이 리포트를 보고 어떤 유형의 문제가 많은지 파악하여 데이터 수집 과정을 개선할 수 있습니다. 여러분이 이 코드를 사용하면 수천, 수만 개의 데이터를 자동으로 검증하여 고품질 데이터셋만 선별할 수 있습니다.
실무에서는 더 많은 검증 단계를 추가하세요: LLM으로 답변의 정확성 체크, 벡터 유사도로 의미적 중복 탐지, 도메인 전문가의 샘플 검토 등. CI/CD 파이프라인에 통합하여 새로운 데이터가 추가될 때마다 자동으로 검증하고, 실패하면 커밋을 거부하는 방식도 효과적입니다.
실전 팁
💡 검증 파이프라인을 Git pre-commit hook에 통합하세요. 품질 검증을 통과하지 못한 데이터는 커밋되지 않도록 강제할 수 있습니다.
💡 실패한 데이터는 버리지 말고 별도로 저장하세요. 나중에 검토하여 검증 규칙을 개선하거나 수동으로 수정할 수 있습니다.
💡 통과율을 모니터링하세요. 갑자기 통과율이 낮아지면 데이터 소스나 수집 과정에 문제가 생긴 것일 수 있습니다.
💡 LLM을 활용한 의미적 검증도 추가하세요. "답변이 질문과 관련 있는가?", "사실적으로 정확한가?" 같은 고급 검증이 가능합니다.
💡 검증 단계는 빠른 것부터 실행하세요. 포맷 체크(밀리초) -> 중복 체크(초) -> LLM 검증(수 초) 순서로 하면 효율적입니다.
9. 실전 테스트와 반복 개선 - 데이터셋 품질 검증하기
시작하며
여러분이 완벽하다고 생각한 데이터셋으로 파인튜닝했는데 실제 서비스에서 성능이 기대에 못 미치는 경험을 해본 적 있나요? 예를 들어, 학습 loss는 계속 감소하는데 실제 사용자 만족도는 낮거나, 특정 시나리오에서만 계속 실패하는 경우가 있습니다.
이런 문제는 데이터셋이 실제 사용 사례를 제대로 반영하지 못하거나, 테스트 방법이 부적절하기 때문에 발생합니다. 실험실에서는 완벽해 보여도 현실 세계는 예상치 못한 변수가 많습니다.
데이터와 현실의 간극이 성능 차이로 나타납니다. 바로 이럴 때 필요한 것이 실전 테스트와 반복 개선 프로세스입니다.
소규모로 배포하여 실제 사용 데이터를 수집하고, 실패 사례를 분석하여 데이터셋을 보강하는 순환 과정을 통해 지속적으로 개선할 수 있습니다.
개요
간단히 말해서, 실전 테스트와 반복 개선은 학습한 모델을 실제 환경에서 테스트하고, 발견된 문제를 데이터셋에 반영하여 다시 학습하는 지속적인 개선 사이클입니다. 일반적인 프로세스는: 1) 초기 데이터셋으로 학습, 2) 소규모 베타 테스트(전체 사용자의 5-10%), 3) 실패 케이스와 사용자 피드백 수집, 4) 부족한 부분을 보완한 데이터 추가, 5) 재학습 및 A/B 테스트, 6) 점진적 확대 배포입니다.
예를 들어, 1주일 베타 테스트 -> 실패 케이스 100개 수집 -> 데이터 보강 -> 재학습 -> 2주일 A/B 테스트 -> 전체 배포 같은 식입니다. 기존에는 한 번 학습하고 끝났다면, 이제는 데이터-학습-테스트-개선의 순환을 여러 번 반복합니다.
첫 버전은 70점이어도 괜찮고, 매 반복마다 5-10점씩 개선하는 것이 목표입니다. 핵심 활동은 첫째, 실패 케이스 체계적 수집(로깅, 사용자 리포트), 둘째, 실패 원인 분석(데이터 부족?
품질 문제? 모델 한계?), 셋째, 타겟 데이터 추가(부족한 시나리오 집중 보강), 넷째, 정량적 평가 메트릭(정확도, F1 스코어, 사용자 만족도)입니다.
데이터 중심 개발(Data-Centric AI) 접근법이 핵심입니다.
코드 예제
# 실전 테스트 및 반복 개선 시스템
from typing import List, Dict, Tuple
from collections import defaultdict, Counter
from datetime import datetime
import json
class ProductionMonitor:
def __init__(self):
self.failure_cases = [] # 실패 케이스 수집
self.success_cases = [] # 성공 케이스
self.user_feedback = [] # 사용자 피드백
def log_interaction(self, user_query: str, model_response: str,
is_success: bool, user_rating: int = None,
metadata: Dict = None):
"""실제 사용 상황 로깅"""
interaction = {
'timestamp': datetime.now().isoformat(),
'query': user_query,
'response': model_response,
'success': is_success,
'rating': user_rating,
'metadata': metadata or {}
}
if is_success:
self.success_cases.append(interaction)
else:
self.failure_cases.append(interaction)
if user_rating is not None:
self.user_feedback.append(interaction)
def analyze_failures(self) -> Dict:
"""실패 케이스 분석"""
if not self.failure_cases:
return {"message": "실패 케이스 없음"}
# 실패 유형 분류 (간단한 키워드 기반)
failure_types = defaultdict(list)
for case in self.failure_cases:
query = case['query'].lower()
# 키워드 기반 분류 (실무에서는 더 정교한 분류 필요)
if '오류' in query or 'error' in query:
failure_types['troubleshooting'].append(case)
elif '어떻게' in query or 'how' in query:
failure_types['how_to'].append(case)
elif '무엇' in query or 'what' in query:
failure_types['concept'].append(case)
else:
failure_types['other'].append(case)
# 통계 생성
analysis = {
'total_failures': len(self.failure_cases),
'failure_by_type': {
ftype: len(cases)
for ftype, cases in failure_types.items()
},
'most_common_type': max(
failure_types.items(),
key=lambda x: len(x[1])
)[0] if failure_types else None,
'sample_failures': [
{'query': c['query'], 'response': c['response'][:100]}
for c in self.failure_cases[:5]
]
}
return analysis
def generate_improvement_dataset(self, num_samples: int = 10) -> List[Dict]:
"""실패 케이스 기반 개선 데이터 생성"""
improvement_data = []
# 가장 빈번한 실패 케이스 선택
sorted_failures = sorted(
self.failure_cases,
key=lambda x: x['timestamp'],
reverse=True
)[:num_samples]
for case in sorted_failures:
# 주석: 실제로는 여기서 전문가가 올바른 답변을 작성하거나
# 더 강력한 LLM(GPT-4 등)으로 개선된 답변을 생성
improvement_example = {
"messages": [
{"role": "user", "content": case['query']},
{
"role": "assistant",
"content": "[전문가가 작성한 올바른 답변]"
# 실제로는 여기에 개선된 답변을 넣어야 함
}
],
"metadata": {
"source": "production_failure",
"original_response": case['response'],
"failure_timestamp": case['timestamp']
}
}
improvement_data.append(improvement_example)
return improvement_data
def calculate_metrics(self) -> Dict:
"""성능 메트릭 계산"""
total = len(self.success_cases) + len(self.failure_cases)
if total == 0:
return {"message": "데이터 없음"}
success_rate = len(self.success_cases) / total * 100
# 사용자 평점 분석
ratings = [f['rating'] for f in self.user_feedback if f['rating']]
avg_rating = sum(ratings) / len(ratings) if ratings else 0
metrics = {
'total_interactions': total,
'success_rate': round(success_rate, 2),
'failure_rate': round(100 - success_rate, 2),
'avg_user_rating': round(avg_rating, 2),
'rating_distribution': dict(Counter(ratings))
}
return metrics
class IterativeImprovement:
def __init__(self, initial_dataset: List[Dict]):
self.dataset = initial_dataset
self.version = 1
self.improvement_history = []
def add_improvement_data(self, new_data: List[Dict],
reason: str):
"""개선 데이터 추가 및 버전 관리"""
self.dataset.extend(new_data)
self.version += 1
self.improvement_history.append({
'version': self.version,
'added_count': len(new_data),
'total_count': len(self.dataset),
'reason': reason,
'timestamp': datetime.now().isoformat()
})
def get_current_dataset(self) -> List[Dict]:
"""현재 최신 데이터셋 반환"""
return self.dataset
def get_improvement_report(self) -> Dict:
"""개선 이력 리포트"""
return {
'current_version': self.version,
'total_examples': len(self.dataset),
'improvement_history': self.improvement_history
}
# 사용 예시 시뮬레이션
monitor = ProductionMonitor()
# 실제 사용 로깅 시뮬레이션
monitor.log_interaction(
"파이썬 데코레이터 오류 해결법?",
"죄송합니다, 잘 모르겠습니다.",
is_success=False,
user_rating=1
)
monitor.log_interaction(
"리스트 컴프리헨션 예제?",
"[x*2 for x in range(5)]는 [0,2,4,6,8]을 생성합니다.",
is_success=True,
user_rating=5
)
# 실패 분석
failure_analysis = monitor.analyze_failures()
print("실패 분석:")
print(json.dumps(failure_analysis, indent=2, ensure_ascii=False))
# 성능 메트릭
metrics = monitor.calculate_metrics()
print("\n성능 메트릭:")
print(json.dumps(metrics, indent=2, ensure_ascii=False))
# 개선 데이터 생성
improvement_data = monitor.generate_improvement_dataset(num_samples=5)
# 반복 개선
improver = IterativeImprovement(initial_dataset=[])
improver.add_improvement_data(
improvement_data,
reason="데코레이터 관련 트러블슈팅 실패 케이스 보강"
)
print("\n개선 리포트:")
print(json.dumps(improver.get_improvement_report(), indent=2, ensure_ascii=False))