实时端点 Real-time endpoints & WebSockets

对于在 fal 运行时上部署的应用程序；除了常规 HTTP 端点之外，开发人员还可以选择在原始 WebSockets 或 fal 的（无状态）实时应用程序框架之上实现辅助接口。

在 fal.App 下，对于处理实时连接的任何端点，可以使用 @fal.realtime() 装饰器代替 @fal.endpoint，以自动使接口与 fal 的实时客户端兼容。这些函数不提供任何会话状态，旨在用于减少总体延迟（使用 fal 的二进制协议）并消除固定连接建立开销。

对于想要使用自己的实时协议构建有状态应用程序的高级用户，可以使用 is_websocket=True 标志初始化 @fal.endpoint，底层函数将接收原始 WebSocket 连接并可以选择以任何方式使用它。

py
import fal
from pydantic import BaseModel

class Input(BaseModel):
    prompt: str = Field()

class Output(BaseModel):
    output: str = Field()

class RealtimeApp(fal.App):
    @fal.endpoint("/")
    def generate(self, input: Input) -> Output:
        return Output(output=input.prompt)

    @fal.endpoint("/ws", is_websocket=True)
    async def generate_ws(self, websocket: WebSocket) -> None:
        await websocket.accept()
        for _ in range(3):
            await websocket.send_json({"message": "Hello world!"})
        await websocket.close()

    @fal.realtime("/realtime")
    def generate_rt(self, input: Input) -> Output:
        return Output(output=input.prompt)