From eeb59ec4f2ac342c50c3652e8a5f0d4f63a284c4 Mon Sep 17 00:00:00 2001 From: Zane Date: Thu, 16 Jul 2026 09:37:37 +0800 Subject: [PATCH] feat(stt): support Fun-ASR-Flash in Tongyi-Qianwen provider (#16844) ## What this PR does Adds support for Alibaba Cloud's hosted Fun-ASR-Flash snapshots to the existing Tongyi-Qianwen speech-to-text provider. - registers `fun-asr-flash-2026-06-15` as a speech-to-text model; - routes only `fun-asr-flash*` models to the documented workspace-native multimodal-generation endpoint; - supports local audio through size-checked data URIs as well as URL/data-URI inputs; - uses the documented SSE response mode for incremental streaming transcription; - closes the streamed HTTP response on completion, failure, or early consumer cancellation; - preserves the existing `dashscope.MultiModalConversation` path for all other Qwen audio models; - keeps RAGFlow's existing synchronous and streaming adapter interfaces. ## Why Fun-ASR-Flash does not use the legacy Qwen audio request shape currently used by `QWenSeq2txt`. Its synchronous API expects `input_audio` at: `/api/v1/services/aigc/multimodal-generation/generation` Without a narrowly scoped adapter path, the hosted model cannot be selected successfully through RAGFlow's Tongyi-Qianwen speech-to-text provider. Closes #16843. ## Compatibility The new behavior is gated by the `fun-asr-flash` model-name prefix. Existing Qwen audio models continue through the original code path unchanged. ## Validation - `pytest test/unit_test/rag/llm/test_sequence2txt_model.py`: 10 passed - Ruff check: passed - Ruff format check: passed - `llm_factories.json` validation: passed - Real hosted-API validation with WAV audio - Real RAGFlow upload/indexing validation with MP3 audio The unit tests cover the native Fun-ASR-Flash request, regression behavior for the legacy Qwen path, SSE streaming, and early response cleanup. ## Documentation - https://help.aliyun.com/document_detail/2979031.html - https://help.aliyun.com/document_detail/2869541.html ### Why a dedicated adapter path is necessary (official evidence) Alibaba Cloud's [Fun-ASR RESTful API reference](https://help.aliyun.com/en/model-studio/fun-asr-recorded-speech-recognition-http-api) makes the incompatibilities with RAGFlow's existing Qwen audio path explicit: | Adapter change | Official API requirement | Why the existing path is insufficient | | --- | --- | --- | | Call the workspace-native HTTP endpoint | The Fun-ASR-Flash synchronous section states that SDK calls are not supported and specifies `POST /api/v1/services/aigc/multimodal-generation/generation`. | The existing adapter calls `dashscope.MultiModalConversation`, so a direct HTTP path is required. | | Use the `input_audio` message shape | `input.messages`, `content`, `type: input_audio`, `input_audio`, and `input_audio.data` are documented as required for an audio request. | The existing Qwen path sends the legacy `audio` content shape, which does not match this API contract. | | Send `parameters.format` | The request schema marks `parameters` and `format` as **Required**, and says the value must match the actual audio format. | The legacy request has no Fun-ASR-Flash `parameters.format` field, so the adapter must derive and send it. | | Encode local files as Data URIs | `input_audio.data` accepts either a public URL or a Base64 Data URI; the reference gives the exact `data:{MIME_TYPE};base64,...` form. | RAGFlow supplies local file paths, which the remote API cannot read directly. | | Parse `output.text` | The documented non-streaming response returns the accumulated transcription in `output.text`. | The legacy Qwen response parser reads `output.choices[].message.content`, so a separate response parser is required. | | Enforce the Base64 input limit | The reference requires the Base64-encoded audio to remain within the 10 MB input limit. | The adapter checks encoded size before reading/sending local audio and directs oversized inputs to the existing public-URL path. | | Use SSE for streaming | The reference specifies `X-DashScope-SSE: enable` and documents intermediate and final SSE events. | The adapter parses those events instead of wrapping one blocking response as a synthetic stream. | | Release streamed responses | Streaming responses must be closed when iteration completes or stops early. | A `finally` cleanup releases the HTTP response on completion, errors, and consumer cancellation. | `sample_rate` is documented as **Optional**. The implementation omits it instead of declaring a fixed value that may not match remote or compressed audio. The [official speech-to-text model list](https://help.aliyun.com/en/model-studio/asr-model/) separately confirms that `fun-asr-flash-2026-06-15` is an offline HTTP model with a five-minute audio limit. --------- Signed-off-by: LauraGPT Co-authored-by: openhands Co-authored-by: LauraGPT --- conf/llm_factories.json | 7 + rag/llm/sequence2txt_model.py | 142 ++++++++++++- .../rag/llm/test_sequence2txt_model.py | 193 ++++++++++++++++++ 3 files changed, 341 insertions(+), 1 deletion(-) create mode 100644 test/unit_test/rag/llm/test_sequence2txt_model.py diff --git a/conf/llm_factories.json b/conf/llm_factories.json index 46fbd10db6..288e08e5c2 100644 --- a/conf/llm_factories.json +++ b/conf/llm_factories.json @@ -441,6 +441,13 @@ ], "is_tools": false }, + { + "llm_name": "fun-asr-flash-2026-06-15", + "model_type": [ + "speech2text" + ], + "is_tools": false + }, { "llm_name": "qwen-mt-flash", "max_tokens": 8192, diff --git a/rag/llm/sequence2txt_model.py b/rag/llm/sequence2txt_model.py index 993e7a8248..cd207535d5 100644 --- a/rag/llm/sequence2txt_model.py +++ b/rag/llm/sequence2txt_model.py @@ -21,6 +21,7 @@ import re from abc import ABC import tempfile import logging +from urllib.parse import urlparse import requests from openai import OpenAI @@ -83,14 +84,34 @@ class FuturMixSeq2txt(GPTSeq2txt): class QWenSeq2txt(Base): _FACTORY_NAME = "Tongyi-Qianwen" + _FUN_ASR_FLASH_PREFIX = "fun-asr-flash" + _FUN_ASR_BASE64_MAX_SIZE = 10 * 1024 * 1024 + _DASHSCOPE_API_BASE = "https://dashscope.aliyuncs.com/api/v1" + _AUDIO_MIME_FORMATS = { + "audio/mpeg": "mp3", + "audio/mp3": "mp3", + "audio/wav": "wav", + "audio/wave": "wav", + "audio/x-wav": "wav", + } - def __init__(self, key, model_name="qwen-audio-asr", **kwargs): + def __init__(self, key, model_name="qwen-audio-asr", base_url=None, **kwargs): import dashscope dashscope.api_key = key + self.api_key = key self.model_name = model_name + self.base_url = (base_url or self._DASHSCOPE_API_BASE).rstrip("/") def transcription(self, audio_path): + # Fun-ASR-Flash uses DashScope's workspace-scoped native multimodal + # endpoint and payload instead of MultiModalConversation. + if self.model_name.startswith(self._FUN_ASR_FLASH_PREFIX): + return self._transcribe_fun_asr_flash(audio_path) + + return self._transcribe_qwen_audio(audio_path) + + def _transcribe_qwen_audio(self, audio_path): import dashscope if audio_path.startswith("http"): @@ -108,7 +129,126 @@ class QWenSeq2txt(Base): text = "**ERROR**: " + str(e) return text, num_tokens_from_string(text) + @classmethod + def _fun_asr_audio_format(cls, audio_path): + """Derive the Fun-ASR audio format from a data URI, URL, or path.""" + + if audio_path.startswith("data:"): + mime_type = audio_path[5:].split(";", 1)[0].lower() + if not mime_type.startswith("audio/"): + raise ValueError(f"Unsupported audio data URI MIME type: {mime_type or 'missing'}") + audio_format = cls._AUDIO_MIME_FORMATS.get(mime_type, mime_type.split("/", 1)[1].removeprefix("x-")) + else: + path = urlparse(audio_path).path if audio_path.startswith(("http://", "https://")) else audio_path + audio_format = os.path.splitext(path)[1].lower().lstrip(".") + + if audio_format == "wave": + audio_format = "wav" + if not audio_format: + raise ValueError("Cannot determine audio format; use a URL/path extension or an audio data URI MIME type") + return audio_format + + @classmethod + def _validate_fun_asr_base64_size(cls, encoded_size): + if encoded_size > cls._FUN_ASR_BASE64_MAX_SIZE: + raise ValueError("Fun-ASR-Flash Base64 audio exceeds the 10 MB encoded-input limit; provide a publicly accessible URL (for example, OSS) instead") + + def _fun_asr_flash_request(self, audio_path, *, stream=False): + audio_format = self._fun_asr_audio_format(audio_path) + + if audio_path.startswith(("http://", "https://")): + audio_input = audio_path + elif audio_path.startswith("data:"): + _, separator, encoded_audio = audio_path.partition(",") + if not separator: + raise ValueError("Invalid audio data URI: missing Base64 payload") + self._validate_fun_asr_base64_size(len(encoded_audio.encode("utf-8"))) + audio_input = audio_path + else: + file_size = os.path.getsize(audio_path) + encoded_size = 4 * ((file_size + 2) // 3) + self._validate_fun_asr_base64_size(encoded_size) + mime_type = "audio/mpeg" if audio_format == "mp3" else f"audio/{audio_format}" + with open(audio_path, "rb") as audio_file: + audio_input = f"data:{mime_type};base64,{base64.b64encode(audio_file.read()).decode('utf-8')}" + + api_base = self.base_url + if api_base.endswith("/compatible-mode/v1"): + api_base = api_base[: -len("/compatible-mode/v1")] + "/api/v1" + url = f"{api_base}/services/aigc/multimodal-generation/generation" + payload = { + "model": self.model_name, + "input": { + "messages": [ + { + "role": "user", + "content": [{"type": "input_audio", "input_audio": {"data": audio_input}}], + } + ] + }, + # sample_rate is optional in the Fun-ASR-Flash API. Omitting it + # avoids declaring incorrect metadata for remote or compressed audio. + "parameters": {"format": audio_format}, + } + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + "X-DashScope-SSE": "enable" if stream else "disable", + } + return url, headers, payload + + def _transcribe_fun_asr_flash(self, audio_path): + try: + url, headers, payload = self._fun_asr_flash_request(audio_path) + + response = requests.post(url, headers=headers, json=payload, timeout=60) + response.raise_for_status() + result = response.json() + text = result.get("text") or result.get("output", {}).get("text") + if not text: + raise ValueError("Missing transcription text in Fun-ASR-Flash response") + text = text.strip() + return text, num_tokens_from_string(text) + except Exception as e: + logging.exception("Fun-ASR-Flash transcription failed for model %s", self.model_name) + return "**ERROR**: " + str(e), 0 + + def _stream_fun_asr_flash(self, audio_path): + response = None + try: + url, headers, payload = self._fun_asr_flash_request(audio_path, stream=True) + response = requests.post(url, headers=headers, json=payload, timeout=60, stream=True) + response.raise_for_status() + + full = "" + for line in response.iter_lines(decode_unicode=True): + if not line or not line.startswith("data:"): + continue + event_data = line[5:].strip() + if not event_data or event_data == "[DONE]": + continue + result = json.loads(event_data) + text = result.get("text") or result.get("output", {}).get("text") + if not text: + continue + full = text.strip() + yield {"event": "delta", "text": full} + + if not full: + raise ValueError("Missing transcription text in Fun-ASR-Flash stream") + yield {"event": "final", "text": full} + except Exception as e: + logging.exception("Fun-ASR-Flash streaming transcription failed for model %s", self.model_name) + yield {"event": "error", "text": "**ERROR**: " + str(e)} + finally: + if response is not None: + response.close() + def stream_transcription(self, audio_path): + if self.model_name.startswith(self._FUN_ASR_FLASH_PREFIX): + yield from self._stream_fun_asr_flash(audio_path) + return + import dashscope if audio_path.startswith("http"): diff --git a/test/unit_test/rag/llm/test_sequence2txt_model.py b/test/unit_test/rag/llm/test_sequence2txt_model.py new file mode 100644 index 0000000000..42f3c16bfe --- /dev/null +++ b/test/unit_test/rag/llm/test_sequence2txt_model.py @@ -0,0 +1,193 @@ +# +# Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import base64 +import json +from unittest.mock import MagicMock, patch + +from rag.llm.sequence2txt_model import QWenSeq2txt + + +def test_fun_asr_flash_uses_native_request_format(tmp_path): + audio_path = tmp_path / "sample.wav" + audio_path.write_bytes(b"RIFF-test-audio") + response = MagicMock() + response.json.return_value = {"output": {"text": "transcribed text"}} + + with patch("rag.llm.sequence2txt_model.requests.post", return_value=response) as post: + model = QWenSeq2txt( + "test-key", + "fun-asr-flash-2026-06-15", + base_url="https://workspace.example.com/compatible-mode/v1", + ) + text, _ = model.transcription(str(audio_path)) + + assert text == "transcribed text" + response.raise_for_status.assert_called_once_with() + request = post.call_args + assert request.args[0] == "https://workspace.example.com/api/v1/services/aigc/multimodal-generation/generation" + assert request.kwargs["headers"]["X-DashScope-SSE"] == "disable" + assert request.kwargs["json"]["parameters"] == {"format": "wav"} + audio_data = request.kwargs["json"]["input"]["messages"][0]["content"][0]["input_audio"]["data"] + assert audio_data == f"data:audio/wav;base64,{base64.b64encode(audio_path.read_bytes()).decode('utf-8')}" + + +def test_qwen_audio_asr_keeps_existing_dashscope_path(): + response = {"output": {"choices": [{"message": MagicMock(content=[{"text": "legacy text"}])}]}} + + with patch("dashscope.MultiModalConversation.call", return_value=response) as call: + model = QWenSeq2txt("test-key", "qwen-audio-asr") + text, _ = model.transcription("https://example.com/sample.wav") + + assert text == "legacy text" + call.assert_called_once_with( + model="qwen-audio-asr", + messages=[ + {"role": "system", "content": [{"text": ""}]}, + {"role": "user", "content": [{"audio": "https://example.com/sample.wav"}]}, + ], + result_format="message", + asr_options={"enable_lid": True, "enable_itn": False}, + ) + + +def test_fun_asr_flash_stream_uses_sse(): + response = MagicMock() + response.iter_lines.return_value = [ + "id:1", + "event:result", + f"data:{json.dumps({'output': {'text': 'stream'}})}", + "", + "id:2", + "event:result", + f"data:{json.dumps({'output': {'text': 'stream text'}})}", + "", + ] + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + + with patch("rag.llm.sequence2txt_model.requests.post", return_value=response) as post: + events = list(model.stream_transcription("data:audio/wav;base64,dGVzdA==")) + + response.raise_for_status.assert_called_once_with() + assert post.call_args.kwargs["headers"]["X-DashScope-SSE"] == "enable" + assert post.call_args.kwargs["stream"] is True + assert events == [ + {"event": "delta", "text": "stream"}, + {"event": "delta", "text": "stream text"}, + {"event": "final", "text": "stream text"}, + ] + + +def test_fun_asr_flash_stream_closes_response_when_consumer_stops_early(): + response = MagicMock() + response.iter_lines.return_value = [f"data:{json.dumps({'output': {'text': 'stream'}})}"] + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + + with patch("rag.llm.sequence2txt_model.requests.post", return_value=response): + stream = model.stream_transcription("data:audio/wav;base64,dGVzdA==") + assert next(stream) == {"event": "delta", "text": "stream"} + stream.close() + + response.close.assert_called_once_with() + + +def test_fun_asr_flash_handles_top_level_text_response(): + response = MagicMock() + response.json.return_value = {"text": "transcribed text"} + + with patch("rag.llm.sequence2txt_model.requests.post", return_value=response): + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + text, _ = model.transcription("data:audio/wav;base64,dGVzdA==") + + assert text == "transcribed text" + + +def test_fun_asr_flash_derives_format_from_data_uri(): + response = MagicMock() + response.json.return_value = {"output": {"text": "transcribed text"}} + audio_data = "data:audio/mpeg;base64,dGVzdA==" + + with patch("rag.llm.sequence2txt_model.requests.post", return_value=response) as post: + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + text, _ = model.transcription(audio_data) + + assert text == "transcribed text" + assert post.call_args.kwargs["json"]["parameters"] == {"format": "mp3"} + assert post.call_args.kwargs["json"]["input"]["messages"][0]["content"][0]["input_audio"]["data"] == audio_data + + +def test_fun_asr_flash_derives_format_from_url_path(): + response = MagicMock() + response.json.return_value = {"output": {"text": "transcribed text"}} + audio_url = "https://example.com/sample.opus?signature=test" + + with patch("rag.llm.sequence2txt_model.requests.post", return_value=response) as post: + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + text, _ = model.transcription(audio_url) + + assert text == "transcribed text" + assert post.call_args.kwargs["json"]["parameters"] == {"format": "opus"} + + +def test_fun_asr_flash_rejects_extensionless_url(caplog): + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + + with patch("rag.llm.sequence2txt_model.requests.post") as post: + text, tokens = model.transcription("https://example.com/audio") + + post.assert_not_called() + assert text.startswith("**ERROR**: Cannot determine audio format") + assert tokens == 0 + assert "Fun-ASR-Flash transcription failed" in caplog.text + + +def test_fun_asr_flash_rejects_local_audio_over_base64_limit(): + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + base64_limit = 8 + largest_allowed_raw_size = (base64_limit // 4) * 3 + + with ( + patch.object(QWenSeq2txt, "_FUN_ASR_BASE64_MAX_SIZE", base64_limit), + patch("rag.llm.sequence2txt_model.os.path.getsize", return_value=largest_allowed_raw_size + 1), + patch("rag.llm.sequence2txt_model.requests.post") as post, + ): + text, tokens = model.transcription("large.wav") + + post.assert_not_called() + assert text.startswith("**ERROR**: Fun-ASR-Flash Base64 audio exceeds the 10 MB encoded-input limit") + assert tokens == 0 + + +def test_fun_asr_flash_rejects_data_uri_over_base64_limit(): + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + base64_limit = 8 + audio_data = f"data:audio/wav;base64,{'A' * (base64_limit + 1)}" + + with patch.object(QWenSeq2txt, "_FUN_ASR_BASE64_MAX_SIZE", base64_limit), patch("rag.llm.sequence2txt_model.requests.post") as post: + text, tokens = model.transcription(audio_data) + + post.assert_not_called() + assert text.startswith("**ERROR**: Fun-ASR-Flash Base64 audio exceeds the 10 MB encoded-input limit") + assert tokens == 0 + + +def test_fun_asr_flash_stream_emits_only_error_event_on_failure(): + model = QWenSeq2txt("test-key", "fun-asr-flash-2026-06-15") + + with patch("rag.llm.sequence2txt_model.requests.post", side_effect=RuntimeError("failed")): + events = list(model.stream_transcription("data:audio/wav;base64,dGVzdA==")) + + assert events == [{"event": "error", "text": "**ERROR**: failed"}]