整合千问realtime模型

package nexus.io.voice.agent.bridge;

import java.util.concurrent.CompletableFuture;

/**
 * 统一抽象：任何“实时语音/音频”模型桥接都实现这个接口。
 * VoiceSocketHandler 只依赖此接口。
 */
public interface RealtimeModelBridge {

  //public RealtimeModelBridge(RealtimeBridgeCallback sender,String url,String model,String voiceName)

  /**
   * 建立到模型的会话连接，并完成必要的 session/setup 配置。
   */
  CompletableFuture<Void> connect(RealtimeSetup setup);

  /**
   * 发送一段上行音频（浏览器推来的 PCM16 16kHz mono little-endian）。
   */
  CompletableFuture<Void> sendPcm16k(byte[] pcm16k);

  /**
   * 结束当前音频输入/触发模型生成：
   * - Gemini：sendAudioStreamEnd()
   * - Qwen Manual：commit + response.create
   * - Qwen Server VAD：可以是 no-op（或作为手动触发的补充）
   */
  CompletableFuture<Void> endAudioInput();

  /**
   * 可选：发送文本输入（聊天框）。
   */
  CompletableFuture<Void> sendText(String text);

  /**
   * 关闭会话并释放资源。
   */
  CompletableFuture<Void> close();
}

package nexus.io.voice.agent.bridge;

import java.util.Arrays;
import java.util.Base64;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

import com.alibaba.dashscope.audio.omni.OmniRealtimeCallback;
import com.alibaba.dashscope.audio.omni.OmniRealtimeConfig;
import com.alibaba.dashscope.audio.omni.OmniRealtimeConversation;
import com.alibaba.dashscope.audio.omni.OmniRealtimeModality;
import com.alibaba.dashscope.audio.omni.OmniRealtimeParam;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import nexus.io.tio.utils.environment.EnvUtils;
import nexus.io.tio.utils.hutool.StrUtil;
import nexus.io.tio.utils.json.JsonUtils;
import nexus.io.voice.agent.model.WsVoiceAgentResponseMessage;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class QwenOmniRealtimeBridge implements RealtimeModelBridge {

  // 中国区（北京）
  private String url = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime";
  private String model = "qwen3-omni-flash-realtime";
  private String voiceName = "Cherry";

  private final RealtimeBridgeCallback callback;

  private volatile OmniRealtimeConversation conversation;
  private final AtomicBoolean connected = new AtomicBoolean(false);

  // 前端上行是 16k PCM16；Qwen3-Omni-Flash-Realtime 下行通常是 24k PCM16（pcm24）
  // 注意：DashScope 事件里音频是 base64；仍可给浏览器发 bytes（二进制）
  public QwenOmniRealtimeBridge(RealtimeBridgeCallback callback) {
    this.callback = callback;
  }

  public QwenOmniRealtimeBridge(RealtimeBridgeCallback callback, String url, String model, String voiceName) {
    this.callback = callback;
    if (url != null) {
      this.url = url;
    }
    if (model != null) {
      this.model = model;
    }

    if (voiceName != null) {
      this.voiceName = voiceName;
    }
  }

  public CompletableFuture<Void> connect(RealtimeSetup setup) {
    return CompletableFuture.runAsync(() -> {
      try {
        String apiKey = EnvUtils.getStr("DASHSCOPE_API_KEY");
        if (StrUtil.isBlank(apiKey)) {
          throw new IllegalStateException("DASHSCOPE_API_KEY is empty");
        }

        OmniRealtimeParam param = OmniRealtimeParam.builder().model(model).apikey(apiKey).url(url).build();

        this.conversation = new OmniRealtimeConversation(param, new OmniRealtimeCallback() {
          @Override
          public void onOpen() {
            connected.set(true);
            sendJson(new WsVoiceAgentResponseMessage("qwen_connected", model));
          }

          @Override
          public void onClose(int code, String reason) {
            connected.set(false);
            sendJson(new WsVoiceAgentResponseMessage("close", reason));
            callback.close("dashscope closed: " + code + ", " + reason);
          }

          @Override
          public void onEvent(JsonObject event) {
            handleEvent(event);
          }
        });

        conversation.connect();

        // 会话配置：建议默认 server_vad + text/audio + transcription
        OmniRealtimeConfig cfg = buildSessionConfig(setup);
        conversation.updateSession(cfg);

        sendJson(new WsVoiceAgentResponseMessage("setup_sent_to_model"));

      } catch (NoApiKeyException e) {
        log.error("NoApiKeyException", e);
        sendError("no_api_key", e.getMessage());
        callback.close("no api key");
      } catch (Exception e) {
        log.error("connect error", e);
        sendError("connect_error", e.getMessage());
        callback.close("connect failed");
      }
    });
  }

  public CompletableFuture<Void> close() {
    return CompletableFuture.runAsync(() -> {
      try {
        OmniRealtimeConversation c = this.conversation;
        if (c != null) {
          c.close(1000, "server close");
        }
      } catch (Exception ignore) {
      } finally {
        connected.set(false);
        callback.close("close");
      }
    });
  }

  /**
   * 浏览器推来的 16k PCM16 bytes（little-endian）
   * DashScope 需要 base64 后通过 input_audio_buffer.append 事件发送（SDK 封装为 appendAudio）。
   */
  public CompletableFuture<Void> sendPcm16k(byte[] pcm16k) {
    return CompletableFuture.runAsync(() -> {
      OmniRealtimeConversation c = this.conversation;
      if (c == null || !connected.get() || pcm16k == null || pcm16k.length == 0) {
        return;
      }
      try {
        String b64 = Base64.getEncoder().encodeToString(pcm16k);
        c.appendAudio(b64);
      } catch (Exception e) {
        log.error("appendAudio failed", e);
        sendError("append_audio_failed", e.getMessage());
      }
    });
  }

  /**
   * Manual 模式：把现有的 audio_end 映射为 commit + createResponse
   * 如果使用 server_vad，可不调用这个方法。
   */
  public CompletableFuture<Void> commitAndCreateResponse() {
    return CompletableFuture.runAsync(() -> {
      OmniRealtimeConversation c = this.conversation;
      if (c == null || !connected.get())
        return;
      try {
        c.commit();
        c.createResponse(null, null);
      } catch (Exception e) {
        log.error("commit/createResponse failed", e);
        sendError("commit_failed", e.getMessage());
      }
    });
  }

  /**
   * 文本输入：SDK 里有的版本支持发文本项；若只做语音助手，可先不实现。
   * 如果确实要支持“输入框发文本”，建议用 Realtime 的对话项事件（conversation.item.create）
   * 或使用 SDK 提供的文本接口（视 SDK 版本而定）。
   */
  public CompletableFuture<Void> sendText(String text) {
    // 先给出最安全的行为：转为 instructions 追加或忽略
    return CompletableFuture.completedFuture(null);
  }

  private OmniRealtimeConfig buildSessionConfig(RealtimeSetup setup) {
    // 把的 setup 组合成 instructions
    String instructions = buildInstructions(setup);

    // server_vad（通话模式）：enableTurnDetection(true)
    // manual（按下即说）：enableTurnDetection(false)，并在 audio_end 时 commit+createResponse
    boolean useServerVad = true;

    List<OmniRealtimeModality> modalities = Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT);
    OmniRealtimeConfig.OmniRealtimeConfigBuilder b = OmniRealtimeConfig.builder()
        //
        .modalities(modalities).voice(voiceName)
        //
        .enableInputAudioTranscription(true)
        //
        .enableTurnDetection(useServerVad);
    //

    if (instructions != null) {
      b.parameters(Map.of("instructions", instructions));
    }

    return b.build();
  }

  private String buildInstructions(RealtimeSetup setup) {
    if (setup == null) {
      return null;
    }

    StringBuilder sb = new StringBuilder();
    if (StrUtil.notBlank(setup.getSystem_prompt()))
      sb.append(setup.getSystem_prompt()).append("\n");
    if (StrUtil.notBlank(setup.getJob_description()))
      sb.append(setup.getJob_description()).append("\n");
    if (StrUtil.notBlank(setup.getResume()))
      sb.append(setup.getResume()).append("\n");
    if (StrUtil.notBlank(setup.getGreeting()))
      sb.append(setup.getGreeting()).append("\n");
    if (StrUtil.notBlank(setup.getQuestions()))
      sb.append(setup.getQuestions()).append("\n");
    return sb.length() == 0 ? null : sb.toString();
  }

  private void handleEvent(JsonObject event) {
    try {
      String type = event.has("type") ? event.get("type").getAsString() : "";

      switch (type) {

      // 会话创建/更新
      case "session.created":
        sendJson(new WsVoiceAgentResponseMessage("setup_complete"));
        break;
      case "session.updated":
        // 可选：记录配置
        break;

      // 服务端 VAD 生命周期（可用于前端“打断播放”）
      case "input_audio_buffer.speech_started":
        sendJson(new WsVoiceAgentResponseMessage("speech_started"));
        break;
      case "input_audio_buffer.speech_stopped":
        sendJson(new WsVoiceAgentResponseMessage("speech_stopped"));
        break;

      // 用户输入转写完成
      case "conversation.item.input_audio_transcription.completed": {
        String transcript = event.has("transcript") ? event.get("transcript").getAsString() : "";
        sendJson(new WsVoiceAgentResponseMessage("transcript_in", transcript));
        break;
      }

      // 模型输出字幕（增量/完成）
      case "response.audio_transcript.delta": {
        String delta = event.has("delta") ? event.get("delta").getAsString() : "";
        sendJson(new WsVoiceAgentResponseMessage("transcript_out", delta));
        break;
      }
      case "response.audio_transcript.done": {
        String transcript = event.has("transcript") ? event.get("transcript").getAsString() : "";
        // 也可以用 text 事件发完整句
        sendJson(new WsVoiceAgentResponseMessage("text", transcript));
        break;
      }

      // 模型输出音频（base64）
      case "response.audio.delta": {
        String b64 = event.has("delta") ? event.get("delta").getAsString() : "";
        if (StrUtil.notBlank(b64)) {
          byte[] pcm = Base64.getDecoder().decode(b64);
          callback.sendBinary(pcm);
        }
        break;
      }
      case "response.audio.done":
        // 音频段结束（可选）
        break;

      // 一轮完成
      case "response.done":
        sendJson(new WsVoiceAgentResponseMessage("turn_complete"));
        break;

      // 错误
      case "error":
        // 不同版本字段可能是 error/message
        sendError("remote_error", event.toString());
        break;

      default:
        // 需要排查时打开
        // sendJson(new WsVoiceAgentResponseMessage("evt", event.toString()));
        break;
      }
    } catch (Exception e) {
      log.error("handleEvent error", e);
      sendError("handle_event_error", e.getMessage());
    }
  }

  private void sendJson(WsVoiceAgentResponseMessage msg) {
    try {
      String json = JsonUtils.toSkipNullJson(msg);
      callback.sendText(json);
    } catch (Exception e) {
      callback.sendText("{\"type\":\"error\",\"message\":\"serialize error\"}");
    }
  }

  private void sendError(String where, String message) {
    WsVoiceAgentResponseMessage m = new WsVoiceAgentResponseMessage("error");
    m.setWhere(where);
    m.setMessage(message == null ? "" : message);
    sendJson(m);
  }

  @Override
  public CompletableFuture<Void> endAudioInput() {
    return CompletableFuture.completedFuture(null);
  }
}

package nexus.io.voice.agent.bridge;

import nexus.io.consts.ModelPlatformName;
import nexus.io.tio.utils.environment.EnvUtils;

public class RealtimeModelBridgeFactory {

  public static RealtimeModelBridge createBridge(String platform, RealtimeBridgeCallback callback) {
    if (platform == null) {
      platform = EnvUtils.getStr("vioce.agent.platform");
    }
    RealtimeModelBridge bridge = null;
    if (ModelPlatformName.GOOGLE.equals(platform)) {
      bridge = new GoogleGeminiRealtimeBridge(callback);

    } else if (ModelPlatformName.BAILIAN.equals(platform)) {
      bridge = new QwenOmniRealtimeBridge(callback);

    } else {
      bridge = new GoogleGeminiRealtimeBridge(callback);
    }
    return bridge;
  }
}

package nexus.io.voice.agent.handler;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.tio.http.common.HttpRequest;
import nexus.io.tio.http.common.HttpResponse;
import nexus.io.tio.utils.json.JsonUtils;
import nexus.io.tio.websocket.common.WebSocketRequest;
import nexus.io.tio.websocket.common.WebSocketResponse;
import nexus.io.tio.websocket.common.WebSocketSessionContext;
import nexus.io.tio.websocket.server.handler.IWebSocketHandler;
import nexus.io.voice.agent.audio.SessionAudioRecorder;
import nexus.io.voice.agent.bridge.RealtimeBridgeCallback;
import nexus.io.voice.agent.bridge.RealtimeModelBridge;
import nexus.io.voice.agent.bridge.RealtimeModelBridgeFactory;
import nexus.io.voice.agent.bridge.RealtimeSetup;
import nexus.io.voice.agent.callback.WsRealtimeBridgeCallback;
import nexus.io.voice.agent.consts.VoiceAgentConst;
import nexus.io.voice.agent.model.WsVoiceAgentRequestMessage;
import nexus.io.voice.agent.model.WsVoiceAgentResponseMessage;
import nexus.io.voice.agent.model.WsVoiceAgentType;
import nexus.io.voice.agent.utils.ChannelContextUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class VoiceSocketHandler implements IWebSocketHandler {
  // 一个前端连接一个 bridge
  private static final Map<String, RealtimeModelBridge> BRIDGES = new ConcurrentHashMap<>();

  @Override
  public HttpResponse handshake(HttpRequest httpRequest, HttpResponse response, ChannelContext channelContext)
      throws Exception {
    log.info("请求信息: {}", httpRequest);
    return response;
  }

  @Override
  public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext)
      throws Exception {
    log.info("握手完成: {}", httpRequest);
  }

  @Override
  public Object onClose(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
    String k = ChannelContextUtils.key(channelContext);
    RealtimeModelBridge bridge = BRIDGES.remove(k);
    if (bridge != null) {
      bridge.close();
    }
    Tio.remove(channelContext, "客户端主动关闭连接");
    return null;
  }

  @Override
  public Object onBytes(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
    String k = ChannelContextUtils.key(channelContext);
    // 前端推：16k PCM mono 裸流,记录用户上行音频（前端发来 16k PCM）
    try {
      SessionAudioRecorder.appendUserPcm(k, bytes);
    } catch (Exception ex) {
      log.warn("appendUserPcm failed: {}", ex.getMessage());
    }

    RealtimeModelBridge bridge = BRIDGES.get(ChannelContextUtils.key(channelContext));
    if (bridge != null) {
      bridge.sendPcm16k(bytes);
    }
    return null;
  }

  @Override
  public Object onText(WebSocketRequest wsRequest, String text, ChannelContext channelContext) throws Exception {
    WebSocketSessionContext wsSessionContext = (WebSocketSessionContext) channelContext.get();
    String path = wsSessionContext.getHandshakeRequest().getRequestLine().path;
    log.info("路径：{}，收到消息：{}", path, text);

    String t = text == null ? "" : text.trim();

    // 先尝试解析为 JSON -> WsMessage
    WsVoiceAgentRequestMessage msg = null;
    try {
      msg = JsonUtils.parse(t, WsVoiceAgentRequestMessage.class);
    } catch (Exception je) {
      // 解析失败：降级为普通文本处理
      log.debug("收到非 JSON 文本或无法解析为 WsMessage", je.getMessage());
      return null;
    } catch (Throwable e) {
      log.error("解析收到的消息异常", e);
      return null;
    }
    RealtimeModelBridge bridge = BRIDGES.get(ChannelContextUtils.key(channelContext));

    if (bridge == null && msg != null && msg.getType() != null) {
      String typeStr = msg.getType().trim().toUpperCase();
      WsVoiceAgentType typeEnum = null;
      try {
        typeEnum = WsVoiceAgentType.valueOf(typeStr);
      } catch (Exception ex) {
        // 未识别的 type，降级处理
        log.debug("未知的 type: {}", typeStr);
      }
      switch (typeEnum) {
      case SETUP:
        String platform = msg.getPlatform();
        String systemPrompt = msg.getSystem_prompt();
        String user_prompt = msg.getUser_prompt();
        String job_description = msg.getJob_description();
        String resume = msg.getResume();
        String questions = msg.getQuestions();
        String greeting = msg.getGreeting();

        RealtimeSetup realtimeSetup = new RealtimeSetup(systemPrompt, user_prompt, job_description, resume, questions,
            greeting);

        connectLLM(channelContext, platform, realtimeSetup);
        // 回显确认
        String json = toJson(new WsVoiceAgentResponseMessage(WsVoiceAgentType.SETUP_RECEIVED.name()));
        Tio.send(channelContext, WebSocketResponse.fromText(json, VoiceAgentConst.CHARSET));
        break;
      default:
        break;
      }
      return null;
    }

    if (bridge == null) {
      String respJson = toJson(new WsVoiceAgentResponseMessage(WsVoiceAgentType.ERROR.name(), "no bridge"));
      Tio.send(channelContext, WebSocketResponse.fromText(respJson, VoiceAgentConst.CHARSET));
      return null;
    }

    try {
      if (msg != null && msg.getType() != null) {
        String typeStr = msg.getType().trim().toUpperCase();
        WsVoiceAgentType typeEnum = null;
        try {
          typeEnum = WsVoiceAgentType.valueOf(typeStr);
        } catch (Exception ex) {
          // 未识别的 type，降级处理
          log.debug("未知的 type: {}", typeStr);
        }

        if (typeEnum != null) {
          switch (typeEnum) {
          case AUDIO_END:
            bridge.endAudioInput();
            break;

          case TEXT:
            String userText = msg.getText() == null ? "" : msg.getText();
            bridge.sendText(userText);
            break;

          case CLOSE:
            bridge.close();
            Tio.remove(channelContext, "client requested close");
            break;

          default:
            // 其它类型：回显原始 JSON
            Tio.send(channelContext, WebSocketResponse.fromText(
                toJson(new WsVoiceAgentResponseMessage(WsVoiceAgentType.IGNORED.name(), t)), VoiceAgentConst.CHARSET));
            break;
          }
        }
      }
    } catch (Exception e) {
      log.error(e.getMessage(), e);
    }
    return null;
  }

  private String toJson(WsVoiceAgentResponseMessage wsVoiceAgentResponseMessage) {
    return JsonUtils.toSkipNullJson(wsVoiceAgentResponseMessage);
  }

  private void connectLLM(ChannelContext channelContext, String platform, RealtimeSetup setup) {
    String k = ChannelContextUtils.key(channelContext);

    // 启动 recorder（用户是 16k，模型默认 24k）
    try {
      SessionAudioRecorder.start(k, 16000, 24000);
    } catch (Exception e) {
      log.warn("start recorder failed: {}", e.getMessage());
    }

    RealtimeBridgeCallback callback = new WsRealtimeBridgeCallback(channelContext);
    callback.start(setup);
    RealtimeModelBridge bridge = RealtimeModelBridgeFactory.createBridge(platform, callback);
    BRIDGES.put(k, bridge);
    // 连接 Gemini Live（异步）
    bridge.connect(setup);
  }

}

conversation.updateSession(OmniRealtimeConfig.builder()
    .modalities(Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT))
    .voice("Cherry")
    .enableTurnDetection(true)
    .enableInputAudioTranscription(true)
    // 关键：指定输入转写 ASR 模型（仅支持 gummy-realtime-v1）
    .inputAudioTranscription("gummy-realtime-v1")   // 如果的 SDK builder 有这个方法
    // 或者：.inputAudioTranscriptionModel("gummy-realtime-v1")
    .parameters(Map.of(
        "instructions", "是…",
        "smooth_output", true
    ))
    .build()
);

case "conversation.item.input_audio_transcription.completed":
    System.out.println("用户: " + event.get("transcript").getAsString());
    break;