当同一应用程序有多个可用副本时，没有定义行为来选择使用哪个副本来处理特定请求，假设所有副本对于给定的一组输入都会表现相同。
对于保存状态并可能包含某些参数集的内存缓存的应用程序来说，这可能并非如此。例如，如果您正在为一个可以运行任何扩散模型但只能在内存中保留 3 个不同模型的应用程序提供服务，那么您希望根据用户提供的输入最小化缓存未命中次数（因为从头开始加载该模型会导致严重的性能损失）。
这就是语义感知路由提示发挥作用的地方。应用程序可以提供专业化的提示，并允许 fal 的路由器为特定请求选择合适的副本，而不是平等对待每个副本。为了使此逻辑有效工作，用户需要提供带有语义识别字符串提示的 X-Fal-Runner-Hint 标头，并且应用程序应实现返回提示列表的 provide_hints() 方法。如果任何副本中有匹配项，则 fal 的路由器会将请求发送到该副本。但是，如果没有匹配项，它将回退到标准路由算法。
py
from typing import Any
import fal
from fal.toolkit import Image
from pydantic import BaseModel
class Input(BaseModel):
model: str = Field()
prompt: str = Field()
class Output(BaseModel):
image: Image = Field()
class AnyModelRunner(fal.App):
def setup(self) -> None:
self.models = {}
def provide_hints(self) -> list[str]:
# Choose to specialize on already loaded models; at first this will be empty
# so we'll be picked for any request, but as slowly the cache builds up, the
# replica will be more preferable compared to others.
return self.models.keys()
def load_model(self, name: str) -> Any:
from diffusers import DiffusionPipeline
if name in self.models:
return self.models[name]
pipeline = DiffusionPipeline.from_pretrained(name)
pipeline.to("cuda")
self.models[name] = pipeline
return pipeline
@fal.endpoint("/")
def run_model(self, input: Input) -> Output:
model = self.load_model(input.model)
result = model(input.prompt)
return Output(image=Image.from_pil(result.images[0]))
最后更新于 2024 年 6 月 21 日