在搭建 AI 会话 API 时,选择 WebSocket 还是 HTTP + Stream 返回

// /server/index.ts
import { Hono } from 'hono';
import { serve } from '@hono/node-server';
import { WebSocketServer } from 'ws';

const app = new Hono();
const wss = new WebSocketServer({ port: 3001 });

const sessions = new Map<string, { isPaused: boolean }>();

wss.on('connection', (ws) => {
  const sessionId = crypto.randomUUID();
  sessions.set(sessionId, { isPaused: false });

  ws.on('message', async (message) => {
    const { action, data } = JSON.parse(message.toString());

    if (action === 'start') {
      const { prompt } = data;
      const stream = await fetchOpenAIStream(prompt, sessionId);
      ws.send(JSON.stringify({ action: 'stream', data: stream }));
    }

    if (action === 'pause') {
      sessions.get(sessionId)!.isPaused = true;
    }

    if (action === 'resume') {
      sessions.get(sessionId)!.isPaused = false;
    }
  });

  ws.on('close', () => {
    sessions.delete(sessionId);
  });
});

async function fetchOpenAIStream(prompt: string, sessionId: string) {
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages: [{ role: 'user', content: prompt }],
      stream: true
    })
  });

  const reader = response.body?.getReader();
  const decoder = new TextDecoder();

  let result = '';

  while (true) {
    const { done, value } = await reader!.read();
    if (done) break;

    const isPaused = sessions.get(sessionId)?.isPaused;
    if (isPaused) continue;

    result += decoder.decode(value);
    wss.clients.forEach((client) => {
      if (client.readyState === 1) {
        client.send(JSON.stringify({ action: 'update', data: result }));
      }
    });
  }

  return result;
}

// 处理 LLM 文本流 + TTS 语音合成
async function processLLMWithTTS(prompt: string, sessionId: string, ws: WebSocket) {
  const llmResponse = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages: [{ role: 'user', content: prompt }],
      stream: true
    })
  });

  const reader = llmResponse.body?.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader!.read();
    if (done) break;

    const session = sessions.get(sessionId);
    if (session?.isStopped) break;    // 💥 立即终止
    if (session?.isPaused) continue;  // ⏸️ 暂停

    const textChunk = decoder.decode(value);
    const audioChunk = await synthesizeTTS(textChunk);

    ws.send(JSON.stringify({ action: 'audio', data: audioChunk }));
  }
}

// 模拟 TTS 音频合成 (使用真实 API 替换)
async function synthesizeTTS(text: string): Promise<string> {
  const response = await fetch('https://api.tts-provider.com/v1/stream', {
    method: 'POST',
    headers: { 'Authorization': `Bearer ${process.env.TTS_API_KEY}` },
    body: JSON.stringify({ text })
  });

  const audioData = await response.arrayBuffer();
  return Buffer.from(audioData).toString('base64'); // 以 Base64 格式返回
}

serve({ fetch: app.fetch, port: 3000 });

// /client/hooks/useWebSocket.ts

import { useEffect, useRef, useState } from 'react';

export const useWebSocket = (url: string) => {
  const ws = useRef<WebSocket | null>(null);
  const [messages, setMessages] = useState<string[]>([]);
  const [isPaused, setIsPaused] = useState(false);

  useEffect(() => {
    ws.current = new WebSocket(url);

    ws.current.onmessage = (event) => {
      const { action, data } = JSON.parse(event.data);
      if (action === 'update') {
        setMessages((prev) => [...prev, data]);
      }
    };

    ws.current.onclose = () => console.warn('WebSocket closed');
    ws.current.onerror = (err) => console.error('WebSocket error', err);

    return () => ws.current?.close();
  }, [url]);

  const sendMessage = (action: string, data: any) => {
    ws.current?.send(JSON.stringify({ action, data }));
  };

  const start = (prompt: string) => sendMessage('start', { prompt });
  const pause = () => {
    sendMessage('pause', {});
    setIsPaused(true);
  };
  const resume = () => {
    sendMessage('resume', {});
    setIsPaused(false);
  };

  return { messages, start, pause, resume, isPaused };
};

useEffect(() => {
  const connect = () => {
    ws.current = new WebSocket(url);

    ws.current.onopen = () => console.log('WebSocket connected');
    ws.current.onclose = () => {
      console.warn('WebSocket closed, reconnecting in 3s...');
      setTimeout(connect, 3000); // 3s 重连机制
    };
    ws.current.onerror = (err) => console.error('WebSocket error', err);
  };

  connect();

  return () => ws.current?.close();
}, [url]);