diff --git a/bootstrap/src/main/resources/application.yaml b/bootstrap/src/main/resources/application.yaml index 1816f5214..3c6338037 100644 --- a/bootstrap/src/main/resources/application.yaml +++ b/bootstrap/src/main/resources/application.yaml @@ -114,6 +114,19 @@ rag: lease-seconds: 30 poll-interval-ms: 200 + voice: + executors: + websocket-lifecycle: + core-pool-size: 2 + max-pool-size: 32 + keep-alive-seconds: 60 + thread-name-prefix: websocket_lifecycle_executor_ + playback: + core-pool-size: 2 + max-pool-size: 8 + keep-alive-seconds: 60 + thread-name-prefix: voice_playback_executor_ + memory: history-keep-turns: 8 summary-enabled: true @@ -199,9 +212,11 @@ ai: bailian: url: https://dashscope.aliyuncs.com api-key: ${BAILIAN_API_KEY:} + workspace: ${BAILIAN_WORKSPACE:} endpoints: chat: /compatible-mode/v1/chat/completions rerank: /api/v1/services/rerank/text-rerank/text-rerank + tts: /api-ws/v1/inference aihubmix: url: https://aihubmix.com api-key: ${AIHUBMIX_API_KEY:} @@ -297,6 +312,25 @@ ai: provider: bailian model: qwen-vl-max + tts: + default-model: cosyvoice-v3-flash + timeout-ms: 3000 + candidates: + - id: cosyvoice-v3-flash + provider: bailian + model: cosyvoice-v3-flash + priority: 1 + voice: longxiaochun_v3 + + websocket: + connect-timeout-ms: 10000 + task-start-timeout-ms: 10000 + task-packet-idle-timeout-ms: 10000 + max-total-per-model: 8 + max-idle-per-model: 8 + idle-timeout-ms: 25000 + eviction-interval-ms: 30000 + # MinerU SaaS API 配置（PDF / Word / PPT 走 MinerU 解析） mineru: api-url: https://mineru.net/api/v4 diff --git a/bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackEventHandlerTest.java b/bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackEventHandlerTest.java new file mode 100644 index 000000000..4ddc4f55b --- /dev/null +++ b/bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackEventHandlerTest.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.service.handler; + +import com.nageoffer.ai.ragent.framework.web.StreamTaskManager; +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.List; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +class VoicePlaybackEventHandlerTest { + + private static final String TASK_ID = "task-1"; + + @Test + void sendsAudioAndCompletesThroughSse() throws Exception { + SseEmitter emitter = mock(SseEmitter.class); + StreamTaskManager taskManager = mock(StreamTaskManager.class); + VoicePlaybackEventHandler handler = new VoicePlaybackEventHandler(emitter, TASK_ID, taskManager); + + handler.onAudio(new byte[]{1, 2}); + handler.onComplete(); + + verify(taskManager).register(any(String.class), any(Runnable.class)); + verify(emitter, times(3)).send(any(SseEmitter.SseEventBuilder.class)); + verify(taskManager).unregister(TASK_ID); + verify(emitter).complete(); + } + + @Test + void reportsOnlyFirstError() { + SseEmitter emitter = mock(SseEmitter.class); + StreamTaskManager taskManager = mock(StreamTaskManager.class); + VoicePlaybackEventHandler handler = new VoicePlaybackEventHandler(emitter, TASK_ID, taskManager); + RuntimeException failure = new RuntimeException("failed"); + + handler.onError(failure); + handler.onError(new RuntimeException("duplicate")); + + verify(taskManager).unregister(TASK_ID); + verify(emitter).completeWithError(failure); + } + + @Test + void cancelsTaskWhenEmitterCompletes() { + SseEmitter emitter = mock(SseEmitter.class); + StreamTaskManager taskManager = mock(StreamTaskManager.class); + new VoicePlaybackEventHandler(emitter, TASK_ID, taskManager); + ArgumentCaptor callbacks = ArgumentCaptor.forClass(Runnable.class); + verify(emitter, times(2)).onCompletion(callbacks.capture()); + + List completionCallbacks = callbacks.getAllValues(); + completionCallbacks.get(1).run(); + + verify(taskManager).cancel(TASK_ID); + verify(taskManager, never()).unregister(TASK_ID); + } + + @Test + void bindsProviderCancellationHandleToTask() { + SseEmitter emitter = mock(SseEmitter.class); + StreamTaskManager taskManager = mock(StreamTaskManager.class); + StreamCancellationHandle providerHandle = mock(StreamCancellationHandle.class); + VoicePlaybackEventHandler handler = new VoicePlaybackEventHandler(emitter, TASK_ID, taskManager); + ArgumentCaptor handle = ArgumentCaptor.forClass(Runnable.class); + + handler.onTaskStarted(providerHandle); + verify(taskManager).bindHandle(eq(TASK_ID), handle.capture()); + handle.getValue().run(); + + verify(providerHandle).cancel(); + } +} diff --git a/bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackTaskRunnerTest.java b/bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackTaskRunnerTest.java new file mode 100644 index 000000000..14ba9edbb --- /dev/null +++ b/bootstrap/src/test/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackTaskRunnerTest.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.service.handler; + +import com.nageoffer.ai.ragent.infra.voice.tts.TtsService; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class VoicePlaybackTaskRunnerTest { + + private static final String TEXT = "你好"; + + @Test + void startsTtsThroughPlaybackExecutor() { + TtsService ttsService = mock(TtsService.class); + VoicePlaybackEventHandler callback = mock(VoicePlaybackEventHandler.class); + VoicePlaybackTaskRunner runner = new VoicePlaybackTaskRunner(ttsService, Runnable::run); + + runner.run(TEXT, callback); + + verify(ttsService).synthesize(TEXT, callback, callback); + } + + @Test + void skipsTtsWhenTaskWasCancelled() { + TtsService ttsService = mock(TtsService.class); + VoicePlaybackEventHandler callback = mock(VoicePlaybackEventHandler.class); + when(callback.isCancelled()).thenReturn(true); + VoicePlaybackTaskRunner runner = new VoicePlaybackTaskRunner(ttsService, Runnable::run); + + runner.run(TEXT, callback); + + verify(ttsService, never()).synthesize(TEXT, callback, callback); + } + + @Test + void delegatesStartFailureToCallback() { + TtsService ttsService = mock(TtsService.class); + VoicePlaybackEventHandler callback = mock(VoicePlaybackEventHandler.class); + RuntimeException failure = new RuntimeException("failed"); + doThrow(failure).when(ttsService).synthesize(TEXT, callback, callback); + VoicePlaybackTaskRunner runner = new VoicePlaybackTaskRunner(ttsService, Runnable::run); + + runner.run(TEXT, callback); + + verify(callback).onStartFailure(failure); + } + + @Test + void delegatesExecutorRejectionToCallback() { + TtsService ttsService = mock(TtsService.class); + VoicePlaybackEventHandler callback = mock(VoicePlaybackEventHandler.class); + Executor rejectingExecutor = command -> { + throw new RejectedExecutionException("busy"); + }; + VoicePlaybackTaskRunner runner = new VoicePlaybackTaskRunner(ttsService, rejectingExecutor); + + runner.run(TEXT, callback); + + verify(callback).onRejected(any(RejectedExecutionException.class)); + } +} diff --git a/frontend/src/components/chat/FeedbackButtons.tsx b/frontend/src/components/chat/FeedbackButtons.tsx index 9fff23a29..99239db7c 100644 --- a/frontend/src/components/chat/FeedbackButtons.tsx +++ b/frontend/src/components/chat/FeedbackButtons.tsx @@ -4,6 +4,7 @@ import { toast } from "sonner"; import { Button } from "@/components/ui/button"; import { ThumbDownFilledIcon, ThumbUpFilledIcon } from "@/components/chat/ThumbIcons"; +import { VoicePlayButton } from "@/components/chat/VoicePlayButton"; import { DropdownMenu, DropdownMenuContent, @@ -19,6 +20,8 @@ interface FeedbackButtonsProps { messageId: string; feedback: FeedbackValue; content: string; + playing?: boolean; + onTogglePlay?: () => void; className?: string; alwaysVisible?: boolean; } @@ -31,6 +34,8 @@ export function FeedbackButtons({ messageId, feedback, content, + playing, + onTogglePlay, className, alwaysVisible }: FeedbackButtonsProps) { @@ -171,6 +176,7 @@ export function FeedbackButtons({ + {onTogglePlay ? : null} + ); +} diff --git a/frontend/src/hooks/useVoicePlayback.ts b/frontend/src/hooks/useVoicePlayback.ts new file mode 100644 index 000000000..bf5f640a9 --- /dev/null +++ b/frontend/src/hooks/useVoicePlayback.ts @@ -0,0 +1,237 @@ +import { useChatStore } from "@/stores/chatStore"; + +import { createStreamResponse } from "@/hooks/useStreamResponse"; +import { storage } from "@/utils/storage"; + +const API_BASE_URL = (import.meta.env.VITE_API_BASE_URL || "").replace(/\/$/, ""); +const PLAY_URL = `${API_BASE_URL}/rag/v3/voice/play`; +const STOP_URL = `${API_BASE_URL}/rag/v3/voice/stop`; + +interface AudioMetaPayload { + taskId?: string; +} + +// 播放器单例 全局只播一条 +let streamRef: ReturnType | null = null; +let requestStopRef: (() => void) | null = null; +let mediaSourceRef: MediaSource | null = null; +let sourceBufferRef: SourceBuffer | null = null; +let audioElRef: HTMLAudioElement | null = null; +let objectUrlRef: string | null = null; +let appendQueue: Uint8Array[] = []; +let appendPending = false; + +function stopInternal() { + const requestStop = requestStopRef; + streamRef = null; + requestStopRef = null; + requestStop?.(); + if (audioElRef) { + audioElRef.pause(); + audioElRef.removeAttribute("src"); + audioElRef = null; + } + if (objectUrlRef) { + URL.revokeObjectURL(objectUrlRef); + objectUrlRef = null; + } + if (mediaSourceRef && mediaSourceRef.readyState !== "closed") { + try { + mediaSourceRef.endOfStream(); + } catch { + // 流已关闭 + } + } + mediaSourceRef = null; + + sourceBufferRef = null; + appendQueue = []; + appendPending = false; +} + +/** + * MSE 流式播放 MP3 + */ +function playInternal(messageId: string) { + stopInternal(); + + if (!("MediaSource" in window)) { + setPlaying(null); + return; + } + if (!MediaSource.isTypeSupported("audio/mpeg")) { + setPlaying(null); + return; + } + + let playStarted = false; + let streamEnded = false; + let taskId: string | null = null; + let cancelRequested = false; + let stopRequested = false; + const mediaSource = new MediaSource(); + mediaSourceRef = mediaSource; + const audio = new Audio(); + audioElRef = audio; + + const requestStop = () => { + cancelRequested = true; + if (!taskId || stopRequested) return; + stopRequested = true; + const token = storage.getToken(); + fetch(`${STOP_URL}?taskId=${encodeURIComponent(taskId)}`, { + method: "POST", + headers: token ? { Authorization: token } : undefined + }).catch(() => null); + }; + + const endStreamIfReady = () => { + if ( + !streamEnded || + appendPending || + appendQueue.length > 0 || + !sourceBufferRef || + sourceBufferRef.updating || + mediaSource.readyState !== "open" + ) { + return; + } + try { + mediaSource.endOfStream(); + } catch { + // 流已关闭 + } + }; + + const flushAppendQueue = () => { + if (appendPending || !sourceBufferRef || appendQueue.length === 0) return; + const next = appendQueue.shift()!; + appendPending = true; + try { + sourceBufferRef.appendBuffer(next); + // 首帧入缓冲后开始播放 + if (!playStarted) { + playStarted = true; + audio.play().catch(() => { + if (audioElRef === audio) { + stopInternal(); + setPlaying(null); + } + }); + } + } catch { + appendPending = false; + } + }; + + mediaSource.addEventListener("sourceopen", () => { + try { + sourceBufferRef = mediaSource.addSourceBuffer("audio/mpeg"); + // MP3 没有可供 MSE 排序的时间戳 + sourceBufferRef.mode = "sequence"; + sourceBufferRef.addEventListener("updateend", () => { + appendPending = false; + flushAppendQueue(); + endStreamIfReady(); + }); + // 补排 sourceopen 前收到的音频帧 + flushAppendQueue(); + endStreamIfReady(); + } catch { + setPlaying(null); + } + }); + + audio.onerror = () => setPlaying(null); + audio.onended = () => setPlaying(null); + + const handlers = { + onEvent(event: string, payload: unknown) { + if (event === "audio-meta") { + const meta = payload as AudioMetaPayload; + taskId = meta?.taskId ?? null; + if (cancelRequested) { + requestStop(); + } + } + if (streamRef !== stream) return; + if (event === "audio") { + const frame = payload as { base64?: string }; + if (!frame?.base64) return; + const bytes = Uint8Array.from(atob(frame.base64), (c) => c.charCodeAt(0)); + appendQueue.push(bytes); + flushAppendQueue(); + } else if (event === "done") { + streamRef = null; + requestStopRef = null; + streamEnded = true; + endStreamIfReady(); + } + }, + onError() { + if (streamRef !== stream) return; + streamRef = null; + requestStopRef = null; + // 保留已缓冲音频 + streamEnded = true; + endStreamIfReady(); + setPlaying(null); + } + }; + + const token = storage.getToken(); + const stream = createStreamResponse( + { + url: `${PLAY_URL}?messageId=${encodeURIComponent(messageId)}`, + headers: token ? { Authorization: token } : undefined, + retryCount: 0 + }, + handlers + ); + streamRef = stream; + requestStopRef = requestStop; + + if (objectUrlRef) { + URL.revokeObjectURL(objectUrlRef); + } + objectUrlRef = URL.createObjectURL(mediaSource); + audio.src = objectUrlRef; + + stream.start().catch(() => { + if (streamRef === stream) { + stopInternal(); + setPlaying(null); + } + }); + setPlaying(messageId); +} + +function setPlaying(messageId: string | null) { + useChatStore.setState({ playingMessageId: messageId }); +} + +/** + * 停止当前语音播放 + */ +export function stopVoicePlayback() { + stopInternal(); + setPlaying(null); +} + +/** + * 消息语音播放 + */ +export function useVoicePlayback() { + const playingId = useChatStore((state) => state.playingMessageId); + + const togglePlay = (messageId: string) => { + if (playingId === messageId) { + stopInternal(); + setPlaying(null); + return; + } + playInternal(messageId); + }; + + return { playingId, togglePlay }; +} diff --git a/frontend/src/stores/chatStore.ts b/frontend/src/stores/chatStore.ts index bccfcd74f..c5fa43a80 100644 --- a/frontend/src/stores/chatStore.ts +++ b/frontend/src/stores/chatStore.ts @@ -22,6 +22,7 @@ import { } from "@/services/chatService"; import { buildQuery } from "@/utils/helpers"; import { createStreamResponse } from "@/hooks/useStreamResponse"; +import { stopVoicePlayback } from "@/hooks/useVoicePlayback"; import { storage } from "@/utils/storage"; interface ChatState { @@ -42,6 +43,8 @@ interface ChatState { openedSourceMessageId: string | null; // 展开推荐面板后需滚入视口的消息；每次请求都是新对象，供 MessageList 一次性响应 recommendReveal: { id: string } | null; + // 当前正在语音播放的消息 ID 全局同时只播一条 + playingMessageId: string | null; fetchSessions: () => Promise; createSession: () => Promise; deleteSession: (sessionId: string) => Promise; @@ -107,6 +110,7 @@ export const useChatStore = create((set, get) => ({ cancelRequested: false, openedSourceMessageId: null, recommendReveal: null, + playingMessageId: null, fetchSessions: async () => { set({ isLoading: true }); try { @@ -144,6 +148,8 @@ export const useChatStore = create((set, get) => ({ if (state.isStreaming) { get().cancelGeneration(); } + // 切换会话时停止语音播放 + stopVoicePlayback(); set({ currentSessionId: null, messages: [], @@ -163,6 +169,10 @@ export const useChatStore = create((set, get) => ({ deleteSession: async (sessionId) => { try { await deleteSessionRequest(sessionId); + // 删除当前会话时停止语音播放 + if (get().currentSessionId === sessionId) { + stopVoicePlayback(); + } set((state) => ({ sessions: state.sessions.filter((session) => session.id !== sessionId), messages: state.currentSessionId === sessionId ? [] : state.messages, @@ -196,6 +206,8 @@ export const useChatStore = create((set, get) => ({ if (get().isStreaming) { get().cancelGeneration(); } + // 切换会话时停止语音播放 + stopVoicePlayback(); set({ isLoading: true, currentSessionId: sessionId, diff --git a/infra-ai/pom.xml b/infra-ai/pom.xml index 4f25996f3..7e07a87dd 100644 --- a/infra-ai/pom.xml +++ b/infra-ai/pom.xml @@ -21,5 +21,10 @@ com.squareup.okhttp3 okhttp-jvm + + + org.apache.commons + commons-pool2 + diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/config/AIModelProperties.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/config/AIModelProperties.java index 23da76f33..c91c0ccc2 100644 --- a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/config/AIModelProperties.java +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/config/AIModelProperties.java @@ -61,6 +61,11 @@ public class AIModelProperties { */ private ModelGroup vlm = new ModelGroup(); + /** + * 语音合成模型组配置 + */ + private ModelGroup tts = new ModelGroup(); + /** * 模型选择策略配置 */ @@ -71,6 +76,11 @@ public class AIModelProperties { */ private Stream stream = new Stream(); + /** + * WebSocket 连接及任务生命周期配置 + */ + private WebSocketConfig websocket = new WebSocketConfig(); + /** * 模型组配置类 * 包含默认模型与候选模型列表 @@ -89,6 +99,11 @@ public static class ModelGroup { */ private List candidates = new ArrayList<>(); + /** + * 模型组调用超时预算 + */ + private Long timeoutMs; + /** * 默认档位名（仅 chat 使用） * 未显式指定 Tier 覆盖时的默认档位（兜底档） @@ -108,6 +123,33 @@ public static class ModelGroup { private Map tiers = new HashMap<>(); } + /** + * WebSocket 配置 + */ + @Data + public static class WebSocketConfig { + + private long connectTimeoutMs; + + private long taskStartTimeoutMs; + + private long taskPacketIdleTimeoutMs; + + private int maxTotalPerModel; + + private int maxIdlePerModel; + + /** + * 连接空闲驱逐超时 0 表示不驱逐 + */ + private long idleTimeoutMs; + + /** + * 空闲驱逐扫描间隔 + */ + private long evictionIntervalMs; + } + /** * 档位配置类 * 定义单个档位的有序候选与超时预算 @@ -173,6 +215,11 @@ public static class ModelCandidate { * 是否支持思考链功能 */ private Boolean supportsThinking = false; + + /** + * 语音类模型默认音色 + */ + private String voice; } /** @@ -192,6 +239,11 @@ public static class ProviderConfig { */ private String apiKey; + /** + * 供应商工作空间标识 + */ + private String workspace; + /** * 端点映射配置 * key: 端点类型，value: 端点路径 diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/enums/ModelCapability.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/enums/ModelCapability.java index 336384707..8b4f066aa 100644 --- a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/enums/ModelCapability.java +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/enums/ModelCapability.java @@ -44,7 +44,13 @@ public enum ModelCapability { * 重排序能力 * 对搜索结果进行重新排序，提高相关性 */ - RERANK("Rerank"); + RERANK("Rerank"), + + /** + * 文本转语音能力 + */ + TTS("TTS"); + /** * 能力的显示名称 diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelSelector.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelSelector.java index a4c11e0b2..0cd217f94 100644 --- a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelSelector.java +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelSelector.java @@ -38,7 +38,7 @@ * 负责根据配置和当前需求选择合适的模型候选列表 *

* chat 组走档位机制：任务 → 档位（tier）→ 档位内有序候选； - * embedding/rerank/vlm 组走 defaultModel + priority 的传统排序 + * embedding/rerank/vlm/tts 组走 defaultModel + priority 的传统排序 */ @Slf4j @Component @@ -96,6 +96,10 @@ public List selectVlmCandidates() { return selectCandidates(properties.getVlm()); } + public List selectTtsCandidates() { + return selectCandidates(properties.getTts()); + } + // ==================== chat：档位机制 ==================== private String resolveTierName(AIModelProperties.ModelGroup group, boolean thinking, Tier override) { @@ -183,7 +187,7 @@ private Map buildRegistry(List selectCandidates(AIModelProperties.ModelGroup group) { if (group == null || group.getCandidates() == null) { @@ -191,7 +195,7 @@ private List selectCandidates(AIModelProperties.ModelGroup group) { } List orderedCandidates = filterAndSortCandidates(group.getCandidates(), group.getDefaultModel()); - return buildAvailableTargets(orderedCandidates); + return buildAvailableTargets(orderedCandidates, group.getTimeoutMs()); } /** @@ -211,12 +215,13 @@ private List filterAndSortCandidates(List buildAvailableTargets(List candidates) { + private List buildAvailableTargets(List candidates, + Long timeoutMs) { Map providers = properties.getProviders(); - // embedding/rerank/vlm 无档位预算，超时走 HTTP 客户端默认 + // embedding/rerank/vlm 未配置 timeoutMs 时走 HTTP 客户端默认；TTS 的 timeoutMs 用于首包等待 return candidates.stream() - .map(candidate -> buildModelTarget(candidate, providers, null)) + .map(candidate -> buildModelTarget(candidate, providers, timeoutMs)) .filter(Objects::nonNull) .collect(Collectors.toList()); } diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelTarget.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelTarget.java index b69afb248..9a1e9f871 100644 --- a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelTarget.java +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/model/ModelTarget.java @@ -27,7 +27,7 @@ * @param id 模型唯一标识符 * @param candidate 模型候选配置，包含模型的具体参数和设置 * @param provider 提供商配置，包含模型提供商的相关信息 - * @param timeoutMs 本次调用的超时预算（毫秒），来自命中的档位配置；null 表示不额外限制，走 HTTP 客户端默认 + * @param timeoutMs 本次调用的超时预算 */ public record ModelTarget( String id, diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/AbstractWebSocketTtsClient.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/AbstractWebSocketTtsClient.java new file mode 100644 index 000000000..1179eae91 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/AbstractWebSocketTtsClient.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import com.nageoffer.ai.ragent.infra.voice.websocket.VoiceConnection; +import com.nageoffer.ai.ragent.infra.voice.websocket.VoiceStreamCallback; +import com.nageoffer.ai.ragent.infra.voice.websocket.WebSocketTaskExecutor; +import com.nageoffer.ai.ragent.infra.voice.websocket.WebSocketTaskSession; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * 基于 WebSocket 的 TTS 客户端模板 + */ +public abstract class AbstractWebSocketTtsClient> + implements TtsClient, AutoCloseable { + + /** + * continue-task 单次发送的文本长度上限 + */ + private static final int CHUNK_MAX_LEN = 80; + + private final WebSocketTaskExecutor taskExecutor; + + protected AbstractWebSocketTtsClient(Executor executor, AIModelProperties.WebSocketConfig poolConfig) { + this.taskExecutor = new WebSocketTaskExecutor<>(this::createConnection, poolConfig, executor); + } + + @Override + public final StreamCancellationHandle synthesize(String text, TtsCallback callback, ModelTarget target) { + AtomicBoolean audioReceived = new AtomicBoolean(); + VoiceStreamCallback streamCallback = adaptCallback(callback, audioReceived); + WebSocketTaskSession session = taskExecutor.openTask( + target, + buildTaskParam(target), + streamCallback, + () -> !audioReceived.get() + ); + try { + // 按协议限制分块发送 + for (String chunk : splitChunks(text)) { + session.send(chunk); + } + session.finish().whenComplete((ignored, throwable) -> { + if (throwable != null) { + streamCallback.onError(throwable); + } + }); + return session::cancel; + } catch (RuntimeException exception) { + session.cancel(); + throw exception; + } + } + + /** + * 按长度上限分块 + */ + private List splitChunks(String text) { + List chunks = new ArrayList<>(); + for (int start = 0; start < text.length(); start += CHUNK_MAX_LEN) { + String chunk = text.substring(start, Math.min(start + CHUNK_MAX_LEN, text.length())); + if (!chunk.isBlank()) { + chunks.add(chunk); + } + } + return chunks; + } + + protected abstract P buildTaskParam(ModelTarget target); + + protected abstract C createConnection(ModelTarget target); + + private VoiceStreamCallback adaptCallback(TtsCallback callback, + AtomicBoolean audioReceived) { + return new VoiceStreamCallback<>() { + @Override + protected void onValidPacket(byte[] packet) { + if (packet.length > 0) { + audioReceived.set(true); + } + callback.onAudio(packet); + } + + @Override + protected void onTaskComplete() { + callback.onComplete(); + } + + @Override + protected void onTaskError(Throwable throwable) { + callback.onError(throwable); + } + }; + } + + @Override + public final void close() { + taskExecutor.close(); + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/BaiLianTtsClient.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/BaiLianTtsClient.java new file mode 100644 index 000000000..c74e9ede5 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/BaiLianTtsClient.java @@ -0,0 +1,224 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.google.gson.Gson; +import com.google.gson.JsonObject; +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.enums.ModelCapability; +import com.nageoffer.ai.ragent.infra.enums.ModelProvider; +import com.nageoffer.ai.ragent.infra.http.HttpResponseHelper; +import com.nageoffer.ai.ragent.infra.http.ModelClientErrorType; +import com.nageoffer.ai.ragent.infra.http.ModelClientException; +import com.nageoffer.ai.ragent.infra.http.ModelUrlResolver; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import com.nageoffer.ai.ragent.infra.voice.websocket.VoiceConnection; +import jakarta.annotation.PreDestroy; +import okhttp3.Request; +import okhttp3.WebSocket; +import okio.ByteString; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; + +import java.util.concurrent.Executor; + +/** + * 阿里云百炼 CosyVoice TTS 客户端 + */ +@Component +public class BaiLianTtsClient extends AbstractWebSocketTtsClient< + BaiLianTtsClient.BaiLianTtsTaskParam, + BaiLianTtsClient.BaiLianTtsConnection> { + + private final WebSocket.Factory webSocketFactory; + private final AIModelProperties.WebSocketConfig websocketConfig; + + public BaiLianTtsClient(@Qualifier("streamingHttpClient") WebSocket.Factory webSocketFactory, + @Qualifier("webSocketLifecycleExecutor") Executor taskExecutor, + AIModelProperties properties) { + super(taskExecutor, properties.getWebsocket()); + this.webSocketFactory = webSocketFactory; + this.websocketConfig = properties.getWebsocket(); + } + + @Override + public String provider() { + return ModelProvider.BAI_LIAN.getId(); + } + + @Override + protected BaiLianTtsConnection createConnection(ModelTarget target) { + return new BaiLianTtsConnection(target, webSocketFactory, websocketConfig); + } + + @Override + protected BaiLianTtsTaskParam buildTaskParam(ModelTarget target) { + return new BaiLianTtsTaskParam( + HttpResponseHelper.requireModel(target, "TTS"), + requireVoice(target) + ); + } + + /** + * 获取候选模型配置的音色 + */ + private String requireVoice(ModelTarget target) { + String voice = target.candidate().getVoice(); + if (voice == null || voice.isBlank()) { + throw new ModelClientException("TTS 未配置默认音色，modelId=" + target.id(), + ModelClientErrorType.CLIENT_ERROR, null); + } + return voice; + } + + @PreDestroy + public void destroy() { + close(); + } + + record BaiLianTtsTaskParam( + String model, + String voice + ) { + } + + static final class BaiLianTtsConnection + extends VoiceConnection { + + private final Gson gson = new Gson(); + + private BaiLianTtsConnection(ModelTarget target, + WebSocket.Factory webSocketFactory, + AIModelProperties.WebSocketConfig websocketConfig) { + super(target, webSocketFactory, websocketConfig); + } + + @Override + protected Request buildWebSocketRequest() { + AIModelProperties.ProviderConfig provider = HttpResponseHelper.requireProvider(target(), "TTS"); + HttpResponseHelper.requireApiKey(provider, "TTS"); + String url = toWebSocketUrl(ModelUrlResolver.resolveUrl(provider, target().candidate(), ModelCapability.TTS)); + Request.Builder request = new Request.Builder() + .url(url) + .addHeader("Authorization", "Bearer " + provider.getApiKey()); + if (provider.getWorkspace() != null && !provider.getWorkspace().isBlank()) { + request.addHeader("X-DashScope-WorkSpace", provider.getWorkspace()); + } + return request.build(); + } + + @Override + protected void doStartTask(String taskId, BaiLianTtsTaskParam param) { + JsonObject parameters = new JsonObject(); + parameters.addProperty("text_type", "PlainText"); + parameters.addProperty("voice", param.voice()); + parameters.addProperty("format", "mp3"); + + JsonObject payload = new JsonObject(); + payload.addProperty("task_group", "audio"); + payload.addProperty("task", "tts"); + payload.addProperty("function", "SpeechSynthesizer"); + payload.addProperty("model", param.model()); + payload.add("parameters", parameters); + payload.add("input", new JsonObject()); + + sendJson(command("run-task", taskId, payload)); + } + + @Override + protected void doSend(String taskId, String text) { + JsonObject input = new JsonObject(); + input.addProperty("text", text); + JsonObject payload = new JsonObject(); + payload.add("input", input); + sendJson(command("continue-task", taskId, payload)); + } + + @Override + protected void doFinishTask(String taskId) { + JsonObject payload = new JsonObject(); + payload.add("input", new JsonObject()); + sendJson(command("finish-task", taskId, payload)); + } + + @Override + protected void doCancelTask(String taskId) { + JsonObject input = new JsonObject(); + input.addProperty("directive", "cancel"); + JsonObject payload = new JsonObject(); + payload.add("input", input); + sendJson(command("finish-task", taskId, payload)); + } + + private JsonObject command(String action, String taskId, JsonObject payload) { + JsonObject header = new JsonObject(); + header.addProperty("action", action); + header.addProperty("task_id", taskId); + header.addProperty("streaming", "duplex"); + + JsonObject command = new JsonObject(); + command.add("header", header); + command.add("payload", payload); + return command; + } + + private void sendJson(JsonObject message) { + WebSocket current = webSocket(); + if (!current.send(gson.toJson(message))) { + throw new ModelClientException("TTS WebSocket 发送失败，modelId=" + modelId(), + ModelClientErrorType.NETWORK_ERROR, null); + } + } + + @Override + protected void handleTextMessage(String text) { + JsonObject response = gson.fromJson(text, JsonObject.class); + JsonObject header = response.getAsJsonObject("header"); + if (header == null || !header.has("event")) { + return; + } + String responseTaskId = header.has("task_id") ? header.get("task_id").getAsString() : null; + validateResponseTaskId(responseTaskId); + String event = header.get("event").getAsString(); + switch (event) { + case "task-started" -> markTaskStarted(); + case "task-finished" -> markTaskFinished(); + case "task-failed" -> failTask(header); + default -> { + // 忽略无需处理的元信息事件 + } + } + } + + @Override + protected byte[] decodeBinaryMessage(ByteString bytes) { + return bytes.toByteArray(); + } + + private void failTask(JsonObject header) { + String errorCode = header.has("error_code") ? header.get("error_code").getAsString() : "unknown"; + String errorMessage = header.has("error_message") ? header.get("error_message").getAsString() : "unknown"; + markTaskFailed(new ModelClientException( + "TTS 任务失败，modelId=" + modelId() + ": " + errorCode + " - " + errorMessage, + ModelClientErrorType.SERVER_ERROR, + null + )); + } + + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/BinaryProbeStreamBridge.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/BinaryProbeStreamBridge.java new file mode 100644 index 000000000..e2386cb1e --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/BinaryProbeStreamBridge.java @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * 音频流首包探测桥接器 + */ +final class BinaryProbeStreamBridge implements TtsCallback { + + private enum Disposition { + PENDING, + COMMITTED, + DISCARDED + } + + private final TtsCallback downstream; + private final CompletableFuture probe = new CompletableFuture<>(); + private final Object lock = new Object(); + private final List buffer = new ArrayList<>(); + private Disposition disposition = Disposition.PENDING; + private boolean terminated; + + BinaryProbeStreamBridge(TtsCallback downstream) { + this.downstream = downstream; + } + + @Override + public void onAudio(byte[] audio) { + if (audio.length == 0) { + return; + } + accept(ProbeResult.success(), false, () -> downstream.onAudio(audio)); + } + + @Override + public void onComplete() { + accept(ProbeResult.noContent(), true, downstream::onComplete); + } + + @Override + public void onError(Throwable throwable) { + accept(ProbeResult.error(throwable), true, () -> downstream.onError(throwable)); + } + + ProbeResult awaitFirstAudio(long timeout, TimeUnit unit) throws InterruptedException { + try { + return probe.get(timeout, unit); + } catch (TimeoutException exception) { + return ProbeResult.timeout(); + } catch (ExecutionException exception) { + return ProbeResult.error(exception.getCause()); + } + } + + void commit() { + synchronized (lock) { + disposition = Disposition.COMMITTED; + buffer.forEach(Runnable::run); + buffer.clear(); + } + } + + void discard() { + synchronized (lock) { + disposition = Disposition.DISCARDED; + buffer.clear(); + } + } + + private void accept(ProbeResult result, boolean terminal, Runnable action) { + synchronized (lock) { + if (terminated || disposition == Disposition.DISCARDED) { + return; + } + terminated = terminal; + probe.complete(result); + if (disposition == Disposition.PENDING) { + buffer.add(action); + return; + } + action.run(); + } + } + + static final class ProbeResult { + + enum Type { + SUCCESS, + ERROR, + TIMEOUT, + NO_CONTENT + } + + private final Type type; + private final Throwable error; + + private ProbeResult(Type type, Throwable error) { + this.type = type; + this.error = error; + } + + Type getType() { + return type; + } + + Throwable getError() { + return error; + } + + boolean isSuccess() { + return type == Type.SUCCESS; + } + + private static ProbeResult success() { + return new ProbeResult(Type.SUCCESS, null); + } + + private static ProbeResult error(Throwable throwable) { + return new ProbeResult(Type.ERROR, throwable); + } + + private static ProbeResult timeout() { + return new ProbeResult(Type.TIMEOUT, null); + } + + private static ProbeResult noContent() { + return new ProbeResult(Type.NO_CONTENT, null); + } + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/RoutingTtsService.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/RoutingTtsService.java new file mode 100644 index 000000000..0f86d2593 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/RoutingTtsService.java @@ -0,0 +1,216 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.nageoffer.ai.ragent.framework.errorcode.BaseErrorCode; +import com.nageoffer.ai.ragent.framework.exception.RemoteException; +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import com.nageoffer.ai.ragent.infra.http.ModelClientErrorType; +import com.nageoffer.ai.ragent.infra.http.ModelClientException; +import com.nageoffer.ai.ragent.infra.model.ModelHealthStore; +import com.nageoffer.ai.ragent.infra.model.ModelSelector; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Primary; +import org.springframework.stereotype.Service; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * TTS 模型路由服务 + */ +@Slf4j +@Service +@Primary +public class RoutingTtsService implements TtsService { + + private static final String STREAM_FAILED_MESSAGE = "TTS 流式任务失败"; + private static final String STREAM_TIMEOUT_MESSAGE = "TTS 首音频超时"; + private static final String STREAM_NO_CONTENT_MESSAGE = "TTS 未返回有效音频"; + + private final ModelSelector selector; + private final ModelHealthStore healthStore; + private final Map clientsByProvider; + + public RoutingTtsService(ModelSelector selector, + ModelHealthStore healthStore, + List clients) { + this.selector = selector; + this.healthStore = healthStore; + this.clientsByProvider = clients.stream() + .collect(Collectors.toMap(TtsClient::provider, Function.identity())); + } + + @Override + public StreamCancellationHandle synthesize(String text, TtsCallback callback, TtsTaskObserver taskObserver) { + List targets = selector.selectTtsCandidates(); + if (targets.isEmpty()) { + throw notifyAllFailed(callback, null); + } + + Throwable lastError = null; + for (ModelTarget target : targets) { + TtsClient client = resolveClient(target); + if (client == null) { + lastError = new ModelClientException( + "TTS 提供商客户端缺失，provider=" + target.candidate().getProvider() + + "，modelId=" + target.id(), + ModelClientErrorType.CLIENT_ERROR, + null + ); + continue; + } + ModelHealthStore.CallPermit permit = healthStore.allowCall(target.id()); + if (permit == null) { + continue; + } + + try { + BinaryProbeStreamBridge bridge = new BinaryProbeStreamBridge(callback); + StreamCancellationHandle handle; + try { + handle = client.synthesize(text, bridge, target); + } catch (RuntimeException exception) { + bridge.discard(); + lastError = exception; + if (exception instanceof ModelClientException clientException + && clientException.getErrorType() == ModelClientErrorType.RATE_LIMITED) { + log.warn("TTS 暂无可用调用容量，modelId={}", target.id()); + continue; + } + healthStore.markFailure(target.id()); + log.warn("TTS 任务启动失败，modelId={}，provider={}", + target.id(), target.candidate().getProvider(), exception); + continue; + } + + taskObserver.onTaskStarted(handle); + if (taskObserver.isCancelled()) { + bridge.discard(); + return handle; + } + + BinaryProbeStreamBridge.ProbeResult result = awaitFirstAudio(bridge, handle, target); + if (taskObserver.isCancelled()) { + bridge.discard(); + return handle; + } + if (result.isSuccess()) { + healthStore.markSuccess(target.id()); + bridge.commit(); + return handle; + } + + bridge.discard(); + healthStore.markFailure(target.id()); + cancelQuietly(handle, target); + lastError = buildLastErrorAndLog(result, target); + } finally { + // 归还半开探测名额 + healthStore.releaseHalfOpenPermit(permit); + } + } + + throw notifyAllFailed(callback, lastError); + } + + private TtsClient resolveClient(ModelTarget target) { + TtsClient client = clientsByProvider.get(target.candidate().getProvider()); + if (client == null) { + log.warn("TTS 提供商客户端缺失: provider={}，modelId={}", + target.candidate().getProvider(), target.id()); + } + return client; + } + + private BinaryProbeStreamBridge.ProbeResult awaitFirstAudio(BinaryProbeStreamBridge bridge, + StreamCancellationHandle handle, + ModelTarget target) { + try { + long timeoutMs = target.timeoutMs(); + return bridge.awaitFirstAudio(timeoutMs, TimeUnit.MILLISECONDS); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + RemoteException interrupted = new RemoteException( + "TTS 首音频等待被中断", exception, BaseErrorCode.REMOTE_ERROR); + bridge.onError(interrupted); + bridge.commit(); + cancelQuietly(handle, target); + throw interrupted; + } + } + + private Throwable buildLastErrorAndLog(BinaryProbeStreamBridge.ProbeResult result, ModelTarget target) { + String provider = target.candidate().getProvider(); + switch (result.getType()) { + case ERROR -> { + Throwable error = result.getError() != null + ? result.getError() + : new ModelClientException(STREAM_FAILED_MESSAGE, ModelClientErrorType.SERVER_ERROR, null); + log.warn("TTS 失败模型: modelId={}，provider={}，原因: 流式任务失败，切换下一个模型", + target.id(), provider, error); + return error; + } + case TIMEOUT -> { + ModelClientException timeout = new ModelClientException( + STREAM_TIMEOUT_MESSAGE, ModelClientErrorType.NETWORK_ERROR, null); + log.warn("TTS 失败模型: modelId={}，provider={}，原因: 首音频超时，切换下一个模型", + target.id(), provider); + return timeout; + } + case NO_CONTENT -> { + ModelClientException noContent = new ModelClientException( + STREAM_NO_CONTENT_MESSAGE, ModelClientErrorType.INVALID_RESPONSE, null); + log.warn("TTS 失败模型: modelId={}，provider={}，原因: 未返回有效音频，切换下一个模型", + target.id(), provider); + return noContent; + } + default -> { + ModelClientException unknown = new ModelClientException( + STREAM_FAILED_MESSAGE, ModelClientErrorType.SERVER_ERROR, null); + log.warn("TTS 失败模型: modelId={}，provider={}，原因: 流式任务失败（未知类型），切换下一个模型", + target.id(), provider); + return unknown; + } + } + } + + private RuntimeException notifyAllFailed(TtsCallback callback, Throwable lastError) { + RemoteException failure = new RemoteException( + "所有 TTS 模型均调用失败", + lastError, + BaseErrorCode.REMOTE_ERROR + ); + callback.onError(failure); + return failure; + } + + private void cancelQuietly(StreamCancellationHandle handle, ModelTarget target) { + try { + handle.cancel(); + } catch (RuntimeException exception) { + log.warn("TTS 失败候选取消异常，modelId={}，provider={}", + target.id(), target.candidate().getProvider(), exception); + } + } + +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsCallback.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsCallback.java new file mode 100644 index 000000000..023defa8b --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsCallback.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +/** + * TTS 流式回调 + */ +public interface TtsCallback { + + void onAudio(byte[] audio); + + void onComplete(); + + void onError(Throwable throwable); +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsClient.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsClient.java new file mode 100644 index 000000000..9a1b7dfc6 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsClient.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; + +/** + * TTS 客户端接口 + */ +public interface TtsClient { + + String provider(); + + StreamCancellationHandle synthesize(String text, TtsCallback callback, ModelTarget target); +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsService.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsService.java new file mode 100644 index 000000000..57b512c20 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsService.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; + +/** + * 文本转语音服务 + */ +public interface TtsService { + + default StreamCancellationHandle synthesize(String text, TtsCallback callback) { + return synthesize(text, callback, handle -> { + }); + } + + StreamCancellationHandle synthesize(String text, TtsCallback callback, TtsTaskObserver taskObserver); +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsTaskObserver.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsTaskObserver.java new file mode 100644 index 000000000..dca6d45c9 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/tts/TtsTaskObserver.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; + +/** + * TTS 任务生命周期观察器 + */ +@FunctionalInterface +public interface TtsTaskObserver { + + /** + * 供应商任务已启动 + */ + void onTaskStarted(StreamCancellationHandle handle); + + /** + * 当前调用是否已取消 + */ + default boolean isCancelled() { + return false; + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceConnection.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceConnection.java new file mode 100644 index 000000000..5abfc7c4c --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceConnection.java @@ -0,0 +1,520 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.http.ModelClientErrorType; +import com.nageoffer.ai.ragent.infra.http.ModelClientException; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.WebSocket; +import okhttp3.WebSocketListener; +import okio.ByteString; + +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +/** + * 可复用 Voice WebSocket 连接模板 + */ +public abstract class VoiceConnection implements AutoCloseable { + + private final ModelTarget target; + private final WebSocket.Factory webSocketFactory; + private final AIModelProperties.WebSocketConfig webSocketConfig; + private final AtomicReference state = new AtomicReference<>(VoiceConnectionState.CONNECTING); + private final AtomicReference currentTaskId = new AtomicReference<>(); + private final AtomicReference> currentCallback = new AtomicReference<>(); + private final AtomicBoolean cancelling = new AtomicBoolean(); + private final CompletableFuture connectionReady = new CompletableFuture<>(); + private volatile CompletableFuture taskStarted; + private volatile CompletableFuture taskFinished; + private final AtomicReference> packetReceivedSignal = new AtomicReference<>(); + private volatile long lastPacketReceivedMs; + private volatile WebSocket webSocket; + + protected VoiceConnection(ModelTarget target, + WebSocket.Factory webSocketFactory, + AIModelProperties.WebSocketConfig webSocketConfig) { + this.target = target; + this.webSocketFactory = webSocketFactory; + this.webSocketConfig = webSocketConfig; + } + + /** + * 建立 WebSocket 并完成连接级初始化 + */ + public final void connect() { + try { + openWebSocket(); + await(connectionReady, webSocketConfig.getConnectTimeoutMs()); + if (!state.compareAndSet(VoiceConnectionState.CONNECTING, VoiceConnectionState.IDLE)) { + throw new IllegalStateException("Voice 连接在初始化期间失效，modelId=" + modelId()); + } + } catch (Exception exception) { + markBroken(); + throw wrap("Voice WebSocket 建连失败，modelId=" + modelId(), exception); + } + } + + /** + * 在当前空闲连接上启动任务 + */ + public final void startTask(String taskId, P param, VoiceStreamCallback callback) { + if (!state.compareAndSet(VoiceConnectionState.IDLE, VoiceConnectionState.TASK_STARTING)) { + throw new IllegalStateException("Voice 连接当前不可启动任务，modelId=" + modelId() + "，state=" + state.get()); + } + currentTaskId.set(taskId); + currentCallback.set(callback); + taskStarted = new CompletableFuture<>(); + taskFinished = new CompletableFuture<>(); + packetReceivedSignal.set(new CompletableFuture<>()); + lastPacketReceivedMs = 0L; + try { + doStartTask(taskId, param); + await(taskStarted, webSocketConfig.getTaskStartTimeoutMs()); + if (!state.compareAndSet(VoiceConnectionState.TASK_STARTING, VoiceConnectionState.TASK_RUNNING)) { + throw new IllegalStateException("Voice 任务启动期间连接失效，modelId=" + modelId() + "，taskId=" + taskId); + } + } catch (Exception exception) { + RuntimeException failure = wrap("Voice 任务启动失败，modelId=" + modelId() + "，taskId=" + taskId, + exception); + markBrokenAndNotify(failure); + throw failure; + } + } + + /** + * 向当前任务发送内容 + */ + public final void send(String taskId, I request) { + requireCurrentTask(taskId, VoiceConnectionState.TASK_RUNNING); + try { + doSend(taskId, request); + } catch (Exception exception) { + RuntimeException failure = wrap("Voice 任务数据发送失败，modelId=" + modelId() + "，taskId=" + taskId, + exception); + markBrokenAndNotify(failure); + throw failure; + } + } + + /** + * 结束当前任务并等待供应商终态 + */ + public final void finishTask(String taskId) { + requireCurrentTask(taskId, VoiceConnectionState.TASK_RUNNING); + if (!state.compareAndSet(VoiceConnectionState.TASK_RUNNING, VoiceConnectionState.TASK_FINISHING)) { + throw new IllegalStateException("Voice 任务当前不可结束，modelId=" + modelId() + "，taskId=" + taskId + + "，state=" + state.get()); + } + try { + doFinishTask(taskId); + awaitTaskTerminated(taskId); + completeTask(taskId, VoiceConnectionState.TASK_FINISHING); + } catch (Exception exception) { + RuntimeException failure = wrap("Voice 任务结束失败，modelId=" + modelId() + "，taskId=" + taskId, + exception); + markBrokenAndNotify(failure); + throw failure; + } + } + + /** + * 取消当前任务并禁用连接复用 + */ + public final void cancelTask(String taskId) { + if (state.get() == VoiceConnectionState.IDLE && currentTaskId.get() == null) { + return; + } + requireCurrentTask(taskId, VoiceConnectionState.TASK_RUNNING, VoiceConnectionState.TASK_FINISHING, + VoiceConnectionState.TASK_CANCELLING); + VoiceConnectionState previous = state.getAndUpdate(current -> switch (current) { + case TASK_RUNNING, TASK_FINISHING -> VoiceConnectionState.TASK_CANCELLING; + default -> current; + }); + if (previous != VoiceConnectionState.TASK_RUNNING + && previous != VoiceConnectionState.TASK_FINISHING + && previous != VoiceConnectionState.TASK_CANCELLING) { + throw new IllegalStateException("Voice 任务当前不可取消，modelId=" + modelId() + "，taskId=" + taskId + + "，state=" + previous); + } + cancelling.set(true); + try { + if (previous != VoiceConnectionState.TASK_CANCELLING) { + doCancelTask(taskId); + } + awaitTaskTerminated(taskId); + completeTask(taskId, VoiceConnectionState.TASK_CANCELLING); + } catch (Exception exception) { + markBroken(); + throw wrap("Voice 任务取消失败，modelId=" + modelId() + "，taskId=" + taskId, exception); + } finally { + cancelling.set(false); + } + } + + public final String modelId() { + return target.id(); + } + + public final ModelTarget target() { + return target; + } + + public final boolean isReusable() { + return state.get() == VoiceConnectionState.IDLE; + } + + /** + * 处理物理连接异常 + */ + private void connectionBroken(Throwable cause) { + if (markBroken()) { + notifyTaskError(cause); + } + } + + @Override + public final void close() { + VoiceConnectionState previous = state.getAndSet(VoiceConnectionState.CLOSED); + if (previous == VoiceConnectionState.CLOSED) { + return; + } + resetTaskContext(); + try { + closeWebSocket(); + } catch (Exception exception) { + throw wrap("Voice WebSocket 关闭失败，modelId=" + modelId(), exception); + } finally { + webSocket = null; + } + } + + /** + * 构建供应商连接请求 + */ + protected abstract Request buildWebSocketRequest() throws Exception; + + protected final WebSocket webSocket() { + return webSocket; + } + + protected final String toWebSocketUrl(String url) { + if (url.startsWith("https://")) { + return "wss://" + url.substring("https://".length()); + } + if (url.startsWith("http://")) { + return "ws://" + url.substring("http://".length()); + } + return url; + } + + protected abstract void doStartTask(String taskId, P param) throws Exception; + + protected abstract void doSend(String taskId, I request) throws Exception; + + protected abstract void doFinishTask(String taskId) throws Exception; + + protected abstract void doCancelTask(String taskId) throws Exception; + + /** + * 处理供应商文本帧 + */ + protected abstract void handleTextMessage(String text); + + /** + * 将供应商二进制帧转换为业务数据包 + */ + protected abstract O decodeBinaryMessage(ByteString bytes); + + private void openWebSocket() throws Exception { + Request request = buildWebSocketRequest(); + webSocket = webSocketFactory.newWebSocket(request, new Listener()); + } + + private void closeWebSocket() { + WebSocket current = webSocket; + if (current != null && !current.close(1000, "normal")) { + current.cancel(); + } + } + + /** + * 校验供应商事件是否属于当前任务 + */ + protected final void validateResponseTaskId(String responseTaskId) { + String expectedTaskId = currentTaskId.get(); + if (!Objects.equals(expectedTaskId, responseTaskId)) { + throw new ModelClientException( + "Voice WebSocket 响应 taskId 不一致，expected=" + expectedTaskId + "，actual=" + responseTaskId, + ModelClientErrorType.INVALID_RESPONSE, + null + ); + } + } + + /** + * 标记供应商任务已启动 + */ + protected final void markTaskStarted() { + CompletableFuture started = taskStarted; + if (started != null) { + started.complete(null); + } + } + + /** + * 标记供应商任务已正常结束 + */ + protected final void markTaskFinished() { + VoiceStreamCallback callback = currentCallback.get(); + if (callback != null) { + callback.onComplete(); + } + completeTaskFinished(null); + } + + /** + * 标记供应商任务失败 + */ + protected final void markTaskFailed(Throwable throwable) { + if (cancelling.get()) { + // 供应商可能用 task-failed 响应取消请求，此时不再上报业务错误 + markBroken(); + completeTaskFinished(null); + return; + } + CompletableFuture started = taskStarted; + if (started != null) { + started.completeExceptionally(throwable); + } + completeTaskFinished(throwable); + connectionBroken(throwable); + } + + private void failConnection(Throwable throwable) { + if (cancelling.get()) { + // 取消可能导致连接关闭，此时不再上报业务错误 + markBroken(); + completeTaskFinished(null); + return; + } + connectionReady.completeExceptionally(throwable); + CompletableFuture started = taskStarted; + if (started != null) { + started.completeExceptionally(throwable); + } + completeTaskFinished(throwable); + connectionBroken(throwable); + } + + private void completeTaskFinished(Throwable throwable) { + CompletableFuture finished = taskFinished; + if (finished == null) { + return; + } + if (throwable == null) { + finished.complete(null); + } else { + finished.completeExceptionally(throwable); + } + } + + private void awaitTaskTerminated(String taskId) throws Exception { + CompletableFuture finished = taskFinished; + long frameIdleTimeoutMs = webSocketConfig.getTaskPacketIdleTimeoutMs(); + Long configuredFirstPacketTimeoutMs = target.timeoutMs(); + long firstPacketTimeoutMs = configuredFirstPacketTimeoutMs != null + ? configuredFirstPacketTimeoutMs + : frameIdleTimeoutMs; + long waitStartedMs = System.currentTimeMillis(); + long firstPacketDeadline = waitStartedMs + firstPacketTimeoutMs; + long minimumFinishDeadline = waitStartedMs + frameIdleTimeoutMs; + CompletableFuture terminationSignal = finished.handle((ignored, throwable) -> null); + while (!finished.isDone()) { + CompletableFuture packetSignal = packetReceivedSignal.get(); + long last = lastPacketReceivedMs; + long deadline = last > 0 + ? Math.max(minimumFinishDeadline, last + frameIdleTimeoutMs) + : firstPacketDeadline; + long remainingMs = deadline - System.currentTimeMillis(); + if (remainingMs <= 0) { + if (finished.isDone() || last != lastPacketReceivedMs) { + continue; + } + String phase = last > 0 ? "帧间空闲" : "首帧等待"; + throw new TimeoutException("finish-task 后" + phase + "超时，taskId=" + taskId); + } + try { + CompletableFuture.anyOf(terminationSignal, packetSignal) + .get(remainingMs, TimeUnit.MILLISECONDS); + } catch (TimeoutException ignored) { + // 收帧和超时可能同时发生，回到循环后按最新时间重新判断 + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw interrupted; + } finally { + if (packetSignal.isDone()) { + packetReceivedSignal.compareAndSet(packetSignal, new CompletableFuture<>()); + } + } + } + await(finished, frameIdleTimeoutMs); + } + + private void await(CompletableFuture future, long timeoutMs) throws Exception { + try { + future.get(timeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException exception) { + Throwable cause = exception.getCause(); + if (cause instanceof Exception checkedException) { + throw checkedException; + } + throw new RuntimeException(cause); + } + } + + private boolean markBroken() { + while (true) { + VoiceConnectionState current = state.get(); + if (current == VoiceConnectionState.CLOSED) { + return false; + } + if (current == VoiceConnectionState.BROKEN + || state.compareAndSet(current, VoiceConnectionState.BROKEN)) { + return true; + } + } + } + + private void markBrokenAndNotify(Throwable throwable) { + if (markBroken()) { + notifyTaskError(throwable); + } + } + + private void notifyTaskError(Throwable throwable) { + VoiceStreamCallback callback = currentCallback.get(); + if (callback != null) { + callback.onError(throwable); + } + } + + private void completeTask(String taskId, VoiceConnectionState terminalState) { + // finish 和 cancel 可能并发，只有先完成状态迁移的一方负责释放租约 + if (!Objects.equals(taskId, currentTaskId.get())) { + return; + } + VoiceConnectionState completedState = terminalState == VoiceConnectionState.TASK_CANCELLING + ? VoiceConnectionState.BROKEN + : VoiceConnectionState.IDLE; + if (!state.compareAndSet(terminalState, completedState)) { + return; + } + resetTaskContext(); + } + + private void resetTaskContext() { + taskStarted = null; + taskFinished = null; + packetReceivedSignal.set(null); + lastPacketReceivedMs = 0L; + currentTaskId.set(null); + currentCallback.set(null); + } + + private void requireCurrentTask(String taskId, VoiceConnectionState... allowedStates) { + if (!Objects.equals(taskId, currentTaskId.get())) { + throw new IllegalStateException("Voice taskId 与当前任务不一致，modelId=" + modelId() + + "，taskId=" + taskId + "，currentTaskId=" + currentTaskId.get()); + } + VoiceConnectionState currentState = state.get(); + for (VoiceConnectionState allowedState : allowedStates) { + if (currentState == allowedState) { + return; + } + } + throw new IllegalStateException("Voice 任务状态不允许当前操作，modelId=" + modelId() + "，taskId=" + taskId + + "，state=" + currentState); + } + + private final class Listener extends WebSocketListener { + + @Override + public void onOpen(WebSocket webSocket, Response response) { + connectionReady.complete(null); + } + + @Override + public void onMessage(WebSocket webSocket, String text) { + try { + handleTextMessage(text); + } catch (RuntimeException exception) { + failConnection(exception); + } + } + + @Override + public void onMessage(WebSocket webSocket, ByteString bytes) { + try { + lastPacketReceivedMs = System.currentTimeMillis(); + CompletableFuture packetSignal = packetReceivedSignal.get(); + if (packetSignal != null) { + packetSignal.complete(null); + } + O packet = decodeBinaryMessage(bytes); + VoiceStreamCallback callback = currentCallback.get(); + if (callback != null) { + callback.onPacket(packet); + } + } catch (RuntimeException exception) { + failConnection(exception); + } + } + + @Override + public void onClosed(WebSocket webSocket, int code, String reason) { + if (state.get() != VoiceConnectionState.CLOSED) { + failConnection(new ModelClientException( + "Voice WebSocket 已关闭: " + code + " - " + reason, + ModelClientErrorType.NETWORK_ERROR, + null + )); + } + } + + @Override + public void onFailure(WebSocket webSocket, Throwable throwable, Response response) { + failConnection(throwable); + } + } + + private RuntimeException wrap(String message, Exception exception) { + if (exception instanceof ModelClientException modelClientException) { + return modelClientException; + } + return new ModelClientException(message, ModelClientErrorType.NETWORK_ERROR, null, exception); + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceConnectionState.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceConnectionState.java new file mode 100644 index 000000000..4b7b5b40b --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceConnectionState.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +/** + * WebSocket 连接与任务状态 + */ +public enum VoiceConnectionState { + + CONNECTING, + IDLE, + TASK_STARTING, + TASK_RUNNING, + TASK_FINISHING, + TASK_CANCELLING, + BROKEN, + CLOSED +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceStreamCallback.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceStreamCallback.java new file mode 100644 index 000000000..f2ef2231a --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/VoiceStreamCallback.java @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Voice 流式回调 + */ +public abstract class VoiceStreamCallback { + + private final AtomicBoolean terminated = new AtomicBoolean(); + + /** + * 接收业务数据包 + */ + public final void onPacket(E packet) { + if (terminated.get()) { + return; + } + onValidPacket(packet); + } + + /** + * 任务正常完成 + */ + public final void onComplete() { + if (terminated.compareAndSet(false, true)) { + onTaskComplete(); + } + } + + /** + * 任务失败 + */ + public final void onError(Throwable throwable) { + if (terminated.compareAndSet(false, true)) { + onTaskError(throwable); + } + } + + /** + * 接收有效数据包 + */ + protected void onValidPacket(E packet) { + } + + /** + * 处理任务完成 + */ + protected void onTaskComplete() { + } + + /** + * 处理任务失败 + */ + protected void onTaskError(Throwable throwable) { + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketConnectionLease.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketConnectionLease.java new file mode 100644 index 000000000..7e2604206 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketConnectionLease.java @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.pool2.impl.GenericObjectPool; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * WebSocket 连接租约 + */ +@Slf4j +public final class WebSocketConnectionLease> implements AutoCloseable { + + private final GenericObjectPool pool; + private final C connection; + private final AtomicBoolean invalidated = new AtomicBoolean(); + private final AtomicBoolean closed = new AtomicBoolean(); + + WebSocketConnectionLease(GenericObjectPool pool, C connection) { + this.pool = pool; + this.connection = connection; + } + + public C connection() { + return connection; + } + + public void invalidate() { + invalidated.set(true); + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + try { + if (!invalidated.get()) { + pool.returnObject(connection); + } else { + pool.invalidateObject(connection); + } + } catch (Exception exception) { + log.warn("Voice 连接释放失败，modelId={}", connection.modelId(), exception); + } + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketExecutor.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketExecutor.java new file mode 100644 index 000000000..96fb7cd48 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketExecutor.java @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +import com.nageoffer.ai.ragent.framework.errorcode.BaseErrorCode; +import com.nageoffer.ai.ragent.framework.exception.RemoteException; +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.http.ModelClientErrorType; +import com.nageoffer.ai.ragent.infra.http.ModelClientException; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import org.apache.commons.pool2.BasePooledObjectFactory; +import org.apache.commons.pool2.PooledObject; +import org.apache.commons.pool2.impl.DefaultPooledObject; +import org.apache.commons.pool2.impl.GenericObjectPool; +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; + +import java.time.Duration; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; + +/** + * 按 modelId 管理 Voice WebSocket 连接池 + */ +public final class WebSocketExecutor> implements AutoCloseable { + + private final Function connectionFactory; + private final AIModelProperties.WebSocketConfig config; + private final Map> poolsByModelId = new ConcurrentHashMap<>(); + private final AtomicBoolean closed = new AtomicBoolean(); + + public WebSocketExecutor(Function connectionFactory, + AIModelProperties.WebSocketConfig config) { + this.connectionFactory = connectionFactory; + this.config = config; + } + + /** + * 从模型连接池借用空闲 WebSocket + */ + public WebSocketConnectionLease acquire(ModelTarget target) { + if (closed.get()) { + throw new IllegalStateException("Voice 连接池已关闭"); + } + String modelId = target.id(); + + GenericObjectPool pool = poolsByModelId.computeIfAbsent(modelId, ignored -> createPool(target)); + try { + C connection = pool.borrowObject(); + return new WebSocketConnectionLease<>(pool, connection); + } catch (NoSuchElementException exception) { + Throwable cause = exception.getCause(); + // Commons Pool 仅在创建连接失败时保留 cause + if (cause == null) { + throw new ModelClientException("Voice 连接池无可用连接，modelId=" + modelId, + ModelClientErrorType.RATE_LIMITED, null, exception); + } + if (cause instanceof ModelClientException modelClientException) { + throw modelClientException; + } + throw new ModelClientException("Voice 连接创建失败，modelId=" + modelId, + ModelClientErrorType.NETWORK_ERROR, null, cause); + } catch (Exception exception) { + if (exception instanceof RuntimeException runtimeException) { + throw runtimeException; + } + throw new RemoteException("Voice 连接借用失败，modelId=" + modelId, + exception, BaseErrorCode.REMOTE_ERROR); + } + } + + private GenericObjectPool createPool(ModelTarget target) { + GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig<>(); + poolConfig.setMaxTotal(config.getMaxTotalPerModel()); + poolConfig.setMaxIdle(config.getMaxIdlePerModel()); + poolConfig.setBlockWhenExhausted(false); + poolConfig.setTestOnBorrow(true); + poolConfig.setTestOnReturn(true); + if (config.getIdleTimeoutMs() > 0) { + poolConfig.setMinEvictableIdleDuration(Duration.ofMillis(config.getIdleTimeoutMs())); + poolConfig.setTimeBetweenEvictionRuns(Duration.ofMillis(config.getEvictionIntervalMs())); + // -1 表示每轮检查全部空闲连接 + poolConfig.setNumTestsPerEvictionRun(-1); + } + + return new GenericObjectPool<>(new BasePooledObjectFactory<>() { + @Override + public C create() throws Exception { + C connection = connectionFactory.apply(target); + try { + connection.connect(); + return connection; + } catch (Exception exception) { + closeQuietly(connection); + throw exception; + } + } + + @Override + public PooledObject wrap(C connection) { + return new DefaultPooledObject<>(connection); + } + + @Override + public boolean validateObject(PooledObject pooledObject) { + return pooledObject.getObject().isReusable(); + } + + @Override + public void destroyObject(PooledObject pooledObject) { + closeQuietly(pooledObject.getObject()); + } + }, poolConfig); + } + + private void closeQuietly(C connection) { + try { + connection.close(); + } catch (RuntimeException ignored) { + // 销毁失败不应阻塞连接池关闭 + } + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + poolsByModelId.values().forEach(GenericObjectPool::close); + poolsByModelId.clear(); + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketTaskExecutor.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketTaskExecutor.java new file mode 100644 index 000000000..e6c571a74 --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketTaskExecutor.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; + +import java.util.UUID; +import java.util.concurrent.Executor; +import java.util.function.BooleanSupplier; +import java.util.function.Function; + +/** + * WebSocket 任务执行器 + */ +public class WebSocketTaskExecutor> implements AutoCloseable { + + private final WebSocketExecutor webSocketExecutor; + private final Executor taskExecutor; + + public WebSocketTaskExecutor(Function connectionFactory, + AIModelProperties.WebSocketConfig poolConfig, + Executor taskExecutor) { + this.webSocketExecutor = new WebSocketExecutor<>(connectionFactory, poolConfig); + this.taskExecutor = taskExecutor; + } + + public WebSocketTaskSession openTask(ModelTarget target, + P taskParam, + VoiceStreamCallback callback, + BooleanSupplier invalidateOnFinish) { + WebSocketConnectionLease lease = webSocketExecutor.acquire(target); + String taskId = generateTaskId(); + try { + lease.connection().startTask(taskId, taskParam, callback); + return new WebSocketTaskSession<>(taskId, lease.connection(), lease, taskExecutor, invalidateOnFinish); + } catch (RuntimeException exception) { + lease.invalidate(); + lease.close(); + throw exception; + } + } + + protected String generateTaskId() { + return UUID.randomUUID().toString(); + } + + @Override + public void close() { + webSocketExecutor.close(); + } +} diff --git a/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketTaskSession.java b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketTaskSession.java new file mode 100644 index 000000000..4ee66a10c --- /dev/null +++ b/infra-ai/src/main/java/com/nageoffer/ai/ragent/infra/voice/websocket/WebSocketTaskSession.java @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.websocket; + +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; +import java.util.function.BooleanSupplier; + +/** + * WebSocket 任务会话 + */ +@Slf4j +public final class WebSocketTaskSession { + + private enum Lifecycle { + ACTIVE, + FINISHING, + CANCELLING, + RELEASED + } + + private final String taskId; + private final VoiceConnection connection; + private final WebSocketConnectionLease lease; + private final Executor taskExecutor; + private final BooleanSupplier invalidateOnFinish; + private final CompletableFuture completion = new CompletableFuture<>(); + private final Object lifecycleLock = new Object(); + private Lifecycle lifecycle = Lifecycle.ACTIVE; + + > WebSocketTaskSession(String taskId, + C connection, + WebSocketConnectionLease lease, + Executor taskExecutor, + BooleanSupplier invalidateOnFinish) { + this.taskId = taskId; + this.connection = connection; + this.lease = lease; + this.taskExecutor = taskExecutor; + this.invalidateOnFinish = invalidateOnFinish; + } + + public void send(I input) { + synchronized (lifecycleLock) { + connection.send(taskId, input); + } + } + + public CompletionStage finish() { + synchronized (lifecycleLock) { + if (lifecycle != Lifecycle.ACTIVE) { + return completion; + } + lifecycle = Lifecycle.FINISHING; + try { + taskExecutor.execute(this::finishTask); + } catch (RuntimeException exception) { + lease.invalidate(); + lifecycle = Lifecycle.RELEASED; + lease.close(); + completion.completeExceptionally(exception); + } + } + return completion; + } + + public void cancel() { + synchronized (lifecycleLock) { + lease.invalidate(); + if (lifecycle == Lifecycle.RELEASED || lifecycle == Lifecycle.CANCELLING) { + return; + } + lifecycle = Lifecycle.CANCELLING; + } + + Throwable failure = null; + try { + connection.cancelTask(taskId); + } catch (RuntimeException exception) { + failure = exception; + log.warn("WebSocket 任务取消失败，modelId={}，taskId={}", connection.modelId(), taskId, exception); + } finally { + synchronized (lifecycleLock) { + lifecycle = Lifecycle.RELEASED; + } + lease.close(); + if (failure == null) { + completion.complete(null); + } else { + completion.completeExceptionally(failure); + } + } + } + + private void finishTask() { + Throwable failure = null; + try { + connection.finishTask(taskId); + } catch (RuntimeException exception) { + failure = exception; + lease.invalidate(); + } + + boolean release; + synchronized (lifecycleLock) { + release = lifecycle == Lifecycle.FINISHING; + if (release) { + lifecycle = Lifecycle.RELEASED; + } + } + if (!release) { + return; + } + + if (failure == null && invalidateOnFinish.getAsBoolean()) { + lease.invalidate(); + } + lease.close(); + if (failure == null) { + completion.complete(null); + } else { + completion.completeExceptionally(failure); + } + } +} diff --git a/infra-ai/src/test/java/com/nageoffer/ai/ragent/infra/voice/tts/BaiLianTtsClientTest.java b/infra-ai/src/test/java/com/nageoffer/ai/ragent/infra/voice/tts/BaiLianTtsClientTest.java new file mode 100644 index 000000000..23cfc7426 --- /dev/null +++ b/infra-ai/src/test/java/com/nageoffer/ai/ragent/infra/voice/tts/BaiLianTtsClientTest.java @@ -0,0 +1,471 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.google.gson.Gson; +import com.google.gson.JsonObject; +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import okhttp3.Protocol; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.WebSocket; +import okhttp3.WebSocketListener; +import okio.ByteString; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class BaiLianTtsClientTest { + + private final Gson gson = new Gson(); + + @Test + void sendsMinimalCosyVoiceProtocolAndReturnsMp3() { + FakeWebSocketFactory factory = new FakeWebSocketFactory(); + BaiLianTtsClient client = new BaiLianTtsClient(factory, Runnable::run, properties()); + RecordingCallback callback = new RecordingCallback(); + + StreamCancellationHandle handle = client.synthesize( + "你好，世界。", + callback, + target() + ); + + JsonObject runTask = factory.messages.get(0); + JsonObject runPayload = runTask.getAsJsonObject("payload"); + JsonObject parameters = runPayload.getAsJsonObject("parameters"); + assertEquals(Set.of("text_type", "voice", "format"), parameters.keySet()); + assertEquals("PlainText", parameters.get("text_type").getAsString()); + assertEquals("longxiaochun", parameters.get("voice").getAsString()); + assertEquals("mp3", parameters.get("format").getAsString()); + assertEquals("audio", runPayload.get("task_group").getAsString()); + assertEquals("cosyvoice-v3-flash", runPayload.get("model").getAsString()); + + assertEquals(List.of("run-task", "continue-task", "finish-task"), factory.actions()); + assertEquals(1, factory.taskIds().stream().distinct().count()); + assertEquals("你好，世界。", factory.messages.get(1) + .getAsJsonObject("payload") + .getAsJsonObject("input") + .get("text") + .getAsString()); + assertEquals("Bearer test-key", factory.request.header("Authorization")); + assertEquals("test-workspace", factory.request.header("X-DashScope-WorkSpace")); + assertTrue(factory.request.url().isHttps()); + assertEquals("dashscope.aliyuncs.com", factory.request.url().host()); + assertEquals("/api-ws/v1/inference", factory.request.url().encodedPath()); + assertArrayEquals(FakeWebSocketFactory.MP3_AUDIO, callback.audio); + assertTrue(callback.completed); + + handle.cancel(); + assertEquals(0, factory.closeCount.get()); + client.close(); + } + + @Test + void splitsLongTextIntoSeparateContinueTasks() { + FakeWebSocketFactory factory = new FakeWebSocketFactory(); + BaiLianTtsClient client = new BaiLianTtsClient(factory, Runnable::run, properties()); + + String longText = "测试".repeat(41); + client.synthesize(longText, new RecordingCallback(), target()); + + // run-task + 2 个 continue-task + finish-task + assertEquals(List.of("run-task", "continue-task", "continue-task", "finish-task"), factory.actions()); + assertEquals(longText.substring(0, 80), factory.messages.get(1) + .getAsJsonObject("payload").getAsJsonObject("input").get("text").getAsString()); + assertEquals(longText.substring(80), factory.messages.get(2) + .getAsJsonObject("payload").getAsJsonObject("input").get("text").getAsString()); + + client.close(); + } + + @Test + void waitsForFirstAudioUsingTtsTimeout() { + FakeWebSocketFactory factory = new FakeWebSocketFactory(true, false, 450L, 0L); + BaiLianTtsClient client = new BaiLianTtsClient(factory, Runnable::run, properties(300L)); + RecordingCallback callback = new RecordingCallback(); + + client.synthesize("延迟首帧", callback, target()); + + assertArrayEquals(FakeWebSocketFactory.MP3_AUDIO, callback.audio); + assertTrue(callback.completed); + assertNull(callback.error); + client.close(); + } + + @Test + void keepsFullFinishGraceWhenAudioArrivesBeforeFinishDirective() { + FakeWebSocketFactory factory = new FakeWebSocketFactory( + true, false, 0L, 200L, true, 200L); + BaiLianTtsClient client = new BaiLianTtsClient(factory, Runnable::run, properties(300L)); + RecordingCallback callback = new RecordingCallback(); + + client.synthesize("提前返回音频", callback, target()); + + assertArrayEquals(FakeWebSocketFactory.MP3_AUDIO, callback.audio); + assertTrue(callback.completed); + assertNull(callback.error); + client.close(); + } + + @Test + void renewsFinishWaitForEveryAudioFrame() { + FakeWebSocketFactory factory = new FakeWebSocketFactory( + true, false, 0L, 200L, 3, 200L); + BaiLianTtsClient client = new BaiLianTtsClient(factory, Runnable::run, properties(300L)); + RecordingCallback callback = new RecordingCallback(); + + client.synthesize("持续流式返回音频", callback, target()); + + assertArrayEquals(FakeWebSocketFactory.MP3_AUDIO, callback.audio); + assertTrue(callback.completed); + assertNull(callback.error); + client.close(); + } + + @Test + void timesOutWhenAudioFrameGapExceedsConfiguredIdleTimeout() { + FakeWebSocketFactory factory = new FakeWebSocketFactory(true, false, 0L, 450L); + BaiLianTtsClient client = new BaiLianTtsClient(factory, Runnable::run, properties(300L)); + RecordingCallback callback = new RecordingCallback(); + + client.synthesize("帧间超时", callback, target()); + + assertArrayEquals(FakeWebSocketFactory.MP3_AUDIO, callback.audio); + assertFalse(callback.completed); + assertNotNull(callback.error); + client.close(); + } + + @Test + void sendsCancelDirectiveBeforeFirstAudio() throws Exception { + FakeWebSocketFactory factory = new FakeWebSocketFactory(false); + ExecutorService executor = Executors.newSingleThreadExecutor(); + BaiLianTtsClient client = new BaiLianTtsClient(factory, executor, properties()); + RecordingCallback callback = new RecordingCallback(); + + try { + StreamCancellationHandle handle = client.synthesize("取消播放", callback, target()); + assertTrue(factory.awaitNormalFinish()); + + handle.cancel(); + + JsonObject cancel = factory.messages.get(factory.messages.size() - 1); + assertEquals("finish-task", cancel.getAsJsonObject("header").get("action").getAsString()); + assertEquals("cancel", cancel.getAsJsonObject("payload") + .getAsJsonObject("input") + .get("directive") + .getAsString()); + assertNull(callback.audio); + assertTrue(callback.completed); + assertEquals(1, factory.closeCount.get()); + } finally { + client.close(); + executor.shutdownNow(); + } + } + + @Test + void invalidatesConnectionWhenCancelReceivesTaskFailed() throws Exception { + FakeWebSocketFactory factory = new FakeWebSocketFactory(false, true); + ExecutorService executor = Executors.newSingleThreadExecutor(); + BaiLianTtsClient client = new BaiLianTtsClient(factory, executor, properties()); + + try { + StreamCancellationHandle handle = client.synthesize("取消播放", new RecordingCallback(), target()); + assertTrue(factory.awaitNormalFinish()); + + handle.cancel(); + + assertEquals(1, factory.closeCount.get()); + } finally { + client.close(); + executor.shutdownNow(); + } + } + + private ModelTarget target() { + AIModelProperties.ModelCandidate candidate = new AIModelProperties.ModelCandidate(); + candidate.setId("cosyvoice-v3-flash"); + candidate.setProvider("bailian"); + candidate.setModel("cosyvoice-v3-flash"); + candidate.setVoice("longxiaochun"); + + AIModelProperties.ProviderConfig provider = new AIModelProperties.ProviderConfig(); + provider.setUrl("https://dashscope.aliyuncs.com"); + provider.setApiKey("test-key"); + provider.setWorkspace("test-workspace"); + provider.setEndpoints(Map.of("tts", "/api-ws/v1/inference")); + return new ModelTarget(candidate.getId(), candidate, provider, 1000L); + } + + private AIModelProperties properties() { + return properties(1000L); + } + + private AIModelProperties properties(long taskPacketIdleTimeoutMs) { + AIModelProperties.WebSocketConfig config = new AIModelProperties.WebSocketConfig(); + config.setConnectTimeoutMs(1000L); + config.setTaskStartTimeoutMs(1000L); + config.setTaskPacketIdleTimeoutMs(taskPacketIdleTimeoutMs); + config.setMaxTotalPerModel(1); + config.setMaxIdlePerModel(1); + + AIModelProperties properties = new AIModelProperties(); + properties.setWebsocket(config); + return properties; + } + + private static final class RecordingCallback implements TtsCallback { + + private byte[] audio; + private boolean completed; + private Throwable error; + + @Override + public void onAudio(byte[] audio) { + this.audio = audio; + } + + @Override + public void onComplete() { + completed = true; + } + + @Override + public void onError(Throwable throwable) { + error = throwable; + } + } + + private final class FakeWebSocketFactory implements WebSocket.Factory { + + private static final byte[] MP3_AUDIO = {73, 68, 51}; + + private final List messages = new ArrayList<>(); + private final AtomicInteger closeCount = new AtomicInteger(); + private final CountDownLatch normalFinish = new CountDownLatch(1); + private final boolean autoFinish; + private final boolean failOnCancel; + private final long firstAudioDelayMs; + private final long finishDelayMs; + private final boolean audioBeforeFinish; + private final long continueTaskDelayMs; + private final int audioFrameCount; + private final long audioFrameIntervalMs; + private Request request; + + private FakeWebSocketFactory() { + this(true, false, 0L, 0L); + } + + private FakeWebSocketFactory(boolean autoFinish) { + this(autoFinish, false, 0L, 0L); + } + + private FakeWebSocketFactory(boolean autoFinish, boolean failOnCancel) { + this(autoFinish, failOnCancel, 0L, 0L); + } + + private FakeWebSocketFactory(boolean autoFinish, boolean failOnCancel, + long firstAudioDelayMs, long finishDelayMs) { + this(autoFinish, failOnCancel, firstAudioDelayMs, finishDelayMs, false, 0L, 1, 0L); + } + + private FakeWebSocketFactory(boolean autoFinish, boolean failOnCancel, + long firstAudioDelayMs, long finishDelayMs, + boolean audioBeforeFinish, long continueTaskDelayMs) { + this(autoFinish, failOnCancel, firstAudioDelayMs, finishDelayMs, + audioBeforeFinish, continueTaskDelayMs, 1, 0L); + } + + private FakeWebSocketFactory(boolean autoFinish, boolean failOnCancel, + long firstAudioDelayMs, long finishDelayMs, + int audioFrameCount, long audioFrameIntervalMs) { + this(autoFinish, failOnCancel, firstAudioDelayMs, finishDelayMs, + false, 0L, audioFrameCount, audioFrameIntervalMs); + } + + private FakeWebSocketFactory(boolean autoFinish, boolean failOnCancel, + long firstAudioDelayMs, long finishDelayMs, + boolean audioBeforeFinish, long continueTaskDelayMs, + int audioFrameCount, long audioFrameIntervalMs) { + this.autoFinish = autoFinish; + this.failOnCancel = failOnCancel; + this.firstAudioDelayMs = firstAudioDelayMs; + this.finishDelayMs = finishDelayMs; + this.audioBeforeFinish = audioBeforeFinish; + this.continueTaskDelayMs = continueTaskDelayMs; + this.audioFrameCount = audioFrameCount; + this.audioFrameIntervalMs = audioFrameIntervalMs; + } + + @Override + public WebSocket newWebSocket(Request request, WebSocketListener listener) { + this.request = request; + FakeWebSocket webSocket = new FakeWebSocket(request, listener); + Response response = new Response.Builder() + .request(request) + .protocol(Protocol.HTTP_1_1) + .code(101) + .message("Switching Protocols") + .build(); + listener.onOpen(webSocket, response); + return webSocket; + } + + private List actions() { + return messages.stream() + .map(message -> message.getAsJsonObject("header").get("action").getAsString()) + .toList(); + } + + private List taskIds() { + return messages.stream() + .map(message -> message.getAsJsonObject("header").get("task_id").getAsString()) + .toList(); + } + + private boolean awaitNormalFinish() throws InterruptedException { + return normalFinish.await(1, TimeUnit.SECONDS); + } + + private final class FakeWebSocket implements WebSocket { + + private final Request request; + private final WebSocketListener listener; + + private FakeWebSocket(Request request, WebSocketListener listener) { + this.request = request; + this.listener = listener; + } + + @Override + public Request request() { + return request; + } + + @Override + public long queueSize() { + return 0; + } + + @Override + public boolean send(String text) { + JsonObject message = gson.fromJson(text, JsonObject.class); + messages.add(message); + String action = message.getAsJsonObject("header").get("action").getAsString(); + String taskId = message.getAsJsonObject("header").get("task_id").getAsString(); + if ("run-task".equals(action)) { + listener.onMessage(this, event(taskId, "task-started")); + } else if ("continue-task".equals(action) && audioBeforeFinish) { + listener.onMessage(this, ByteString.of(MP3_AUDIO)); + sleep(continueTaskDelayMs); + } else if ("finish-task".equals(action)) { + JsonObject input = message.getAsJsonObject("payload").getAsJsonObject("input"); + if (input.has("directive")) { + listener.onMessage(this, event(taskId, failOnCancel ? "task-failed" : "task-finished")); + } else { + normalFinish.countDown(); + if (autoFinish) { + respondWithAudio(taskId); + } + } + } + return true; + } + + private void respondWithAudio(String taskId) { + Runnable response = () -> { + if (!audioBeforeFinish) { + sleep(firstAudioDelayMs); + for (int frame = 0; frame < audioFrameCount; frame++) { + if (frame > 0) { + sleep(audioFrameIntervalMs); + } + listener.onMessage(this, ByteString.of(MP3_AUDIO)); + } + } + sleep(finishDelayMs); + listener.onMessage(this, event(taskId, "task-finished")); + }; + if (firstAudioDelayMs == 0L && finishDelayMs == 0L) { + response.run(); + return; + } + Thread responseThread = new Thread(response, "fake-tts-response"); + responseThread.setDaemon(true); + responseThread.start(); + } + + private void sleep(long delayMs) { + if (delayMs <= 0L) { + return; + } + try { + Thread.sleep(delayMs); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + } + } + + @Override + public boolean send(ByteString bytes) { + return true; + } + + @Override + public boolean close(int code, String reason) { + closeCount.incrementAndGet(); + listener.onClosed(this, code, reason == null ? "" : reason); + return true; + } + + @Override + public void cancel() { + } + + private String event(String taskId, String event) { + JsonObject header = new JsonObject(); + header.addProperty("task_id", taskId); + header.addProperty("event", event); + JsonObject response = new JsonObject(); + response.add("header", header); + response.add("payload", new JsonObject()); + return gson.toJson(response); + } + } + } +} diff --git a/infra-ai/src/test/java/com/nageoffer/ai/ragent/infra/voice/tts/RoutingTtsServiceTest.java b/infra-ai/src/test/java/com/nageoffer/ai/ragent/infra/voice/tts/RoutingTtsServiceTest.java new file mode 100644 index 000000000..0098e208a --- /dev/null +++ b/infra-ai/src/test/java/com/nageoffer/ai/ragent/infra/voice/tts/RoutingTtsServiceTest.java @@ -0,0 +1,290 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.infra.voice.tts; + +import com.nageoffer.ai.ragent.framework.exception.RemoteException; +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import com.nageoffer.ai.ragent.infra.config.AIModelProperties; +import com.nageoffer.ai.ragent.infra.http.ModelClientErrorType; +import com.nageoffer.ai.ragent.infra.http.ModelClientException; +import com.nageoffer.ai.ragent.infra.model.ModelHealthStore; +import com.nageoffer.ai.ragent.infra.model.ModelSelector; +import com.nageoffer.ai.ragent.infra.model.ModelTarget; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class RoutingTtsServiceTest { + + @Test + void usesModelHealthStoreForConsecutiveFailures() { + AIModelProperties properties = singleModelProperties(); + ModelHealthStore healthStore = new ModelHealthStore(properties); + FailingTtsClient client = new FailingTtsClient("test"); + RoutingTtsService service = service(properties, healthStore, List.of(client)); + + assertThrows(RemoteException.class, + () -> service.synthesize("第一次", new RecordingCallback())); + assertFalse(healthStore.isUnavailable("tts-model")); + + assertThrows(RemoteException.class, + () -> service.synthesize("第二次", new RecordingCallback())); + assertTrue(healthStore.isUnavailable("tts-model")); + assertEquals(2, client.attempts.get()); + assertEquals(2, client.invalidatingCancellations.get()); + } + + @Test + void discardsFailedCandidateEventsAndCommitsFirstAudioFromFallback() { + AIModelProperties properties = fallbackProperties(); + ModelHealthStore healthStore = new ModelHealthStore(properties); + FailingTtsClient primary = new FailingTtsClient("primary"); + SuccessfulTtsClient backup = new SuccessfulTtsClient("backup", new byte[]{2, 3}); + RoutingTtsService service = service(properties, healthStore, List.of(primary, backup)); + RecordingCallback callback = new RecordingCallback(); + + StreamCancellationHandle handle = service.synthesize("你好", callback); + + assertArrayEquals(new byte[]{2, 3}, callback.audioEvents.get(0)); + assertEquals(1, callback.audioEvents.size()); + assertTrue(callback.completed); + assertEquals(0, callback.errors.get()); + assertEquals(1, primary.attempts.get()); + assertEquals(1, backup.attempts.get()); + handle.cancel(); + } + + @Test + void releasesHalfOpenPermitWhenStartedTaskIsCancelledBeforeFirstAudio() { + AIModelProperties properties = fallbackProperties(); + properties.getSelection().setFailureThreshold(1); + properties.getSelection().setOpenDurationMs(0L); + ModelHealthStore healthStore = new ModelHealthStore(properties); + CancelCompletingTtsClient primary = new CancelCompletingTtsClient("primary"); + SuccessfulTtsClient backup = new SuccessfulTtsClient("backup", new byte[]{2, 3}); + RoutingTtsService service = service(properties, healthStore, List.of(primary, backup)); + RecordingCallback callback = new RecordingCallback(); + AtomicBoolean cancelled = new AtomicBoolean(); + healthStore.markFailure("primary-model"); + + service.synthesize("立即取消", callback, new TtsTaskObserver() { + @Override + public void onTaskStarted(StreamCancellationHandle handle) { + cancelled.set(true); + handle.cancel(); + } + + @Override + public boolean isCancelled() { + return cancelled.get(); + } + }); + + assertEquals(1, primary.cancellations.get()); + assertEquals(0, backup.attempts.get()); + assertTrue(callback.audioEvents.isEmpty()); + assertFalse(callback.completed); + assertEquals(0, callback.errors.get()); + assertFalse(healthStore.isUnavailable("primary-model")); + assertNotNull(healthStore.allowCall("primary-model")); + } + + @Test + void releasesHalfOpenPermitWhenConnectionPoolIsExhausted() { + AIModelProperties properties = singleModelProperties(); + properties.getSelection().setFailureThreshold(1); + properties.getSelection().setOpenDurationMs(0L); + ModelHealthStore healthStore = new ModelHealthStore(properties); + PoolExhaustedTtsClient client = new PoolExhaustedTtsClient("test"); + RoutingTtsService service = service(properties, healthStore, List.of(client)); + healthStore.markFailure("tts-model"); + + assertThrows(RemoteException.class, + () -> service.synthesize("连接池耗尽", new RecordingCallback())); + + assertEquals(1, client.attempts.get()); + assertFalse(healthStore.isUnavailable("tts-model")); + assertNotNull(healthStore.allowCall("tts-model")); + } + + private RoutingTtsService service(AIModelProperties properties, + ModelHealthStore healthStore, + List clients) { + return new RoutingTtsService( + new ModelSelector(properties, healthStore), + healthStore, + clients + ); + } + + private AIModelProperties singleModelProperties() { + AIModelProperties properties = new AIModelProperties(); + properties.getProviders().put("test", new AIModelProperties.ProviderConfig()); + properties.getTts().setDefaultModel("tts-model"); + properties.getTts().setTimeoutMs(1000L); + properties.getTts().setCandidates(List.of(candidate("tts-model", "test", 1))); + return properties; + } + + private AIModelProperties fallbackProperties() { + AIModelProperties properties = new AIModelProperties(); + properties.getProviders().put("primary", new AIModelProperties.ProviderConfig()); + properties.getProviders().put("backup", new AIModelProperties.ProviderConfig()); + properties.getTts().setDefaultModel("primary-model"); + properties.getTts().setTimeoutMs(1000L); + properties.getTts().setCandidates(List.of( + candidate("primary-model", "primary", 1), + candidate("backup-model", "backup", 2) + )); + return properties; + } + + private AIModelProperties.ModelCandidate candidate(String id, String provider, int priority) { + AIModelProperties.ModelCandidate candidate = new AIModelProperties.ModelCandidate(); + candidate.setId(id); + candidate.setProvider(provider); + candidate.setModel(id); + candidate.setPriority(priority); + return candidate; + } + + private static final class FailingTtsClient implements TtsClient { + + private final String provider; + private final AtomicInteger attempts = new AtomicInteger(); + private final AtomicInteger invalidatingCancellations = new AtomicInteger(); + + private FailingTtsClient(String provider) { + this.provider = provider; + } + + @Override + public String provider() { + return provider; + } + + @Override + public StreamCancellationHandle synthesize(String text, TtsCallback callback, ModelTarget target) { + attempts.incrementAndGet(); + callback.onError(new IllegalStateException("provider failed")); + return invalidatingCancellations::incrementAndGet; + } + } + + private static final class SuccessfulTtsClient implements TtsClient { + + private final String provider; + private final byte[] audio; + private final AtomicInteger attempts = new AtomicInteger(); + + private SuccessfulTtsClient(String provider, byte[] audio) { + this.provider = provider; + this.audio = audio; + } + + @Override + public String provider() { + return provider; + } + + @Override + public StreamCancellationHandle synthesize(String text, TtsCallback callback, ModelTarget target) { + attempts.incrementAndGet(); + callback.onAudio(audio); + callback.onComplete(); + return () -> { + }; + } + } + + private static final class CancelCompletingTtsClient implements TtsClient { + + private final String provider; + private final AtomicInteger cancellations = new AtomicInteger(); + + private CancelCompletingTtsClient(String provider) { + this.provider = provider; + } + + @Override + public String provider() { + return provider; + } + + @Override + public StreamCancellationHandle synthesize(String text, TtsCallback callback, ModelTarget target) { + return () -> { + cancellations.incrementAndGet(); + callback.onComplete(); + }; + } + } + + private static final class PoolExhaustedTtsClient implements TtsClient { + + private final String provider; + private final AtomicInteger attempts = new AtomicInteger(); + + private PoolExhaustedTtsClient(String provider) { + this.provider = provider; + } + + @Override + public String provider() { + return provider; + } + + @Override + public StreamCancellationHandle synthesize(String text, TtsCallback callback, ModelTarget target) { + attempts.incrementAndGet(); + throw new ModelClientException("pool exhausted", ModelClientErrorType.RATE_LIMITED, null); + } + } + + private static final class RecordingCallback implements TtsCallback { + + private final List audioEvents = new ArrayList<>(); + private final AtomicInteger errors = new AtomicInteger(); + private boolean completed; + + @Override + public void onAudio(byte[] audio) { + audioEvents.add(audio); + } + + @Override + public void onComplete() { + completed = true; + } + + @Override + public void onError(Throwable throwable) { + errors.incrementAndGet(); + } + } +} diff --git a/pom.xml b/pom.xml index fe9ae4e85..2681f4b30 100644 --- a/pom.xml +++ b/pom.xml @@ -28,6 +28,7 @@ 0.22.0 1.18 5.3.2 + 2.12.1 1.1.2 2.0.2 3.0.6 @@ -178,6 +179,12 @@ ${okhttp.version} + + org.apache.commons + commons-pool2 + ${commons-pool2.version} + + io.modelcontextprotocol.sdk mcp diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/config/ThreadPoolExecutorConfig.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/config/ThreadPoolExecutorConfig.java index 18a711ba9..d02e78644 100644 --- a/rag/src/main/java/com/nageoffer/ai/ragent/rag/config/ThreadPoolExecutorConfig.java +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/config/ThreadPoolExecutorConfig.java @@ -19,6 +19,7 @@ import cn.hutool.core.thread.ThreadFactoryBuilder; import com.alibaba.ttl.threadpool.TtlExecutors; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -173,6 +174,54 @@ public Executor modelStreamExecutor() { return TtlExecutors.getTtlExecutor(executor); } + /** + * Voice WebSocket 任务收尾线程池 + */ + @Bean + public Executor webSocketLifecycleExecutor( + @Value("${rag.voice.executors.websocket-lifecycle.core-pool-size}") int corePoolSize, + @Value("${rag.voice.executors.websocket-lifecycle.max-pool-size}") int maxPoolSize, + @Value("${rag.voice.executors.websocket-lifecycle.keep-alive-seconds}") long keepAliveSeconds, + @Value("${rag.voice.executors.websocket-lifecycle.thread-name-prefix}") String threadNamePrefix) { + ThreadPoolExecutor executor = new ThreadPoolExecutor( + corePoolSize, + maxPoolSize, + keepAliveSeconds, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + ThreadFactoryBuilder.create() + .setNamePrefix(threadNamePrefix) + .build(), + new ThreadPoolExecutor.AbortPolicy() + ); + executor.allowCoreThreadTimeOut(true); + return TtlExecutors.getTtlExecutor(executor); + } + + /** + * 消息语音播放合成线程池 + */ + @Bean + public Executor voicePlaybackExecutor( + @Value("${rag.voice.executors.playback.core-pool-size}") int corePoolSize, + @Value("${rag.voice.executors.playback.max-pool-size}") int maxPoolSize, + @Value("${rag.voice.executors.playback.keep-alive-seconds}") long keepAliveSeconds, + @Value("${rag.voice.executors.playback.thread-name-prefix}") String threadNamePrefix) { + ThreadPoolExecutor executor = new ThreadPoolExecutor( + corePoolSize, + maxPoolSize, + keepAliveSeconds, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + ThreadFactoryBuilder.create() + .setNamePrefix(threadNamePrefix) + .build(), + new ThreadPoolExecutor.AbortPolicy() + ); + executor.allowCoreThreadTimeOut(true); + return TtlExecutors.getTtlExecutor(executor); + } + /** * SSE 排队后执行入口线程池 */ diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/controller/VoicePlaybackController.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/controller/VoicePlaybackController.java new file mode 100644 index 000000000..ea641ae1f --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/controller/VoicePlaybackController.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.controller; + +import com.nageoffer.ai.ragent.framework.convention.Result; +import com.nageoffer.ai.ragent.framework.idempotent.IdempotentSubmit; +import com.nageoffer.ai.ragent.framework.web.Results; +import com.nageoffer.ai.ragent.rag.config.RAGDefaultProperties; +import com.nageoffer.ai.ragent.rag.service.VoicePlaybackService; +import lombok.RequiredArgsConstructor; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * 消息语音播放控制器 + */ +@RestController +@RequiredArgsConstructor +public class VoicePlaybackController { + + private final VoicePlaybackService voicePlaybackService; + private final RAGDefaultProperties ragDefaultProperties; + + /** + * 发起消息语音播放 + */ + @GetMapping(value = "/rag/v3/voice/play", produces = "text/event-stream;charset=UTF-8") + public SseEmitter play(@RequestParam String messageId) { + SseEmitter emitter = new SseEmitter(ragDefaultProperties.getSseTimeoutMs()); + voicePlaybackService.play(messageId, emitter); + return emitter; + } + + /** + * 停止指定播放任务 + */ + @IdempotentSubmit + @PostMapping(value = "/rag/v3/voice/stop") + public Result stop(@RequestParam String taskId) { + voicePlaybackService.stop(taskId); + return Results.success(); + } +} diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/AudioFramePayload.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/AudioFramePayload.java new file mode 100644 index 000000000..cd0e0d2b3 --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/AudioFramePayload.java @@ -0,0 +1,24 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.dto; + +/** + * 音频帧事件载荷 + */ +public record AudioFramePayload(String base64) { +} diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/AudioMetaPayload.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/AudioMetaPayload.java new file mode 100644 index 000000000..da38aea7d --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/dto/AudioMetaPayload.java @@ -0,0 +1,24 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.dto; + +/** + * 音频元信息事件载荷 + */ +public record AudioMetaPayload(String taskId) { +} diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java index 87f748f50..0b55d3185 100644 --- a/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/enums/SSEEventType.java @@ -53,7 +53,17 @@ public enum SSEEventType { /** * 拒绝事件 */ - REJECT("reject"); + REJECT("reject"), + + /** + * 音频帧事件 + */ + AUDIO("audio"), + + /** + * 音频元信息事件 + */ + AUDIO_META("audio-meta"); private final String value; diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/VoicePlaybackService.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/VoicePlaybackService.java new file mode 100644 index 000000000..3c65a9438 --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/VoicePlaybackService.java @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.service; + +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * 消息语音播放服务 + */ +public interface VoicePlaybackService { + + /** + * 播放指定消息 + */ + void play(String messageId, SseEmitter emitter); + + /** + * 停止指定播放任务 + */ + void stop(String taskId); +} diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamCallbackFactory.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamCallbackFactory.java index 6bf189941..e99b2ff8d 100644 --- a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamCallbackFactory.java +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/StreamCallbackFactory.java @@ -62,4 +62,11 @@ public StreamCallback createChatEventHandler(SseEmitter emitter, return new StreamChatEventHandler(params); } + + /** + * 创建语音播放事件处理器 + */ + public VoicePlaybackEventHandler createVoicePlaybackEventHandler(SseEmitter emitter, String taskId) { + return new VoicePlaybackEventHandler(emitter, taskId, taskManager); + } } diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackEventHandler.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackEventHandler.java new file mode 100644 index 000000000..945c65f65 --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackEventHandler.java @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.service.handler; + +import com.nageoffer.ai.ragent.framework.web.SseEmitterSender; +import com.nageoffer.ai.ragent.framework.web.StreamTaskManager; +import com.nageoffer.ai.ragent.infra.chat.StreamCancellationHandle; +import com.nageoffer.ai.ragent.infra.voice.tts.TtsCallback; +import com.nageoffer.ai.ragent.infra.voice.tts.TtsTaskObserver; +import com.nageoffer.ai.ragent.rag.dto.AudioFramePayload; +import com.nageoffer.ai.ragent.rag.dto.AudioMetaPayload; +import com.nageoffer.ai.ragent.rag.dto.CompletionPayload; +import com.nageoffer.ai.ragent.rag.enums.SSEEventType; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.Base64; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * 语音播放 SSE 事件处理器 + */ +@Slf4j +public class VoicePlaybackEventHandler implements TtsCallback, TtsTaskObserver { + + private final String taskId; + private final SseEmitterSender sender; + private final StreamTaskManager taskManager; + private final AtomicBoolean terminated = new AtomicBoolean(); + private final AtomicBoolean firstFrameLogged = new AtomicBoolean(); + + public VoicePlaybackEventHandler(SseEmitter emitter, String taskId, StreamTaskManager taskManager) { + this.taskId = taskId; + this.sender = new SseEmitterSender(emitter); + this.taskManager = taskManager; + initialize(emitter); + } + + private void initialize(SseEmitter emitter) { + taskManager.register(taskId, () -> { + sender.sendEvent(SSEEventType.CANCEL.value(), new CompletionPayload(null, null)); + sender.sendEvent(SSEEventType.DONE.value(), "[DONE]"); + sender.complete(); + }); + bindEmitterCancellation(emitter); + sender.sendEvent(SSEEventType.AUDIO_META.value(), new AudioMetaPayload(taskId)); + log.info("播放任务发起，taskId={}", taskId); + } + + @Override + public void onTaskStarted(StreamCancellationHandle handle) { + taskManager.bindHandle(taskId, wrapCancellationHandle(handle)::cancel); + } + + @Override + public boolean isCancelled() { + return taskManager.isCancelled(taskId); + } + + String taskId() { + return taskId; + } + + @Override + public void onAudio(byte[] audio) { + if (terminated.get() || isCancelled()) { + return; + } + if (firstFrameLogged.compareAndSet(false, true)) { + log.info("播放任务首帧下发，taskId={}", taskId); + } + sender.sendEvent(SSEEventType.AUDIO.value(), new AudioFramePayload( + Base64.getEncoder().encodeToString(audio))); + } + + @Override + public void onComplete() { + if (isCancelled() || !terminated.compareAndSet(false, true)) { + return; + } + log.info("播放任务完成，taskId={}", taskId); + sender.sendEvent(SSEEventType.DONE.value(), "[DONE]"); + taskManager.unregister(taskId); + sender.complete(); + } + + @Override + public void onError(Throwable throwable) { + terminateWithError("播放任务失败，taskId={}", throwable); + } + + public void onStartFailure(Throwable throwable) { + if (isCancelled()) { + log.info("播放任务已取消，忽略启动异常，taskId={}", taskId); + taskManager.unregister(taskId); + return; + } + terminateWithError("播放任务启动失败，taskId={}", throwable); + } + + public void onRejected(Throwable throwable) { + terminateWithError("播放任务线程池拒绝，taskId={}", throwable); + } + + private void terminateWithError(String message, Throwable throwable) { + if (isCancelled() || !terminated.compareAndSet(false, true)) { + return; + } + log.error(message, taskId, throwable); + taskManager.unregister(taskId); + sender.fail(throwable); + } + + private void bindEmitterCancellation(SseEmitter emitter) { + Runnable cancel = () -> { + if (terminated.compareAndSet(false, true) && !isCancelled()) { + taskManager.cancel(taskId); + } + }; + emitter.onCompletion(cancel); + emitter.onTimeout(cancel); + emitter.onError(ignored -> cancel.run()); + } + + private StreamCancellationHandle wrapCancellationHandle(StreamCancellationHandle handle) { + // 取消指令与已在途的 finish-task 存在竞态，停止后不复用当前连接 + return () -> { + try { + handle.cancel(); + log.info("播放任务已取消，taskId={}", taskId); + } catch (RuntimeException exception) { + log.warn("播放任务取消失败，taskId={}", taskId, exception); + } + }; + } +} diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackTaskRunner.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackTaskRunner.java new file mode 100644 index 000000000..8f5de609e --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/handler/VoicePlaybackTaskRunner.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.service.handler; + +import com.nageoffer.ai.ragent.infra.voice.tts.TtsService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; + +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; + +/** + * 在线程池中启动语音播放任务 + */ +@Slf4j +@Component +public class VoicePlaybackTaskRunner { + + private final TtsService ttsService; + private final Executor voicePlaybackExecutor; + + public VoicePlaybackTaskRunner(TtsService ttsService, + @Qualifier("voicePlaybackExecutor") Executor voicePlaybackExecutor) { + this.ttsService = ttsService; + this.voicePlaybackExecutor = voicePlaybackExecutor; + } + + public void run(String text, VoicePlaybackEventHandler callback) { + try { + voicePlaybackExecutor.execute(() -> synthesize(text, callback)); + } catch (RejectedExecutionException exception) { + callback.onRejected(exception); + } + } + + private void synthesize(String text, VoicePlaybackEventHandler callback) { + if (callback.isCancelled()) { + return; + } + try { + ttsService.synthesize(text, callback, callback); + log.info("播放任务已启动，taskId={}", callback.taskId()); + } catch (RuntimeException exception) { + callback.onStartFailure(exception); + } + } +} diff --git a/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/impl/VoicePlaybackServiceImpl.java b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/impl/VoicePlaybackServiceImpl.java new file mode 100644 index 000000000..26d8420d7 --- /dev/null +++ b/rag/src/main/java/com/nageoffer/ai/ragent/rag/service/impl/VoicePlaybackServiceImpl.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +package com.nageoffer.ai.ragent.rag.service.impl; + +import cn.hutool.core.util.StrUtil; +import cn.hutool.core.util.IdUtil; +import com.baomidou.mybatisplus.core.toolkit.Wrappers; +import com.nageoffer.ai.ragent.framework.context.UserContext; +import com.nageoffer.ai.ragent.framework.exception.ClientException; +import com.nageoffer.ai.ragent.framework.web.StreamTaskManager; +import com.nageoffer.ai.ragent.rag.dao.entity.ConversationMessageDO; +import com.nageoffer.ai.ragent.rag.dao.mapper.ConversationMessageMapper; +import com.nageoffer.ai.ragent.rag.service.VoicePlaybackService; +import com.nageoffer.ai.ragent.rag.service.handler.StreamCallbackFactory; +import com.nageoffer.ai.ragent.rag.service.handler.VoicePlaybackEventHandler; +import com.nageoffer.ai.ragent.rag.service.handler.VoicePlaybackTaskRunner; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * 消息语音播放服务默认实现 + */ +@Service +@RequiredArgsConstructor +public class VoicePlaybackServiceImpl implements VoicePlaybackService { + + private static final String ROLE_ASSISTANT = "assistant"; + + private final ConversationMessageMapper conversationMessageMapper; + private final StreamCallbackFactory callbackFactory; + private final VoicePlaybackTaskRunner taskRunner; + private final StreamTaskManager taskManager; + + @Override + public void play(String messageId, SseEmitter emitter) { + String taskId = IdUtil.getSnowflakeNextIdStr(); + ConversationMessageDO message = loadAssistantMessage(messageId, UserContext.getUserId()); + String text = message.getContent(); + if (StrUtil.isBlank(text)) { + throw new ClientException("消息内容为空，无法播放"); + } + + VoicePlaybackEventHandler callback = callbackFactory.createVoicePlaybackEventHandler(emitter, taskId); + taskRunner.run(text, callback); + } + + @Override + public void stop(String taskId) { + taskManager.cancel(taskId); + } + + /** + * 定位当前用户的 assistant 消息 + */ + private ConversationMessageDO loadAssistantMessage(String messageId, String userId) { + ConversationMessageDO message = conversationMessageMapper.selectOne( + Wrappers.lambdaQuery(ConversationMessageDO.class) + .eq(ConversationMessageDO::getId, messageId) + .eq(ConversationMessageDO::getUserId, userId) + .eq(ConversationMessageDO::getRole, ROLE_ASSISTANT) + .eq(ConversationMessageDO::getDeleted, 0) + ); + if (message == null) { + throw new ClientException("消息不存在"); + } + return message; + } +} diff --git a/rag/src/test/java/com/nageoffer/ai/ragent/infra/model/ModelSelectorTest.java b/rag/src/test/java/com/nageoffer/ai/ragent/infra/model/ModelSelectorTest.java index 6bf61f287..e26daa51a 100644 --- a/rag/src/test/java/com/nageoffer/ai/ragent/infra/model/ModelSelectorTest.java +++ b/rag/src/test/java/com/nageoffer/ai/ragent/infra/model/ModelSelectorTest.java @@ -161,4 +161,17 @@ private static AIModelProperties buildProperties() { List targets = selector.selectChatCandidates(false); assertEquals(List.of("qwen3-local", "gpt-5.4"), ids(targets)); } + + @Test + void TTS模型组超时写入ModelTarget统一超时() { + AIModelProperties.ModelCandidate ttsCandidate = cand("tts-model", "bailian", "tts-model", false); + properties.getTts().setDefaultModel("tts-model"); + properties.getTts().setCandidates(List.of(ttsCandidate)); + properties.getTts().setTimeoutMs(1000L); + + List targets = selector.selectTtsCandidates(); + + assertEquals(List.of("tts-model"), ids(targets)); + assertEquals(1000L, targets.get(0).timeoutMs()); + } }