zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-17 05:07:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(stt): support Fun-ASR-Flash in Tongyi-Qianwen provider (#16844)

Browse Source 
## 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 <LauraGPT@users.noreply.github.com>
Co-authored-by: openhands <openhands@all-hands.dev>
Co-authored-by: LauraGPT <LauraGPT@users.noreply.github.com>
This commit is contained in:
Zane
2026-07-16 09:37:37 +08:00
committed by GitHub
parent bda703b588
commit eeb59ec4f2
3 changed files with 341 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
conf/llm_factories.json
View File

@@ -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,

142
rag/llm/sequence2txt_model.py
View File

@@ -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"):

193
test/unit_test/rag/llm/test_sequence2txt_model.py Normal file
View File

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