Tutorial Lengkap Replicate: Menjalankan dan Deploy Model ML via API

Replicate adalah platform cloud yang memungkinkan Anda menjalankan model machine learning melalui API tanpa perlu mengelola infrastruktur GPU sendiri. Dengan Replicate, Anda bisa mengakses ribuan model open-source mulai dari image generation, LLM, speech-to-text, hingga video generation hanya dengan beberapa baris kode.

Platform ini sangat populer di kalangan developer karena kemudahannya: tidak perlu setup server, tidak perlu konfigurasi CUDA, dan model langsung siap digunakan melalui REST API atau Python SDK. Replicate juga menyediakan fitur untuk mendeploy model custom Anda sendiri menggunakan tool bernama Cog.

Dalam tutorial ini, kita akan mempelajari cara menggunakan Replicate mulai dari setup awal, menjalankan prediksi, streaming response, menggunakan webhooks, fine-tuning model, hingga packaging model custom dengan Cog.

Instalasi dan Setup

Membuat Akun dan API Token

Langkah pertama adalah membuat akun di Replicate. Setelah mendaftar, Anda bisa mendapatkan API token dari halaman Account Settings.

# Set API token sebagai environment variable
export REPLICATEAPITOKEN="r8yourapitokenhere"

Instalasi Python SDK

# Instalasi menggunakan pip
pip install replicate

Atau menggunakan uv (recommended)
uv pip install replicate

Verifikasi Instalasi

import replicate
import os

Pastikan token sudah di-set
assert os.environ.get("REPLICATEAPITOKEN"), "Token belum di-set!"

Test koneksi dengan menjalankan model sederhana
output = replicate.run(
    "meta/meta-llama-3.1-8b-instruct",
    input={"prompt": "Hello, world!"}
)
print("".join(output))

Menjalankan Prediksi (Basic Usage)

Text Generation dengan LLM

Cara paling sederhana untuk menjalankan model di Replicate adalah menggunakan fungsi replicate.run().

import replicate

Menjalankan Llama 3.1 untuk text generation
output = replicate.run(
    "meta/meta-llama-3.1-70b-instruct",
    input={
        "prompt": "Jelaskan konsep machine learning dalam 3 paragraf",
        "maxtokens": 512,

        "temperature": 0.7,
        "topp": 0.9,
        "systemprompt": "Kamu adalah asisten AI yang membantu menjelaskan konsep teknologi dalam bahasa Indonesia."

    }
)

Output adalah iterator, gabungkan menjadi string
result = "".join(output)
print(result)

import replicate

Generate gambar menggunakan SDXL
output = replicate.run(
    "stability-ai/sdxl:39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    input={
        "prompt": "A futuristic city skyline at sunset, cyberpunk style, highly detailed",
        "negativeprompt": "blurry, low quality, distorted",
        "width": 1024,
        "height": 1024,
        "numoutputs": 1,

        "scheduler": "KEULER",
        "numinferencesteps": 30,
        "guidancescale": 7.5

    }
)

Output berupa list URL gambar
for i, url in enumerate(output):
    print(f"Image {i+1}: {url}")

import replicate

Menggunakan LLaVA untuk image understanding
output = replicate.run(
    "yorickvp/llava-v1.6-34b:41ecfbfb261e6c1adf3ad896c9066ca98346996d7c4045c5bc944a79d430f174",
    input={
        "image": "https://example.com/photo.jpg",
        "prompt": "Describe this image in detail"
    }
)

result = "".join(output)
print(result)

import replicate

Transkripsi audio menggunakan Whisper
output = replicate.run(
    "openai/whisper:4d50797a6f35677e3f7e36c5b0d0c3c50e8a0b0e4e0e0e0e0e0e0e0e0e0e0e",
    input={
        "audio": open("recording.mp3", "rb"),
        "model": "large-v3",
        "language": "id",
        "translate": False
    }
)

print(output["transcription"])

import replicate

Membuat prediksi tanpa menunggu hasilnya
prediction = replicate.predictions.create(
    model="stability-ai/sdxl",
    input={
        "prompt": "A beautiful landscape painting in the style of Monet",
        "width": 1024,
        "height": 1024
    }
)

print(f"Prediction ID: {prediction.id}")
print(f"Status: {prediction.status}")

import replicate
import time

Buat prediksi
prediction = replicate.predictions.create(
    model="stability-ai/sdxl",
    input={"prompt": "A cat wearing a hat"}
)

Polling status
while prediction.status not in ["succeeded", "failed", "canceled"]:
    time.sleep(2)
    prediction.reload()
    print(f"Status: {prediction.status}")

if prediction.status == "succeeded":
    print(f"Output: {prediction.output}")
else:
    print(f"Error: {prediction.error}")

import replicate

prediction = replicate.predictions.create(
    model="meta/meta-llama-3.1-70b-instruct",
    input={"prompt": "Write a very long essay..."}
)

Cancel jika tidak diperlukan lagi
replicate.predictions.cancel(prediction.id)
print(f"Status: {prediction.status}")

import replicate

Stream output token by token
for event in replicate.stream(
    "meta/meta-llama-3.1-8b-instruct",
    input={
        "prompt": "Tulis cerita pendek tentang robot yang belajar memasak",
        "maxtokens": 500
    }
):
    print(str(event), end="", flush=True)

print()  # New line di akhir

Streaming dengan Server-Sent Events

import replicate

Menggunakan event types untuk kontrol lebih detail
for event in replicate.stream(
    "meta/meta-llama-3.1-70b-instruct",
    input={
        "prompt": "Explain quantum computing",
        "maxtokens": 300

    }
):
    if event.event == "output":
        print(str(event), end="", flush=True)
    elif event.event == "done":
        print("\n--- Stream selesai ---")

import replicate

Buat prediksi dengan webhook
prediction = replicate.predictions.create(
    model="stability-ai/sdxl",
    input={
        "prompt": "A photo-realistic portrait of a cat astronaut"
    },
    webhook="https://your-server.com/api/replicate-webhook",
    webhookeventsfilter=["completed"]

)

print(f"Prediction {prediction.id} dibuat, webhook akan dipanggil saat selesai")

from fastapi import FastAPI, Request
import json

app = FastAPI()

@app.post("/api/replicate-webhook")
async def handlewebhook(request: Request):
    payload = await request.json()

    predictionid = payload["id"]

    status = payload["status"]

    if status == "succeeded":
        output = payload["output"]
        print(f"Prediction {predictionid} selesai!")
        print(f"Output: {output}")
        # Proses output (simpan ke database, kirim ke user, dll)
    elif status == "failed":
        error = payload["error"]
        print(f"Prediction {predictionid} gagal: {error}")


    return {"status": "ok"}

const express = require('express');
const app = express();

app.use(express.json());

app.post('/api/replicate-webhook', (req, res) => {
  const { id, status, output, error } = req.body;

  if (status === 'succeeded') {
    console.log(Prediction ${id} selesai:, output);
    // Proses output
  } else if (status === 'failed') {
    console.error(Prediction ${id} gagal:, error);
  }

  res.json({ status: 'ok' });
});

app.listen(3000);

import replicate

Menggunakan file lokal sebagai input
with open("inputimage.jpg", "rb") as f:
    output = replicate.run(
        "sczhou/codeformer:7de2ea26c616d5bf2245ad0d5e24f0ff9a6204578a5c876db53142edd9d2cd56",
        input={
            "image": f,
            "fidelity": 0.7,
            "upscale": 2
        }
    )

print(f"Enhanced image URL: {output}")

Download Output

import replicate
import httpx

Generate gambar
output = replicate.run(
    "stability-ai/sdxl",
    input={"prompt": "A serene mountain lake at dawn"}
)

Download hasil
for i, url in enumerate(output):
    response = httpx.get(url)
    with open(f"output{i}.png", "wb") as f:

        f.write(response.content)
    print(f"Saved output{i}.png")

Menggunakan REST API Langsung

Selain Python SDK, Anda bisa menggunakan REST API secara langsung dengan curl atau HTTP client apapun.

Membuat Prediksi via curl

curl -s -X POST "https://api.replicate.com/v1/predictions" \
  -H "Authorization: Bearer $REPLICATEAPITOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "version": "39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    "input": {
      "prompt": "A beautiful sunset over the ocean"
    }
  }'

Menggunakan Official Model

# Untuk official models, gunakan endpoint model langsung
curl -s -X POST "https://api.replicate.com/v1/models/meta/meta-llama-3.1-8b-instruct/predictions" \
  -H "Authorization: Bearer $REPLICATEAPITOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "prompt": "What is machine learning?"
    }
  }'

Mengecek Status via curl

curl -s "https://api.replicate.com/v1/predictions/PREDICTIONID" \
  -H "Authorization: Bearer $REPLICATEAPITOKEN"

# macOS
brew install cog

Linux
sudo curl -o /usr/local/bin/cog -L "https://github.com/replicate/cog/releases/latest/download/cog$(uname -s)$(uname -m)"

sudo chmod +x /usr/local/bin/cog

my-model/
  cog.yaml        # Konfigurasi environment
  predict.py      # Kode prediksi
  weights/         # Model weights (opsional)

build:
  pythonversion: "3.11"
  pythonpackages:

"torch==2.1.0"
"torchvision==0.16.0"
"transformers==4.36.0"
"Pillow==10.1.0"
  gpu: true

  cuda: "12.1"
  systempackages:
"libgl1-mesa-glx"

predict: "predict.py:Predictor"

Menulis Predictor

import torch
from cog import BasePredictor, Input, Path
from transformers import AutoModelForCausalLM, AutoTokenizer

class Predictor(BasePredictor):
    def setup(self):
        """Load model ke memori saat container start."""
        self.modelname = "microsoft/DialoGPT-medium"

        self.tokenizer = AutoTokenizer.frompretrained(self.modelname)

        self.model = AutoModelForCausalLM.frompretrained(self.modelname)

        self.model.eval()

    def predict(
        self,
        prompt: str = Input(description="Input text prompt"),
        maxlength: int = Input(description="Maximum response length", default=100, ge=1, le=500),
        temperature: float = Input(description="Sampling temperature", default=0.7, ge=0.1, le=2.0),
    ) -> str:
        """Jalankan prediksi pada input."""
        inputids = self.tokenizer.encode(prompt + self.tokenizer.eostoken, returntensors="pt")


        with torch.nograd():
            output = self.model.generate(
                inputids,

                maxlength=maxlength,

                temperature=temperature,
                dosample=True,
                topp=0.9,

                padtokenid=self.tokenizer.eostokenid

            )

        response = self.tokenizer.decode(output[:, inputids.shape[-1]:][0], skipspecialtokens=True)
        return response

Predictor dengan Image Output

from cog import BasePredictor, Input, Path
from diffusers import StableDiffusionPipeline
import torch

class Predictor(BasePredictor):
    def setup(self):
        self.pipe = StableDiffusionPipeline.frompretrained(

            "runwayml/stable-diffusion-v1-5",
            torchdtype=torch.float16
        ).to("cuda")

    def predict(
        self,
        prompt: str = Input(description="Text prompt for image generation"),
        numinferencesteps: int = Input(description="Number of denoising steps", default=30, ge=1, le=100),
        guidancescale: float = Input(description="Guidance scale", default=7.5, ge=1.0, le=20.0),

    ) -> Path:
        """Generate image dari text prompt."""
        image = self.pipe(
            prompt,
            numinferencesteps=numinferencesteps,

            guidancescale=guidancescale

        ).images[0]

        outputpath = Path("/tmp/output.png")
        image.save(outputpath)

        return outputpath

Test Lokal dan Push

# Test prediksi secara lokal
cog predict -i prompt="Hello world"

Build Docker image
cog build -t my-model

Push ke Replicate
cog push r8.im/username/my-model

Fine-tuning Model

Replicate mendukung fine-tuning untuk beberapa model populer.

Fine-tuning SDXL

import replicate

Buat training
training = replicate.trainings.create(
    model="stability-ai/sdxl",
    version="39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    input={
        "inputimages": "https://example.com/training-images.zip",

        "tokenstring": "TOK",
        "captionprefix": "a photo of TOK, ",

        "maxtrainsteps": 1000,

        "usefacedetectioninstead": False,
        "learningrate": 1e-4

    },
    destination="username/my-custom-sdxl"
)

print(f"Training ID: {training.id}")
print(f"Status: {training.status}")

import replicate
import time

training = replicate.trainings.get("TRAININGID")

while training.status not in ["succeeded", "failed", "canceled"]:
    time.sleep(30)
    training.reload()
    print(f"Status: {training.status}")
    if training.logs:
        print(f"Logs: {training.logs[-200:]}")

if training.status == "succeeded":
    print(f"Model berhasil di-fine-tune!")
    print(f"Version: {training.output['version']}")

Menggunakan Model yang Sudah Di-fine-tune

import replicate

Gunakan model custom hasil fine-tuning
output = replicate.run(
    "username/my-custom-sdxl:versionid",

    input={
        "prompt": "A photo of TOK in a beautiful garden",
        "numoutputs": 4
    }
)

for url in output:
    print(url)

Deployment dan Model Management

Membuat Deployment

import replicate

Buat deployment dengan hardware khusus
deployment = replicate.deployments.create(
    name="my-llm-deployment",
    model="meta/meta-llama-3.1-8b-instruct",
    hardware="gpu-a40-large",
    mininstances=1,

    maxinstances=5
)

print(f"Deployment created: {deployment.name}")

Menjalankan Prediksi pada Deployment

import replicate

Jalankan prediksi pada deployment spesifik
deployment = replicate.deployments.get("username/my-llm-deployment")

prediction = deployment.predictions.create(
    input={
        "prompt": "Explain the theory of relativity",
        "maxtokens": 256

    }
)

prediction.wait()
print(prediction.output)

import replicate

Cari model berdasarkan keyword
models = replicate.models.search("text-to-image")
for model in models:
    print(f"{model.owner}/{model.name}: {model.description}")

from langchaincommunity.llms import Replicate

llm = Replicate(
    model="meta/meta-llama-3.1-70b-instruct",
    modelkwargs={

        "temperature": 0.7,
        "maxtokens": 500,
        "topp": 0.9

    }
)

response = llm.invoke("Apa itu neural network?")
print(response)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import replicate

app = FastAPI()

@app.post("/generate")
async def generatetext(prompt: str):
    output = replicate.run(
        "meta/meta-llama-3.1-8b-instruct",
        input={"prompt": prompt, "maxtokens": 500}

    )
    return {"response": "".join(output)}

@app.post("/generate/stream")
async def generatestream(prompt: str):
    def stream():
        for event in replicate.stream(
            "meta/meta-llama-3.1-8b-instruct",
            input={"prompt": prompt, "maxtokens": 500}

        ):
            yield str(event)

    return StreamingResponse(stream(), mediatype="text/plain")

Best Practices

1. Error Handling yang Tepat

import replicate
from replicate.exceptions import ReplicateError, ModelError

try:
    output = replicate.run(
        "stability-ai/sdxl",
        input={"prompt": "A beautiful landscape"}
    )
except ModelError as e:
    print(f"Model error: {e}")
except ReplicateError as e:
    print(f"API error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

2. Menggunakan Versi Model yang Spesifik

# Selalu pin versi model untuk production
output = replicate.run(
    "stability-ai/sdxl:39ed52f2a78e934b3ba6e2a89f5b1c712de7dfea535525255b1aa35c5565e08b",
    input={"prompt": "test"}
)

3. Batch Processing

import replicate
import asyncio

async def batchpredict(prompts):

    predictions = []
    for prompt in prompts:
        prediction = replicate.predictions.create(
            model="meta/meta-llama-3.1-8b-instruct",
            input={"prompt": prompt, "maxtokens": 200}
        )
        predictions.append(prediction)

    results = []
    for prediction in predictions:
        prediction.wait()
        results.append("".join(prediction.output))

    return results

prompts = [
    "Jelaskan apa itu Python",
    "Jelaskan apa itu JavaScript",
    "Jelaskan apa itu Rust"
]

results = asyncio.run(batchpredict(prompts))

for prompt, result in zip(prompts, results):
    print(f"Q: {prompt}")
    print(f"A: {result}\n")

import replicate

Cek hardware requirements dan pricing sebelum menjalankan model
model = replicate.models.get("meta/meta-llama-3.1-70b-instruct")
latestversion = model.latestversion


print(f"Model: {model.owner}/{model.name}")
print(f"Description: {model.description}")

List semua prediksi untuk monitoring cost
predictions = replicate.predictions.list()
for p in predictions:
    if p.metrics and "predicttime" in p.metrics:
        print(f"ID: {p.id}, Time: {p.metrics['predicttime']:.2f}s, Status: {p.status}")

import replicate
import time

def runwithretry(model, inputdata, maxretries=3, timeout=300):

    for attempt in range(maxretries):
        try:
            prediction = replicate.predictions.create(
                model=model,
                input=inputdata

            )

            starttime = time.time()
            while prediction.status not in ["succeeded", "failed", "canceled"]:
                if time.time() - starttime > timeout:

                    replicate.predictions.cancel(prediction.id)
                    raise TimeoutError(f"Prediction timed out after {timeout}s")
                time.sleep(2)
                prediction.reload()

            if prediction.status == "succeeded":
                return prediction.output

            raise Exception(f"Prediction failed: {prediction.error}")

        except TimeoutError:
            if attempt < maxretries - 1:
                print(f"Attempt {attempt + 1} timed out, retrying...")
                continue
            raise

    raise Exception(f"Failed after {max_retries} attempts")

Kesimpulan

Replicate menyederhanakan proses menjalankan dan mendeploy model machine learning dengan menyediakan API yang mudah digunakan dan infrastruktur GPU yang terkelola. Berikut ringkasan poin-poin penting:

• Kemudahan Akses: Jalankan ribuan model open-source tanpa setup infrastruktur melalui Python SDK atau REST API.
• Streaming: Gunakan replicate.stream() untuk mendapatkan output LLM secara real-time, memberikan pengalaman pengguna yang lebih baik.
• Webhooks: Untuk task berdurasi panjang, gunakan webhooks agar server Anda mendapat notifikasi otomatis saat prediksi selesai.
• Cog: Package model custom Anda ke dalam container yang siap deploy ke Replicate menggunakan Cog.
• Fine-tuning: Manfaatkan fitur fine-tuning untuk mengadaptasi model seperti SDXL dengan dataset custom Anda.
• Deployment: Buat deployment khusus dengan hardware dan scaling yang bisa dikonfigurasi sesuai kebutuhan.
• Integrasi: Replicate terintegrasi dengan baik dengan framework seperti LangChain dan FastAPI.

Dengan memahami fitur-fitur ini, Anda bisa membangun aplikasi AI yang powerful tanpa harus mengelola kompleksitas infrastruktur GPU secara langsung.