Prerequisites #

Operator usage prerequisites: Enable the Business Risk Identification product - Audio Risk Identification - Audio On-Demand Service.

Operator parameters #

Input #

Output #

A struct containing the risk identification result

• Decision: Determination result (PASS, BLOCK, REVIEW)
• Message: Information returned by the server
• risk_result: Complete detection result in JSON format

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use Daft (for distributed scenarios) to run the operator for audio content risk identification.

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.audio import AudioRiskRec
from daft.las.functions.udf import las_udf

if __name__ == "__main__":

    """
    算子使用前置条件：开通业务风险识别产品-音频风险识别-音频点播服务
    app_id: 在业务风险识别产品中开通的应用ID
    biztype: 在业务风险识别产品中配置的音频风险识别场景
    本示例代码中使用模拟值
    """
    
    # Caution: AudioRiskRec requires the app_id and biztype parameters. Example values are used here.
    
    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import logging

        import ray

        def configure_logging():
            logging.basicConfig(
                level=logging.INFO,
                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S.%s".format(),
            )
            logging.getLogger("tracing.span").setLevel(logging.WARNING)
            logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
            logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

        ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
        daft.set_runner_ray()
    daft.set_execution_config(actor_udf_ready_timeout=600)
    daft.set_execution_config(min_cpu_per_task=0)

    # Build the URL using environment variables
    tos_dir_url = os.getenv("TOS_DIR_URL", "las-cn-beijing-public-online.tos-cn-beijing.volces.com")
    samples = {
        "audio_id": ["1"],
        "audio_path": [
            f"https://{tos_dir_url}/public/shared_audio_dataset/sample.mp3"
        ],
    }
    df = daft.from_pydict(samples)

    # Replace with valid app_id and biztype for actual use
    constructor_kwargs = {"app_id": 123456, "biztype": "test_biztype"}

    # Use Daft for distributed processing
    df = df.with_column(
        "parsed_result",
        las_udf(AudioRiskRec, construct_args=constructor_kwargs, concurrency=1)(col("audio_id"), col("audio_path")),
    )
    df = df.with_column("Decision", col("parsed_result")["Decision"])
    df = df.with_column("Message", col("parsed_result")["Message"])
    df = df.with_column("risk_result", col("parsed_result")["risk_result"])

    df.select("audio_id", "audio_path", "Decision", "Message").show()
    # ╭──────────┬────────────────────────────────┬──────────┬─────────╮
    # │ audio_id ┆ audio_path                     ┆ Decision ┆ Message │
    # │ ---      ┆ ---                            ┆ ---      ┆ ---     │
    # │ Utf8     ┆ Utf8                           ┆ Utf8     ┆ Utf8    │
    # ╞══════════╪════════════════════════════════╪══════════╪═════════╡
    # │ 1        ┆ https://las-cn-beijing-publi-… ┆ PASS     ┆ success │
    # ╰──────────┴────────────────────────────────┴──────────┴─────────╯