整合Gemini realtime模型

    <dependency>
      <groupId>com.google.genai</groupId>
      <artifactId>google-genai</artifactId>
      <version>${google-genai.version}</version>
    </dependency>

package nexus.io.voice.agent.bridge;

public interface RealtimeBridgeCallback {
  void sendText(String json);

  void sendBinary(byte[] bytes);

  void close(String reason);

  void session(String sessionId);

  void turnComplete(String role, String text);

  void start(RealtimeSetup setup);
}

package nexus.io.voice.agent.bridge;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;

@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain=true)
public class RealtimeSetup {
  private String system_prompt; // 当 type == "setup" 时的系统提示
  private String user_prompt; // 当 type == "setup" 时的用户提示
  private String job_description;
  private String resume;
  private String questions;
  private String greeting;
}

package nexus.io.voice.agent.bridge;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

import com.google.genai.AsyncSession;
import com.google.genai.Client;
import com.google.genai.types.ActivityHandling;
import com.google.genai.types.AudioTranscriptionConfig;
import com.google.genai.types.AutomaticActivityDetection;
import com.google.genai.types.Blob;
import com.google.genai.types.ClientOptions;
import com.google.genai.types.Content;
import com.google.genai.types.EndSensitivity;
import com.google.genai.types.LiveConnectConfig;
import com.google.genai.types.LiveSendClientContentParameters;
import com.google.genai.types.LiveSendRealtimeInputParameters;
import com.google.genai.types.LiveServerContent;
import com.google.genai.types.LiveServerMessage;
import com.google.genai.types.Modality;
import com.google.genai.types.Part;
import com.google.genai.types.PrebuiltVoiceConfig;
import com.google.genai.types.RealtimeInputConfig;
import com.google.genai.types.SpeechConfig;
import com.google.genai.types.StartSensitivity;
import com.google.genai.types.ThinkingConfig;
import com.google.genai.types.TurnCoverage;
import com.google.genai.types.VoiceConfig;
import nexus.io.voice.agent.model.WsVoiceAgentResponseMessage;
import nexus.io.gemini.GeminiClient;
import nexus.io.tio.utils.hutool.StrUtil;
import nexus.io.tio.utils.json.JsonUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class GeminiLiveBridge {

  private static final String MODEL = "models/gemini-2.5-flash-native-audio-preview-12-2025";

  // 输入输出音频 mime
  private static final String INPUT_MIME = "audio/pcm;rate=16000";
  private static final String OUTPUT_MIME_PREFIX = "audio/pcm"; // 输出是 audio/pcm（24k）

  private final Object transcriptLock = new Object();
  private final StringBuilder turnUserTranscript = new StringBuilder();
  private final StringBuilder turnAssistantTranscript = new StringBuilder();

  private final Client client;
  private volatile AsyncSession session;
  private final RealtimeBridgeCallback callback;

  public GeminiLiveBridge(RealtimeBridgeCallback sender) {
    this.callback = sender;

    Client.Builder b = Client.builder().apiKey(GeminiClient.GEMINI_API_KEY);

    ClientOptions clientOptions = ClientOptions.builder().build();

    b.clientOptions(clientOptions);

    this.client = b.build();
  }

  public CompletableFuture<Void> connect(RealtimeSetup realtimeSetup) {
    LiveConnectConfig config = buildLiveConfig();

    // AsyncLive.connect(model, config) -> CompletableFuture<AsyncSession>
    return client.async.live.connect(MODEL, config).thenCompose(sess -> {
      this.session = sess;
      String sessionId = sess.sessionId();
      callback.session(sessionId);
      send(new WsVoiceAgentResponseMessage("gemini_connected", sessionId));

      // 如果连接建立时已存在 prompts（可能前端先发 setup），立即发送到模型
      try {
        sendPromptsIfAny(sess, realtimeSetup);
      } catch (Exception ex) {
        log.error("send setup prompts error (connect)", ex);
        send(new WsVoiceAgentResponseMessage("error", safe(ex.getMessage())));
      }

      // 注册 receive 回调（只注册一次）
      CompletableFuture<Void> receiveFuture = sess.receive(this::onGeminiMessage);
      receiveFuture.whenComplete((v, ex) -> {
        log.info("v:{},ex:{}", v, ex);
        if (ex != null) {

        }
      });
      return receiveFuture;
    }).exceptionally(ex -> {
      log.error("Gemini live connect failed", ex);
      String safe = safe(ex.getMessage());
      send(new WsVoiceAgentResponseMessage("error", safe));
      callback.close("gemini connect failed");
      return null;
    });
  }

  public CompletableFuture<Void> close() {
    try {
      AsyncSession s = this.session;
      if (s != null) {
        return s.close().exceptionally(ex -> null);
      }
    } finally {
      try {
        client.close();
        callback.close("close");
      } catch (Exception ignore) {
      }
    }
    return CompletableFuture.completedFuture(null);
  }

  /** 前端推来的 16k PCM 裸流 */
  public CompletableFuture<Void> sendPcm16k(byte[] pcm16k) {
    AsyncSession s = this.session;
    if (s == null)
      return CompletableFuture.completedFuture(null);

    Blob audioBlob = Blob.builder().mimeType(INPUT_MIME).data(pcm16k).build();

    LiveSendRealtimeInputParameters params = LiveSendRealtimeInputParameters.builder().audio(audioBlob).build();

    return s.sendRealtimeInput(params).exceptionally(ex -> {
      String message = ex.getMessage();
      log.error(message);
      send(new WsVoiceAgentResponseMessage("error", safe(message)));
      if ("org.java_websocket.exceptions.WebsocketNotConnectedException".equals(message)) {
        close();
      }
      return null;
    });
  }

  /** 前端说“音频结束”（可选，用于提示流结束） */
  public CompletableFuture<Void> sendAudioStreamEnd() {
    AsyncSession s = this.session;
    if (s == null) {
      return CompletableFuture.completedFuture(null);
    }

    LiveSendRealtimeInputParameters params = LiveSendRealtimeInputParameters.builder().audioStreamEnd(true).build();

    return s.sendRealtimeInput(params).exceptionally(ex -> {
      String message = ex.getMessage();
      log.error(message);
      send(new WsVoiceAgentResponseMessage("error", safe(message)));
      if ("org.java_websocket.exceptions.WebsocketNotConnectedException".equals(message)) {
        close();
      }
      return null;
    });
  }

  /** 前端发文本输入（可选） */
  public CompletableFuture<Void> sendText(String text) {
    AsyncSession s = this.session;
    if (s == null) {
      return CompletableFuture.completedFuture(null);
    }

    Content userMessage = Content.fromParts(Part.fromText(text));

    LiveSendClientContentParameters cc = LiveSendClientContentParameters.builder().turns(List.of(userMessage))
        .turnComplete(true).build();

    return s.sendClientContent(cc).exceptionally(ex -> {
      log.error(ex.getMessage());
      send(new WsVoiceAgentResponseMessage("error", safe(ex.getMessage())));
      return null;
    });
  }

  private void sendPromptsIfAny(AsyncSession s, RealtimeSetup realtimeSetup) {
    if (realtimeSetup == null) {
      return;
    }
    String systemPrompt = realtimeSetup.getSystem_prompt();
    String job_description = realtimeSetup.getJob_description();
    String resume = realtimeSetup.getResume();
    String questions = realtimeSetup.getQuestions();
    String greeting = realtimeSetup.getGreeting();

    List<Content> initialTurns = new ArrayList<>();
    if (StrUtil.notBlank(systemPrompt)) {
      initialTurns.add(Content.fromParts(Part.fromText(systemPrompt)));
    }

    if (StrUtil.notBlank(job_description)) {
      initialTurns.add(Content.fromParts(Part.fromText(job_description)));
    }

    if (StrUtil.notBlank(resume)) {
      initialTurns.add(Content.fromParts(Part.fromText(resume)));
    }

    if (StrUtil.notBlank(questions) || StrUtil.notBlank(greeting)) {
      initialTurns.add(Content.fromParts(Part.fromText(greeting + "\n\n" + questions)));
    }

    if (!initialTurns.isEmpty()) {
      // 初始化指令通常作为 context 而非单次完成的用户 turn，可按需调整 turnComplete
      LiveSendClientContentParameters cc = LiveSendClientContentParameters.builder().turns(initialTurns)
          .turnComplete(true).build();

      s.sendClientContent(cc).exceptionally(ex -> {
        log.error(ex.getMessage());
        send(new WsVoiceAgentResponseMessage("error", safe(ex.getMessage())));
        return null;
      });

      send(new WsVoiceAgentResponseMessage("setup_sent_to_model"));
    }
  }

  private LiveConnectConfig buildLiveConfig() {
    // 自动VAD配置：AutomaticActivityDetection/RealtimeInputConfig
    AutomaticActivityDetection vad = AutomaticActivityDetection.builder().disabled(false)
        .startOfSpeechSensitivity(StartSensitivity.Known.START_SENSITIVITY_HIGH)
        .endOfSpeechSensitivity(EndSensitivity.Known.END_SENSITIVITY_LOW).prefixPaddingMs(100).silenceDurationMs(500)
        .build();

    RealtimeInputConfig realtimeInput = RealtimeInputConfig.builder().automaticActivityDetection(vad)
        .activityHandling(ActivityHandling.Known.START_OF_ACTIVITY_INTERRUPTS)
        .turnCoverage(TurnCoverage.Known.TURN_INCLUDES_ONLY_ACTIVITY).build();

    PrebuiltVoiceConfig prebuiltVoiceConfig = PrebuiltVoiceConfig.builder().voiceName("Puck").build();
    VoiceConfig voiceConfig = VoiceConfig.builder().prebuiltVoiceConfig(prebuiltVoiceConfig).build();
    SpeechConfig speech = SpeechConfig.builder().voiceConfig(voiceConfig).build();

    ThinkingConfig thinkingConfig = ThinkingConfig.builder().thinkingBudget(0).build();
    AudioTranscriptionConfig inputAudioTranscription = AudioTranscriptionConfig.builder().build();
    LiveConnectConfig liveConnectConfig = LiveConnectConfig.builder()
        .responseModalities(List.of(new Modality(Modality.Known.AUDIO)))
        //
        .speechConfig(speech).thinkingConfig(thinkingConfig).realtimeInputConfig(realtimeInput)
        // 可选：转写
        .inputAudioTranscription(inputAudioTranscription).outputAudioTranscription(inputAudioTranscription)
        //
        .build();

    return liveConnectConfig;
  }

  private void onGeminiMessage(LiveServerMessage msg) {
    try {
      if (msg.setupComplete().isPresent()) {
        send(new WsVoiceAgentResponseMessage("setup_complete"));
      }

      Optional<LiveServerContent> serverContentOpt = msg.serverContent();
      if (serverContentOpt.isPresent()) {
        LiveServerContent sc = serverContentOpt.get();

        // 输入转写
        sc.inputTranscription().ifPresent(t -> {
          Optional<String> optional = t.text();
          String text = optional.orElse(null);
          send(new WsVoiceAgentResponseMessage("transcript_in", text));
          appendTurnTranscript("user", text);
        });

        // 输出转写
        sc.outputTranscription().ifPresent(t -> {
          Optional<String> optional = t.text();
          String text = optional.orElse(null);
          send(new WsVoiceAgentResponseMessage("transcript_out", text));
          appendTurnTranscript("model", text);
        });

        // 模型输出（音频/文本 part）
        sc.modelTurn().ifPresent(content -> {
          content.parts().ifPresent(parts -> {
            for (Part p : parts) {
              // 文本
              p.text().ifPresent(txt -> send(new WsVoiceAgentResponseMessage("text", txt)));

              // 音频（inlineData）
              p.inlineData().ifPresent(blob -> {
                String mt = blob.mimeType().orElse("");
                byte[] data = blob.data().orElse(null);
                if (data != null && mt.startsWith(OUTPUT_MIME_PREFIX)) {
                  // 直接二进制回传（建议前端按 24k PCM 播放）
                  callback.sendBinary(data);
                } else if (data != null) {
                  // 兜底：非音频inlineData，走base64文本
                  String b64 = Base64.getEncoder().encodeToString(data);
                  WsVoiceAgentResponseMessage m = new WsVoiceAgentResponseMessage("inline_data");
                  m.setText(b64);
                  send(m);
                }
              });

              // functionCall 等后续再扩展
              p.functionCall().ifPresent(fc -> {
                WsVoiceAgentResponseMessage m = new WsVoiceAgentResponseMessage("function_call");
                m.setText(fc.name().orElse(""));
                send(m);
              });
            }
          });
        });

        // turnComplete
        if (sc.turnComplete().orElse(false)) {
          send(new WsVoiceAgentResponseMessage("turn_complete"));
          flushTurnTranscriptOnComplete();
        }
      }

      // goAway（服务端提示将断开）
      msg.goAway().ifPresent(g -> {
        String timeLeft = g.timeLeft().map(Duration::toString) // 例如 PT30S / PT1M
            .orElse("");
        WsVoiceAgentResponseMessage m = new WsVoiceAgentResponseMessage("go_away");
        m.setText(timeLeft);
        send(m);
      });

      // usage
      msg.usageMetadata().ifPresent(u -> {
        WsVoiceAgentResponseMessage m = new WsVoiceAgentResponseMessage("usage");
        // 将 usage 信息编码到 text 字段或扩展 DTO；此处把三个数字拼接（可按需改成专门字段）
        Optional<Integer> promptTokenCount = u.promptTokenCount();
        Optional<Integer> responseTokenCount = u.responseTokenCount();
        Optional<Integer> totalTokenCount = u.totalTokenCount();
        m.setPromptTokenCount(promptTokenCount);
        m.setResponseTokenCount(responseTokenCount);
        m.setTotalTokenCount(totalTokenCount);
        send(m);
      });

    } catch (Exception e) {
      log.error("onGeminiMessage error", e);
      send(new WsVoiceAgentResponseMessage("error", safe(e.getMessage())));
    }
  }

  private void send(WsVoiceAgentResponseMessage msg) {
    try {
      String json = JsonUtils.toSkipNullJson(msg);
      callback.sendText(json);
    } catch (Exception e) {
      log.error("serialize message error", e);
      // fallback minimal message
//      WsVoiceAgentResponseMessage wsVoiceAgentResponseMessage = new WsVoiceAgentResponseMessage();
//      wsVoiceAgentResponseMessage.setType(WsVoiceAgentType.ERROR.name());
//      wsVoiceAgentResponseMessage.setMessage("serialize error");
      // 手写性能高
      callback.sendText("{\"type\":\"error\",\"message\":\"serialize error\"}");
    }
  }

  private static String safe(String s) {
    if (s == null) {
      return "";
    }
    return s.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n").replace("\r", "");
  }

  private void appendTurnTranscript(String role, String text) {
    if (text == null || text.isEmpty())
      return;
    synchronized (transcriptLock) {
      StringBuilder sb = "user".equals(role) ? turnUserTranscript : turnAssistantTranscript;
      if (sb.length() > 0)
        sb.append(' ');
      sb.append(text);
    }
  }

  private void flushTurnTranscriptOnComplete() {
    String userText;
    String assistantText;
    synchronized (transcriptLock) {
      userText = turnUserTranscript.toString().trim();
      assistantText = turnAssistantTranscript.toString().trim();
      turnUserTranscript.setLength(0);
      turnAssistantTranscript.setLength(0);
    }

    // 一次 turn complete，分别按角色回调（各一次；如果为空则不回调）
    try {
      if (!userText.isEmpty()) {
        callback.turnComplete("user", userText);
      }
      if (!assistantText.isEmpty()) {
        callback.turnComplete("assistant", assistantText);
      }
    } catch (Exception e) {
      log.error("turnComplete callback error", e);
    }
  }
}

package nexus.io.voice.agent.callback;

import java.nio.file.Path;

import nexus.io.voice.agent.utils.ChannelContextUtils;
import nexus.io.jfinal.aop.Aop;
import nexus.io.media.NativeMedia;
import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.tio.utils.hutool.StrUtil;
import nexus.io.tio.utils.snowflake.SnowflakeIdUtils;
import nexus.io.tio.websocket.common.WebSocketResponse;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class WsRealtimeBridgeCallback implements RealtimeBridgeCallback {

  private ChannelContext channelContext;
  private String sessionId;

  public WsRealtimeBridgeCallback(ChannelContext channelContext) {
    this.channelContext = channelContext;
    this.sessionId = ChannelContextUtils.key(channelContext);
  }

  @Override
  public void sendText(String json) {
    WebSocketResponse wsResp = WebSocketResponse.fromText(json, InterviewConst.CHARSET);
    Tio.send(channelContext, wsResp);
  }

  @Override
  public void sendBinary(byte[] bytes) {
    WebSocketResponse wsResp = WebSocketResponse.fromBytes(bytes);
    Tio.send(channelContext, wsResp);
  }

  @Override
  public void close(String reason) {
    Tio.remove(channelContext, reason);
  }

  @Override
  public void session(String sessionId) {
  }

  @Override
  public void turnComplete(String role, String text) {
    
    // log.info("role:{},text:{}", role, text);
  }
}

package nexus.io.voice.agent.utils;

import nexus.io.tio.core.ChannelContext;

public class ChannelContextUtils {

  public static String key(ChannelContext ctx) {
    // 用 tio 自己的唯一标识
    return ctx.getId();
  }

}

package nexus.io.voice.agent.consts;

public interface InterviewConst {
  
  String CHARSET = "utf-8";

}

package nexus.io.voice.agent.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * WebSocket 前端发来的消息结构
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class WsVoiceAgentRequestMessage {
  private String type; // 消息类型：setup | text | audio_end | close | ...
  private String text; // 当 type == "text" 时的文本
  private String system_prompt; // 当 type == "setup" 时的系统提示
  private String user_prompt; // 当 type == "setup" 时的用户提示
  private String job_description;
  private String resume;
  private String questions;
  private String greeting;
}

package nexus.io.voice.agent.model;

import java.util.Optional;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 后端 -> 前端 统一消息体
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class WsVoiceAgentResponseMessage {

  private String type;

  // 通用字段
  private String text;
  private String message;
  private String where;

  // setup / session
  private String sessionId;

  // goAway
  private String timeLeft;

  // usage
  private Integer promptTokenCount;
  private Integer responseTokenCount;
  private Integer totalTokenCount;

  // function call
  private String name;

  public WsVoiceAgentResponseMessage(String type) {
    this.type = type;
  }

  public WsVoiceAgentResponseMessage(String type, String text) {
    this.type = type;
    this.text = text;
  }

  public WsVoiceAgentResponseMessage(String type, Optional<String> optional) {
    this.type = type;
    this.text = optional.orElse(null);
  }

  public void setPromptTokenCount(Optional<Integer> promptTokenCount) {
    this.promptTokenCount = promptTokenCount != null ? promptTokenCount.orElse(null) : null;
  }

  public void setResponseTokenCount(Optional<Integer> responseTokenCount) {
    this.responseTokenCount = responseTokenCount != null ? responseTokenCount.orElse(null) : null;
  }

  public void setTotalTokenCount(Optional<Integer> totalTokenCount) {
    this.totalTokenCount = totalTokenCount != null ? totalTokenCount.orElse(null) : null;
  }
}

package nexus.io.voice.agent.model;

/**
 * WebSocket 消息类型枚举
 */
public enum WsVoiceAgentType {
  SETUP,
  //
  TEXT,
  //
  AUDIO_END,
  //
  CLOSE,

  // server -> client 响应类型（也可复用同一枚举）
  ERROR,
  //
  SETUP_RECEIVED,
  //
  SETUP_SENT_TO_MODEL,
  //
  GEMINI_CONNECTED,
  //
  SETUP_COMPLETE,
  //
  TRANSCRIPT_IN,
  //
  TRANSCRIPT_OUT,
  //
  TURN_COMPLETE,
  //
  GO_AWAY,
  //
  USAGE,
  //
  INLINE_DATA,
  //
  FUNCTION_CALL,
  //
  IGNORED
}

package nexus.io.voice.agent.handler;

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

import nexus.io.voice.agent.audio.SessionAudioRecorder;
import nexus.io.voice.agent.bridge.GeminiLiveBridge;
import nexus.io.voice.agent.bridge.RealtimeBridgeCallback;
import nexus.io.voice.agent.bridge.RealtimeSetup;
import nexus.io.voice.agent.callback.WsRealtimeBridgeCallback;
import nexus.io.voice.agent.consts.InterviewConst;
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 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 lombok.extern.slf4j.Slf4j;

@Slf4j
public class VoiceSocketHandler implements IWebSocketHandler {
  // 一个前端连接一个 bridge
  private static final Map<String, GeminiLiveBridge> 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);
    GeminiLiveBridge 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());
    }

    GeminiLiveBridge 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;
    }
    GeminiLiveBridge 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 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, realtimeSetup);
        // 回显确认
        String json = toJson(new WsVoiceAgentResponseMessage(WsVoiceAgentType.SETUP_RECEIVED.name()));
        Tio.send(channelContext, WebSocketResponse.fromText(json, InterviewConst.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, InterviewConst.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.sendAudioStreamEnd();
            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)), InterviewConst.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, RealtimeSetup setup) {
    String k = ChannelContextUtils.key(channelContext);

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

    GeminiLiveBridge bridge = new GeminiLiveBridge(callback);
    BRIDGES.put(k, bridge);
    // 连接 Gemini Live（异步）
    bridge.connect(setup);
  }
}