Merge pull request #3124 from xinnan-tech/py-add-ASRv2

使用双流优化版接口（速度提升），增加豆包流式语音识别模型2.0模型选择
2026-07-22 07:03:53 +08:00 · 2026-04-22 11:29:57 +08:00
parent 8a156e99f4 e9a2489e48
commit ddfbacc717
5 changed files with 97 additions and 106 deletions
@@ -0,0 +1,45 @@
+-- 修复豆包语音合成模型2.0 provider_code 重复问题，添加 ASR 2.0 支持
+
+-- ==================== 豆包语音合成模型2.0 ====================
+-- 删除 TTS 2.0 供应器（不再需要单独的供应器）
+delete from `ai_model_provider` where id = 'SYSTEM_TTS_HSDSTTS_V2';
+
+-- ==================== 豆包语音识别(流式) ====================
+-- 修正原有豆包语音识别(流式)供应器，移除cluster字段，添加resource_id字段
+UPDATE `ai_model_provider` SET `fields` = '[{"key":"appid","type":"string","label":"应用ID"},{"key":"access_token","type":"string","label":"访问令牌"},{"key":"boosting_table_name","type":"string","label":"热词文件名称"},{"key":"correct_table_name","type":"string","label":"替换词文件名称"},{"key":"output_dir","type":"string","label":"输出目录"},{"key":"end_window_size","type":"number","label":"静音判定时长(ms)"},{"key":"enable_multilingual","type":"boolean","label":"是否开启多语种识别模式"},{"key":"language","type":"string","label":"指定语言编码"},{"key":"resource_id","type":"string","label":"资源ID"}]' WHERE `id` = 'SYSTEM_ASR_DoubaoStreamASR';
+
+-- 修正原有豆包语音识别(流式)配置，移除cluster字段，添加resource_id默认值
+UPDATE `ai_model_config` SET `config_json` = JSON_REMOVE(JSON_SET(`config_json`, '$.resource_id', 'volc.bigasr.sauc.duration'), '$.cluster') WHERE `id` = 'ASR_DoubaoStreamASR';
+
+-- ==================== 豆包语音识别模型2.0 ====================
+
+-- 插入豆包语音识别模型2.0配置
+delete from `ai_model_config` where id = 'ASR_DoubaoStreamASRV2';
+INSERT INTO `ai_model_config` VALUES ('ASR_DoubaoStreamASRV2', 'ASR', 'DoubaoStreamASRV2', '豆包语音识别模型2.0', 0, 1, '{
+  "type": "doubao_stream",
+  "appid": "",
+  "access_token": "",
+  "resource_id": "volc.seedasr.sauc.duration",
+  "end_window_size": 200,
+  "enable_multilingual": false,
+  "language": "zh-CN",
+  "output_dir": "tmp/"
+}', NULL, NULL, 6, NULL, NULL, NULL, NULL);
+
+-- 豆包语音识别模型2.0配置说明文档
+UPDATE `ai_model_config` SET
+`doc_link` = 'https://www.volcengine.com/docs/6561/109979',
+`remark` = '豆包语音识别模型2.0配置说明（基于火山引擎seed-asr）：
+1. 访问 https://www.volcengine.com/ 注册并开通火山引擎账号
+2. 访问 https://console.volcengine.com/speech/service/10038 开通豆包流式语音识别模型2.0
+3. 在页面底部获取appid和access_token
+4. 资源ID有两种：小时版（volc.seedasr.sauc.duration）和并发版（volc.seedasr.sauc.concurrent）
+   - 小时版：固定为：volc.seedasr.sauc.duration（豆包语音识别模型2.0）
+   - 并发版：固定为：volc.seedasr.sauc.concurrent（豆包语音识别模型2.0）
+
+详细参数文档：https://www.volcengine.com/docs/6561/109979
+
+注意：
+- 豆包语音识别模型2.0使用volc.seedasr.sauc.duration资源ID，与豆包语音识别(流式)（volc.bigasr.sauc.duration）不同
+- 语音识别模型2.0价格更为便宜，建议在高并发场景下使用并发版资源ID
+' WHERE `id` = 'ASR_DoubaoStreamASRV2';
@@ -620,3 +620,10 @@ databaseChangeLog:
        - sqlFile:
            encoding: utf8
            path: classpath:db/changelog/202604201719.sql
+  - changeSet:
+      id: 202604211700
+      author: RanChen
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202604211700.sql
@@ -396,7 +396,28 @@ ASR:
    type: doubao_stream
    appid: 你的火山引擎语音合成服务appid
    access_token: 你的火山引擎语音合成服务access_token
-    cluster: volcengine_input_common
+    # 资源ID：小时版 volc.bigasr.sauc.duration，并发版 volc.bigasr.sauc.concurrent
+    resource_id: volc.bigasr.sauc.duration
+    # 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738
+    boosting_table_name: （选填）你的热词文件名称
+    correct_table_name: （选填）你的替换词文件名称
+    # 是否开启多语种识别模式
+    enable_multilingual: False
+    # 多语种识别当该键为空时，该模型支持中英文、上海话、闽南语，四川、陕西、粤语识别。当将其设置为特定键时，它可以识别指定语言。
+    # 详细语言列表参考 https://www.volcengine.com/docs/6561/1354869
+    # language: zh-cn
+    # 静音判定时长(ms)，默认200ms
+    end_window_size: 200
+    output_dir: tmp/
+  DoubaoStreamASRV2:
+    # 豆包语音识别模型2.0（基于火山引擎seed-asr）
+    # 开通地址：https://console.volcengine.com/speech/service/10038
+    # 价格更为便宜，建议在高并发场景下使用
+    type: doubao_stream
+    appid: 你的火山引擎语音合成服务appid
+    access_token: 你的火山引擎语音合成服务access_token
+    # 资源ID：小时版 volc.seedasr.sauc.duration，并发版 volc.seedasr.sauc.concurrent
+    resource_id: volc.seedasr.sauc.duration
    # 热词、替换词使用流程：https://www.volcengine.com/docs/6561/155738
    boosting_table_name: （选填）你的热词文件名称
    correct_table_name: （选填）你的替换词文件名称
@@ -29,8 +29,10 @@ class ASRProvider(ASRProviderBase):

        # 配置参数
        self.appid = str(config.get("appid"))
-        self.cluster = config.get("cluster")
        self.access_token = config.get("access_token")
+        # 资源ID，用于区分不同的ASR模型（默认1.0模型小时版，v2版本使用seed-asr）
+        self.resource_id = config.get("resource_id", "volc.bigasr.sauc.duration")
+
        self.boosting_table_name = config.get("boosting_table_name", "")
        self.correct_table_name = config.get("correct_table_name", "")
        self.output_dir = config.get("output_dir", "tmp/")
@@ -44,7 +46,7 @@ class ASRProvider(ASRProviderBase):
        if self.enable_multilingual:
            self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream"
        else:
-            self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel"
+            self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_async"
        self.uid = config.get("uid", "streaming_asr_service")
        self.workflow = config.get(
            "workflow", "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate"
@@ -283,7 +285,6 @@ class ASRProvider(ASRProviderBase):
        req = {
            "app": {
                "appid": self.appid,
-                "cluster": self.cluster,
                "token": self.access_token,
            },
            "user": {"uid": self.uid},
@@ -322,7 +323,7 @@ class ASRProvider(ASRProviderBase):
        return {
            "X-Api-App-Key": self.appid,
            "X-Api-Access-Key": self.access_token,
-            "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
+            "X-Api-Resource-Id": self.resource_id,
            "X-Api-Connect-Id": str(uuid.uuid4()),
        }

@@ -385,9 +386,17 @@ class ASRProvider(ASRProviderBase):
                    "payload_msg": error_msg,
                }

-            # 获取JSON数据（跳过12字节头部）
+            # 获取JSON数据
            try:
-                json_data = res[12:].decode("utf-8")
+                # 检查字节8-11是否为有效的JSON长度字段
+                # 格式：4字节头 + 4字节序列号 + 4字节长度 + JSON数据
+                length = int.from_bytes(res[8:12], "big")
+                if length > 0 and length <= len(res) - 12:
+                    # 有长度字段，从字节12开始读取指定长度的JSON
+                    json_data = res[12:12 + length].decode("utf-8")
+                else:
+                    # 无长度字段或长度无效，尝试直接解析
+                    json_data = res[8:].decode("utf-8")
                result = json.loads(json_data)
                logger.bind(tag=TAG).debug(f"成功解析JSON响应: {result}")
                return {"payload_msg": result}
@@ -65,69 +65,8 @@ class BaseASRTester:
 class DoubaoStreamASRTester(BaseASRTester):
    def __init__(self):
        super().__init__("DoubaoStreamASR")
-
-    def _generate_header(
-        self,
-        version=0x01,
-        message_type=0x01,
-        message_type_specific_flags=0x00,
-        serial_method=0x01,
-        compression_type=0x01,
-        reserved_data=0x00,
-        extension_header: bytes = b"",
-    ):
-        """生成协议头（修复：使用正确的Header格式）"""
-        header = bytearray()
-        header_size = int(len(extension_header) / 4) + 1
-        header.append((version << 4) | header_size)
-        header.append((message_type << 4) | message_type_specific_flags)
-        header.append((serial_method << 4) | compression_type)
-        header.append(reserved_data)
-        header.extend(extension_header)
-        return header
-
-    def _generate_audio_default_header(self):
-        """生成音频数据Header"""
-        return self._generate_header(
-            version=0x01,
-            message_type=0x02,
-            message_type_specific_flags=0x00,  # 普通音频帧
-            serial_method=0x01,
-            compression_type=0x01,
-        )
-
-    def _generate_last_audio_header(self):
-        """生成最后一帧音频的Header（标记音频结束）"""
-        return self._generate_header(
-            version=0x01,
-            message_type=0x02,
-            message_type_specific_flags=0x02,  # 0x02表示这是最后一帧
-            serial_method=0x01,
-            compression_type=0x01,
-        )
-
-    def _parse_response(self, res: bytes) -> dict:
-        try:
-            if len(res) < 4:
-                return {"error": "响应数据长度不足"}
-            header = res[:4]
-            message_type = header[1] >> 4
-            if message_type == 0x0F:
-                code = int.from_bytes(res[4:8], "big", signed=False)
-                msg_length = int.from_bytes(res[8:12], "big", signed=False)
-                error_msg = json.loads(res[12:].decode("utf-8"))
-                return {
-                    "code": code,
-                    "msg_length": msg_length,
-                    "payload_msg": error_msg
-                }
-            try:
-                json_data = res[12:].decode("utf-8")
-                return {"payload_msg": json.loads(json_data)}
-            except (UnicodeDecodeError, json.JSONDecodeError):
-                return {"error": "JSON解析失败"}
-        except Exception:
-            return {"error": "解析响应失败"}
+        from core.providers.asr.doubao_stream import ASRProvider as DoubaoStreamProvider
+        self.provider = DoubaoStreamProvider(self.asr_config, delete_audio_file=False)

    async def test(self, test_count=5):
        if not self.test_audio_files:
@@ -138,59 +77,29 @@ class DoubaoStreamASRTester(BaseASRTester):
        latencies = []
        for i in range(test_count):
            try:
-                ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel"
-                appid = self.asr_config["appid"]
-                access_token = self.asr_config["access_token"]
-                cluster = self.asr_config.get("cluster", "volcengine_input_common")
-                uid = self.asr_config.get("uid", "streaming_asr_service")
-
                start_time = time.time()

-                headers = {
-                    "X-Api-App-Key": appid,
-                    "X-Api-Access-Key": access_token,
-                    "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
-                    "X-Api-Connect-Id": str(uuid.uuid4())
-                }
+                headers = self.provider.token_auth()

                async with websockets.connect(
-                    ws_url,
+                    self.provider.ws_url,
                    additional_headers=headers,
                    max_size=1000000000,
                    ping_interval=None,
                    ping_timeout=None,
                    close_timeout=10
                ) as ws:
-                    request_params = {
-                        "app": {"appid": appid, "cluster": cluster, "token": access_token},
-                        "user": {"uid": uid},
-                        "request": {
-                            "reqid": str(uuid.uuid4()),
-                            "workflow": "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate",
-                            "show_utterances": True,
-                            "result_type": "single",
-                            "sequence": 1
-                        },
-                        "audio": {
-                            "format": "pcm",
-                            "codec": "pcm",
-                            "rate": 16000,
-                            "language": "zh-CN",
-                            "bits": 16,
-                            "channel": 1,
-                            "sample_rate": 16000
-                        }
-                    }
+                    request_params = self.provider.construct_request(str(uuid.uuid4()))

                    payload_bytes = str.encode(json.dumps(request_params))
                    payload_bytes = gzip.compress(payload_bytes)
-                    full_client_request = self._generate_header()
+                    full_client_request = bytearray(self.provider.generate_header())
                    full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
                    full_client_request.extend(payload_bytes)
                    await ws.send(full_client_request)

                    init_res = await ws.recv()
-                    result = self._parse_response(init_res)
+                    result = self.provider.parse_response(init_res)
                    if "code" in result and result["code"] != 1000:
                        raise Exception(f"初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}")

@@ -200,7 +109,7 @@ class DoubaoStreamASRTester(BaseASRTester):

                    # 发送音频数据（使用最后一帧标记，告诉服务端音频已结束）
                    payload = gzip.compress(audio_data)
-                    audio_request = bytearray(self._generate_last_audio_header())  # 修复：使用最后一帧Header
+                    audio_request = bytearray(self.provider.generate_last_audio_default_header())
                    audio_request.extend(len(payload).to_bytes(4, "big"))
                    audio_request.extend(payload)
                    await ws.send(audio_request)