Complete Replicate Tutorial: Run and Deploy ML Models via API

Replicate is a cloud platform that allows you to run machine learning models through an API without managing GPU infrastructure yourself. With Replicate, you can access thousands of open-source models ranging from image generation, LLMs, speech-to-text, to video generation with just a few lines of code.

The platform is extremely popular among developers for its simplicity: no server setup required, no CUDA configuration needed, and models are ready to use through REST API or Python SDK. Replicate also provides tools to deploy your custom models using an open-source tool called Cog.

In this tutorial, we will learn how to use Replicate from initial setup, running predictions, streaming responses, using webhooks, fine-tuning models, to packaging custom models with Cog.

Installation and Setup

Creating an Account and API Token

The first step is to create an account on Replicate. After signing up, you can obtain your API token from the Account Settings page.

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

Installing the Python SDK

# Install using pip
pip install replicate

Or using uv (recommended)
uv pip install replicate

Verifying Installation

import replicate
import os

Make sure token is set
assert os.environ.get("REPLICATEAPITOKEN"), "Token not set!"

Test connection by running a simple model
output = replicate.run(
    "meta/meta-llama-3.1-8b-instruct",
    input={"prompt": "Hello, world!"}
)
print("".join(output))

Running Predictions (Basic Usage)

Text Generation with LLMs

The simplest way to run a model on Replicate is using the replicate.run() function.

import replicate

Run Llama 3.1 for text generation
output = replicate.run(
    "meta/meta-llama-3.1-70b-instruct",
    input={
        "prompt": "Explain the concept of machine learning in 3 paragraphs",
        "maxtokens": 512,

        "temperature": 0.7,
        "topp": 0.9,
        "systemprompt": "You are a helpful AI assistant that explains technology concepts clearly."

    }
)

Output is an iterator, join into a string
result = "".join(output)
print(result)

import replicate

Generate images using 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 is a list of image URLs
for i, url in enumerate(output):
    print(f"Image {i+1}: {url}")

import replicate

Use LLaVA for 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

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

print(output["transcription"])

import replicate

Create prediction without waiting for results
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

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

Poll for 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 if no longer needed
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": "Write a short story about a robot learning to cook",
        "maxtokens": 500
    }
):
    print(str(event), end="", flush=True)

print()  # New line at the end

Streaming with Server-Sent Events

import replicate

Use event types for more detailed control
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 complete ---")

import replicate

Create prediction with 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} created, webhook will be called upon completion")

from fastapi import FastAPI, Request

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} completed!")
        print(f"Output: {output}")
        # Process output (save to database, send to user, etc.)
    elif status == "failed":
        error = payload["error"]
        print(f"Prediction {predictionid} failed: {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} completed:, output);
    // Process output
  } else if (status === 'failed') {
    console.error(Prediction ${id} failed:, error);
  }

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

app.listen(3000);

import replicate

Use local file as 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}")

Downloading Output

import replicate
import httpx

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

Download results
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")

Using the REST API Directly

In addition to the Python SDK, you can use the REST API directly with curl or any HTTP client.

Creating Predictions 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"
    }
  }'

Using Official Models

# For official models, use the model endpoint directly
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?"
    }
  }'

Checking 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        # Environment configuration
  predict.py      # Prediction code
  weights/         # Model weights (optional)

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"

Writing a Predictor

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

class Predictor(BasePredictor):
    def setup(self):
        """Load model into memory when container starts."""
        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:
        """Run prediction on 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 with 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 from text prompt."""
        image = self.pipe(
            prompt,
            numinferencesteps=numinferencesteps,

            guidancescale=guidancescale

        ).images[0]

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

        return outputpath

Testing Locally and Pushing

# Test prediction locally
cog predict -i prompt="Hello world"

Build Docker image
cog build -t my-model

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

Fine-tuning Models

Replicate supports fine-tuning for several popular models.

Fine-tuning SDXL

import replicate

Create 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 fine-tuned successfully!")
    print(f"Version: {training.output['version']}")

Using a Fine-tuned Model

import replicate

Use custom fine-tuned model
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)

Deployments and Model Management

Creating a Deployment

import replicate

Create deployment with dedicated hardware
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}")

Running Predictions on a Deployment

import replicate

Run prediction on a specific deployment
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

Search models by 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("What is a 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. Proper Error Handling

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. Pin Model Versions

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

3. Batch Processing

import replicate

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 = [
    "Explain what Python is",
    "Explain what JavaScript is",
    "Explain what Rust is"
]

results = batchpredict(prompts)

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

import replicate

Check hardware requirements and pricing before running models
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 all predictions for cost monitoring
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")

Conclusion

Replicate simplifies the process of running and deploying machine learning models by providing an easy-to-use API and managed GPU infrastructure. Here is a summary of the key takeaways:

• Easy Access: Run thousands of open-source models without infrastructure setup through the Python SDK or REST API.
• Streaming: Use replicate.stream() for real-time LLM output, providing a better user experience for chat-style applications.
• Webhooks: For long-running tasks, use webhooks so your server receives automatic notifications when predictions complete.
• Cog: Package your custom models into deployment-ready containers using Cog, Replicate's open-source packaging tool.
• Fine-tuning: Leverage the fine-tuning feature to adapt models like SDXL with your custom datasets.
• Deployments: Create dedicated deployments with configurable hardware and scaling to match your workload requirements.
• Integration: Replicate integrates well with popular frameworks like LangChain and FastAPI for building production applications.

By understanding these features, you can build powerful AI applications without dealing with the complexity of GPU infrastructure management directly.