【2025年版】AI生成コンテンツ技術設定完全マニュアル：品質とパフォーマンスを最大化する実装ガイド

graph TD
    A[Content Request API] --> B[Load Balancer]
    B --> C[AI Model Router]
    C --> D[GPT-4.5 Endpoint]
    C --> E[Claude 3.7 Endpoint]
    C --> F[Custom Model Endpoint]
    
    G[Quality Checker] --> H[Content Validation]
    H --> I[Human Review Queue]
    H --> J[Auto Publish]
    
    K[Monitoring System] --> L[Performance Metrics]
    K --> M[Quality Metrics]
    K --> N[Cost Tracking]

# docker-compose.yml (基本構成)
version: '3.8'
services:
  ai-content-api:
    image: ai-content-platform:latest
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - CLAUDE_API_KEY=${CLAUDE_API_KEY}
      - REDIS_URL=redis://redis:6379
      - DATABASE_URL=postgresql://postgres:password@postgres:5432/ai_content
    ports:
      - "8000:8000"
    depends_on:
      - redis
      - postgres
      
  quality-checker:
    image: content-quality-checker:latest
    environment:
      - CONTENT_API_URL=http://ai-content-api:8000
      - QUALITY_THRESHOLD=0.75
    
  redis:
    image: redis:7-alpine
    
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: ai_content
      POSTGRES_PASSWORD: password

# config/settings.py
import os
from typing import Dict, Any

class AIModelConfig:
    """AI モデル設定の基底クラス"""
    
    def __init__(self, environment: str = "production"):
        self.environment = environment
        self.base_config = self._load_base_config()
        
    def _load_base_config(self) -> Dict[str, Any]:
        """環境別基本設定"""
        configs = {
            "development": {
                "gpt4_5": {
                    "temperature": 0.9,
                    "max_tokens": 2000,
                    "top_p": 0.95,
                    "frequency_penalty": 0.1,
                    "presence_penalty": 0.1,
                    "timeout": 60,
                    "retry_count": 2
                },
                "claude_3_7": {
                    "temperature": 0.8,
                    "max_tokens": 4000,
                    "top_p": 0.9,
                    "timeout": 90,
                    "retry_count": 3
                }
            },
            "production": {
                "gpt4_5": {
                    "temperature": 0.7,
                    "max_tokens": 3000,
                    "top_p": 0.85,
                    "frequency_penalty": 0.2,
                    "presence_penalty": 0.15,
                    "timeout": 30,
                    "retry_count": 3
                },
                "claude_3_7": {
                    "temperature": 0.6,
                    "max_tokens": 4000,
                    "top_p": 0.8,
                    "timeout": 45,
                    "retry_count": 3
                }
            }
        }
        return configs.get(self.environment, configs["production"])

# models/gpt45_optimizer.py
class GPT45ContentOptimizer:
    """GPT-4.5 コンテンツタイプ別最適化設定"""
    
    CONTENT_TYPE_SETTINGS = {
        "blog_article": {
            "temperature": 0.7,
            "max_tokens": 3500,
            "top_p": 0.9,
            "frequency_penalty": 0.3,  # 重複表現を避ける
            "presence_penalty": 0.2,   # トピックの多様性
            "system_message": """あなたは経験豊富なコンテンツライターです。
                                 読者にとって価値のある、独自性のある記事を作成してください。""",
            "quality_check": {
                "min_uniqueness": 0.8,
                "max_repetition": 0.15,
                "required_structure": ["intro", "body", "conclusion"]
            }
        },
        
        "product_description": {
            "temperature": 0.5,
            "max_tokens": 800,
            "top_p": 0.85,
            "frequency_penalty": 0.4,  # より厳格な重複回避
            "presence_penalty": 0.1,
            "system_message": """あなたはプロダクトマーケティングの専門家です。
                                 魅力的で正確な商品説明を作成してください。""",
            "quality_check": {
                "min_uniqueness": 0.9,
                "max_repetition": 0.1,
                "required_elements": ["features", "benefits", "specifications"]
            }
        },
        
        "technical_documentation": {
            "temperature": 0.3,
            "max_tokens": 4000,
            "top_p": 0.8,
            "frequency_penalty": 0.1,  # 技術用語の重複は許容
            "presence_penalty": 0.05,
            "system_message": """あなたは技術文書作成の専門家です。
                                 正確で理解しやすい技術説明を作成してください。""",
            "quality_check": {
                "min_accuracy": 0.95,
                "max_ambiguity": 0.1,
                "required_structure": ["overview", "details", "examples"]
            }
        }
    }
    
    def get_optimized_settings(self, content_type: str, 
                             custom_params: Dict = None) -> Dict:
        """最適化されたモデル設定を取得"""
        base_settings = self.CONTENT_TYPE_SETTINGS.get(
            content_type, 
            self.CONTENT_TYPE_SETTINGS["blog_article"]
        )
        
        if custom_params:
            # カスタムパラメーターで上書き
            base_settings.update(custom_params)
            
        return base_settings

# models/claude37_reasoning.py
class Claude37ReasoningOptimizer:
    """Claude 3.7 推論モード最適化"""
    
    def __init__(self):
        self.reasoning_patterns = {
            "analytical": {
                "thinking_depth": "deep",
                "step_by_step": True,
                "verification": True,
                "reasoning_display": "visible"
            },
            "creative": {
                "thinking_depth": "moderate",
                "brainstorming": True,
                "multiple_perspectives": True,
                "reasoning_display": "hidden"
            },
            "factual": {
                "thinking_depth": "thorough",
                "fact_checking": True,
                "source_verification": True,
                "reasoning_display": "summary"
            }
        }
    
    def optimize_for_reasoning(self, content_type: str, 
                             complexity: str = "medium") -> Dict:
        """推論タスク用の最適化設定"""
        base_config = {
            "model": "claude-3-7-sonnet-20250219",
            "temperature": 0.4,  # 推論では低めに設定
            "max_tokens": 4000,
            "reasoning_mode": True,
            "thinking_budget": self._get_thinking_budget(complexity)
        }
        
        pattern = self.reasoning_patterns.get(content_type, "analytical")
        base_config.update(pattern)
        
        return base_config
    
    def _get_thinking_budget(self, complexity: str) -> int:
        """思考時間の予算設定"""
        budgets = {
            "simple": 1000,    # トークン
            "medium": 2000,
            "complex": 5000,
            "expert": 10000
        }
        return budgets.get(complexity, 2000)

# models/multimodal_processor.py
import base64
from typing import List, Dict, Union

class MultimodalContentProcessor:
    """マルチモーダルコンテンツ処理システム"""
    
    def __init__(self):
        self.supported_formats = {
            "image": ["png", "jpg", "jpeg", "webp"],
            "audio": ["mp3", "wav", "m4a"],
            "video": ["mp4", "avi", "mov"]
        }
        
        self.processing_config = {
            "image_analysis": {
                "max_resolution": "1920x1080",
                "quality": "high",
                "detail_level": "comprehensive",
                "ocr_enabled": True,
                "object_detection": True
            },
            "audio_processing": {
                "transcription": True,
                "speaker_identification": False,
                "emotion_analysis": True,
                "language_detection": True
            },
            "content_synthesis": {
                "cross_modal_consistency": True,
                "context_awareness": True,
                "accessibility_features": True
            }
        }
    
    async def process_multimodal_content(self, 
                                       text_input: str,
                                       media_files: List[Dict],
                                       output_format: str = "enhanced_text") -> Dict:
        """マルチモーダルコンテンツの統合処理"""
        
        # メディアファイルの前処理
        processed_media = await self._preprocess_media(media_files)
        
        # GPT-4.5のマルチモーダル機能を活用
        multimodal_prompt = self._create_multimodal_prompt(
            text_input, processed_media
        )
        
        response = await self._call_multimodal_api(
            multimodal_prompt, 
            self.processing_config
        )
        
        return {
            "enhanced_content": response.text,
            "media_insights": response.media_analysis,
            "quality_score": await self._calculate_quality_score(response),
            "accessibility_data": response.accessibility_info
        }
    
    async def _preprocess_media(self, media_files: List[Dict]) -> List[Dict]:
        """メディアファイルの前処理"""
        processed = []
        for media in media_files:
            if media["type"] == "image":
                # 画像の最適化・圧縮
                optimized = await self._optimize_image(media["data"])
                processed.append({
                    "type": "image",
                    "data": optimized,
                    "metadata": media.get("metadata", {})
                })
            elif media["type"] == "audio":
                # 音声の前処理
                transcription = await self._transcribe_audio(media["data"])
                processed.append({
                    "type": "audio_text",
                    "data": transcription,
                    "metadata": media.get("metadata", {})
                })
        return processed

# quality/content_evaluator.py
import asyncio
from typing import Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum

class QualityLevel(Enum):
    EXCELLENT = 90
    GOOD = 75
    ACCEPTABLE = 60
    POOR = 40
    UNACCEPTABLE = 0

@dataclass
class QualityMetrics:
    accuracy: float
    uniqueness: float
    readability: float
    relevance: float
    engagement: float
    compliance: float
    overall_score: float

class ContentQualityEvaluator:
    """AI生成コンテンツ品質評価システム"""
    
    def __init__(self):
        self.evaluation_weights = {
            "accuracy": 0.25,      # 情報の正確性
            "uniqueness": 0.20,    # 独自性・オリジナリティ
            "readability": 0.15,   # 読みやすさ
            "relevance": 0.15,     # 関連性・適合性
            "engagement": 0.15,    # エンゲージメント予測
            "compliance": 0.10     # コンプライアンス適合
        }
        
        self.quality_checkers = {
            "fact_checker": FactChecker(),
            "plagiarism_detector": PlagiarismDetector(),
            "readability_analyzer": ReadabilityAnalyzer(),
            "compliance_checker": ComplianceChecker()
        }
    
    async def evaluate_content(self, content: str, 
                             metadata: Dict = None) -> QualityMetrics:
        """コンテンツの総合品質評価"""
        
        # 並列で各品質指標を評価
        evaluation_tasks = [
            self._evaluate_accuracy(content, metadata),
            self._evaluate_uniqueness(content),
            self._evaluate_readability(content),
            self._evaluate_relevance(content, metadata),
            self._evaluate_engagement(content),
            self._evaluate_compliance(content)
        ]
        
        results = await asyncio.gather(*evaluation_tasks)
        
        # 重み付きスコア計算
        weighted_scores = {}
        for i, metric in enumerate(self.evaluation_weights.keys()):
            weighted_scores[metric] = results[i]
        
        overall_score = sum(
            score * self.evaluation_weights[metric]
            for metric, score in weighted_scores.items()
        )
        
        return QualityMetrics(
            accuracy=weighted_scores["accuracy"],
            uniqueness=weighted_scores["uniqueness"],
            readability=weighted_scores["readability"],
            relevance=weighted_scores["relevance"],
            engagement=weighted_scores["engagement"],
            compliance=weighted_scores["compliance"],
            overall_score=overall_score
        )
    
    async def _evaluate_accuracy(self, content: str, 
                               metadata: Dict) -> float:
        """情報の正確性評価"""
        # ファクトチェックAPI連携
        fact_check_results = await self.quality_checkers[
            "fact_checker"
        ].verify_facts(content)
        
        # 専門分野の場合は追加検証
        if metadata and metadata.get("domain") in ["medical", "legal", "financial"]:
            domain_check = await self._domain_specific_check(
                content, metadata["domain"]
            )
            return (fact_check_results + domain_check) / 2
        
        return fact_check_results
    
    async def _evaluate_uniqueness(self, content: str) -> float:
        """独自性・オリジナリティ評価"""
        # 盗用検出
        plagiarism_score = await self.quality_checkers[
            "plagiarism_detector"
        ].check_plagiarism(content)
        
        # 言い回しの独自性チェック
        originality_score = await self._check_originality(content)
        
        # コンテンツの新規性評価
        novelty_score = await self._check_novelty(content)
        
        return (plagiarism_score + originality_score + novelty_score) / 3

# monitoring/quality_monitor.py
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List
import logging

class QualityMonitor:
    """リアルタイム品質監視システム"""
    
    def __init__(self):
        self.quality_thresholds = {
            "critical": 40,    # この値以下は即座に停止
            "warning": 60,     # この値以下は警告
            "target": 80       # 目標品質レベル
        }
        
        self.monitoring_metrics = [
            "average_quality_score",
            "quality_variance",
            "rejection_rate",
            "processing_time",
            "cost_per_quality_unit"
        ]
        
        self.alert_handlers = {
            "critical": self._handle_critical_alert,
            "warning": self._handle_warning_alert,
            "info": self._handle_info_alert
        }
    
    async def start_monitoring(self):
        """品質監視の開始"""
        logging.info("Quality monitoring started")
        
        while True:
            try:
                current_metrics = await self._collect_metrics()
                await self._analyze_quality_trends(current_metrics)
                await self._check_thresholds(current_metrics)
                
                # 5分間隔での監視
                await asyncio.sleep(300)
                
            except Exception as e:
                logging.error(f"Quality monitoring error: {e}")
                await asyncio.sleep(60)  # エラー時は短い間隔で再試行
    
    async def _collect_metrics(self) -> Dict:
        """現在の品質メトリクスを収集"""
        # 過去1時間のコンテンツ品質データを取得
        recent_scores = await self._get_recent_quality_scores(
            timedelta(hours=1)
        )
        
        if not recent_scores:
            return {"status": "no_data"}
        
        return {
            "timestamp": datetime.now(),
            "average_quality": sum(recent_scores) / len(recent_scores),
            "quality_variance": self._calculate_variance(recent_scores),
            "sample_count": len(recent_scores),
            "min_quality": min(recent_scores),
            "max_quality": max(recent_scores),
            "trend": self._calculate_trend(recent_scores)
        }
    
    async def _analyze_quality_trends(self, metrics: Dict):
        """品質トレンドの分析と予測"""
        if metrics.get("status") == "no_data":
            return
        
        # トレンド分析
        if metrics["trend"] < -5:  # 品質が急激に低下
            await self._trigger_alert("warning", {
                "message": "Quality trend declining rapidly",
                "current_average": metrics["average_quality"],
                "trend": metrics["trend"]
            })
        
        # 分散チェック（品質の安定性）
        if metrics["quality_variance"] > 200:  # 品質のばらつきが大きい
            await self._trigger_alert("info", {
                "message": "High quality variance detected",
                "variance": metrics["quality_variance"],
                "recommendation": "Review model parameters"
            })
    
    async def _handle_critical_alert(self, alert_data: Dict):
        """緊急アラートの処理"""
        # AI生成の一時停止
        await self._pause_ai_generation()
        
        # 管理者への緊急通知
        await self._send_emergency_notification(alert_data)
        
        # 自動復旧の試行
        await self._attempt_auto_recovery()
    
    async def _attempt_auto_recovery(self):
        """自動復旧の試行"""
        recovery_actions = [
            self._reset_model_parameters,
            self._clear_model_cache,
            self._switch_to_backup_model,
            self._restart_quality_checkers
        ]
        
        for action in recovery_actions:
            try:
                await action()
                # 復旧確認
                if await self._verify_recovery():
                    logging.info(f"Auto-recovery successful: {action.__name__}")
                    await self._resume_ai_generation()
                    return
            except Exception as e:
                logging.error(f"Recovery action failed: {action.__name__}: {e}")
        
        # 全ての自動復旧が失敗した場合
        await self._escalate_to_human_intervention()

# security/ai_security_manager.py
import hashlib
import hmac
import jwt
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from cryptography.fernet import Fernet

class AISecurityManager:
    """AI生成コンテンツのセキュリティ管理"""
    
    def __init__(self, encryption_key: str, jwt_secret: str):
        self.encryption_key = encryption_key
        self.jwt_secret = jwt_secret
        self.cipher = Fernet(encryption_key.encode())
        
        self.security_policies = {
            "data_retention": {
                "generated_content": timedelta(days=365),
                "user_prompts": timedelta(days=90),
                "quality_metrics": timedelta(days=180),
                "error_logs": timedelta(days=30)
            },
            "access_control": {
                "api_key_rotation": timedelta(days=30),
                "session_timeout": timedelta(hours=8),
                "max_failed_attempts": 5,
                "lockout_duration": timedelta(minutes=30)
            },
            "audit_requirements": {
                "log_all_requests": True,
                "log_generated_content": True,
                "log_quality_scores": True,
                "anonymize_user_data": True
            }
        }
    
    def encrypt_sensitive_data(self, data: str) -> str:
        """機密データの暗号化"""
        return self.cipher.encrypt(data.encode()).decode()
    
    def decrypt_sensitive_data(self, encrypted_data: str) -> str:
        """暗号化データの復号"""
        return self.cipher.decrypt(encrypted_data.encode()).decode()
    
    def generate_secure_api_token(self, user_id: str, 
                                 permissions: List[str]) -> str:
        """セキュアなAPIトークン生成"""
        payload = {
            "user_id": user_id,
            "permissions": permissions,
            "issued_at": datetime.utcnow().timestamp(),
            "expires_at": (datetime.utcnow() + timedelta(hours=24)).timestamp()
        }
        
        return jwt.encode(payload, self.jwt_secret, algorithm="HS256")
    
    def validate_api_token(self, token: str) -> Optional[Dict]:
        """APIトークンの検証"""
        try:
            payload = jwt.decode(token, self.jwt_secret, algorithms=["HS256"])
            
            # 有効期限チェック
            if datetime.utcnow().timestamp() > payload["expires_at"]:
                return None
                
            return payload
        except jwt.InvalidTokenError:
            return None
    
    async def audit_content_generation(self, request_data: Dict, 
                                     response_data: Dict, 
                                     user_context: Dict):
        """コンテンツ生成の監査ログ"""
        audit_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_context.get("user_id"),
            "request_hash": self._hash_request(request_data),
            "model_used": request_data.get("model"),
            "content_type": request_data.get("content_type"),
            "quality_score": response_data.get("quality_score"),
            "processing_time": response_data.get("processing_time"),
            "cost": response_data.get("cost"),
            "compliance_flags": response_data.get("compliance_flags", [])
        }
        
        # 個人情報の匿名化
        audit_entry = await self._anonymize_audit_data(audit_entry)
        
        # 監査ログの保存
        await self._store_audit_log(audit_entry)
    
    def _hash_request(self, request_data: Dict) -> str:
        """リクエストデータのハッシュ化"""
        # 機密情報を除外してハッシュ化
        filtered_data = {
            k: v for k, v in request_data.items() 
            if k not in ["api_key", "user_token", "personal_data"]
        }
        
        data_string = str(sorted(filtered_data.items()))
        return hashlib.sha256(data_string.encode()).hexdigest()

# compliance/compliance_checker.py
import re
from typing import Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum

class ComplianceLevel(Enum):
    COMPLIANT = "compliant"
    WARNING = "warning"  
    VIOLATION = "violation"

@dataclass
class ComplianceResult:
    level: ComplianceLevel
    violations: List[str]
    warnings: List[str]
    recommendations: List[str]
    confidence_score: float

class ComplianceChecker:
    """コンプライアンス自動チェックシステム"""
    
    def __init__(self):
        self.compliance_rules = {
            "gdpr": {
                "personal_data_patterns": [
                    r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
                    r'\b[\w\.-]+@[\w\.-]+\.\w+\b',  # Email
                    r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',  # Credit Card
                ],
                "required_disclosures": [
                    "data processing purpose",
                    "data retention period",
                    "user rights information"
                ]
            },
            "medical": {
                "prohibited_claims": [
                    r'cure[sd]?\s+(?:cancer|diabetes|covid)',
                    r'treat[sd]?\s+(?:depression|anxiety)',
                    r'diagnos[es]\s+(?:any|all)\s+(?:disease|condition)'
                ],
                "required_disclaimers": [
                    "not medical advice",
                    "consult healthcare provider",
                    "individual results may vary"
                ]
            },
            "financial": {
                "risk_disclosure": [
                    r'guaranteed?\s+returns?',
                    r'no\s+risk',
                    r'safe\s+investment'
                ],
                "required_warnings": [
                    "past performance disclaimer",
                    "risk of loss warning",
                    "regulatory compliance note"
                ]
            }
        }
        
        self.industry_validators = {
            "healthcare": HealthcareComplianceValidator(),
            "finance": FinanceComplianceValidator(),
            "legal": LegalComplianceValidator(),
            "education": EducationComplianceValidator()
        }
    
    async def check_compliance(self, content: str, 
                             industry: str = "general",
                             target_region: str = "global") -> ComplianceResult:
        """総合コンプライアンスチェック"""
        
        violations = []
        warnings = []
        recommendations = []
        
        # GDPR/プライバシー関連チェック
        if target_region in ["eu", "global"]:
            gdpr_result = await self._check_gdpr_compliance(content)
            violations.extend(gdpr_result["violations"])
            warnings.extend(gdpr_result["warnings"])
        
        # 業界固有のコンプライアンスチェック
        if industry != "general" and industry in self.industry_validators:
            industry_result = await self.industry_validators[industry].validate(content)
            violations.extend(industry_result["violations"])
            warnings.extend(industry_result["warnings"])
            recommendations.extend(industry_result["recommendations"])
        
        # コンテンツの有害性チェック
        toxicity_result = await self._check_content_toxicity(content)
        violations.extend(toxicity_result["violations"])
        
        # 全体的なコンプライアンスレベル決定
        if violations:
            level = ComplianceLevel.VIOLATION
            confidence_score = 0.9  # 違反がある場合は高い確信度
        elif warnings:
            level = ComplianceLevel.WARNING
            confidence_score = 0.7
        else:
            level = ComplianceLevel.COMPLIANT
            confidence_score = 0.8
        
        return ComplianceResult(
            level=level,
            violations=violations,
            warnings=warnings,
            recommendations=recommendations,
            confidence_score=confidence_score
        )
    
    async def _check_gdpr_compliance(self, content: str) -> Dict:
        """GDPR コンプライアンスチェック"""
        violations = []
        warnings = []
        
        # 個人データパターンの検出
        for pattern in self.compliance_rules["gdpr"]["personal_data_patterns"]:
            if re.search(pattern, content, re.IGNORECASE):
                violations.append(f"Potential personal data detected: {pattern}")
        
        # 必要な開示情報の確認
        required_disclosures = self.compliance_rules["gdpr"]["required_disclosures"]
        for disclosure in required_disclosures:
            if disclosure.replace(" ", "").lower() not in content.replace(" ", "").lower():
                warnings.append(f"Missing GDPR disclosure: {disclosure}")
        
        return {"violations": violations, "warnings": warnings}
    
    async def _check_content_toxicity(self, content: str) -> Dict:
        """コンテンツの有害性チェック"""
        violations = []
        
        # 有害なコンテンツパターン
        toxic_patterns = [
            r'hate\s+speech',
            r'discriminat[eioy]',
            r'harassment',
            r'violent\s+content'
        ]
        
        for pattern in toxic_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                violations.append(f"Potentially toxic content: {pattern}")
        
        return {"violations": violations}

# optimization/model_selector.py
from typing import Dict, List, Optional, Tuple
import asyncio
from dataclasses import dataclass

@dataclass
class ModelPerformance:
    accuracy: float
    speed: float  # tokens per second
    cost_per_token: float
    quality_consistency: float
    specialization_score: float

class DynamicModelSelector:
    """動的モデル選択システム"""
    
    def __init__(self):
        self.model_capabilities = {
            "gpt-4.5": ModelPerformance(
                accuracy=0.92,
                speed=45.0,
                cost_per_token=0.00003,
                quality_consistency=0.88,
                specialization_score=0.85
            ),
            "claude-3.7-sonnet": ModelPerformance(
                accuracy=0.90,
                speed=38.0,
                cost_per_token=0.000025,
                quality_consistency=0.92,
                specialization_score=0.90
            ),
            "o3-pro": ModelPerformance(
                accuracy=0.95,
                speed=12.0,
                cost_per_token=0.00015,
                quality_consistency=0.95,
                specialization_score=0.98
            )
        }
        
        self.optimization_strategies = {
            "cost_optimized": {
                "weight_accuracy": 0.3,
                "weight_speed": 0.2,
                "weight_cost": 0.4,
                "weight_quality": 0.1
            },
            "quality_optimized": {
                "weight_accuracy": 0.4,
                "weight_speed": 0.1,
                "weight_cost": 0.1,
                "weight_quality": 0.4
            },
            "speed_optimized": {
                "weight_accuracy": 0.2,
                "weight_speed": 0.5,
                "weight_cost": 0.2,
                "weight_quality": 0.1
            },
            "balanced": {
                "weight_accuracy": 0.25,
                "weight_speed": 0.25,
                "weight_cost": 0.25,
                "weight_quality": 0.25
            }
        }
    
    async def select_optimal_model(self, 
                                 content_requirements: Dict,
                                 optimization_strategy: str = "balanced") -> Tuple[str, Dict]:
        """最適なモデルの選択"""
        
        # 要件の解析
        content_type = content_requirements.get("type", "general")
        quality_threshold = content_requirements.get("min_quality", 0.7)
        max_cost = content_requirements.get("max_cost_per_request", 1.0)
        max_processing_time = content_requirements.get("max_processing_time", 60)
        
        # 各モデルのスコア計算
        model_scores = {}
        strategy = self.optimization_strategies[optimization_strategy]
        
        for model_name, performance in self.model_capabilities.items():
            # 制約チェック
            if not self._meets_constraints(performance, quality_threshold, 
                                         max_cost, max_processing_time, 
                                         content_requirements):
                continue
            
            # 総合スコア計算
            score = (
                performance.accuracy * strategy["weight_accuracy"] +
                (performance.speed / 100) * strategy["weight_speed"] +
                (1 - performance.cost_per_token * 1000) * strategy["weight_cost"] +
                performance.quality_consistency * strategy["weight_quality"]
            )
            
            # コンテンツタイプ固有の調整
            if content_type in ["technical", "legal", "medical"]:
                score += performance.specialization_score * 0.2
            
            model_scores[model_name] = score
        
        if not model_scores:
            raise ValueError("No model meets the specified constraints")
        
        # 最高スコアのモデルを選択
        best_model = max(model_scores.items(), key=lambda x: x[1])
        
        # 推奨設定を取得
        recommended_settings = await self._get_model_settings(
            best_model[0], content_requirements
        )
        
        return best_model[0], recommended_settings
    
    def _meets_constraints(self, performance: ModelPerformance,
                          min_quality: float, max_cost: float,
                          max_time: float, requirements: Dict) -> bool:
        """制約条件のチェック"""
        
        # 品質制約
        if performance.accuracy < min_quality:
            return False
        
        # コスト制約（概算）
        estimated_tokens = requirements.get("estimated_tokens", 2000)
        estimated_cost = performance.cost_per_token * estimated_tokens
        if estimated_cost > max_cost:
            return False
        
        # 処理時間制約（概算）
        estimated_time = estimated_tokens / performance.speed
        if estimated_time > max_time:
            return False
        
        return True
    
    async def _get_model_settings(self, model_name: str, 
                                requirements: Dict) -> Dict:
        """モデル固有の推奨設定"""
        
        base_settings = {
            "gpt-4.5": {
                "temperature": 0.7,
                "max_tokens": 2000,
                "top_p": 0.9,
                "frequency_penalty": 0.1,
                "presence_penalty": 0.1
            },
            "claude-3.7-sonnet": {
                "temperature": 0.6,
                "max_tokens": 4000,
                "top_p": 0.8,
                "reasoning_mode": True
            },
            "o3-pro": {
                "temperature": 0.4,
                "max_tokens": 4000,
                "thinking_budget": 5000,
                "verification_enabled": True
            }
        }
        
        settings = base_settings.get(model_name, base_settings["gpt-4.5"]).copy()
        
        # 要件に基づく動的調整
        content_type = requirements.get("type", "general")
        if content_type in ["creative", "marketing"]:
            settings["temperature"] = min(settings["temperature"] + 0.2, 1.0)
        elif content_type in ["technical", "legal"]:
            settings["temperature"] = max(settings["temperature"] - 0.2, 0.1)
        
        return settings

# scaling/auto_scaler.py
import asyncio
from typing import Dict, List
import aiohttp
from datetime import datetime, timedelta

class AIContentAutoScaler:
    """AI コンテンツ生成の自動スケーリング"""
    
    def __init__(self):
        self.scaling_metrics = {
            "request_rate": {"current": 0, "threshold_scale_up": 100, "threshold_scale_down": 20},
            "avg_response_time": {"current": 0, "threshold_scale_up": 30, "threshold_scale_down": 5},
            "error_rate": {"current": 0, "threshold_scale_up": 0.05, "threshold_scale_down": 0.01},
            "queue_length": {"current": 0, "threshold_scale_up": 50, "threshold_scale_down": 10}
        }
        
        self.model_instances = {
            "gpt-4.5": {"active": 2, "max": 10, "min": 1},
            "claude-3.7": {"active": 1, "max": 8, "min": 1},
            "o3-pro": {"active": 1, "max": 3, "min": 1}  # 高コストモデルは制限
        }
        
        self.load_balancer = LoadBalancer()
    
    async def monitor_and_scale(self):
        """監視とスケーリングのメインループ"""
        while True:
            try:
                # メトリクス収集
                current_metrics = await self._collect_current_metrics()
                
                # スケーリング判断
                scaling_decision = await self._analyze_scaling_needs(current_metrics)
                
                # スケーリング実行
                if scaling_decision["action"] != "no_action":
                    await self._execute_scaling(scaling_decision)
                
                # ロードバランサー設定更新
                await self.load_balancer.update_instance_weights()
                
                await asyncio.sleep(30)  # 30秒間隔で監視
                
            except Exception as e:
                logging.error(f"Auto-scaling error: {e}")
                await asyncio.sleep(60)
    
    async def _collect_current_metrics(self) -> Dict:
        """現在のシステムメトリクス収集"""
        metrics = {}
        
        # リクエスト率の計算
        recent_requests = await self._get_request_count(timedelta(minutes=5))
        metrics["request_rate"] = recent_requests / 5  # per minute
        
        # 平均応答時間
        metrics["avg_response_time"] = await self._get_avg_response_time(timedelta(minutes=5))
        
        # エラー率
        recent_errors = await self._get_error_count(timedelta(minutes=5))
        metrics["error_rate"] = recent_errors / max(recent_requests, 1)
        
        # キューの長さ
        metrics["queue_length"] = await self._get_queue_length()
        
        return metrics
    
    async def _analyze_scaling_needs(self, current_metrics: Dict) -> Dict:
        """スケーリング必要性の分析"""
        scale_up_signals = 0
        scale_down_signals = 0
        
        for metric_name, metric_data in self.scaling_metrics.items():
            current_value = current_metrics.get(metric_name, 0)
            
            if current_value > metric_data["threshold_scale_up"]:
                scale_up_signals += 1
            elif current_value < metric_data["threshold_scale_down"]:
                scale_down_signals += 1
        
        # スケーリング判断
        if scale_up_signals >= 2:  # 2つ以上の指標が閾値超過
            return {
                "action": "scale_up",
                "priority_model": self._get_highest_demand_model(),
                "scale_factor": min(2, scale_up_signals)
            }
        elif scale_down_signals >= 3 and self._can_scale_down():  # より慎重にスケールダウン
            return {
                "action": "scale_down",
                "target_model": self._get_lowest_utilization_model(),
                "scale_factor": 1
            }
        else:
            return {"action": "no_action"}
    
    async def _execute_scaling(self, decision: Dict):
        """スケーリングの実行"""
        action = decision["action"]
        
        if action == "scale_up":
            model = decision["priority_model"]
            scale_factor = decision["scale_factor"]
            
            current_instances = self.model_instances[model]["active"]
            max_instances = self.model_instances[model]["max"]
            
            new_instances = min(
                current_instances + scale_factor,
                max_instances
            )
            
            if new_instances > current_instances:
                await self._start_model_instances(model, new_instances - current_instances)
                self.model_instances[model]["active"] = new_instances
                logging.info(f"Scaled up {model}: {current_instances} -> {new_instances}")
        
        elif action == "scale_down":
            model = decision["target_model"]
            current_instances = self.model_instances[model]["active"]
            min_instances = self.model_instances[model]["min"]
            
            new_instances = max(current_instances - 1, min_instances)
            
            if new_instances < current_instances:
                await self._stop_model_instances(model, current_instances - new_instances)
                self.model_instances[model]["active"] = new_instances
                logging.info(f"Scaled down {model}: {current_instances} -> {new_instances}")

# monitoring/dashboard.py
from typing import Dict, List
import asyncio
import json
from datetime import datetime, timedelta

class AIContentMonitoringDashboard:
    """AI コンテンツ生成システム監視ダッシュボード"""
    
    def __init__(self):
        self.kpi_definitions = {
            "system_health": {
                "uptime_percentage": {"target": 99.9, "critical": 99.0},
                "avg_response_time": {"target": 5.0, "critical": 15.0},
                "error_rate": {"target": 0.01, "critical": 0.05}
            },
            "content_quality": {
                "avg_quality_score": {"target": 80.0, "critical": 60.0},
                "human_approval_rate": {"target": 95.0, "critical": 85.0},
                "uniqueness_score": {"target": 85.0, "critical": 70.0}
            },
            "cost_efficiency": {
                "cost_per_content": {"target": 0.50, "critical": 2.00},
                "token_utilization": {"target": 0.80, "critical": 0.50},
                "processing_efficiency": {"target": 0.75, "critical": 0.50}
            },
            "user_satisfaction": {
                "user_rating": {"target": 4.5, "critical": 3.5},
                "content_acceptance": {"target": 0.90, "critical": 0.70},
                "revision_rate": {"target": 0.20, "critical": 0.50}
            }
        }
        
        self.alert_channels = {
            "email": {"enabled": True, "threshold": "warning"},
            "slack": {"enabled": True, "threshold": "info"},
            "sms": {"enabled": True, "threshold": "critical"},
            "webhook": {"enabled": True, "threshold": "all"}
        }
    
    async def generate_dashboard_data(self, time_range: timedelta = timedelta(hours=24)) -> Dict:
        """ダッシュボード表示用データの生成"""
        end_time = datetime.now()
        start_time = end_time - time_range
        
        dashboard_data = {
            "timestamp": end_time.isoformat(),
            "time_range": str(time_range),
            "system_status": await self._get_system_status(),
            "kpi_summary": await self._calculate_kpi_summary(start_time, end_time),
            "performance_trends": await self._get_performance_trends(start_time, end_time),
            "recent_alerts": await self._get_recent_alerts(timedelta(hours=6)),
            "cost_analysis": await self._get_cost_analysis(start_time, end_time),
            "quality_distribution": await self._get_quality_distribution(start_time, end_time)
        }
        
        return dashboard_data
    
    async def _calculate_kpi_summary(self, start_time: datetime, end_time: datetime) -> Dict:
        """KPI サマリーの計算"""
        kpi_summary = {}
        
        for category, kpis in self.kpi_definitions.items():
            category_data = {}
            
            for kpi_name, thresholds in kpis.items():
                current_value = await self._get_kpi_value(kpi_name, start_time, end_time)
                
                # ステータス判定
                if current_value >= thresholds["target"]:
                    status = "excellent"
                elif current_value >= thresholds["critical"]:
                    status = "good"
                else:
                    status = "critical"
                
                category_data[kpi_name] = {
                    "current_value": current_value,
                    "target": thresholds["target"],
                    "critical_threshold": thresholds["critical"],
                    "status": status,
                    "trend": await self._calculate_trend(kpi_name, start_time, end_time)
                }
            
            kpi_summary[category] = category_data
        
        return kpi_summary
    
    async def _get_performance_trends(self, start_time: datetime, end_time: datetime) -> Dict:
        """パフォーマンストレンドの取得"""
        # 時間軸でのデータポイント生成（1時間間隔）
        time_points = []
        current_time = start_time
        while current_time <= end_time:
            time_points.append(current_time)
            current_time += timedelta(hours=1)
        
        trends = {
            "request_volume": [],
            "average_quality": [],
            "response_time": [],
            "cost_per_hour": [],
            "error_rate": []
        }
        
        for time_point in time_points:
            hour_start = time_point
            hour_end = time_point + timedelta(hours=1)
            
            trends["request_volume"].append({
                "timestamp": time_point.isoformat(),
                "value": await self._get_request_count(hour_start, hour_end)
            })
            
            trends["average_quality"].append({
                "timestamp": time_point.isoformat(),
                "value": await self._get_avg_quality_score(hour_start, hour_end)
            })
            
            trends["response_time"].append({
                "timestamp": time_point.isoformat(),
                "value": await self._get_avg_response_time(hour_start, hour_end)
            })
            
            trends["cost_per_hour"].append({
                "timestamp": time_point.isoformat(),
                "value": await self._get_hourly_cost(hour_start, hour_end)
            })
            
            trends["error_rate"].append({
                "timestamp": time_point.isoformat(),
                "value": await self._get_error_rate(hour_start, hour_end)
            })
        
        return trends
    
    async def generate_health_report(self) -> Dict:
        """システム健全性レポートの生成"""
        report = {
            "report_id": f"health_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
            "generated_at": datetime.now().isoformat(),
            "executive_summary": await self._generate_executive_summary(),
            "detailed_analysis": await self._generate_detailed_analysis(),
            "recommendations": await self._generate_recommendations(),
            "action_items": await self._generate_action_items()
        }
        
        return report
    
    async def _generate_executive_summary(self) -> Dict:
        """エグゼクティブサマリーの生成"""
        # 過去24時間のKPIサマリー
        kpi_summary = await self._calculate_kpi_summary(
            datetime.now() - timedelta(hours=24),
            datetime.now()
        )
        
        # 全体ステータスの判定
        critical_count = 0
        good_count = 0
        excellent_count = 0
        
        for category in kpi_summary.values():
            for kpi in category.values():
                if kpi["status"] == "critical":
                    critical_count += 1
                elif kpi["status"] == "good":
                    good_count += 1
                elif kpi["status"] == "excellent":
                    excellent_count += 1
        
        if critical_count > 0:
            overall_status = "needs_attention"
            priority = "high"
        elif good_count > excellent_count:
            overall_status = "stable"
            priority = "medium"
        else:
            overall_status = "excellent"
            priority = "low"
        
        return {
            "overall_status": overall_status,
            "priority": priority,
            "critical_issues": critical_count,
            "total_requests_24h": await self._get_request_count_24h(),
            "avg_quality_24h": await self._get_avg_quality_24h(),
            "cost_24h": await self._get_total_cost_24h(),
            "uptime_percentage": await self._get_uptime_percentage()
        }