fal-optimization

クイックリファレンス

最適化	手法	効果
並列リクエスト	バッチ処理での Promise.all()	スループットが5～10倍
ポーリングの回避	Webhookを使用	API呼び出しを削減
シードによるキャッシュ	prompt+seed の結果を保存	再生成を回避
適切な画像サイズ	必要な解像度を使用	コストを削減
ステップ数の削減	推論ステップ数を減らす	高速化、低コスト化

モデル階層	開発	本番環境
画像	FLUX Schnell	FLUX.2 Pro
動画	Runway Turbo	Kling 2.6 Pro

サーバーレス設定	コスト最適化	レイテンシ最適化
min_concurrency	0	1+
keep_alive	120	600+
machine_type	実行可能な最小サイズ	上位ティア

このスキルを使う場面

パフォーマンスとコストの最適化に使用します：

• 生成レイテンシの短縮
• APIコストの削減
• 並列処理の実装
• ポーリングとWebhookの選択
• サーバーレスのスケーリング設定

関連スキル：

• APIのパターンについては、fal-api-referenceを参照してください
• モデルの選択については、fal-model-guideを参照してください
• サーバーレス設定については、fal-serverless-guideを参照してください

---

fal.aiのパフォーマンスとコストの最適化

fal.aiの統合のパフォーマンスを最適化し、コストを削減し、スケールするための戦略です。

パフォーマンスの最適化

クライアント側の最適化

1. キューベースの実行を使用する

生成タスクでは、常に run() より subscribe() を優先してください：

// Recommended: Queue-based with progress tracking
const result = await fal.subscribe("fal-ai/flux/dev", {
  input: { prompt: "test" },
  logs: true,
  onQueueUpdate: (update) => {
    // Show progress to users
    if (update.status === "IN_PROGRESS") {
      console.log("Generating...");
    }
  }
});

// Only use run() for fast endpoints (< 30s)
const quickResult = await fal.run("fal-ai/fast-sdxl", {
  input: { prompt: "quick test" }
});

2. 並列リクエスト

複数のリクエストを同時に処理します：

// JavaScript - Parallel execution
async function generateBatch(prompts: string[]) {
  const results = await Promise.all(
    prompts.map(prompt =>
      fal.subscribe("fal-ai/flux/dev", {
        input: { prompt }
      })
    )
  );
  return results;
}

// With rate limiting
async function generateBatchWithLimit(prompts: string[], limit = 5) {
  const results = [];
  for (let i = 0; i < prompts.length; i += limit) {
    const batch = prompts.slice(i, i + limit);
    const batchResults = await Promise.all(
      batch.map(prompt =>
        fal.subscribe("fal-ai/flux/dev", { input: { prompt } })
      )
    );
    results.push(...batchResults);
    // Small delay between batches
    if (i + limit < prompts.length) {
      await new Promise(r => setTimeout(r, 100));
    }
  }
  return results;
}

# Python - Async parallel
import asyncio
import fal_client

async def generate_batch(prompts: list[str]) -> list[dict]:
    tasks = [
        fal_client.run_async("fal-ai/flux/dev", arguments={"prompt": p})
        for p in prompts
    ]
    return await asyncio.gather(*tasks)

# With semaphore for rate limiting
async def generate_batch_limited(prompts: list[str], limit: int = 5):
    semaphore = asyncio.Semaphore(limit)

    async def generate_one(prompt: str):
        async with semaphore:
            return await fal_client.run_async(
                "fal-ai/flux/dev",
                arguments={"prompt": prompt}
            )

    return await asyncio.gather(*[generate_one(p) for p in prompts])

3. リアルタイムのフィードバックにはストリーミングを使用する

段階的な出力にはストリーミングを使用します：

// Show incremental progress
const stream = await fal.stream("fal-ai/flux/dev", {
  input: { prompt: "A landscape" }
});

for await (const event of stream) {
  updateProgressUI(event);
}

const result = await stream.done();

4. インタラクティブアプリにはWebSocketを使用する

継続的な入力があるリアルタイムアプリケーションの場合：

const connection = fal.realtime.connect("fal-ai/lcm-sd15-i2i", {
  connectionKey: `user-${userId}`,
  throttleInterval: 128,  // Debounce rapid inputs
  onResult: (result) => {
    displayImage(result.images[0].url);
  }
});

// Send updates as user types/draws
inputElement.addEventListener('input', (e) => {
  connection.send({
    prompt: e.target.value,
    image_url: currentImage
  });
});

サーバー側の最適化（サーバーレス）

1. モデルを効率よく読み込む

class OptimizedApp(fal.App):
    machine_type = "GPU-A100"
    requirements = ["torch", "transformers", "accelerate"]

    volumes = {
        "/data": fal.Volume("model-cache")
    }

    def setup(self):
        import torch
        from transformers import AutoModelForCausalLM

        # Use fp16 for faster inference and less memory
        self.model = AutoModelForCausalLM.from_pretrained(
            "model-name",
            torch_dtype=torch.float16,
            device_map="auto",
            cache_dir="/data/models"  # Persistent cache
        )

        # Enable optimizations
        if hasattr(self.model, 'enable_attention_slicing'):
            self.model.enable_attention_slicing()

2. コールドスタートを減らす

class WarmApp(fal.App):
    machine_type = "GPU-A100"
    keep_alive = 600      # 10 minutes warm
    min_concurrency = 1   # Always keep one ready

    # Use lightweight health check
    @fal.endpoint("/health")
    def health(self):
        return {"status": "ok"}

3. メモリー管理

class MemoryEfficientApp(fal.App):
    def setup(self):
        import torch

        # Use mixed precision
        self.model = load_model(torch_dtype=torch.float16)

        # Enable memory-efficient attention (if using transformers)
        self.model.enable_xformers_memory_efficient_attention()

    def teardown(self):
        # Clean up GPU memory
        import torch
        if hasattr(self, 'model'):
            del self.model
        torch.cuda.empty_cache()

    @fal.endpoint("/generate")
    def generate(self, request):
        import torch

        with torch.inference_mode():  # Disable gradient tracking
            result = self.model(request.input)

        return result

コストの最適化

1. 適切なモデルを選ぶ

ニーズ	低コストの選択肢	プレミアムの選択肢
素早い試行	FLUX Schnell ($)	FLUX.1 Dev ($$)
本番環境	FLUX.1 Dev ($$)	FLUX.2 Pro ($$$)
動画プレビュー	Runway Turbo ($$)	Kling Pro ($$$)

// Development: Use fast/cheap models
const preview = await fal.subscribe("fal-ai/flux/schnell", {
  input: { prompt: "test", num_inference_steps: 4 }
});

// Production: Use quality models
const final = await fal.subscribe("fal-ai/flux-2-pro", {
  input: { prompt: "test" }
});

2. 画像サイズを最適化する

大きすぎるサイズではなく、必要なサイズで生成します：

// Don't generate larger than needed
const result = await fal.subscribe("fal-ai/flux/dev", {
  input: {
    prompt: "test",
    // Use preset sizes
    image_size: "square_hd",  // 1024x1024

    // Or specific dimensions
    image_size: { width: 800, height: 600 }
  }
});

3. 推論ステップ数を減らす

許容できる品質を得られる最小のステップ数を見つけます：

// Quick previews: fewer steps
const preview = await fal.subscribe("fal-ai/flux/dev", {
  input: {
    prompt: "test",
    num_inference_steps: 15  // Faster, slightly lower quality
  }
});

// Final render: more steps
const final = await fal.subscribe("fal-ai/flux/dev", {
  input: {
    prompt: "test",
    num_inference_steps: 28  // Default, high quality
  }
});

4. 大量のリクエストにはWebhookを使用する

Webhookを使用してポーリングのオーバーヘッドを回避します：

// Instead of polling
const result = await fal.subscribe("fal-ai/flux/dev", {
  input: { prompt: "test" },
  pollInterval: 1000  // Polling = more API calls
});

// Use webhooks
const { request_id } = await fal.queue.submit("fal-ai/flux/dev", {
  input: { prompt: "test" },
  webhookUrl: "https://your-server.com/webhook"
});
// No polling needed - result delivered to webhook

5. 結果をキャッシュする

再現可能な出力にはシードを使用します：

// Cache key based on prompt + seed
const cacheKey = `${prompt}-${seed}`;
const cached = await cache.get(cacheKey);

if (cached) {
  return cached;
}

const result = await fal.subscribe("fal-ai/flux/dev", {
  input: { prompt, seed }
});

await cache.set(cacheKey, result);
return result;

6. サーバーレスのコストを最適化する

class CostOptimizedApp(fal.App):
    machine_type = "GPU-A10G"   # Cheaper than A100 if sufficient
    min_concurrency = 0         # Scale to zero when not used
    keep_alive = 120            # Shorter keep-alive

    # Use appropriate GPU for model size
    # T4: < 16GB VRAM models
    # A10G: 16-24GB VRAM models
    # A100: 24-80GB VRAM models

スケーリング戦略

1. 水平スケーリング

class ScalableApp(fal.App):
    machine_type = "GPU-A100"
    min_concurrency = 2    # Always have 2 instances
    max_concurrency = 20   # Scale up to 20

    # fal handles auto-scaling based on queue depth

2. リクエストのバッチ処理

class BatchApp(fal.App):
    @fal.endpoint("/batch")
    def batch_generate(self, prompts: list[str]) -> list[dict]:
        # Process multiple prompts in one request
        results = []
        for prompt in prompts:
            result = self.model(prompt)
            results.append(result)
        return results

3. 優先度付きキュー

優先度に応じて異なるエンドポイントを使用します：

class PriorityApp(fal.App):
    machine_type = "GPU-A100"

    @fal.endpoint("/high-priority")
    def high_priority(self, request):
        # Separate endpoint for important requests
        return self.process(request)

    @fal.endpoint("/standard")
    def standard(self, request):
        # Standard processing
        return self.process(request)

モニタリングとデバッグ

1. ログを追加する

import logging

class MonitoredApp(fal.App):
    def setup(self):
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        self.logger.info("App starting up")
        # Load model
        self.logger.info("Model loaded successfully")

    @fal.endpoint("/generate")
    def generate(self, request):
        import time
        start = time.time()

        result = self.process(request)

        elapsed = time.time() - start
        self.logger.info(f"Request processed in {elapsed:.2f}s")

        return result

2. メトリクスを追跡する

// Client-side timing
const start = Date.now();

const result = await fal.subscribe("fal-ai/flux/dev", {
  input: { prompt: "test" },
  onQueueUpdate: (update) => {
    if (update.status === "IN_QUEUE") {
      console.log(`Queue position: ${update.queue_position}`);
    }
  }
});

const elapsed = Date.now() - start;
console.log(`Total time: ${elapsed}ms`);

// Track in your analytics
analytics.track("fal_generation", {
  model: "flux/dev",
  elapsed_ms: elapsed,
  queue_time_ms: result.timings?.queue,
  inference_time_ms: result.timings?.inference
});

3. エラーをモニタリングする

try {
  const result = await fal.subscribe("fal-ai/flux/dev", {
    input: { prompt: "test" }
  });
} catch (error) {
  // Log to error tracking service
  errorTracker.captureException(error, {
    tags: {
      model: "flux/dev",
      type: error.constructor.name
    },
    extra: {
      status: error.status,
      body: error.body
    }
  });

  // Handle gracefully
  return fallbackResult();
}

チェックリスト

本番環境への移行前

• キューベースの実行（subscribe）を使用している
• ユースケースに適したモデルを選択している
• 画像サイズを最適化している
• エラー処理を実装している
• レート制限を設定している
• キャッシュ戦略を定義している

サーバーレスのデプロイ

• モデルのサイズに合ったマシンタイプを選択している
• モデルをリクエストごとではなく setup() で読み込んでいる
• 大規模なモデルには永続ボリュームを使用している
• シークレットを適切に設定している
• ヘルスチェックエンドポイントを設定している
• ロギングを有効にしている

コスト管理

• ゼロへのスケールを有効にしている（min_concurrency = 0）
• 適切な keep_alive の設定を使用している
• 開発には低コストのモデルを使用している
• 可能な場合はバッチ処理を使用している
• ポーリングではなくWebhookコールバックを使用している

モニタリング

• レイテンシを追跡している
• エラー率をモニタリングしている
• コストを追跡している
• キューの深さに関するアラートを設定している