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"}]