Merge pull request #3436 from xinnan-tech/perf/hello-handshake-fast

hello 握手首屏响应从 4s 降到 200ms
This commit is contained in:
wengzh authored and GitHub committed 2026-10-27 14:34:13 +08:00
commit 6c352e5978
8 files changed
+84 -24

No files matched your search

+4
View File
@@ -10,6 +10,7 @@ from core.http_server import SimpleHttpServer
from core.websocket_server import WebSocketServer
from core.utils.util import check_ffmpeg_installed
from core.utils.gc_manager import get_gc_manager
from core.handle.helloHandle import preheat_wakeup_audio_cache
TAG = __name__
logger = setup_logging()
@@ -64,6 +65,9 @@ async def main():
# 添加 stdin 监控任务
stdin_task = asyncio.create_task(monitor_stdin())
# 预热唤醒词音频编码缓存,让首条连接就命中
asyncio.create_task(preheat_wakeup_audio_cache())
# 启动全局GC管理器(5分钟清理一次)
gc_manager = get_gc_manager(interval_seconds=300)
await gc_manager.start()
+1 -1
View File
@@ -781,7 +781,7 @@ class ConnectionHandler:
self.voiceprint_provider = voiceprint_provider
self.logger.bind(tag=TAG).info("声纹识别功能已在连接时动态启用")
else:
self.logger.bind(tag=TAG).warning("声纹识别功能启用但配置不完整")
self.logger.bind(tag=TAG).debug("声纹识别功能启用但配置不完整")
else:
self.logger.bind(tag=TAG).info("声纹识别功能未启用")
except Exception as e:
+49 -6
View File
@@ -4,6 +4,7 @@ import uuid
import random
import asyncio
from typing import TYPE_CHECKING
from config.logger import setup_logging
if TYPE_CHECKING:
from core.connection import ConnectionHandler
@@ -13,9 +14,14 @@ from core.providers.tts.dto.dto import SentenceType
from core.utils.wakeup_word import WakeupWordsConfig
from core.handle.sendAudioHandle import sendAudioMessage, send_tts_message
from core.utils.util import remove_punctuation_and_length, opus_datas_to_wav_bytes
from core.providers.tools.device_mcp import MCPClient, send_mcp_initialize_message
from core.providers.tools.device_mcp import (
MCPClient,
send_mcp_initialize_message,
send_mcp_tools_list_request,
)
TAG = __name__
_wakeup_logger = setup_logging()
WAKEUP_CONFIG = {
"refresh_time": 10,
@@ -63,6 +69,8 @@ async def handleHelloMessage(conn: "ConnectionHandler", msg_json):
# The device waits for the server hello before processing MCP messages.
if features and features.get("mcp"):
asyncio.create_task(send_mcp_initialize_message(conn))
# 并发预发 tools/list,让帧排在 audio 流之前到达 client
asyncio.create_task(send_mcp_tools_list_request(conn))
async def checkWakeupWords(conn: "ConnectionHandler", text):
@@ -105,19 +113,19 @@ async def checkWakeupWords(conn: "ConnectionHandler", text):
}
# 获取音频数据
opus_packets = await audio_to_data(response.get("file_path"), use_cache=False)
opus_packets = await audio_to_data(response.get("file_path"), use_cache=True)
# 播放唤醒词回复
conn.client_abort = False
# 将唤醒词回复视为新会话,生成新的 sentence_id,确保流控器重置
conn.sentence_id = str(uuid.uuid4().hex)
conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}")
await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text"))
await sendAudioMessage(conn, SentenceType.LAST, [], None)
# 补充对话
conn.dialogue.put(Message(role="assistant", content=response.get("text")))
conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}")
await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text"))
asyncio.create_task(_finish_wakeup_reply(conn))
# 检查是否需要更新唤醒词回复
if time.time() - response.get("time", 0) > WAKEUP_CONFIG["refresh_time"]:
@@ -126,6 +134,16 @@ async def checkWakeupWords(conn: "ConnectionHandler", text):
return True
async def _finish_wakeup_reply(conn: "ConnectionHandler"):
"""等音频队列清空后再发 stop(时序:全部音频帧→stop),期间不阻塞接收循环"""
try:
await sendAudioMessage(conn, SentenceType.LAST, [], None)
except asyncio.CancelledError:
raise
except Exception as e:
conn.logger.bind(tag=TAG).error(f"唤醒词回复收尾失败: {e}")
async def wakeupWordsResponse(conn: "ConnectionHandler"):
if not conn.tts:
return
@@ -160,7 +178,32 @@ async def wakeupWordsResponse(conn: "ConnectionHandler"):
f.write(wav_bytes)
# 更新配置
wakeup_words_config.update_wakeup_response(voice, file_path, result)
# 覆写文件后 mtime 变化使旧缓存失效,此处立刻预编码新文件,
# 下次唤醒直接命中缓存,编码成本不占用唤醒响应路径
await audio_to_data(file_path, use_cache=True)
finally:
# 确保在任何情况下都释放锁
if _wakeup_response_lock.locked():
_wakeup_response_lock.release()
async def preheat_wakeup_audio_cache():
"""进程启动时预热唤醒词音频编码缓存
把 wakeup_words/{voice}.wav 在 Opus 编码结果一次性算好塞进全局 cache,
后续每条连接上来就能直接命中,省掉 wav→PCM→Opus 的数百毫秒开销。
"""
# 进程启动时 tts 还未初始化,无法拿到 conn.tts.voice,按常见 fallback 列表预热
preheat_voices = ["default", "female-shaonv", "male-qn-jingying"]
fallback_path = "config/assets/wakeup_words_short.wav"
for voice in preheat_voices:
file_path = fallback_path
try:
response = wakeup_words_config.get_wakeup_response(voice)
if response and response.get("file_path"):
file_path = response["file_path"]
await audio_to_data(file_path, use_cache=True)
except Exception as e:
_wakeup_logger.bind(tag=TAG).warning(
f"预热 voice={voice} 失败: {e}"
)
@@ -12,10 +12,8 @@ from core.providers.tts.dto.dto import SentenceType
from core.utils.audioRateController import AudioRateController
TAG = __name__
# 音频帧时长(毫秒)
AUDIO_FRAME_DURATION = 60
# 预缓冲包数量,直接发送以减少延迟
PRE_BUFFER_COUNT = 5
AUDIO_FRAME_DURATION = 60 # 音频帧时长(毫秒),与 Opus 编码 60ms/帧对齐
PRE_BUFFER_COUNT = 2 # 首段发送时不入队的帧数,避免流控延迟拉高首帧到达时间
async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text, sentence_id=None):
@@ -62,11 +62,6 @@ async def handle_mcp_message(
logger.bind(tag=TAG).debug(
f"客户端MCP服务器信息: name={name}, version={version}"
)
await asyncio.sleep(1)
logger.bind(tag=TAG).debug("初始化完成,开始请求MCP工具列表")
await send_mcp_tools_list_request(conn)
return
elif msg_id == 2: # mcpToolsListID
@@ -24,7 +24,7 @@ class ServerMCPManager:
self.config_path = get_project_dir() + "data/.mcp_server_settings.json"
if not os.path.exists(self.config_path):
self.config_path = ""
logger.bind(tag=TAG).warning(
logger.bind(tag=TAG).debug(
f"请检查mcp服务配置文件:data/.mcp_server_settings.json"
)
self.clients: Dict[str, ServerMCPClient] = {}
@@ -5,6 +5,7 @@
import os
import asyncio
import hashlib
import threading
from typing import Dict, Any, TYPE_CHECKING
@@ -120,8 +121,9 @@ class PromptManager:
# 使用传入的提示词并缓存(如果有设备ID)
if device_id:
device_cache_key = f"device_prompt:{device_id}"
self.cache_manager.set(self.CacheType.DEVICE_PROMPT, device_cache_key, user_prompt)
self.cache_manager.set(
self.CacheType.DEVICE_PROMPT, device_cache_key, user_prompt
)
self.logger.bind(tag=TAG).debug(f"设备 {device_id} 的提示词已缓存")
self.logger.bind(tag=TAG).info(f"使用快速提示词: {user_prompt[:50]}...")
@@ -253,6 +255,18 @@ class PromptManager:
if not self.base_prompt_template:
return user_prompt
# 缓存键须覆盖所有渲染输入:emoji 开关、设备 prompt 变更否则会命中过期提示词
emoji_enabled = kwargs.get("emoji_enabled", True)
prompt_hash = hashlib.md5(user_prompt.encode("utf-8")).hexdigest()[:8]
enhanced_cache_key = (
f"enhanced_prompt:{device_id}:{emoji_enabled}:{prompt_hash}"
)
cached_prompt = self.cache_manager.get(
self.CacheType.DEVICE_PROMPT, enhanced_cache_key
)
if cached_prompt is not None:
return cached_prompt
try:
# 获取最新的时间信息(不缓存)
today_date, today_weekday, lunar_date = self._get_current_time_info()
@@ -301,11 +315,13 @@ class PromptManager:
*args,
**kwargs,
)
device_cache_key = f"device_prompt:{device_id}"
self.cache_manager.set(
self.CacheType.DEVICE_PROMPT, device_cache_key, enhanced_prompt
self.CacheType.DEVICE_PROMPT,
enhanced_cache_key,
enhanced_prompt,
ttl=300,
)
self.logger.bind(tag=TAG).info(
self.logger.bind(tag=TAG).debug(
f"构建增强提示词成功,长度: {len(enhanced_prompt)}"
)
return enhanced_prompt
+6 -2
View File
@@ -310,8 +310,12 @@ async def audio_to_data(
from core.utils.cache.manager import cache_manager
from core.utils.cache.config import CacheType
# 生成缓存键,包含文件路径和编码类型
cache_key = f"{audio_file_path}:{is_opus}"
# 缓存键含 mtime:文件被原地覆写时旧缓存自动失效(如唤醒词回复重新 TTS),避免播到过期音频
try:
mtime = os.path.getmtime(audio_file_path)
cache_key = f"{audio_file_path}:{mtime}:{is_opus}"
except OSError:
cache_key = f"{audio_file_path}:{is_opus}"
# 尝试从缓存获取结果
if use_cache: