README.md



AIfeng/2025-07-11 13:36:00


豆包ASR语音识别服务

基于豆包（Doubao）语音识别API的通用ASR服务，支持流式和非流式语音识别，提供简洁易用的Python接口。


🚀 特性


多种识别模式: 支持流式和非流式语音识别

多格式支持: 支持WAV、MP3、PCM等音频格式

灵活配置: 支持配置文件、环境变量、代码配置等多种方式

异步支持: 基于asyncio的异步API，支持高并发

实时回调: 流式识别支持实时结果回调

错误处理: 完善的错误处理和重试机制

易于集成: 简洁的API设计，易于集成到现有项目


📦 安装


依赖要求

pip install websockets aiofiles


项目结构

asr/doubao/
├── __init__.py          # 模块初始化和公共API
├── config.json          # 默认配置文件
├── config_manager.py    # 配置管理器
├── protocol.py          # 豆包协议处理
├── audio_utils.py       # 音频处理工具
├── asr_client.py        # ASR客户端核心
├── service_factory.py   # 服务工厂和便捷接口
├── example.py           # 使用示例
└── README.md           # 项目文档


🔧 配置


1. 获取API密钥

访问豆包开放平台获取：


app_key: 应用密钥

access_key: 访问密钥


2. 配置方式


方式1: 环境变量（推荐）

export DOUBAO_APP_KEY="your_app_key"
export DOUBAO_ACCESS_KEY="your_access_key"


方式2: 配置文件

创建 config.json：

{
  "auth_config": {
    "app_key": "your_app_key",
    "access_key": "your_access_key"
  },
  "asr_config": {
    "streaming_mode": true,
    "enable_punc": true,
    "seg_duration": 200
  }
}


方式3: 代码配置

service = create_asr_service(
    app_key="your_app_key",
    access_key="your_access_key"
)


🎯 快速开始


1. 简单文件识别

import asyncio
from asr.doubao import recognize_file

async def simple_recognition():
    result = await recognize_file(
        audio_path="path/to/your/audio.wav",
        app_key="your_app_key",
        access_key="your_access_key",
        streaming=True
    )
    print(f"识别结果: {result}")

# 运行
asyncio.run(simple_recognition())


2. 流式识别with实时回调

import asyncio
from asr.doubao import create_asr_service

async def streaming_recognition():
    # 定义结果回调函数
    def on_result(result):
        if result.get('payload_msg'):
            print(f"实时结果: {result['payload_msg']}")

    # 创建服务实例
    service = create_asr_service(
        app_key="your_app_key",
        access_key="your_access_key",
        streaming=True
    )

    try:
        result = await service.recognize_file(
            "path/to/your/audio.wav",
            result_callback=on_result
        )
        print(f"最终结果: {result}")
    finally:
        await service.close()

# 运行
asyncio.run(streaming_recognition())


2.1 优化的文本输出（推荐）

针对豆包ASR输出完整报文但只需要文本的问题，提供了专门的结果处理器：

import asyncio
from asr.doubao import create_asr_service
from asr.doubao.result_processor import create_text_only_callback

async def optimized_streaming():
    # 只处理文本内容的回调函数
    def on_text(text: str):
        print(f"识别文本: {text}")
        # 流式数据特点：后一次覆盖前一次，最终结果会不停刷新

    # 创建优化的回调（自动提取text字段）
    optimized_callback = create_text_only_callback(
        user_callback=on_text,
        enable_streaming_log=False  # 关闭中间结果日志
    )

    service = create_asr_service(
        app_key="your_app_key",
        access_key="your_access_key",
        streaming=True
    )

    try:
        await service.recognize_file(
            "path/to/your/audio.wav",
            result_callback=optimized_callback
        )
    finally:
        await service.close()

# 运行
asyncio.run(optimized_streaming())


3. 音频数据识别

import asyncio
from asr.doubao import recognize_audio_data

async def data_recognition():
    # 读取音频数据
    with open("path/to/your/audio.wav", "rb") as f:
        audio_data = f.read()

    result = await recognize_audio_data(
        audio_data=audio_data,
        audio_format="wav",
        app_key="your_app_key",
        access_key="your_access_key"
    )
    print(f"识别结果: {result}")

# 运行
asyncio.run(data_recognition())


4. 同步方式（简单场景）

from asr.doubao import run_recognition

# 同步识别
result = run_recognition(
    audio_path="path/to/your/audio.wav",
    app_key="your_app_key",
    access_key="your_access_key"
)
print(f"识别结果: {result}")


📚 详细用法


服务实例管理

from asr.doubao import DoubaoASRService, create_asr_service

# 创建服务实例
service = create_asr_service(
    app_key="your_app_key",
    access_key="your_access_key",
    streaming=True,
    debug=True
)

# 执行多次识别（复用连接）
result1 = await service.recognize_file("audio1.wav")
result2 = await service.recognize_file("audio2.wav")

# 关闭服务
await service.close()


批量识别

async def batch_recognition(audio_files):
    service = create_asr_service(
        app_key="your_app_key",
        access_key="your_access_key"
    )

    results = []
    try:
        for audio_file in audio_files:
            result = await service.recognize_file(audio_file)
            results.append({
                'file': audio_file,
                'result': result
            })
    finally:
        await service.close()

    return results

# 使用
audio_files = ["audio1.wav", "audio2.wav", "audio3.wav"]
results = await batch_recognition(audio_files)


自定义配置

custom_config = {
    'asr_config': {
        'enable_punc': True,
        'seg_duration': 300,  # 自定义分段时长
        'streaming_mode': True
    },
    'connection_config': {
        'timeout': 60,  # 自定义超时时间
        'retry_times': 5
    },
    'logging_config': {
        'enable_debug': True
    }
}

service = create_asr_service(
    app_key="your_app_key",
    access_key="your_access_key",
    custom_config=custom_config
)


配置文件使用

# 使用配置文件
result = await recognize_file(
    audio_path="audio.wav",
    config_path="asr/doubao/config.json"
)

# 或者
service = create_asr_service(config_path="asr/doubao/config.json")


🔧 配置参数


ASR配置 (asr_config)


参数
类型
默认值
说明


ws_url
str
wss://openspeech.bytedance.com/api/v3/sauc/bigmodel
流式识别WebSocket URL


ws_url_nostream
str
wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream
非流式识别WebSocket URL


resource_id
str
volc.bigasr.sauc.duration
资源ID


model_name
str
bigmodel
模型名称


enable_punc
bool
true
是否启用标点符号


streaming_mode
bool
true
是否启用流式模式


seg_duration
int
200
音频分段时长(ms)


mp3_seg_size
int
1000
MP3分段大小(bytes)


认证配置 (auth_config)


参数
类型
说明


app_key
str
应用密钥


access_key
str
访问密钥


音频配置 (audio_config)


参数
类型
默认值
说明


default_format
str
wav
默认音频格式


default_rate
int
16000
默认采样率


default_bits
int
16
默认位深度


default_channel
int
1
默认声道数


supported_formats
list
["wav", "mp3", "pcm"]
支持的音频格式


连接配置 (connection_config)


参数
类型
默认值
说明


max_size
int
1000000000
最大消息大小


timeout
int
30
连接超时时间(秒)


retry_times
int
3
重试次数


retry_delay
int
1
重试延迟(秒)


🎵 支持的音频格式


格式
扩展名
说明


WAV
.wav
无损音频格式，推荐使用


MP3
.mp3
压缩音频格式


PCM
.pcm
原始音频数据


音频要求


采样率: 推荐16kHz，支持8kHz、16kHz、24kHz、48kHz

位深度: 推荐16bit

声道: 推荐单声道(mono)

编码: PCM编码


🔍 API参考


便捷函数


recognize_file()


async def recognize_file(
    audio_path: str,
    app_key: str = None,
    access_key: str = None,
    config_path: str = None,
    streaming: bool = True,
    result_callback: callable = None,
    **kwargs
) -> dict:


识别音频文件。

参数:


audio_path: 音频文件路径

app_key: 应用密钥

access_key: 访问密钥

config_path: 配置文件路径

streaming: 是否使用流式识别

result_callback: 结果回调函数

**kwargs: 其他配置参数


返回: 识别结果字典


recognize_audio_data()


async def recognize_audio_data(
    audio_data: bytes,
    audio_format: str,
    app_key: str = None,
    access_key: str = None,
    config_path: str = None,
    streaming: bool = True,
    result_callback: callable = None,
    **kwargs
) -> dict:


识别音频数据。

参数:


audio_data: 音频数据(bytes)

audio_format: 音频格式("wav", "mp3", "pcm")
其他参数同recognize_file()


run_recognition()


def run_recognition(
    audio_path: str = None,
    audio_data: bytes = None,
    audio_format: str = None,
    **kwargs
) -> dict:


同步方式执行识别。


create_asr_service()


def create_asr_service(
    app_key: str = None,
    access_key: str = None,
    config_path: str = None,
    custom_config: dict = None,
    **kwargs
) -> DoubaoASRService:


创建ASR服务实例。


核心类


DoubaoASRService


主要服务类，提供高级API。

方法:


recognize_file(audio_path, result_callback=None): 识别文件

recognize_audio_data(audio_data, audio_format, result_callback=None): 识别音频数据

get_status(): 获取服务状态

close(): 关闭服务


DoubaoASRClient


底层客户端类，处理WebSocket通信。


ConfigManager


配置管理器，处理配置加载、验证、合并等。

方法:


load_config(config_path): 加载配置文件

save_config(config, config_path): 保存配置文件

validate_config(config): 验证配置

merge_configs(base_config, override_config): 合并配置

create_default_config(): 创建默认配置


DoubaoResultProcessor


结果处理器，专门处理豆包ASR流式识别结果，解决输出完整报文但只需要文本的问题。

特性:


自动提取payload_msg.result.text字段
处理流式数据覆盖更新特性
可配置日志输出级别
支持自定义文本回调函数


方法:


extract_text_from_result(result): 从完整结果中提取文本

process_streaming_result(result): 处理流式结果

create_optimized_callback(user_callback): 创建优化的回调函数

get_current_result(): 获取当前识别状态

reset(): 重置处理器状态


便捷函数:


create_text_only_callback(user_callback, enable_streaming_log): 创建只处理文本的回调

extract_text_only(result): 快速提取文本内容


使用示例:

from asr.doubao.result_processor import DoubaoResultProcessor, create_text_only_callback

# 方式1: 使用处理器类
processor = DoubaoResultProcessor(text_only=True, enable_streaming_log=False)
callback = processor.create_optimized_callback(lambda text: print(f"文本: {text}"))

# 方式2: 使用便捷函数
callback = create_text_only_callback(
    user_callback=lambda text: print(f"文本: {text}"),
    enable_streaming_log=False
)

# 在ASR服务中使用
service = create_asr_service(...)
await service.recognize_file("audio.wav", result_callback=callback)


🧪 测试

运行测试套件：

python -m pytest test/test_doubao_asr.py -v


或者直接运行测试文件：

python test/test_doubao_asr.py


测试包括：


单元测试
集成测试
性能测试
错误处理测试


🔧 故障排除


常见问题


1. 认证失败

AuthenticationError: Invalid app_key or access_key


解决方案:


检查app_key和access_key是否正确
确认密钥是否已激活
检查网络连接


2. 音频格式不支持

AudioFormatError: Unsupported audio format


解决方案:


确认音频格式为WAV、MP3或PCM
检查音频文件是否损坏
转换音频格式到支持的格式


3. 连接超时

ConnectionTimeoutError: Connection timeout


解决方案:


检查网络连接
增加timeout配置
检查防火墙设置


4. 音频文件过大

FileSizeError: Audio file too large


解决方案:


分割音频文件
压缩音频质量
使用流式识别


调试模式

启用调试模式获取详细日志：

service = create_asr_service(
    app_key="your_app_key",
    access_key="your_access_key",
    debug=True
)


或在配置文件中设置：

{
  "logging_config": {
    "enable_debug": true,
    "log_requests": true,
    "log_responses": true
  }
}


📈 性能优化


1. 连接复用

对于批量识别，使用服务实例复用连接：

service = create_asr_service(...)
try:
    for audio_file in audio_files:
        result = await service.recognize_file(audio_file)
finally:
    await service.close()


2. 并发处理

使用asyncio进行并发识别：

import asyncio

async def concurrent_recognition(audio_files):
    tasks = []
    for audio_file in audio_files:
        task = recognize_file(audio_file, ...)
        tasks.append(task)

    results = await asyncio.gather(*tasks)
    return results


3. 音频预处理


使用合适的音频格式和参数
预先分割大文件
去除静音段


🤝 贡献

欢迎提交Issue和Pull Request！


开发环境设置


克隆项目
安装依赖：pip install -r requirements.txt

运行测试：python -m pytest

提交代码前请确保测试通过


📄 许可证

MIT License


🔗 相关链接


豆包语音识别API文档
豆包开放平台
WebSocket协议


📞 支持

如有问题，请：


查看本文档的故障排除部分
搜索已有的Issue
创建新的Issue并提供详细信息
联系技术支持


作者: AIfeng

版本: 1.0.0

更新时间: 2025-07-11