音频检测与切分-阿里云帮助中心

本文介绍 Python DataFrame API 中的静音检测、语音活动检测和音频切分算子。

使用限制

仅实时计算引擎 VVR 11.8 及以上版本支持。
三个音频切分算子都是 UDTF，需要配合 DataFrame.join_lateral 使用。
本文使用音频片段引用 (AUDIO_CLIP_REF_TYPE) 和解码后的音频波形 (AUDIO_WAVEFORM_TYPE) 表示 pyflink.multimodal.types 中定义的 DataFrame API 类型。

算子清单

分类	算子	说明
音频检测	audio_silence_detection	检测音频波形中的低振幅静音区间。
音频检测	audio_detect_speech	使用 WebRTC VAD 检测音频波形中的语音活动区间。
音频切分	audio_split_by_duration	按指定固定时长依次切分音频，并根据 `segment_type` 输出实际音频波形或片段引用。适合处理超过下游单次输入上限的长音频，例如将一小时录音切成 30 秒片段并行送入 ASR；最后一段可以更短。
	audio_split_by_timestamp	按调用方提供的开始和结束时间范围切分音频，并按原顺序输出音频波形或片段引用。适合已经有字幕、日志、会议纪要或人工标注时间轴的场景，例如按字幕时间提取对应音频，或按会议纪要定位发言片段。
	audio_split_by_speech	根据已有语音活动范围，对区间做扩边、合并、短段过滤和长段拆分，再输出音频波形或片段引用。适合处理含大量静音的通话、访谈或 Podcast，例如配合 `audio_detect_speech` 跳过空白，只生成适合 ASR 的人声片段。

时间范围类型

静音检测和语音活动检测返回以下 DataFrame API 类型：

DataType.list(
    DataType.struct({
        "start_ms": DataType.int64(),
        "end_ms": DataType.int64(),
        "duration_ms": DataType.int64()
    })
)

时间范围采用半开区间 [start_ms, end_ms)，单位为毫秒。

通用 Runtime 参数

函数签名保留各算子实际支持的 Runtime 参数。为避免重复，算子参数表只说明业务参数，Runtime 参数统一说明如下。

参数	类型	默认值	适用范围	说明
`concurrency`	`Optional[int]`	`None`	本页全部算子	UDF 或 UDTF 并发度。`None` 表示使用框架默认值。

音频检测

audio_silence_detection

检测音频波形中的低振幅静音区间。

输入类型： AUDIO_WAVEFORM_TYPE，解码后的音频波形列。

函数签名：

audio_silence_detection(
    *columns,
    threshold_db,
    min_silence_ms=0,
    concurrency=None
)

参数	类型	默认值	说明
`threshold_db`	`float`	必填	相对于满幅的静音阈值，必须小于或等于 0，能量不高于该阈值的帧视为静音。
`min_silence_ms`	`int`	`0`	最短静音区间，单位为毫秒；更短的区间不返回。

算子按以下方式判断静音：

多声道音频先将同一时刻的各声道取平均，得到单声道音频。
将音频分成较短的重叠片段，使用 RMS（均方根）估算每个片段的平均音量。
将平均音量表示为 dBFS。dBFS 越接近 0 表示声音越大，数值越小表示声音越小。
连续低于或等于 threshold_db 且时长不少于 min_silence_ms 的部分，作为静音区间返回。

例如，threshold_db=-45.0 时，某个片段的平均音量为 -30 dBFS，不会被判定为静音；为 -50 dBFS 时，会被判定为静音。

该算子只按音量判断静音，不使用 VAD 或 ASR 模型，因此“静音区间”不等同于“非语音区间”。空值输入返回 None。

返回类型：

DataType.list(
    DataType.struct({
        "start_ms": DataType.int64(),    # 区间起始时间
        "end_ms": DataType.int64(),      # 区间结束时间
        "duration_ms": DataType.int64()  # 区间持续时长
    })
)

from pyflink.dataframe import col
from pyflink.multimodal.operators import audio_silence_detection

result = df.with_column(
    "silence_ranges",
    audio_silence_detection(
        col("waveform"),
        threshold_db=-45.0,
        min_silence_ms=300
    )
)

audio_detect_speech

使用 WebRTC VAD 检测音频波形中的语音活动区间。

输入类型： AUDIO_WAVEFORM_TYPE，单声道音频波形列。采样率必须为 8000、16000、32000 或 48000 Hz。

函数签名：

audio_detect_speech(
    *columns,
    aggressiveness=0,
    frame_ms=30,
    min_speech_ms=0,
    merge_gap_ms=0,
    max_segments=1024,
    concurrency=None
)

参数	类型	默认值	说明
`aggressiveness`	`int`	`0`	WebRTC VAD 模式，取值范围为 `[0, 3]`。
`frame_ms`	`int`	`30`	VAD 帧长，只能为 10、20 或 30 毫秒。
`min_speech_ms`	`int`	`0`	丢弃短于该时长的语音区间，单位为毫秒。
`merge_gap_ms`	`int`	`0`	合并间隔不超过该值的相邻语音区间，单位为毫秒。
`max_segments`	`int`	`1024`	最多返回的语音区间数，必须为正整数。

返回结果按时间排序。空值输入返回 None。

返回类型：

DataType.list(
    DataType.struct({
        "start_ms": DataType.int64(),    # 区间起始时间戳
        "end_ms": DataType.int64(),      # 区间结束时间戳
        "duration_ms": DataType.int64()  # 区间持续时长
    })
)

from pyflink.multimodal.operators import (
    audio_detect_speech,
    audio_standardize,
)

# 先进行标准化处理
speech_ready = audio_standardize(
    col("waveform"),
    sample_rate=16000,
    channels=1
)

# 识别语音活动区间
result = df.with_column(
    "speech_ranges",
    audio_detect_speech(
        speech_ready,
        aggressiveness=2,
        min_speech_ms=200,
        merge_gap_ms=100
    )
)

音频切分

audio_split_by_duration

按固定时长切分音频。

输入类型：

segment_type="audio"：AUDIO_WAVEFORM_TYPE；
segment_type="ref"：DataType.string()（音频 URL）或 AUDIO_CLIP_REF_TYPE，要切分的音频列。

输入类型必须与 segment_type 匹配；输入为空值时不输出行。

函数签名：

audio_split_by_duration(
    *columns,
    segment_duration_ms,
    segment_type="audio",
    max_segments=1024,
    concurrency=None
)

参数	类型	默认值	说明
`segment_duration_ms`	`int`	必填	每个片段的目标时长，单位为毫秒，必须大于 0。
`segment_type`	`str`	`"audio"`	`"audio"` 输出为物化波形；`"ref"` 输出音频片段引用。
`max_segments`	`int`	`1024`	每个输入允许的最大片段数，必须大于 0；超过限制时抛出异常。

返回类型： UDTF 输出单列 segment。

segment_type="audio" 时为 AUDIO_WAVEFORM_TYPE；
segment_type="ref" 时为 AUDIO_CLIP_REF_TYPE。

行为说明：

segment_type="audio" 要求输入为解码后的音频波形。
segment_type="ref" 要求输入为 URI 或音频片段引用。算子会探测实际音频时长，并结合已有片段引用的边界进行裁剪；输出只保存片段引用，不复制音频内容。
最后一个片段可能会短于 segment_duration_ms。
需要生成的片段数超过 max_segments 时抛出异常，不会静默截断。

from pyflink.multimodal.operators import audio_split_by_duration

segments = df.join_lateral(
    audio_split_by_duration(
        col("audio_uri"),
        segment_duration_ms=30_000,
        segment_type="ref",
    ).alias("segment")
)

audio_split_by_timestamp

按调用方提供的时间范围切分音频。

输入类型：

第一个输入列
- segment_type="audio"：AUDIO_WAVEFORM_TYPE；
- segment_type="ref"：DataType.string() （音频 URL）或 AUDIO_CLIP_REF_TYPE。
不使用第二个输入列：通过 timestamps 传入固定的 Python 字面量时间范围。
使用第二个输入列：不设置 timestamps，由第二个输入列逐行提供时间范围。该列的 DataFrame API 类型为：

DataType.list(
    DataType.struct({
        "start_ms": DataType.int64(),
        "end_ms": DataType.int64(),
    })
)

函数签名：

audio_split_by_timestamp(
    *columns,
    timestamps=None,
    segment_type="audio",
    concurrency=None
)

参数	类型	默认值	说明
`timestamps`	`Optional[Union[List[Any], Tuple[Any, ...]]]`	`None`	固定时间范围，外层必须是 Python 列表或元组。每项可以是 `Row`、包含 `start_ms` 和 `end_ms` 属性的对象、包含同名键的字典，或二元列表/元组。起止时间必须为非负 `int`，且 `end_ms` 不得小于 `start_ms`。`None` 表示使用第二个输入列逐行提供时间范围，不表示不切分。
`segment_type`	`str`	`"audio"`	`"audio"` 返回物化波形；`"ref"` 返回音频片段引用。

timestamps 参数和第二个时间范围输入列不能同时使用。URI 输入的范围不会根据实际音频时长裁剪；有边界的音频片段引用只根据已有引用边界裁剪。音频或时间范围输入为空值时不输出行。

返回类型：UDTF 输出单列 segment。

segment_type="audio" 时为 AUDIO_WAVEFORM_TYPE；
segment_type="ref" 时为 AUDIO_CLIP_REF_TYPE。

from pyflink.dataframe import col
from pyflink.multimodal.operators import audio_split_by_timestamp

ranges = [
    {"start_ms": 1_000, "end_ms": 3_500},
    {"start_ms": 8_000, "end_ms": 12_000},
]

# 不使用第二个输入列：通过 Python 参数提供固定时间范围。
literal_segments = df.join_lateral(
    audio_split_by_timestamp(
        col("audio_uri"),
        timestamps=ranges,
        segment_type="ref"
    ).alias("segment")
)

# 使用第二个输入列：每行可以提供不同的时间范围。
column_segments = df.join_lateral(
    audio_split_by_timestamp(
        col("audio_uri"),
        col("ranges"),
        segment_type="ref"
    ).alias("segment")
)

audio_split_by_speech

根据上游已经生成的语音活动范围切分音频。该算子本身不检测语音，可将 audio_detect_speech 的结果作为语音活动范围输入。

输入类型：

segment_type="audio"：AUDIO_WAVEFORM_TYPE + DataType.list(DataType.struct(...))（语音活动区间列）；
segment_type="ref"：DataType.string() 或 AUDIO_CLIP_REF_TYPE + DataType.list(DataType.struct(...))（语音活动区间列）。

第一个输入列是与 segment_type 匹配的音频列，第二个输入列是语音活动范围数组列，算子根据 “语音范围数组列”（第二个输入列），对音频列进行划分。

函数签名：

audio_split_by_speech(
    *columns,
    segment_type="audio",
    pre_padding_ms=0,
    post_padding_ms=0,
    merge_gap_ms=0,
    min_segment_ms=0,
    max_segment_ms=None,
    max_segments=1024,
    concurrency=None
)

参数	类型	默认值	说明
`segment_type`	`str`	`"audio"`	`"audio"` 返回物化波形；`"ref"` 返回音频片段引用。
`pre_padding_ms`	`int`	`0`	每段开始前增加的时长，单位为毫秒，必须大于或等于 0。
`post_padding_ms`	`int`	`0`	每段结束后增加的时长，单位为毫秒，必须大于或等于 0。
`merge_gap_ms`	`int`	`0`	合并间隔不超过该值的区间，单位为毫秒。
`min_segment_ms`	`int`	`0`	丢弃短于该值的区间，单位为毫秒。
`max_segment_ms`	`Optional[int]`	`None`	把过长区间继续切分为不超过该长度的片段；`None` 表示不限制。
`max_segments`	`int`	`1024`	最大结果片段数，必须大于 0。

超出实际音频时长的范围会被裁剪，完全越界的范围会被丢弃。音频或语音活动范围输入为空值时不输出行。

返回类型： UDTF 单列 segment。

segment_type="audio" 时为 AUDIO_WAVEFORM_TYPE；
segment_type="ref" 时为 AUDIO_CLIP_REF_TYPE。

from pyflink.multimodal.operators import audio_split_by_speech

# 生成语音活动区间
detected_ranges = audio_detect_speech(
    col("speech_ready"),
    aggressiveness=2,
    min_speech_ms=100,
    merge_gap_ms=50
)

# 进行区间切分
expression_segments = df.join_lateral(
    audio_split_by_speech(
        col("speech_ready"),
        detected_ranges,
        pre_padding_ms=200,
        post_padding_ms=200,
        merge_gap_ms=100,
    ).alias("segment")
)