From 9596c52ea2f54dc6640b0908626f91ea2babb1fc Mon Sep 17 00:00:00 2001 From: taiyi Date: Thu, 11 Jun 2026 15:05:12 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dtoekn=E6=9D=83=E9=99=90?= =?UTF-8?q?=E5=92=8C=E8=AF=AD=E9=9F=B3=E8=81=8A=E5=A4=A9=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/api/chat_router.py | 35 +- backend/api/models/chat.py | 4 +- backend/api/voice_chat_router.py | 461 ++++++++++++-------- backend/models/database.py | 2 +- backend/scripts/add_chat_log_mode_column.py | 60 +++ backend/services/chat_service.py | 58 ++- backend/services/realtime_voice_service.py | 52 ++- frontend/src/stores/chatStore.js | 193 ++++++-- frontend/src/views/ChatView.vue | 98 +++-- 9 files changed, 655 insertions(+), 308 deletions(-) create mode 100644 backend/scripts/add_chat_log_mode_column.py diff --git a/backend/api/chat_router.py b/backend/api/chat_router.py index 9595756..0247797 100644 --- a/backend/api/chat_router.py +++ b/backend/api/chat_router.py @@ -153,8 +153,8 @@ async def get_chat_history_route( for item in result["data"]: history_items.append(ChatHistoryItem( id=item["id"], - user_msg=item["user_msg"], - ai_msg=item["ai_msg"], + user_msg=item.get("user_msg") or "", + ai_msg=item.get("ai_msg") or "", created_at=item["created_at"], mode=item.get("mode") )) @@ -249,6 +249,9 @@ async def websocket_chat_endpoint(websocket: WebSocket, user_id: str): })) full_text = "" + stream_trace_id = None + stream_tokens_used = None + stream_duration_ms = 0 try: async for chunk in send_message_stream( db_session=db_session, @@ -261,6 +264,9 @@ async def websocket_chat_endpoint(websocket: WebSocket, user_id: str): if isinstance(chunk, dict): remaining_tokens = None if chunk.get("type") == "stats" and not user_id.startswith("guest_"): + stream_trace_id = chunk.get("trace_id") + stream_tokens_used = chunk.get("tokens_used") + stream_duration_ms = chunk.get("duration_ms", 0) charge_result = await token_service.charge_tokens( user_id=numeric_user_id, mode="text_chat", @@ -307,6 +313,31 @@ async def websocket_chat_endpoint(websocket: WebSocket, user_id: str): })) continue + # 写入聊天历史记录（ChatLog） + try: + input_tokens = 0 + output_tokens = 0 + if isinstance(stream_tokens_used, dict): + input_tokens = int(stream_tokens_used.get("input") or stream_tokens_used.get("input_tokens") or 0) + output_tokens = int(stream_tokens_used.get("output") or stream_tokens_used.get("output_tokens") or 0) + chat_log = ChatLog( + user_id=numeric_user_id, + trace_id=stream_trace_id or str(uuid.uuid4()), + pet_id=pet_id, + bg_id=background_id, + user_msg=msg_content, + ai_msg=full_text, + conversation_id=conversation_id or str(uuid.uuid4()), + tokens_input=input_tokens, + tokens_output=output_tokens, + duration_ms=stream_duration_ms, + mode="text_chat" + ) + db_session.add(chat_log) + await db_session.commit() + except Exception as log_err: + logger.error(f"Failed to save text chat ChatLog: {log_err}") + await websocket.send_text(json.dumps({ "type": "typing_end", "timestamp": datetime.now().isoformat() diff --git a/backend/api/models/chat.py b/backend/api/models/chat.py index 77fe38f..43fda0a 100644 --- a/backend/api/models/chat.py +++ b/backend/api/models/chat.py @@ -23,8 +23,8 @@ class SendMessageResponse(BaseModel): class ChatHistoryItem(BaseModel): """聊天历史项""" id: int - user_msg: str - ai_msg: str + user_msg: Optional[str] = "" + ai_msg: Optional[str] = "" created_at: datetime mode: Optional[str] = Field(None, description="消息模式: text_chat | voice_chat | realtime_call") diff --git a/backend/api/voice_chat_router.py b/backend/api/voice_chat_router.py index 668eb36..3133539 100644 --- a/backend/api/voice_chat_router.py +++ b/backend/api/voice_chat_router.py @@ -405,16 +405,31 @@ async def websocket_voice_chat(websocket: WebSocket, user_id: str): session_state["last_persisted_user_msg"] = "" session_state["last_persisted_ai_len"] = 0 - client = await handle_start_session( + result = await handle_start_session( websocket, user_id, message, client ) - if client: + if result: + client, _session_config = result + # 保存会话配置到 session_state，供自动重连使用 + session_state['_session_config'] = _session_config # 启动事件接收任务 receive_task = asyncio.create_task( receive_events_loop(websocket, client, user_id, "voice_chat", session_state) ) elif msg_type == "audio_data": + # 接收音频数据 - 检查会话是否已终止 + if session_state.get("session_dead"): + try: + await websocket.send_text(json.dumps({ + "type": "session_stopped", + "reason": "session_dead", + "message": "语音会话已结束，请重新开始", + "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + continue # 接收音频数据 - 检查token是否已耗尽 if session_state.get("token_exhausted"): try: @@ -429,29 +444,56 @@ async def websocket_voice_chat(websocket: WebSocket, user_id: str): except Exception: pass return - if client and client.is_session_active: + # 使用最新的 client（可能已被 receive_events_loop 重连替换） + active_client = session_state.get('_reconnected_client') or client + if active_client and active_client.is_session_active: audio_b64 = message.get("data", "") if audio_b64: audio_data = base64.b64decode(audio_b64) - await client.send_audio(audio_data) + try: + await active_client.send_audio(audio_data) + except Exception as e: + logger.error(f"[VOICE_SEND_AUDIO_FAILED] user_id={user_id} error={e}") + try: + await websocket.send_text(json.dumps({ + "type": "session_stopped", + "reason": "audio_send_failed", + "message": "语音连接中断，请重新开始", + "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + return else: - logger.warning("Received audio but session not active") + logger.warning(f"[VOICE_SESSION_DEAD] user_id={user_id} client={client is not None} active={client.is_session_active if client else False}") + # 会话已结束但前端还在发音频，通知前端重新开始 + try: + await websocket.send_text(json.dumps({ + "type": "session_stopped", + "reason": "session_not_active", + "message": "语音会话已结束，请重新开始", + "timestamp": datetime.now().isoformat() + })) + except Exception: + pass elif msg_type == "end_speech": # 用户停止说话（push_to_talk模式下通知服务端） logger.info(f"User {user_id} ended speech") - if client and client.is_session_active: + active_client = session_state.get('_reconnected_client') or client + if active_client and active_client.is_session_active: try: - await client.send_end_asr() + await active_client.send_end_asr() except Exception as e: logger.warning(f"Failed to send EndASR: {e}") elif msg_type == "text_query": # 发送文本query - if client and client.is_session_active: + active_client = session_state.get('_reconnected_client') or client + if active_client and active_client.is_session_active: content = message.get("content", "") if content: - await client.send_text_query(content) + await active_client.send_text_query(content) else: await websocket.send_text(json.dumps({ "type": "error", @@ -461,9 +503,10 @@ async def websocket_voice_chat(websocket: WebSocket, user_id: str): elif msg_type == "say_hello": # 发送打招呼 - if client and client.is_session_active: + active_client = session_state.get('_reconnected_client') or client + if active_client and active_client.is_session_active: content = message.get("content", "你好！") - await client.say_hello(content) + await active_client.say_hello(content) elif msg_type == "stop_session": # 停止会话 @@ -644,6 +687,7 @@ async def websocket_realtime_call(websocket: WebSocket, user_id: str): session_state["conversation_id"] = f"realtime-{user_id}-{uuid.uuid4().hex[:12]}" session_state["trace_id"] = f"trace-{user_id}-{uuid.uuid4().hex[:12]}" session_state["token_charged"] = False + session_state['_session_config'] = config receive_task = asyncio.create_task( receive_events_loop(websocket, client, user_id, "voice_realtime_call", session_state) ) @@ -1014,7 +1058,7 @@ async def handle_start_session( user_id: str, message: dict, existing_client: Optional[RealtimeVoiceClient] -) -> Optional[RealtimeVoiceClient]: +): """处理开始会话请求""" # 如果已有客户端，先关闭 @@ -1091,7 +1135,7 @@ async def handle_start_session( })) logger.info(f"Session started for user {user_id}, session_id={client.session_id}") - return client + return client, config except Exception as e: logger.error(f"Failed to start session: {str(e)}") @@ -1148,183 +1192,256 @@ async def receive_events_loop( charge_mode: str, session_state: Optional[Dict[str, Any]] = None ) -> None: - """接收服务端事件并转发到WebSocket""" - + """接收服务端事件并转发到WebSocket，连接断开时自动重连""" + session_state = session_state or {} + MAX_RECONNECT_ATTEMPTS = 3 + RECONNECT_DELAY = 2 # 秒 + stop_event = asyncio.Event() periodic_task = asyncio.create_task( periodic_token_check(websocket, user_id, charge_mode, session_state, stop_event) ) - try: - async for event in client.receive_events(): - try: - # 转换事件为JSON消息 - msg = { - "type": event.event_type, - "event_id": event.event_id, - "timestamp": datetime.now().isoformat() - } - - # 添加事件数据 - if event.data: - msg["data"] = event.data - - # 处理音频数据 - if event.audio_data: - msg["audio"] = base64.b64encode(event.audio_data).decode('utf-8') - msg["format"] = "pcm_s16le" - msg["sample_rate"] = 24000 - if event.event_type == 'asr_result' and isinstance(event.data, dict): - text = event.data.get('text') or '' - if text and not event.data.get('is_interim', False): - session_state['user_msg'] = text - await _append_voice_log_chunk( - user_id=user_id, - mode=charge_mode, - session_state=session_state, - user_msg=text, - ai_msg=None, - tokens_used=None, - persist_kind='user' - ) - elif event.event_type == 'chat_response' and isinstance(event.data, dict): - delta = event.data.get('content') or '' - if delta: - session_state['ai_msg'] = (session_state.get('ai_msg') or '') + delta - await _append_voice_log_chunk( - user_id=user_id, - mode=charge_mode, - session_state=session_state, - user_msg=None, - ai_msg=session_state['ai_msg'], - tokens_used=None, - persist_kind='ai' - ) - elif event.event_type == 'usage' and isinstance(event.data, dict): - usage_data = event.data.get('usage', {}) - session_state['tokens_used'] = usage_data - - await websocket.send_text(json.dumps(msg)) - - # 特殊事件处理 - if event.event_type == 'asr_start': - logger.debug(f"ASR started: {event.data}") - # 每轮对话重置扣费标记，确保本轮 usage 事件触发扣费 - session_state['token_charged'] = False - _tc = await _check_user_has_tokens(user_id) - if not _tc.get("has_tokens"): - logger.warning(f"[VOICE_TOKEN_EXHAUSTED] user_id={user_id}") - try: - await websocket.send_text(json.dumps({ - "type": "error", - "error": "insufficient_tokens", - "message": "Token不足，语音会话已结束", - "available_tokens": 0, - "timestamp": datetime.now().isoformat() - })) - except Exception: - pass + current_client = client + should_reconnect = True + + while should_reconnect: + reconnect_attempt = 0 + connected = False + + # 如果当前客户端已断开，尝试重连 + if not current_client.is_connected: + while reconnect_attempt < MAX_RECONNECT_ATTEMPTS: + reconnect_attempt += 1 + logger.info(f"[VOICE_RECONNECT] user_id={user_id} attempt={reconnect_attempt}/{MAX_RECONNECT_ATTEMPTS}") + try: + await websocket.send_text(json.dumps({ + "type": "reconnecting", + "attempt": reconnect_attempt, + "max_attempts": MAX_RECONNECT_ATTEMPTS, + "message": f"语音连接中断，正在重连({reconnect_attempt}/{MAX_RECONNECT_ATTEMPTS})...", + "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + + try: + # 关闭旧客户端 + try: + await current_client.close() + except Exception: + pass + + # 创建新客户端并连接 + new_client = await realtime_voice_service.create_client(user_id) + if not await new_client.connect(): + logger.warning(f"[VOICE_RECONNECT] connect failed, attempt={reconnect_attempt}") + await asyncio.sleep(RECONNECT_DELAY) + continue + + # 使用保存的配置重建会话 + saved_config = session_state.get('_session_config') + if not saved_config: + logger.error("[VOICE_RECONNECT] No saved session config, cannot reconnect") break - elif event.event_type == 'usage': - usage_data = event.data.get('usage', {}) if isinstance(event.data, dict) else {} - # 避免重复扣费：如果已经扣费成功则跳过 - if session_state.get("token_charged"): - logger.info( - "[VOICE_CHARGE_SKIP] user_id=%s mode=%s reason=%s already_charged=True", - user_id, charge_mode, "usage_event_duplicate" - ) - # 仍然更新 tokens_used 以便 ChatLog 记录 - session_state['tokens_used'] = usage_data - else: - token_charge_config = get_token_charge_config() - fallback_amount_map = { - "voice_chat": token_charge_config.get("voice_chat_fallback_tokens", 200), - "voice_realtime_call": token_charge_config.get("voice_realtime_call_fallback_tokens", 200), - } + if not await new_client.start_session(saved_config): + logger.warning(f"[VOICE_RECONNECT] start_session failed, attempt={reconnect_attempt}") + await asyncio.sleep(RECONNECT_DELAY) + continue - charge_result = await _charge_user_tokens( - user_id=user_id, - mode=charge_mode, - tokens_used=usage_data, - fallback_amount=fallback_amount_map.get(charge_mode, 200) - ) - if charge_result.get("success"): - session_state['tokens_used'] = usage_data - session_state['token_charged'] = True - # 通知客户端剩余token数，便于前端实时更新统计 - try: - await websocket.send_text(json.dumps({ - "type": "token_update", - "remaining_tokens": charge_result.get("available_tokens", 0), - "consumed": charge_result.get("amount", 0), - "mode": charge_mode, - "timestamp": datetime.now().isoformat() - })) - except Exception: - pass - else: - session_state['token_charged'] = False - remaining = charge_result.get("available_tokens", 0) - logger.warning( - f"Failed to charge tokens for user {user_id} in {charge_mode}: " - f"{charge_result.get('message')}, remaining={remaining}" + # 重连成功 + current_client = new_client + # 更新 session_state 中的 client 引用，让外层 handler 也能用新 client + session_state['_reconnected_client'] = new_client + session_state['session_dead'] = False + connected = True + logger.info(f"[VOICE_RECONNECT] Success! new session_id={new_client.session_id}") + try: + await websocket.send_text(json.dumps({ + "type": "reconnected", + "session_id": new_client.session_id, + "message": "语音连接已恢复", + "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + break + + except Exception as e: + logger.error(f"[VOICE_RECONNECT] Error: {e}") + await asyncio.sleep(RECONNECT_DELAY) + + if not connected: + # 所有重连尝试都失败 + logger.error(f"[VOICE_RECONNECT_FAILED] user_id={user_id} all {MAX_RECONNECT_ATTEMPTS} attempts failed") + should_reconnect = False + break + + # 运行事件循环 + loop_should_reconnect = False + try: + event_count = 0 + async for event in current_client.receive_events(): + event_count += 1 + try: + msg = { + "type": event.event_type, + "event_id": event.event_id, + "timestamp": datetime.now().isoformat() + } + if event.data: + msg["data"] = event.data + if event.audio_data: + msg["audio"] = base64.b64encode(event.audio_data).decode('utf-8') + msg["format"] = "pcm_s16le" + msg["sample_rate"] = 24000 + + if event.event_type == 'asr_result' and isinstance(event.data, dict): + text = event.data.get('text') or '' + if text and not event.data.get('is_interim', False): + session_state['user_msg'] = text + await _append_voice_log_chunk( + user_id=user_id, mode=charge_mode, session_state=session_state, + user_msg=text, ai_msg=None, tokens_used=None, persist_kind='user' ) + elif event.event_type == 'chat_response' and isinstance(event.data, dict): + delta = event.data.get('content') or '' + if delta: + session_state['ai_msg'] = (session_state.get('ai_msg') or '') + delta + await _append_voice_log_chunk( + user_id=user_id, mode=charge_mode, session_state=session_state, + user_msg=None, ai_msg=session_state['ai_msg'], tokens_used=None, persist_kind='ai' + ) + elif event.event_type == 'usage' and isinstance(event.data, dict): + session_state['tokens_used'] = event.data.get('usage', {}) + + await websocket.send_text(json.dumps(msg)) + + # 特殊事件处理 + if event.event_type == 'asr_start': + session_state['token_charged'] = False + try: + _tc = await _check_user_has_tokens(user_id) + except Exception as e: + logger.warning(f"[VOICE_TOKEN_CHECK_ERROR] user_id={user_id} error={e}") + _tc = {"has_tokens": True, "available_tokens": -1} + if not _tc.get("has_tokens"): try: - await websocket.send_text(json.dumps({ - "type": "token_update", - "remaining_tokens": remaining, - "consumed": 0, - "mode": charge_mode, - "warning": charge_result.get("message", "扣费失败"), - "timestamp": datetime.now().isoformat() - })) + await asyncio.sleep(0.5) + _tc2 = await _check_user_has_tokens(user_id) except Exception: - pass - # 扣费失败时终止事件循环（余额不够最低消费也要停） - if charge_result.get("reason") == "insufficient_tokens": - logger.warning( - f"[VOICE_TOKEN_EXHAUSTED] user_id={user_id} mode={charge_mode} " - f"remaining={remaining}, breaking event loop and closing" - ) - session_state["token_exhausted"] = True + _tc2 = {"has_tokens": True} + if not _tc2.get("has_tokens"): + logger.warning(f"[VOICE_TOKEN_EXHAUSTED] user_id={user_id}") try: await websocket.send_text(json.dumps({ - "type": "error", - "error": "insufficient_tokens", - "message": "Token已耗尽，语音会话已结束", - "available_tokens": 0, + "type": "error", "error": "insufficient_tokens", + "message": "Token不足，语音会话已结束", + "available_tokens": 0, "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + session_state["token_exhausted"] = True + break + elif event.event_type == 'usage': + usage_data = event.data.get('usage', {}) if isinstance(event.data, dict) else {} + if session_state.get("token_charged"): + session_state['tokens_used'] = usage_data + else: + token_charge_config = get_token_charge_config() + fallback_amount_map = { + "voice_chat": token_charge_config.get("voice_chat_fallback_tokens", 200), + "voice_realtime_call": token_charge_config.get("voice_realtime_call_fallback_tokens", 200), + } + charge_result = await _charge_user_tokens( + user_id=user_id, mode=charge_mode, tokens_used=usage_data, + fallback_amount=fallback_amount_map.get(charge_mode, 200) + ) + if charge_result.get("success"): + session_state['tokens_used'] = usage_data + session_state['token_charged'] = True + try: + await websocket.send_text(json.dumps({ + "type": "token_update", + "remaining_tokens": charge_result.get("available_tokens", 0), + "consumed": charge_result.get("amount", 0), + "mode": charge_mode, "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + else: + session_state['token_charged'] = False + remaining = charge_result.get("available_tokens", 0) + try: + await websocket.send_text(json.dumps({ + "type": "token_update", "remaining_tokens": remaining, + "consumed": 0, "mode": charge_mode, + "warning": charge_result.get("message", "扣费失败"), "timestamp": datetime.now().isoformat() })) except Exception: pass - break - elif event.event_type == 'dialog_error': - logger.warning( - f"DialogCommonError 599: status_code={event.data.get('status_code')} " - f"message={event.data.get('message')}" - ) - elif event.event_type == 'session_failed': - logger.error(f"Session failed: {event.data.get('error')}") - break - - except Exception as e: - logger.error(f"Error forwarding event: {str(e)}") - - except asyncio.CancelledError: - logger.info("Event receive loop cancelled") - except Exception as e: - logger.error(f"Event receive loop error: {str(e)}") - finally: - stop_event.set() - if not periodic_task.done(): - periodic_task.cancel() - try: - await periodic_task - except asyncio.CancelledError: - pass - await _persist_voice_session_history(user_id, charge_mode, session_state) + if charge_result.get("reason") == "insufficient_tokens": + session_state["token_exhausted"] = True + try: + await websocket.send_text(json.dumps({ + "type": "error", "error": "insufficient_tokens", + "message": "Token已耗尽，语音会话已结束", + "available_tokens": 0, "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + break + elif event.event_type == 'dialog_error': + logger.warning(f"DialogError 599: {event.data.get('message')}") + loop_should_reconnect = True + break + elif event.event_type == 'session_failed': + logger.error(f"Session failed: {event.data.get('error')}") + loop_should_reconnect = True + break + + except Exception as e: + logger.error(f"Error forwarding event: {str(e)}") + + except asyncio.CancelledError: + logger.info("Event receive loop cancelled") + except Exception as e: + logger.error(f"Event receive loop error: {str(e)}") + loop_should_reconnect = True + + # 如果 async for 正常结束（火山引擎静默断开，不抛异常）， + # 也需要触发重连。仅在连接确实断开时重连，避免打断正常结束的会话。 + if not loop_should_reconnect and event_count > 0 and not current_client.is_connected: + logger.info("[VOICE_SILENT_DISCONNECTION] Event loop ended without exception, client disconnected, triggering reconnect") + loop_should_reconnect = True + + # 内层循环结束后判断是否需要重连 + if loop_should_reconnect and not session_state.get("token_exhausted"): + continue # 继续外层 while 循环进行重连 + + should_reconnect = False + + # 最终清理——只在彻底结束时执行 + stop_event.set() + if not periodic_task.done(): + periodic_task.cancel() + try: + await periodic_task + except asyncio.CancelledError: + pass + session_state['session_dead'] = True + if not session_state.get("token_exhausted"): + try: + await websocket.send_text(json.dumps({ + "type": "session_stopped", "reason": "event_loop_ended", + "message": "语音会话已结束", "timestamp": datetime.now().isoformat() + })) + except Exception: + pass + await _persist_voice_session_history(user_id, charge_mode, session_state) # ============ 工具函数 ============ diff --git a/backend/models/database.py b/backend/models/database.py index 121ea8c..4195396 100644 --- a/backend/models/database.py +++ b/backend/models/database.py @@ -90,7 +90,7 @@ class ChatLog(Base): duration_ms = Column(Integer, nullable=True) conversation_id = Column(String(64), nullable=False) mode = Column(String(20), nullable=True, default="text_chat") # text_chat | voice_chat | realtime_call - created_at = Column(DateTime, default=datetime.utcnow) + created_at = Column(DateTime, default=datetime.now) class UserWallet(Base): """用户钱包模型""" diff --git a/backend/scripts/add_chat_log_mode_column.py b/backend/scripts/add_chat_log_mode_column.py new file mode 100644 index 0000000..2fa0fd2 --- /dev/null +++ b/backend/scripts/add_chat_log_mode_column.py @@ -0,0 +1,60 @@ +""" +数据库迁移脚本：为 t_chat_log 表添加 mode 列 +区分文本聊天、语音聊天、实时通话的 token 消耗 +""" + +import pymysql + + +def add_chat_log_mode_column(): + """迁移数据库，添加 mode 列""" + connection = pymysql.connect( + host='localhost', + user='root', + password='taiyi1224', + database='pet_companion', + charset='utf8mb4' + ) + + try: + with connection.cursor() as cursor: + cursor.execute(""" + SELECT COLUMN_NAME + FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_NAME = 't_chat_log' + AND COLUMN_NAME = 'mode' + """) + existing_column = cursor.fetchone() + + if existing_column: + print("✓ mode 列已存在，无需迁移") + else: + cursor.execute(""" + ALTER TABLE t_chat_log + ADD COLUMN `mode` VARCHAR(20) NULL DEFAULT 'text_chat' AFTER `conversation_id` + """) + print("✓ 已成功添加 mode 列") + + # 将现有记录的 mode 初始化为 text_chat + cursor.execute(""" + UPDATE t_chat_log + SET `mode` = 'text_chat' + WHERE `mode` IS NULL + """) + print("✓ 已将现有记录的 mode 初始化为 text_chat") + + connection.commit() + print("✓ 迁移完成！") + + except Exception as e: + print(f"✗ 迁移失败: {e}") + connection.rollback() + raise + finally: + connection.close() + + +if __name__ == "__main__": + print("开始数据库迁移...") + add_chat_log_mode_column() + print("迁移完成！") diff --git a/backend/services/chat_service.py b/backend/services/chat_service.py index 1f4b642..a833629 100644 --- a/backend/services/chat_service.py +++ b/backend/services/chat_service.py @@ -8,8 +8,25 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select from sqlalchemy import select as sa_select import uuid -from datetime import datetime +from datetime import datetime, timedelta, timezone import logging + +# 北京时间时区 +_BEIJING_TZ = timezone(timedelta(hours=8)) + + +def _utc_to_beijing(dt: datetime) -> str: + """将 datetime 转为北京时间 ISO 格式字符串（带 +08:00） + + 数据库中所有 created_at 统一存储北京时间（datetime.now()）， + 此处直接补上 +08:00 时区标记，不做额外转换。 + """ + if dt is None: + return None + if dt.tzinfo is None: + # 数据库存的是 naive 北京时间，补上 +08:00 时区标记 + dt = dt.replace(tzinfo=_BEIJING_TZ) + return dt.isoformat() import asyncio from models.database import ChatLog, Pet, Background, BackgroundPetConfig, UserWallet @@ -186,7 +203,7 @@ async def get_chat_history( "id": log.id, "user_msg": log.user_msg, "ai_msg": log.ai_msg, - "created_at": log.created_at.isoformat() if log.created_at else None, + "created_at": _utc_to_beijing(log.created_at), "mode": getattr(log, "mode", None) or "text_chat" }) @@ -295,32 +312,11 @@ async def send_message_stream(db_session: AsyncSession, user_id: int, message: s full_response += chunk yield chunk - chat_log = ChatLog( - user_id=user_id, - trace_id=trace_id, - pet_id=pet_id, - bg_id=background_id, - user_msg=message, - ai_msg=full_response, - conversation_id=conversation_id, - tokens_input=input_tokens, - tokens_output=output_tokens, - duration_ms=duration_ms, - mode="text_chat" + logger.info( + f"Stream message processed with Doubao: user={user_id}, pet={pet_id}, " + f"response_length={len(full_response)}" ) - # 使用独立会话保存记录，避免受路由 get_db() 生命周期影响 - async with AsyncSessionLocal() as save_session: - save_session.add(chat_log) - await save_session.commit() - - # 查询用户实际剩余token - wallet_result = await save_session.execute( - sa_select(UserWallet).where(UserWallet.user_id == user_id) - ) - wallet = wallet_result.scalar_one_or_none() - remaining_tokens = (wallet.available_tokens or 0) if wallet else 0 - yield { "type": "stats", "tokens_used": { @@ -329,16 +325,12 @@ async def send_message_stream(db_session: AsyncSession, user_id: int, message: s "total": input_tokens + output_tokens }, "duration_ms": duration_ms, - "remaining_tokens": remaining_tokens, "conversation_id": conversation_id, - "trace_id": trace_id + "trace_id": trace_id, + "full_response": full_response, + "user_msg": message } - logger.info( - f"Stream message processed with Doubao: user={user_id}, pet={pet_id}, " - f"response_length={len(full_response)}" - ) - except Exception as e: logger.error(f"Error in send_message_stream: {str(e)}") yield f"抱歉，处理消息时出现错误：{str(e)}" diff --git a/backend/services/realtime_voice_service.py b/backend/services/realtime_voice_service.py index 1453316..f11b1b7 100644 --- a/backend/services/realtime_voice_service.py +++ b/backend/services/realtime_voice_service.py @@ -194,35 +194,37 @@ class RealtimeVoiceClient: """ try: logger.info(f"Connecting to Realtime Voice API: {self.API_URL}") - + headers = self._get_headers() - + # 兼容不同版本的websockets库 # 14.0+ 使用 additional_headers, 之前的版本使用 extra_headers + # ping_interval=20 启用心跳保活，防止NAT/防火墙静默断开连接 + # ping_timeout=20 确保心跳超时时能及时检测到连接已死 + ws_kwargs = { + "additional_headers": headers, + "ping_interval": None, # 火山引擎不支持 WebSocket ping/pong，不能开 + "max_size": 10 * 1024 * 1024, # 10MB + "close_timeout": 5, + } try: # 优先尝试 additional_headers (新版本) - self._ws = await websockets.connect( - self.API_URL, - additional_headers=headers, - ping_interval=None, - max_size=10 * 1024 * 1024 # 10MB - ) + self._ws = await websockets.connect(self.API_URL, **ws_kwargs) except TypeError: try: # 尝试 extra_headers (旧版本) - self._ws = await websockets.connect( - self.API_URL, - extra_headers=headers, - ping_interval=None, - max_size=10 * 1024 * 1024 # 10MB - ) + ws_kwargs.pop("additional_headers", None) + ws_kwargs["extra_headers"] = headers + self._ws = await websockets.connect(self.API_URL, **ws_kwargs) except TypeError: # 如果都不行，尝试作为普通参数 (极旧版本或特殊情况) + ws_kwargs.pop("extra_headers", None) self._ws = await websockets.connect( self.API_URL, headers=headers, ping_interval=None, - max_size=10 * 1024 * 1024 + max_size=10 * 1024 * 1024, + close_timeout=5, ) # 获取logid @@ -445,25 +447,33 @@ class RealtimeVoiceClient: async def receive_event(self) -> Optional[RealtimeEvent]: """ 接收单个事件 - + Returns: RealtimeEvent或None """ if not self._ws: return None - + try: - response = await self._ws.recv() + # 加60秒超时，防止连接已死但未检测到时 recv() 永远卡住 + response = await asyncio.wait_for(self._ws.recv(), timeout=60) parsed = parse_response(response) - + return self._convert_to_event(parsed) - except ConnectionClosed: - logger.info("WebSocket connection closed") + except asyncio.TimeoutError: + logger.warning("[VOICE_RECV_TIMEOUT] 60s no response from Volcano Engine, connection may be dead") + self._is_connected = False + self._is_session_active = False + return None + except ConnectionClosed as e: + logger.info(f"WebSocket connection closed: code={e.code} reason={e.reason}") self._is_connected = False self._is_session_active = False return None except Exception as e: logger.error(f"Failed to receive event: {str(e)}") + self._is_connected = False + self._is_session_active = False return None async def receive_events(self): diff --git a/frontend/src/stores/chatStore.js b/frontend/src/stores/chatStore.js index 0897498..f27b024 100644 --- a/frontend/src/stores/chatStore.js +++ b/frontend/src/stores/chatStore.js @@ -50,6 +50,8 @@ export const useChatStore = defineStore('chat', () => { // 音频相关 const audioContext = ref(null) + let _recordingSource = null // 当前录音的 MediaStreamSource + let _recordingProcessor = null // 当前录音的 ScriptProcessorNode const mediaRecorder = ref(null) const audioChunks = ref([]) const permissionStream = ref(null) @@ -81,6 +83,9 @@ export const useChatStore = defineStore('chat', () => { const audioBufferMap = ref({}) // 按 question_id 缓冲音频数据 const pendingChatContent = ref('') // 缓冲的AI文字回复 + // 请求代次：每次发送新消息时递增，用于丢弃过时的响应 + let _chatRequestGeneration = 0 + // ============ 计算属性 ============ const currentPetId = computed(() => { const pet = JSON.parse(localStorage.getItem('selectedPet') || '{}') @@ -117,6 +122,9 @@ export const useChatStore = defineStore('chat', () => { const bgId = options.backgroundId || currentBackgroundId.value const conversationId = options.conversationId || currentConversationId.value + // 递增请求代次，用于区分新旧响应 + const myGeneration = ++_chatRequestGeneration + const userMessage = { id: Date.now(), content, @@ -125,7 +133,19 @@ export const useChatStore = defineStore('chat', () => { status: 'sending' } + // 提前创建 AI 流式消息，这样旧响应能精确找到自己的消息而不是用 findLastIndex + const aiMessageId = Date.now() + 1 + const aiMessage = { + id: aiMessageId, + content: '', + isUser: false, + timestamp: new Date().toISOString(), + status: 'streaming', + isStreaming: true + } + messages.value.push(userMessage) + messages.value.push(aiMessage) isLoading.value = true console.log('[ChatStore] User message added:', userMessage) @@ -154,12 +174,32 @@ export const useChatStore = defineStore('chat', () => { } } + // 通用的消息结束处理：标记用户消息和AI消息为完成状态 + const finalizeMessage = () => { + const uIdx = messages.value.findIndex(m => m.id === userMessage.id) + if (uIdx !== -1) messages.value[uIdx].status = 'sent' + const aIdx = messages.value.findIndex(m => m.id === aiMessageId) + if (aIdx !== -1) { + messages.value[aIdx].status = 'completed' + messages.value[aIdx].isStreaming = false + if (finalResponse.response) { + messages.value[aIdx].content = finalResponse.response + } + } + } + const fail = (error) => { cleanup() const index = messages.value.findIndex(m => m.id === userMessage.id) if (index !== -1) { messages.value[index].status = 'failed' } + // AI 消息标记为失败而非卡在 streaming + const aIdx = messages.value.findIndex(m => m.id === aiMessageId) + if (aIdx !== -1 && messages.value[aIdx].status === 'streaming') { + messages.value[aIdx].status = 'failed' + messages.value[aIdx].isStreaming = false + } isLoading.value = false ElMessage.error('消息发送失败，请重试') reject(error) @@ -187,6 +227,9 @@ export const useChatStore = defineStore('chat', () => { } socket.onmessage = (event) => { + // 判断是否是旧响应（有更新的请求已经发出） + const isStale = _chatRequestGeneration !== myGeneration + let data try { data = JSON.parse(event.data) @@ -199,24 +242,17 @@ export const useChatStore = defineStore('chat', () => { case 'connected': break case 'typing_start': - isAiProcessing.value = true + // 只有当前代次才设置 isAiProcessing + if (!isStale) isAiProcessing.value = true break case 'message_chunk': { const chunk = data.content || '' if (!chunk) break finalResponse.response += chunk - const lastAiMsgIndex = messages.value.findLastIndex(m => !m.isUser && m.status === 'streaming') - if (lastAiMsgIndex !== -1) { - messages.value[lastAiMsgIndex].content = finalResponse.response - } else { - messages.value.push({ - id: Date.now() + 1, - content: finalResponse.response, - isUser: false, - timestamp: new Date().toISOString(), - status: 'streaming', - isStreaming: true - }) + // 精确按 ID 更新自己的消息（不用 findLastIndex，避免并发冲突） + const aIdx = messages.value.findIndex(m => m.id === aiMessageId) + if (aIdx !== -1) { + messages.value[aIdx].content = finalResponse.response } break } @@ -231,28 +267,19 @@ export const useChatStore = defineStore('chat', () => { } break case 'typing_end': { - isAiProcessing.value = false - const index = messages.value.findIndex(m => m.id === userMessage.id) - if (index !== -1) messages.value[index].status = 'sent' - const streamingMsg = messages.value.find(m => m.isStreaming) - if (streamingMsg) { - streamingMsg.status = 'completed' - streamingMsg.isStreaming = false - } - if (finalResponse.response) { - const lastAiMsgIndex = messages.value.findLastIndex(m => !m.isUser) - if (lastAiMsgIndex !== -1) { - messages.value[lastAiMsgIndex].status = 'completed' - messages.value[lastAiMsgIndex].content = finalResponse.response - } - } + // 无论新旧，都要 finalize 消息——文本完整保存到聊天历史 + finalizeMessage() if (!finalResponse.conversation_id) { finalResponse.conversation_id = currentConversationId.value } - currentConversationId.value = finalResponse.conversation_id - isLoading.value = false + // 只有当前代次才更新全局状态 + if (!isStale) { + isAiProcessing.value = false + currentConversationId.value = finalResponse.conversation_id + isLoading.value = false + resolve(finalResponse) + } cleanup() - resolve(finalResponse) break } case 'token_update': @@ -268,7 +295,13 @@ export const useChatStore = defineStore('chat', () => { if (data.error === 'insufficient_tokens') { _tokenExhausted = true } - fail(new Error(data.message || data.error || '聊天失败')) + // 旧响应的错误也妥善处理消息状态 + if (isStale) { + finalizeMessage() + cleanup() + } else { + fail(new Error(data.message || data.error || '聊天失败')) + } break } } @@ -278,8 +311,16 @@ export const useChatStore = defineStore('chat', () => { } socket.onclose = () => { - if (isLoading.value) { + // 如果 WebSocket 意外关闭但消息还在 streaming，妥善结束 + const aIdx = messages.value.findIndex(m => m.id === aiMessageId) + if (aIdx !== -1 && messages.value[aIdx].status === 'streaming') { + finalizeMessage() + } + // 确保 promise 不会永远挂起（typing_end 可能没收到） + if (_chatRequestGeneration === myGeneration) { + isAiProcessing.value = false isLoading.value = false + resolve(finalResponse) } } }) @@ -569,8 +610,21 @@ export const useChatStore = defineStore('chat', () => { break case 'session_stopped': - console.log('[ChatStore] Voice session stopped') + console.log('[ChatStore] Voice session stopped', data.reason) resetAudioState() + voiceSessionReady.value = false + // 通知用户会话已结束 + if (data.reason === 'dialog_error') { + ElMessage.warning(data.message || '语音对话出错，会话已结束') + } else if (data.reason === 'audio_send_failed') { + ElMessage.warning('语音连接中断，请重新开始') + } else if (data.reason === 'session_failed') { + ElMessage.error('语音会话异常结束，请重新开始') + } else { + ElMessage.info('语音会话已结束') + } + // 断开语音WebSocket，用户需要重新点击按钮开始新会话 + disconnectVoiceWebSocket() // 触发统计更新 triggerStatsUpdate() break @@ -1326,8 +1380,20 @@ export const useChatStore = defineStore('chat', () => { await ensureAudioContext(audioContext.value); + // 断开上一次录音遗留的音频节点，防止泄漏 + if (_recordingSource) { + try { _recordingSource.disconnect() } catch (_) {} + _recordingSource = null + } + if (_recordingProcessor) { + try { _recordingProcessor.disconnect() } catch (_) {} + _recordingProcessor = null + } + const source = audioContext.value.createMediaStreamSource(stream) const processor = audioContext.value.createScriptProcessor(4096, 1, 1) + _recordingSource = source + _recordingProcessor = processor source.connect(processor) processor.connect(audioContext.value.destination) @@ -1413,14 +1479,25 @@ export const useChatStore = defineStore('chat', () => { return } + // 不再关闭 AudioContext，复用同一个实例以避免移动端浏览器 + // AudioContext 创建上限导致 2-3 次录音后无法再创建新实例的问题。 + // 断开录制相关的音频节点，AudioContext 保持存活供下次录音复用。 + if (_recordingSource) { + try { _recordingSource.disconnect() } catch (_) {} + _recordingSource = null + } + if (_recordingProcessor) { + try { _recordingProcessor.disconnect() } catch (_) {} + _recordingProcessor = null + } try { - if (audioContext.value) { - audioContext.value.close() + if (audioContext.value && audioContext.value.state !== 'closed') { + // 挂起而非关闭，下次录音时 resume 即可 + audioContext.value.suspend().catch(() => {}) } } catch (error) { - console.warn('[ChatStore] AudioContext.close() failed:', error) + console.warn('[ChatStore] AudioContext.suspend() failed:', error) } - audioContext.value = null }) } @@ -1436,6 +1513,7 @@ export const useChatStore = defineStore('chat', () => { // 持续运行的音频队列处理器 let audioQueueRunning = false let nextScheduleTime = 0 + let activeBufferSources = [] // 追踪所有已调度的 BufferSource，用于 stop 时全部停止 async function startAudioQueueProcessor() { if (audioQueueRunning) return @@ -1451,6 +1529,10 @@ export const useChatStore = defineStore('chat', () => { await ensureAudioContext(currentAudioContext) nextScheduleTime = currentAudioContext.currentTime + // 连续空轮询计数，用于延迟 suspend 避免频繁 suspend/resume + let emptyPolls = 0 + const IDLE_SUSPEND_THRESHOLD = 50 // 50 × 20ms = 1秒无新数据后 suspend + try { while (audioQueueRunning) { // 取出当前队列中所有块 @@ -1461,11 +1543,22 @@ export const useChatStore = defineStore('chat', () => { } if (chunks.length === 0) { + emptyPolls++ + // 长时间无新数据时挂起 AudioContext，减少移动端音频源竞争 + if (emptyPolls >= IDLE_SUSPEND_THRESHOLD && currentAudioContext?.state === 'running') { + try { await currentAudioContext.suspend() } catch (_) {} + } // 无新块，短暂等待后继续检查 await new Promise(r => setTimeout(r, 20)) continue } + // 收到新数据，确保 AudioContext 处于运行状态 + emptyPolls = 0 + if (currentAudioContext?.state === 'suspended') { + await ensureAudioContext(currentAudioContext) + } + // 确保调度时间不早于当前时间（防止累积延迟） if (nextScheduleTime < currentAudioContext.currentTime) { nextScheduleTime = currentAudioContext.currentTime @@ -1483,6 +1576,10 @@ export const useChatStore = defineStore('chat', () => { isPlayingQueue.value = false isPlaying.value = false audioQueueRunning = false + // 处理器退出时挂起 AudioContext，避免持续占用音频输出 + if (currentAudioContext?.state === 'running') { + try { await currentAudioContext.suspend() } catch (_) {} + } } } @@ -1490,6 +1587,11 @@ export const useChatStore = defineStore('chat', () => { audioQueueRunning = false isPlayingQueue.value = false isPlaying.value = false + // 停止所有已调度但尚未播放完的 BufferSource + for (const src of activeBufferSources) { + try { src.stop() } catch (_) {} + } + activeBufferSources = [] } /** @@ -1515,6 +1617,12 @@ export const useChatStore = defineStore('chat', () => { source.buffer = audioBuffer source.connect(currentAudioContext.destination) source.start(startTime) + // 追踪所有活跃的 BufferSource，用于 stop 时全部停止 + activeBufferSources.push(source) + source.onended = () => { + const idx = activeBufferSources.indexOf(source) + if (idx !== -1) activeBufferSources.splice(idx, 1) + } return { source, duration } } @@ -1597,6 +1705,15 @@ export const useChatStore = defineStore('chat', () => { pendingChatContent.value = '' currentQuestionId.value = null audioBufferMap.value = {} + // 清理录音音频节点引用 + if (_recordingSource) { + try { _recordingSource.disconnect() } catch (_) {} + _recordingSource = null + } + if (_recordingProcessor) { + try { _recordingProcessor.disconnect() } catch (_) {} + _recordingProcessor = null + } } /** diff --git a/frontend/src/views/ChatView.vue b/frontend/src/views/ChatView.vue index 2c53391..ec9f9b9 100644 --- a/frontend/src/views/ChatView.vue +++ b/frontend/src/views/ChatView.vue @@ -571,7 +571,6 @@ export default { const conversationPanelText = ref('') const showConversationBubbles = ref(false) const backgroundAudioEnabled = ref(false) - const preVoiceVideoMuted = ref(true) // 进入语音模式前背景视频的 muted 状态 const showBackgroundAudioHint = ref(false) const lastConversationCompletedAt = ref(null) const measuredUserBubbleHeight = ref(72) @@ -678,7 +677,11 @@ export default { try { video.loop = true video.muted = !forceUnmute - video.volume = 1 + // 语音/通话模式下不强制恢复音量，由 isPlaying watcher 管理避让 + const inVoiceOrCall = chatMode.value === 'voice' || chatMode.value === 'call' + if (!inVoiceOrCall || !chatStore.isPlaying) { + video.volume = 1 + } await video.play() backgroundAudioEnabled.value = !video.muted @@ -761,7 +764,9 @@ export default { const handleBackgroundVideoReady = async () => { enforceInlineBackgroundVideo() - await tryEnableBackgroundAudio({ forceUnmute: true }) + // 语音/通话模式下不强制 unmute，避免与 AI 语音冲突 + const inVoiceOrCall = chatMode.value === 'voice' || chatMode.value === 'call' + await tryEnableBackgroundAudio({ forceUnmute: !inVoiceOrCall }) } const handleBackgroundVideoEnded = async () => { @@ -774,7 +779,9 @@ export default { video.currentTime = 0 } catch {} - await tryEnableBackgroundAudio({ forceUnmute: true }) + // 语音/通话模式下不强制 unmute，避免与 AI 语音冲突 + const inVoiceOrCall = chatMode.value === 'voice' || chatMode.value === 'call' + await tryEnableBackgroundAudio({ forceUnmute: !inVoiceOrCall }) } const showConversationPanelText = (text) => { @@ -1625,6 +1632,11 @@ export default { return } + // 如果上一个AI回复还在播放/加载中，先停止 + if (chatStore.isLoading || chatStore.isPlaying || chatStore.isAiProcessing) { + chatStore.stopAudio() + } + const messageText = inputMessage.value.trim() inputMessage.value = '' sending.value = true @@ -2071,15 +2083,16 @@ export default { const size = 200 const maxPages = 20 + const params = { pet_id: petId, bg_id: bgId, size } + const userId = userStore.getUserId + if (userId && typeof userId === 'string' && userId.startsWith('guest_')) { + params.user_id = userId + } + while (page <= maxPages) { - const response = await chatStore.loadChatHistory( - petId, - bgId, - page, - size, - null, - null - ) + params.page = page + const res = await chatAPI.getHistory(params) + const response = res.data const list = Array.isArray(response?.data) ? response.data : [] list.forEach(item => { @@ -2117,30 +2130,34 @@ export default { return } - let startDate = null - let endDate = null + const params = { + pet_id: petId, + bg_id: bgId, + page: historyPage.value, + size: historyPageSize.value + } + + const userId = userStore.getUserId + if (userId && typeof userId === 'string' && userId.startsWith('guest_')) { + params.user_id = userId + } + if (Array.isArray(historyDateRange.value) && historyDateRange.value.length === 2) { const [start, end] = historyDateRange.value if (start) { const startDt = new Date(start) startDt.setHours(0, 0, 0, 0) - startDate = startDt.toISOString() + params.start_date = startDt.toISOString() } if (end) { const endDt = new Date(end) endDt.setHours(23, 59, 59, 999) - endDate = endDt.toISOString() + params.end_date = endDt.toISOString() } } - const response = await chatStore.loadChatHistory( - petId, - bgId, - historyPage.value, - historyPageSize.value, - startDate, - endDate - ) + const res = await chatAPI.getHistory(params) + const response = res.data chatHistory.value = response?.data || [] historyTotal.value = response?.total || 0 @@ -2259,15 +2276,7 @@ export default { ElMessage.warning('麦克风权限已被拒绝，某些功能可能无法使用。请在浏览器设置中开启麦克风权限。') } - watch(() => chatStore.messages, (newMessages, oldMessages) => { - console.log('[ChatView] Messages changed!') - console.log('[ChatView] New messages length:', newMessages?.length) - console.log('[ChatView] New messages:', newMessages) - console.log('[ChatView] Old messages length:', oldMessages?.length) - - const latestAi = newMessages?.slice().reverse().find(m => !m.isUser) - console.log('[ChatView] Latest AI message found:', latestAi) - + watch(() => chatStore.messages, (newMessages) => { if (Array.isArray(newMessages) && newMessages.length) { newMessages.forEach((message) => { if (!message?.id || savedHistoryMessageKeys.has(message.id)) return @@ -2359,15 +2368,26 @@ export default { } }) - // 语音/通话模式下静音背景视频，避免双音频源竞争导致卡顿 - watch(chatMode, (mode) => { + // AI 回复播放时自动降低背景视频音量，结束后恢复，两者同时播放 + const BG_VOLUME_DUCKED = 0.15 // AI 播放时背景音量降至 15% + const BG_VOLUME_NORMAL = 1 // 正常背景音量 + let bgVolumeBeforeDuck = BG_VOLUME_NORMAL + + watch(() => chatStore.isPlaying, (playing) => { const video = backgroundVideoRef.value if (!video) return - if (mode === 'voice' || mode === 'call') { - preVoiceVideoMuted.value = video.muted - video.muted = true + + // 仅在语音/通话模式下做音量避让，其他模式不受影响 + const inVoiceOrCall = chatMode.value === 'voice' || chatMode.value === 'call' + if (!inVoiceOrCall) return + + if (playing) { + // AI 开始播放 → 记住当前音量并降低 + bgVolumeBeforeDuck = video.volume + video.volume = BG_VOLUME_DUCKED } else { - video.muted = preVoiceVideoMuted.value + // AI 停止播放 → 恢复之前的音量 + video.volume = bgVolumeBeforeDuck } })