【完全ガイド】Replicate API設定・使い方｜画像生成AIを簡単に実装する方法

生成AIを自分のアプリやサービスに組み込みたいと思ったことはありませんか？「Stable Diffusionで画像を生成したい」「動画を自動生成したい」と思っても、GPUサーバーの準備や環境構築のハードルが高くて諦めてしまう方も多いのではないでしょうか。

そんな悩みを解決してくれるのがReplicate APIです。

この記事を読めば、プログラミング初心者でもReplicate APIを使ってAI機能を実装できるようになります。

Replicate APIとは？

Replicate（レプリケート）は、機械学習モデルをクラウド上で実行できるプラットフォームです。開発者は自分でGPUサーバーを用意したり、複雑な環境構築をしたりすることなく、APIを呼び出すだけで様々なAIモデルを利用できます。

Replicateの基本的な仕組み

Replicateの仕組みを簡単に説明すると、以下のようになります：

1. 開発者がモデルを公開: 世界中の開発者が作成した機械学習モデルがReplicateにデプロイされています
2. ユーザーがAPIを呼び出す: あなたはAPIキーを使ってこれらのモデルを呼び出します
3. クラウド上で実行: モデルはReplicateのGPUサーバー上で実行されます
4. 結果を受け取る: 生成された画像やテキストなどの結果が返されます

なぜReplicateが注目されているのか

Replicateが多くの開発者に支持されている理由は以下の通りです：

1. GPUの準備が不要 高価なGPUを購入したり、クラウドGPUをレンタルしたりする必要がありません。Replicateが全て管理してくれます。

2. 環境構築が不要 PyTorchやTensorFlow、CUDAなどの複雑な環境構築が一切不要です。APIを呼び出すだけで動きます。

3. 数千のモデルがすぐに使える Stable Diffusion、FLUX、Whisper、LLaMAなど、数千のオープンソースモデルがすぐに利用可能です。

4. スケーラビリティ アクセスが急増しても自動的にスケールするため、インフラの心配が不要です。

5. 従量課金制 使った分だけの支払いで、最小限のコストからスタートできます。

Replicateで利用できるモデルの種類

Replicateでは非常に多種多様なAIモデルを利用できます。主なカテゴリを紹介します：

Replicate APIの料金体系

Replicate APIは従量課金制を採用しており、使った分だけ料金が発生します。初期費用や月額固定費用はありません。

基本的な料金構造

Replicateの料金は、主に以下の要素で決まります：

1. GPU実行時間: モデルがGPU上で実行された時間
2. 使用するGPUの種類: より高性能なGPUほど料金が高い
3. モデルの種類: 一部のモデルには固定料金が設定されている場合も

GPU別の料金表

※料金は変動する場合があります。最新の料金はReplicate公式サイトでご確認ください。

人気モデルの実際のコスト目安

実際に人気のあるモデルを使用した場合の料金目安をまとめました：

画像生成（1枚あたり）

動画生成（1動画あたり）

音声認識（Whisper）

無料クレジット

新規登録ユーザーには無料クレジットが付与されます。これを使って様々なモデルを試すことができます。

費用を抑えるコツ

1. 適切なモデルを選ぶ

• 用途に合った最小限のモデルを選択
• 例：簡単な画像生成ならFLUX.1 [schnell]、高品質が必要ならFLUX.1 [pro]

2. 解像度を最適化

• 不必要に高い解像度を指定しない
• 後で拡大が必要なら、Real-ESRGANなどのアップスケーラーを使う

3. バッチ処理の活用

• 複数の画像を一度に生成する場合、バッチ処理で効率化

4. キャッシュの活用

• 同じプロンプトでの再生成を避ける
• 結果をローカルに保存して再利用

アカウント登録とAPIキー取得

Replicate APIを使い始めるには、まずアカウントを作成してAPIキーを取得する必要があります。以下の手順に従って設定を進めましょう。

Step 1: アカウント登録

1. Replicateの公式サイトにアクセス

ブラウザで https://replicate.com にアクセスします。

2. サインアップをクリック

トップページ右上の「Sign up」ボタン、または「Get started」ボタンをクリックします。

3. 認証方法を選択

以下の方法でアカウントを作成できます：

• GitHubアカウント: 最も簡単でおすすめ
• Googleアカウント: Gmailユーザーに便利
• メールアドレス: 上記アカウントを使わない場合

4. 利用規約に同意

利用規約とプライバシーポリシーを確認し、同意してアカウントを作成します。

5. 登録完了

これでReplicateアカウントの作成は完了です。ダッシュボードにアクセスできるようになります。

Step 2: APIキーの取得

1. アカウント設定にアクセス

ログイン後、右上のプロフィールアイコンをクリックし、「Account」または「API tokens」を選択します。

直接URLでアクセスする場合は：

https://replicate.com/account/api-tokens

2. APIトークンを作成

「Create token」ボタンをクリックします。

3. トークン名を設定（任意）

トークンに名前を付けることができます。複数のプロジェクトで使い分ける場合は、識別しやすい名前を付けましょう。

例：

• my-blog-project
• image-generation-app
• test-development

4. トークンをコピー

生成されたAPIトークンをコピーします。

:::caution::: セキュリティ上の重要な注意

• APIトークンは一度しか表示されません。必ずコピーして安全な場所に保存してください
• トークンをGitHubなどの公開リポジトリにコミットしないでください
• トークンが漏洩した場合は、すぐに無効化して新しいトークンを生成してください :::

Step 3: 環境変数の設定

APIトークンは環境変数として設定することをおすすめします。コードに直接書くとセキュリティリスクがあります。

macOS/Linux の場合

ターミナルで以下のコマンドを実行します：

# 一時的な設定（現在のターミナルセッションのみ）
export REPLICATE_API_TOKEN=r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxx

# 永続的な設定（.bashrcまたは.zshrcに追加）
echo 'export REPLICATE_API_TOKEN=r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxx' >> ~/.zshrc
source ~/.zshrc

Windows（PowerShell）の場合

# 一時的な設定
$env:REPLICATE_API_TOKEN = "r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxx"

# 永続的な設定（システム環境変数）
[System.Environment]::SetEnvironmentVariable("REPLICATE_API_TOKEN", "r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxx", "User")

Windows（コマンドプロンプト）の場合

set REPLICATE_API_TOKEN=r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxx

Node.js プロジェクトの場合（.envファイル）

プロジェクトルートに.envファイルを作成：

REPLICATE_API_TOKEN=r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxx

.gitignoreに追加することを忘れずに：

.env

Step 4: 設定の確認

環境変数が正しく設定されているか確認しましょう。

ターミナルで確認

echo $REPLICATE_API_TOKEN

トークンが表示されれば設定完了です。

Pythonで確認

import os
print(os.environ.get('REPLICATE_API_TOKEN'))

これでReplicate APIを使用する準備が整いました。

Python での基本的な使い方

PythonはReplicate APIを使用する上で最も一般的な言語です。公式SDKが充実しており、簡単にAI機能を実装できます。

Python SDKのインストール

まず、Replicate公式のPythonパッケージをインストールします：

pip install replicate

または、プロジェクトで仮想環境を使用している場合：

# venv環境の場合
python -m venv venv
source venv/bin/activate  # macOS/Linux
.\venv\Scripts\activate   # Windows

pip install replicate

requirements.txtを使用する場合

replicate>=0.22.0

pip install -r requirements.txt

最初のAPI呼び出し

環境変数REPLICATE_API_TOKENが設定されていれば、以下のコードで簡単にAPIを呼び出せます：

import replicate

# FLUX.1 [schnell] で画像を生成
output = replicate.run(
    "black-forest-labs/flux-schnell",
    input={
        "prompt": "A beautiful sunset over the ocean, dramatic clouds, golden hour lighting"
    }
)

# 結果のURLを表示
print(output)

このコードを実行すると、生成された画像のURLが返されます。

replicate.run() の詳細

replicate.run()関数は、モデルを同期的に実行します。処理が完了するまで待機し、結果を返します。

基本的な構文

output = replicate.run(
    "owner/model-name:version",  # モデルの指定
    input={                       # 入力パラメータ
        "param1": "value1",
        "param2": "value2"
    }
)

モデルの指定方法

モデルは以下の形式で指定します：

1. 最新バージョンを使用: owner/model-name
2. 特定バージョンを指定: owner/model-name:version-hash

# 最新バージョン（推奨）
output = replicate.run(
    "black-forest-labs/flux-schnell",
    input={"prompt": "A cat"}
)

# 特定バージョンを指定
output = replicate.run(
    "black-forest-labs/flux-schnell:5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637",
    input={"prompt": "A cat"}
)

画像生成の実践例

FLUX.1 [schnell] を使った画像生成の詳細な例を見てみましょう：

import replicate
import urllib.request
import os

def generate_image(prompt, output_path="output.png"):
    """
    FLUX.1 [schnell] で画像を生成し、ローカルに保存する

    Args:
        prompt: 生成したい画像の説明
        output_path: 保存先のファイルパス

    Returns:
        保存したファイルのパス
    """

    # 画像を生成
    output = replicate.run(
        "black-forest-labs/flux-schnell",
        input={
            "prompt": prompt,
            "num_outputs": 1,           # 生成する画像の数
            "aspect_ratio": "1:1",      # アスペクト比
            "output_format": "png",     # 出力形式
            "output_quality": 90,       # 出力品質（JPEGの場合）
        }
    )

    # 結果はURLのリスト
    image_url = output[0]
    print(f"Generated image URL: {image_url}")

    # 画像をダウンロードして保存
    urllib.request.urlretrieve(image_url, output_path)
    print(f"Image saved to: {output_path}")

    return output_path

# 使用例
if __name__ == "__main__":
    prompt = """
    A professional photo of a sleek gaming setup with RGB lighting,
    mechanical keyboard, gaming mouse, and ultrawide monitor,
    dark room with purple ambient lighting, high quality, 4k
    """

    generate_image(prompt, "gaming_setup.png")

非同期処理（Predictions API）

大きなモデルや時間のかかる処理の場合、非同期で実行することをおすすめします：

import replicate
import time

# 非同期で予測を開始
prediction = replicate.predictions.create(
    model="black-forest-labs/flux-dev",
    input={
        "prompt": "A detailed fantasy landscape with mountains and waterfalls",
        "num_inference_steps": 50,
        "guidance_scale": 7.5
    }
)

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

# ステータスをポーリングして完了を待つ
while prediction.status not in ["succeeded", "failed", "canceled"]:
    time.sleep(2)
    prediction = replicate.predictions.get(prediction.id)
    print(f"Status: {prediction.status}")

# 結果を確認
if prediction.status == "succeeded":
    print(f"Output: {prediction.output}")
else:
    print(f"Failed: {prediction.error}")

ストリーミング出力

一部のモデル（特にテキスト生成モデル）はストリーミング出力に対応しています：

import replicate

# LLaMAでテキスト生成（ストリーミング）
for event in replicate.stream(
    "meta/llama-2-70b-chat",
    input={
        "prompt": "Explain quantum computing in simple terms:",
        "max_new_tokens": 500
    }
):
    print(str(event), end="", flush=True)

print()  # 最後に改行

複数画像の生成

一度のリクエストで複数の画像を生成することも可能です：

import replicate

# 複数画像を一度に生成
output = replicate.run(
    "black-forest-labs/flux-schnell",
    input={
        "prompt": "A cute robot assistant, cartoon style, friendly expression",
        "num_outputs": 4,  # 4枚の画像を生成
        "aspect_ratio": "1:1"
    }
)

# 結果は複数のURLのリスト
for i, url in enumerate(output):
    print(f"Image {i+1}: {url}")

エラーハンドリング

API呼び出し時のエラーを適切に処理しましょう：

import replicate
from replicate.exceptions import ReplicateError, ModelError

def safe_generate_image(prompt):
    """エラーハンドリングを含む画像生成"""

    try:
        output = replicate.run(
            "black-forest-labs/flux-schnell",
            input={"prompt": prompt}
        )
        return output[0]

    except replicate.exceptions.ReplicateError as e:
        # Replicate API固有のエラー
        print(f"Replicate API Error: {e}")
        return None

    except Exception as e:
        # その他のエラー
        print(f"Unexpected error: {e}")
        return None

# 使用例
result = safe_generate_image("A beautiful garden")
if result:
    print(f"Success: {result}")
else:
    print("Image generation failed")

実践的なユーティリティクラス

プロジェクトで使いやすいようにクラスとして整理した例：

import replicate
import urllib.request
import os
from typing import Optional, List
from dataclasses import dataclass
from enum import Enum

class AspectRatio(Enum):
    SQUARE = "1:1"
    LANDSCAPE = "16:9"
    PORTRAIT = "9:16"
    WIDE = "21:9"
    STANDARD = "4:3"

@dataclass
class ImageGenerationResult:
    urls: List[str]
    prompt: str
    model: str

class ReplicateImageGenerator:
    """Replicate APIを使った画像生成クラス"""

    # 利用可能なモデル
    MODELS = {
        "flux-schnell": "black-forest-labs/flux-schnell",
        "flux-dev": "black-forest-labs/flux-dev",
        "sdxl": "stability-ai/sdxl",
        "sd15": "stability-ai/stable-diffusion",
    }

    def __init__(self, model: str = "flux-schnell"):
        """
        Args:
            model: 使用するモデル名（flux-schnell, flux-dev, sdxl, sd15）
        """
        if model not in self.MODELS:
            raise ValueError(f"Unknown model: {model}. Available: {list(self.MODELS.keys())}")
        self.model = self.MODELS[model]
        self.model_name = model

    def generate(
        self,
        prompt: str,
        num_outputs: int = 1,
        aspect_ratio: AspectRatio = AspectRatio.SQUARE,
        save_to: Optional[str] = None
    ) -> ImageGenerationResult:
        """
        画像を生成する

        Args:
            prompt: 生成する画像の説明
            num_outputs: 生成する画像の数（1-4）
            aspect_ratio: アスペクト比
            save_to: 保存先ディレクトリ（指定時は自動保存）

        Returns:
            ImageGenerationResult: 生成結果
        """

        output = replicate.run(
            self.model,
            input={
                "prompt": prompt,
                "num_outputs": min(num_outputs, 4),
                "aspect_ratio": aspect_ratio.value
            }
        )

        urls = list(output) if isinstance(output, list) else [output]

        # 自動保存
        if save_to:
            os.makedirs(save_to, exist_ok=True)
            for i, url in enumerate(urls):
                filename = f"{self.model_name}_{i+1}.png"
                filepath = os.path.join(save_to, filename)
                urllib.request.urlretrieve(url, filepath)
                print(f"Saved: {filepath}")

        return ImageGenerationResult(
            urls=urls,
            prompt=prompt,
            model=self.model
        )

# 使用例
if __name__ == "__main__":
    generator = ReplicateImageGenerator("flux-schnell")

    result = generator.generate(
        prompt="A cozy coffee shop interior, warm lighting, plants",
        num_outputs=2,
        aspect_ratio=AspectRatio.LANDSCAPE,
        save_to="./generated_images"
    )

    print(f"Generated {len(result.urls)} images")
    for url in result.urls:
        print(url)

JavaScript/Node.js での使い方

JavaScript/Node.jsでもReplicate APIを簡単に利用できます。Webアプリケーションやサーバーサイドでの実装に最適です。

パッケージのインストール

npm または yarn を使ってインストールします：

# npm
npm install replicate

# yarn
yarn add replicate

# pnpm
pnpm add replicate

基本的な使い方

CommonJS形式

const Replicate = require("replicate");

const replicate = new Replicate({
  auth: process.env.REPLICATE_API_TOKEN,
});

async function generateImage() {
  const output = await replicate.run(
    "black-forest-labs/flux-schnell",
    {
      input: {
        prompt: "A futuristic cityscape at night with neon lights"
      }
    }
  );

  console.log(output);
}

generateImage();

ES Modules形式

import Replicate from "replicate";

const replicate = new Replicate({
  auth: process.env.REPLICATE_API_TOKEN,
});

const output = await replicate.run(
  "black-forest-labs/flux-schnell",
  {
    input: {
      prompt: "A serene Japanese garden with cherry blossoms"
    }
  }
);

console.log(output);

TypeScriptでの使用

TypeScriptを使う場合、型定義も含まれています：

import Replicate from "replicate";

const replicate = new Replicate({
  auth: process.env.REPLICATE_API_TOKEN!,
});

interface FluxInput {
  prompt: string;
  num_outputs?: number;
  aspect_ratio?: string;
  output_format?: "webp" | "png" | "jpg";
}

async function generateWithFlux(options: FluxInput): Promise<string[]> {
  const output = await replicate.run(
    "black-forest-labs/flux-schnell",
    { input: options }
  );

  return output as string[];
}

// 使用例
const images = await generateWithFlux({
  prompt: "A professional headshot photo, studio lighting",
  num_outputs: 2,
  aspect_ratio: "1:1"
});

images.forEach((url, i) => {
  console.log(`Image ${i + 1}: ${url}`);
});

画像のダウンロード保存

生成した画像をローカルに保存する例：

import Replicate from "replicate";
import fs from "fs";
import https from "https";
import path from "path";

const replicate = new Replicate();

async function downloadImage(url, filepath) {
  return new Promise((resolve, reject) => {
    const file = fs.createWriteStream(filepath);
    https.get(url, (response) => {
      response.pipe(file);
      file.on("finish", () => {
        file.close();
        resolve(filepath);
      });
    }).on("error", (err) => {
      fs.unlink(filepath, () => {});
      reject(err);
    });
  });
}

async function generateAndSave(prompt, outputDir = "./output") {
  // ディレクトリ作成
  if (!fs.existsSync(outputDir)) {
    fs.mkdirSync(outputDir, { recursive: true });
  }

  // 画像生成
  const output = await replicate.run(
    "black-forest-labs/flux-schnell",
    {
      input: {
        prompt: prompt,
        num_outputs: 1
      }
    }
  );

  // ダウンロード
  const timestamp = Date.now();
  const filepath = path.join(outputDir, `generated_${timestamp}.png`);
  await downloadImage(output[0], filepath);

  console.log(`Saved to: ${filepath}`);
  return filepath;
}

// 実行
generateAndSave("A magical forest with glowing mushrooms");

fetch APIを使った方法（ブラウザ互換）

Node.js 18以降、またはブラウザ環境では、SDKを使わずにfetch APIで直接呼び出すこともできます：

async function generateImageWithFetch(prompt) {
  const response = await fetch("https://api.replicate.com/v1/predictions", {
    method: "POST",
    headers: {
      "Authorization": `Token ${process.env.REPLICATE_API_TOKEN}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      version: "5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637",
      input: {
        prompt: prompt
      }
    }),
  });

  const prediction = await response.json();
  console.log("Prediction created:", prediction.id);

  // ポーリングして結果を待つ
  let result = prediction;
  while (result.status !== "succeeded" && result.status !== "failed") {
    await new Promise(resolve => setTimeout(resolve, 2000));

    const statusResponse = await fetch(
      `https://api.replicate.com/v1/predictions/${prediction.id}`,
      {
        headers: {
          "Authorization": `Token ${process.env.REPLICATE_API_TOKEN}`,
        },
      }
    );
    result = await statusResponse.json();
    console.log("Status:", result.status);
  }

  if (result.status === "succeeded") {
    return result.output;
  } else {
    throw new Error(`Prediction failed: ${result.error}`);
  }
}

// 使用
generateImageWithFetch("A beautiful mountain landscape")
  .then(output => console.log("Output:", output))
  .catch(err => console.error("Error:", err));

Next.js APIルートでの使用例

Next.jsのAPIルートからReplicate APIを呼び出す例：

// pages/api/generate-image.ts
import type { NextApiRequest, NextApiResponse } from "next";
import Replicate from "replicate";

const replicate = new Replicate({
  auth: process.env.REPLICATE_API_TOKEN,
});

export default async function handler(
  req: NextApiRequest,
  res: NextApiResponse
) {
  if (req.method !== "POST") {
    return res.status(405).json({ error: "Method not allowed" });
  }

  const { prompt } = req.body;

  if (!prompt) {
    return res.status(400).json({ error: "Prompt is required" });
  }

  try {
    const output = await replicate.run(
      "black-forest-labs/flux-schnell",
      {
        input: {
          prompt: prompt,
          num_outputs: 1
        }
      }
    );

    res.status(200).json({ images: output });
  } catch (error) {
    console.error("Replicate error:", error);
    res.status(500).json({ error: "Image generation failed" });
  }
}

フロントエンド側の呼び出し

// components/ImageGenerator.tsx
"use client";

import { useState } from "react";

export function ImageGenerator() {
  const [prompt, setPrompt] = useState("");
  const [imageUrl, setImageUrl] = useState<string | null>(null);
  const [loading, setLoading] = useState(false);

  const handleGenerate = async () => {
    setLoading(true);
    try {
      const response = await fetch("/api/generate-image", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ prompt }),
      });

      const data = await response.json();
      setImageUrl(data.images[0]);
    } catch (error) {
      console.error("Error:", error);
    } finally {
      setLoading(false);
    }
  };

  return (
    <div>
      <input
        type="text"
        value={prompt}
        onChange={(e) => setPrompt(e.target.value)}
        placeholder="Enter your prompt..."
      />
      <button onClick={handleGenerate} disabled={loading}>
        {loading ? "Generating..." : "Generate"}
      </button>

      {imageUrl && <img src={imageUrl} alt="Generated" />}
    </div>
  );
}

cURL での使い方

コマンドラインからReplicate APIを直接呼び出すこともできます。テストやシェルスクリプトでの利用に便利です。

基本的なAPI呼び出し

予測の作成（非同期）

curl -s -X POST \
  -H "Authorization: Token $REPLICATE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "version": "5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637",
    "input": {
      "prompt": "A cute cat wearing a tiny hat"
    }
  }' \
  https://api.replicate.com/v1/predictions

レスポンス例

{
  "id": "abc123xyz",
  "version": "5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637",
  "status": "starting",
  "input": {
    "prompt": "A cute cat wearing a tiny hat"
  },
  "urls": {
    "get": "https://api.replicate.com/v1/predictions/abc123xyz",
    "cancel": "https://api.replicate.com/v1/predictions/abc123xyz/cancel"
  }
}

予測結果の取得

curl -s \
  -H "Authorization: Token $REPLICATE_API_TOKEN" \
  https://api.replicate.com/v1/predictions/abc123xyz

同期的な実行（wait パラメータ）

結果が出るまで待機する方法：

curl -s -X POST \
  -H "Authorization: Token $REPLICATE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Prefer: wait" \
  -d '{
    "version": "5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637",
    "input": {
      "prompt": "A beautiful sunset"
    }
  }' \
  https://api.replicate.com/v1/predictions

シェルスクリプトでの自動化

画像生成からダウンロードまでを自動化するスクリプト：

#!/bin/bash

PROMPT="$1"
OUTPUT_DIR="${2:-./output}"

if [ -z "$PROMPT" ]; then
  echo "Usage: $0 \"your prompt\" [output_directory]"
  exit 1
fi

mkdir -p "$OUTPUT_DIR"

echo "Creating prediction..."
PREDICTION=$(curl -s -X POST \
  -H "Authorization: Token $REPLICATE_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d "{
    \"version\": \"5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637\",
    \"input\": {
      \"prompt\": \"$PROMPT\"
    }
  }" \
  https://api.replicate.com/v1/predictions)

PREDICTION_ID=$(echo $PREDICTION | jq -r '.id')
echo "Prediction ID: $PREDICTION_ID"

# ポーリング
while true; do
  RESULT=$(curl -s \
    -H "Authorization: Token $REPLICATE_API_TOKEN" \
    https://api.replicate.com/v1/predictions/$PREDICTION_ID)

  STATUS=$(echo $RESULT | jq -r '.status')
  echo "Status: $STATUS"

  if [ "$STATUS" = "succeeded" ]; then
    IMAGE_URL=$(echo $RESULT | jq -r '.output[0]')
    echo "Image URL: $IMAGE_URL"

    # ダウンロード
    FILENAME="$OUTPUT_DIR/generated_$(date +%s).png"
    curl -s -o "$FILENAME" "$IMAGE_URL"
    echo "Saved to: $FILENAME"
    break
  elif [ "$STATUS" = "failed" ]; then
    echo "Prediction failed"
    echo $RESULT | jq '.error'
    exit 1
  fi

  sleep 2
done

使用方法：

chmod +x generate.sh
./generate.sh "A cyberpunk city at night" ./my_images

jqを使ったレスポンス処理

jqコマンドを使うと、JSONレスポンスを簡単に処理できます：

# 出力URLだけを取得
curl -s -H "Authorization: Token $REPLICATE_API_TOKEN" \
  https://api.replicate.com/v1/predictions/abc123 | jq -r '.output[]'

# ステータスだけを取得
curl -s -H "Authorization: Token $REPLICATE_API_TOKEN" \
  https://api.replicate.com/v1/predictions/abc123 | jq -r '.status'

# 全ての予測一覧を取得
curl -s -H "Authorization: Token $REPLICATE_API_TOKEN" \
  https://api.replicate.com/v1/predictions | jq '.results[] | {id, status, model}'

人気モデルの紹介と使い方

Replicateには数千のモデルが公開されていますが、ここでは特に人気のあるモデルを紹介します。

画像生成モデル

FLUX.1 シリーズ（Black Forest Labs）

FLUX.1は2024年に登場した最新の画像生成モデルで、Stable Diffusionの次世代として注目されています。

FLUX.1 [schnell] - 高速版

• 最速の画像生成（1-4ステップで生成）
• 料金：約$0.003/画像
• 用途：プロトタイピング、大量生成

import replicate

output = replicate.run(
    "black-forest-labs/flux-schnell",
    input={
        "prompt": "A professional product photo of a sleek smartphone",
        "num_outputs": 1,
        "aspect_ratio": "1:1",
        "output_format": "png",
        "output_quality": 90
    }
)

FLUX.1 [dev] - 開発版

• 高品質な画像生成
• 料金：約$0.025/画像
• 用途：一般的な画像生成

output = replicate.run(
    "black-forest-labs/flux-dev",
    input={
        "prompt": "An intricate steampunk pocket watch with gears and brass details",
        "num_inference_steps": 28,
        "guidance_scale": 3.5,
        "aspect_ratio": "1:1"
    }
)

FLUX.1 [pro] - プロ版

• 最高品質
• 料金：約$0.055/画像
• 用途：商用利用、高品質が必要な場合

output = replicate.run(
    "black-forest-labs/flux-pro",
    input={
        "prompt": "A photorealistic portrait of a young professional in business attire",
        "aspect_ratio": "3:4",
        "safety_tolerance": 2
    }
)

Stable Diffusion XL (SDXL)

高品質な1024x1024画像を生成できる定番モデル：

output = replicate.run(
    "stability-ai/sdxl",
    input={
        "prompt": "A majestic dragon flying over ancient mountains, epic fantasy art",
        "negative_prompt": "ugly, blurry, low quality, distorted",
        "width": 1024,
        "height": 1024,
        "num_inference_steps": 30,
        "guidance_scale": 7.5,
        "scheduler": "K_EULER_ANCESTRAL"
    }
)

Stable Diffusion 1.5

軽量で高速、様々な用途に対応：

output = replicate.run(
    "stability-ai/stable-diffusion",
    input={
        "prompt": "A cozy cabin in the woods during autumn",
        "negative_prompt": "blurry, bad quality",
        "width": 512,
        "height": 512,
        "num_inference_steps": 50,
        "guidance_scale": 7.5
    }
)

画像編集モデル

背景除去（Remove Background）

output = replicate.run(
    "cjwbw/rembg",
    input={
        "image": "https://example.com/photo.jpg"
    }
)
# 背景が透明になった画像のURLが返される

超解像度化（Real-ESRGAN）

画像を高解像度化：

output = replicate.run(
    "nightmareai/real-esrgan",
    input={
        "image": "https://example.com/low-res.jpg",
        "scale": 4,  # 4倍に拡大
        "face_enhance": True  # 顔を強調
    }
)

インペインティング（画像の一部を編集）

output = replicate.run(
    "stability-ai/stable-diffusion-inpainting",
    input={
        "prompt": "A red sports car",
        "image": "https://example.com/scene.jpg",
        "mask": "https://example.com/mask.jpg",  # 編集したい領域を白で指定
        "num_inference_steps": 50
    }
)

動画生成モデル

Stable Video Diffusion

画像から動画を生成：

output = replicate.run(
    "stability-ai/stable-video-diffusion",
    input={
        "input_image": "https://example.com/starting-frame.jpg",
        "frames_per_second": 6,
        "motion_bucket_id": 127,  # 動きの大きさ（1-255）
        "cond_aug": 0.02
    }
)
# 動画のURLが返される

AnimateDiff

テキストから動画を生成：

output = replicate.run(
    "lucataco/animate-diff",
    input={
        "prompt": "A butterfly flying through a garden, smooth motion",
        "negative_prompt": "blurry, distorted",
        "num_frames": 16,
        "num_inference_steps": 25
    }
)

音声認識モデル

Whisper（OpenAI）

音声からテキストを生成：

output = replicate.run(
    "openai/whisper",
    input={
        "audio": "https://example.com/audio.mp3",
        "model": "large-v3",
        "language": "ja",  # 日本語
        "translate": False,
        "transcription": "plain text"
    }
)
# {"text": "認識されたテキスト", "segments": [...]}

日本語音声の文字起こし

def transcribe_japanese(audio_url):
    """日本語音声を文字起こし"""
    output = replicate.run(
        "openai/whisper",
        input={
            "audio": audio_url,
            "model": "large-v3",
            "language": "ja",
            "transcription": "plain text"
        }
    )
    return output["text"]

# 使用例
text = transcribe_japanese("https://example.com/japanese-audio.mp3")
print(text)

テキスト生成モデル

LLaMA 2

Meta社のオープンソースLLM：

output = replicate.run(
    "meta/llama-2-70b-chat",
    input={
        "prompt": "What are the key differences between Python and JavaScript?",
        "system_prompt": "You are a helpful programming assistant.",
        "max_new_tokens": 500,
        "temperature": 0.7,
        "top_p": 0.9
    }
)

Code Llama

コード生成に特化：

output = replicate.run(
    "meta/codellama-34b-instruct",
    input={
        "prompt": "Write a Python function to calculate fibonacci numbers:",
        "max_new_tokens": 500,
        "temperature": 0.1
    }
)

モデルの検索と選び方

Replicateには数千のモデルがあります。目的に合ったモデルを見つけるコツ：

1. 公式サイトで検索

• https://replicate.com/explore でカテゴリ別に検索
• 「image generation」「video」「audio」などで絞り込み

2. 人気度と実行回数を確認

• 実行回数が多いモデルは信頼性が高い傾向
• スター数やフォーク数も参考に

3. 料金と速度のバランス

• 用途に応じて適切なモデルを選択
• テスト時は安価なモデル、本番は高品質モデル

4. ライセンスの確認

• 商用利用可能かどうか
• 帰属表示が必要かどうか

実践的なコード例集

ここでは、実際のプロジェクトで使える実践的なコード例を紹介します。

1. バッチ画像生成システム

複数のプロンプトから画像を一括生成：

import replicate
import asyncio
import aiohttp
import os
from typing import List, Dict
from dataclasses import dataclass
from datetime import datetime

@dataclass
class GenerationTask:
    prompt: str
    filename: str
    aspect_ratio: str = "1:1"

@dataclass
class GenerationResult:
    task: GenerationTask
    url: str
    success: bool
    error: str = ""

class BatchImageGenerator:
    """バッチ画像生成クラス"""

    def __init__(self, output_dir: str = "./generated"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def generate_batch(self, tasks: List[GenerationTask]) -> List[GenerationResult]:
        """
        複数の画像を順次生成

        Args:
            tasks: 生成タスクのリスト

        Returns:
            生成結果のリスト
        """
        results = []

        for i, task in enumerate(tasks, 1):
            print(f"Generating {i}/{len(tasks)}: {task.filename}")

            try:
                output = replicate.run(
                    "black-forest-labs/flux-schnell",
                    input={
                        "prompt": task.prompt,
                        "aspect_ratio": task.aspect_ratio
                    }
                )

                url = output[0]
                results.append(GenerationResult(
                    task=task,
                    url=url,
                    success=True
                ))

            except Exception as e:
                results.append(GenerationResult(
                    task=task,
                    url="",
                    success=False,
                    error=str(e)
                ))

        return results

    async def download_images(self, results: List[GenerationResult]):
        """生成された画像を一括ダウンロード"""

        async with aiohttp.ClientSession() as session:
            tasks = []
            for result in results:
                if result.success:
                    filepath = os.path.join(self.output_dir, result.task.filename)
                    tasks.append(self._download_one(session, result.url, filepath))

            await asyncio.gather(*tasks)

    async def _download_one(self, session, url: str, filepath: str):
        """1つの画像をダウンロード"""
        async with session.get(url) as response:
            content = await response.read()
            with open(filepath, "wb") as f:
                f.write(content)
            print(f"Downloaded: {filepath}")

# 使用例
if __name__ == "__main__":
    generator = BatchImageGenerator("./batch_output")

    # 生成タスクを定義
    tasks = [
        GenerationTask(
            prompt="A serene Japanese garden with cherry blossoms",
            filename="japanese_garden.png",
            aspect_ratio="16:9"
        ),
        GenerationTask(
            prompt="A futuristic city skyline at sunset",
            filename="future_city.png",
            aspect_ratio="16:9"
        ),
        GenerationTask(
            prompt="A cozy coffee shop interior with warm lighting",
            filename="coffee_shop.png",
            aspect_ratio="4:3"
        ),
        GenerationTask(
            prompt="An underwater scene with colorful coral reef",
            filename="underwater.png",
            aspect_ratio="16:9"
        ),
    ]

    # 生成実行
    results = generator.generate_batch(tasks)

    # 結果を表示
    for result in results:
        if result.success:
            print(f"✓ {result.task.filename}: {result.url}")
        else:
            print(f"✗ {result.task.filename}: {result.error}")

    # 画像をダウンロード
    asyncio.run(generator.download_images(results))

2. 画像のスタイル変換パイプライン

元画像を様々なスタイルに変換：

import replicate
from typing import Dict, List

class ImageStyleTransformer:
    """画像スタイル変換クラス"""

    STYLES = {
        "anime": "anime style, colorful, detailed illustration",
        "oil_painting": "oil painting style, impressionist, brushstrokes visible",
        "watercolor": "watercolor painting, soft edges, flowing colors",
        "pencil_sketch": "pencil sketch, detailed line art, black and white",
        "cyberpunk": "cyberpunk style, neon colors, futuristic",
        "vintage": "vintage photograph, sepia tones, nostalgic",
        "pixel_art": "pixel art style, 8-bit, retro gaming",
    }

    def transform(
        self,
        image_url: str,
        style: str,
        strength: float = 0.75
    ) -> str:
        """
        画像をスタイル変換

        Args:
            image_url: 元画像のURL
            style: スタイル名（STYLES参照）
            strength: 変換の強さ（0.0-1.0）

        Returns:
            変換後の画像URL
        """
        if style not in self.STYLES:
            raise ValueError(f"Unknown style: {style}. Available: {list(self.STYLES.keys())}")

        style_prompt = self.STYLES[style]

        output = replicate.run(
            "stability-ai/sdxl",
            input={
                "prompt": f"Transform this image into {style_prompt}",
                "image": image_url,
                "prompt_strength": strength,
                "num_inference_steps": 30
            }
        )

        return output[0]

    def transform_all_styles(self, image_url: str) -> Dict[str, str]:
        """全スタイルに変換"""
        results = {}

        for style_name in self.STYLES:
            print(f"Transforming to {style_name}...")
            try:
                results[style_name] = self.transform(image_url, style_name)
            except Exception as e:
                results[style_name] = f"Error: {e}"

        return results

# 使用例
transformer = ImageStyleTransformer()
result = transformer.transform(
    "https://example.com/photo.jpg",
    "anime",
    strength=0.8
)
print(f"Anime version: {result}")

3. 音声文字起こしシステム

長時間の音声を効率的に文字起こし：

import replicate
from dataclasses import dataclass
from typing import List, Optional
import json

@dataclass
class TranscriptionSegment:
    start: float
    end: float
    text: str

@dataclass
class TranscriptionResult:
    text: str
    segments: List[TranscriptionSegment]
    language: str
    duration: float

class AudioTranscriber:
    """音声文字起こしクラス"""

    SUPPORTED_LANGUAGES = ["en", "ja", "zh", "ko", "es", "fr", "de", "it", "pt", "ru"]

    def transcribe(
        self,
        audio_url: str,
        language: Optional[str] = None,
        include_timestamps: bool = True
    ) -> TranscriptionResult:
        """
        音声を文字起こし

        Args:
            audio_url: 音声ファイルのURL
            language: 言語コード（Noneで自動検出）
            include_timestamps: タイムスタンプを含めるか

        Returns:
            TranscriptionResult
        """

        input_params = {
            "audio": audio_url,
            "model": "large-v3",
            "transcription": "srt" if include_timestamps else "plain text"
        }

        if language:
            input_params["language"] = language

        output = replicate.run(
            "openai/whisper",
            input=input_params
        )

        # 結果をパース
        segments = []
        if "segments" in output:
            for seg in output["segments"]:
                segments.append(TranscriptionSegment(
                    start=seg["start"],
                    end=seg["end"],
                    text=seg["text"]
                ))

        return TranscriptionResult(
            text=output.get("text", ""),
            segments=segments,
            language=output.get("detected_language", language or "unknown"),
            duration=segments[-1].end if segments else 0.0
        )

    def transcribe_to_srt(self, audio_url: str, output_path: str, language: Optional[str] = None):
        """SRT字幕ファイルとして出力"""

        result = self.transcribe(audio_url, language, include_timestamps=True)

        srt_content = ""
        for i, seg in enumerate(result.segments, 1):
            start_time = self._format_srt_time(seg.start)
            end_time = self._format_srt_time(seg.end)
            srt_content += f"{i}\n{start_time} --> {end_time}\n{seg.text.strip()}\n\n"

        with open(output_path, "w", encoding="utf-8") as f:
            f.write(srt_content)

        print(f"SRT saved to: {output_path}")
        return output_path

    def _format_srt_time(self, seconds: float) -> str:
        """秒数をSRT形式の時間文字列に変換"""
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        millis = int((seconds % 1) * 1000)
        return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"

# 使用例
transcriber = AudioTranscriber()

# 日本語音声の文字起こし
result = transcriber.transcribe(
    "https://example.com/japanese-podcast.mp3",
    language="ja"
)
print(f"言語: {result.language}")
print(f"長さ: {result.duration:.1f}秒")
print(f"テキスト: {result.text[:200]}...")

# SRTファイルとして保存
transcriber.transcribe_to_srt(
    "https://example.com/video-audio.mp3",
    "subtitles.srt",
    language="ja"
)

4. 商品画像自動生成システム

EC向けの商品画像を自動生成：

import replicate
from typing import List, Dict
from dataclasses import dataclass

@dataclass
class ProductImageConfig:
    name: str
    description: str
    background: str = "clean white background, studio lighting"
    angles: List[str] = None

    def __post_init__(self):
        if self.angles is None:
            self.angles = ["front view", "45 degree angle", "side view"]

class ProductImageGenerator:
    """商品画像生成クラス"""

    BACKGROUNDS = {
        "studio": "clean white background, professional studio lighting, soft shadows",
        "lifestyle": "modern lifestyle setting, natural lighting, styled environment",
        "minimal": "minimalist background, subtle gradient, clean aesthetic",
        "nature": "natural outdoor setting, soft natural lighting",
    }

    def generate_product_images(
        self,
        config: ProductImageConfig,
        background_style: str = "studio"
    ) -> Dict[str, str]:
        """
        商品の複数アングル画像を生成

        Args:
            config: 商品設定
            background_style: 背景スタイル

        Returns:
            アングル名: URL の辞書
        """
        background = self.BACKGROUNDS.get(background_style, self.BACKGROUNDS["studio"])
        results = {}

        for angle in config.angles:
            prompt = f"""
            Professional product photography of {config.name},
            {config.description},
            {angle},
            {background},
            high resolution, 4K quality, sharp focus,
            commercial product shot
            """.strip().replace("\n", " ")

            print(f"Generating {angle}...")

            output = replicate.run(
                "black-forest-labs/flux-dev",
                input={
                    "prompt": prompt,
                    "aspect_ratio": "1:1",
                    "num_inference_steps": 28
                }
            )

            results[angle] = output[0]

        return results

    def generate_with_background_removal(self, image_url: str) -> str:
        """背景を透明に"""
        output = replicate.run(
            "cjwbw/rembg",
            input={"image": image_url}
        )
        return output

# 使用例
generator = ProductImageGenerator()

config = ProductImageConfig(
    name="wireless earbuds",
    description="sleek black wireless earbuds with charging case, premium design, matte finish",
    angles=["front view", "case open showing earbuds", "single earbud close-up"]
)

images = generator.generate_product_images(config, background_style="studio")

for angle, url in images.items():
    print(f"{angle}: {url}")

# 背景を透明にしたバージョンも生成
for angle, url in images.items():
    transparent = generator.generate_with_background_removal(url)
    print(f"{angle} (transparent): {transparent}")

5. AIアート生成ツール

様々なアートスタイルで画像を生成：

import replicate
from enum import Enum
from typing import Optional

class ArtStyle(Enum):
    PHOTOREALISTIC = "photorealistic, highly detailed, 8k resolution, professional photography"
    ANIME = "anime style, vibrant colors, cel shading, Japanese animation"
    OIL_PAINTING = "oil painting, classic art style, rich textures, visible brushstrokes"
    WATERCOLOR = "watercolor painting, soft edges, flowing colors, artistic"
    DIGITAL_ART = "digital art, concept art, detailed illustration, artstation style"
    SURREALISM = "surrealist art style, dreamlike, Salvador Dali inspired"
    POP_ART = "pop art style, bold colors, Andy Warhol inspired"
    MINIMALIST = "minimalist design, clean lines, simple shapes, modern"
    COMIC_BOOK = "comic book style, bold outlines, dynamic composition, halftone"
    CYBERPUNK = "cyberpunk aesthetic, neon lights, futuristic, dark atmosphere"

class AIArtGenerator:
    """AIアート生成クラス"""

    NEGATIVE_PROMPTS = {
        ArtStyle.PHOTOREALISTIC: "cartoon, illustration, painting, blurry, low quality",
        ArtStyle.ANIME: "photorealistic, 3D render, western cartoon",
        ArtStyle.OIL_PAINTING: "digital, photograph, smooth, anime",
        # 他のスタイルも同様に定義
    }

    def generate_art(
        self,
        subject: str,
        style: ArtStyle,
        model: str = "flux-dev",
        additional_details: Optional[str] = None
    ) -> str:
        """
        アート作品を生成

        Args:
            subject: 描画対象
            style: アートスタイル
            model: 使用モデル
            additional_details: 追加の詳細

        Returns:
            生成画像のURL
        """

        prompt_parts = [subject, style.value]
        if additional_details:
            prompt_parts.append(additional_details)

        prompt = ", ".join(prompt_parts)

        model_map = {
            "flux-schnell": "black-forest-labs/flux-schnell",
            "flux-dev": "black-forest-labs/flux-dev",
            "sdxl": "stability-ai/sdxl"
        }

        model_id = model_map.get(model, model_map["flux-dev"])

        output = replicate.run(
            model_id,
            input={
                "prompt": prompt,
                "aspect_ratio": "1:1"
            }
        )

        return output[0]

    def generate_variations(
        self,
        subject: str,
        styles: list[ArtStyle] = None
    ) -> dict[str, str]:
        """複数スタイルのバリエーションを生成"""

        if styles is None:
            styles = [ArtStyle.PHOTOREALISTIC, ArtStyle.ANIME, ArtStyle.OIL_PAINTING]

        results = {}
        for style in styles:
            print(f"Generating {style.name}...")
            results[style.name] = self.generate_art(subject, style)

        return results

# 使用例
art_gen = AIArtGenerator()

# 単一スタイルで生成
cyberpunk_art = art_gen.generate_art(
    subject="A samurai warrior standing in rain",
    style=ArtStyle.CYBERPUNK,
    additional_details="neon-lit Tokyo streets, reflections on wet pavement"
)
print(f"Cyberpunk art: {cyberpunk_art}")

# 複数スタイルのバリエーション
variations = art_gen.generate_variations(
    subject="A majestic lion",
    styles=[ArtStyle.PHOTOREALISTIC, ArtStyle.ANIME, ArtStyle.OIL_PAINTING, ArtStyle.POP_ART]
)

for style_name, url in variations.items():
    print(f"{style_name}: {url}")

Webhook の設定と使い方

Webhookを使うと、予測の完了時に指定したURLへ通知を送ることができます。長時間かかる処理や、大量の処理を行う場合に便利です。

Webhookの仕組み

1. 予測作成時にWebhook URLを指定
2. 予測が完了（成功/失敗）すると、ReplicateがそのURLにPOSTリクエストを送信
3. あなたのサーバーで結果を受け取って処理

Webhookの設定方法

Pythonでの設定

import replicate

# Webhookを使って予測を作成
prediction = replicate.predictions.create(
    model="black-forest-labs/flux-dev",
    input={
        "prompt": "A beautiful landscape"
    },
    webhook="https://your-server.com/webhook/replicate",
    webhook_events_filter=["completed"]  # 完了時のみ通知
)

print(f"Prediction ID: {prediction.id}")
print("Webhook will be called when complete")

webhook_events_filter のオプション

• start: 予測開始時
• output: 出力が生成されたとき（ストリーミング）
• logs: ログ出力時
• completed: 完了時（成功/失敗/キャンセル）

Webhookサーバーの実装例

Flask を使った例

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route("/webhook/replicate", methods=["POST"])
def replicate_webhook():
    """Replicateからのwebhookを受け取る"""

    # リクエストボディを取得
    data = request.json

    prediction_id = data.get("id")
    status = data.get("status")
    output = data.get("output")
    error = data.get("error")

    print(f"Received webhook for prediction: {prediction_id}")
    print(f"Status: {status}")

    if status == "succeeded":
        # 成功時の処理
        print(f"Output: {output}")
        # ここで画像をダウンロード、データベースに保存など
        process_completed_prediction(prediction_id, output)
    elif status == "failed":
        # 失敗時の処理
        print(f"Error: {error}")
        handle_failed_prediction(prediction_id, error)

    return jsonify({"received": True}), 200

def process_completed_prediction(prediction_id, output):
    """完了した予測を処理"""
    # 例：データベースに保存
    print(f"Processing: {prediction_id}")
    # db.save_result(prediction_id, output)

def handle_failed_prediction(prediction_id, error):
    """失敗した予測を処理"""
    print(f"Failed: {prediction_id}, Error: {error}")
    # 通知を送るなど

if __name__ == "__main__":
    app.run(port=5000)

FastAPI を使った例

from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Optional, List, Any

app = FastAPI()

class ReplicateWebhook(BaseModel):
    id: str
    status: str
    output: Optional[Any] = None
    error: Optional[str] = None
    created_at: str
    completed_at: Optional[str] = None

@app.post("/webhook/replicate")
async def replicate_webhook(webhook: ReplicateWebhook):
    """Replicateからのwebhookを受け取る"""

    print(f"Prediction {webhook.id}: {webhook.status}")

    if webhook.status == "succeeded":
        # 成功時の処理
        await process_output(webhook.id, webhook.output)
    elif webhook.status == "failed":
        # 失敗時の処理
        await handle_error(webhook.id, webhook.error)

    return {"status": "received"}

async def process_output(prediction_id: str, output: Any):
    """出力を処理"""
    print(f"Output for {prediction_id}: {output}")

async def handle_error(prediction_id: str, error: str):
    """エラーを処理"""
    print(f"Error for {prediction_id}: {error}")

Webhook URLのセキュリティ

Webhookを安全に運用するためのベストプラクティス：

1. 署名の検証

Replicateは各webhookリクエストに署名を付けます。これを検証することで、リクエストが本物であることを確認できます。

import hmac
import hashlib

def verify_webhook_signature(body: bytes, signature: str, secret: str) -> bool:
    """Webhook署名を検証"""
    expected = hmac.new(
        secret.encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(f"sha256={expected}", signature)

# Flaskでの使用例
@app.route("/webhook/replicate", methods=["POST"])
def replicate_webhook():
    signature = request.headers.get("Webhook-Signature", "")

    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401

    # ... 処理を続行

2. HTTPSを使用

Webhook URLは必ずHTTPSを使用してください。

3. 予測IDの検証

受け取った予測IDが、自分が作成したものであることを確認してください。

ローカル開発でのテスト

ローカル環境でWebhookをテストするには、ngrokやlocaltunnelを使用します：

# ngrokをインストール
brew install ngrok  # macOS

# ローカルサーバーを公開
ngrok http 5000

# 表示されたURLをWebhook URLとして使用
# 例: https://abc123.ngrok.io/webhook/replicate

エラーハンドリングとトラブルシューティング

Replicate APIを使用する際に遭遇する可能性のあるエラーと、その対処法をまとめます。

一般的なエラーと対処法

1. 認証エラー (401 Unauthorized)

症状

replicate.exceptions.ReplicateError: Unauthorized

原因

• APIトークンが設定されていない
• APIトークンが無効または失効している

対処法

import os

# 環境変数を確認
token = os.environ.get("REPLICATE_API_TOKEN")
if not token:
    print("Error: REPLICATE_API_TOKEN is not set")
else:
    print(f"Token starts with: {token[:10]}...")

# 明示的にトークンを設定
import replicate
client = replicate.Client(api_token="r8_xxxxxx")

2. モデルが見つからない (404 Not Found)

症状

replicate.exceptions.ModelError: Model not found

原因

• モデル名のスペルミス
• モデルが削除または非公開になった

対処法

# 正しいモデル名を確認
# https://replicate.com で検索

# バージョンを指定する場合は正確なハッシュを使用
output = replicate.run(
    "black-forest-labs/flux-schnell:5599ed30703defd1d160a25a63321b4dec97101d98b4674bcc56e41f62f35637",
    input={"prompt": "test"}
)

3. 入力パラメータエラー (422 Unprocessable Entity)

症状

replicate.exceptions.ModelError: Invalid input

原因

• 必須パラメータが欠けている
• パラメータの型や値が不正

対処法

# モデルのスキーマを確認
model = replicate.models.get("black-forest-labs/flux-schnell")
print(model.latest_version.openapi_schema["components"]["schemas"]["Input"])

# 正しいパラメータを使用
output = replicate.run(
    "black-forest-labs/flux-schnell",
    input={
        "prompt": "test",  # 必須
        "num_outputs": 1,  # 整数
        "aspect_ratio": "1:1"  # 定められた値
    }
)

4. レート制限 (429 Too Many Requests)

症状

replicate.exceptions.ReplicateError: Rate limit exceeded

原因

• 短時間に大量のリクエストを送信した

対処法

import time
from functools import wraps

def rate_limit_retry(max_retries=3, delay=5):
    """レート制限時にリトライするデコレータ"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except replicate.exceptions.ReplicateError as e:
                    if "rate limit" in str(e).lower() and attempt < max_retries - 1:
                        print(f"Rate limited. Waiting {delay} seconds...")
                        time.sleep(delay * (attempt + 1))  # 指数バックオフ
                    else:
                        raise
        return wrapper
    return decorator

@rate_limit_retry(max_retries=3, delay=5)
def generate_image(prompt):
    return replicate.run(
        "black-forest-labs/flux-schnell",
        input={"prompt": prompt}
    )

5. タイムアウト

症状

• 処理が長時間完了しない

対処法

import replicate
from concurrent.futures import TimeoutError
import asyncio

async def generate_with_timeout(prompt, timeout_seconds=120):
    """タイムアウト付きで画像を生成"""

    prediction = replicate.predictions.create(
        model="black-forest-labs/flux-dev",
        input={"prompt": prompt}
    )

    start_time = asyncio.get_event_loop().time()

    while prediction.status not in ["succeeded", "failed", "canceled"]:
        if asyncio.get_event_loop().time() - start_time > timeout_seconds:
            # キャンセル
            replicate.predictions.cancel(prediction.id)
            raise TimeoutError(f"Prediction timed out after {timeout_seconds}s")

        await asyncio.sleep(2)
        prediction = replicate.predictions.get(prediction.id)

    if prediction.status == "succeeded":
        return prediction.output
    else:
        raise Exception(f"Prediction failed: {prediction.error}")

6. GPU不足 (503 Service Unavailable)

症状

replicate.exceptions.ReplicateError: No available machines

原因

• 人気モデルでGPUが一時的に不足している

対処法

import time
import random

def generate_with_retry(prompt, max_retries=5):
    """リトライ付きで画像を生成"""

    for attempt in range(max_retries):
        try:
            return replicate.run(
                "black-forest-labs/flux-schnell",
                input={"prompt": prompt}
            )
        except replicate.exceptions.ReplicateError as e:
            if "503" in str(e) or "no available" in str(e).lower():
                # 指数バックオフ + ジッター
                wait_time = (2 <strong> attempt) + random.uniform(0, 1)
                print(f"GPU unavailable. Retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
            else:
                raise

    raise Exception("Max retries exceeded")

デバッグのコツ

1. ログを有効にする

import logging
import replicate

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("replicate")
logger.setLevel(logging.DEBUG)

# これでAPIリクエスト/レスポンスの詳細が表示される

2. 予測の詳細を確認

prediction = replicate.predictions.create(
    model="black-forest-labs/flux-schnell",
    input={"prompt": "test"}
)

# 詳細情報を表示
print(f"ID: {prediction.id}")
print(f"Status: {prediction.status}")
print(f"Created at: {prediction.created_at}")
print(f"URLs: {prediction.urls}")
print(f"Metrics: {prediction.metrics}")  # 実行時間など

3. エラーの詳細を取得**

try:
    output = replicate.run(...)
except replicate.exceptions.ReplicateError as e:
    print(f"Error type: {type(e).__name__}")
    print(f"Error message: {e}")
    print(f"Error args: {e.args}")

よくある問題と解決策

ベストプラクティス

Replicate APIを効果的に活用するためのベストプラクティスをまとめます。

1. APIキーの安全な管理

# ❌ 悪い例：コードにハードコード
api_token = "r8_xxxxxxxxxxxx"

# ✓ 良い例：環境変数を使用
import os
api_token = os.environ.get("REPLICATE_API_TOKEN")

# ✓ より良い例：設定ファイルを使用
from dotenv import load_dotenv
load_dotenv()
api_token = os.environ.get("REPLICATE_API_TOKEN")

2. 適切なモデル選択

# 用途に応じたモデル選択
MODEL_CONFIGS = {
    # 高速・低コスト（プロトタイプ、大量生成）
    "fast": "black-forest-labs/flux-schnell",

    # バランス（一般的な用途）
    "balanced": "black-forest-labs/flux-dev",

    # 高品質（商用、最終成果物）
    "quality": "black-forest-labs/flux-pro",
}

def generate_image(prompt, quality="balanced"):
    model = MODEL_CONFIGS.get(quality, MODEL_CONFIGS["balanced"])
    return replicate.run(model, input={"prompt": prompt})

3. エラーハンドリングの実装

import replicate
from typing import Optional
import logging

logger = logging.getLogger(__name__)

class ReplicateService:
    """Replicate APIのラッパークラス"""

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries

    def generate_image(
        self,
        prompt: str,
        model: str = "black-forest-labs/flux-schnell"
    ) -> Optional[str]:
        """
        エラーハンドリング付きで画像を生成

        Returns:
            成功時は画像URL、失敗時はNone
        """
        for attempt in range(self.max_retries):
            try:
                output = replicate.run(
                    model,
                    input={"prompt": prompt}
                )
                return output[0] if output else None

            except replicate.exceptions.ReplicateError as e:
                logger.warning(f"Attempt {attempt + 1} failed: {e}")
                if attempt == self.max_retries - 1:
                    logger.error(f"All attempts failed for prompt: {prompt[:50]}...")
                    return None

            except Exception as e:
                logger.error(f"Unexpected error: {e}")
                return None

        return None

4. コスト管理

from dataclasses import dataclass
from typing import Dict

@dataclass
class CostEstimate:
    model: str
    estimated_cost: float
    currency: str = "USD"

class CostTracker:
    """API使用コストを追跡"""

    # モデルごとの概算コスト（1回あたり）
    COST_PER_RUN: Dict[str, float] = {
        "black-forest-labs/flux-schnell": 0.003,
        "black-forest-labs/flux-dev": 0.025,
        "black-forest-labs/flux-pro": 0.055,
        "stability-ai/sdxl": 0.008,
        "openai/whisper": 0.006,  # 1分あたり
    }

    def __init__(self, budget_limit: float = 10.0):
        self.budget_limit = budget_limit
        self.total_cost = 0.0
        self.run_count = 0

    def estimate_cost(self, model: str, runs: int = 1) -> CostEstimate:
        """コストを見積もる"""
        cost_per_run = self.COST_PER_RUN.get(model, 0.01)
        return CostEstimate(
            model=model,
            estimated_cost=cost_per_run * runs
        )

    def track_run(self, model: str):
        """実行をトラッキング"""
        cost = self.COST_PER_RUN.get(model, 0.01)
        self.total_cost += cost
        self.run_count += 1

        if self.total_cost >= self.budget_limit * 0.8:
            print(f"Warning: Approaching budget limit ({self.total_cost:.2f}/{self.budget_limit:.2f})")

    def can_run(self, model: str) -> bool:
        """予算内で実行可能か確認"""
        estimated = self.estimate_cost(model)
        return (self.total_cost + estimated.estimated_cost) <= self.budget_limit

# 使用例
tracker = CostTracker(budget_limit=5.0)

if tracker.can_run("black-forest-labs/flux-dev"):
    output = replicate.run("black-forest-labs/flux-dev", input={"prompt": "test"})
    tracker.track_run("black-forest-labs/flux-dev")
    print(f"Total cost so far: ${tracker.total_cost:.3f}")
else:
    print("Budget limit reached!")

5. キャッシュの活用

import hashlib
import json
import os
from typing import Optional, Dict, Any

class ResultCache:
    """生成結果をキャッシュ"""

    def __init__(self, cache_dir: str = "./.replicate_cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def _get_cache_key(self, model: str, input_params: Dict[str, Any]) -> str:
        """キャッシュキーを生成"""
        content = json.dumps({"model": model, "input": input_params}, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def get(self, model: str, input_params: Dict[str, Any]) -> Optional[str]:
        """キャッシュから取得"""
        key = self._get_cache_key(model, input_params)
        cache_file = os.path.join(self.cache_dir, f"{key}.json")

        if os.path.exists(cache_file):
            with open(cache_file) as f:
                data = json.load(f)
                print(f"Cache hit: {key}")
                return data.get("output")
        return None

    def set(self, model: str, input_params: Dict[str, Any], output: Any):
        """キャッシュに保存"""
        key = self._get_cache_key(model, input_params)
        cache_file = os.path.join(self.cache_dir, f"{key}.json")

        with open(cache_file, "w") as f:
            json.dump({"model": model, "input": input_params, "output": output}, f)
        print(f"Cached: {key}")

# 使用例
cache = ResultCache()

def generate_with_cache(prompt: str, model: str = "black-forest-labs/flux-schnell"):
    input_params = {"prompt": prompt}

    # キャッシュを確認
    cached = cache.get(model, input_params)
    if cached:
        return cached

    # 新規生成
    output = replicate.run(model, input=input_params)

    # キャッシュに保存
    cache.set(model, input_params, output)

    return output

6. プロンプトエンジニアリング

class PromptBuilder:
    """効果的なプロンプトを構築"""

    @staticmethod
    def build_image_prompt(
        subject: str,
        style: str = "",
        details: str = "",
        quality_tags: bool = True,
        negative: bool = False
    ) -> str:
        """
        構造化されたプロンプトを生成

        Args:
            subject: 主題
            style: スタイル（アニメ、写真風など）
            details: 追加の詳細
            quality_tags: 品質向上タグを追加
            negative: ネガティブプロンプト用
        """
        parts = [subject]

        if style:
            parts.append(style)

        if details:
            parts.append(details)

        if quality_tags and not negative:
            parts.extend([
                "high quality",
                "detailed",
                "professional"
            ])

        return ", ".join(parts)

    @staticmethod
    def build_negative_prompt(exclude_list: list = None) -> str:
        """ネガティブプロンプトを生成"""
        defaults = [
            "low quality",
            "blurry",
            "distorted",
            "ugly",
            "bad anatomy",
            "watermark",
            "text"
        ]

        if exclude_list:
            defaults.extend(exclude_list)

        return ", ".join(defaults)

# 使用例
builder = PromptBuilder()

prompt = builder.build_image_prompt(
    subject="A majestic eagle soaring over mountains",
    style="photorealistic, wildlife photography",
    details="golden hour lighting, dramatic clouds"
)

negative = builder.build_negative_prompt(["cartoon", "illustration"])

print(f"Prompt: {prompt}")
print(f"Negative: {negative}")

他のサービスとの比較

Replicate以外にもAI APIサービスは多数あります。主要なサービスとの比較を紹介します。

画像生成API比較

Replicate vs 他サービスの詳細比較

Replicate vs OpenAI (DALL-E 3)

Replicate vs Stability AI API

Replicate vs RunPod

用途別おすすめ

発展的な使い方

カスタムモデルのデプロイ

自分のモデルをReplicateにデプロイすることもできます：

# Cogをインストール
pip install cog

# 予測スクリプトを作成（predict.py）
# Dockerfile的な設定（cog.yaml）を作成

# ローカルでテスト
cog predict -i prompt="test"

# Replicateにプッシュ
cog push r8.im/your-username/your-model

cog.yaml の例

build:
  python_version: "3.10"
  python_packages:
    - torch==2.0.1
    - diffusers==0.21.0
    - transformers

predict: "predict.py:Predictor"

predict.py の例

from cog import BasePredictor, Input, Path
import torch

class Predictor(BasePredictor):
    def setup(self):
        """モデルをロード"""
        self.model = load_your_model()

    def predict(
        self,
        prompt: str = Input(description="Input prompt"),
        steps: int = Input(default=30, ge=1, le=100),
    ) -> Path:
        """予測を実行"""
        output = self.model.generate(prompt, steps)
        output_path = "/tmp/output.png"
        output.save(output_path)
        return Path(output_path)

Trainings API（ファインチューニング）

一部のモデルでは、カスタムデータでファインチューニングが可能です：

import replicate

# トレーニングを開始
training = replicate.trainings.create(
    model="stability-ai/sdxl",
    version="latest",
    destination="your-username/custom-sdxl",
    input={
        "input_images": "https://your-bucket.s3.amazonaws.com/training-data.zip",
        "token_string": "TOK",
        "caption_prefix": "a photo of TOK, ",
        "max_train_steps": 1000
    }
)

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

# トレーニング完了を待つ
import time
while training.status not in ["succeeded", "failed"]:
    time.sleep(60)
    training = replicate.trainings.get(training.id)
    print(f"Status: {training.status}")

Deployments（専用インスタンス）

高負荷アプリケーション向けに、専用インスタンスをデプロイできます：

# デプロイメントを作成
deployment = replicate.deployments.create(
    model="black-forest-labs/flux-schnell",
    name="my-flux-deployment",
    hardware="gpu-a40-small",
    min_instances=1,
    max_instances=5
)

# デプロイメントを使って予測
output = replicate.deployments.predictions.create(
    deployment="your-username/my-flux-deployment",
    input={"prompt": "A beautiful sunset"}
)

よくある質問（FAQ）

セキュリティとコンプライアンス

APIキーのセキュリティ

APIキーを安全に管理するためのベストプラクティス：

1. 環境変数を使用

# ❌ 悪い例
api_key = "r8_xxxxxxxxxxxxxxxx"

# ✓ 良い例
import os
api_key = os.environ.get("REPLICATE_API_TOKEN")

2. .gitignoreに追加

# .gitignore
.env
.env.local
*.env

3. シークレット管理サービスを使用

• AWS Secrets Manager
• Google Secret Manager
• HashiCorp Vault
• Azure Key Vault

4. ローテーションの実施 定期的にAPIキーをローテーションし、古いキーを無効化します。

データプライバシー

Replicateを使用する際のデータプライバシーに関する考慮事項：

• 入力データの取り扱い: Replicateは入力データをモデルの実行に使用しますが、永続的に保存することは通常ありません
• 出力データの保存: 生成された出力は一定期間Replicateのサーバーに保存されます
• ログ: APIリクエストのログが記録される場合があります

機密性の高いデータを扱う場合は、Replicateのプライバシーポリシーを確認し、必要に応じて独自のインフラでモデルをホストすることを検討してください。

コンプライアンス

Replicateは以下のコンプライアンス基準を満たしています：

• SOC 2 Type II: セキュリティ、可用性、処理の完全性に関する監査
• GDPR: EU一般データ保護規則への対応

特定の業界（医療、金融など）でReplicateを使用する場合は、追加のコンプライアンス要件を確認してください。

まとめ

画像クレジット

本記事はReplicate APIの解説記事であり、特定の画像素材は使用していません。

※出典：Replicate公式ドキュメント

※出典：Replicate API Reference