fal-serverless-guide

快速参考

机器类型	GPU	显存	用途
GPU-T4	T4	16GB	开发、小型模型
GPU-A10G	A10G	24GB	7B-13B 模型
GPU-A100	A100	40/80GB	13B-70B 模型
GPU-H100	H100	80GB	前沿模型

应用属性	用途	示例
machine_type	GPU 选择	"GPU-A100"
requirements	依赖项	["torch", "transformers"]
keep_alive	保持运行时长	300（5 分钟）
min_concurrency	最小实例数	0（缩容至零）
max_concurrency	最大并行数	4

命令	用途
fal deploy app.py::MyApp	部署到 fal
fal run app.py::MyApp	在本地运行
fal logs <app-id>	查看日志
fal secrets set KEY=value	设置密钥

何时使用此技能

适用于自定义模型部署：

• 在 fal 基础设施上部署自定义 ML 模型
• 配置 GPU 实例和自动伸缩
• 为模型权重设置持久化存储
• 创建多端点应用
• 管理密钥和环境变量

相关技能：

• API 集成：请参阅 fal-api-reference
• 优化：请参阅 fal-optimization
• 使用托管模型：请参阅 fal-model-guide

---

fal.ai 无服务器部署指南

在 fal.ai 的无服务器基础设施上部署自定义 ML 模型的完整指南。

概述

fal 无服务器服务提供：

• 从零自动扩展到数千个实例
• GPU 支持（T4、A10G、A100、H100、H200、B200）
• 模型权重的持久化存储
• 密钥管理
• 实时日志和监控
• 按使用量计费

安装

pip install fal

身份验证

# Login to fal
fal auth login

# Or set API key
export FAL_KEY="your-api-key"

基本应用结构

import fal
from pydantic import BaseModel

class RequestModel(BaseModel):
    """Input schema for your endpoint"""
    prompt: str
    max_tokens: int = 100

class ResponseModel(BaseModel):
    """Output schema for your endpoint"""
    text: str
    tokens: int

class MyApp(fal.App):
    # Machine configuration
    machine_type = "GPU-A100"
    num_gpus = 1

    # Dependencies
    requirements = [
        "torch>=2.0.0",
        "transformers>=4.35.0",
        "accelerate"
    ]

    # Scaling configuration
    keep_alive = 300        # Keep instance warm (seconds)
    min_concurrency = 0     # Scale to zero when idle
    max_concurrency = 4     # Max concurrent requests

    def setup(self):
        """
        Called once when container starts.
        Load models and heavy resources here.
        """
        import torch
        from transformers import AutoModelForCausalLM, AutoTokenizer

        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.tokenizer = AutoTokenizer.from_pretrained("model-name")
        self.model = AutoModelForCausalLM.from_pretrained(
            "model-name",
            torch_dtype=torch.float16
        ).to(self.device)

    @fal.endpoint("/predict")
    def predict(self, request: RequestModel) -> ResponseModel:
        """
        Main inference endpoint.
        Called for each request.
        """
        inputs = self.tokenizer(request.prompt, return_tensors="pt")
        inputs = inputs.to(self.device)

        outputs = self.model.generate(
            **inputs,
            max_new_tokens=request.max_tokens
        )

        text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        return ResponseModel(text=text, tokens=len(outputs[0]))

    @fal.endpoint("/health")
    def health(self):
        """Health check endpoint"""
        return {"status": "healthy", "device": self.device}

    def teardown(self):
        """Called when container shuts down (optional)"""
        if hasattr(self, 'model'):
            del self.model
        import torch
        torch.cuda.empty_cache()

机器类型

类型	GPU	显存	用途
CPU	无	-	预处理、轻量级任务
GPU-T4	NVIDIA T4	16GB	开发、小型模型
GPU-A10G	NVIDIA A10G	24GB	中型模型（7B-13B）
GPU-A100	NVIDIA A100	40/80GB	大型模型（13B-70B）
GPU-H100	NVIDIA H100	80GB	前沿性能
GPU-H200	NVIDIA H200	141GB	超大型模型
GPU-B200	NVIDIA B200	192GB	前沿模型（100B+）

多 GPU 配置

class MultiGPUApp(fal.App):
    machine_type = "GPU-H100"
    num_gpus = 4  # Use 4 H100s

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

        self.model = AutoModelForCausalLM.from_pretrained(
            "meta-llama/Llama-2-70b-hf",
            torch_dtype=torch.float16,
            device_map="auto"  # Distribute across GPUs
        )

持久化存储

使用卷在重启后保留数据：

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

    # Define persistent volumes
    volumes = {
        "/data": fal.Volume("model-cache"),
        "/outputs": fal.Volume("generated-outputs")
    }

    def setup(self):
        import os
        from transformers import AutoModel

        cache_dir = "/data/models"
        os.makedirs(cache_dir, exist_ok=True)

        # Model weights persist across cold starts
        self.model = AutoModel.from_pretrained(
            "large-model",
            cache_dir=cache_dir
        )

    @fal.endpoint("/generate")
    def generate(self, request):
        output_path = "/outputs/result.png"
        # Save to persistent storage
        return {"path": output_path}

密钥管理

# Set secrets via CLI
fal secrets set HF_TOKEN=hf_xxx API_KEY=sk_xxx

# List secrets
fal secrets list

# Delete secret
fal secrets delete HF_TOKEN

import os

class SecureApp(fal.App):
    def setup(self):
        # Access secrets as environment variables
        hf_token = os.environ["HF_TOKEN"]

        from huggingface_hub import login
        login(token=hf_token)

        # Now can access gated models
        self.model = load_gated_model()

部署命令

# Deploy application
fal deploy app.py::MyApp

# Deploy with options
fal deploy app.py::MyApp \
  --machine-type GPU-A100 \
  --num-gpus 2 \
  --min-concurrency 1 \
  --max-concurrency 8

# View deployments
fal list

# View logs
fal logs <app-id>

# View real-time logs
fal logs <app-id> --follow

# Delete deployment
fal delete <app-id>

# Run locally for testing
fal run app.py::MyApp

高级模式

图像生成应用

import fal
from pydantic import BaseModel
from typing import Optional
import io

class ImageRequest(BaseModel):
    prompt: str
    negative_prompt: Optional[str] = None
    width: int = 1024
    height: int = 1024
    steps: int = 28
    seed: Optional[int] = None

class ImageResponse(BaseModel):
    image_url: str
    seed: int

class ImageGenerator(fal.App):
    machine_type = "GPU-A100"
    requirements = [
        "torch",
        "diffusers",
        "transformers",
        "accelerate",
        "safetensors"
    ]
    keep_alive = 600
    max_concurrency = 2

    volumes = {
        "/data": fal.Volume("diffusion-models")
    }

    def setup(self):
        import torch
        from diffusers import StableDiffusionXLPipeline

        self.pipe = StableDiffusionXLPipeline.from_pretrained(
            "stabilityai/stable-diffusion-xl-base-1.0",
            torch_dtype=torch.float16,
            cache_dir="/data/models"
        ).to("cuda")

        # Optimize
        self.pipe.enable_model_cpu_offload()

    @fal.endpoint("/generate")
    def generate(self, request: ImageRequest) -> ImageResponse:
        import torch
        import random

        seed = request.seed or random.randint(0, 2**32 - 1)
        generator = torch.Generator("cuda").manual_seed(seed)

        image = self.pipe(
            prompt=request.prompt,
            negative_prompt=request.negative_prompt,
            width=request.width,
            height=request.height,
            num_inference_steps=request.steps,
            generator=generator
        ).images[0]

        # Save and upload to CDN
        path = f"/tmp/output_{seed}.png"
        image.save(path)
        url = fal.upload_file(path)

        return ImageResponse(image_url=url, seed=seed)

流式响应

import fal
from typing import Generator

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

    def setup(self):
        from transformers import AutoModelForCausalLM, AutoTokenizer
        self.tokenizer = AutoTokenizer.from_pretrained("model")
        self.model = AutoModelForCausalLM.from_pretrained("model")

    @fal.endpoint("/stream")
    def stream(self, prompt: str) -> Generator[str, None, None]:
        from transformers import TextIteratorStreamer
        from threading import Thread

        inputs = self.tokenizer(prompt, return_tensors="pt").to("cuda")
        streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True)

        thread = Thread(
            target=self.model.generate,
            kwargs={**inputs, "streamer": streamer, "max_new_tokens": 256}
        )
        thread.start()

        for text in streamer:
            yield text

后台任务

import fal
from typing import Optional
import asyncio

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

    @fal.endpoint("/process")
    async def process(self, data: str) -> dict:
        # Submit background work
        task_id = await self.start_background_task(data)
        return {"task_id": task_id, "status": "processing"}

    async def start_background_task(self, data: str) -> str:
        # Implement your background logic
        import uuid
        task_id = str(uuid.uuid4())
        # Save task to queue/database
        return task_id

多个端点

import fal
from pydantic import BaseModel

class TextRequest(BaseModel):
    text: str

class ImageRequest(BaseModel):
    image_url: str

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

    def setup(self):
        self.text_model = self.load_text_model()
        self.vision_model = self.load_vision_model()

    @fal.endpoint("/analyze-text")
    def analyze_text(self, request: TextRequest) -> dict:
        result = self.text_model(request.text)
        return {"analysis": result}

    @fal.endpoint("/analyze-image")
    def analyze_image(self, request: ImageRequest) -> dict:
        result = self.vision_model(request.image_url)
        return {"analysis": result}

    @fal.endpoint("/")
    def info(self) -> dict:
        return {
            "name": "MultiModal Analyzer",
            "endpoints": ["/analyze-text", "/analyze-image"]
        }

    @fal.endpoint("/health")
    def health(self) -> dict:
        return {"status": "healthy"}

自动伸缩配置

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

    # Scaling options
    min_concurrency = 0     # Scale to zero (cost savings)
    max_concurrency = 10    # Max parallel requests
    keep_alive = 300        # Keep warm for 5 minutes

    # For always-on endpoints
    # min_concurrency = 1   # Always have one instance ready

并发指南

每个请求的 GPU 显存	建议的 max_concurrency
< 4GB	8-10
4-8GB	4-6
8-16GB	2-4
16-40GB	1-2
> 40GB	1

错误处理

import fal
from pydantic import BaseModel

class MyApp(fal.App):
    @fal.endpoint("/predict")
    def predict(self, request: dict):
        try:
            result = self.process(request)
            return {"result": result}
        except ValueError as e:
            # Client error
            raise fal.HTTPException(400, f"Invalid input: {e}")
        except RuntimeError as e:
            # Server error
            raise fal.HTTPException(500, f"Processing failed: {e}")
        except Exception as e:
            # Unexpected error
            raise fal.HTTPException(500, "Internal error")

本地开发

# Run locally
fal run app.py::MyApp

# Test endpoint
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"prompt": "test"}'

# Run with environment variables
FAL_KEY=xxx HF_TOKEN=yyy fal run app.py::MyApp

调用已部署的应用

JavaScript/TypeScript

import { fal } from "@fal-ai/client";

fal.config({ credentials: process.env.FAL_KEY });

const result = await fal.subscribe("your-username/your-app/predict", {
  input: {
    prompt: "Hello world",
    max_tokens: 100
  }
});

Python

import fal_client

result = fal_client.subscribe(
    "your-username/your-app/predict",
    arguments={
        "prompt": "Hello world",
        "max_tokens": 100
    }
)

REST API

curl -X POST "https://queue.fal.run/your-username/your-app/predict" \
  -H "Authorization: Key $FAL_KEY" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Hello world", "max_tokens": 100}'

最佳实践

1. 在 setup() 中加载模型

   • 仅执行一次繁重的初始化，而不是每次请求都执行
   • 对大型权重使用持久化卷

2. 使用合适的机器类型

   • 根据模型大小匹配 GPU 显存
   • 权衡成本与性能

3. 处理冷启动

   • 对频繁访问的端点使用 keep_alive
   • 对延迟敏感型应用使用 min_concurrency=1

4. 优化内存

   • 尽可能使用 fp16/bf16
   • 启用节省内存的注意力机制
   • 在 teardown 中清理 GPU 缓存

5. 监控和调试

   • 定期检查日志：fal logs <app-id> --follow
   • 实现运行状况检查
   • 使用结构化日志记录

6. 安全

   • 使用密钥存储 API 密钥
   • 验证所有输入
   • 不要暴露内部错误

定价

• 按实际使用的计算时长（秒）计费
• 不同 GPU 类型的费率不同
• 缩容至零时不收费
• 请访问 https://fal.ai/pricing 查看当前费率