通过Anyscale上的Ray优化分布式训练、数据整理和批量推理流程。使用Python API在数千个节点上扩展现有的AI库，如PyTorch、vLLM、SGLang和XGBoost。复制代码1import ray 2from sentence_transformers import SentenceTransformer 3 4class SentenceTransformerEmbed: 5 def __init__(self, model: str): 6 self.model = SentenceTransformer( 7 model, device="cuda") 8 9 def __call__(self, batch): 10 texts = batch["text"] 11 embeddings = self.model.encode(texts,...) 12 batch["embedding"] = embeddings 13 return batch 14 15# 从对象存储中加载源文档 16ds = ray.data.read_parquet("s3://documents") 17 18# 在16个GPU工作者上并行计算嵌入 19ds = ds.map_batches( 20 SentenceTransformerEmbed, 21 fn_constructor_kwargs={"model": "bge-large-en-v1.5"}, 22 concurrency=16, 23 num_gpus=1, 24 batch_size=512, 25) 26 27# 将嵌入保存到仓库 28ds.write_parquet("s3://warehouse/embeddings/")
