본문
3. 데이터셋 준비하기
AI&ML/파인튜닝 2025. 12. 16. 08:02
반응형
앞서 살펴본 파인튜닝의 개념을 실제로 적용하기 위해서는 학습을 위한 데이터가 필요합니다.
이제 데이터를 어떻게 준비하고 처리하는지 알아보겠습니다.
- 데이터셋 준비하기
- 첫 번째 방법: 허깅페이스에서 데이터셋 찾기
- 두 번째 방법: 자체 데이터셋 만들기
허깅페이스에서 데이터셋 찾기
- 허깅페이스(Hugging Face)가 뭐야?
- 인공지능 분야에서 사용되는 머신러닝 모델과 데이터셋을 공유하고 협업하는 서비스입니다.
허깅페이스(https://huggingface.co/datasets)에서 금융 관련 키워드로 검색하면 다양한 금융 데이터셋을 찾을 수 있어요.
허깅페이스 데이터셋 샘플
- gbharti/finance-alpaca
- 금융 상담: 투자, 은퇴, 포트폴리오 관련 조언
- 금융 교육: 경제 개념, 용어, 법률 설명
- 실용 조언: 신용카드, 투자 위험, 세금 관련
- 맞춤형 상담: 개인별 상황에 따른 전문 조언
- sujet-ai/Sujet-Finance-Instruct-177k
- 감성 분석: 금융 텍스트의 긍정/부정 분석, 주가 전망
- 금융 특화: 전문 질의응답, 뉴스 분석, 대화형 응답
- 고급 분석: 복잡한 금융 질문 답변, 맥락 기반 상담
자체 데이터셋 만들기
AI를 가르치기 위한 교과서를 만든다고 생각하면 쉬워요!
시작하기 전에 필요한 것들
- 기본 데이터 (AI에게 가르칠 내용)
- 마치 학생에게 줄 교과서처럼!
- GPT-4 API 키
- 도서관 출입증과 같은 것!
- 데이터 검증 도구
- 교과서 내용이 정확한지 검토하는 도구
1. 학습 데이터 모으기
- 첫 번째 방법: 회사 데이터베이스에서 직접 가져오기
- 마치 회사 도서관에서 책을 빌리는 것처럼!
- OO증권의 2024년 1분기 실적 데이터 가져오기 (영업 수익, 순이익, 기타 재무 정보)
# 사내 DB 연결 예시
import pymysql
import pandas as pd
from sqlalchemy import create_engine
# DB 연결 설정
db_connection = create_engine(
'mysql+pymysql://user:password@hostname/database'
)
# 실적 데이터 쿼리
query = """
SELECT quarter, revenue, profit, roe, roa
FROM test_securities_results
WHERE year = 2024 AND quarter = 1
"""
# DB에서 데이터 불러오기
test_data = pd.read_sql(query, db_connection)
# System Prompt에 실시간 데이터 반영
system_prompt = f"""당신은 한국의 금융 전문가입니다.
현재 OO증권의 데이터:
- 2024년 1분기 영업수익: {test_data.iloc[0]['revenue']}
- 2024년 1분기 순이익: {test_data.iloc[0]['profit']}
...
- 두 번째 방법: 문서에서 추출하기
- 마치 신문이나 보고서에서 중요한 내용을 스크랩하는 것처럼!
- 실적 보고서에서 중요 내용 추출
- 시장 분석 보고서에서 인사이트 추출
# 실적보고서와 시장분석 자료 활용
earnings_report = """[실적보고서]..."""
market_analysis = """[시장분석]..."""
system_prompt = f"""다음 OO증권 관련 문서들을 참고해 답변해주세요:
1. 실적 정보:
{earnings_report}
2. 시장 분석:
{market_analysis}
제시된 문서 내용에 기반하여 답변해주세요."""- 세 번째 방법: 파일에서 읽어오기
- # 마치 엑셀 파일에 잘 정리된 성적표를 읽는 것처럼!
- OO증권의 분기별 실적이 정리된 CSV 파일 읽기
- 정리된 데이터를 AI가 이해할 수 있는 형태로 변환
# CSV 파일에서 데이터 읽기
def load_test_financial_data(file_path):
# OO증권 재무데이터 로드
df = pd.read_csv(file_path)
return df
# 데이터 로드
test_financial_data = load_test_financial_data('test_securities_financial_data.csv')
# System Prompt 구성
system_prompt = f"""당신은 한국의 금융 전문가입니다.
다음 OO증권의 분기별 실적 데이터를 참고하여 답변해주세요:
{test_financial_data.to_string()}
제공된 데이터의 범위 내에서만 답변해주세요."""2. AI에게 금융 전문가 역할 부여하기
#마치 신입 직원에게 업무 매뉴얼을 주는 것처럼:
- "당신은 OO증권의 리서치 전문가입니다"
- "이 공시 자료를 참고해서 답변하세요"
- "전문 용어는 쉽게 설명해주세요"
# GPT-4를 금융 전문가로 설정
system_prompt = f"""당신은 OO증권의 리서치 전문가입니다.
다음 공시 데이터를 기반으로 답변해주세요:
{reference_doc}
주의사항:
1. 제공된 실제 데이터만 사용하여 답변
2. 수치는 정확하게 인용
3. 금융 전문 용어는 쉽게 설명3. 질문-답변 세트 만들기
- Q: "OO증권의 2024년 1분기 실적이 어떤가요?"
- A: "2024년 1분기 영업수익은 1조 2,345억원으로..."
qa_pairs = [
{
"질문": "OO증권의 2024년 1분기 영업수익이 얼마인가요?",
"답변": "OO증권의 2024년 1분기 영업수익은 1조 2,345억원..."
},
...
]4. 데이터 저장하기
- CSV 파일로 저장 (엑셀로 열 수 있게)
- JSON 파일로 저장 (컴퓨터가 읽기 쉽게)
- 한글이 깨지지 않도록 관리
# CSV와 JSON 형식으로 저장
def save_dataset(dataset: List[Dict], filename: str):
"""데이터셋을 CSV와 JSON 형식으로 저장"""
try:
# DataFrame 생성
df = pd.DataFrame(dataset)
# CSV 저장 (한글 깨짐 방지)
df.to_csv(
f"{filename}.csv",
index=False,
encoding='utf-8-sig'
)
# JSON 저장 (한글 깨짐 방지 + 들여쓰기)
with open(f"{filename}.json", 'w', encoding='utf-8') as f:
json.dump(
dataset,
f,
ensure_ascii=False,
indent=2
)
print(f"데이터셋 저장 완료:")
print(f"- CSV 파일: {filename}.csv")
print(f"- JSON 파일: {filename}.json")
except Exception as e:
print(f"데이터셋 저장 중 오류 발생: {str(e)}")
raise5. 전체 과정 자동화하기
- 파이프라인 단계
- 1단계: 준비 작업
- 2단계: 질문-답변 세트 만들기
- 3단계: 데이터 검증하기
- 4단계: 데이터 저장하기
- 5단계: 샘플 확인하기
- 예시 코드
import os
import json
from typing import List, Dict
import random
def generate_qa_pair() -> Dict[str, str]:
"""단일 질문-답변 쌍 생성"""
templates = [
{
"instruction": "주식 투자를 처음 시작하려고 합니다. 어떻게 시작해야 할까요?",
"response": "주식 투자를 시작하기 위한 기본 단계를 안내해드리겠습니다:\n1. 증권사 계좌 개설\n2. 기본적인 투자 교육 이수\n3. 소액으로 시작하여 경험 쌓기\n4. 분산 투자로 리스크 관리\n5. 정기적인 투자 성과 모니터링"
},
{
"instruction": "ETF와 주식의 차이점은 무엇인가요?",
"response": "ETF와 주식의 주요 차이점을 설명드리겠습니다:\n1. 분산투자: ETF는 여러 종목에 분산투자, 주식은 단일 기업 투자\n2. 위험도: ETF가 일반적으로 위험도가 낮음\n3. 운용방식: ETF는 인덱스 추종, 주식은 기업 실적에 따라 변동\n4. 거래방식: 둘 다 실시간 거래 가능"
},
{
"instruction": "주식 투자 시 손실 관리는 어떻게 해야 하나요?",
"response": "효과적인 손실 관리 방법을 안내해드립니다:\n1. 투자금 분산\n2. 손절매 기준 설정\n3. 적정 투자 비중 유지\n4. 정기적인 포트폴리오 리밸런싱\n5. 투자 일지 작성으로 실수 분석"
}
]
return random.choice(templates)
def generate_dataset(num_samples: int) -> List[Dict[str, str]]:
"""지정된 수만큼 Q&A 쌍 생성"""
dataset = []
for _ in range(num_samples):
qa_pair = generate_qa_pair()
dataset.append(qa_pair)
return dataset
def validate_response(qa_pair: Dict[str, str]) -> bool:
"""Q&A 쌍 유효성 검증"""
# 필수 키 존재 확인
if not all(key in qa_pair for key in ['instruction', 'response']):
return False
# 내용 길이 검증
if len(qa_pair['instruction']) < 10 or len(qa_pair['response']) < 20:
return False
# 내용 형식 검증
if not qa_pair['instruction'].endswith('?'):
return False
return True
def save_dataset(dataset: List[Dict[str, str]], output_path: str) -> None:
"""데이터셋을 JSON 형식으로 저장"""
# JSON 파일로 저장
with open(f"{output_path}.json", 'w', encoding='utf-8') as f:
json.dump(dataset, f, ensure_ascii=False, indent=2)
# CSV 형식으로도 저장 (backup)
import csv
with open(f"{output_path}.csv", 'w', encoding='utf-8', newline='') as f:
writer = csv.DictWriter(f, fieldnames=['instruction', 'response'])
writer.writeheader()
writer.writerows(dataset)
# 메인 실행 프로그램
def main():
"""데이터셋 생성부터 저장까지 전체 파이프라인"""
try:
print("OO증권 데이터셋 생성 시작...")
# 1. 초기 설정
os.makedirs('output', exist_ok=True) # 출력 폴더 생성
# 2. 데이터셋 생성 (5개의 Q&A 쌍)
dataset = generate_dataset(num_samples=5)
print(f"데이터셋 생성: {len(dataset)}개의 Q&A 쌍")
# 3. 데이터 검증
validated_dataset = [
qa for qa in dataset
if validate_response(qa)
]
print(f"검증 통과: {len(validated_dataset)}개의 Q&A 쌍")
# 4. 데이터셋 저장
output_path = "output/test_securities_qa_dataset"
save_dataset(validated_dataset, output_path)
# 5. 샘플 데이터 확인
print("\n생성된 데이터 샘플:")
for idx, qa in enumerate(validated_dataset[:2], 1):
print(f"\n[샘플 {idx}]")
print(f"Q: {qa['instruction']}")
print(f"A: {qa['response']}")
except Exception as e:
print(f"실행 중 오류 발생: {str(e)}")
raise
else:
print("\n데이터셋 생성 완료!")
if __name__ == "__main__":
main()
반응형
댓글