LLMによる高品質合成データ生成：学習データ不足を解消する実装ガイド

import asyncio
import logging
import json
from typing import List, Dict, Any
from pydantic import BaseModel, ValidationError
from openai import AsyncOpenAI
import os

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 環境変数からAPIキーを取得（実際には.envファイルなどを使用）
client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# データモデルの定義（生成したいデータの構造）
class TicketData(BaseModel):
    id: int
    category: str  # 例: 請求, 技術的問題, アカウント
    sentiment: str  # 例: ポジティブ, ニュートラル, ネガティブ
    text: str

class SyntheticDataGenerator:
    def __init__(self, model_name: str = "gpt-4o"):
        self.model_name = model_name
        self.generated_count = 0
        self.error_count = 0

    def _create_prompt(self, num_examples: int = 5) -> str:
        """プロンプトの作成。Few-Shotの例を含める"""
        examples = """
        Example 1:
        Category: 請求, Sentiment: ネガティブ
        Text: 請求書がまだ届いていません。いつになりますか？

        Example 2:
        Category: 技術的問題, Sentiment: ポジティブ
        Text: 新機能の使い方がとても分かりやすくて助かりました。

        Example 3:
        Category: アカウント, Sentiment: ニュートラル
        Text: パスワードを変更したいのですが、手順を教えてください。
        """
        return f"""
        あなたはカスタマーサポートのデータ生成アシスタントです。
        以下の例に基づいて、カスタマーサポートのチケットデータを生成してください。
        データは有効なJSON形式のリストで返してください。
        各オブジェクトは "id", "category", "sentiment", "text" キーを持つ必要があります。
        "id" は連番で振ってください。

        {examples}

        Generate {num_examples} new unique examples in JSON list format.
        """

    async def _generate_with_retry(self, prompt: str, max_retries: int = 3) -> str:
        """リトライロジックを含めた生成メソッド"""
        for attempt in range(max_retries):
            try:
                response = await client.chat.completions.create(
                    model=self.model_name,
                    messages=[
                        {"role": "system", "content": "You are a helpful data generator."},
                        {"role": "user", "content": prompt}
                    ],
                    temperature=0.7, # 創造性と一貫性のバランス
                    response_format={"type": "json_object"} # JSONモードを強制
                )
                return response.choices[0].message.content
            except Exception as e:
                logger.warning(f"Attempt {attempt + 1} failed: {e}")
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt) # Exponential Backoff
        
        return ""

    def _validate_data(self, raw_data: str) -> List[TicketData]:
        """生成されたJSONデータの検証"""
        try:
            data_json = json.loads(raw_data)
            if "items" in data_json:
                data_list = data_json["items"]
            elif isinstance(data_json, list):
                data_list = data_json
            else:
                # 単一のオブジェクトや予期せぬ構造への対応
                logger.warning("Unexpected JSON structure, trying to adapt...")
                return []

            validated_tickets = []
            for item in data_list:
                try:
                    ticket = TicketData(**item)
                    validated_tickets.append(ticket)
                except ValidationError as ve:
                    logger.error(f"Validation error for item {item}: {ve}")
                    self.error_count += 1
            
            return validated_tickets
        except json.JSONDecodeError as e:
            logger.error(f"JSON decode error: {e}")
            self.error_count += 1
            return []

    async def generate_batch(self, batch_size: int = 5) -> List[TicketData]:
        """1バッチ分のデータを生成・検証する"""
        logger.info(f"Generating batch of {batch_size} items...")
        prompt = self._create_prompt(batch_size)
        
        try:
            raw_content = await self._generate_with_retry(prompt)
            validated_data = self._validate_data(raw_content)
            self.generated_count += len(validated_data)
            logger.info(f"Successfully generated and validated {len(validated_data)} items.")
            return validated_data
        except Exception as e:
            logger.error(f"Failed to generate batch: {e}")
            return []

async def main():
    generator = SyntheticDataGenerator()
    all_tickets = []
    
    # 合計20個のデータを生成する（4バッチ）
    for _ in range(4):
        batch = await generator.generate_batch(batch_size=5)
        all_tickets.extend(batch)
        # APIレートリミットを避けるための少しの待機
        await asyncio.sleep(1)

    logger.info(f"Generation complete. Total valid: {generator.generated_count}, Errors: {generator.error_count}")
    
    # 結果のサンプル表示
    for ticket in all_tickets[:3]:
        print(f"ID: {ticket.id} | Cat: {ticket.category} | Sent: {ticket.sentiment}")
        print(f"Text: {ticket.text}\n")

if __name__ == "__main__":
    asyncio.run(main())