命令行界面 (CLI)

本节包含 CLI 的 API 参考。有关如何使用 CLI 的更多信息，请参阅 教程 - CLI。

distilabel pipeline 子命令的实用工具函数

以下是一些实用工具函数，可帮助在控制台中使用 pipelines。

utils

parse_runtime_parameters(params)

将运行时参数从 CLI 格式解析为 Pipeline.run 方法期望的格式。CLI 格式是元组列表，其中第一个元素是键列表，第二个元素是值。

参数

名称	类型	描述	默认值
params	List[Tuple[List[str], str]]	元组列表，其中第一个元素是键列表，第二个元素是值。	必需

返回

类型	描述
Dict[str, Dict[str, Any]]	一个字典，其中包含 Pipeline.run 方法期望格式的运行时参数。
Dict[str, Dict[str, Any]]	Pipeline.run 方法。

源代码位于 src/distilabel/cli/pipeline/utils.py

def parse_runtime_parameters(
    params: List[Tuple[List[str], str]],
) -> Dict[str, Dict[str, Any]]:
    """Parses the runtime parameters from the CLI format to the format expected by the
    `Pipeline.run` method. The CLI format is a list of tuples, where the first element is
    a list of keys and the second element is the value.

    Args:
        params: A list of tuples, where the first element is a list of keys and the
            second element is the value.

    Returns:
        A dictionary with the runtime parameters in the format expected by the
        `Pipeline.run` method.
    """
    runtime_params = {}
    for keys, value in params:
        current = runtime_params
        for i, key in enumerate(keys):
            if i == len(keys) - 1:
                current[key] = value
            else:
                current = current.setdefault(key, {})
    return runtime_params

valid_http_url(url)

检查 URL 是否为有效的 HTTP URL。

参数

名称	类型	描述	默认值
url	str	要检查的 URL。	必需

返回

类型	描述
bool	True，如果 URL 是有效的 HTTP URL。False，否则。

源代码位于 src/distilabel/cli/pipeline/utils.py

def valid_http_url(url: str) -> bool:
    """Check if the URL is a valid HTTP URL.

    Args:
        url: The URL to check.

    Returns:
        `True`, if the URL is a valid HTTP URL. `False`, otherwise.
    """
    try:
        TypeAdapter(HttpUrl).validate_python(url)  # type: ignore
    except ValidationError:
        return False

    return True

get_config_from_url(url)

从指向 JSON 或 YAML 文件的 URL 加载 pipeline 配置。

参数

名称	类型	描述	默认值
url	str	指向 pipeline 配置文件的 URL。	必需

返回

类型	描述
Dict[str, Any]	pipeline 配置，以字典形式。

引发

类型	描述
ValueError	如果文件格式不受支持。

源代码位于 src/distilabel/cli/pipeline/utils.py

def get_config_from_url(url: str) -> Dict[str, Any]:
    """Loads the pipeline configuration from a URL pointing to a JSON or YAML file.

    Args:
        url: The URL pointing to the pipeline configuration file.

    Returns:
        The pipeline configuration as a dictionary.

    Raises:
        ValueError: If the file format is not supported.
    """
    if not url.endswith((".json", ".yaml", ".yml")):
        raise DistilabelUserError(
            f"Unsupported file format for '{url}'. Only JSON and YAML are supported",
            page="sections/how_to_guides/basic/pipeline/?h=seriali#serializing-the-pipeline",
        )
    response = _download_remote_file(url)

    if url.endswith((".yaml", ".yml")):
        content = response.content.decode("utf-8")
        return yaml.safe_load(content)

    return response.json()

get_pipeline_from_url(url, pipeline_name='pipeline')

将文件下载到当前工作目录，并从 python 脚本加载 pipeline 对象。

参数

名称	类型	描述	默认值
url	str	指向包含 pipeline 定义的 python 脚本的 URL。	必需
pipeline_name	str	脚本中 pipeline 的名称。例如：with Pipeline(...) as pipeline:...。	'pipeline'

返回

类型	描述
BasePipeline	实例化的 pipeline。

引发

类型	描述
ValueError	如果文件格式不受支持。

源代码位于 src/distilabel/cli/pipeline/utils.py

def get_pipeline_from_url(url: str, pipeline_name: str = "pipeline") -> "BasePipeline":
    """Downloads the file to the current working directory and loads the pipeline object
    from a python script.

    Args:
        url: The URL pointing to the python script with the pipeline definition.
        pipeline_name: The name of the pipeline in the script.
            I.e: `with Pipeline(...) as pipeline:...`.

    Returns:
        The pipeline instantiated.

    Raises:
        ValueError: If the file format is not supported.
    """
    if not url.endswith(".py"):
        raise DistilabelUserError(
            f"Unsupported file format for '{url}'. It must be a python file.",
            page="sections/how_to_guides/advanced/cli/#distilabel-pipeline-run",
        )
    response = _download_remote_file(url)

    content = response.content.decode("utf-8")
    script_local = Path.cwd() / Path(url).name
    script_local.write_text(content)

    # Add the current working directory to sys.path
    sys.path.insert(0, os.getcwd())
    module = importlib.import_module(str(Path(url).stem))
    pipeline = getattr(module, pipeline_name, None)
    if not pipeline:
        raise ImportError(
            f"The script must contain an object with the pipeline named: '{pipeline_name}' that can be imported"
        )

    return pipeline

get_pipeline(config_or_script, pipeline_name='pipeline')

从配置文件或远程 python 脚本获取 pipeline。

参数

名称	类型	描述	默认值
config_or_script	str	pipeline 配置文件的路径或 URL，或 python 脚本的 URL。	必需
pipeline_name	str	脚本中 pipeline 的名称。例如：with Pipeline(...) as pipeline:...。	'pipeline'

返回

类型	描述
BasePipeline	pipeline。

引发

类型	描述
ValueError	如果文件格式不受支持。
FileNotFoundError	如果配置文件不存在。

源代码位于 src/distilabel/cli/pipeline/utils.py

def get_pipeline(
    config_or_script: str, pipeline_name: str = "pipeline"
) -> "BasePipeline":
    """Get a pipeline from a configuration file or a remote python script.

    Args:
        config_or_script: The path or URL to the pipeline configuration file
            or URL to a python script.
        pipeline_name: The name of the pipeline in the script.
            I.e: `with Pipeline(...) as pipeline:...`.

    Returns:
        The pipeline.

    Raises:
        ValueError: If the file format is not supported.
        FileNotFoundError: If the configuration file does not exist.
    """
    config = script = None
    if config_or_script.endswith((".json", ".yaml", ".yml")):
        config = config_or_script
    elif config_or_script.endswith(".py"):
        script = config_or_script
    else:
        raise DistilabelUserError(
            "The file must be a valid config file or python script with a pipeline.",
            page="sections/how_to_guides/advanced/cli/#distilabel-pipeline-run",
        )

    if valid_http_url(config_or_script):
        if config:
            data = get_config_from_url(config)
            return Pipeline.from_dict(data)
        return get_pipeline_from_url(script, pipeline_name=pipeline_name)

    if not config:
        raise ValueError(
            f"To run a pipeline from a python script, run it as `python {script}`"
        )

    if Path(config).is_file():
        return Pipeline.from_file(config)

    raise FileNotFoundError(f"File '{config_or_script}' does not exist.")

display_pipeline_information(pipeline)

向控制台显示 pipeline 信息。

参数

名称	类型	描述	默认值
pipeline	BasePipeline	pipeline。	必需

源代码位于 src/distilabel/cli/pipeline/utils.py

def display_pipeline_information(pipeline: "BasePipeline") -> None:
    """Displays the pipeline information to the console.

    Args:
        pipeline: The pipeline.
    """
    from rich.console import Console

    Console().print(_build_pipeline_panel(pipeline))