diff --git a/README.md b/README.md
index 27c8c759..a94c89f2 100644
--- a/README.md
+++ b/README.md
@@ -141,11 +141,10 @@
本项目提供两种部署方式，请根据您的具体需求选择：
#### 🚀 部署方式选择
-
-| 部署方式 | 特点 | 适用场景 | Docker部署文档 | 源码部署文档 | 视频教程 |
+| 部署方式 | 特点 | 适用场景 | 部署文档 | 配置要求 | 视频教程 |
|---------|------|---------|---------|---------|---------|
-| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) | [本地源码只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| - |
-| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) |
+| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker版](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) / [源码部署](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| 如果使用`FunASR`要2核4G，如果全API，要2核2G | - |
+| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker版](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) / [源码部署](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | 如果使用`FunASR`要4核8G，如果全API，要2核4G| [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) |
> 💡 提示：以下是按最新代码部署后的测试平台，有需要可烧录测试，并发为6个，每天会清空数据
@@ -158,9 +157,23 @@ OTA接口地址: https://2662r3426b.vicp.fun/xiaozhi/ota/
Websocket接口地址: wss://2662r3426b.vicp.fun/xiaozhi/v1/
```
+#### 🚩 配置说明和推荐
+> [!Note]
+> 本项目默认的配置是`入门全免费`设置，如果想效果更优，推荐使用`全流式配置`。
+>
+> 本项目自`0.5.2`版本，已支持整个生命周期全流式，相比`0.5`版本以前，响应速度提升约`2.5秒`
+
+
+| 模块名称 | 入门全免费设置 | 全流式配置 |
+|---------|---------|------|
+| ASR(语音识别) | FunASR(本地) | ✅DoubaoASR(火山流式语音识别) |
+| LLM(大模型) | ChatGLMLLM(智谱glm-4-flash) | ✅DoubaoLLM(火山doubao-1-5-pro-32k-250115) |
+| TTS(语音合成) | EdgeTTS(微软语音) | ✅HuoshanDoubleStreamTTS(火山双流式语音合成) |
+| Intent(意图识别) | function_call(函数调用) | ✅function_call(函数调用) |
+| Memory(记忆功能) | mem_local_short(本地短期记忆） | ✅mem_local_short（本地短期记忆） |
+
---
## 功能清单 ✨
-
### 已实现 ✅
| 功能模块 | 描述 |
diff --git a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java
index abac385b..2e3174af 100644
--- a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java
+++ b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java
@@ -227,7 +227,7 @@ public interface Constant {
/**
* 版本号
*/
- public static final String VERSION = "0.5.1";
+ public static final String VERSION = "0.5.2";
/**
* 无效固件URL
diff --git a/main/xiaozhi-server/config/logger.py b/main/xiaozhi-server/config/logger.py
index e5fec630..5f107f5d 100644
--- a/main/xiaozhi-server/config/logger.py
+++ b/main/xiaozhi-server/config/logger.py
@@ -4,7 +4,7 @@ from loguru import logger
from config.config_loader import load_config
from config.settings import check_config_file
-SERVER_VERSION = "0.5.1"
+SERVER_VERSION = "0.5.2"
_logger_initialized = False
@@ -48,7 +48,11 @@ def setup_logging():
# 第一次初始化时配置日志
if not _logger_initialized:
- logger.configure(extra={"selected_module": log_config.get("selected_module", "00000000000000")}) # 新增配置
+ logger.configure(
+ extra={
+ "selected_module": log_config.get("selected_module", "00000000000000")
+ }
+ ) # 新增配置
log_format = log_config.get(
"log_format",
"{time:YYMMDD HH:mm:ss}[{version}_{extra[selected_module]}][{extra[tag]}]-{level}-{message}",
@@ -62,7 +66,9 @@ def setup_logging():
log_format = log_format.replace("{version}", SERVER_VERSION)
log_format = log_format.replace("{selected_module}", selected_module_str)
log_format_file = log_format_file.replace("{version}", SERVER_VERSION)
- log_format_file = log_format_file.replace("{selected_module}", selected_module_str)
+ log_format_file = log_format_file.replace(
+ "{selected_module}", selected_module_str
+ )
log_level = log_config.get("log_level", "INFO")
log_dir = log_config.get("log_dir", "tmp")
@@ -116,15 +122,25 @@ def update_module_string(selected_module_str):
log_format = log_format.replace("{version}", SERVER_VERSION)
log_format = log_format.replace("{selected_module}", selected_module_str)
log_format_file = log_format_file.replace("{version}", SERVER_VERSION)
- log_format_file = log_format_file.replace("{selected_module}", selected_module_str)
+ log_format_file = log_format_file.replace(
+ "{selected_module}", selected_module_str
+ )
logger.remove()
- logger.add(sys.stdout, format=log_format, level=log_config.get("log_level", "INFO"), filter=formatter)
logger.add(
- os.path.join(log_config.get("log_dir", "tmp"), log_config.get("log_file", "server.log")),
+ sys.stdout,
+ format=log_format,
+ level=log_config.get("log_level", "INFO"),
+ filter=formatter,
+ )
+ logger.add(
+ os.path.join(
+ log_config.get("log_dir", "tmp"),
+ log_config.get("log_file", "server.log"),
+ ),
format=log_format_file,
level=log_config.get("log_level", "INFO"),
- filter=formatter
+ filter=formatter,
)
except Exception as e:
diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py
index da0e6b46..e5e47ccc 100644
--- a/main/xiaozhi-server/core/connection.py
+++ b/main/xiaozhi-server/core/connection.py
@@ -84,6 +84,7 @@ class ConnectionHandler:
# 客户端状态相关
self.client_abort = False
+ self.client_is_speaking = False
self.client_listen_mode = "auto"
# 线程任务相关
@@ -624,7 +625,7 @@ class ConnectionHandler:
function_arguments = ""
content_arguments = ""
text_index = 0
-
+ self.client_abort = False
for response in llm_responses:
if self.client_abort:
break
@@ -704,17 +705,23 @@ class ConnectionHandler:
"arguments": function_arguments,
}
- # 处理MCP工具调用
+ # 处理Server端MCP工具调用
if self.mcp_manager.is_mcp_tool(function_name):
result = self._handle_mcp_tool_call(function_call_data)
elif hasattr(self, "mcp_client") and self.mcp_client.has_tool(function_name):
- # 如果是MCP工具调用
+ # 如果是小智端MCP工具调用
self.logger.bind(tag=TAG).debug(
- f"调用MCP工具: {function_name}, 参数: {function_arguments}"
+ f"调用小智端MCP工具: {function_name}, 参数: {function_arguments}"
)
- result = asyncio.run_coroutine_threadsafe(call_mcp_tool(self, self.mcp_client, function_name, function_arguments), self.loop).result()
- self.logger.bind(tag=TAG).debug(f"MCP工具调用结果: {result}")
- result = ActionResponse(action=Action.REQLLM, result=result, response="")
+ try:
+ result = asyncio.run_coroutine_threadsafe(call_mcp_tool(self, self.mcp_client, function_name, function_arguments), self.loop).result()
+ self.logger.bind(tag=TAG).debug(f"MCP工具调用结果: {result}")
+ result = ActionResponse(action=Action.REQLLM, result=result, response="")
+ except Exception as e:
+ self.logger.bind(tag=TAG).error(f"MCP工具调用失败: {e}")
+ result = ActionResponse(
+ action=Action.REQLLM, result="MCP工具调用失败", response=""
+ )
else:
# 处理系统函数
result = self.func_handler.handle_llm_function_call(
@@ -867,6 +874,7 @@ class ConnectionHandler:
self.report_queue.task_done()
def clearSpeakStatus(self):
+ self.client_is_speaking = False
self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态")
async def close(self, ws=None):
diff --git a/main/xiaozhi-server/core/handle/functionHandler.py b/main/xiaozhi-server/core/handle/functionHandler.py
index 5f620880..c5e438ad 100644
--- a/main/xiaozhi-server/core/handle/functionHandler.py
+++ b/main/xiaozhi-server/core/handle/functionHandler.py
@@ -61,7 +61,7 @@ class FunctionHandler:
self.function_registry.register_function("plugin_loader")
self.function_registry.register_function("get_time")
self.function_registry.register_function("get_lunar")
- self.function_registry.register_function("handle_speaker_volume_or_screen_brightness")
+ # self.function_registry.register_function("handle_speaker_volume_or_screen_brightness")
def register_config_functions(self):
"""注册配置中的函数,可以不同客户端使用不同的配置"""
diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py
index a38a33ce..e0c0f4e3 100644
--- a/main/xiaozhi-server/core/handle/helloHandle.py
+++ b/main/xiaozhi-server/core/handle/helloHandle.py
@@ -104,7 +104,13 @@ async def wakeupWordsResponse(conn):
"""唤醒词响应"""
wakeup_word = random.choice(WAKEUP_CONFIG["words"])
- result = conn.llm.response_no_stream(conn.config["prompt"], wakeup_word)
+ question = (
+ "此刻用户正在和你说```"
+ + wakeup_word
+ + "```。\n请你根据以上用户的内容，进行简短回复，文字内容控制在15个字以内。\n"
+ + "请勿对这条内容本身进行任何解释和回应，仅返回对用户的内容的回复。"
+ )
+ result = conn.llm.response_no_stream(conn.config["prompt"], question)
if result is None or result == "":
return
tts_file = await asyncio.to_thread(conn.tts.to_tts, result)
diff --git a/main/xiaozhi-server/core/handle/mcpHandle.py b/main/xiaozhi-server/core/handle/mcpHandle.py
index e03fe72f..4b0e2884 100644
--- a/main/xiaozhi-server/core/handle/mcpHandle.py
+++ b/main/xiaozhi-server/core/handle/mcpHandle.py
@@ -7,33 +7,36 @@ TAG = __name__
class MCPClient:
"""MCPClient，用于管理MCP状态和工具"""
def __init__(self):
- self.tools = []
+ self.tools = {} # Dictionary for O(1) lookup
self.ready = False
self.call_results = {} # To store Futures for tool call responses
self.next_id = 1
self.lock = asyncio.Lock()
+ self._cached_available_tools = None # Cache for get_available_tools
- async def has_tool(self, name: str) -> bool:
- async with self.lock:
- for tool in self.tools:
- if tool["name"] == name:
- return True
- return False
+ def has_tool(self, name: str) -> bool:
+ return name in self.tools
def get_available_tools(self) -> list:
- # async with self.lock:
+ # Check if the cache is valid
+ if self._cached_available_tools is not None:
+ return self._cached_available_tools
+
+ # If cache is not valid, regenerate the list
result = []
- for tool in self.tools:
+ for tool_name, tool_data in self.tools.items():
function_def = {
- "name": tool["name"],
- "description": tool["description"],
+ "name": tool_data["name"],
+ "description": tool_data["description"],
"parameters": {
- "type": tool["inputSchema"].get("type", "object"),
- "properties": tool["inputSchema"].get("properties", {}),
- "required": tool["inputSchema"].get("required", [])
+ "type": tool_data["inputSchema"].get("type", "object"),
+ "properties": tool_data["inputSchema"].get("properties", {}),
+ "required": tool_data["inputSchema"].get("required", [])
}
}
result.append({"type": "function", "function": function_def})
+
+ self._cached_available_tools = result # Store the generated list in cache
return result
async def is_ready(self) -> bool:
@@ -46,8 +49,9 @@ class MCPClient:
async def add_tool(self, tool_data: dict):
async with self.lock:
- self.tools.append(tool_data)
-
+ self.tools[tool_data["name"]] = tool_data
+ self._cached_available_tools = None # Invalidate the cache when a tool is added
+
async def get_next_id(self) -> int:
async with self.lock:
current_id = self.next_id
@@ -232,7 +236,7 @@ async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: str =
if not await mcp_client.is_ready():
raise RuntimeError("MCP客户端尚未准备就绪")
- if not await mcp_client.has_tool(tool_name):
+ if not mcp_client.has_tool(tool_name):
raise ValueError(f"工具 {tool_name} 不存在")
tool_call_id = await mcp_client.get_next_id()
@@ -272,4 +276,4 @@ async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: str =
raise TimeoutError("工具调用请求超时")
except Exception as e:
await mcp_client.cleanup_call_result(tool_call_id)
- raise e
+ raise e
\ No newline at end of file
diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py
index 6776dd0f..b829afb5 100644
--- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py
+++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py
@@ -1,6 +1,7 @@
from core.handle.sendAudioHandle import send_stt_message
from core.handle.intentHandler import handle_user_intent
from core.utils.output_counter import check_device_output_limit
+from core.handle.abortHandle import handleAbortMessage
import time
from core.handle.sendAudioHandle import SentenceType
from core.utils.util import audio_to_data
@@ -12,11 +13,14 @@ async def handleAudioMessage(conn, audio):
if conn.vad is None:
conn.logger.bind(tag=TAG).warning("VAD模块未初始化，继续等待")
return
- if conn.asr is None:
- conn.logger.bind(tag=TAG).warning("ASR模块未初始化，继续等待")
+ if conn.asr is None or not hasattr(conn.asr, "conn") or conn.asr.conn is None:
+ conn.logger.bind(tag=TAG).warning("ASR模块未初始化或通道未就绪，继续等待")
return
# 当前片段是否有人说话
have_voice = conn.vad.is_vad(conn, audio)
+ if have_voice:
+ if conn.client_is_speaking:
+ await handleAbortMessage(conn)
# 设备长时间空闲检测，用于say goodbye
await no_voice_close_connect(conn, have_voice)
# 接收音频
@@ -35,6 +39,8 @@ async def startToChat(conn, text):
):
await max_out_size(conn)
return
+ if conn.client_is_speaking:
+ await handleAbortMessage(conn)
# 首先进行意图分析
intent_handled = await handle_user_intent(conn, text)
@@ -72,7 +78,7 @@ async def no_voice_close_connect(conn, have_voice):
return
prompt = end_prompt.get("prompt")
if not prompt:
- prompt = "请你以“时间过得真快”未来头，用富有感情、依依不舍的话来结束这场对话吧。！"
+ prompt = "请你以```时间过得真快```未来头，用富有感情、依依不舍的话来结束这场对话吧。！"
await startToChat(conn, prompt)
diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py
index 6625838c..85a71c8e 100644
--- a/main/xiaozhi-server/core/handle/sendAudioHandle.py
+++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py
@@ -62,6 +62,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
# 发送结束消息（如果是最后一个文本）
if conn.llm_finish_task and sentenceType == SentenceType.LAST:
await send_tts_message(conn, "stop", None)
+ conn.client_is_speaking = False
if conn.close_after_chat:
await conn.close()
@@ -88,6 +89,7 @@ async def sendAudio(conn, audios, pre_buffer=True):
# 播放剩余音频帧
for opus_packet in remaining_audios:
if conn.client_abort:
+ conn.client_abort = False
return
# 每分钟重置一次计时器
@@ -141,4 +143,5 @@ async def send_stt_message(conn, text):
await conn.websocket.send(
json.dumps({"type": "stt", "text": stt_text, "session_id": conn.session_id})
)
+ conn.client_is_speaking = True
await send_tts_message(conn, "start")
diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py
index ace995e4..0fbae11e 100644
--- a/main/xiaozhi-server/core/handle/textHandle.py
+++ b/main/xiaozhi-server/core/handle/textHandle.py
@@ -59,6 +59,7 @@ async def handleTextMessage(conn, message):
# 如果是唤醒词，且关闭了唤醒词回复，就不用回答
await send_stt_message(conn, original_text)
await send_tts_message(conn, "stop", None)
+ conn.client_is_speaking = False
elif is_wakeup_words:
# 上报纯文字数据（复用ASR上报功能，但不提供音频数据）
enqueue_asr_report(conn, "嘿，你好呀", [])
diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py
index 1a3a384d..707f80bb 100644
--- a/main/xiaozhi-server/core/providers/asr/base.py
+++ b/main/xiaozhi-server/core/providers/asr/base.py
@@ -47,7 +47,7 @@ class ASRProviderBase(ABC):
if self.conn.client_voice_stop:
asr_audio_task = copy.deepcopy(self.conn.asr_audio)
self.conn.asr_audio.clear()
- self.conn.client_abort = False
+
# 音频太短了，无法识别
self.conn.reset_vad_states()
if len(asr_audio_task) > 15:
diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py
index 9ef1d411..da4098ad 100644
--- a/main/xiaozhi-server/core/providers/asr/doubao.py
+++ b/main/xiaozhi-server/core/providers/asr/doubao.py
@@ -33,6 +33,21 @@ class ASRProvider(ASRProviderBase):
self.max_retries = 3
self.retry_delay = 2 # 重试延迟秒数
self.recv_lock = asyncio.Lock() # 添加接收锁
+ self.reconnect_lock = asyncio.Lock() # 添加重连锁
+ self.last_reconnect_time = 0 # 上次重连时间
+ self.reconnect_cooldown = 1 # 增加重连冷却时间到10秒
+ self.reconnect_count = 0 # 当前重连次数
+ self.max_reconnect_count = 3 # 减少最大重连次数到3次
+ self.asr_thread = None # ASR监听线程
+ self.thread_lock = threading.Lock() # 线程管理锁
+ self.is_reconnecting = False # 添加重连状态标志
+
+ # 添加会话管理相关属性
+ self._session_lock = asyncio.Lock() # 会话操作的并发锁
+ self._current_session_id = None # 当前会话ID
+ self._session_started = False # 会话是否已开始
+ self._session_finished = False # 会话是否已结束
+ self._session_close_event = asyncio.Event() # 添加会话关闭事件
self.appid = str(config.get("appid"))
self.cluster = config.get("cluster")
@@ -60,7 +75,6 @@ class ASRProvider(ASRProviderBase):
self.asr_ws = None
self.forward_task = None
self.conn = None
- self.asr_thread = None
###################################################################################
# 豆包流式ASR重写父类的方法--开始
@@ -68,10 +82,18 @@ class ASRProvider(ASRProviderBase):
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
- retry_count = 0
- while retry_count < self.max_retries:
- try:
- # 确保关闭旧的连接
+ async with self._session_lock:
+ # 如果正在重连，等待重连完成
+ if self.is_reconnecting:
+ logger.bind(tag=TAG).info("等待当前重连完成...")
+ await self._session_close_event.wait()
+ self._session_close_event.clear()
+
+ # 如果已有会话未结束，先关闭它
+ if self._session_started and not self._session_finished:
+ logger.bind(tag=TAG).warning(
+ f"发现未关闭的会话 {self._current_session_id}，正在关闭..."
+ )
if self.asr_ws is not None:
try:
await self.asr_ws.close()
@@ -79,62 +101,94 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).warning(f"关闭旧连接时发生错误: {e}")
finally:
self.asr_ws = None
+ self._session_finished = True
+ self._session_close_event.set()
- headers = self.token_auth() if self.auth_method == "token" else None
- self.asr_ws = await websockets.connect(
- self.ws_url,
- additional_headers=headers,
- max_size=1000000000,
- ping_interval=None, # 禁用ping，因为服务器可能不支持
- ping_timeout=None,
- close_timeout=10,
- )
+ # 重置会话状态
+ self._current_session_id = str(uuid.uuid4())
+ self._session_started = True
+ self._session_finished = False
+ self.is_reconnecting = True
- # 发送初始化请求
- request_params = self.construct_request(str(uuid.uuid4()))
- try:
- payload_bytes = str.encode(json.dumps(request_params))
- payload_bytes = gzip.compress(payload_bytes)
- full_client_request = self.generate_header()
- full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
- full_client_request.extend(payload_bytes)
- await self.asr_ws.send(full_client_request)
- logger.bind(tag=TAG).debug(f"发送初始化请求: {request_params}")
- except Exception as e:
- logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}")
- raise e
+ try:
+ retry_count = 0
+ while retry_count < self.max_retries:
+ try:
+ headers = (
+ self.token_auth() if self.auth_method == "token" else None
+ )
+ self.asr_ws = await websockets.connect(
+ self.ws_url,
+ additional_headers=headers,
+ max_size=1000000000,
+ ping_interval=None,
+ ping_timeout=None,
+ close_timeout=10,
+ )
- # 等待初始化响应
- try:
- init_res = await self.asr_ws.recv()
- result = self.parse_response(init_res)
- logger.bind(tag=TAG).info(f"ASR服务初始化响应: {result}")
- except Exception as e:
- logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
- raise e
+ # 发送初始化请求
+ request_params = self.construct_request(
+ self._current_session_id
+ )
+ try:
+ payload_bytes = str.encode(json.dumps(request_params))
+ payload_bytes = gzip.compress(payload_bytes)
+ full_client_request = self.generate_header()
+ full_client_request.extend(
+ (len(payload_bytes)).to_bytes(4, "big")
+ )
+ full_client_request.extend(payload_bytes)
+ await self.asr_ws.send(full_client_request)
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}")
+ raise e
- # 启动接收ASR结果的异步任务
- asr_priority = threading.Thread(
- target=self._start_monitor_asr_response_thread, daemon=True
- )
- asr_priority.start()
- return
+ # 等待初始化响应
+ try:
+ init_res = await self.asr_ws.recv()
+ self.parse_response(init_res)
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
+ raise e
- except websockets.exceptions.WebSocketException as e:
- retry_count += 1
- if retry_count < self.max_retries:
- logger.bind(tag=TAG).warning(
- f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}"
- )
- await asyncio.sleep(self.retry_delay)
- else:
- logger.bind(tag=TAG).error(
- f"WebSocket连接失败，已达到最大重试次数: {e}"
- )
- raise
- except Exception as e:
- logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}")
- raise
+ # 启动接收ASR结果的异步任务
+ with self.thread_lock:
+ if (
+ self.asr_thread is None
+ or not self.asr_thread.is_alive()
+ ):
+ logger.bind(tag=TAG).info("创建新的ASR监听线程...")
+ self.asr_thread = threading.Thread(
+ target=self._start_monitor_asr_response_thread,
+ daemon=True,
+ )
+ self.asr_thread.start()
+ # 等待一小段时间确保线程启动
+ await asyncio.sleep(0.1)
+ if not self.asr_thread.is_alive():
+ logger.bind(tag=TAG).error("ASR监听线程启动失败")
+ raise Exception("ASR监听线程启动失败")
+ logger.bind(tag=TAG).info("ASR监听线程已启动")
+ return
+
+ except websockets.exceptions.WebSocketException as e:
+ retry_count += 1
+ if retry_count < self.max_retries:
+ logger.bind(tag=TAG).warning(
+ f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}"
+ )
+ await asyncio.sleep(self.retry_delay)
+ else:
+ logger.bind(tag=TAG).warning(
+ f"WebSocket连接失败，已达到最大重试次数: {e}"
+ )
+ raise
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}")
+ raise
+ finally:
+ self.is_reconnecting = False
+ self._session_close_event.set()
async def receive_audio(self, audio, _):
if not isinstance(audio, bytes):
@@ -150,7 +204,7 @@ class ASRProvider(ASRProviderBase):
if self.asr_ws:
await self.asr_ws.send(audio_request)
except Exception as e:
- logger.bind(tag=TAG).error(f"发送音频数据时发生错误: {e}")
+ logger.bind(tag=TAG).debug(f"发送音频数据时发生错误: {e}")
###################################################################################
# 豆包流式ASR重写父类的方法--结束
@@ -246,23 +300,61 @@ class ASRProvider(ASRProviderBase):
def _start_monitor_asr_response_thread(self):
# 初始化链接
- asyncio.run_coroutine_threadsafe(
- self._forward_asr_results(), loop=self.conn.loop
- )
+ try:
+ with self.thread_lock:
+ if self.conn is None or self.conn.loop is None:
+ logger.bind(tag=TAG).error(
+ "无法启动ASR监听线程：conn或loop未初始化"
+ )
+ return
+
+ try:
+ logger.bind(tag=TAG).info("开始启动ASR监听...")
+ asyncio.run_coroutine_threadsafe(
+ self._forward_asr_results(), loop=self.conn.loop
+ )
+ logger.bind(tag=TAG).info("ASR监听已启动")
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"启动ASR监听线程失败: {e}")
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"ASR监听线程发生未预期的错误: {e}")
async def _forward_asr_results(self):
try:
while not self.conn.stop_event.is_set():
try:
if self.asr_ws is None:
- logger.bind(tag=TAG).info("尝试重新连接ASR服务...")
- await self.open_audio_channels(self.conn)
- continue
+ # 检查是否需要重连
+ async with self.reconnect_lock:
+ current_time = asyncio.get_event_loop().time()
+ if (
+ current_time - self.last_reconnect_time
+ < self.reconnect_cooldown
+ ):
+ await asyncio.sleep(1)
+ continue
+
+ if self.reconnect_count >= self.max_reconnect_count:
+ logger.bind(tag=TAG).error(
+ "达到最大重连次数限制，停止重连"
+ )
+ await asyncio.sleep(self.reconnect_cooldown)
+ self.reconnect_count = 0
+ continue
+
+ self.last_reconnect_time = current_time
+ self.reconnect_count += 1
+ logger.bind(tag=TAG).info(
+ f"尝试重新连接ASR服务... (第{self.reconnect_count}次)"
+ )
+ await self.open_audio_channels(self.conn)
+ continue
# 使用锁来确保同一时间只有一个协程在接收数据
async with self.recv_lock:
response = await self.asr_ws.recv()
result = self.parse_response(response)
+
# 检查是否需要重连
if result.get("need_reconnect", False):
logger.bind(tag=TAG).info(
@@ -290,6 +382,7 @@ class ASRProvider(ASRProviderBase):
self.text = utterance["text"]
await self.handle_voice_stop(None)
break
+
except websockets.ConnectionClosed:
logger.bind(tag=TAG).debug("ASR服务连接已关闭，准备重连...")
# 确保关闭旧连接
@@ -301,34 +394,16 @@ class ASRProvider(ASRProviderBase):
finally:
self.asr_ws = None
- retry_count = 0
- while (
- retry_count < self.max_retries
- and not self.conn.stop_event.is_set()
- ):
- try:
- logger.bind(tag=TAG).info(
- f"正在进行第{retry_count + 1}次重连尝试..."
- )
- await self.open_audio_channels(self.conn)
- break
- except Exception as e:
- retry_count += 1
- if retry_count < self.max_retries:
- logger.bind(tag=TAG).warning(
- f"重连失败，等待{self.retry_delay}秒后重试: {e}"
- )
- await asyncio.sleep(self.retry_delay)
- else:
- logger.bind(tag=TAG).error(
- f"重连失败，已达到最大重试次数: {e}"
- )
- await asyncio.sleep(
- self.retry_delay
- ) # 继续等待，以便后续重试
+ # 等待冷却时间
+ await asyncio.sleep(self.reconnect_cooldown)
+ continue
+
except Exception as e:
if not self.conn.stop_event.is_set():
- await asyncio.sleep(2) # 增加重试延迟
+ logger.bind(tag=TAG).error(f"ASR监听发生错误: {e}")
+ await asyncio.sleep(self.retry_delay)
+ continue
+
except Exception as e:
logger.bind(tag=TAG).error(f"ASR监听线程发生错误: {e}")
# 确保在发生严重错误时也能继续尝试重连
@@ -422,8 +497,38 @@ class ASRProvider(ASRProviderBase):
f"ASR错误: {error_message} (错误码: {error_code})"
)
- # 如果是识别相关错误（>=1020），标记需要重连
- if error_code >= 1020:
+ # 如果是识别相关错误，标记需要重连
+ if error_code >= 1020 or error_code == 1001:
result["need_reconnect"] = True
return result
+
+ async def close_session(self):
+ """关闭当前会话"""
+ async with self._session_lock:
+ if not self._session_started:
+ logger.bind(tag=TAG).warning("尝试关闭未开始的会话")
+ return
+
+ if self._session_finished:
+ logger.bind(tag=TAG).warning(
+ f"会话 {self._current_session_id} 已经关闭"
+ )
+ return
+
+ try:
+ if self.asr_ws is not None:
+ await self.asr_ws.close()
+ except Exception as e:
+ logger.bind(tag=TAG).warning(f"关闭WebSocket连接时发生错误: {e}")
+ finally:
+ self.asr_ws = None
+ self._session_finished = True
+ self._session_started = False
+ self._current_session_id = None
+ # 重置重连计数
+ self.reconnect_count = 0
+
+ async def close(self):
+ """资源清理方法"""
+ await self.close_session()
diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py
index 0b17eef5..36b6ecbf 100644
--- a/main/xiaozhi-server/core/providers/tts/base.py
+++ b/main/xiaozhi-server/core/providers/tts/base.py
@@ -216,6 +216,7 @@ class TTSProviderBase(ABC):
logger.bind(tag=TAG).error(
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
+ continue
def _audio_play_priority_thread(self):
while not self.conn.stop_event.is_set():
diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
index 25548958..de50c18c 100644
--- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
@@ -6,6 +6,7 @@ import asyncio
import threading
import traceback
import websockets
+import time
from config.logger import setup_logging
from core.utils import opus_encoder_utils
from core.utils.util import check_model_key
@@ -138,6 +139,7 @@ class Response:
class TTSProvider(TTSProviderBase):
def __init__(self, config, delete_audio_file):
super().__init__(config, delete_audio_file)
+ self.ws = None # 初始化ws属性
self.interface_type = InterfaceType.DUAL_STREAM
self.appId = config.get("appid")
self.access_token = config.get("access_token")
@@ -166,59 +168,133 @@ class TTSProvider(TTSProviderBase):
self._current_session_id = None # 当前会话ID
self._session_started = False # 会话是否已开始
self._session_finished = False # 会话是否已结束
+ self._connection_ready = False # 连接是否就绪
+ self._reconnect_attempts = 0 # 重连尝试次数
+ self._max_reconnect_attempts = 3 # 最大重连次数
###################################################################################
# 火山双流式TTS重写父类的方法--开始
###################################################################################
async def open_audio_channels(self, conn):
- await super().open_audio_channels(conn)
- ws_header = {
- "X-Api-App-Key": self.appId,
- "X-Api-Access-Key": self.access_token,
- "X-Api-Resource-Id": self.resource_id,
- "X-Api-Connect-Id": uuid.uuid4(),
- }
- self.ws = await websockets.connect(
- self.ws_url, additional_headers=ws_header, max_size=1000000000
- )
- tts_priority = threading.Thread(
- target=self._start_monitor_tts_response_thread, daemon=True
- )
- tts_priority.start()
+ try:
+ await super().open_audio_channels(conn)
+ await self._ensure_connection()
+ tts_priority = threading.Thread(
+ target=self._start_monitor_tts_response_thread, daemon=True
+ )
+ tts_priority.start()
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"Failed to open audio channels: {str(e)}")
+ self.ws = None
+ raise
+
+ async def _ensure_connection(self):
+ """确保WebSocket连接可用"""
+ try:
+ if self.ws is None:
+ logger.bind(tag=TAG).info("WebSocket连接不存在，开始建立新连接...")
+ ws_header = {
+ "X-Api-App-Key": self.appId,
+ "X-Api-Access-Key": self.access_token,
+ "X-Api-Resource-Id": self.resource_id,
+ "X-Api-Connect-Id": uuid.uuid4(),
+ }
+ self.ws = await websockets.connect(
+ self.ws_url, additional_headers=ws_header, max_size=1000000000
+ )
+ self._connection_ready = True
+ self._reconnect_attempts = 0
+ logger.bind(tag=TAG).info("WebSocket连接建立成功")
+ else:
+ # 尝试发送ping来检查连接是否还活着
+ try:
+ logger.bind(tag=TAG).debug("检查WebSocket连接状态...")
+ pong_waiter = await self.ws.ping()
+ await asyncio.wait_for(pong_waiter, timeout=1.0)
+ logger.bind(tag=TAG).debug("WebSocket连接状态正常")
+ except (asyncio.TimeoutError, websockets.ConnectionClosed):
+ # 如果ping失败，重新建立连接
+ logger.bind(tag=TAG).warning("WebSocket连接已断开，准备重新连接...")
+ try:
+ await self.ws.close()
+ except:
+ pass
+ self.ws = None
+ self._connection_ready = False
+ # 重新建立连接
+ await self._ensure_connection()
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"确保连接失败: {str(e)}")
+ self._connection_ready = False
+ self.ws = None
+ raise
def tts_text_priority_thread(self):
+ logger.bind(tag=TAG).info("TTS文本处理线程启动")
while not self.conn.stop_event.is_set():
try:
+ logger.bind(tag=TAG).debug("等待TTS文本队列消息...")
message = self.tts_text_queue.get(timeout=1)
- logger.bind(tag=TAG).debug(
- f"TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name}"
+ logger.bind(tag=TAG).info(
+ f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
)
if message.sentence_type == SentenceType.FIRST:
# 初始化参数
- future = asyncio.run_coroutine_threadsafe(
- self.start_session(self.conn.sentence_id), loop=self.conn.loop
- )
- future.result()
- self.tts_audio_first_sentence = True
- self.before_stop_play_files.clear()
- elif ContentType.TEXT == message.content_type:
- if message.content_detail:
+ try:
+ logger.bind(tag=TAG).info("开始启动TTS会话...")
future = asyncio.run_coroutine_threadsafe(
- self.text_to_speak(message.content_detail, None),
+ self.start_session(self.conn.sentence_id),
loop=self.conn.loop,
)
future.result()
+ self.tts_audio_first_sentence = True
+ self.before_stop_play_files.clear()
+ logger.bind(tag=TAG).info("TTS会话启动成功")
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}")
+ # 直接跳过当前消息，不重新入队
+ time.sleep(1)
+ continue
+ elif ContentType.TEXT == message.content_type:
+ if message.content_detail:
+ try:
+ logger.bind(tag=TAG).info(
+ f"开始发送TTS文本: {message.content_detail}"
+ )
+ future = asyncio.run_coroutine_threadsafe(
+ self.text_to_speak(message.content_detail, None),
+ loop=self.conn.loop,
+ )
+ future.result()
+ logger.bind(tag=TAG).info("TTS文本发送成功")
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
+ # 直接跳过当前消息，不重新入队
+ time.sleep(1)
+ continue
elif ContentType.FILE == message.content_type:
+ logger.bind(tag=TAG).info(
+ f"添加音频文件到待播放列表: {message.content_file}"
+ )
self.before_stop_play_files.append(
(message.content_file, message.content_detail)
)
if message.sentence_type == SentenceType.LAST:
- future = asyncio.run_coroutine_threadsafe(
- self.finish_session(self.conn.sentence_id), loop=self.conn.loop
- )
- future.result()
+ try:
+ logger.bind(tag=TAG).info("开始结束TTS会话...")
+ future = asyncio.run_coroutine_threadsafe(
+ self.finish_session(self.conn.sentence_id),
+ loop=self.conn.loop,
+ )
+ future.result()
+ logger.bind(tag=TAG).info("TTS会话结束成功")
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
+ # 直接跳过当前消息，不重新入队
+ time.sleep(1)
+ continue
except queue.Empty:
continue
@@ -226,11 +302,30 @@ class TTSProvider(TTSProviderBase):
logger.bind(tag=TAG).error(
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
+ # 如果是WebSocket连接关闭错误，等待一段时间后继续
+ if "non-exist session" in str(e):
+ time.sleep(1)
+ continue
async def text_to_speak(self, text, _):
- # 发送文本
- await self.send_text(self.speaker, text, self.conn.sentence_id)
- return
+ """发送文本到TTS服务"""
+ try:
+ # 确保WebSocket连接可用
+ if not self._connection_ready or self.ws is None:
+ logger.bind(tag=TAG).warning("WebSocket连接不可用，尝试重新连接...")
+ await self._ensure_connection()
+
+ # 发送文本
+ await self.send_text(self.speaker, text, self.conn.sentence_id)
+ return
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
+ # 如果是连接问题，尝试重新连接
+ if isinstance(e, websockets.ConnectionClosed):
+ self._connection_ready = False
+ self.ws = None
+ await self._handle_connection_error()
+ raise
###################################################################################
# 火山双流式TTS重写父类的方法--结束
@@ -308,14 +403,21 @@ class TTSProvider(TTSProviderBase):
async def send_event(
self, header: bytes, optional: bytes | None = None, payload: bytes = None
):
- full_client_request = bytearray(header)
- if optional is not None:
- full_client_request.extend(optional)
- if payload is not None:
- payload_size = len(payload).to_bytes(4, "big", signed=True)
- full_client_request.extend(payload_size)
- full_client_request.extend(payload)
- await self.ws.send(full_client_request)
+ try:
+ full_client_request = bytearray(header)
+ if optional is not None:
+ full_client_request.extend(optional)
+ if payload is not None:
+ payload_size = len(payload).to_bytes(4, "big", signed=True)
+ full_client_request.extend(payload_size)
+ full_client_request.extend(payload)
+ await self.ws.send(full_client_request)
+ except websockets.ConnectionClosed:
+ if await self._handle_connection_error():
+ # 重连成功后重试发送
+ await self.ws.send(full_client_request)
+ else:
+ raise
async def send_text(self, speaker: str, text: str, session_id):
header = Header(
@@ -450,63 +552,154 @@ class TTSProvider(TTSProviderBase):
return
async def start_session(self, session_id):
- logger.bind(tag=TAG).debug(f"开始会话～～{session_id}")
- async with self._session_lock:
- # 如果已有会话未结束，先关闭它
- if self._session_started and not self._session_finished:
- logger.bind(tag=TAG).warning(
- f"发现未关闭的会话 {self._current_session_id}，正在关闭..."
- )
- await self.finish_session(self._current_session_id)
+ logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
+ try:
+ async with self._session_lock:
+ try:
+ # 确保连接可用
+ logger.bind(tag=TAG).info("检查WebSocket连接状态...")
+ await asyncio.wait_for(self._ensure_connection(), timeout=5)
- # 重置会话状态
- self._current_session_id = session_id
- self._session_started = True
- self._session_finished = False
+ # 如果已有会话未结束，先关闭它
+ if self._session_started and not self._session_finished:
+ logger.bind(tag=TAG).warning(
+ f"发现未关闭的会话 {self._current_session_id}，正在关闭..."
+ )
+ try:
+ await asyncio.wait_for(
+ self.finish_session(self._current_session_id), timeout=5
+ )
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"关闭旧会话失败: {str(e)}")
+ # 强制重置会话状态
+ self._session_started = False
+ self._session_finished = True
+ self._current_session_id = None
- header = Header(
- message_type=FULL_CLIENT_REQUEST,
- message_type_specific_flags=MsgTypeFlagWithEvent,
- serial_method=JSON,
- ).as_bytes()
- optional = Optional(
- event=EVENT_StartSession, sessionId=session_id
- ).as_bytes()
- payload = self.get_payload_bytes(
- event=EVENT_StartSession, speaker=self.speaker
- )
- await self.send_event(header, optional, payload)
+ # 重置会话状态
+ self._current_session_id = session_id
+ self._session_started = True
+ self._session_finished = False
+ logger.bind(tag=TAG).info(
+ f"会话状态已更新 - 开始: {self._session_started}, 结束: {self._session_finished}"
+ )
+
+ header = Header(
+ message_type=FULL_CLIENT_REQUEST,
+ message_type_specific_flags=MsgTypeFlagWithEvent,
+ serial_method=JSON,
+ ).as_bytes()
+ optional = Optional(
+ event=EVENT_StartSession, sessionId=session_id
+ ).as_bytes()
+ payload = self.get_payload_bytes(
+ event=EVENT_StartSession, speaker=self.speaker
+ )
+ await asyncio.wait_for(
+ self.send_event(header, optional, payload), timeout=5
+ )
+ logger.bind(tag=TAG).info("会话启动请求已发送")
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
+ self._session_started = False
+ self._session_finished = True
+ self._current_session_id = None
+ raise
+ except asyncio.TimeoutError:
+ logger.bind(tag=TAG).error(f"启动会话超时: {session_id}")
+ # 超时后强制重置会话状态
+ self._session_started = False
+ self._session_finished = True
+ self._current_session_id = None
+ # 尝试关闭WebSocket连接
+ if self.ws:
+ try:
+ await self.ws.close()
+ except:
+ pass
+ self.ws = None
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"启动会话时发生未知错误: {str(e)}")
+ # 发生未知错误时也重置会话状态
+ self._session_started = False
+ self._session_finished = True
+ self._current_session_id = None
async def finish_session(self, session_id):
- logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}")
- async with self._session_lock:
- # 检查会话状态
- if not self._session_started:
- logger.bind(tag=TAG).warning(f"尝试关闭未开始的会话 {session_id}")
- return
+ logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
+ try:
+ async with self._session_lock:
+ try:
+ # 检查会话状态
+ if not self._session_started:
+ logger.bind(tag=TAG).warning(
+ f"尝试关闭未开始的会话 {session_id}"
+ )
+ return
- if self._session_finished:
- logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭")
- return
+ if self._session_finished:
+ logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭")
+ return
- if self._current_session_id != session_id:
- logger.bind(tag=TAG).warning(
- f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}"
- )
- return
+ if self._current_session_id != session_id:
+ logger.bind(tag=TAG).warning(
+ f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}"
+ )
+ # 即使会话ID不匹配，也尝试关闭当前会话
+ if self._current_session_id:
+ session_id = self._current_session_id
- header = Header(
- message_type=FULL_CLIENT_REQUEST,
- message_type_specific_flags=MsgTypeFlagWithEvent,
- serial_method=JSON,
- ).as_bytes()
- optional = Optional(
- event=EVENT_FinishSession, sessionId=session_id
- ).as_bytes()
- payload = str.encode("{}")
- await self.send_event(header, optional, payload)
+ # 确保WebSocket连接可用
+ if self.ws is None:
+ logger.bind(tag=TAG).warning(
+ "WebSocket连接不存在，尝试重新连接..."
+ )
+ await asyncio.wait_for(self._ensure_connection(), timeout=5)
- # 更新会话状态
+ header = Header(
+ message_type=FULL_CLIENT_REQUEST,
+ message_type_specific_flags=MsgTypeFlagWithEvent,
+ serial_method=JSON,
+ ).as_bytes()
+ optional = Optional(
+ event=EVENT_FinishSession, sessionId=session_id
+ ).as_bytes()
+ payload = str.encode("{}")
+ await asyncio.wait_for(
+ self.send_event(header, optional, payload), timeout=5
+ )
+ logger.bind(tag=TAG).info("会话结束请求已发送")
+
+ # 更新会话状态
+ self._session_finished = True
+ self._session_started = False
+ self._current_session_id = None
+ logger.bind(tag=TAG).info(
+ "会话状态已更新 - 开始: False, 结束: True"
+ )
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
+ # 即使发生错误，也要重置会话状态
+ self._session_finished = True
+ self._session_started = False
+ self._current_session_id = None
+ raise
+ except asyncio.TimeoutError:
+ logger.bind(tag=TAG).error(f"关闭会话超时: {session_id}")
+ # 超时后强制重置会话状态
+ self._session_finished = True
+ self._session_started = False
+ self._current_session_id = None
+ # 尝试关闭WebSocket连接
+ if self.ws:
+ try:
+ await self.ws.close()
+ except:
+ pass
+ self.ws = None
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"关闭会话时发生未知错误: {str(e)}")
+ # 发生未知错误时也重置会话状态
self._session_finished = True
self._session_started = False
self._current_session_id = None
@@ -527,3 +720,20 @@ class TTSProvider(TTSProviderBase):
def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False):
opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end)
return opus_datas
+
+ async def _handle_connection_error(self):
+ """处理连接错误"""
+ if self._reconnect_attempts < self._max_reconnect_attempts:
+ self._reconnect_attempts += 1
+ logger.bind(tag=TAG).warning(
+ f"尝试重新连接 (第{self._reconnect_attempts}次)"
+ )
+ try:
+ await self._ensure_connection()
+ return True
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"重新连接失败: {str(e)}")
+ return False
+ else:
+ logger.bind(tag=TAG).error("达到最大重连次数，放弃重连")
+ return False
diff --git a/main/xiaozhi-server/test/test_page.html b/main/xiaozhi-server/test/test_page.html
index fbad025d..567332b8 100644
--- a/main/xiaozhi-server/test/test_page.html
+++ b/main/xiaozhi-server/test/test_page.html
@@ -1484,6 +1484,23 @@
if (message.text && message.text !== '😊') {
addMessage(message.text);
}
+ }else if (message.type === 'mcp') {
+ const payload = message.payload || {};
+ log(`服务器下发: ${JSON.stringify(message)}`, 'info');
+ if (payload) {
+ // 模拟小智客户端行为
+ if(payload.method === 'tools/list'){
+ const replay_message = JSON.stringify({"session_id":"","type":"mcp","payload":{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"self.get_device_status","description":"Provides the real-time information of the device, including the current status of the audio speaker, screen, battery, network, etc.\nUse this tool for: \n1. Answering questions about current condition (e.g. what is the current volume of the audio speaker?)\n2. As the first step to control the device (e.g. turn up / down the volume of the audio speaker, etc.)","inputSchema":{"type":"object","properties":{}}},{"name":"self.audio_speaker.set_volume","description":"Set the volume of the audio speaker. If the current volume is unknown, you must call `self.get_device_status` tool first and then call this tool.","inputSchema":{"type":"object","properties":{"volume":{"type":"integer","minimum":0,"maximum":100}},"required":["volume"]}},{"name":"self.screen.set_brightness","description":"Set the brightness of the screen.","inputSchema":{"type":"object","properties":{"brightness":{"type":"integer","minimum":0,"maximum":100}},"required":["brightness"]}},{"name":"self.screen.set_theme","description":"Set the theme of the screen. The theme can be 'light' or 'dark'.","inputSchema":{"type":"object","properties":{"theme":{"type":"string"}},"required":["theme"]}}]}}})
+ websocket.send(replay_message);
+ log(`回复MCP消息: ${replay_message}`, 'info');
+ } else if(payload.method === 'tools/call'){
+ // 模拟回复
+ const replay_message = JSON.stringify({"session_id":"9f261599","type":"mcp","payload":{"jsonrpc":"2.0","id": payload.id,"result":{"content":[{"type":"text","text":"true"}],"isError":false}}})
+ websocket.send(replay_message);
+ log(`回复MCP消息: ${replay_message}`, 'info');
+ }
+ }
+
} else {
// 未知消息类型
log(`未知消息类型: ${message.type}`, 'info');
@@ -1523,7 +1540,10 @@
device_id: config.deviceId,
device_name: config.deviceName,
device_mac: config.deviceMac,
- token: config.token
+ token: config.token,
+ features: {
+ mcp: true
+ }
};
log('发送hello握手消息', 'info');