3장: 텍스트 데이터 합성 실전

import json
import asyncio
from dataclasses import dataclass, asdict
from openai import AsyncOpenAI
 
client = AsyncOpenAI()
 
 
@dataclass
class InstructionPair:
    instruction: str
    response: str
    domain: str
    difficulty: str
    quality_score: float = 0.0
 
 
INSTRUCTION_GEN_PROMPT = """당신은 AI 학습 데이터 생성 전문가입니다.
주어진 주제와 도메인에 대해 고품질 지시문 5개를 생성하세요.
 
주제: {topic}
도메인: {domain}
난이도: {difficulty}
 
규칙:
- 각 지시문은 명확하고 구체적이어야 합니다.
- 단순 정의 질문보다 분석/비교/적용 질문을 선호합니다.
- 지시문만 생성하고, 응답은 생성하지 마세요.
 
JSON 배열 형식으로 출력하세요:
["지시문1", "지시문2", ...]"""
 
 
RESPONSE_GEN_PROMPT = """다음 지시문에 대해 전문적이고 상세한 응답을 작성하세요.
 
지시문: {instruction}
 
규칙:
- 정확한 정보를 기반으로 답변합니다.
- 구조화된 형식(목록, 단계 등)을 적절히 활용합니다.
- 200~500 단어 범위로 작성합니다.
- 한국어로 작성합니다."""
 
 
async def generate_instructions(
    topic: str, domain: str, difficulty: str
) -> list[str]:
    """주어진 주제에 대한 지시문을 생성합니다."""
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": INSTRUCTION_GEN_PROMPT.format(
                    topic=topic, domain=domain, difficulty=difficulty
                ),
            }
        ],
        temperature=0.8,
        response_format={"type": "json_object"},
    )
    content = response.choices[0].message.content
    return json.loads(content) if content else []
 
 
async def generate_response(instruction: str) -> str:
    """지시문에 대한 응답을 생성합니다."""
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": RESPONSE_GEN_PROMPT.format(
                    instruction=instruction
                ),
            }
        ],
        temperature=0.7,
        max_tokens=1024,
    )
    return response.choices[0].message.content or ""
 
 
async def build_instruction_dataset(
    topics: list[dict],
) -> list[InstructionPair]:
    """전체 지시-응답 데이터셋을 구축합니다."""
    all_pairs: list[InstructionPair] = []
 
    for topic_info in topics:
        instructions = await generate_instructions(
            topic=topic_info["topic"],
            domain=topic_info["domain"],
            difficulty=topic_info["difficulty"],
        )
 
        tasks = [generate_response(inst) for inst in instructions]
        responses = await asyncio.gather(*tasks)
 
        for inst, resp in zip(instructions, responses):
            pair = InstructionPair(
                instruction=inst,
                response=resp,
                domain=topic_info["domain"],
                difficulty=topic_info["difficulty"],
            )
            all_pairs.append(pair)
 
    return all_pairs

SCENARIO_TEMPLATE = {
    "context": "사용자의 배경과 상황 설명",
    "user_goal": "사용자가 달성하려는 목표",
    "assistant_role": "어시스턴트의 역할과 전문성",
    "num_turns": "대화 턴 수 (3~10)",
    "complexity": "대화 복잡도 (단순 / 중간 / 복잡)",
    "special_conditions": "특수 조건 (예: 오류 정정, 주제 전환)",
}
 
SCENARIOS = [
    {
        "context": "Python 초보자가 웹 크롤링을 배우려 합니다.",
        "user_goal": "BeautifulSoup으로 뉴스 기사를 크롤링하고 싶습니다.",
        "assistant_role": "시니어 Python 개발자",
        "num_turns": 6,
        "complexity": "중간",
        "special_conditions": "중간에 에러가 발생하여 디버깅이 필요한 상황 포함",
    },
    {
        "context": "데이터 분석가가 ML 모델 선택에 고민하고 있습니다.",
        "user_goal": "고객 이탈 예측에 적합한 모델을 추천받고 싶습니다.",
        "assistant_role": "ML 엔지니어",
        "num_turns": 8,
        "complexity": "복잡",
        "special_conditions": "데이터 불균형 문제를 중간에 발견하는 전개",
    },
]

CONVERSATION_PROMPT = """다음 시나리오에 따라 자연스러운 멀티턴 대화를 생성하세요.
 
시나리오:
- 맥락: {context}
- 사용자 목표: {user_goal}
- 어시스턴트 역할: {assistant_role}
- 대화 턴 수: {num_turns}턴
- 복잡도: {complexity}
- 특수 조건: {special_conditions}
 
규칙:
1. 각 턴은 자연스러운 한국어 대화체로 작성합니다.
2. 사용자는 점진적으로 질문을 심화시킵니다.
3. 어시스턴트는 이전 대화 맥락을 참조하여 응답합니다.
4. 코드 예시가 필요한 경우 실행 가능한 코드를 포함합니다.
5. 대화가 자연스럽게 종결되어야 합니다.
 
JSON 형식으로 출력:
{{
  "conversation": [
    {{"role": "user", "content": "..."}},
    {{"role": "assistant", "content": "..."}}
  ]
}}"""

CLASSIFICATION_PROMPT = """다음 카테고리 각각에 대해 한국어 텍스트 예시를 생성하세요.
 
카테고리: {categories}
각 카테고리당 생성 수: {num_per_category}개
 
규칙:
- 텍스트는 50~200자 범위입니다.
- 각 텍스트는 해당 카테고리에 명확하게 속해야 합니다.
- 경계 사례(두 카테고리에 걸쳐 있는 경우)도 10% 포함합니다.
- 다양한 어조와 문체를 사용합니다.
 
JSON 형식:
[{{"text": "...", "label": "카테고리명", "confidence": 0.0~1.0}}]"""
 
 
def generate_classification_data(
    categories: list[str],
    num_per_category: int = 50,
) -> list[dict]:
    """분류 학습 데이터를 합성 생성합니다."""
    # LLM에 배치로 요청
    # 경계 사례를 별도 프롬프트로 추가 생성
    # 라벨 일관성 검증 후 반환
    pass

NER_GENERATION_PROMPT = """다음 개체 유형을 포함하는 한국어 문장을 생성하세요.
 
개체 유형: {entity_types}
문장 수: {num_sentences}
 
각 문장에 대해 개체의 시작 위치, 끝 위치, 유형을 표시하세요.
 
JSON 형식:
[
  {{
    "text": "삼성전자는 서울 서초구에 본사를 두고 있습니다.",
    "entities": [
      {{"start": 0, "end": 4, "type": "ORG", "text": "삼성전자"}},
      {{"start": 7, "end": 12, "type": "LOC", "text": "서울 서초구"}}
    ]
  }}
]"""
 
 
def validate_ner_annotations(data: list[dict]) -> list[dict]:
    """NER 어노테이션의 정확성을 검증합니다."""
    validated = []
    for item in data:
        text = item["text"]
        valid = True
        for entity in item["entities"]:
            # 텍스트 내 실제 위치와 일치하는지 확인
            extracted = text[entity["start"]:entity["end"]]
            if extracted != entity["text"]:
                valid = False
                break
        if valid:
            validated.append(item)
    return validated

SUMMARIZATION_PROMPT = """다음 주제에 대해 원본 기사와 그 요약문 쌍을 생성하세요.
 
주제: {topic}
원본 길이: {source_length}자 내외
요약 길이: {summary_length}자 내외
요약 유형: {summary_type}  # extractive 또는 abstractive
 
규칙:
- 원본은 뉴스 기사 스타일로 작성합니다.
- 요약은 원본의 핵심 정보를 정확히 포함해야 합니다.
- abstractive 요약은 원본의 문장을 그대로 복사하지 않습니다.
 
JSON 형식:
{{"source": "원본 텍스트", "summary": "요약 텍스트", "type": "요약 유형"}}"""

TRANSLATION_PROMPT = """다음 영어 지시-응답 쌍을 자연스러운 한국어로 번역하세요.
 
규칙:
- 기술 용어는 한국어 업계에서 통용되는 표현을 사용합니다.
- 문화적 맥락이 맞지 않는 부분은 한국 맥락으로 적절히 변환합니다.
- 코드 주석도 한국어로 번역합니다.
- 단순 직역이 아닌, 한국어 화자가 자연스럽게 느끼는 표현을 사용합니다.
 
영어 원본:
Instruction: {en_instruction}
Response: {en_response}
 
한국어 번역:"""
 
 
LOCALIZATION_RULES = {
    "ko": {
        "currency": "USD -> 원",
        "date_format": "MM/DD/YYYY -> YYYY년 MM월 DD일",
        "names": "영어 이름 -> 한국 이름",
        "examples": "미국 기업 -> 한국 기업",
    }
}

import ast
import subprocess
from dataclasses import dataclass
 
 
@dataclass
class CodeSample:
    instruction: str
    code: str
    language: str
    test_cases: list[str]
    is_valid: bool = False
 
 
CODE_GEN_PROMPT = """다음 프로그래밍 과제에 대한 Python 솔루션을 작성하세요.
 
과제: {task_description}
난이도: {difficulty}
 
규칙:
- 실행 가능한 완전한 코드를 작성합니다.
- 타입 힌트를 포함합니다.
- 독스트링으로 함수를 설명합니다.
- 테스트 케이스 3개를 assert 문으로 작성합니다.
 
응답 형식:
```python
# 솔루션 코드

# 테스트 코드
```"""
 
 
def validate_python_syntax(code: str) -> bool:
    """Python 코드의 구문 유효성을 검사합니다."""
    try:
        ast.parse(code)
        return True
    except SyntaxError:
        return False
 
 
def execute_code_safely(code: str, timeout: int = 10) -> tuple[bool, str]:
    """코드를 안전하게 실행하여 테스트를 검증합니다."""
    try:
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        return result.returncode == 0, result.stderr
    except subprocess.TimeoutExpired:
        return False, "Timeout"
 
 
def generate_and_validate_code(
    task: str, difficulty: str, model_fn, max_retries: int = 3
) -> CodeSample | None:
    """코드를 생성하고 구문/실행 검증을 수행합니다."""
    for attempt in range(max_retries):
        response = model_fn(
            CODE_GEN_PROMPT.format(
                task_description=task,
                difficulty=difficulty,
            )
        )
 
        # 코드 블록 추출
        code = extract_code_block(response)
        if not code:
            continue
 
        # 구문 검증
        if not validate_python_syntax(code):
            continue
 
        # 실행 검증
        success, error = execute_code_safely(code)
        if success:
            return CodeSample(
                instruction=task,
                code=code,
                language="python",
                test_cases=[],
                is_valid=True,
            )
 
    return None
 
 
def extract_code_block(text: str) -> str:
    """마크다운 코드 블록에서 코드를 추출합니다."""
    import re
    pattern = r"```python\n(.*?)```"
    matches = re.findall(pattern, text, re.DOTALL)
    return matches[0].strip() if matches else ""

from enum import Enum
from dataclasses import dataclass
 
 
class DataType(Enum):
    INSTRUCTION = "instruction"
    CONVERSATION = "conversation"
    CLASSIFICATION = "classification"
    NER = "ner"
    SUMMARIZATION = "summarization"
    CODE = "code"
 
 
@dataclass
class GenerationConfig:
    data_type: DataType
    num_samples: int
    domain: str
    difficulty: str
    language: str = "ko"
    temperature: float = 0.7
    max_retries: int = 3
 
 
class TextSynthesisPipeline:
    """텍스트 합성 데이터 통합 파이프라인"""
 
    def __init__(self, model: str = "gpt-4o"):
        self.model = model
        self.generators = {
            DataType.INSTRUCTION: self._gen_instruction,
            DataType.CONVERSATION: self._gen_conversation,
            DataType.CLASSIFICATION: self._gen_classification,
            DataType.NER: self._gen_ner,
            DataType.SUMMARIZATION: self._gen_summarization,
            DataType.CODE: self._gen_code,
        }
 
    async def generate(self, config: GenerationConfig) -> list[dict]:
        """설정에 따라 합성 데이터를 생성합니다."""
        generator = self.generators[config.data_type]
        raw_data = await generator(config)
 
        # 공통 후처리
        filtered = self._filter_quality(raw_data)
        deduplicated = self._deduplicate(filtered)
 
        return deduplicated
 
    def _filter_quality(self, data: list[dict]) -> list[dict]:
        """품질 기준 미달 데이터를 필터링합니다."""
        return [
            item for item in data
            if item.get("quality_score", 0) > 0.7
        ]
 
    def _deduplicate(self, data: list[dict]) -> list[dict]:
        """유사한 데이터를 제거합니다."""
        # 임베딩 기반 유사도 검사
        # ROUGE-L 기반 중복 검사
        return data
 
    async def _gen_instruction(self, config):
        """지시-응답 쌍 생성"""
        pass
 
    async def _gen_conversation(self, config):
        """대화 데이터 생성"""
        pass
 
    async def _gen_classification(self, config):
        """분류 데이터 생성"""
        pass
 
    async def _gen_ner(self, config):
        """NER 데이터 생성"""
        pass
 
    async def _gen_summarization(self, config):
        """요약 데이터 생성"""
        pass
 
    async def _gen_code(self, config):
        """코드 데이터 생성"""
        pass