WebSocket 通信統合ガイド

通信アーキテクチャの概要

フロントエンドは WebSocket を通じてバックエンドと永続的な接続を確立し、双方向のリアルタイム通信を実現します。システムはメッセージタイプに基づくルーティングの仕組みを採用し、ストリーミングデータ転送、状態の同期、エラーからの復旧に対応しています。

接続管理

WebSocket 接続の初期化

const getWebSocketUrl = useCallback(() => {
  // 1. 优先使用传入的URL
  if (websocketUrl) return websocketUrl

  // 2. 使用环境变量
  const envWsUrl = process.env.NEXT_PUBLIC_WEBSOCKET_URL
  if (envWsUrl) return envWsUrl

  // 3. 动态构建URL
  if (typeof window !== "undefined") {
    const protocol = window.location.protocol === "https:" ? "wss:" : "ws:"
    return `${protocol}//${window.location.hostname}:8000/ws/chat`
  }

  return "ws://localhost:8000/ws/chat"
}, [websocketUrl])

接続状態の管理

type ConnectionState = 'disconnected' | 'connecting' | 'connected' | 'reconnecting' | 'error'

interface ConnectionInfo {
  state: ConnectionState
  url: string
  error: string | null
  reconnectAttempts: number
  socket: WebSocket | null
}

自動再接続の仕組み

const scheduleReconnect = useCallback(() => {
  if (state.connection.reconnectAttempts >= maxReconnectAttempts) {
    dispatch({
      type: 'CONNECTION_ERROR',
      payload: { error: "已达到最大重连次数" }
    })
    return
  }

  const attempts = state.connection.reconnectAttempts + 1
  const delay = baseReconnectDelay * Math.pow(2, attempts - 1) // 指数退避算法

  dispatch({ type: 'CONNECTION_RECONNECTING', payload: { attempts } })

  reconnectTimeoutRef.current = setTimeout(() => {
    connectWebSocket()
  }, delay)
}, [state.connection.reconnectAttempts, connectWebSocket])

メッセージプロトコル仕様

基本的なメッセージ形式

interface WebSocketMessage {
  type: string                    // 消息类型标识
  payload: Record<string, any>    // 消息载荷
  timestamp?: number              // 可选时间戳
}

メッセージタイプの分類

クライアントからのメッセージ送信

1. ユーザーのテキスト入力

const sendUserInput = useCallback((text?: string) => {
  const inputText = text || state.conversation.currentInput
  if (!inputText.trim()) return

  const success = sendMessage({
    type: 'USER_TEXT_INPUT',
    payload: { text: inputText },
  })

  if (success) {
    dispatch({ type: 'INPUT_CLEAR' })
    dispatch({ type: 'TURN_END' }) // 清除当前轮次，等待助手响应
  }
}, [state.conversation.currentInput, sendMessage])

2. 現在の対話の中断

const interruptTurn = useCallback(() => {
  dispatch({ type: 'TURN_INTERRUPT' })

  // 发送中断消息到服务器
  sendMessage({
    type: 'INTERRUPT_TURN',
    payload: {},
  })
}, [sendMessage])

3. 音声認識の切り替え

const setListening = useCallback((isListening: boolean) => {
  dispatch({ type: 'UI_SET_LISTENING', payload: { isListening } })

  // 发送ASR监听切换消息
  sendMessage({
    type: 'TOGGLE_ASR_LISTENING',
    payload: { enabled: isListening },
  })
}, [sendMessage])

サーバー応答の処理

1. LLM ストリーミング応答の処理

case 'LLM_CHUNK':
  const { content_delta, is_final } = message.payload
  const currentState = stateRef.current

  // 确保有当前轮次的助手消息
  let currentAssistantMessageId = currentState.conversation.currentTurnId

  if (!currentAssistantMessageId) {
    // 创建新的助手消息
    currentAssistantMessageId = Date.now().toString() + '_assistant'
    dispatch({ type: 'TURN_START', payload: { turnId: currentAssistantMessageId } })

    const assistantMessage: Message = {
      id: currentAssistantMessageId,
      role: 'assistant',
      content: content_delta || '',
      timestamp: Date.now(),
    }
    dispatch({ type: 'MESSAGE_ADD', payload: { message: assistantMessage } })
  } else {
    // 🔒 智能保护机制：在有工具调用的turn中启用静默模式
    const hasToolCallsInTurn = currentState.conversation.hasToolCallsInCurrentTurn

    if (hasToolCallsInTurn) {
      return // 静默模式：等待LLM_FINAL_RESPONSE统一更新
    }

    // 正常的流式追加
    const currentMessage = currentState.conversation.messages.find(m => m.id === currentAssistantMessageId)
    if (currentMessage) {
      const newContent = (currentMessage.content || '') + (content_delta || '')

      dispatch({
        type: 'MESSAGE_UPDATE',
        payload: {
          id: currentAssistantMessageId,
          updates: { content: newContent }
        }
      })
    }
  }

  // 更新流式状态
  if (is_final) {
    dispatch({ type: 'STREAM_END', payload: {} })
  }
  break

2. LLM 最終応答の処理

case 'LLM_FINAL_RESPONSE':
  // 🛡️ 事件去重：防止重复处理相同的LLM_FINAL_RESPONSE
  const responseContent = message.payload.content || ''
  const currentTurnId = stateRef.current.conversation.currentTurnId
  const responseKey = `${currentTurnId}_${responseContent.length}_${responseContent.slice(-20)}`

  if (!window.processedLLMResponses) {
    window.processedLLMResponses = new Set()
  }

  if (window.processedLLMResponses.has(responseKey)) {
    return // 跳过重复的LLM_FINAL_RESPONSE事件
  }

  window.processedLLMResponses.add(responseKey)

  // 智能内容更新逻辑
  const finalCurrentState = stateRef.current
  let finalAssistantMessageId = finalCurrentState.conversation.currentTurnId

  if (finalAssistantMessageId) {
    const currentMessage = finalCurrentState.conversation.messages.find(m => m.id === finalAssistantMessageId)
    const existingContent = currentMessage?.content || ''
    const newContent = message.payload.content || ''
    const hasToolCallsInTurn = finalCurrentState.conversation.hasToolCallsInCurrentTurn

    let finalContent = newContent

    // 🔒 智能保护：如果当前turn有工具调用，智能累积内容
    if (hasToolCallsInTurn && existingContent && newContent) {
      if (existingContent.includes(newContent)) {
        finalContent = existingContent // 防止重复
      } else {
        finalContent = `${existingContent}\n\n${newContent}` // 智能累积
      }
    }

    dispatch({
      type: 'MESSAGE_UPDATE',
      payload: {
        id: finalAssistantMessageId,
        updates: { content: finalContent }
      }
    })
  }

  dispatch({ type: 'STREAM_END', payload: { finalContent: message.payload.content } })
  break

3. ツール呼び出しの処理

case 'TOOL_CALL_START':
  // 解析工具调用参数
  let parsedArgs = {}
  try {
    parsedArgs = message.payload.args_str ? JSON.parse(message.payload.args_str) : {}
  } catch (e) {
    console.error("Failed to parse tool call args:", message.payload.args_str, e)
    parsedArgs = { error: "解析参数失败" }
  }

  const toolCall: ToolCall = {
    name: message.payload.name,
    id: message.payload.id,
    args: parsedArgs,
    status: 'pending',
  }

  dispatch({ type: 'TOOL_CALL_START', payload: { toolCall } })

  // 更新当前助手消息的工具调用列表
  const toolCurrentState = stateRef.current
  let toolAssistantMessageId = toolCurrentState.conversation.currentTurnId

  if (!toolAssistantMessageId) {
    // 创建新的助手消息轮次
    toolAssistantMessageId = Date.now().toString() + '_assistant_with_tools'
    dispatch({ type: 'TURN_START', payload: { turnId: toolAssistantMessageId } })

    const assistantMessage: Message = {
      id: toolAssistantMessageId,
      role: 'assistant',
      content: '',
      timestamp: Date.now(),
      toolCalls: [toolCall]
    }
    dispatch({ type: 'MESSAGE_ADD', payload: { message: assistantMessage } })
  } else {
    // 更新现有助手消息的工具调用
    const currentMessage = toolCurrentState.conversation.messages.find(m => m.id === toolAssistantMessageId)
    const existingToolCalls = currentMessage?.toolCalls || []

    dispatch({
      type: 'MESSAGE_UPDATE',
      payload: {
        id: toolAssistantMessageId,
        updates: {
          toolCalls: [...existingToolCalls, toolCall]
        }
      }
    })
  }
  break

case 'TOOL_CALL_RESULT':
  // 处理工具调用结果（成功或失败）
  if (message.payload.status === 'error') {
    dispatch({
      type: 'TOOL_CALL_ERROR',
      payload: {
        id: message.payload.id,
        error: message.payload.error || '工具调用失败'
      }
    })
  } else {
    dispatch({
      type: 'TOOL_CALL_END',
      payload: {
        id: message.payload.id,
        result: message.payload.result || '执行成功'
      }
    })
  }

  // 更新消息中的工具调用状态
  const resultCurrentState = stateRef.current
  if (resultCurrentState.conversation.currentTurnId) {
    const currentMessage = resultCurrentState.conversation.messages.find(m => m.id === resultCurrentState.conversation.currentTurnId)
    if (currentMessage?.toolCalls) {
      const updatedToolCalls = currentMessage.toolCalls.map(tool =>
        tool.id === message.payload.id ? {
          ...tool,
          result: message.payload.result || (message.payload.status === 'error' ? undefined : '执行成功'),
          error: message.payload.status === 'error' ? (message.payload.error || '工具调用失败') : undefined,
          status: message.payload.status === 'error' ? 'error' : 'success'
        } : tool
      )

      dispatch({
        type: 'MESSAGE_UPDATE',
        payload: {
          id: resultCurrentState.conversation.currentTurnId,
          updates: { toolCalls: updatedToolCalls }
        }
      })
    }
  }
  break

4. 思考チェーンの処理

case 'REASONING_START':
  const currentState2 = stateRef.current
  let currentAssistantMessageId2 = currentState2.conversation.currentTurnId

  // 如果没有当前轮次，创建新的助手消息
  if (!currentAssistantMessageId2) {
    currentAssistantMessageId2 = Date.now().toString() + '_assistant'
    dispatch({ type: 'TURN_START', payload: { turnId: currentAssistantMessageId2 } })

    const assistantMessage: Message = {
      id: currentAssistantMessageId2,
      role: 'assistant',
      content: '',
      timestamp: Date.now(),
      reasoningContent: '',
      isReasoningStreaming: true
    }
    dispatch({ type: 'MESSAGE_ADD', payload: { message: assistantMessage } })
  }

  dispatch({
    type: 'REASONING_START',
    payload: { messageId: currentAssistantMessageId2 }
  })
  break

case 'REASONING_CHUNK':
  const currentState3 = stateRef.current
  const currentAssistantMessageId3 = currentState3.conversation.currentTurnId

  if (currentAssistantMessageId3) {
    dispatch({
      type: 'REASONING_CHUNK',
      payload: {
        messageId: currentAssistantMessageId3,
        content: message.payload.reasoning_delta || ''
      }
    })
  }
  break

case 'REASONING_END':
  const currentState4 = stateRef.current
  const currentAssistantMessageId4 = currentState4.conversation.currentTurnId

  if (currentAssistantMessageId4) {
    dispatch({
      type: 'REASONING_END',
      payload: { messageId: currentAssistantMessageId4 }
    })
  }
  break

5. ASR 結果の処理

case 'ASR_FINAL_RESULT':
  // 只有在语音模式下才处理ASR结果，避免在键盘模式下干扰用户输入
  const asrCurrentState = stateRef.current
  if (asrCurrentState.ui.inputMode === 'voice') {
    dispatch({
      type: 'UI_SET_ASR_RESULT',
      payload: { result: message.payload.text }
    })
  }
  // 键盘模式下忽略ASR结果
  break

状態同期の仕組み

システム状態の更新

case 'SYSTEM_STATUS':
  dispatch({ type: 'SYSTEM_STATUS_UPDATE', payload: message.payload })

  // 特殊处理ASR状态
  if (message.payload.module === 'asr_service') {
    dispatch({
      type: 'UI_SET_LISTENING',
      payload: { isListening: message.payload.status === 'listening' }
    })
  }
  break

case 'TTS_STATUS':
  dispatch({
    type: 'SYSTEM_STATUS_UPDATE',
    payload: {
      module: 'tts_player',
      status: message.payload.status
    }
  })
  break

セッションラウンドの管理

case 'TURN_ENDED':
  dispatch({ type: 'TURN_END' })
  break

case 'TURN_INTERRUPTED':
  dispatch({ type: 'TURN_INTERRUPT' })
  break

️ エラー処理と復旧

接続エラーの処理

const connectWebSocket = useCallback((url?: string) => {
  const wsUrl = url || getWebSocketUrl()

  dispatch({ type: 'CONNECTION_CONNECTING', payload: { url: wsUrl } })

  try {
    const newSocket = new WebSocket(wsUrl)

    newSocket.onopen = () => {
      dispatch({ type: 'CONNECTION_CONNECTED', payload: { socket: newSocket } })
      dispatch({ type: 'CONNECTION_RESET_ATTEMPTS' })
    }

    newSocket.onerror = (error) => {
      console.error("WebSocket error:", error)
      dispatch({
        type: 'CONNECTION_ERROR',
        payload: { error: "WebSocket连接错误" }
      })
    }

    newSocket.onclose = (event) => {
      socketRef.current = null

      if (event.code !== 1000) {
        // 非正常关闭，尝试重连
        scheduleReconnect()
      } else {
        dispatch({ type: 'CONNECTION_DISCONNECTED', payload: {} })
      }
    }

  } catch (error) {
    dispatch({
      type: 'CONNECTION_ERROR',
      payload: { error: "创建WebSocket连接失败" }
    })
  }
}, [getWebSocketUrl])

メッセージ送信失敗の処理

const sendMessage = useCallback((message: WebSocketMessage): boolean => {
  if (!socketRef.current || state.connection.state !== 'connected') {
    dispatch({
      type: 'SYSTEM_ERROR_ADD',
      payload: {
        message: "WebSocket未连接",
        source: "websocket",
        severity: 'warning'
      }
    })
    return false
  }

  try {
    socketRef.current.send(JSON.stringify(message))
    return true
  } catch (error) {
    console.error("Failed to send WebSocket message:", error)
    dispatch({
      type: 'SYSTEM_ERROR_ADD',
      payload: {
        message: "发送消息失败",
        source: "websocket",
        severity: 'error'
      }
    })
    return false
  }
}, [state.connection.state])

メッセージ解析エラーの処理

newSocket.onmessage = (event) => {
  try {
    const message: WebSocketMessage = JSON.parse(event.data)
    dispatch({ type: 'SYSTEM_MESSAGE_RECEIVED', payload: { message } })
    handleWebSocketMessage(message)
  } catch (error) {
    console.error("Failed to parse WebSocket message:", error)
    dispatch({
      type: 'SYSTEM_ERROR_ADD',
      payload: {
        message: "解析WebSocket消息失败",
        source: "websocket",
        severity: 'error'
      }
    })
  }
}

⚡ パフォーマンスの最適化

メッセージ処理の最適化

// 使用useRef避免闭包问题，确保获取最新状态
const stateRef = useRef(state)
useEffect(() => {
  stateRef.current = state
}, [state])

// 消息处理函数使用最新状态
const handleWebSocketMessage = useCallback((message: WebSocketMessage) => {
  const currentState = stateRef.current
  // 处理逻辑...
}, []) // 无依赖，避免重复创建函数

再接続戦略の最適化

// 指数退避算法，避免频繁重连
const baseReconnectDelay = 1000 // 1秒
const maxReconnectAttempts = 5

const delay = baseReconnectDelay * Math.pow(2, attempts - 1)
// attempts=1: 1秒
// attempts=2: 2秒
// attempts=3: 4秒
// attempts=4: 8秒
// attempts=5: 16秒

メモリリークの防止

useEffect(() => {
  return () => {
    // 清理重连定时器
    if (reconnectTimeoutRef.current) {
      clearTimeout(reconnectTimeoutRef.current)
    }

    // 关闭WebSocket连接
    if (socketRef.current) {
      socketRef.current.close()
    }

    // 清理全局去重状态
    if (typeof window !== 'undefined' && window.processedLLMResponses) {
      window.processedLLMResponses.clear()
    }
  }
}, [])

テストとデバッグ

WebSocket 接続のテスト

// 开发模式下的连接状态监控
useEffect(() => {
  if (process.env.NODE_ENV === 'development') {
    console.log('WebSocket状态变更:', {
      state: state.connection.state,
      url: state.connection.url,
      attempts: state.connection.reconnectAttempts,
      error: state.connection.error
    })
  }
}, [state.connection])

メッセージフローのデバッグ

const handleWebSocketMessage = useCallback((message: WebSocketMessage) => {
  // 开发模式下记录所有消息
  if (process.env.NODE_ENV === 'development') {
    console.group(`📩 WebSocket消息: ${message.type}`)
    console.log('载荷:', message.payload)
    console.log('时间戳:', new Date().toISOString())
    console.groupEnd()
  }

  // 消息处理逻辑...
}, [])

WebSocket 通信の統合はフロントエンドシステムの中核となる基盤であり、充実した接続管理、メッセージ処理、エラー復旧の仕組みにより、バックエンドとの安定した通信と良好なユーザー体験を確保します。