Cara Mengintegrasikan Deepseek AI API untuk Pengembang: Panduan Lengkap

Base URL: https://api.deepseek.com
Chat Completions: /v1/chat/completions
Embeddings: /v1/embeddings
Fine-tuning: /v1/fine-tuning

import os
import requests
from typing import Dict, List, Optional

class DeepSeekClient:
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv('DEEPSEEK_API_KEY')
        self.base_url = "https://api.deepseek.com"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
    
    def chat_completion(
        self, 
        messages: List[Dict], 
        model: str = "deepseek-chat",
        temperature: float = 0.7,
        max_tokens: int = 1000,
        stream: bool = False
    ) -> Dict:
        """
        Generate chat completion using DeepSeek API
        """
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": stream
        }
        
        response = requests.post(
            f"{self.base_url}/v1/chat/completions",
            headers=self.headers,
            json=payload
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
    
    def enable_deep_think(self, messages: List[Dict]) -> Dict:
        """
        Activate DeepThink mode for complex reasoning
        """
        # Add system prompt to enable reasoning mode
        system_message = {
            "role": "system",
            "content": "Think step by step and show your reasoning process."
        }
        
        enhanced_messages = [system_message] + messages
        
        return self.chat_completion(
            messages=enhanced_messages,
            model="deepseek-reasoner",
            temperature=0.1  # Lower temperature for more consistent reasoning
        )

# Contoh penggunaan
def main():
    client = DeepSeekClient()
    
    # Basic chat completion
    messages = [
        {"role": "user", "content": "Explain quantum computing in simple terms"}
    ]
    
    response = client.chat_completion(messages)
    print("Basic Response:", response['choices'][0]['message']['content'])
    
    # DeepThink mode untuk reasoning kompleks
    complex_question = [
        {
            "role": "user", 
            "content": "If I have 10 apples and give away 30% to my friend, then buy 5 more apples, and my friend gives me back half of what I gave them, how many apples do I have in total?"
        }
    ]
    
    reasoning_response = client.enable_deep_think(complex_question)
    print("\nReasoning Response:", reasoning_response['choices'][0]['message']['content'])

if __name__ == "__main__":
    main()

const axios = require('axios');

class DeepSeekClient {
    constructor(apiKey = process.env.DEEPSEEK_API_KEY) {
        this.apiKey = apiKey;
        this.baseURL = 'https://api.deepseek.com';
        this.headers = {
            'Authorization': `Bearer ${this.apiKey}`,
            'Content-Type': 'application/json'
        };
    }

    async chatCompletion(messages, options = {}) {
        const payload = {
            model: options.model || 'deepseek-chat',
            messages: messages,
            temperature: options.temperature || 0.7,
            max_tokens: options.max_tokens || 1000,
            stream: options.stream || false
        };

        try {
            const response = await axios.post(
                `${this.baseURL}/v1/chat/completions`,
                payload,
                { headers: this.headers }
            );
            return response.data;
        } catch (error) {
            throw new Error(`DeepSeek API Error: ${error.response?.status} - ${error.response?.data}`);
        }
    }

    async analyzeCode(code, language) {
        const messages = [
            {
                role: "system",
                content: `You are an expert code reviewer. Analyze the following ${language} code and provide feedback on optimization, bugs, and best practices.`
            },
            {
                role: "user",
                content: code
            }
        ];

        return await this.chatCompletion(messages, { 
            model: 'deepseek-coder',
            temperature: 0.2 
        });
    }
}

// Contoh penggunaan
async function example() {
    const client = new DeepSeekClient();
    
    // Analisis kode
    const pythonCode = `
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
    
print(fibonacci(10))
    `;
    
    const codeAnalysis = await client.analyzeCode(pythonCode, 'Python');
    console.log('Code Analysis:', codeAnalysis.choices[0].message.content);
}

example().catch(console.error);

import time
import random
from functools import wraps

def retry_with_exponential_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    
                    delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f} seconds...")
                    time.sleep(delay)
            
            return None
        return wrapper
    return decorator

class RobustDeepSeekClient(DeepSeekClient):
    @retry_with_exponential_backoff(max_retries=3)
    def safe_chat_completion(self, messages, **kwargs):
        return self.chat_completion(messages, **kwargs)

import json

def stream_chat_completion(client, messages):
    """
    Handle streaming responses for real-time output
    """
    payload = {
        "model": "deepseek-chat",
        "messages": messages,
        "stream": True
    }
    
    response = requests.post(
        f"{client.base_url}/v1/chat/completions",
        headers=client.headers,
        json=payload,
        stream=True
    )
    
    for line in response.iter_lines():
        if line:
            line = line.decode('utf-8')
            if line.startswith('data: '):
                data = line[6:]  # Remove 'data: ' prefix
                if data.strip() == '[DONE]':
                    break
                try:
                    chunk = json.loads(data)
                    delta = chunk['choices'][0]['delta']
                    if 'content' in delta:
                        yield delta['content']
                except json.JSONDecodeError:
                    continue

class OpenAIToDeepSeekAdapter:
    """
    Adapter untuk memudahkan migrasi dari OpenAI ke DeepSeek
    """
    
    MODEL_MAPPING = {
        'gpt-4': 'deepseek-chat',
        'gpt-4-turbo': 'deepseek-chat',
        'gpt-3.5-turbo': 'deepseek-chat',
        'code-davinci-002': 'deepseek-coder'
    }
    
    def __init__(self, deepseek_client):
        self.client = deepseek_client
    
    def chat_completions_create(self, model, messages, **kwargs):
        # Map OpenAI model names to DeepSeek equivalents
        deepseek_model = self.MODEL_MAPPING.get(model, 'deepseek-chat')
        
        # Convert OpenAI parameters to DeepSeek format
        deepseek_params = self._convert_parameters(kwargs)
        
        return self.client.chat_completion(
            messages=messages,
            model=deepseek_model,
            **deepseek_params
        )
    
    def _convert_parameters(self, openai_params):
        # Convert OpenAI-specific parameters
        converted = {}
        
        param_mapping = {
            'max_tokens': 'max_tokens',
            'temperature': 'temperature',
            'top_p': 'top_p',
            'frequency_penalty': 'frequency_penalty',
            'presence_penalty': 'presence_penalty'
        }
        
        for openai_param, deepseek_param in param_mapping.items():
            if openai_param in openai_params:
                converted[deepseek_param] = openai_params[openai_param]
        
        return converted

# Migrasi mudah
openai_client = OpenAI()  # Client lama
deepseek_client = DeepSeekClient()
adapter = OpenAIToDeepSeekAdapter(deepseek_client)

# Ganti client tanpa mengubah kode banyak
response = adapter.chat_completions_create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)

class EcommerceAIAssistant:
    def __init__(self):
        self.client = DeepSeekClient()
        self.product_db = ProductDatabase()  # Database produk
    
    def customer_service_bot(self, user_query, user_context):
        system_prompt = """
        Anda adalah asisten customer service untuk toko online Indonesia.
        Jawab dengan ramah, informatif, dan gunakan Bahasa Indonesia yang natural.
        Jika ada masalah teknis, berikan solusi step-by-step.
        """
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Context: {user_context}\nPertanyaan: {user_query}"}
        ]
        
        return self.client.chat_completion(messages, temperature=0.3)
    
    def product_recommendation(self, user_history, preferences):
        # Analisis behavior user dan berikan rekomendasi
        prompt = f"""
        Berdasarkan riwayat pembelian: {user_history}
        Dan preferensi: {preferences}
        
        Berikan 5 rekomendasi produk yang paling relevan.
        Format output sebagai JSON dengan struktur:
        {{"recommendations": [
            {{"product_id": "id", "reason": "alasan rekomendasi", "score": 0.95}}
        ]}}
        """
        
        response = self.client.chat_completion([
            {"role": "user", "content": prompt}
        ])
        
        return json.loads(response['choices'][0]['message']['content'])

# Implementasi
assistant = EcommerceAIAssistant()

# Customer service
cs_response = assistant.customer_service_bot(
    "Produk yang saya beli tidak sesuai deskripsi",
    "User ID: 12345, Order ID: ORD789, Product: Sepatu Nike"
)

# Product recommendation  
recommendations = assistant.product_recommendation(
    user_history=["electronics", "fashion", "books"],
    preferences={"brand": "local", "price_range": "medium"}
)

import pandas as pd
import matplotlib.pyplot as plt

class PopulationDataAnalyzer:
    def __init__(self, local_model_path):
        # Deploy DeepSeek model secara lokal
        self.client = LocalDeepSeekClient(local_model_path)
        
    def analyze_demographic_trends(self, data_file):
        # Load data
        df = pd.read_csv(data_file)
        
        # Generate insight menggunakan AI
        summary_prompt = f"""
        Analisis data kependudukan dengan statistik:
        - Total populasi: {len(df)}
        - Usia rata-rata: {df['age'].mean():.1f}
        - Distribusi gender: {df['gender'].value_counts().to_dict()}
        - Top 5 provinsi: {df['province'].value_counts().head().to_dict()}
        
        Berikan analisis trend demografi, insight penting, dan rekomendasi kebijakan.
        Format output dalam bahasa Indonesia yang formal untuk laporan pemerintah.
        """
        
        analysis = self.client.chat_completion([
            {"role": "user", "content": summary_prompt}
        ])
        
        return analysis['choices'][0]['message']['content']
    
    def generate_monthly_report(self, data_file):
        # Analisis komprehensif
        insights = self.analyze_demographic_trends(data_file)
        
        # Generate visualizations
        self.create_charts(data_file)
        
        # Compile report
        report_template = f"""
        # LAPORAN BULANAN ANALISIS KEPENDUDUKAN
        
        ## Executive Summary
        {insights}
        
        ## Rekomendasi Tindak Lanjut
        [Generated by AI analysis]
        
        ## Lampiran
        - Chart 1: Distribusi Usia
        - Chart 2: Sebaran Geografis
        - Chart 3: Trend 12 Bulan Terakhir
        """
        
        return report_template

# Implementation
analyzer = PopulationDataAnalyzer("/path/to/local/deepseek/model")
monthly_report = analyzer.generate_monthly_report("population_data.csv")

class PersonalizedTutor:
    def __init__(self):
        self.client = DeepSeekClient()
        self.learning_tracker = LearningProgressTracker()
    
    def adaptive_explanation(self, concept, student_level, learning_style):
        system_prompt = f"""
        Anda adalah tutor pribadi yang adaptif.
        Level siswa: {student_level}
        Gaya belajar: {learning_style}
        
        Jelaskan konsep dengan cara yang sesuai dengan profil siswa.
        Gunakan analogi, contoh, dan latihan yang relevan.
        """
        
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Tolong jelaskan: {concept}"}
        ]
        
        return self.client.enable_deep_think(messages)
    
    def generate_practice_questions(self, topic, difficulty, count=5):
        prompt = f"""
        Generate {count} practice questions untuk topic: {topic}
        Difficulty level: {difficulty}
        
        Format JSON:
        {{
            "questions": [
                {{
                    "question": "soal",
                    "options": ["A", "B", "C", "D"],
                    "correct_answer": "B",
                    "explanation": "penjelasan"
                }}
            ]
        }}
        """
        
        response = self.client.chat_completion([
            {"role": "user", "content": prompt}
        ], model="deepseek-reasoner")
        
        return json.loads(response['choices'][0]['message']['content'])

# Personalized learning experience
tutor = PersonalizedTutor()

# Adaptive explanation
explanation = tutor.adaptive_explanation(
    concept="Photosynthesis",
    student_level="middle_school",
    learning_style="visual_kinesthetic"
)

# Auto-generated practice
questions = tutor.generate_practice_questions(
    topic="Algebra basics",
    difficulty="beginner",
    count=10
)

# ❌ Prompt yang kurang efektif
bad_prompt = "Buatkan artikel tentang AI"

# ✅ Prompt yang efektif
good_prompt = """
Role: Anda adalah technical writer berpengalaman di bidang AI.

Task: Tulis artikel blog 1000 kata tentang "AI dalam Healthcare"

Context: 
- Target audience: Healthcare professionals
- Tone: Professional tapi accessible
- Focus: Practical applications dan benefits

Structure:
1. Introduction (150 kata)
2. 3 main use cases (500 kata)
3. Challenges dan solutions (200 kata)  
4. Conclusion (150 kata)

Style: Gunakan data konkret, contoh real-world, dan avoid technical jargon yang unnecessary.
"""

class PromptOptimizer:
    @staticmethod
    def create_structured_prompt(role, task, context, examples=None):
        prompt = f"""
        ROLE: {role}
        
        TASK: {task}
        
        CONTEXT: {context}
        """
        
        if examples:
            prompt += f"\n\nEXAMPLES:\n{examples}"
        
        prompt += "\n\nOUTPUT:"
        
        return prompt
    
    @staticmethod
    def few_shot_learning(task, examples):
        prompt = f"Task: {task}\n\n"
        
        for i, example in enumerate(examples, 1):
            prompt += f"Example {i}:\n"
            prompt += f"Input: {example['input']}\n"
            prompt += f"Output: {example['output']}\n\n"
        
        return prompt

def optimize_for_deepseek(prompt, task_type="general"):
    """
    Optimize prompts specifically for DeepSeek models
    """
    optimizations = {
        "reasoning": {
            "prefix": "Think step by step and show your reasoning:",
            "suffix": "\n\nExplain your thinking process.",
            "temperature": 0.1
        },
        "creative": {
            "prefix": "Be creative and imaginative:",
            "suffix": "\n\nMake it engaging and original.",
            "temperature": 0.9
        },
        "analytical": {
            "prefix": "Analyze systematically and provide evidence:",
            "suffix": "\n\nSupport your conclusions with data.",
            "temperature": 0.3
        }
    }
    
    config = optimizations.get(task_type, optimizations["general"])
    
    optimized_prompt = f"{config['prefix']} {prompt} {config['suffix']}"
    
    return optimized_prompt, config['temperature']

import hashlib
import json
from functools import lru_cache
import redis

class DeepSeekCache:
    def __init__(self, redis_client=None):
        self.redis = redis_client or redis.Redis(host='localhost', port=6379, db=0)
        self.local_cache = {}
    
    def get_cache_key(self, messages, model, temperature):
        # Create unique key untuk request
        content = json.dumps({
            "messages": messages,
            "model": model,
            "temperature": temperature
        }, sort_keys=True)
        
        return hashlib.md5(content.encode()).hexdigest()
    
    def get_cached_response(self, cache_key):
        # Try local cache first (fastest)
        if cache_key in self.local_cache:
            return self.local_cache[cache_key]
        
        # Try Redis cache
        cached = self.redis.get(cache_key)
        if cached:
            response = json.loads(cached)
            self.local_cache[cache_key] = response  # Update local cache
            return response
        
        return None
    
    def cache_response(self, cache_key, response, ttl=3600):
        # Cache to both local and Redis
        self.local_cache[cache_key] = response
        self.redis.setex(cache_key, ttl, json.dumps(response))

class CachedDeepSeekClient(DeepSeekClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache = DeepSeekCache()
    
    def chat_completion(self, messages, **kwargs):
        # Generate cache key
        cache_key = self.cache.get_cache_key(
            messages, 
            kwargs.get('model', 'deepseek-chat'),
            kwargs.get('temperature', 0.7)
        )
        
        # Try to get from cache
        cached_response = self.cache.get_cached_response(cache_key)
        if cached_response:
            return cached_response
        
        # If not cached, make API call
        response = super().chat_completion(messages, **kwargs)
        
        # Cache the response
        self.cache.cache_response(cache_key, response)
        
        return response

class BatchProcessor:
    def __init__(self, client, batch_size=10, delay=1.0):
        self.client = client
        self.batch_size = batch_size
        self.delay = delay
    
    async def process_batch(self, requests):
        """
        Process multiple requests in batches untuk avoid rate limiting
        """
        results = []
        
        for i in range(0, len(requests), self.batch_size):
            batch = requests[i:i + self.batch_size]
            batch_results = []
            
            # Process batch concurrently
            tasks = []
            for request in batch:
                task = asyncio.create_task(
                    self.client.chat_completion(**request)
                )
                tasks.append(task)
            
            batch_results = await asyncio.gather(*tasks, return_exceptions=True)
            results.extend(batch_results)
            
            # Rate limiting delay
            if i + self.batch_size < len(requests):
                await asyncio.sleep(self.delay)
        
        return results

# Usage
async def bulk_analysis():
    processor = BatchProcessor(DeepSeekClient())
    
    requests = [
        {"messages": [{"role": "user", "content": f"Analyze text {i}"}]}
        for i in range(100)
    ]
    
    results = await processor.process_batch(requests)
    return results

import time
import logging
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class APIUsageMetrics:
    timestamp: float
    model: str
    input_tokens: int
    output_tokens: int
    response_time: float
    cost: float
    success: bool
    error_message: str = None

class DeepSeekMonitor:
    def __init__(self):
        self.metrics: List[APIUsageMetrics] = []
        self.logger = logging.getLogger(__name__)
    
    def log_request(self, model, input_tokens, output_tokens, 
                   response_time, success, error=None):
        
        cost = self.calculate_cost(model, input_tokens, output_tokens)
        
        metric = APIUsageMetrics(
            timestamp=time.time(),
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            response_time=response_time,
            cost=cost,
            success=success,
            error_message=error
        )
        
        self.metrics.append(metric)
        
        # Log ke file
        self.logger.info(f"API Call - Model: {model}, Cost: ${cost:.4f}, "
                        f"Time: {response_time:.2f}s, Success: {success}")
    
    def calculate_cost(self, model, input_tokens, output_tokens):
        # DeepSeek pricing (as of 2025)
        pricing = {
            "deepseek-chat": {"input": 0.14, "output": 2.19},  # per 1M tokens
            "deepseek-coder": {"input": 0.14, "output": 2.19},
            "deepseek-reasoner": {"input": 0.14, "output": 2.19}
        }
        
        rates = pricing.get(model, pricing["deepseek-chat"])
        
        input_cost = (input_tokens / 1_000_000) * rates["input"]
        output_cost = (output_tokens / 1_000_000) * rates["output"]
        
        return input_cost + output_cost
    
    def get_daily_summary(self):
        today = time.time() - 86400  # Last 24 hours
        recent_metrics = [m for m in self.metrics if m.timestamp > today]
        
        if not recent_metrics:
            return {}
        
        total_cost = sum(m.cost for m in recent_metrics)
        total_requests = len(recent_metrics)
        successful_requests = sum(1 for m in recent_metrics if m.success)
        avg_response_time = sum(m.response_time for m in recent_metrics) / total_requests
        
        return {
            "total_requests": total_requests,
            "successful_requests": successful_requests,
            "success_rate": successful_requests / total_requests,
            "total_cost": total_cost,
            "avg_response_time": avg_response_time,
            "avg_cost_per_request": total_cost / total_requests if total_requests > 0 else 0
        }

class MonitoredDeepSeekClient(DeepSeekClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.monitor = DeepSeekMonitor()
    
    def chat_completion(self, messages, **kwargs):
        start_time = time.time()
        
        try:
            response = super().chat_completion(messages, **kwargs)
            response_time = time.time() - start_time
            
            # Extract token counts from response
            usage = response.get('usage', {})
            input_tokens = usage.get('prompt_tokens', 0)
            output_tokens = usage.get('completion_tokens', 0)
            
            self.monitor.log_request(
                model=kwargs.get('model', 'deepseek-chat'),
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                response_time=response_time,
                success=True
            )
            
            return response
            
        except Exception as e:
            response_time = time.time() - start_time
            
            self.monitor.log_request(
                model=kwargs.get('model', 'deepseek-chat'),
                input_tokens=0,
                output_tokens=0,
                response_time=response_time,
                success=False,
                error=str(e)
            )
            
            raise e

import os
from cryptography.fernet import Fernet

class SecureAPIKeyManager:
    def __init__(self, key_file="api_keys.enc"):
        self.key_file = key_file
        self.cipher_key = self._get_or_create_cipher_key()
        self.fernet = Fernet(self.cipher_key)
    
    def _get_or_create_cipher_key(self):
        key_path = ".cipher_key"
        
        if os.path.exists(key_path):
            with open(key_path, 'rb') as f:
                return f.read()
        else:
            key = Fernet.generate_key()
            with open(key_path, 'wb') as f:
                f.write(key)
            return key
    
    def store_api_key(self, service_name, api_key):
        encrypted_key = self.fernet.encrypt(api_key.encode())
        
        keys = self._load_keys()
        keys[service_name] = encrypted_key.decode()
        self._save_keys(keys)
    
    def get_api_key(self, service_name):
        keys = self._load_keys()
        
        if service_name not in keys:
            raise ValueError(f"API key for {service_name} not found")
        
        encrypted_key = keys[service_name].encode()
        return self.fernet.decrypt(encrypted_key).decode()
    
    def _load_keys(self):
        if not os.path.exists(self.key_file):
            return {}
        
        with open(self.key_file, 'r') as f:
            return json.load(f)
    
    def _save_keys(self, keys):
        with open(self.key_file, 'w') as f:
            json.dump(keys, f)

# Usage
key_manager = SecureAPIKeyManager()
key_manager.store_api_key("deepseek", "your-api-key-here")

# In your application
api_key = key_manager.get_api_key("deepseek")
client = DeepSeekClient(api_key)

import re
from typing import List, Dict

class InputValidator:
    def __init__(self):
        self.max_message_length = 10000
        self.forbidden_patterns = [
            r'<script.*?>.*?</script>',  # XSS prevention
            r'<.*?javascript.*?>',
            r'\b(DROP|DELETE|INSERT|UPDATE)\b',  # SQL injection prevention
            r'system\(\s*["\'].*?["\'].*?\)',  # Command injection
        ]
    
    def validate_messages(self, messages: List[Dict]) -> bool:
        """
        Validate input messages for security dan content policy
        """
        for message in messages:
            if not self.validate_message(message):
                return False
        return True
    
    def validate_message(self, message: Dict) -> bool:
        content = message.get('content', '')
        
        # Check message length
        if len(content) > self.max_message_length:
            raise ValueError(f"Message too long: {len(content)} chars (max: {self.max_message_length})")
        
        # Check for forbidden patterns
        for pattern in self.forbidden_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                raise ValueError(f"Forbidden pattern detected: {pattern}")
        
        return True
    
    def sanitize_input(self, text: str) -> str:
        """
        Sanitize input text
        """
        # Remove potential HTML tags
        text = re.sub(r'<[^>]+>', '', text)
        
        # Remove excessive whitespace
        text = re.sub(r'\s+', ' ', text).strip()
        
        # Truncate if too long
        if len(text) > self.max_message_length:
            text = text[:self.max_message_length] + "... [truncated]"
        
        return text

class SecureDeepSeekClient(DeepSeekClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.validator = InputValidator()
    
    def chat_completion(self, messages, **kwargs):
        # Validate input
        self.validator.validate_messages(messages)
        
        # Sanitize messages
        sanitized_messages = []
        for message in messages:
            sanitized_message = message.copy()
            sanitized_message['content'] = self.validator.sanitize_input(message['content'])
            sanitized_messages.append(sanitized_message)
        
        return super().chat_completion(sanitized_messages, **kwargs)

class GeopoliticalRiskMitigation:
    def __init__(self):
        self.fallback_providers = ['openai', 'anthropic', 'cohere']
        self.current_provider = 'deepseek'
    
    def setup_multi_provider_client(self):
        """
        Setup multiple AI providers untuk fallback
        """
        self.clients = {
            'deepseek': DeepSeekClient(),
            'openai': OpenAIClient(),
            'anthropic': AnthropicClient()
        }
    
    def adaptive_request(self, messages, **kwargs):
        """
        Try primary provider, fallback to alternatives if needed
        """
        for provider in [self.current_provider] + self.fallback_providers:
            try:
                if provider in self.clients:
                    return self.clients[provider].chat_completion(messages, **kwargs)
            except Exception as e:
                logging.warning(f"Provider {provider} failed: {e}")
                continue
        
        raise Exception("All AI providers failed")

class DataSovereigntyManager:
    def __init__(self):
        self.data_classification = {
            'public': ['general_queries', 'educational_content'],
            'internal': ['business_logic', 'customer_data'],
            'confidential': ['financial_data', 'personal_info']
        }
    
    def route_request(self, content, classification):
        """
        Route requests based on data sensitivity
        """
        if classification == 'confidential':
            # Use local on-premise model
            return self.local_deepseek_client.chat_completion(content)
        elif classification == 'internal':
            # Use private cloud deployment
            return self.private_cloud_client.chat_completion(content)
        else:
            # Safe to use public API
            return self.public_api_client.chat_completion(content)

import asyncio
from asyncio import Semaphore

class RateLimitedClient:
    def __init__(self, client, max_requests_per_minute=60):
        self.client = client
        self.semaphore = Semaphore(max_requests_per_minute)
        self.request_times = []
    
    async def rate_limited_request(self, messages, **kwargs):
        async with self.semaphore:
            # Clean old request times
            current_time = time.time()
            self.request_times = [t for t in self.request_times 
                                if current_time - t < 60]
            
            # Check if we need to wait
            if len(self.request_times) >= 60:
                wait_time = 60 - (current_time - self.request_times[0])
                if wait_time > 0:
                    await asyncio.sleep(wait_time)
            
            # Make request
            self.request_times.append(current_time)
            return await self.client.chat_completion(messages, **kwargs)

class ConsistencyManager:
    def __init__(self, client):
        self.client = client
        self.consistency_cache = {}
    
    def get_consistent_response(self, messages, consistency_key=None, 
                              min_consensus=3):
        """
        Get multiple responses dan return consensus
        """
        if consistency_key and consistency_key in self.consistency_cache:
            return self.consistency_cache[consistency_key]
        
        responses = []
        for _ in range(min_consensus):
            response = self.client.chat_completion(messages, temperature=0.1)
            responses.append(response['choices'][0]['message']['content'])
        
        # Simple consensus: return most common response
        from collections import Counter
        consensus = Counter(responses).most_common(1)[0][0]
        
        if consistency_key:
            self.consistency_cache[consistency_key] = consensus
        
        return consensus

class QualityValidator:
    def __init__(self):
        self.quality_metrics = {
            'min_length': 50,
            'max_length': 5000,
            'forbidden_phrases': ['I cannot', 'As an AI', 'I apologize'],
            'required_elements': []
        }
    
    def validate_response(self, response, task_type="general"):
        """
        Validate response quality based on task type
        """
        content = response['choices'][0]['message']['content']
        
        issues = []
        
        # Length check
        if len(content) < self.quality_metrics['min_length']:
            issues.append("Response too short")
        
        if len(content) > self.quality_metrics['max_length']:
            issues.append("Response too long")
        
        # Content quality checks
        for phrase in self.quality_metrics['forbidden_phrases']:
            if phrase.lower() in content.lower():
                issues.append(f"Contains problematic phrase: {phrase}")
        
        # Task-specific validation
        if task_type == "code":
            if "```" not in content:
                issues.append("Code response should contain code blocks")
        
        elif task_type == "analysis":
            if not any(word in content.lower() for word in ['analysis', 'conclusion', 'findings']):
                issues.append("Analysis response lacks analytical elements")
        
        return {
            'is_valid': len(issues) == 0,
            'issues': issues,
            'quality_score': max(0, 1 - len(issues) * 0.2)
        }

class QualityAssuredClient(DeepSeekClient):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.validator = QualityValidator()
        self.max_retries = 3
    
    def chat_completion_with_quality_check(self, messages, task_type="general", **kwargs):
        for attempt in range(self.max_retries):
            response = super().chat_completion(messages, **kwargs)
            
            quality_check = self.validator.validate_response(response, task_type)
            
            if quality_check['is_valid']:
                return response
            
            # If quality is poor, adjust prompt and retry
            if attempt < self.max_retries - 1:
                # Add quality instruction to prompt
                enhanced_messages = messages.copy()
                enhanced_messages.append({
                    "role": "system",
                    "content": f"Previous response had issues: {', '.join(quality_check['issues'])}. Please provide a better response."
                })
                messages = enhanced_messages
        
        # Return best attempt even if not perfect
        return response