Panduan Membangun AI Handle Terintegrasi DeepSeek, Gemini, dan Provider Lain

JAVASCORNER.CO.ID, JAKARTA - Dalam pengembangan aplikasi AI modern, ketergantungan pada satu penyedia layanan LLM (Large Language Model) adalah risiko besar. Bayangkan aplikasi chatbot customer service Anda tiba-tiba down karena API OpenAI mengalami gangguan, atau biaya melonjak karena semua request diproses oleh model premium. Inilah mengapa arsitektur Multi-Provider yang disebut "AI Handle" menjadi krusial. Artikel ini akan memandu Anda membangun Universal AI Handler yang terintegrasi dengan DeepSeek, Google Gemini, OpenAI, dan Anthropic, menggunakan pendekatan LLM Gateway dengan Circuit Breaker dan Fallback Mechanism.

• Arsitektur AI Handle yang Benar
• Perbandingan Tools Pendukung
• Persiapan Environment
• Implementasi Kode: Universal AI Handle
• Penjelasan Mendalam Kode
• Testing dan Simulasi
• Tabel Perbandingan Strategi Handle
• Rekomendasi Production Grade

Arsitektur AI Handle yang Benar

Sebelum masuk ke kode, pahami tiga layer penting dalam arsitektur ini yang membedakannya dengan sekadar "panggil API biasa":

Dalam tutorial ini, kita akan membangun Gateway Layer sendiri menggunakan Python. Handler ini cerdas: jika DeepSeek sibuk (rate limit), otomatis pindah ke Gemini; jika Gemini lambat, pindah ke OpenAI.

Perbandingan Tools Pendukung

Sebelum membuat kode, kenali tools yang bisa mempermudah hidup Anda:

Persiapan Environment

Kita akan menggunakan Python dan FastAPI. Siapkan API Keys untuk DeepSeek, Google Gemini, dan OpenAI.

pip install fastapi uvicorn openai google-generativeai tenacity httpx

Buat file .env:

# DeepSeek (Murah untuk Reasoning)
DEEPSEEK_API_KEY=sk-...
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

# Google Gemini
GEMINI_API_KEY=AI...
GEMINI_MODEL=gemini-1.5-flash

# OpenAI (Cadangan)
OPENAI_API_KEY=sk-...

Implementasi Kode: Universal AI Handle

Berikut adalah implementasi AI Handler dengan kemampuan:

1. Fallback Otomatis (Provider A mati -> Provider B)
2. Retry dengan Exponential Backoff (Coba ulang jika gagal)
3. Circuit Breaker Sederhana (Hentikan sementara provider yang error terus)

import os
import asyncio
from typing import List, Dict, Any
from datetime import datetime, timedelta
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import AsyncOpenAI
import google.generativeai as genai
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# -------------------- KONFIGURASI --------------------
app = FastAPI(title="Universal AI Gateway")

# Inisialisasi Client
deepseek_client = AsyncOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)
openai_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))

# -------------------- CIRCUIT BREAKER SEDERHANA --------------------
class CircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED" # CLOSED = Normal, OPEN = Jangan dipakai

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if datetime.now() > self.last_failure_time + timedelta(seconds=self.recovery_timeout):
                self.state = "HALF_OPEN"
                print(f"[Circuit Breaker] Mencoba memulihkan...")
            else:
                raise Exception(f"Circuit Breaker OPEN untuk provider. Coba lagi nanti.")

        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failure_count = 0
                print(f"[Circuit Breaker] Pulih! Kembali ke CLOSED.")
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = datetime.now()
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                print(f"[Circuit Breaker] Gagal {self.failure_count} kali. Beralih ke OPEN.")
            raise e

# -------------------- LOGIC HANDLER --------------------
class AIHandler:
    def __init__(self):
        self.providers = {
            "deepseek": {
                "client": deepseek_client,
                "model": "deepseek-chat",
                "priority": 1, # Prioritas utama
                "breaker": CircuitBreaker()
            },
            "gemini": {
                "client": None, # Gemini punya cara panggil beda
                "model": os.getenv("GEMINI_MODEL", "gemini-1.5-flash"),
                "priority": 2,
                "breaker": CircuitBreaker()
            },
            "openai": {
                "client": openai_client,
                "model": "gpt-3.5-turbo", # Murah untuk fallback
                "priority": 3,
                "breaker": CircuitBreaker()
            }
        }

    async def _call_gemini(self, model: str, messages: List[Dict]):
        """Wrapper khusus karena Gemini pakai library sendiri"""
        gen_model = genai.GenerativeModel(model)
        # Gabungkan pesan sistem dan user
        prompt = messages[-1]["content"] 
        response = await gen_model.generate_content_async(prompt)
        return response.text

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(Exception)
    )
    async def _execute_provider(self, provider_name: str, config: Dict, messages: List[Dict]):
        """Eksekusi ke provider tertentu dengan retry logic"""
        print(f"[Attempt] Mencoba provider: {provider_name}")
        
        if provider_name == "gemini":
            response_text = await self._call_gemini(config["model"], messages)
            return {"content": response_text, "provider": provider_name}
        else:
            client = config["client"]
            response = await client.chat.completions.create(
                model=config["model"],
                messages=messages,
                max_tokens=1000
            )
            return {"content": response.choices[0].message.content, "provider": provider_name}

    async def chat(self, messages: List[Dict]) -> Dict[str, Any]:
        """
        Logic Utama: Coba prioritas 1, gagal? pindah ke prioritas 2.
        Dilindungi oleh Circuit Breaker.
        """
        # Urutkan berdasarkan priority
        sorted_providers = sorted(self.providers.items(), key=lambda x: x[1]["priority"])
        
        last_exception = None
        for name, config in sorted_providers:
            try:
                # Circuit Breaker akan mengecek apakah provider ini sedang "Open"
                result = await config["breaker"].call(
                    self._execute_provider, name, config, messages
                )
                return result
            except Exception as e:
                last_exception = e
                print(f"[Failover] Provider {name} gagal: {str(e)}. Pindah ke berikutnya...")
                continue
        
        # Jika semua provider gagal
        raise Exception(f"Semua provider gagal. Error terakhir: {str(last_exception)}")

# -------------------- API ENDPOINT --------------------
class ChatRequest(BaseModel):
    message: str

@app.post("/v1/chat/completions")
async def chat_completion(request: ChatRequest):
    handler = AIHandler()
    messages = [{"role": "user", "content": request.message}]
    
    try:
        result = await handler.chat(messages)
        return {
            "status": "success",
            "data": result["content"],
            "served_by": result["provider"],
            "timestamp": datetime.now().isoformat()
        }
    except Exception as e:
        raise HTTPException(status_code=503, detail=str(e))

@app.get("/health")
async def health_check():
    return {"status": "Gateway Active", "strategy": "Multi-Provider with Fallback"}

# Jalankan dengan: uvicorn main:app --reload

Penjelasan Mendalam Kode

1. Circuit Breaker

Dalam kode di atas, jika DeepSeek gagal 3 kali berturut-turut, statusnya menjadi OPEN. Selama 60 detik, handler tidak akan mencoba DeepSeek lagi, langsung melompat ke Gemini. Ini mencegah "Retry Storm" yang bisa memperparah error.

2. Exponential Backoff

Fungsi @retry akan mencoba ulang request yang gagal (misal karena network error) dengan jeda: 2 detik, lalu 4 detik, lalu 10 detik. Ini lebih sopan terhadap server provider daripada meng-spam request setiap detik.

3. Prioritas Berdasarkan Biaya (Cost-Aware)

Dalam contoh, saya set priority: 1 untuk DeepSeek. Mengapa? Karena DeepSeek biasanya jauh lebih murah daripada OpenAI atau Gemini untuk kualitas reasoning yang setara. Dengan arsitektur ini, Anda bisa menghemat biaya hingga 70% dengan mengarahkan traffic ke model murah terlebih dahulu.

Testing dan Simulasi

Coba jalankan server:

uvicorn main:app --reload

Lakukan request:

curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"message": "Jelaskan tentang black hole dalam 2 kalimat"}'

Output yang diharapkan:

{
  "status": "success",
  "data": "Lubang hitam adalah region...",
  "served_by": "deepseek",
  "timestamp": "2024-..."
}

Skenario Gagal: Matikan internet, atau masukkan API Key salah. Maka output akan berubah menjadi:

{
  "status": "success",
  "data": "...",
  "served_by": "gemini", // Otomatis pindah!
  "timestamp": "..."
}

Tabel Perbandingan Strategi Handle

Rekomendasi Production Grade

Untuk skala produksi besar, kode di atas bisa dikembangkan dengan:

1. Menggunakan LiteLLM Proxy: Daripada bikin dari nol, gunakan LiteLLM yang sudah mendukung 100+ provider dan memiliki fitur load balancing.
2. Observability: Integrasikan dengan Langfuse atau Helicone untuk melihat berapa banyak request yang gagal di DeepSeek dan berapa yang di-fallback ke Gemini.
3. Semantic Caching: Cache tidak berdasarkan teks exact match, tapi berdasarkan kemiripan vektor (vector similarity). Jika user nanya "Halo" dan "Hai", jawabannya bisa sama.

Membangun AI Handle yang benar bukan hanya tentang menggabungkan beberapa API Key. Ini tentang mendesain resiliensi. Dengan menggabungkan prioritas biaya (DeepSeek), fallback otomatis (Gemini), dan circuit breaker, Anda memastikan aplikasi AI Anda tetap hidup, murah, dan cepat. Gunakan kode di atas sebagai fondasi, lalu sesuaikan dengan kebutuhan bisnis Anda. Jangan sampai aplikasi Anda mati hanya karena satu provider sedang maintenance.