mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-14 19:45:08 +08:00
fix(stt): replace DashScope realtime WS replay with synchronous Qwen3-ASR HTTP recognition (#580)
This commit is contained in:
mateaix
parent
54a2c3c0a3
commit
a578c8ef0e
@ -119,8 +119,8 @@ public class TalkModeWebSocketHandler extends AbstractWebSocketHandler {
|
|||||||
|
|
||||||
// 2. STT: 音频转文字
|
// 2. STT: 音频转文字
|
||||||
// 前端用 WavRecorder（Web Audio API + 手写 PCM WAV 编码）— 见
|
// 前端用 WavRecorder（Web Audio API + 手写 PCM WAV 编码）— 见
|
||||||
// mateclaw-ui/src/utils/wavEncoder.ts. WebM/Opus 被 DashScope
|
// mateclaw-ui/src/utils/wavEncoder.ts. WAV 是所有 STT provider
|
||||||
// Paraformer 拒收，WAV 是所有 STT provider 都接受的最大公约数。
|
// 都接受的最大公约数，也让后端能对 PCM 做静音预检。
|
||||||
Map<String, Object> sttResult = sttService.transcribe(audioData, "audio.wav", "audio/wav", null);
|
Map<String, Object> sttResult = sttService.transcribe(audioData, "audio.wav", "audio/wav", null);
|
||||||
if (!Boolean.TRUE.equals(sttResult.get("success"))) {
|
if (!Boolean.TRUE.equals(sttResult.get("success"))) {
|
||||||
sendJson(session, Map.of("type", "error", "message", "Speech recognition failed: " + sttResult.get("error")));
|
sendJson(session, Map.of("type", "error", "message", "Speech recognition failed: " + sttResult.get("error")));
|
||||||
|
|||||||
@ -6,21 +6,18 @@ import java.nio.ByteOrder;
|
|||||||
/**
|
/**
|
||||||
* Strip the RIFF/WAVE header off a WAV blob to expose raw PCM samples.
|
* Strip the RIFF/WAVE header off a WAV blob to expose raw PCM samples.
|
||||||
*
|
*
|
||||||
* <p>DashScope's realtime ASR expects the {@code parameters.format = "pcm"}
|
* <p>Used for pre-flight audio diagnostics: the web recorder (see
|
||||||
* input as **bare 16-bit signed little-endian PCM**, not WAV. The frontend
|
* {@code mateclaw-ui/src/utils/wavEncoder.ts}) emits a 16 kHz mono 16-bit
|
||||||
* (see {@code mateclaw-ui/src/utils/wavEncoder.ts}) emits a 16 kHz mono
|
* WAV with the canonical 44-byte header, and unwrapping it lets STT
|
||||||
* 16-bit WAV with the canonical 44-byte header — this helper unwraps it.
|
* providers run a peak/RMS silence check on the raw samples before paying
|
||||||
*
|
* for a recognition call — "mic captured nothing" then surfaces as a
|
||||||
* <p>Why not just send the WAV: DashScope rejects with "format mismatch"
|
* precise local error instead of an empty transcript.
|
||||||
* because the first 44 bytes look like garbage when interpreted as PCM
|
|
||||||
* samples — they're the RIFF magic + format chunk metadata.
|
|
||||||
*
|
*
|
||||||
* <p>Limitations: handles only the canonical 44-byte WAV layout produced by
|
* <p>Limitations: handles only the canonical 44-byte WAV layout produced by
|
||||||
* MateClaw's WavRecorder. WAVs with extra chunks (LIST, JUNK, …) before the
|
* MateClaw's WavRecorder. WAVs with extra chunks (LIST, JUNK, …) before the
|
||||||
* data chunk would need a chunk-walking parser. We don't currently accept
|
* data chunk would need a chunk-walking parser. Callers should gate on
|
||||||
* arbitrary uploads, so the tighter scope is fine; if this changes,
|
* {@link #isCanonicalWav} and skip the diagnostics for anything else,
|
||||||
* extend {@link #extract} to scan for the {@code "data"} chunk header
|
* rather than treating non-WAV input as an error.
|
||||||
* instead of assuming offset 36.
|
|
||||||
*/
|
*/
|
||||||
public final class WavPcmExtractor {
|
public final class WavPcmExtractor {
|
||||||
|
|
||||||
@ -32,6 +29,17 @@ public final class WavPcmExtractor {
|
|||||||
|
|
||||||
private WavPcmExtractor() {}
|
private WavPcmExtractor() {}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* True when the bytes carry the RIFF/WAVE magic and are long enough to
|
||||||
|
* hold the canonical 44-byte header. Cheap gate for callers that only
|
||||||
|
* want PCM diagnostics on inputs {@link #extract} can actually handle.
|
||||||
|
*/
|
||||||
|
public static boolean isCanonicalWav(byte[] bytes) {
|
||||||
|
return bytes != null && bytes.length >= CANONICAL_HEADER_BYTES
|
||||||
|
&& bytes[0] == 'R' && bytes[1] == 'I' && bytes[2] == 'F' && bytes[3] == 'F'
|
||||||
|
&& bytes[8] == 'W' && bytes[9] == 'A' && bytes[10] == 'V' && bytes[11] == 'E';
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Extract raw PCM bytes from a WAV blob. Throws when the input is too short
|
* Extract raw PCM bytes from a WAV blob. Throws when the input is too short
|
||||||
* or the magic header bytes don't look like RIFF/WAVE — better to fail loud
|
* or the magic header bytes don't look like RIFF/WAVE — better to fail loud
|
||||||
|
|||||||
@ -1,114 +1,96 @@
|
|||||||
package vip.mate.stt.provider;
|
package vip.mate.stt.provider;
|
||||||
|
|
||||||
|
import cn.hutool.http.HttpRequest;
|
||||||
|
import cn.hutool.http.HttpResponse;
|
||||||
import com.fasterxml.jackson.databind.JsonNode;
|
import com.fasterxml.jackson.databind.JsonNode;
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
import vip.mate.llm.service.ModelProviderService;
|
import vip.mate.llm.service.ModelProviderService;
|
||||||
|
import vip.mate.stt.AudioMimeTypes;
|
||||||
import vip.mate.stt.SttProvider;
|
import vip.mate.stt.SttProvider;
|
||||||
import vip.mate.stt.SttRequest;
|
import vip.mate.stt.SttRequest;
|
||||||
import vip.mate.stt.SttResult;
|
import vip.mate.stt.SttResult;
|
||||||
import vip.mate.stt.WavPcmExtractor;
|
import vip.mate.stt.WavPcmExtractor;
|
||||||
import vip.mate.system.model.SystemSettingsDTO;
|
import vip.mate.system.model.SystemSettingsDTO;
|
||||||
|
|
||||||
import java.net.URI;
|
import java.util.Base64;
|
||||||
import java.net.http.HttpClient;
|
|
||||||
import java.net.http.WebSocket;
|
|
||||||
import java.nio.ByteBuffer;
|
|
||||||
import java.time.Duration;
|
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.UUID;
|
|
||||||
import java.util.concurrent.CompletionStage;
|
|
||||||
import java.util.concurrent.CountDownLatch;
|
|
||||||
import java.util.concurrent.ExecutionException;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
|
||||||
import java.util.concurrent.TimeoutException;
|
|
||||||
import java.util.concurrent.atomic.AtomicReference;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* DashScope STT Provider — Paraformer Realtime via WebSocket.
|
* DashScope STT Provider — synchronous HTTP recognition via Qwen3-ASR.
|
||||||
*
|
*
|
||||||
* <p>DashScope's only sync-callable STT path is the realtime WebSocket API
|
* <p>Recognizes a complete recorded clip with a single
|
||||||
* — there is no <code>/audio/transcriptions</code> endpoint on either the
|
* {@code POST /compatible-mode/v1/chat/completions} call: the audio travels
|
||||||
* native or OpenAI-compatible HTTP surface (verified empirically, returns
|
* as a base64 {@code input_audio} content part and the transcript comes back
|
||||||
* 404). The earlier sync-HTTP version of this provider was speculative and
|
* as the assistant message content. One request, one response — no session
|
||||||
* has been replaced by this one.
|
* protocol, no timing constraints.
|
||||||
*
|
*
|
||||||
* <h2>Wire protocol</h2>
|
* <h2>Why HTTP recognition instead of the realtime WebSocket</h2>
|
||||||
* Documented at <i>Aliyun DashScope Realtime ASR</i>. Message exchange:
|
* An earlier version of this provider replayed the recorded clip through the
|
||||||
* <ol>
|
* {@code paraformer-realtime-v2} WebSocket. That endpoint is built for live
|
||||||
* <li>Open WS to {@value #WS_URL} with {@code Authorization: bearer
|
* microphone streams: its server-side VAD assumes audio arrives at wall-clock
|
||||||
* <api-key>} header.</li>
|
* pace, and a replayed clip that falls outside that contract is silently
|
||||||
* <li>Client sends a {@code run-task} text frame with task_id +
|
* discarded — the protocol completes cleanly ({@code task-started} →
|
||||||
* paraformer-realtime-v2 model + format/sample-rate parameters.</li>
|
* {@code task-finished}) with <b>zero</b> {@code result-generated} events
|
||||||
* <li>Server replies with {@code task-started} text frame.</li>
|
* (see issue #580). Pacing the replay with 100ms sleeps per chunk made it
|
||||||
* <li>Client streams raw 16-bit PCM bytes as binary frames (chunked at
|
* work in some environments, but:
|
||||||
* ~100ms each = {@value #CHUNK_BYTES} bytes for 16 kHz mono).</li>
|
* <ul>
|
||||||
* <li>Server emits {@code result-generated} events as transcripts come
|
* <li>the VAD sensitivity remained — users still hit 0-event failures;</li>
|
||||||
* in. Each event carries a sentence keyed by {@code begin_time};
|
* <li>every transcription cost at least the clip's own duration in
|
||||||
* later events with the same {@code begin_time} update the same
|
* wall-clock time (10s of speech ≥ 10s of paced streaming), with a
|
||||||
* sentence (interim → final).</li>
|
* worker thread parked in {@code Thread.sleep} the whole way;</li>
|
||||||
* <li>Client sends {@code finish-task} text frame; server replies with
|
* <li>only canonical 16-bit PCM WAV could be sent, so voice notes from IM
|
||||||
* {@code task-finished}; both sides close.</li>
|
* channels (ogg/opus/amr/m4a) always failed over to Whisper.</li>
|
||||||
* </ol>
|
* </ul>
|
||||||
|
* The synchronous recognition endpoint is the purpose-built API for
|
||||||
|
* "recorded clip in, text out": latency is a single round trip regardless of
|
||||||
|
* clip length, and it accepts wav/mp3/ogg/opus/m4a/amr/webm and more, which
|
||||||
|
* also makes IM-channel voice notes first-class here.
|
||||||
*
|
*
|
||||||
* <p>The {@link SttProvider} interface is sync — we bridge the async WS
|
* <h2>Wire format</h2>
|
||||||
* conversation to a blocking call via {@link CountDownLatch} (run-task ack
|
* OpenAI-compatible chat completion with an audio content part:
|
||||||
* + task-finished ack) plus an overall hard timeout. The whole transcribe
|
* <pre>{@code
|
||||||
* call returns either a full transcript or a domain-typed
|
* {"model":"qwen3-asr-flash",
|
||||||
* {@link SttResult#failure} after at most {@value #OVERALL_TIMEOUT_MS}ms.
|
* "messages":[{"role":"user","content":[
|
||||||
|
* {"type":"input_audio","input_audio":{"data":"data:;base64,...","format":"wav"}}]}],
|
||||||
|
* "stream":false,
|
||||||
|
* "asr_options":{"language":"zh"}} // omitted → auto language detection
|
||||||
|
* }</pre>
|
||||||
|
* Response: standard chat completion; transcript at
|
||||||
|
* {@code choices[0].message.content}. Errors arrive as HTTP 4xx/5xx with an
|
||||||
|
* {@code error.code} / {@code error.message} body.
|
||||||
*/
|
*/
|
||||||
@Slf4j
|
@Slf4j
|
||||||
@Component
|
@Component
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
public class DashScopeSttProvider implements SttProvider {
|
public class DashScopeSttProvider implements SttProvider {
|
||||||
|
|
||||||
/** DashScope WS endpoint for realtime inference (audio/text/multimodal). */
|
/** OpenAI-compatible chat completions endpoint carrying ASR requests. */
|
||||||
static final URI WS_URL = URI.create("wss://dashscope.aliyuncs.com/api-ws/v1/inference/");
|
static final String ASR_ENDPOINT =
|
||||||
|
"https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions";
|
||||||
|
|
||||||
/** Default model — paraformer-realtime-v2 is the canonical 2024+ realtime ASR. */
|
/** Default recognition model — multilingual, auto language detection. */
|
||||||
static final String DEFAULT_MODEL = "paraformer-realtime-v2";
|
static final String DEFAULT_MODEL = "qwen3-asr-flash";
|
||||||
|
|
||||||
/** Default sample rate in Hz. Must match the actual WAV — the helper reads it. */
|
/** Overall budget for the single HTTP round trip. */
|
||||||
static final int DEFAULT_SAMPLE_RATE_HZ = 16_000;
|
static final int HTTP_TIMEOUT_MS = 60_000;
|
||||||
|
|
||||||
/** ~100ms of 16 kHz / 16-bit / mono PCM. DashScope recommends 100-300ms chunks. */
|
|
||||||
static final int CHUNK_BYTES = 3200;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* How long to sleep between chunks. Paraformer-Realtime expects audio to
|
|
||||||
* arrive at roughly the natural recording rate; if we dump the whole clip
|
|
||||||
* in tens of milliseconds the server discards the stream and replies with
|
|
||||||
* task-finished + zero result-generated events. The official Python SDK
|
|
||||||
* does the same with {@code time.sleep(0.1)} between chunks. Matches
|
|
||||||
* {@link #CHUNK_BYTES} (100ms of audio → 100ms wall sleep).
|
|
||||||
*/
|
|
||||||
static final long CHUNK_PACING_MS = 100L;
|
|
||||||
|
|
||||||
/** How long to wait for the WS handshake + task-started ack before giving up. */
|
|
||||||
static final long TASK_STARTED_TIMEOUT_MS = 10_000L;
|
|
||||||
|
|
||||||
/** Overall budget for a single transcribe — beyond this we abort the WS. */
|
|
||||||
static final long OVERALL_TIMEOUT_MS = 60_000L;
|
|
||||||
|
|
||||||
private final ModelProviderService modelProviderService;
|
private final ModelProviderService modelProviderService;
|
||||||
private final ObjectMapper objectMapper;
|
private final ObjectMapper objectMapper;
|
||||||
/** Shared HttpClient — JDK's WebSocket builder doesn't reuse the underlying
|
|
||||||
* connection pool when you allocate a fresh client per call, so making
|
|
||||||
* this a field saves a connection-pool spin-up on every transcribe. */
|
|
||||||
private final HttpClient httpClient = HttpClient.newHttpClient();
|
|
||||||
|
|
||||||
@Override public String id() { return "dashscope"; }
|
@Override public String id() { return "dashscope"; }
|
||||||
@Override public String label() { return "DashScope (Paraformer Realtime)"; }
|
@Override public String label() { return "DashScope (Qwen3 ASR)"; }
|
||||||
@Override public boolean requiresCredential() { return true; }
|
@Override public boolean requiresCredential() { return true; }
|
||||||
@Override public int autoDetectOrder() { return 150; }
|
@Override public int autoDetectOrder() { return 150; }
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Per-language priority. Paraformer is the strongest mainstream Chinese
|
* Per-language priority. DashScope's ASR family is the strongest
|
||||||
* STT, so push it ahead of Whisper on zh — see {@link SttProvider} javadoc
|
* mainstream Chinese STT, so push it ahead of Whisper on zh — see
|
||||||
* for the routing rationale.
|
* {@link SttProvider} javadoc for the routing rationale.
|
||||||
*/
|
*/
|
||||||
@Override
|
@Override
|
||||||
public int autoDetectOrder(String language) {
|
public int autoDetectOrder(String language) {
|
||||||
@ -136,124 +118,51 @@ public class DashScopeSttProvider implements SttProvider {
|
|||||||
return SttResult.failure("DashScope API Key 未配置");
|
return SttResult.failure("DashScope API Key 未配置");
|
||||||
}
|
}
|
||||||
byte[] audio = request.getAudioData();
|
byte[] audio = request.getAudioData();
|
||||||
if (audio == null || audio.length < WavPcmExtractor.CANONICAL_HEADER_BYTES) {
|
if (audio == null || audio.length == 0) {
|
||||||
return SttResult.failure("音频为空或过短");
|
return SttResult.failure("音频为空");
|
||||||
}
|
|
||||||
byte[] pcm = WavPcmExtractor.extract(audio);
|
|
||||||
int sampleRate = WavPcmExtractor.sampleRate(audio);
|
|
||||||
String model = request.getModel() != null ? request.getModel() : DEFAULT_MODEL;
|
|
||||||
String taskId = UUID.randomUUID().toString().replace("-", "");
|
|
||||||
|
|
||||||
// Peak/RMS check — the silence path is a failure mode worth its
|
|
||||||
// own log line so users can tell "mic captured nothing" from
|
|
||||||
// "DashScope rejected real audio". Successful calls log peak/rms
|
|
||||||
// at DEBUG only; a healthy call shouldn't produce a per-request
|
|
||||||
// INFO log every time the user holds the talk button.
|
|
||||||
int[] peakRms = computePcmPeakRms(pcm);
|
|
||||||
if (peakRms[0] == 0) {
|
|
||||||
log.warn("[DashScope STT] PCM is silent (peak=0, bytes={}) — check mic permission / frontend recording",
|
|
||||||
pcm.length);
|
|
||||||
return SttResult.failure(
|
|
||||||
"音频为静音（PCM peak=0）— 检查麦克风权限或前端录制实现");
|
|
||||||
}
|
|
||||||
log.debug("[DashScope STT] PCM stats — bytes={} samples={} peak={} rms={} sampleRate={}",
|
|
||||||
pcm.length, pcm.length / 2, peakRms[0], peakRms[1], sampleRate);
|
|
||||||
|
|
||||||
DashScopeSession session = new DashScopeSession(taskId, objectMapper);
|
|
||||||
WebSocket ws;
|
|
||||||
try {
|
|
||||||
ws = httpClient.newWebSocketBuilder()
|
|
||||||
.header("Authorization", "bearer " + apiKey)
|
|
||||||
.connectTimeout(Duration.ofMillis(TASK_STARTED_TIMEOUT_MS))
|
|
||||||
.buildAsync(WS_URL, session)
|
|
||||||
.get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS);
|
|
||||||
} catch (TimeoutException e) {
|
|
||||||
return SttResult.failure("DashScope WS 握手超时");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
try {
|
// Silence gate — only for WAV, where we can read PCM directly.
|
||||||
// 1. run-task. Envelope dumped at DEBUG only — the JSON is
|
// "Mic captured nothing" is by far the most common voice-input
|
||||||
// identical across calls modulo task_id + language hint, so
|
// failure; catching it here yields a precise error instead of an
|
||||||
// logging it on every transcribe just clutters logs.
|
// empty transcript from the model. Non-WAV inputs (IM voice
|
||||||
String runTask = buildRunTask(taskId, model, sampleRate, request.getLanguage());
|
// notes) skip the gate and go straight to the API.
|
||||||
log.debug("[DashScope STT] run-task envelope: {}", runTask);
|
if (WavPcmExtractor.isCanonicalWav(audio)) {
|
||||||
ws.sendText(runTask, true).get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS);
|
byte[] pcm = WavPcmExtractor.extract(audio);
|
||||||
|
int[] peakRms = computePcmPeakRms(pcm);
|
||||||
// 2. wait for task-started ack
|
if (peakRms[0] == 0) {
|
||||||
if (!session.awaitTaskStarted(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
|
log.warn("[DashScope STT] PCM is silent (peak=0, bytes={}) — check mic permission / frontend recording",
|
||||||
return SttResult.failure("DashScope task-started 超时");
|
pcm.length);
|
||||||
}
|
|
||||||
if (session.failed()) {
|
|
||||||
return SttResult.failure("DashScope: " + session.errorMessage());
|
|
||||||
}
|
|
||||||
|
|
||||||
// 3. stream PCM chunks at real-time pace. Paraformer-Realtime
|
|
||||||
// is built for live mic input and silently drops audio when it
|
|
||||||
// arrives faster than wall-clock — symptom is 0 chars
|
|
||||||
// transcribed even though the protocol completes successfully
|
|
||||||
// (no task-failed). Sleep 100ms between 100ms chunks so total
|
|
||||||
// send time ≈ audio duration, matching what DashScope's own
|
|
||||||
// SDK examples do (time.sleep(0.1) per chunk).
|
|
||||||
int chunksSent = 0;
|
|
||||||
long sendStart = System.currentTimeMillis();
|
|
||||||
for (int offset = 0; offset < pcm.length; offset += CHUNK_BYTES) {
|
|
||||||
int len = Math.min(CHUNK_BYTES, pcm.length - offset);
|
|
||||||
ByteBuffer chunk = ByteBuffer.wrap(pcm, offset, len);
|
|
||||||
ws.sendBinary(chunk, true).get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS);
|
|
||||||
chunksSent++;
|
|
||||||
Thread.sleep(CHUNK_PACING_MS);
|
|
||||||
// Cheap fail-fast: if the server already said we're done /
|
|
||||||
// failed mid-stream, stop sending so we don't waste seconds
|
|
||||||
// sleeping on a dead connection.
|
|
||||||
if (session.failed() || session.taskFinishedRaised()) break;
|
|
||||||
}
|
|
||||||
long sendDuration = System.currentTimeMillis() - sendStart;
|
|
||||||
log.debug("[DashScope STT] streamed {} chunks ({} bytes) in {} ms",
|
|
||||||
chunksSent, pcm.length, sendDuration);
|
|
||||||
|
|
||||||
// 4. finish-task
|
|
||||||
ws.sendText(buildFinishTask(taskId), true).get(TASK_STARTED_TIMEOUT_MS, TimeUnit.MILLISECONDS);
|
|
||||||
|
|
||||||
// 5. wait for task-finished
|
|
||||||
if (!session.awaitTaskFinished(OVERALL_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
|
|
||||||
return SttResult.failure("DashScope task-finished 超时");
|
|
||||||
}
|
|
||||||
if (session.failed()) {
|
|
||||||
return SttResult.failure("DashScope: " + session.errorMessage());
|
|
||||||
}
|
|
||||||
|
|
||||||
String text = session.aggregatedText();
|
|
||||||
log.info("[DashScope STT] Transcribed {} chars from {} result-events "
|
|
||||||
+ "(model={}, sampleRate={}, pcmBytes={})",
|
|
||||||
text.length(), session.resultEventCount(), model, sampleRate, pcm.length);
|
|
||||||
if (text.isEmpty() && session.resultEventCount() == 0) {
|
|
||||||
// Distinct failure mode: protocol completed cleanly but
|
|
||||||
// server never sent a single result-generated event.
|
|
||||||
// Almost always means the audio was discarded for
|
|
||||||
// pacing/format reasons. Surface as a typed failure so
|
|
||||||
// the fallback chain (Whisper) can still try.
|
|
||||||
return SttResult.failure(
|
return SttResult.failure(
|
||||||
"DashScope 收到 0 个识别事件——可能是音频格式或节奏问题");
|
"音频为静音（PCM peak=0）— 检查麦克风权限或前端录制实现");
|
||||||
}
|
|
||||||
return SttResult.success(text);
|
|
||||||
} finally {
|
|
||||||
// Best-effort close. abort() is fire-and-forget; we don't need to wait.
|
|
||||||
try {
|
|
||||||
ws.sendClose(WebSocket.NORMAL_CLOSURE, "done");
|
|
||||||
} catch (Exception ignored) {
|
|
||||||
ws.abort();
|
|
||||||
}
|
}
|
||||||
|
log.debug("[DashScope STT] PCM stats — bytes={} peak={} rms={}",
|
||||||
|
pcm.length, peakRms[0], peakRms[1]);
|
||||||
}
|
}
|
||||||
} catch (TimeoutException e) {
|
|
||||||
log.warn("[DashScope STT] timeout: {}", e.getMessage());
|
String model = (request.getModel() != null && !request.getModel().isBlank())
|
||||||
return SttResult.failure("DashScope STT 超时: " + e.getMessage());
|
? request.getModel() : DEFAULT_MODEL;
|
||||||
} catch (ExecutionException e) {
|
String format = resolveFormat(request.getFileName(), request.getContentType());
|
||||||
Throwable cause = e.getCause() != null ? e.getCause() : e;
|
String body = buildRequestBody(model, audio, format, request.getLanguage());
|
||||||
log.error("[DashScope STT] WS error: {}", cause.getMessage(), cause);
|
|
||||||
return SttResult.failure("DashScope STT WS 错误: " + cause.getMessage());
|
HttpResponse response = HttpRequest.post(ASR_ENDPOINT)
|
||||||
} catch (InterruptedException e) {
|
.header("Authorization", "Bearer " + apiKey.trim())
|
||||||
Thread.currentThread().interrupt();
|
.header("Content-Type", "application/json")
|
||||||
return SttResult.failure("DashScope STT 被中断");
|
.body(body)
|
||||||
|
.timeout(HTTP_TIMEOUT_MS)
|
||||||
|
.execute();
|
||||||
|
|
||||||
|
if (response.getStatus() != 200) {
|
||||||
|
String error = parseErrorMessage(response.body());
|
||||||
|
log.warn("[DashScope STT] HTTP {} — {}", response.getStatus(), error);
|
||||||
|
return SttResult.failure("DashScope STT 失败: HTTP " + response.getStatus()
|
||||||
|
+ (error.isEmpty() ? "" : " — " + error));
|
||||||
|
}
|
||||||
|
|
||||||
|
String text = parseTranscript(response.body());
|
||||||
|
log.info("[DashScope STT] Transcribed {} chars (model={}, format={}, audioBytes={})",
|
||||||
|
text.length(), model, format, audio.length);
|
||||||
|
return SttResult.success(text);
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
log.error("[DashScope STT] Error: {}", e.getMessage(), e);
|
log.error("[DashScope STT] Error: {}", e.getMessage(), e);
|
||||||
return SttResult.failure("DashScope STT 异常: " + e.getMessage());
|
return SttResult.failure("DashScope STT 异常: " + e.getMessage());
|
||||||
@ -264,35 +173,93 @@ public class DashScopeSttProvider implements SttProvider {
|
|||||||
/* Wire-format helpers (package-private for unit testing). */
|
/* Wire-format helpers (package-private for unit testing). */
|
||||||
/* ====================================================================== */
|
/* ====================================================================== */
|
||||||
|
|
||||||
String buildRunTask(String taskId, String model, int sampleRate, String language) throws Exception {
|
/**
|
||||||
Map<String, Object> parameters = new LinkedHashMap<>();
|
* Build the recognition request. The audio rides in a
|
||||||
parameters.put("format", "pcm");
|
* {@code data:;base64,} URI — the separate {@code format} field tells the
|
||||||
parameters.put("sample_rate", sampleRate);
|
* service how to decode it, so the URI needs no media type.
|
||||||
// Language hint when supplied — paraformer-realtime-v2 supports
|
*/
|
||||||
// "zh", "en", "ja", "ko" via language_hints. Skip when null/blank
|
String buildRequestBody(String model, byte[] audio, String format, String language) throws Exception {
|
||||||
// to let the model auto-detect.
|
Map<String, Object> inputAudio = new LinkedHashMap<>();
|
||||||
if (language != null && !language.isBlank()) {
|
inputAudio.put("data", "data:;base64," + Base64.getEncoder().encodeToString(audio));
|
||||||
// Strip locale suffix (zh-CN → zh).
|
inputAudio.put("format", format);
|
||||||
String hint = language.toLowerCase();
|
|
||||||
int dash = hint.indexOf('-');
|
|
||||||
if (dash > 0) hint = hint.substring(0, dash);
|
|
||||||
parameters.put("language_hints", new String[]{hint});
|
|
||||||
}
|
|
||||||
|
|
||||||
Map<String, Object> payload = Map.of(
|
Map<String, Object> payload = new LinkedHashMap<>();
|
||||||
"task_group", "audio",
|
payload.put("model", model);
|
||||||
"task", "asr",
|
payload.put("messages", List.of(Map.of(
|
||||||
"function", "recognition",
|
"role", "user",
|
||||||
"model", model,
|
"content", List.of(Map.of(
|
||||||
"parameters", parameters,
|
"type", "input_audio",
|
||||||
"input", Map.of());
|
"input_audio", inputAudio)))));
|
||||||
Map<String, Object> message = Map.of(
|
payload.put("stream", false);
|
||||||
"header", Map.of(
|
|
||||||
"action", "run-task",
|
// Language hint when supplied ("zh", "en", "ja", ...). Strip the
|
||||||
"task_id", taskId,
|
// locale suffix (zh-CN → zh); omit entirely to let the model
|
||||||
"streaming", "duplex"),
|
// auto-detect among its supported languages.
|
||||||
"payload", payload);
|
String hint = stripLocale(language);
|
||||||
return objectMapper.writeValueAsString(message);
|
if (hint != null) {
|
||||||
|
payload.put("asr_options", Map.of("language", hint));
|
||||||
|
}
|
||||||
|
return objectMapper.writeValueAsString(payload);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** {@code zh-CN → zh}; null/blank → null (auto-detect). */
|
||||||
|
static String stripLocale(String language) {
|
||||||
|
if (language == null || language.isBlank()) return null;
|
||||||
|
String hint = language.toLowerCase();
|
||||||
|
int dash = hint.indexOf('-');
|
||||||
|
return dash > 0 ? hint.substring(0, dash) : hint;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Derive the {@code input_audio.format} value ("wav", "mp3", ...) from
|
||||||
|
* the upload's filename/content-type. Falls back to "wav", matching the
|
||||||
|
* web recorder's output.
|
||||||
|
*/
|
||||||
|
static String resolveFormat(String fileName, String contentType) {
|
||||||
|
String resolved = AudioMimeTypes.resolveFileName(fileName, contentType);
|
||||||
|
int dot = resolved.lastIndexOf('.');
|
||||||
|
return dot >= 0 ? resolved.substring(dot + 1) : "wav";
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Extract the transcript from a chat-completion response. Content is
|
||||||
|
* normally a plain string; tolerate the content-part array form
|
||||||
|
* ({@code [{"text": "..."}]}) that multimodal-capable endpoints may emit.
|
||||||
|
*/
|
||||||
|
String parseTranscript(String json) throws Exception {
|
||||||
|
JsonNode content = objectMapper.readTree(json)
|
||||||
|
.path("choices").path(0).path("message").path("content");
|
||||||
|
if (content.isTextual()) {
|
||||||
|
return content.asText();
|
||||||
|
}
|
||||||
|
if (content.isArray()) {
|
||||||
|
StringBuilder sb = new StringBuilder();
|
||||||
|
for (JsonNode part : content) {
|
||||||
|
sb.append(part.path("text").asText(""));
|
||||||
|
}
|
||||||
|
return sb.toString();
|
||||||
|
}
|
||||||
|
return "";
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Pull a human-readable message out of an error body. DashScope's
|
||||||
|
* compatible mode wraps errors as {@code {"error":{"code","message"}}};
|
||||||
|
* the native surface uses top-level {@code code}/{@code message}.
|
||||||
|
* Returns "" when the body isn't parseable JSON.
|
||||||
|
*/
|
||||||
|
String parseErrorMessage(String body) {
|
||||||
|
if (body == null || body.isBlank()) return "";
|
||||||
|
try {
|
||||||
|
JsonNode root = objectMapper.readTree(body);
|
||||||
|
JsonNode error = root.has("error") ? root.path("error") : root;
|
||||||
|
String code = error.path("code").asText("");
|
||||||
|
String message = error.path("message").asText("");
|
||||||
|
if (code.isEmpty()) return message;
|
||||||
|
return message.isEmpty() ? code : code + " — " + message;
|
||||||
|
} catch (Exception e) {
|
||||||
|
return "";
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -327,168 +294,4 @@ public class DashScopeSttProvider implements SttProvider {
|
|||||||
int rms = (int) Math.sqrt((double) sumSq / sampleCount);
|
int rms = (int) Math.sqrt((double) sumSq / sampleCount);
|
||||||
return new int[]{peak, rms};
|
return new int[]{peak, rms};
|
||||||
}
|
}
|
||||||
|
|
||||||
String buildFinishTask(String taskId) throws Exception {
|
|
||||||
Map<String, Object> message = Map.of(
|
|
||||||
"header", Map.of(
|
|
||||||
"action", "finish-task",
|
|
||||||
"task_id", taskId,
|
|
||||||
"streaming", "duplex"),
|
|
||||||
"payload", Map.of("input", Map.of()));
|
|
||||||
return objectMapper.writeValueAsString(message);
|
|
||||||
}
|
|
||||||
|
|
||||||
/* ====================================================================== */
|
|
||||||
/* WebSocket.Listener: collects events and signals task-started/finished. */
|
|
||||||
/* ====================================================================== */
|
|
||||||
|
|
||||||
/**
|
|
||||||
* State machine for one DashScope ASR conversation. Package-private so
|
|
||||||
* unit tests can drive it with synthetic JSON without hitting the network.
|
|
||||||
*/
|
|
||||||
static class DashScopeSession implements WebSocket.Listener {
|
|
||||||
private final String taskId;
|
|
||||||
private final ObjectMapper mapper;
|
|
||||||
private final CountDownLatch taskStarted = new CountDownLatch(1);
|
|
||||||
private final CountDownLatch taskFinished = new CountDownLatch(1);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Sentence buffer keyed by begin_time. DashScope emits multiple
|
|
||||||
* {@code result-generated} events for the same sentence as it gets
|
|
||||||
* refined (interim → final); each new event for a given begin_time
|
|
||||||
* supersedes the previous text. LinkedHashMap preserves arrival
|
|
||||||
* order, which roughly matches speech order, for the final concat.
|
|
||||||
*/
|
|
||||||
private final Map<Long, String> sentencesByBeginTime = new LinkedHashMap<>();
|
|
||||||
|
|
||||||
/** Buffer for fragmented text frames (WS allows partial messages). */
|
|
||||||
private final StringBuilder textFrameBuf = new StringBuilder();
|
|
||||||
|
|
||||||
private final AtomicReference<String> errorMessage = new AtomicReference<>();
|
|
||||||
|
|
||||||
/** Counts result-generated events — distinguishes "server got our audio
|
|
||||||
* but recognised nothing" (>0 events, all empty text) from "server
|
|
||||||
* saw zero audio frames" (0 events). Helps diagnose pacing /
|
|
||||||
* format issues. */
|
|
||||||
private int resultEventCount;
|
|
||||||
|
|
||||||
DashScopeSession(String taskId, ObjectMapper mapper) {
|
|
||||||
this.taskId = taskId;
|
|
||||||
this.mapper = mapper;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
|
|
||||||
textFrameBuf.append(data);
|
|
||||||
if (last) {
|
|
||||||
handleMessage(textFrameBuf.toString());
|
|
||||||
textFrameBuf.setLength(0);
|
|
||||||
}
|
|
||||||
webSocket.request(1);
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void onError(WebSocket webSocket, Throwable error) {
|
|
||||||
errorMessage.compareAndSet(null, "WS error: " + error.getMessage());
|
|
||||||
taskStarted.countDown();
|
|
||||||
taskFinished.countDown();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
|
|
||||||
// If the server closes before task-finished, unblock waiters.
|
|
||||||
if (taskFinished.getCount() > 0) {
|
|
||||||
errorMessage.compareAndSet(null,
|
|
||||||
"WS closed before task-finished (status=" + statusCode + ", reason=" + reason + ")");
|
|
||||||
}
|
|
||||||
taskStarted.countDown();
|
|
||||||
taskFinished.countDown();
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
/** Package-private hook so unit tests can drive {@link DashScopeSession} without a real WebSocket. */
|
|
||||||
void handleMessage(String json) {
|
|
||||||
// Always trace the raw frame at DEBUG — this is invaluable when
|
|
||||||
// the protocol completes "successfully" but produces no
|
|
||||||
// transcripts. Without seeing every frame it's impossible to
|
|
||||||
// tell whether DashScope sent us a status-update / warning we
|
|
||||||
// ignored, or just stayed silent between task-started and
|
|
||||||
// task-finished.
|
|
||||||
log.debug("[DashScope STT] frame: {}", json);
|
|
||||||
try {
|
|
||||||
JsonNode node = mapper.readTree(json);
|
|
||||||
String event = node.path("header").path("event").asText();
|
|
||||||
switch (event) {
|
|
||||||
case "task-started" -> taskStarted.countDown();
|
|
||||||
case "result-generated" -> {
|
|
||||||
resultEventCount++;
|
|
||||||
JsonNode sentence = node.path("payload").path("output").path("sentence");
|
|
||||||
if (sentence.isObject()) {
|
|
||||||
long beginTime = sentence.path("begin_time").asLong(0L);
|
|
||||||
String text = sentence.path("text").asText("");
|
|
||||||
// Always overwrite — later events for the same begin_time
|
|
||||||
// carry the more-final transcript.
|
|
||||||
sentencesByBeginTime.put(beginTime, text);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
case "task-finished" -> taskFinished.countDown();
|
|
||||||
case "task-failed" -> {
|
|
||||||
String msg = node.path("header").path("error_message").asText("unknown");
|
|
||||||
String code = node.path("header").path("error_code").asText("");
|
|
||||||
errorMessage.compareAndSet(null,
|
|
||||||
code.isEmpty() ? msg : (code + " — " + msg));
|
|
||||||
// Wake both latches so the caller can return the typed
|
|
||||||
// failure instead of timing out for the full budget.
|
|
||||||
taskStarted.countDown();
|
|
||||||
taskFinished.countDown();
|
|
||||||
}
|
|
||||||
// Anything else (status updates, model warnings, beta
|
|
||||||
// events) gets surfaced at INFO so it shows up without
|
|
||||||
// turning DEBUG on. If DashScope rolls out a new event
|
|
||||||
// type we should know about, this catches it.
|
|
||||||
default -> log.info("[DashScope STT] unhandled event '{}' frame={}", event, json);
|
|
||||||
}
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.warn("[DashScope STT] failed to parse WS message: {}", e.getMessage());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
boolean awaitTaskStarted(long timeout, TimeUnit unit) throws InterruptedException {
|
|
||||||
return taskStarted.await(timeout, unit);
|
|
||||||
}
|
|
||||||
|
|
||||||
boolean awaitTaskFinished(long timeout, TimeUnit unit) throws InterruptedException {
|
|
||||||
return taskFinished.await(timeout, unit);
|
|
||||||
}
|
|
||||||
|
|
||||||
boolean failed() {
|
|
||||||
return errorMessage.get() != null;
|
|
||||||
}
|
|
||||||
|
|
||||||
String errorMessage() {
|
|
||||||
return errorMessage.get();
|
|
||||||
}
|
|
||||||
|
|
||||||
/** True once task-finished has been observed — used by the sender
|
|
||||||
* loop to bail out early instead of pacing through dead-WS sleeps. */
|
|
||||||
boolean taskFinishedRaised() {
|
|
||||||
return taskFinished.getCount() == 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
int resultEventCount() {
|
|
||||||
return resultEventCount;
|
|
||||||
}
|
|
||||||
|
|
||||||
String aggregatedText() {
|
|
||||||
// Concat in begin_time order. Different sentences typically don't
|
|
||||||
// need separator characters because Chinese text streams are
|
|
||||||
// already glued; for safety against missed punctuation we leave
|
|
||||||
// a soft join ("") rather than space — Whisper-style space
|
|
||||||
// joining produces odd-looking Chinese transcripts.
|
|
||||||
StringBuilder sb = new StringBuilder();
|
|
||||||
sentencesByBeginTime.values().forEach(sb::append);
|
|
||||||
return sb.toString();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -8,17 +8,18 @@ import java.nio.ByteOrder;
|
|||||||
|
|
||||||
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
|
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
|
||||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Pinned behaviour for the WAV → raw-PCM helper.
|
* Pinned behaviour for the WAV → raw-PCM helper.
|
||||||
*
|
*
|
||||||
* <p>Why this matters: DashScope's realtime ASR rejects bare WAV with
|
* <p>Why this matters: the peak/RMS silence pre-check reads raw samples off
|
||||||
* "format mismatch" because the first 44 bytes look like garbage when
|
* the frontend's WAV blob before any recognition call is made. A wrong
|
||||||
* interpreted as PCM. {@link WavPcmExtractor} is the chokepoint that
|
* header offset would feed header bytes into the PCM math and misreport
|
||||||
* converts the frontend's WAV blob to the bytes DashScope actually wants.
|
* silence vs signal; a wrong sample-rate read would break any consumer
|
||||||
* Wrong header offset → silent garbage transcripts; wrong sample-rate read
|
* that needs the true capture rate.
|
||||||
* → audibly distorted.
|
|
||||||
*/
|
*/
|
||||||
class WavPcmExtractorTest {
|
class WavPcmExtractorTest {
|
||||||
|
|
||||||
@ -49,6 +50,15 @@ class WavPcmExtractorTest {
|
|||||||
assertThrows(IllegalArgumentException.class, () -> WavPcmExtractor.extract(null));
|
assertThrows(IllegalArgumentException.class, () -> WavPcmExtractor.extract(null));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("isCanonicalWav: true for RIFF/WAVE, false for junk / short / null")
|
||||||
|
void isCanonicalWav_gates() {
|
||||||
|
assertTrue(WavPcmExtractor.isCanonicalWav(buildWav(16_000, 16, new byte[8])));
|
||||||
|
assertFalse(WavPcmExtractor.isCanonicalWav(new byte[64])); // no magic
|
||||||
|
assertFalse(WavPcmExtractor.isCanonicalWav(new byte[10])); // too short
|
||||||
|
assertFalse(WavPcmExtractor.isCanonicalWav(null));
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("sampleRate: reads 16 kHz from the canonical header offset")
|
@DisplayName("sampleRate: reads 16 kHz from the canonical header offset")
|
||||||
void sampleRate_reads16kHz() {
|
void sampleRate_reads16kHz() {
|
||||||
|
|||||||
@ -5,223 +5,151 @@ import com.fasterxml.jackson.databind.ObjectMapper;
|
|||||||
import org.junit.jupiter.api.BeforeEach;
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
import org.junit.jupiter.api.DisplayName;
|
import org.junit.jupiter.api.DisplayName;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import vip.mate.stt.provider.DashScopeSttProvider.DashScopeSession;
|
|
||||||
|
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.Base64;
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
|
|
||||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Unit tests for the message-handling state machine of
|
* Unit tests for {@link DashScopeSttProvider}'s wire-format helpers. The
|
||||||
* {@link DashScopeSttProvider}. The end-to-end WebSocket flow can't be
|
* HTTP round trip itself isn't exercised (no mock server); request-body
|
||||||
* exercised without a mock WS server, but the JSON parsing + transcript
|
* construction, response/transcript parsing, and error extraction are the
|
||||||
* aggregation + latch transitions are fully testable in isolation by
|
* parts with encoding rules worth pinning:
|
||||||
* driving {@link DashScopeSession#handleMessage(String)} directly.
|
|
||||||
*
|
*
|
||||||
* <p>What these tests guard against:
|
|
||||||
* <ul>
|
* <ul>
|
||||||
* <li>"Two events for the same begin_time" — the second event must
|
* <li>The audio must ride as a {@code data:;base64,} URI plus an explicit
|
||||||
* <b>overwrite</b> the first (interim → final), not append.
|
* {@code format} field — dropping either breaks decoding server-side.</li>
|
||||||
* Otherwise you get duplicated text in the final transcript.</li>
|
* <li>{@code asr_options} must be omitted entirely when no language hint
|
||||||
* <li>Sentence ordering — multi-sentence speech must come out in
|
* is supplied, so the service auto-detects.</li>
|
||||||
* arrival order regardless of begin_time int values.</li>
|
* <li>Transcript extraction must tolerate both plain-string and
|
||||||
* <li>task-failed must surface the error message on both latches so
|
* content-part-array response shapes.</li>
|
||||||
* the caller doesn't time out for the full 60s budget.</li>
|
|
||||||
* </ul>
|
* </ul>
|
||||||
*/
|
*/
|
||||||
class DashScopeSttProviderTest {
|
class DashScopeSttProviderTest {
|
||||||
|
|
||||||
private DashScopeSession session;
|
|
||||||
private DashScopeSttProvider provider;
|
private DashScopeSttProvider provider;
|
||||||
private ObjectMapper mapper;
|
private ObjectMapper mapper;
|
||||||
|
|
||||||
@BeforeEach
|
@BeforeEach
|
||||||
void setUp() {
|
void setUp() {
|
||||||
mapper = new ObjectMapper();
|
mapper = new ObjectMapper();
|
||||||
session = new DashScopeSession("test-task-id", mapper);
|
|
||||||
provider = new DashScopeSttProvider(null, mapper);
|
provider = new DashScopeSttProvider(null, mapper);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("task-started event releases the start latch")
|
@DisplayName("buildRequestBody serialises model, base64 audio, format and stream=false")
|
||||||
void taskStarted_releasesLatch() throws Exception {
|
void buildRequestBody_coreShape() throws Exception {
|
||||||
session.handleMessage("""
|
byte[] audio = "fake-wav-bytes".getBytes(StandardCharsets.UTF_8);
|
||||||
{"header":{"task_id":"test-task-id","event":"task-started"},"payload":{}}
|
String json = provider.buildRequestBody("qwen3-asr-flash", audio, "wav", null);
|
||||||
""");
|
|
||||||
assertTrue(session.awaitTaskStarted(100, TimeUnit.MILLISECONDS));
|
|
||||||
assertFalse(session.failed());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("result-generated builds transcript text")
|
|
||||||
void resultGenerated_appendsToTranscript() {
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"task_id":"test-task-id","event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":0,"end_time":1500,"text":"你好"}}}}
|
|
||||||
""");
|
|
||||||
assertEquals("你好", session.aggregatedText());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("interim updates for the same begin_time overwrite (not append)")
|
|
||||||
void resultGenerated_overwritesSameBeginTime() {
|
|
||||||
// Real DashScope behaviour: each sentence starts as a partial
|
|
||||||
// transcript and gets refined on subsequent events. Both events
|
|
||||||
// share the same begin_time. If we appended instead of overwriting
|
|
||||||
// we'd produce "你你好" instead of "你好".
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":0,"end_time":500,"text":"你"}}}}
|
|
||||||
""");
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":0,"end_time":1500,"text":"你好"}}}}
|
|
||||||
""");
|
|
||||||
assertEquals("你好", session.aggregatedText());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("multiple sentences concatenate in arrival order")
|
|
||||||
void resultGenerated_concatenatesSentencesInOrder() {
|
|
||||||
// Different begin_time → different sentences. Final transcript is
|
|
||||||
// the concat of all sentences in arrival order (LinkedHashMap).
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":0,"end_time":1500,"text":"你好"}}}}
|
|
||||||
""");
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":1500,"end_time":3000,"text":"世界"}}}}
|
|
||||||
""");
|
|
||||||
assertEquals("你好世界", session.aggregatedText());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("task-finished releases the finish latch")
|
|
||||||
void taskFinished_releasesLatch() throws Exception {
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"task-finished"},"payload":{}}
|
|
||||||
""");
|
|
||||||
assertTrue(session.awaitTaskFinished(100, TimeUnit.MILLISECONDS));
|
|
||||||
assertFalse(session.failed());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("task-failed surfaces error message and unblocks both latches")
|
|
||||||
void taskFailed_surfacesErrorAndUnblocks() throws Exception {
|
|
||||||
// Critical for fail-fast behaviour: without this the caller would
|
|
||||||
// time out after the full 60s OVERALL_TIMEOUT_MS instead of seeing
|
|
||||||
// the typed error within milliseconds.
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"task-failed",
|
|
||||||
"error_code":"InvalidParameter.SampleRate",
|
|
||||||
"error_message":"sample rate not supported"},
|
|
||||||
"payload":{}}
|
|
||||||
""");
|
|
||||||
assertTrue(session.awaitTaskStarted(100, TimeUnit.MILLISECONDS));
|
|
||||||
assertTrue(session.awaitTaskFinished(100, TimeUnit.MILLISECONDS));
|
|
||||||
assertTrue(session.failed());
|
|
||||||
assertTrue(session.errorMessage().contains("InvalidParameter.SampleRate"));
|
|
||||||
assertTrue(session.errorMessage().contains("sample rate not supported"));
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("resultEventCount tracks every result-generated event (regardless of text)")
|
|
||||||
void resultEventCount_isIncrementedPerEvent() {
|
|
||||||
// Distinguishing "server got our audio but didn't recognise anything"
|
|
||||||
// (>0 events with empty text) from "server saw 0 audio frames"
|
|
||||||
// (0 events) is the diagnostic that fingered the chunk-pacing bug.
|
|
||||||
// Pin the counter behaviour so it doesn't regress.
|
|
||||||
assertEquals(0, session.resultEventCount());
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":0,"text":"hi"}}}}
|
|
||||||
""");
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"result-generated"},
|
|
||||||
"payload":{"output":{"sentence":{"begin_time":1000,"text":""}}}}
|
|
||||||
""");
|
|
||||||
assertEquals(2, session.resultEventCount());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("taskFinishedRaised flips once task-finished arrives — sender uses it to bail out early")
|
|
||||||
void taskFinishedRaised_signalsSender() {
|
|
||||||
// The sender loop polls this between paced chunks so a server that
|
|
||||||
// closes the stream early doesn't make us sleep through the rest of
|
|
||||||
// the audio for nothing.
|
|
||||||
assertFalse(session.taskFinishedRaised());
|
|
||||||
session.handleMessage("""
|
|
||||||
{"header":{"event":"task-finished"},"payload":{}}
|
|
||||||
""");
|
|
||||||
assertTrue(session.taskFinishedRaised());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("malformed JSON doesn't crash the session")
|
|
||||||
void malformedJson_isLoggedNotThrown() {
|
|
||||||
// The session is fed straight from WS frames — corrupt input must
|
|
||||||
// not bubble up into the WebSocket.Listener and tear down the
|
|
||||||
// connection.
|
|
||||||
session.handleMessage("not valid json");
|
|
||||||
session.handleMessage("{\"missing_header\":true}");
|
|
||||||
// No event released either latch; session is still waiting.
|
|
||||||
assertFalse(session.failed());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
@DisplayName("buildRunTask serialises the documented run-task envelope")
|
|
||||||
void buildRunTask_envelopeShape() throws Exception {
|
|
||||||
// The wire format is documented by Aliyun — pin it so future
|
|
||||||
// refactors don't accidentally drop a required field.
|
|
||||||
String json = provider.buildRunTask(
|
|
||||||
"abcd1234efgh5678", "paraformer-realtime-v2", 16_000, "zh-CN");
|
|
||||||
JsonNode node = mapper.readTree(json);
|
JsonNode node = mapper.readTree(json);
|
||||||
assertEquals("run-task", node.path("header").path("action").asText());
|
|
||||||
assertEquals("abcd1234efgh5678", node.path("header").path("task_id").asText());
|
assertEquals("qwen3-asr-flash", node.path("model").asText());
|
||||||
assertEquals("duplex", node.path("header").path("streaming").asText());
|
assertEquals(false, node.path("stream").asBoolean(true));
|
||||||
assertEquals("audio", node.path("payload").path("task_group").asText());
|
|
||||||
assertEquals("asr", node.path("payload").path("task").asText());
|
JsonNode content = node.path("messages").path(0).path("content").path(0);
|
||||||
assertEquals("recognition", node.path("payload").path("function").asText());
|
assertEquals("user", node.path("messages").path(0).path("role").asText());
|
||||||
assertEquals("paraformer-realtime-v2", node.path("payload").path("model").asText());
|
assertEquals("input_audio", content.path("type").asText());
|
||||||
assertEquals("pcm", node.path("payload").path("parameters").path("format").asText());
|
assertEquals("wav", content.path("input_audio").path("format").asText());
|
||||||
assertEquals(16_000, node.path("payload").path("parameters").path("sample_rate").asInt());
|
|
||||||
// language_hints strips the locale: zh-CN → zh
|
String data = content.path("input_audio").path("data").asText();
|
||||||
assertEquals("zh", node.path("payload").path("parameters").path("language_hints").get(0).asText());
|
assertTrue(data.startsWith("data:;base64,"), "audio must be a base64 data URI");
|
||||||
|
assertEquals(Base64.getEncoder().encodeToString(audio),
|
||||||
|
data.substring("data:;base64,".length()));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("buildRunTask omits language_hints when language is null")
|
@DisplayName("buildRequestBody adds asr_options.language with locale stripped")
|
||||||
void buildRunTask_skipsLanguageHintsWhenNull() throws Exception {
|
void buildRequestBody_languageHint() throws Exception {
|
||||||
// Null language means "let DashScope auto-detect" — sending an
|
String json = provider.buildRequestBody("qwen3-asr-flash", new byte[]{1}, "wav", "zh-CN");
|
||||||
// empty array would flag as a parameter error on some accounts.
|
|
||||||
String json = provider.buildRunTask("task1", "paraformer-realtime-v2", 16_000, null);
|
|
||||||
JsonNode node = mapper.readTree(json);
|
JsonNode node = mapper.readTree(json);
|
||||||
assertTrue(node.path("payload").path("parameters").path("language_hints").isMissingNode(),
|
assertEquals("zh", node.path("asr_options").path("language").asText());
|
||||||
"language_hints should be omitted when language is null");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("buildFinishTask serialises the documented finish-task envelope")
|
@DisplayName("buildRequestBody omits asr_options when language is null — auto-detect")
|
||||||
void buildFinishTask_envelopeShape() throws Exception {
|
void buildRequestBody_omitsAsrOptionsWhenNoLanguage() throws Exception {
|
||||||
String json = provider.buildFinishTask("abcd1234");
|
// An empty or null language means "let the service detect the
|
||||||
JsonNode node = mapper.readTree(json);
|
// language"; sending asr_options with a null/blank language field
|
||||||
assertEquals("finish-task", node.path("header").path("action").asText());
|
// would be rejected as a parameter error.
|
||||||
assertEquals("abcd1234", node.path("header").path("task_id").asText());
|
String json = provider.buildRequestBody("qwen3-asr-flash", new byte[]{1}, "wav", null);
|
||||||
assertEquals("duplex", node.path("header").path("streaming").asText());
|
assertTrue(mapper.readTree(json).path("asr_options").isMissingNode());
|
||||||
// payload.input is required to be an empty object — DashScope
|
String jsonBlank = provider.buildRequestBody("qwen3-asr-flash", new byte[]{1}, "wav", " ");
|
||||||
// rejects requests where it's missing or null.
|
assertTrue(mapper.readTree(jsonBlank).path("asr_options").isMissingNode());
|
||||||
assertTrue(node.path("payload").path("input").isObject());
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("stripLocale: zh-CN → zh, en → en, null/blank → null")
|
||||||
|
void stripLocale_variants() {
|
||||||
|
assertEquals("zh", DashScopeSttProvider.stripLocale("zh-CN"));
|
||||||
|
assertEquals("zh", DashScopeSttProvider.stripLocale("ZH-Hant"));
|
||||||
|
assertEquals("en", DashScopeSttProvider.stripLocale("en"));
|
||||||
|
assertNull(DashScopeSttProvider.stripLocale(null));
|
||||||
|
assertNull(DashScopeSttProvider.stripLocale(" "));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("resolveFormat maps filename/content-type to the input_audio format value")
|
||||||
|
void resolveFormat_variants() {
|
||||||
|
assertEquals("wav", DashScopeSttProvider.resolveFormat("clip.wav", null));
|
||||||
|
assertEquals("mp3", DashScopeSttProvider.resolveFormat(null, "audio/mpeg"));
|
||||||
|
assertEquals("ogg", DashScopeSttProvider.resolveFormat("note.ogg", "audio/ogg"));
|
||||||
|
assertEquals("webm", DashScopeSttProvider.resolveFormat(null, "audio/webm; codecs=opus"));
|
||||||
|
// Unknown everything → wav (matches the web recorder's output).
|
||||||
|
assertEquals("wav", DashScopeSttProvider.resolveFormat(null, null));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("parseTranscript reads plain-string message content")
|
||||||
|
void parseTranscript_stringContent() throws Exception {
|
||||||
|
String response = """
|
||||||
|
{"choices":[{"message":{"role":"assistant","content":"你好世界"},
|
||||||
|
"finish_reason":"stop"}],"usage":{}}
|
||||||
|
""";
|
||||||
|
assertEquals("你好世界", provider.parseTranscript(response));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("parseTranscript concatenates content-part arrays")
|
||||||
|
void parseTranscript_arrayContent() throws Exception {
|
||||||
|
String response = """
|
||||||
|
{"choices":[{"message":{"content":[{"text":"你好"},{"text":"世界"}]}}]}
|
||||||
|
""";
|
||||||
|
assertEquals("你好世界", provider.parseTranscript(response));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("parseTranscript returns empty string on missing/odd shapes instead of throwing")
|
||||||
|
void parseTranscript_missingContent() throws Exception {
|
||||||
|
assertEquals("", provider.parseTranscript("{}"));
|
||||||
|
assertEquals("", provider.parseTranscript("{\"choices\":[]}"));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("parseErrorMessage handles compatible-mode and native error bodies")
|
||||||
|
void parseErrorMessage_variants() {
|
||||||
|
assertEquals("InvalidApiKey — Invalid API-key provided.",
|
||||||
|
provider.parseErrorMessage("""
|
||||||
|
{"error":{"code":"InvalidApiKey","message":"Invalid API-key provided."}}
|
||||||
|
"""));
|
||||||
|
assertEquals("Throttling — Requests throttled.",
|
||||||
|
provider.parseErrorMessage("""
|
||||||
|
{"code":"Throttling","message":"Requests throttled."}
|
||||||
|
"""));
|
||||||
|
assertEquals("just a message",
|
||||||
|
provider.parseErrorMessage("{\"message\":\"just a message\"}"));
|
||||||
|
assertEquals("", provider.parseErrorMessage("not json"));
|
||||||
|
assertEquals("", provider.parseErrorMessage(null));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("computePcmPeakRms returns 0,0 on silence; non-zero on synthetic tone")
|
@DisplayName("computePcmPeakRms returns 0,0 on silence; non-zero on synthetic tone")
|
||||||
void computePcmPeakRms_distinguishesSilenceFromSignal() {
|
void computePcmPeakRms_distinguishesSilenceFromSignal() {
|
||||||
// The diagnostic distinguishing "mic captured silence" (peak=0) from
|
// The diagnostic distinguishing "mic captured silence" (peak=0) from
|
||||||
// "DashScope rejected non-empty audio" (peak>0 but 0 events) is a
|
// "provider rejected non-empty audio" is a critical user-visible
|
||||||
// critical user-visible signal — pin its math.
|
// signal — pin its math.
|
||||||
byte[] silent = new byte[1000]; // all zeros
|
byte[] silent = new byte[1000]; // all zeros
|
||||||
int[] silentStats = DashScopeSttProvider.computePcmPeakRms(silent);
|
int[] silentStats = DashScopeSttProvider.computePcmPeakRms(silent);
|
||||||
assertEquals(0, silentStats[0]);
|
assertEquals(0, silentStats[0]);
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user