语音识别示例

场景描述

在音频数据处理场景中，需要将存储在数据库表中的音频二进制数据通过语音识别模型（Whisper）转换为文本。通过Class形式的Scalar UDF，在__init__中加载Whisper模型，在__call__中对每段音频执行识别并返回文本结果。本示例从sample-files.com下载一段真实语音音频作为测试数据，写入数据库表后供UDF处理。

约束与限制

• UDF运行时需通过pip安装transformers包，请确保运行时环境网络可用。
• Scalar UDF的__init__参数仅支持传入常量。
• 注册UDF前建议先调用delete_function清理同名函数，避免重复执行时报错。

前提条件

• 已安装Aura DataFrame SDK。具体操作，请参见Aura DataFrame SDK获取与安装。
• 已创建AuraJobV2端点，并获取端点名称。具体操作，请参见创建Aura引擎端点。
• 已创建LakeFormation实例，并在该实例中创建Catalog及数据库。具体操作，请参见创建LakeFormation实例并规划元数据。
• 已创建并行文件系统。具体操作，请参见创建并行文件系统。
• 代码中涉及os.getenv()中的参数已配置到环境变量。
• 运行时环境要求Python版本为3.11。

准备工作

本示例使用Whisper语音识别模型，需要准备模型文件、音频文件和Python依赖。

1. 安装Python依赖。

   pip install torch==2.6.0
   pip install transformers==5.3.0
   pip install av
   pip install scipy

2. 安装huggingface命令行工具。

   pip install huggingface-hub

3. 下载测试音频文件到示例脚本同级目录，重命名为test_audio.wav：

   下载一段WAV格式的语音文件，建议10秒以内。

4. 下载Whisper base模型文件到示例脚本同级目录。

   hf download openai/whisper-base --local-dir ./whisper_base

   如果hf download命令无法使用，可通过浏览器手动下载：

   1. 在Hugging Face网站搜索openai/whisper-base，进入模型主页。
   2. 单击“Files and versions”标签页。
   3. 下载以下文件，保存到本地whisper_base目录下：
      • config.json
      • pytorch_model.bin
      • preprocessor_config.json
      • tokenizer_config.json
      • tokenizer.json
      • vocab.json
      • merges.txt
      • normalizer.json
      • special_tokens_map.json
      • added_tokens.json
      • generation_config.json
   4. 将whisper_base目录放置在与示例脚本同级目录下。

示例

本示例定义一个基于Whisper模型的语音识别Scalar UDF，将测试音频写入数据库后，通过map_batches方法调用UDF将音频二进制数据转换为文本。

1. 定义Class UDF类SpeechRecognitionUDF，在__init__中通过transformers加载Whisper语音识别模型，在__call__中用av库解码音频二进制数据并重采样到16kHz后执行识别。
2. 连接Aura服务并设置UDF工作空间。
3. 读取测试音频文件，建表并插入音频二进制数据。
4. 删除已存在的同名UDF（支持重复执行）。
5. 注册Scalar UDF到目标数据库，指定依赖包和模型文件。
6. 加载audio_data数据集。
7. 通过map_batches方法调用UDF进行语音识别，生成hyp_text列。
8. 选择相关列并执行查询打印结果。
9. 关闭Aura连接。

import os
import ibis
import aura_frame
from aura_frame.multimodal import ai_lake



# 定义Scalar UDF：语音识别
class SpeechRecognitionUDF:
    def __init__(self, model_path: str) -> None:
        import torch
        from transformers import pipeline
        self._pipeline = pipeline(
            "automatic-speech-recognition",
            model=model_path,
            device=0 if torch.cuda.is_available() else -1,
        )
    def __call__(self, audios: bytes) -> str:
        import numpy as np
        import av
        import io
        from scipy.signal import resample_poly
        container = av.open(io.BytesIO(audios))
        stream = container.streams.audio[0]
        frames = []
        for frame in container.decode(stream):
            frames.append(frame.to_ndarray())
        if not frames:
            raise ValueError("Audio data is None.")
        audio_np = np.concatenate(frames, axis=1)
        if audio_np.ndim > 1:
            audio_np = audio_np.mean(axis=0)
        if audio_np.dtype == np.int16:
            audio_np = audio_np.astype(np.float32) / 32768.0
        elif audio_np.dtype == np.int32:
            audio_np = audio_np.astype(np.float32) / 2147483648.0
        elif audio_np.dtype != np.float32:
            audio_np = audio_np.astype(np.float32)
        # 重采样到 16kHz（Whisper 要求）
        target_sr = 16000
        if stream.rate != target_sr:
            audio_np = resample_poly(audio_np, target_sr, stream.rate)
        result = self._pipeline({"raw": audio_np, "sampling_rate": target_sr})
        return result["text"]

try:
    # 读取测试音频文件
    audio_path = "test_audio.wav"
    with open(audio_path, "rb") as f:
        audio_bytes = f.read()
    # 建表并插入音频数据
    table_name = "audio_data"
    table_schema = ibis.schema({"id": "int32", "audio_data": "binary"})
    con.sql(f'DROP TABLE IF EXISTS "{target_database}"."{table_name}";')
    con.create_table(
        table_name,
        schema=table_schema,
        database=target_database,
        table_format="orc",
        if_not_exists=True,
    )
    hex_str = audio_bytes.hex()
    con.sql(f'''
    INSERT INTO "{target_database}"."{table_name}" (id, audio_data) VALUES
        (1, '\\x{hex_str}'::bytea)
    ''')
    print("音频数据写入完成")
    # 删除已存在的同名UDF（支持重复执行）
    con.delete_function("speechrecognitionudf", database=target_database, if_it_exists=True)
    # 注册Scalar UDF：语音识别
    con.create_scalar_function(
        SpeechRecognitionUDF,
        database=target_database,
        packages=["transformers==5.3.0"],
        imports=["./whisper_base"],
    )
    # 加载音频数据集
    ds = con.load_dataset("audio_data", database=target_database)
    # 使用Scalar UDF进行语音识别
    udf = con.get_function("speechrecognitionudf", database=target_database)
    ds = ds.map_batches(
        fn=udf,
        on=[ds.audio_data],
        as_col="hyp_text",
        num_cpus=2,
        memory=4096,
        arch="x86",
        model_path="./whisper_base",
    )
    ds = ds.select_columns(ds.id, ds.audio_data, ds.hyp_text)
    df = ds.execute()
    print(df)
finally:
    con.close()

相关文档

• Session APIs
• Dataset APIs
• Table APIs
• User-Defined Function APIs