RayData 多模态与AI推理算子

使用 Data-Juicer Mapper 算子进行数据转换

LindormRay 提供 Dataset.datajuicer.map(op_name, **op_kwargs) 方法对接 Data-Juicer Mapper 算子，通过 op_name 指定算子名称，并通过 op_kwargs 配置对应算子的参数。

以下示例中的 Data-Juicer Mapper 算子（默认处理 text 字段）依次清理 HTML 标签、邮箱地址和链接，并为邮箱清理算子指定并发数及 CPU 资源。处理后 id 列保留，text 列更新为清洗后的内容：

import ray

ray.init()

ds = ray.data.from_items(
    [
        {
            "id": 1,
            "text": "<p>Contact admin@example.com for product information.</p>",
        },
        {
            "id": 2,
            "text": "<div>Read the user guide at https://example.com/docs.</div>",
        },
    ]
)

cleaned_ds = (
    ds.datajuicer.map("clean_html_mapper")
    .datajuicer.map("clean_email_mapper", concurrency=2, num_cpus=1)
    .datajuicer.map("clean_links_mapper")
)

print(cleaned_ds.take_all())

使用 Data-Juicer Filter 算子进行数据过滤

LindormRay 提供 Dataset.datajuicer.filter(op_name, **op_kwargs) 方法对接 Data-Juicer Filter 算子，通过 op_name 指定算子名称，并通过 op_kwargs 配置对应算子的参数。

以下示例中的 Data-Juicer Filter 算子（默认处理 text 字段）接着过滤 cleaned_ds，仅保留长度为 10～200 个字符且单词数为 3～50 的文本：

filtered_ds = (
    cleaned_ds.datajuicer.filter(
        "text_length_filter",
        min_len=10,
        max_len=200,
        concurrency=2,
        num_cpus=1,
    )
    .datajuicer.filter(
        "words_num_filter",
        min_num=3,
        max_num=50,
    )
)

print(filtered_ds.take_all())

Data-Juicer Mapper 和 Filter 可以相互串联，也可以与 Ray Data 原生算子组合。Filter 结果可能包含 Data-Juicer 生成的统计字段；例如，可以在输出前选择需要的业务列：

output_ds = filtered_ds.select_columns(["id", "text"])

使用 prompt 算子生成文本

prompt 算子根据输入文本生成回复，可用于摘要、改写和问答。输入列可以是字符串，也可以是非空的消息列表；输出列为生成的文本。

• 参数说明：

  参数	说明
  公共参数（input、output、url、headers）	详见上方公共参数列表。
  model	可选，服务使用的模型名称。
  system_message	可选，添加到每条请求开头的系统提示词。

• 使用示例：

  以下示例对 Dataset 中的文本生成摘要：

  import os
  import ray
  
  ray.init()
  
  ds = ray.data.from_items([{"text": "Ray Data 可以并行处理大规模数据。"}])
  
  result_ds = ds.ai.prompt(
      input="text",
      output="summary",
      url=os.environ["PROMPT_ENDPOINT"],
      headers={"Authorization": f"Bearer {os.environ['INFERENCE_API_KEY']}"},
      model="<model-name>",
      system_message="请用一句话概括输入内容。",
  )
  
  print(result_ds.take(1))
  

使用 embed_text 算子生成文本向量

embed_text 算子将文本转换为向量，可用于语义检索和相似度计算。输入列必须是字符串；输出列为浮点数向量。

• 参数说明：

  参数	说明
  公共参数（input、output、url、headers）	详见上方公共参数列表。
  model	可选，服务使用的模型名称。
  dimensions	可选，输出向量的维度。是否支持及可选值取决于模型。

• 使用示例：

  以下示例为 Dataset 中的文本生成向量。

  import os
  import ray
  
  ray.init()
  
  ds = ray.data.from_items([{"text": "Ray Data 可以并行处理大规模数据。"}])
  
  result_ds = ds.ai.embed_text(
      input="text",
      output="embedding",
      url=os.environ["EMBEDDING_ENDPOINT"],
      headers={"Authorization": f"Bearer {os.environ['INFERENCE_API_KEY']}"},
      model="<embedding-model-name>",
  )
  
  print(result_ds.take(1))
  

使用 inference 算子调用推理服务

inference 算子可将自定义结构的请求传给推理服务，适用于服务支持的文本、图像等推理场景。输入列必须是字典，其内容应符合目标服务的请求格式；输出列保存服务返回的完整响应。

• 参数说明：

  参数	说明
  公共参数（input、output、url、headers）	详见上方公共参数列表。

• 使用示例：

  以下示例使用 Dataset 中的请求内容调用推理服务。

  import os
  import ray
  
  ray.init()
  
  request_ds = ray.data.from_items(
      [
          {
              "request": {
                  "model": "<model-name>",
                  "messages": [
                      {
                          "role": "user",
                          "content": "Explain Ray Data in one sentence.",
                      }
                  ],
              },
          }
      ]
  )
  
  response_ds = request_ds.ai.inference(
      input="request",
      output="response",
      url=os.environ["INFERENCE_ENDPOINT"],
      headers={"Authorization": f"Bearer {os.environ['INFERENCE_API_KEY']}"},
  )
  
  print(response_ds.take(1))