mmops SDK 使用指南

获取发布包

当前 0.1.0 版本的 wheel 包对象为公开读，可以直接下载，无需 OSS 凭证：

curl -fL -o ./mmops-0.1.0-py3-none-any.whl \
  'https://adb-mmops.oss-cn-beijing.aliyuncs.com/sdk/0.1.0/mmops-0.1.0-py3-none-any.whl'

也可以直接访问 mmops 0.1.0 wheel 下载地址。

安装 SDK 与依赖

在独立的虚拟环境中安装 wheel 包。以下示例使用 Python 3.12，命令中的 wheel 文件名替换为实际获取到的文件：

python3.12 -m venv .venv-mmops
source .venv-mmops/bin/activate
python -m pip install './mmops-0.1.0-py3-none-any.whl[ray]'
python -m pip check
python -c 'import mmops; print(mmops.__version__)'

[ray] 安装 Ray Dataset 的运行依赖。如果本地已安装同版本 wheel，重新安装时请使用 python -m pip install --force-reinstall --no-cache-dir './mmops-0.1.0-py3-none-any.whl[ray]'。

在 AnalyticDB Ray 上运行时，依赖版本须与资源组的实际 Python 和 Ray 版本一致。

可选依赖组

按实际使用的算子追加 extras。例如同时处理 HTML、图片和 OSS 文件：

python -m pip install './mmops-0.1.0-py3-none-any.whl[ray,office,media,oss]'

各 extra 的用途如下：

wheel 包仅包含 SDK，其余依赖由 pip 从可用的包源安装，模型权重和系统工具需另外准备。完全离线部署时，请提前准备所选依赖的安装包。

输入列与结果列

以下约定适用于内置 Dataset 算子，各算子的具体字段请参见离线算子概述中的算子目录：

算子默认只返回接口声明的结果列，原输入列不会自动全部保留。keep_columns 与标准结果列重名时会报错，需先用 Ray Data 原生的 rename_columns() 改名。

split_text 每行输出一个 chunks 列表，不会自动将一行展开为多行。如需每个片段占一行，请在后续使用 flat_map()。

通用算子从 ds.mmops 调用，行业领域算子分别从 ds.mmops.game、ds.mmops.product 和 ds.mmops.embodied 调用，算子名称与输入输出请参见游戏领域算子概述、商品场景算子概述和具身智能算子概述。

创建和加工 Dataset

Ray Data 负责创建、读取和写出数据集。以下示例展示筛选、生成字段、文本分块与展开：

import ray
from mmops.ray import enable_accessor


enable_accessor()
ray.init(num_cpus=4, include_dashboard=False)
try:
    source = ray.data.from_items([
        {"source_id": "doc-1", "body": "  第一段。\n\n第二段。  "},
        {"source_id": "doc-2", "body": "   "},
    ])
    cleaned = source.filter(lambda row: bool(row["body"].strip()))
    cleaned = cleaned.map(lambda row: {
        "source_id": row["source_id"], "text": row["body"].strip(),
    })
    chunks = cleaned.mmops.split_text(
        text_column="text",
        keep_columns=["source_id"],
        config={"strategy": "paragraph", "max_chars": 80, "overlap_chars": 0},
    )
    flattened = chunks.flat_map(lambda row: [
        {"source_id": row["source_id"],
         "chunk_index": chunk["chunk_index"], "text": chunk["content"]}
        for chunk in row["chunks"]
    ])
    result = flattened.select_columns(["source_id", "chunk_index", "text"])
    print(sorted(result.take(2), key=lambda row: row["chunk_index"]))
finally:
    ray.shutdown()

输出如下：

[{'source_id': 'doc-1', 'chunk_index': 0, 'text': '第一段。'},
 {'source_id': 'doc-1', 'chunk_index': 1, 'text': '第二段。'}]

其中 filter()、map()、flat_map() 和 select_columns() 是 Ray Data 原生方法，mmops.split_text() 是 SDK 提供的算子。后续可以继续接入字段提取、向量化或质量筛选等算子。

读取和写出数据

在已初始化 Ray 的脚本中，使用 Ray Data 原生读写接口：

# 输入路径需事先存在，文件中的字段即 Dataset 列。
source = ray.data.read_parquet("/shared/input/documents/")
# 其他常用格式：
# source = ray.data.read_json("/shared/input/documents.jsonl")
# source = ray.data.read_csv("/shared/input/documents.csv")

result = source.mmops.split_text(
    text_column="text", keep_columns=["source_id"],
    config={"strategy": "length", "max_chars": 800, "overlap_chars": 80},
)
result.write_parquet("/shared/output/document-chunks/job-001/")
# 或：result.write_json("/shared/output/document-chunks-json/job-001/")

每次任务使用独立的输出目录，避免重复执行时混入旧分片。多行结果不保证全局顺序，请使用 source_id、chunk_index 等字段关联记录。

文档解析后分块

安装 [ray,office] 后，可以在已初始化 Ray 的脚本中将 HTML 解析与文本分块串联。以下示例的输入文件需事先存在：

source = ray.data.from_items([
    {"source_id": "page-1", "source_uri": "/shared/input/guide.html"},
])
parsed = source.mmops.extract_html(
    document_column="source_uri", keep_columns=["source_id", "source_uri"],
)
result = parsed.mmops.split_text(
    text_column="text", keep_columns=["source_id", "source_uri"],
    config={"strategy": "length", "max_chars": 800, "overlap_chars": 80},
)
result.write_parquet("/shared/output/html-chunks/job-001/")

extract_html 只提取静态正文和表格，不执行页面中的 JavaScript。其他格式请从文档算子概述中选择，每一步均需显式保留后续要用的来源列。

函数 UDF

自定义处理通过 Ray Data 原生接口加入流水线：逐行处理使用 map()，批量处理使用 map_batches()。以下示例先规范文本空白，再调用内置的分块算子：

import re

import ray
from mmops.ray import enable_accessor


def normalize_text(batch):
    return {
        "source_id": batch["source_id"],
        "text": [re.sub(r"\s+", " ", text).strip() for text in batch["text"]],
    }


enable_accessor()
ray.init(num_cpus=4, include_dashboard=False)
try:
    source = ray.data.from_items([
        {"source_id": "doc-1", "text": "  Ray   Data\n 使用说明  "},
    ])
    cleaned = source.map_batches(
        normalize_text, batch_format="numpy", batch_size=32, num_cpus=1,
    )
    result = cleaned.mmops.split_text(
        text_column="text", keep_columns=["source_id"],
        config={"strategy": "length", "max_chars": 80, "overlap_chars": 0},
    )
    print(result.take(1))
finally:
    ray.shutdown()

类 UDF

需要在批次之间复用模型、客户端或其他状态时，使用可调用类。__init__ 在每个 Actor 内初始化，__call__ 处理后续批次。以下类复用编译后的正则表达式：

class NormalizeText:
    def __init__(self):
        self.whitespace = re.compile(r"\s+")

    def __call__(self, batch):
        return {
            "source_id": batch["source_id"],
            "text": [self.whitespace.sub(" ", text).strip() for text in batch["text"]],
        }


# 在上面的脚本中，用这一段替换 cleaned = source.map_batches(...)。
cleaned = source.map_batches(
    NormalizeText,
    compute=ray.data.ActorPoolStrategy(size=1),
    batch_format="numpy", batch_size=32, num_cpus=1,
)

模型权重也可以在类的构造方法中加载，供该 Actor 的后续批次复用。使用本地 GPU 模型时，请在 map_batches() 中按需设置 num_gpus。完整的输入校验、状态管理和重试约定，请参见自定义离线算子（UDF）。

在 AnalyticDB Ray 上提交作业

先按托管Ray服务准备资源组，并在资源组详情中获取可访问的 HTTP 连接地址，例如 http://<ADB_RAY_HOST>:8265。提交端须能访问该地址，执行节点须能访问 wheel 包地址和所用的 pip 源。

提交前请核对资源组的 Python 和 Ray 版本。基于DMS Notebook进行Ray交互式开发使用 Python 3.11.13 和 Ray 2.49.2，以下示例按 Ray 2.49.2 配置，实际版本不同时请同步调整。建议先用下面的无模型任务验证目标资源组。

在独立目录 adb-mmops-job 中准备两个文件：

adb-mmops-job/
  main.py
  runtime_env.yaml

runtime_env.yaml 安装 SDK，并固定资源组的实际 Ray 版本：

pip:
  packages:
    - "ray[data]==2.49.2"
    - "mmops[ray] @ https://adb-mmops.oss-cn-beijing.aliyuncs.com/sdk/0.1.0/mmops-0.1.0-py3-none-any.whl"
  pip_check: true

main.py 连接作业所在的集群，执行文本分块并检查输出：

import ray
from mmops.ray import enable_accessor


ray.init(address="auto")
enable_accessor()
try:
    source = ray.data.from_items([
        {"source_id": "doc-1", "text": "ADB Ray 上运行 mmops。"},
    ])
    result = source.mmops.split_text(
        text_column="text", keep_columns=["source_id"],
        config={"strategy": "length", "max_chars": 800, "overlap_chars": 80},
    )
    rows = result.take(1)
    assert rows[0]["chunks"][0]["content"] == "ADB Ray 上运行 mmops。"
    print(rows)
finally:
    ray.shutdown()

在独立的提交环境中安装匹配版本的 CLI，再进入作业目录提交作业。连接地址替换为实际值：

python -m pip install 'ray[default]==2.49.2'
cd adb-mmops-job
ray job submit \
  --address 'http://<ADB_RAY_HOST>:8265' \
  --working-dir . \
  --runtime-env runtime_env.yaml \
  -- python main.py

提交作业时请注意以下事项：

• 工作目录只放作业文件，不要放虚拟环境或密钥，因为 Ray 会将该目录打包上传。

• 作业提交层的 runtime_env 同时应用于 Driver、Task 和 Actor。仅在 ray.init(runtime_env=...) 中配置依赖不会安装到 Driver 本身。更多信息，请参见 Ray 依赖管理。

• 按任务需要添加 pdf、office、media、oss 等 extras。FFmpeg、LibreOffice 等系统工具需在执行环境中另外准备。

• 模型配置通过提交层的 env_vars 或节点环境提供，凭证不要写入上传目录。

• 同版本 wheel 包被覆盖后，Ray 仍可能复用已有的依赖环境。此时可在 wheel 包地址末尾添加或更新 #sha256=<本次发布的 SHA256>，使 pip 配置随产物变化并校验下载内容。

交互调试可以使用 ray://<ADB_RAY_HOST>:10001 连接，但 Notebook 与集群的 Python、Ray 版本须一致，且 SDK 须安装在 Notebook 自身的环境中。批处理请使用上述 Ray Jobs 方式。

连接其他 Ray 集群

上述本地示例使用 ray.init(num_cpus=4, ...)。连接已准备好的 Ray 集群时，将初始化改为：

import os
import ray
from mmops.ray import enable_accessor


enable_accessor()
env_vars = {name: value for name, value in os.environ.items()
            if name.startswith("MMOPS_")}
ray.init(address="auto", runtime_env={"env_vars": env_vars})

运行前，请在 Driver 和每个 Worker 的镜像或环境中安装相同版本的 wheel 包、Ray 与所需依赖。runtime_env.env_vars 传递环境配置，但文件路径指向的模型上下文或权重仍需在节点上实际可读。如果代码中通过 api_key_env 指定了其他凭证变量名，也需将该变量传入执行环境。

文件与对象存储

集群中的 /shared 表示所有 Worker 节点均可读写的共享挂载目录。该目录需要实际准备，不是 AnalyticDB Ray 默认提供的目录。Driver 的临时目录、下载目录和 file:// 路径不会自动分发到其他节点，因此前文的临时文件示例仅适用于本地运行。AnalyticDB Ray 的 Worker 磁盘用于临时数据，正式结果请写入已配置的持久化存储。

SDK 访问 OSS 文件时，需要安装 [oss]，并向执行进程注入 MMOPS_OSS_ENDPOINT、MMOPS_OSS_ACCESS_KEY_ID 和 MMOPS_OSS_ACCESS_KEY_SECRET。使用 STS 临时凭证时，还需提供 MMOPS_OSS_SECURITY_TOKEN。这些变量仅用于 SDK 的文件访问，不会自动配置 Ray 或 Arrow 的数据集 filesystem。快速上手时，可以先将结果表写入共享挂载目录；如需直接在 OSS 上读写结果数据集，请参见下文。

python -m pip install --force-reinstall './mmops-0.1.0-py3-none-any.whl[ray,ossfs]'
python -m pip check

import uuid

import ray
from mmops.storage import oss_filesystem

# 前置条件：已初始化 Ray，并已配置 MMOPS_OSS_* 环境变量。
filesystem = oss_filesystem()
path = f"your-bucket/mmops/test/check-{uuid.uuid4().hex}/"
expected = [{"id": 1, "text": "第一条数据"}, {"id": 2, "text": "第二条数据"}]

ray.data.from_items(expected).write_parquet(path, filesystem=filesystem)
rows = ray.data.read_parquet(path, filesystem=filesystem).take_all()
assert sorted(rows, key=lambda row: row["id"]) == expected
print(f"OSS 写入、回读校验成功：oss://{path}")

批次与资源

内置算子的 batch_size 是 Ray 每批处理的输入行数，增大该值不代表模型请求会自动合并。concurrency 控制执行实例数，num_cpus 是每个实例申请的资源。实际并行度还受数据分块和集群可用资源的限制。

内置的图片理解、语音识别等算子通过远端模型服务调用，不在 Worker 节点上申请本地 GPU。自定义 UDF按需声明 num_gpus，并确保模型权重对所有执行节点可见。