import os
|
import sys
|
import copy
|
import json
|
import re
|
import uuid
|
import time
|
import queue
|
import asyncio
|
import threading
|
import traceback
|
import subprocess
|
import websockets
|
import opuslib_next
|
import numpy as np
|
|
from core.utils.util import (
|
extract_json_from_string,
|
check_vad_update,
|
check_asr_update,
|
filter_sensitive_info,
|
)
|
from typing import Dict, Any
|
from collections import deque
|
from core.utils.modules_initialize import (
|
initialize_modules,
|
initialize_tts,
|
initialize_asr,
|
)
|
from core.handle.reportHandle import report, enqueue_tool_report
|
from core.providers.tts.default import DefaultTTS
|
from concurrent.futures import ThreadPoolExecutor
|
from core.utils.dialogue import Message, Dialogue
|
from core.providers.asr.dto.dto import InterfaceType
|
from core.handle.textHandle import handleTextMessage
|
from core.providers.tools.unified_tool_handler import UnifiedToolHandler
|
from plugins_func.loadplugins import auto_import_modules
|
from plugins_func.register import Action, ActionResponse, all_function_registry, module_func_map
|
from core.auth import AuthenticationError
|
from config.config_loader import get_private_config_from_api
|
from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType
|
from config.logger import setup_logging, build_module_string, create_connection_logger
|
from config.manage_api_client import DeviceNotFoundException, DeviceBindException, generate_and_save_chat_title
|
from core.utils.prompt_manager import PromptManager
|
from core.utils.voiceprint_provider import VoiceprintProvider
|
from core.utils.util import get_system_error_response
|
from core.utils import textUtils
|
from core.websocket_handshake import is_client_hello, is_priority_listen_detect
|
|
|
TAG = __name__
|
|
auto_import_modules("plugins_func.functions")
|
|
|
class TTSException(RuntimeError):
|
pass
|
|
# direct_answer 虚拟工具定义
|
# 不是真实工具，是路由机制：将"调不调工具"的二选一变为"调哪个"的多选，防止小模型误触发真实工具
|
DIRECT_ANSWER_TOOL = {
|
"type": "function",
|
"function": {
|
"name": "direct_answer",
|
"description": "当用户的请求不匹配其他任何工具时，可用此选项直接回复。将回复内容写在response参数里。",
|
"parameters": {
|
"type": "object",
|
"properties": {
|
"response": {
|
"type": "string",
|
"description": "你回复用户的完整内容",
|
},
|
},
|
"required": ["response"],
|
},
|
},
|
}
|
|
|
class ConnectionHandler:
|
def __init__(
|
self,
|
config: Dict[str, Any],
|
_vad,
|
_asr,
|
_llm,
|
_memory,
|
_intent,
|
server=None,
|
):
|
self.common_config = config
|
self.config = copy.deepcopy(config)
|
self.session_id = str(uuid.uuid4())
|
self.logger = setup_logging()
|
self.server = server # 保存server实例的引用
|
|
self.need_bind = False # 是否需要绑定设备
|
self.bind_completed_event = asyncio.Event()
|
self.bind_code = None # 绑定设备的验证码
|
self.last_bind_prompt_time = 0 # 上次播放绑定提示的时间戳(秒)
|
self.bind_prompt_interval = 60 # 绑定提示播放间隔(秒)
|
|
self.read_config_from_api = self.config.get("read_config_from_api", False)
|
|
self.websocket: websockets.ServerConnection | None = None
|
self.headers = None
|
self.device_id = None
|
self.client_ip = None
|
self.prompt = None
|
self.welcome_msg = None
|
self.max_output_size = 0
|
self.chat_history_conf = 0
|
self.audio_format = "opus"
|
self.sample_rate = 24000 # 默认采样率，从客户端 hello 消息中动态更新
|
|
# 客户端状态相关
|
self.client_abort = False
|
self.client_is_speaking = False
|
self.client_listen_mode = "auto"
|
self.client_aec = False # 是否启用了服务端AEC
|
|
# 线程任务相关
|
self.loop = None # 在 handle_connection 中获取运行中的事件循环
|
self.stop_event = threading.Event()
|
self.executor = ThreadPoolExecutor(max_workers=5)
|
self._tts_channel_lock = threading.Lock()
|
|
# 添加上报线程池
|
self.report_queue = queue.Queue()
|
self.report_thread = None
|
# 未来可以通过修改此处，调节asr的上报和tts的上报，目前默认都开启
|
self.report_asr_enable = self.read_config_from_api
|
self.report_tts_enable = self.read_config_from_api
|
|
# 依赖的组件
|
self.vad = None
|
self.asr = None
|
self.tts = None
|
self._asr = _asr
|
self._vad = _vad
|
self.llm = _llm
|
self.memory = _memory
|
self.intent = _intent
|
|
# 为每个连接单独管理声纹识别
|
self.voiceprint_provider = None
|
|
# vad相关变量
|
self.client_audio_buffer = bytearray()
|
self.client_have_voice = False
|
self.client_voice_window = deque(maxlen=5)
|
self.first_activity_time = 0.0 # 记录首次活动的时间（毫秒）
|
self.last_activity_time = 0.0 # 统一的活动时间戳（毫秒）
|
self.vad_last_voice_time = 0.0 # 记录用户最后一次说话的时间（毫秒）
|
self.client_voice_stop = False
|
self.last_is_voice = False
|
|
# asr相关变量
|
# 因为实际部署时可能会用到公共的本地ASR，不能把变量暴露给公共ASR
|
# 所以涉及到ASR的变量，需要在这里定义，属于connection的私有变量
|
self.asr_audio = [] # 存储PCM帧列表，供VAD和ASR共享
|
self.asr_audio_queue = queue.Queue()
|
self.current_speaker = None # 存储当前说话人
|
self.introduced_speakers = set() # 已"首次引入"的说话人，控制只在首轮带名字
|
self.system_introduced_speakers = set() # 已在 system 注入过身份的说话人，控制 system 身份只首轮出现
|
|
# llm相关变量
|
self.dialogue = Dialogue()
|
|
# tts相关变量
|
self.sentence_id = None
|
# 处理TTS响应没有文本返回
|
self.tts_MessageText = ""
|
|
# iot相关变量
|
self.iot_descriptors = {}
|
self.func_handler = None
|
|
self.cmd_exit = self.config["exit_commands"]
|
|
# 是否在聊天结束后关闭连接
|
self.close_after_chat = False
|
self.load_function_plugin = False
|
self.intent_type = "nointent"
|
|
self.timeout_seconds = (
|
int(self.config.get("close_connection_no_voice_time", 120)) + 60
|
) # 在原来第一道关闭的基础上加60秒，进行二道关闭
|
self.timeout_task = None
|
|
# {"mcp":true} 表示启用MCP功能
|
self.features = None
|
|
# 标记连接是否来自MQTT
|
self.conn_from_mqtt_gateway = False
|
|
# 初始化提示词管理器
|
self.prompt_manager = PromptManager(self.config, self.logger)
|
|
# 初始化通话状态
|
self.calling = False
|
# 标记当前是否为来电接听模式
|
self.incoming_call = None
|
|
# Phase 1 App Runtime（连接内存，不建表）
|
self.active_app_id = "chat"
|
self.current_task_id = "conversation"
|
self.interaction_mode = "CHAT"
|
self.current_source = "default_chat"
|
self.tool_policy = "inherit_agent"
|
self.tool_allowlist = []
|
self.original_system_prompt = None
|
self.awaiting_voice_app_choice = False
|
self.shadowing_index = 0
|
self.shadowing_prompt = ""
|
self.shadowing_clips = []
|
self.shadowing_awaiting_user = False
|
self.shadowing_clip_gen = 0
|
self.shadowing_timeout_task = None
|
self.shadowing_assess_index = 0
|
self.last_asr_language = None
|
self.task_session = None
|
self.last_asr_audio_snapshot = None
|
self.shadowing_listen_pcm = []
|
self.shadowing_audio_saved_index = -1
|
|
async def handle_connection(self, ws: websockets.ServerConnection):
|
try:
|
# 获取运行中的事件循环（必须在异步上下文中）
|
self.loop = asyncio.get_running_loop()
|
|
# 获取并验证headers
|
self.headers = dict(ws.request.headers)
|
real_ip = self.headers.get("x-real-ip") or self.headers.get(
|
"x-forwarded-for"
|
)
|
if real_ip:
|
self.client_ip = real_ip.split(",")[0].strip()
|
else:
|
self.client_ip = ws.remote_address[0]
|
self.logger.bind(tag=TAG).info(
|
f"{self.client_ip} conn - Headers: {self.headers}"
|
)
|
|
self.device_id = self.headers.get("device-id", None)
|
|
# 认证通过,继续处理
|
self.websocket = ws
|
if self.server is not None:
|
await self.server.register_device_handler(self.device_id, self)
|
|
# 检查是否来自MQTT连接
|
request_path = ws.request.path
|
self.conn_from_mqtt_gateway = request_path.endswith("?from=mqtt_gateway")
|
if self.conn_from_mqtt_gateway:
|
self.logger.bind(tag=TAG).info("连接来自:MQTT网关")
|
|
# 初始化活动时间戳
|
self.first_activity_time = time.time() * 1000
|
self.last_activity_time = time.time() * 1000
|
|
# 启动超时检查任务
|
self.timeout_task = asyncio.create_task(self._check_timeout())
|
|
# 启动AEC缓存清理任务
|
self._aec_cache_cleanup_task = asyncio.create_task(self._check_aec_cache_expiry())
|
|
self.welcome_msg = copy.deepcopy(self.config.get("xiaozhi") or {})
|
self.welcome_msg["session_id"] = self.session_id
|
|
# 从配置中读取采样率
|
self.sample_rate = self.welcome_msg["audio_params"]["sample_rate"]
|
self.logger.bind(tag=TAG).info(f"配置输出音频采样率为: {self.sample_rate}")
|
|
# 在后台初始化配置和组件（完全不阻塞主循环）
|
asyncio.create_task(self._background_initialize())
|
|
try:
|
async for message in self.websocket:
|
await self._route_message(message)
|
except websockets.exceptions.ConnectionClosed:
|
self.logger.bind(tag=TAG).info("客户端断开连接")
|
|
except AuthenticationError as e:
|
self.logger.bind(tag=TAG).error(f"Authentication failed: {str(e)}")
|
return
|
except Exception as e:
|
stack_trace = traceback.format_exc()
|
self.logger.bind(tag=TAG).error(f"Connection error: {str(e)}-{stack_trace}")
|
return
|
finally:
|
try:
|
await self._save_and_close(ws)
|
except Exception as final_error:
|
self.logger.bind(tag=TAG).error(f"最终清理时出错: {final_error}")
|
# 确保即使保存记忆失败，也要关闭连接
|
try:
|
await self.close(ws)
|
except Exception as close_error:
|
self.logger.bind(tag=TAG).error(
|
f"强制关闭连接时出错: {close_error}"
|
)
|
|
async def _save_and_close(self, ws):
|
"""保存记忆并关闭连接"""
|
try:
|
# 守护线程1：独立生成标题（不依赖记忆模型）
|
if self.session_id:
|
def generate_title_task():
|
try:
|
loop = asyncio.new_event_loop()
|
asyncio.set_event_loop(loop)
|
loop.run_until_complete(
|
generate_and_save_chat_title(self.session_id)
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"生成标题失败: {e}")
|
finally:
|
try:
|
loop.close()
|
except Exception:
|
pass
|
|
threading.Thread(target=generate_title_task, daemon=True).start()
|
|
# 守护线程2：走老流程记忆保存（仅记忆，不含标题）
|
if self.memory:
|
# 使用线程池异步保存记忆
|
def save_memory_task():
|
try:
|
# 创建新事件循环（避免与主循环冲突）
|
loop = asyncio.new_event_loop()
|
asyncio.set_event_loop(loop)
|
loop.run_until_complete(
|
self.memory.save_memory(
|
self.dialogue.dialogue, self.session_id
|
)
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"保存记忆失败: {e}")
|
finally:
|
try:
|
loop.close()
|
except Exception:
|
pass
|
|
# 启动线程保存记忆，不等待完成
|
threading.Thread(target=save_memory_task, daemon=True).start()
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"保存记忆失败: {e}")
|
finally:
|
# 立即关闭连接，不等待记忆保存完成
|
try:
|
await self.close(ws)
|
except Exception as close_error:
|
self.logger.bind(tag=TAG).error(
|
f"保存记忆后关闭连接失败: {close_error}"
|
)
|
|
async def _discard_message_with_bind_prompt(self):
|
"""丢弃消息并检查是否需要播放绑定提示"""
|
current_time = time.time()
|
# 检查是否需要播放绑定提示
|
if current_time - self.last_bind_prompt_time >= self.bind_prompt_interval:
|
self.last_bind_prompt_time = current_time
|
# 复用现有的绑定提示逻辑
|
from core.handle.receiveAudioHandle import check_bind_device
|
|
asyncio.create_task(check_bind_device(self))
|
|
async def _route_message(self, message):
|
"""消息路由"""
|
# hello 必须立刻应答。绑定状态还在拉配置时如果丢掉 hello，
|
# 设备会卡在 handshake/hello 等待，表现为连不上语音服务。
|
if is_client_hello(message) or is_priority_listen_detect(message):
|
await handleTextMessage(self, message)
|
return
|
|
# 检查是否已经获取到真实的绑定状态
|
if not self.bind_completed_event.is_set():
|
# 还没有获取到真实状态，等待直到获取到真实状态或超时
|
try:
|
await asyncio.wait_for(self.bind_completed_event.wait(), timeout=1)
|
except asyncio.TimeoutError:
|
# 超时仍未获取到真实状态，丢弃消息
|
await self._discard_message_with_bind_prompt()
|
return
|
|
# 已经获取到真实状态，检查是否需要绑定
|
if self.need_bind:
|
# 需要绑定，丢弃消息
|
await self._discard_message_with_bind_prompt()
|
return
|
|
# 不需要绑定，继续处理消息
|
|
if isinstance(message, str):
|
await handleTextMessage(self, message)
|
elif isinstance(message, bytes):
|
if self.vad is None or self.asr is None:
|
return
|
|
# 处理来自MQTT网关的音频包
|
if self.conn_from_mqtt_gateway and len(message) >= 16:
|
handled = await self._process_mqtt_audio_message(message)
|
if handled:
|
return
|
|
# 入口处直接解码PCM，避免VAD和ASR重复解码
|
pcm_frame = self._decode_opus_packet(message)
|
if pcm_frame:
|
self.asr_audio_queue.put(pcm_frame)
|
|
async def _process_mqtt_audio_message(self, message):
|
"""
|
处理来自MQTT网关的音频消息，解析16字节头部并提取音频数据，在入队前进行AEC处理
|
|
Args:
|
message: 包含头部的音频消息
|
|
Returns:
|
bool: 是否成功处理了消息
|
"""
|
try:
|
# 解析timestamp
|
timestamp = int.from_bytes(message[8:12], "big")
|
|
audio_data = message[16:]
|
# 入口直接解码PCM
|
pcm_frame = self._decode_opus_packet(audio_data)
|
if not pcm_frame:
|
return True
|
|
# AEC处理：如果timestamp>0且启用了AEC
|
if timestamp > 0 and self.client_aec:
|
pcm_frame = self._apply_aec(timestamp, pcm_frame)
|
|
self.asr_audio_queue.put(pcm_frame)
|
return True
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"解析WebSocket音频包失败: {e}")
|
|
# 处理失败，返回False表示需要继续处理
|
return False
|
|
def _apply_aec(self, timestamp: int, pcm_frame: bytes) -> bytes:
|
"""应用AEC处理 - 综合算法：互相关延迟估计 + Wiener滤波 + 频谱减法"""
|
try:
|
if not pcm_frame or len(pcm_frame) == 0:
|
return pcm_frame
|
|
if not hasattr(self, "aec_audio_cache") or not self.aec_audio_cache:
|
return pcm_frame
|
|
mic_audio = np.frombuffer(pcm_frame, dtype=np.int16).astype(np.float32)
|
mic_rms = np.sqrt(np.mean(mic_audio ** 2))
|
|
if mic_rms < 100:
|
return pcm_frame
|
|
sorted_timestamps = sorted(self.aec_audio_cache.keys())
|
if len(sorted_timestamps) < 2:
|
return pcm_frame
|
|
# ========== 匹配参考帧（对数功率谱匹配） ==========
|
n = len(mic_audio)
|
|
# 找最接近的timestamp作为起点
|
closest_idx = min(range(len(sorted_timestamps)), key=lambda i: abs(sorted_timestamps[i] - timestamp))
|
|
# 预计算 mic_audio 的对数功率谱（循环内共用，避免重复FFT）
|
mic_window = np.hanning(n)
|
mic_fft = np.fft.rfft(mic_audio * mic_window)
|
mic_psd = np.abs(mic_fft) ** 2
|
mic_log_psd = 10 * np.log10(mic_psd + 1e-8)
|
mic_P_xx = np.dot(mic_log_psd, mic_log_psd)
|
|
# 用对数功率谱匹配找最佳帧：前后各找2帧
|
best_corr = -1
|
best_ref_idx = closest_idx
|
best_ref_rms = 0.0
|
|
for offset in range(-2, 3): # T-2, T-1, T, T+1, T+2
|
test_idx = closest_idx + offset
|
if test_idx < 0 or test_idx >= len(sorted_timestamps):
|
continue
|
test_ts = sorted_timestamps[test_idx]
|
test_ref = np.frombuffer(self.aec_audio_cache[test_ts], dtype=np.int16).astype(np.float32)
|
test_ref_rms = np.sqrt(np.mean(test_ref ** 2))
|
if test_ref_rms < 50:
|
continue
|
|
# 对数功率谱相关性
|
test_window = np.hanning(len(test_ref))
|
test_fft = np.fft.rfft(test_ref * test_window)
|
test_psd = np.abs(test_fft) ** 2
|
test_log_psd = 10 * np.log10(test_psd + 1e-8)
|
P_xy = np.dot(mic_log_psd, test_log_psd)
|
P_yy = np.dot(test_log_psd, test_log_psd)
|
corr = abs(P_xy) / (np.sqrt(mic_P_xx) * np.sqrt(P_yy) + 1e-8)
|
|
if corr > best_corr:
|
best_corr = corr
|
best_ref_idx = test_idx
|
best_ref_rms = test_ref_rms
|
|
best_ts = sorted_timestamps[best_ref_idx]
|
best_ref = np.frombuffer(self.aec_audio_cache[best_ts], dtype=np.int16).astype(np.float32)
|
ref_rms = best_ref_rms
|
|
if ref_rms < 50:
|
return pcm_frame
|
|
# 对齐参考信号（直接截取相同长度）
|
aligned_ref = best_ref[:n]
|
if len(aligned_ref) < n:
|
aligned_ref = np.pad(aligned_ref, (0, n - len(aligned_ref)))
|
|
# ========== 频域 AEC 处理（谱减法） ==========
|
# 时域信号经过声学路径后相位失真，导致时域相关性低且P_xy正负不定
|
# 频域幅度谱不受相位影响，对数功率谱相关性稳定在0.97+
|
# 公式：result_mag = max(|mic_fft| - |ref_fft| * scale * coef, 0)
|
|
mic_mag = np.abs(mic_fft)
|
mic_phase = np.angle(mic_fft)
|
ref_fft = np.fft.rfft(aligned_ref * np.hanning(n))
|
ref_mag = np.abs(ref_fft)
|
|
# 频域计算回声比例 scale
|
scale = np.sum(mic_mag * ref_mag) / (np.dot(ref_mag, ref_mag) + 1e-8)
|
|
# 自适应系数：根据scale和coh动态调整
|
# scale大（回声强）-> coef大；coh高（匹配准）-> coef大
|
raw_coef = 1.0 + scale * 3 + (best_corr - 0.97) * 30
|
coef = max(0.5, min(3.0, raw_coef))
|
|
# 谱减法（过减 + 半波整流）
|
echo_mag = ref_mag * scale * coef
|
result_mag = np.maximum(mic_mag - echo_mag * 1.5, mic_mag * 0.1)
|
|
# 保留相位重建信号
|
result_fft = result_mag * np.exp(1j * mic_phase)
|
output = np.fft.irfft(result_fft, n)
|
|
# 高置信度是纯回声时，再压一下确保VAD检测不到
|
if best_corr >= 0.97 and ref_rms > 500:
|
output = output * 0.3
|
|
# 后处理：限幅
|
output = np.clip(output, -32768, 32767)
|
|
# 转换为bytes
|
result = output.astype(np.int16).tobytes()
|
|
return result
|
|
except Exception as e:
|
self.logger.bind(tag=TAG).warning(f"[AEC] 处理失败: {e}")
|
return pcm_frame
|
|
def _decode_opus_packet(self, opus_packet: bytes) -> bytes:
|
"""
|
解码Opus数据包为PCM数据
|
|
Args:
|
opus_packet: Opus编码的音频数据
|
|
Returns:
|
bytes: 解码后的PCM数据，失败返回None
|
"""
|
try:
|
if not opus_packet or len(opus_packet) == 0:
|
return None
|
|
self._init_connection_state(self)
|
pcm_frame = self._connection_opus_decoder.decode(opus_packet, 960)
|
return pcm_frame
|
except Exception as e:
|
self.logger.bind(tag=TAG).debug(f"Opus解码失败: {e}")
|
return None
|
|
def _init_connection_state(self, conn):
|
"""为连接初始化独立的Opus解码器"""
|
if not hasattr(conn, "_connection_opus_decoder"):
|
conn._connection_opus_decoder = opuslib_next.Decoder(16000, 1)
|
|
async def handle_restart(self, message):
|
"""处理服务器重启请求"""
|
try:
|
|
self.logger.bind(tag=TAG).info("收到服务器重启指令，准备执行...")
|
|
# 发送确认响应
|
await self.websocket.send(
|
json.dumps(
|
{
|
"type": "server",
|
"status": "success",
|
"message": "服务器重启中...",
|
"content": {"action": "restart"},
|
}
|
)
|
)
|
|
# 异步执行重启操作
|
def restart_server():
|
"""实际执行重启的方法"""
|
time.sleep(1)
|
self.logger.bind(tag=TAG).info("执行服务器重启...")
|
subprocess.Popen(
|
[sys.executable, "app.py"],
|
stdin=sys.stdin,
|
stdout=sys.stdout,
|
stderr=sys.stderr,
|
start_new_session=True,
|
)
|
os._exit(0)
|
|
# 使用线程执行重启避免阻塞事件循环
|
threading.Thread(target=restart_server, daemon=True).start()
|
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"重启失败: {str(e)}")
|
await self.websocket.send(
|
json.dumps(
|
{
|
"type": "server",
|
"status": "error",
|
"message": f"Restart failed: {str(e)}",
|
"content": {"action": "restart"},
|
}
|
)
|
)
|
|
def _tts_config_ready(self) -> bool:
|
selected = (self.config.get("selected_module") or {}).get("TTS")
|
providers = self.config.get("TTS") or {}
|
return bool(selected) and selected in providers
|
|
async def _open_tts_channels(self):
|
"""Open TTS worker threads. Must run on the connection event loop after TTS config exists."""
|
if self.tts is None and not self._tts_config_ready():
|
self.logger.bind(tag=TAG).info("TTS配置尚未从智控台加载，延后打开通道")
|
return
|
with self._tts_channel_lock:
|
if self.tts is None:
|
self.tts = self._initialize_tts()
|
tts = self.tts
|
if tts is None:
|
self.logger.bind(tag=TAG).warning("TTS实例不可用，无法打开通道")
|
return
|
await tts.open_audio_channels(self)
|
play = getattr(tts, "audio_play_priority_thread", None)
|
text_th = getattr(tts, "tts_priority_thread", None)
|
if play and play.is_alive() and text_th and text_th.is_alive():
|
self.logger.bind(tag=TAG).info("TTS通道已打开")
|
else:
|
self.logger.bind(tag=TAG).error("TTS通道打开后工作线程未运行")
|
|
def _ensure_tts_channels(self):
|
"""Start TTS worker threads once; used after private-config init."""
|
try:
|
if self.loop is None:
|
return
|
fut = asyncio.run_coroutine_threadsafe(self._open_tts_channels(), self.loop)
|
fut.result(timeout=15)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"打开TTS通道失败: {e}")
|
|
def _initialize_components(self):
|
try:
|
if self.tts is not None:
|
self._ensure_tts_channels()
|
if self.need_bind:
|
self.bind_completed_event.set()
|
return
|
self.selected_module_str = build_module_string(
|
self.config.get("selected_module", {})
|
)
|
self.logger = create_connection_logger(self.selected_module_str)
|
|
"""初始化组件"""
|
if self.config.get("prompt") is not None:
|
user_prompt = self.config["prompt"]
|
# 使用快速提示词进行初始化
|
prompt = self.prompt_manager.get_quick_prompt(user_prompt)
|
self.change_system_prompt(prompt)
|
self.logger.bind(tag=TAG).info(
|
f"快速初始化组件: prompt成功 {prompt[:50]}..."
|
)
|
|
"""初始化本地组件"""
|
if self.vad is None:
|
self.vad = self._vad
|
if self.asr is None:
|
self.asr = self._initialize_asr()
|
|
# 初始化声纹识别
|
self._initialize_voiceprint()
|
# 打开语音识别通道
|
asyncio.run_coroutine_threadsafe(
|
self.asr.open_audio_channels(self), self.loop
|
)
|
|
"""加载记忆"""
|
self._initialize_memory()
|
"""加载意图识别"""
|
self._initialize_intent()
|
"""初始化上报线程"""
|
self._init_report_threads()
|
"""更新系统提示词"""
|
self._init_prompt_enhancement()
|
"""注入工具调用few-shot示例（仅function_call模式）"""
|
self._inject_tool_call_fewshot()
|
if self.original_system_prompt is None:
|
self.original_system_prompt = self.prompt
|
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"实例化组件失败: {e}")
|
|
def _init_prompt_enhancement(self):
|
|
# 更新上下文信息
|
self.prompt_manager.update_context_info(self, self.client_ip)
|
enhanced_prompt = self.prompt_manager.build_enhanced_prompt(
|
self.config["prompt"],
|
self.device_id,
|
self.client_ip,
|
emoji_enabled=(self.features or {}).get("emoji", True),
|
)
|
if enhanced_prompt:
|
self.change_system_prompt(enhanced_prompt)
|
self.logger.bind(tag=TAG).debug("系统提示词已增强更新")
|
|
def _inject_tool_call_fewshot(self):
|
"""注入工具调用 few-shot 示例到对话历史。
|
结构：正样本（工具调用示例）放在动态 system 之前，可命中前缀缓存；
|
负样本（直接回答示例）放在动态 system 之后、紧挨真实用户消息，
|
确保模型在处理用户消息前最后看到的是"不调工具"的行为模式。
|
"""
|
if self.intent_type != "function_call":
|
return
|
if not hasattr(self, "func_handler") or self.func_handler is None:
|
return
|
|
tools = self.func_handler.get_functions()
|
if not tools:
|
return
|
|
tool_names = {t.get("function", {}).get("name") for t in tools}
|
|
# === few-shot 示例（is_temporary）===
|
# 展示 direct_answer 携带 response 参数的用法，一次调用完成回复
|
|
# 示例1：direct_answer（回复内容写在 response 参数里，无需递归）
|
da_tc_id = "fewshot_da_001"
|
self.dialogue.put(Message(role="user", content="给我讲个故事吧", is_temporary=True))
|
self.dialogue.put(Message(
|
role="assistant",
|
tool_calls=[{
|
"id": da_tc_id,
|
"function": {"arguments": '{"response": "好呀，你想听什么类型的呀？童话、冒险还是搞笑的？选一个我给你开讲~"}', "name": "direct_answer"},
|
"type": "function", "index": 0,
|
}],
|
is_temporary=True,
|
))
|
self.dialogue.put(Message(
|
role="tool", tool_call_id=da_tc_id,
|
content="已直接回复", is_temporary=True,
|
))
|
|
# 示例2：真实工具调用（handle_exit_intent）
|
if "handle_exit_intent" in tool_names:
|
tc_id = "fewshot_exit_001"
|
self.dialogue.put(Message(role="user", content="拜拜", is_temporary=True))
|
self.dialogue.put(Message(
|
role="assistant",
|
tool_calls=[{
|
"id": tc_id,
|
"function": {"arguments": '{"say_goodbye": "再见，下次再聊~"}', "name": "handle_exit_intent"},
|
"type": "function", "index": 0,
|
}],
|
is_temporary=True,
|
))
|
self.dialogue.put(Message(
|
role="tool", tool_call_id=tc_id,
|
content="退出意图已处理", is_temporary=True,
|
))
|
self.dialogue.put(Message(
|
role="assistant", content="再见，下次再聊~", is_temporary=True,
|
))
|
|
self.logger.bind(tag=TAG).debug("已注入工具调用 few-shot 示例")
|
|
def _init_report_threads(self):
|
"""初始化ASR和TTS上报线程"""
|
if not self.read_config_from_api or self.need_bind:
|
return
|
if self.chat_history_conf == 0:
|
return
|
if self.report_thread is None or not self.report_thread.is_alive():
|
self.report_thread = threading.Thread(
|
target=self._report_worker, daemon=True
|
)
|
self.report_thread.start()
|
self.logger.bind(tag=TAG).info("TTS上报线程已启动")
|
|
def _initialize_tts(self):
|
"""初始化TTS"""
|
tts = None
|
if not self.need_bind and self._tts_config_ready():
|
tts = initialize_tts(self.config)
|
elif not self._tts_config_ready():
|
self.logger.bind(tag=TAG).info("跳过TTS初始化：配置中没有可用的TTS模块")
|
return None
|
|
if tts is None:
|
tts = DefaultTTS(self.config, delete_audio_file=True)
|
|
return tts
|
|
def _initialize_asr(self):
|
"""初始化ASR"""
|
if (
|
self._asr is not None
|
and hasattr(self._asr, "interface_type")
|
and self._asr.interface_type == InterfaceType.LOCAL
|
):
|
# 如果公共ASR是本地服务，则直接返回
|
# 因为本地一个实例ASR，可以被多个连接共享
|
asr = self._asr
|
else:
|
# 如果公共ASR是远程服务，则初始化一个新实例
|
# 因为远程ASR，涉及到websocket连接和接收线程，需要每个连接一个实例
|
asr = initialize_asr(self.config)
|
|
return asr
|
|
def _initialize_voiceprint(self):
|
"""为当前连接初始化声纹识别"""
|
try:
|
voiceprint_config = self.config.get("voiceprint", {})
|
if voiceprint_config:
|
voiceprint_provider = VoiceprintProvider(voiceprint_config)
|
if voiceprint_provider is not None and voiceprint_provider.enabled:
|
self.voiceprint_provider = voiceprint_provider
|
self.logger.bind(tag=TAG).info("声纹识别功能已在连接时动态启用")
|
else:
|
self.logger.bind(tag=TAG).warning("声纹识别功能启用但配置不完整")
|
else:
|
self.logger.bind(tag=TAG).info("声纹识别功能未启用")
|
except Exception as e:
|
self.logger.bind(tag=TAG).warning(f"声纹识别初始化失败: {str(e)}")
|
|
async def _background_initialize(self):
|
"""在后台初始化配置和组件（完全不阻塞主循环）"""
|
try:
|
# 异步获取差异化配置
|
await self._initialize_private_config_async()
|
# 在线程池中初始化组件
|
self.executor.submit(self._initialize_components)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"后台初始化失败: {e}")
|
|
async def _initialize_private_config_async(self):
|
"""从接口异步获取差异化配置（异步版本，不阻塞主循环）"""
|
if not self.read_config_from_api:
|
self.need_bind = False
|
self.bind_completed_event.set()
|
return
|
try:
|
begin_time = time.time()
|
private_config = await get_private_config_from_api(
|
self.config,
|
self.headers.get("device-id"),
|
self.headers.get("client-id", self.headers.get("device-id")),
|
)
|
private_config["delete_audio"] = bool(self.config.get("delete_audio", True))
|
private_config["tts_timeout"] = self.config.get("tts_timeout", 15)
|
self.logger.bind(tag=TAG).info(
|
f"{time.time() - begin_time} 秒，异步获取差异化配置成功: {json.dumps(filter_sensitive_info(private_config), ensure_ascii=False)}"
|
)
|
self.need_bind = False
|
self.bind_completed_event.set()
|
except DeviceNotFoundException as e:
|
self.need_bind = True
|
private_config = {}
|
except DeviceBindException as e:
|
self.need_bind = True
|
self.bind_code = e.bind_code
|
private_config = {}
|
except Exception as e:
|
self.need_bind = True
|
self.logger.bind(tag=TAG).error(f"异步获取差异化配置失败: {e}")
|
private_config = {}
|
|
init_llm, init_tts, init_memory, init_intent = (
|
False,
|
False,
|
False,
|
False,
|
)
|
|
init_vad = check_vad_update(self.common_config, private_config)
|
init_asr = check_asr_update(self.common_config, private_config)
|
|
if init_vad:
|
self.config["VAD"] = private_config["VAD"]
|
self.config["selected_module"]["VAD"] = private_config["selected_module"][
|
"VAD"
|
]
|
if init_asr:
|
self.config["ASR"] = private_config["ASR"]
|
self.config["selected_module"]["ASR"] = private_config["selected_module"][
|
"ASR"
|
]
|
if private_config.get("TTS", None) is not None:
|
init_tts = True
|
self.config["TTS"] = private_config["TTS"]
|
self.config["selected_module"]["TTS"] = private_config["selected_module"][
|
"TTS"
|
]
|
if private_config.get("LLM", None) is not None:
|
init_llm = True
|
self.config["LLM"] = private_config["LLM"]
|
self.config["selected_module"]["LLM"] = private_config["selected_module"][
|
"LLM"
|
]
|
if private_config.get("VLLM", None) is not None:
|
self.config["VLLM"] = private_config["VLLM"]
|
self.config["selected_module"]["VLLM"] = private_config["selected_module"][
|
"VLLM"
|
]
|
if private_config.get("Memory", None) is not None:
|
init_memory = True
|
self.config["Memory"] = private_config["Memory"]
|
self.config["selected_module"]["Memory"] = private_config[
|
"selected_module"
|
]["Memory"]
|
if private_config.get("Intent", None) is not None:
|
init_intent = True
|
self.config["Intent"] = private_config["Intent"]
|
model_intent = private_config.get("selected_module", {}).get("Intent", {})
|
self.config["selected_module"]["Intent"] = model_intent
|
# 加载插件配置
|
if model_intent != "Intent_nointent":
|
plugin_from_server = private_config.get("plugins", {})
|
for plugin, config_str in plugin_from_server.items():
|
plugin_from_server[plugin] = json.loads(config_str)
|
# 将模块级别的插件配置复制到各具体函数名，
|
# 方便后续 per-function 查找描述、news_sources 等配置
|
for module_name, func_names in module_func_map.items():
|
if module_name in plugin_from_server:
|
module_config = plugin_from_server[module_name]
|
for func_name in func_names:
|
if func_name not in plugin_from_server:
|
plugin_from_server[func_name] = module_config
|
self.config["plugins"] = plugin_from_server
|
# 将模块级别的插件名展开为具体函数名
|
expanded_functions = []
|
for plugin_key in plugin_from_server.keys():
|
if plugin_key in all_function_registry:
|
expanded_functions.append(plugin_key)
|
elif plugin_key in module_func_map:
|
expanded_functions.extend(module_func_map[plugin_key])
|
else:
|
expanded_functions.append(plugin_key)
|
self.config["Intent"][self.config["selected_module"]["Intent"]][
|
"functions"
|
] = expanded_functions
|
if private_config.get("prompt", None) is not None:
|
self.config["prompt"] = private_config["prompt"]
|
# 获取声纹信息
|
if private_config.get("voiceprint", None) is not None:
|
self.config["voiceprint"] = private_config["voiceprint"]
|
if private_config.get("summaryMemory", None) is not None:
|
self.config["summaryMemory"] = private_config["summaryMemory"]
|
if private_config.get("device_max_output_size", None) is not None:
|
self.max_output_size = int(private_config["device_max_output_size"])
|
if private_config.get("chat_history_conf", None) is not None:
|
self.chat_history_conf = int(private_config["chat_history_conf"])
|
if private_config.get("mcp_endpoint", None) is not None:
|
self.config["mcp_endpoint"] = private_config["mcp_endpoint"]
|
if private_config.get("context_providers", None) is not None:
|
self.config["context_providers"] = private_config["context_providers"]
|
|
# 注入替换词到 TTS 模块配置
|
if private_config.get("correct_words", None) is not None:
|
select_tts_module = self.config["selected_module"]["TTS"]
|
self.config["TTS"][select_tts_module]["correct_words"] = private_config[
|
"correct_words"
|
]
|
|
# 使用 run_in_executor 在线程池中执行 initialize_modules，避免阻塞主循环
|
try:
|
modules = await self.loop.run_in_executor(
|
None, # 使用默认线程池
|
initialize_modules,
|
self.logger,
|
private_config,
|
init_vad,
|
init_asr,
|
init_llm,
|
init_tts,
|
init_memory,
|
init_intent,
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"初始化组件失败: {e}")
|
modules = {}
|
if modules.get("tts", None) is not None:
|
self.tts = modules["tts"]
|
if self.tts is not None:
|
try:
|
await self._open_tts_channels()
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"打开TTS通道失败: {e}")
|
if modules.get("vad", None) is not None:
|
self.vad = modules["vad"]
|
if modules.get("asr", None) is not None:
|
self.asr = modules["asr"]
|
if modules.get("llm", None) is not None:
|
self.llm = modules["llm"]
|
if modules.get("intent", None) is not None:
|
self.intent = modules["intent"]
|
if modules.get("memory", None) is not None:
|
self.memory = modules["memory"]
|
|
def _initialize_memory(self):
|
if self.memory is None:
|
return
|
"""初始化记忆模块"""
|
self.memory.init_memory(
|
role_id=self.device_id,
|
llm=self.llm,
|
summary_memory=self.config.get("summaryMemory", None),
|
save_to_file=not self.read_config_from_api,
|
)
|
|
# 获取记忆总结配置
|
memory_config = self.config["Memory"]
|
memory_type = self.config["Memory"][self.config["selected_module"]["Memory"]][
|
"type"
|
]
|
# 如果使用 nomen 或 mem_report_only，直接返回
|
if memory_type == "nomem" or memory_type == "mem_report_only":
|
return
|
# 使用 mem_local_short 模式
|
elif memory_type == "mem_local_short":
|
memory_llm_name = memory_config[self.config["selected_module"]["Memory"]][
|
"llm"
|
]
|
if memory_llm_name and memory_llm_name in self.config["LLM"]:
|
# 如果配置了专用LLM，则创建独立的LLM实例
|
from core.utils import llm as llm_utils
|
|
memory_llm_config = self.config["LLM"][memory_llm_name]
|
memory_llm_type = memory_llm_config.get("type", memory_llm_name)
|
memory_llm = llm_utils.create_instance(
|
memory_llm_type, memory_llm_config
|
)
|
self.logger.bind(tag=TAG).info(
|
f"为记忆总结创建了专用LLM: {memory_llm_name}, 类型: {memory_llm_type}"
|
)
|
self.memory.set_llm(memory_llm)
|
else:
|
# 否则使用主LLM
|
self.memory.set_llm(self.llm)
|
self.logger.bind(tag=TAG).info("使用主LLM作为意图识别模型")
|
|
def _initialize_intent(self):
|
if self.intent is None:
|
return
|
self.intent_type = self.config["Intent"][
|
self.config["selected_module"]["Intent"]
|
]["type"]
|
if self.intent_type == "function_call" or self.intent_type == "intent_llm":
|
self.load_function_plugin = True
|
"""初始化意图识别模块"""
|
# 获取意图识别配置
|
intent_config = self.config["Intent"]
|
intent_type = self.config["Intent"][self.config["selected_module"]["Intent"]][
|
"type"
|
]
|
|
# 如果使用 nointent，直接返回
|
if intent_type == "nointent":
|
return
|
# 使用 intent_llm 模式
|
elif intent_type == "intent_llm":
|
intent_llm_name = intent_config[self.config["selected_module"]["Intent"]][
|
"llm"
|
]
|
|
if intent_llm_name and intent_llm_name in self.config["LLM"]:
|
# 如果配置了专用LLM，则创建独立的LLM实例
|
from core.utils import llm as llm_utils
|
|
intent_llm_config = self.config["LLM"][intent_llm_name]
|
intent_llm_type = intent_llm_config.get("type", intent_llm_name)
|
intent_llm = llm_utils.create_instance(
|
intent_llm_type, intent_llm_config
|
)
|
self.logger.bind(tag=TAG).info(
|
f"为意图识别创建了专用LLM: {intent_llm_name}, 类型: {intent_llm_type}"
|
)
|
self.intent.set_llm(intent_llm)
|
else:
|
# 否则使用主LLM
|
self.intent.set_llm(self.llm)
|
self.logger.bind(tag=TAG).info("使用主LLM作为意图识别模型")
|
|
"""加载统一工具处理器"""
|
self.func_handler = UnifiedToolHandler(self)
|
|
# 异步初始化工具处理器
|
if hasattr(self, "loop") and self.loop:
|
asyncio.run_coroutine_threadsafe(self.func_handler._initialize(), self.loop)
|
|
def change_system_prompt(self, prompt):
|
self.prompt = prompt
|
# 更新系统prompt至上下文
|
self.dialogue.update_system_message(self.prompt)
|
|
def chat(self, query, depth=0):
|
# 保存当前任务的sentence_id到局部变量，避免被新任务覆盖
|
current_sentence_id = None
|
|
if query is not None:
|
self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}")
|
|
# 为最顶层时新建会话ID和发送FIRST请求
|
if depth == 0:
|
current_sentence_id = str(uuid.uuid4().hex)
|
self.sentence_id = current_sentence_id # 更新共享属性
|
self.dialogue.put(Message(role="user", content=query))
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.FIRST,
|
content_type=ContentType.ACTION,
|
)
|
)
|
else:
|
# 递归调用时，使用当前的sentence_id
|
current_sentence_id = self.sentence_id
|
|
# 设置最大递归深度，避免无限循环，可根据实际需求调整
|
MAX_DEPTH = 5
|
force_final_answer = False # 标记是否强制最终回答
|
|
if depth >= MAX_DEPTH:
|
self.logger.bind(tag=TAG).debug(
|
f"已达到最大工具调用深度 {MAX_DEPTH}，将强制基于现有信息回答"
|
)
|
force_final_answer = True
|
# 添加系统指令，要求 LLM 基于现有信息回答
|
self.dialogue.put(
|
Message(
|
role="user",
|
content="[系统提示] 已达到最大工具调用次数限制，请你基于目前已经获取的所有信息，直接给出最终答案。不要再尝试调用任何工具。",
|
)
|
)
|
|
# Define intent functions
|
functions = None
|
# 达到最大深度时，禁用工具调用，强制 LLM 直接回答
|
if (
|
self.intent_type == "function_call"
|
and hasattr(self, "func_handler")
|
and not force_final_answer
|
):
|
functions = list(self.func_handler.get_functions())
|
# 仅在第一层调用时注入 direct_answer 虚拟工具
|
# 递归调用（depth>0）不注入，避免模型在生成文本回复时再次调 direct_answer 导致循环
|
if functions is not None and depth == 0:
|
functions.append(DIRECT_ANSWER_TOOL)
|
|
response_message = []
|
|
try:
|
# 使用带记忆的对话
|
memory_str = None
|
# 仅当query非空（代表用户询问）时查询记忆
|
if self.memory is not None and query:
|
future = asyncio.run_coroutine_threadsafe(
|
self.memory.query_memory(query), self.loop
|
)
|
memory_str = future.result()
|
|
# 仅在该说话人首次出现时把身份注入 system，之后靠对话历史首轮保留，
|
# 避免每轮在 system 重复出现名字诱导模型反复称呼
|
speaker_for_system = None
|
cs = (self.current_speaker or "").strip()
|
if cs and cs != "未知说话人" and cs not in self.system_introduced_speakers:
|
self.system_introduced_speakers.add(cs)
|
speaker_for_system = cs
|
|
if self.intent_type == "function_call" and functions is not None:
|
# 使用支持functions的streaming接口
|
llm_responses = self.llm.response_with_functions(
|
self.session_id,
|
self.dialogue.get_llm_dialogue_with_memory(
|
memory_str, self.config.get("voiceprint", {}), speaker_for_system
|
),
|
functions=functions,
|
)
|
else:
|
llm_responses = self.llm.response(
|
self.session_id,
|
self.dialogue.get_llm_dialogue_with_memory(
|
memory_str, self.config.get("voiceprint", {}), speaker_for_system
|
),
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}")
|
return None
|
|
# 处理流式响应
|
tool_call_flag = False
|
# 支持多个并行工具调用 - 使用列表存储
|
tool_calls_list = [] # 格式: [{"id": "", "name": "", "arguments": ""}]
|
content_arguments = ""
|
emotion_flag = True
|
try:
|
for response in llm_responses:
|
if self.client_abort:
|
break
|
if self.intent_type == "function_call" and functions is not None:
|
content, tools_call = response
|
if "content" in response:
|
content = response["content"]
|
tools_call = None
|
if content is not None and len(content) > 0:
|
content_arguments += content
|
|
if not tool_call_flag and content_arguments.startswith("<tool_call>"):
|
# print("content_arguments", content_arguments)
|
tool_call_flag = True
|
|
if tools_call is not None and len(tools_call) > 0:
|
tool_call_flag = True
|
self._merge_tool_calls(tool_calls_list, tools_call)
|
|
# 流式提取 direct_answer 的 response 参数，实时送 TTS
|
# 使用安全缓冲区，防止 JSON 闭合符号泄漏到 TTS
|
_DA_STREAM_BUFFER = 5
|
for tc in tool_calls_list:
|
if tc["name"] == "direct_answer" and tc.get("arguments"):
|
da_text = self._extract_direct_answer_response(tc["arguments"])
|
sent_len = tc.get("_da_sent", 0)
|
if da_text and len(da_text) > sent_len:
|
safe_end = max(sent_len, len(da_text) - _DA_STREAM_BUFFER)
|
if safe_end > sent_len:
|
new_part = da_text[sent_len:safe_end]
|
# 清理 delta 中可能泄漏的 JSON 闭合垃圾
|
new_part = self._clean_response_garbage(new_part)
|
if new_part:
|
tc["_da_sent"] = safe_end
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.MIDDLE,
|
content_type=ContentType.TEXT,
|
content_detail=new_part,
|
)
|
)
|
else:
|
content = response
|
|
# 在llm回复中获取情绪表情，一轮对话只在开头获取一次
|
if emotion_flag and content is not None and content.strip():
|
if (self.features or {}).get("emoji", True):
|
asyncio.run_coroutine_threadsafe(
|
textUtils.get_emotion(self, content),
|
self.loop,
|
)
|
emotion_flag = False
|
|
if content is not None and len(content) > 0:
|
if not tool_call_flag:
|
response_message.append(content)
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.MIDDLE,
|
content_type=ContentType.TEXT,
|
content_detail=content,
|
)
|
)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"LLM stream processing error: {e}")
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.MIDDLE,
|
content_type=ContentType.TEXT,
|
content_detail=get_system_error_response(self.config),
|
)
|
)
|
if depth == 0:
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.LAST,
|
content_type=ContentType.ACTION,
|
)
|
)
|
return
|
# 处理function call
|
if tool_call_flag:
|
bHasError = False
|
# 处理基于文本的工具调用格式
|
if len(tool_calls_list) == 0 and content_arguments:
|
a = extract_json_from_string(content_arguments)
|
if a is not None:
|
try:
|
content_arguments_json = json.loads(a)
|
tool_calls_list.append(
|
{
|
"id": str(uuid.uuid4().hex),
|
"name": content_arguments_json["name"],
|
"arguments": json.dumps(
|
content_arguments_json["arguments"],
|
ensure_ascii=False,
|
),
|
}
|
)
|
except Exception as e:
|
bHasError = True
|
response_message.append(a)
|
else:
|
bHasError = True
|
response_message.append(content_arguments)
|
if bHasError:
|
self.logger.bind(tag=TAG).error(
|
f"function call error: {content_arguments}"
|
)
|
|
if not bHasError and len(tool_calls_list) > 0:
|
# 处理 direct_answer 虚拟工具
|
direct_answer_calls = [tc for tc in tool_calls_list if tc["name"] == "direct_answer"]
|
real_tool_calls = [tc for tc in tool_calls_list if tc["name"] != "direct_answer"]
|
|
if direct_answer_calls:
|
self.logger.bind(tag=TAG).debug(
|
f"模型选择 direct_answer，流式已播报，写入对话历史"
|
)
|
for tc in direct_answer_calls:
|
da_response = self._extract_direct_answer_response(tc.get("arguments", "{}"))
|
if da_response:
|
# 刷新流式缓冲区中未发送的部分
|
sent_len = tc.get("_da_sent", 0)
|
remaining = da_response[sent_len:]
|
if remaining:
|
remaining = self._clean_response_garbage(remaining)
|
if remaining:
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.MIDDLE,
|
content_type=ContentType.TEXT,
|
content_detail=remaining,
|
)
|
)
|
# 写入对话历史
|
da_response = self._clean_response_garbage(da_response)
|
self.tts.store_tts_text(current_sentence_id, da_response)
|
self.dialogue.put(Message(role="assistant", content=da_response))
|
|
if not real_tool_calls:
|
if depth == 0:
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.LAST,
|
content_type=ContentType.ACTION,
|
)
|
)
|
return
|
|
tool_calls_list = real_tool_calls
|
|
if not bHasError and len(tool_calls_list) > 0:
|
self.logger.bind(tag=TAG).debug(
|
f"检测到 {len(tool_calls_list)} 个工具调用"
|
)
|
|
# LLM 流式阶段已播报过的文本
|
streamed_text = ""
|
if len(response_message) > 0:
|
streamed_text = "".join(response_message)
|
self.tts.store_tts_text(current_sentence_id, streamed_text)
|
self.dialogue.put(Message(role="assistant", content=streamed_text))
|
response_message.clear()
|
|
# 收集所有工具调用的 Future
|
futures_with_data = []
|
for tool_call_data in tool_calls_list:
|
self.logger.bind(tag=TAG).debug(
|
f"function_name={tool_call_data['name']}, function_id={tool_call_data['id']}, function_arguments={tool_call_data['arguments']}"
|
)
|
|
# 使用公共方法上报工具调用
|
tool_input = json.loads(tool_call_data.get("arguments") or "{}")
|
enqueue_tool_report(self, tool_call_data['name'], tool_input)
|
|
future = asyncio.run_coroutine_threadsafe(
|
self.func_handler.handle_llm_function_call(
|
self, tool_call_data
|
),
|
self.loop,
|
)
|
futures_with_data.append((future, tool_call_data, tool_input))
|
|
# 工具调用超时时间，可配置，默认30秒
|
tool_call_timeout = int(self.config.get("tool_call_timeout", 30))
|
# 等待协程结束（实际等待时长为最慢的那个）
|
tool_results = []
|
|
for future, tool_call_data, tool_input in futures_with_data:
|
try:
|
result = future.result(timeout=tool_call_timeout)
|
tool_results.append((result, tool_call_data))
|
# 使用公共方法上报工具调用结果
|
enqueue_tool_report(self, tool_call_data['name'], tool_input, str(result.result) if result.result else None, report_tool_call=False)
|
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(
|
f"工具调用超时或异常: {tool_call_data['name']}, 错误: {e}"
|
)
|
# 超时时返回错误响应，避免整个流程卡死
|
tool_results.append((
|
ActionResponse(action=Action.ERROR, result="哎呀，网络遇到点问题，请稍后再试下！"),
|
tool_call_data
|
))
|
# 上报工具调用错误
|
enqueue_tool_report(self, tool_call_data['name'], tool_input, str(e), report_tool_call=False)
|
|
# 统一处理工具调用结果
|
if tool_results:
|
self._handle_function_result(tool_results, depth=depth, streamed_text=streamed_text)
|
|
# 存储对话内容
|
if len(response_message) > 0:
|
text_buff = "".join(response_message)
|
self.tts.store_tts_text(current_sentence_id, text_buff)
|
self.dialogue.put(Message(role="assistant", content=text_buff))
|
|
if depth == 0:
|
self.tts.tts_text_queue.put(
|
TTSMessageDTO(
|
sentence_id=current_sentence_id,
|
sentence_type=SentenceType.LAST,
|
content_type=ContentType.ACTION,
|
)
|
)
|
# 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
|
self.logger.bind(tag=TAG).debug(
|
lambda: json.dumps(
|
self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False
|
)
|
)
|
|
return True
|
|
def _handle_function_result(self, tool_results, depth, streamed_text=""):
|
need_llm_tools = []
|
record_tools = []
|
|
for result, tool_call_data in tool_results:
|
if result.action in [
|
Action.RESPONSE,
|
Action.NOTFOUND,
|
Action.ERROR,
|
]:
|
text = result.response if result.response else result.result
|
if streamed_text and text in streamed_text:
|
self.logger.bind(tag=TAG).debug(
|
f"Skipping duplicate TTS for tool {tool_call_data['name']}, already streamed"
|
)
|
else:
|
self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
|
self.tts.store_tts_text(self.sentence_id, text)
|
self.dialogue.put(Message(role="assistant", content=text))
|
elif result.action == Action.REQLLM:
|
need_llm_tools.append((result, tool_call_data))
|
elif result.action == Action.RECORD:
|
record_tools.append((result, tool_call_data))
|
else:
|
pass
|
|
# Action.RECORD：写入完整工具调用链（assistant(tool_calls) → tool(result) → assistant(response)）
|
# 模型从历史中学到工具调用模式，不额外调用LLM
|
if record_tools:
|
# 构造 assistant 消息（含 tool_calls），记录"模型调用了哪些工具"
|
all_tool_calls = [
|
{
|
"id": tool_call_data["id"],
|
"function": {
|
"arguments": (
|
"{}"
|
if tool_call_data["arguments"] == ""
|
else tool_call_data["arguments"]
|
),
|
"name": tool_call_data["name"],
|
},
|
"type": "function",
|
"index": idx,
|
}
|
for idx, (_, tool_call_data) in enumerate(record_tools)
|
]
|
self.dialogue.put(Message(role="assistant", tool_calls=all_tool_calls))
|
|
# 写入每条工具的执行结果，记录"工具返回了什么"
|
for result, tool_call_data in record_tools:
|
text = result.result or ""
|
self.dialogue.put(
|
Message(
|
role="tool",
|
tool_call_id=(
|
str(uuid.uuid4())
|
if tool_call_data["id"] is None
|
else tool_call_data["id"]
|
),
|
content=text,
|
)
|
)
|
|
# 用固定文本作为最终回复，补全标准三段式，保证下一条消息是 user 而非接 tool
|
response_parts = []
|
for result, _ in record_tools:
|
resp = result.response or result.result
|
if resp:
|
response_parts.append(resp)
|
if response_parts:
|
self.dialogue.put(Message(role="assistant", content="，".join(response_parts)))
|
|
if need_llm_tools:
|
all_tool_calls = [
|
{
|
"id": tool_call_data["id"],
|
"function": {
|
"arguments": (
|
"{}"
|
if tool_call_data["arguments"] == ""
|
else tool_call_data["arguments"]
|
),
|
"name": tool_call_data["name"],
|
},
|
"type": "function",
|
"index": idx,
|
}
|
for idx, (_, tool_call_data) in enumerate(need_llm_tools)
|
]
|
self.dialogue.put(Message(role="assistant", tool_calls=all_tool_calls))
|
|
for result, tool_call_data in need_llm_tools:
|
text = result.result
|
if text is not None and len(text) > 0:
|
self.dialogue.put(
|
Message(
|
role="tool",
|
tool_call_id=(
|
str(uuid.uuid4())
|
if tool_call_data["id"] is None
|
else tool_call_data["id"]
|
),
|
content=text,
|
)
|
)
|
|
self.chat(None, depth=depth + 1)
|
|
def _report_worker(self):
|
"""聊天记录上报工作线程"""
|
while not self.stop_event.is_set():
|
try:
|
# 从队列获取数据，设置超时以便定期检查停止事件
|
item = self.report_queue.get(timeout=1)
|
if item is None: # 检测毒丸对象
|
break
|
try:
|
# 检查线程池状态
|
if self.executor is None:
|
continue
|
# 提交任务到线程池
|
self.executor.submit(self._process_report, *item)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}")
|
except queue.Empty:
|
continue
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"聊天记录上报工作线程异常: {e}")
|
|
self.logger.bind(tag=TAG).info("聊天记录上报线程已退出")
|
|
def _process_report(self, type, text, audio_data, report_time, task_instance_id=None, step_index=None):
|
"""处理上报任务"""
|
try:
|
# 执行异步上报（在事件循环中运行）
|
asyncio.run(report(self, type, text, audio_data, report_time, task_instance_id, step_index))
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"上报处理异常: {e}")
|
finally:
|
# 标记任务完成
|
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):
|
"""资源清理方法"""
|
try:
|
if self.server is not None:
|
await self.server.unregister_device_handler(self.device_id, self)
|
try:
|
from core.app_runtime.session import end_session, schedule_close_parent_except
|
|
end_session(self, "disconnected")
|
schedule_close_parent_except(self, "disconnected")
|
except Exception as session_error:
|
self.logger.bind(tag=TAG).warning(f"断线 flush TaskSession 失败: {session_error}")
|
|
# 清理 VAD 连接资源
|
if (
|
hasattr(self, "vad")
|
and self.vad
|
and hasattr(self.vad, "release_conn_resources")
|
):
|
self.vad.release_conn_resources(self)
|
|
# 清理opus解码器
|
if hasattr(self, "_connection_opus_decoder"):
|
try:
|
delattr(self, "_connection_opus_decoder")
|
except Exception:
|
pass
|
|
# 清理音频缓冲区
|
if hasattr(self, "audio_buffer"):
|
self.audio_buffer.clear()
|
|
# 取消超时任务
|
if self.timeout_task and not self.timeout_task.done():
|
self.timeout_task.cancel()
|
try:
|
await self.timeout_task
|
except asyncio.CancelledError:
|
pass
|
self.timeout_task = None
|
|
# 取消AEC缓存清理任务
|
if hasattr(self, "_aec_cache_cleanup_task") and self._aec_cache_cleanup_task and not self._aec_cache_cleanup_task.done():
|
self._aec_cache_cleanup_task.cancel()
|
try:
|
await self._aec_cache_cleanup_task
|
except asyncio.CancelledError:
|
pass
|
self._aec_cache_cleanup_task = None
|
|
# 清理AEC缓存
|
if hasattr(self, "aec_audio_cache"):
|
self.aec_audio_cache.clear()
|
self.aec_audio_cache_time.clear()
|
|
# 清理工具处理器资源
|
if hasattr(self, "func_handler") and self.func_handler:
|
try:
|
await self.func_handler.cleanup()
|
except Exception as cleanup_error:
|
self.logger.bind(tag=TAG).error(
|
f"清理工具处理器时出错: {cleanup_error}"
|
)
|
|
# 触发停止事件
|
if self.stop_event:
|
self.stop_event.set()
|
|
# 清空任务队列
|
self.clear_queues()
|
|
# 关闭WebSocket连接
|
try:
|
if ws:
|
# 安全地检查WebSocket状态并关闭
|
try:
|
if hasattr(ws, "closed") and not ws.closed:
|
await ws.close()
|
elif hasattr(ws, "state") and ws.state.name != "CLOSED":
|
await ws.close()
|
else:
|
# 如果没有closed属性，直接尝试关闭
|
await ws.close()
|
except Exception:
|
# 如果关闭失败，忽略错误
|
pass
|
elif self.websocket:
|
try:
|
if (
|
hasattr(self.websocket, "closed")
|
and not self.websocket.closed
|
):
|
await self.websocket.close()
|
elif (
|
hasattr(self.websocket, "state")
|
and self.websocket.state.name != "CLOSED"
|
):
|
await self.websocket.close()
|
else:
|
# 如果没有closed属性，直接尝试关闭
|
await self.websocket.close()
|
except Exception:
|
# 如果关闭失败，忽略错误
|
pass
|
except Exception as ws_error:
|
self.logger.bind(tag=TAG).error(f"关闭WebSocket连接时出错: {ws_error}")
|
|
if self.tts:
|
await self.tts.close()
|
if self.asr:
|
await self.asr.close()
|
|
# 最后关闭线程池（避免阻塞）
|
if self.executor:
|
try:
|
self.executor.shutdown(wait=False)
|
except Exception as executor_error:
|
self.logger.bind(tag=TAG).error(
|
f"关闭线程池时出错: {executor_error}"
|
)
|
self.executor = None
|
self.logger.bind(tag=TAG).info("连接资源已释放")
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"关闭连接时出错: {e}")
|
finally:
|
# 确保停止事件被设置
|
if self.stop_event:
|
self.stop_event.set()
|
|
def clear_queues(self):
|
"""清空所有任务队列"""
|
if self.tts:
|
self.logger.bind(tag=TAG).debug(
|
f"开始清理: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}"
|
)
|
|
# 使用非阻塞方式清空队列
|
for q in [
|
self.tts.tts_text_queue,
|
self.tts.tts_audio_queue,
|
self.report_queue,
|
]:
|
if not q:
|
continue
|
while True:
|
try:
|
q.get_nowait()
|
except queue.Empty:
|
break
|
|
# 重置音频流控器（取消后台任务并清空队列）
|
if hasattr(self, "audio_rate_controller") and self.audio_rate_controller:
|
self.audio_rate_controller.reset()
|
self.logger.bind(tag=TAG).debug("已重置音频流控器")
|
|
self.logger.bind(tag=TAG).debug(
|
f"清理结束: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}"
|
)
|
|
def reset_audio_states(self):
|
"""
|
重置所有音频相关状态(VAD + ASR)
|
"""
|
# Reset VAD states
|
self.client_audio_buffer.clear()
|
self.client_have_voice = False
|
self.client_voice_stop = False
|
self.client_voice_window.clear()
|
self.last_is_voice = False
|
self.vad_last_voice_time = 0.0
|
|
# Clear ASR buffers
|
self.asr_audio.clear()
|
|
self.logger.bind(tag=TAG).debug("All audio states reset.")
|
|
def chat_and_close(self, text):
|
"""Chat with the user and then close the connection"""
|
try:
|
# Use the existing chat method
|
self.chat(text)
|
|
# After chat is complete, close the connection
|
self.close_after_chat = True
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"Chat and close error: {str(e)}")
|
|
async def _check_timeout(self):
|
"""检查连接超时"""
|
try:
|
while not self.stop_event.is_set():
|
last_activity_time = self.last_activity_time
|
if self.need_bind:
|
last_activity_time = self.first_activity_time
|
|
# 检查是否超时（只有在时间戳已初始化的情况下）
|
if last_activity_time > 0.0:
|
current_time = time.time() * 1000
|
if current_time - last_activity_time > self.timeout_seconds * 1000:
|
if not self.stop_event.is_set():
|
self.logger.bind(tag=TAG).info("连接超时，准备关闭")
|
# 设置停止事件，防止重复处理
|
self.stop_event.set()
|
# 使用 try-except 包装关闭操作，确保不会因为异常而阻塞
|
try:
|
await self.close(self.websocket)
|
except Exception as close_error:
|
self.logger.bind(tag=TAG).error(
|
f"超时关闭连接时出错: {close_error}"
|
)
|
break
|
# 每10秒检查一次，避免过于频繁
|
await asyncio.sleep(10)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}")
|
finally:
|
self.logger.bind(tag=TAG).info("超时检查任务已退出")
|
|
async def _check_aec_cache_expiry(self):
|
"""定期清理过期的AEC缓存"""
|
try:
|
while not self.stop_event.is_set():
|
if hasattr(self, "aec_audio_cache") and self.aec_audio_cache:
|
current_time = time.time()
|
expired_keys = [
|
ts for ts, cache_time in list(self.aec_audio_cache_time.items())
|
if current_time - cache_time > 120 # 2分钟过期
|
]
|
for ts in expired_keys:
|
self.aec_audio_cache.pop(ts, None)
|
self.aec_audio_cache_time.pop(ts, None)
|
if expired_keys:
|
self.logger.bind(tag=TAG).debug(f"[AEC] 清理过期缓存 {len(expired_keys)} 条")
|
# 每30秒检查一次
|
await asyncio.sleep(30)
|
except Exception as e:
|
self.logger.bind(tag=TAG).error(f"AEC缓存清理任务出错: {e}")
|
|
@staticmethod
|
def _extract_direct_answer_response(arguments_str):
|
"""从 direct_answer 的参数中提取 response 值。
|
优先使用 json.loads 标准解析，流式阶段 fallback 到字符串提取。
|
"""
|
if not arguments_str:
|
return ""
|
# 优先尝试标准 JSON 解析（适用于完整且格式正确的 JSON）
|
try:
|
data = json.loads(arguments_str)
|
if isinstance(data, dict) and "response" in data:
|
return data["response"]
|
except (json.JSONDecodeError, TypeError):
|
pass
|
# Fallback：流式阶段 JSON 可能不完整，使用字符串提取
|
marker = '"response": "'
|
idx = arguments_str.find(marker)
|
if idx < 0:
|
marker = '"response":"'
|
idx = arguments_str.find(marker)
|
if idx < 0:
|
return ""
|
start = idx + len(marker)
|
raw = arguments_str[start:]
|
# 去掉末尾的 JSON 闭合符号（如果已完整）
|
if raw.endswith('"}'):
|
raw = raw[:-2]
|
elif raw.endswith('"'):
|
raw = raw[:-1]
|
# 处理 JSON 转义
|
raw = raw.replace('\\"', '"').replace('\\n', '\n').replace('\\\\', '\\')
|
return raw
|
|
@staticmethod
|
def _clean_response_garbage(text):
|
"""清理 response 中可能泄漏的 JSON 闭合符号。
|
模型有时会在 response 内容中生成 JSON 闭合字符（如 ）"}} 或 '})，
|
这些不是故事内容的一部分，需要去除。
|
"""
|
if not text:
|
return text
|
# 清理独立一行的 JSON 闭合垃圾（如 ）"}} '}} "}} }} } ）
|
_garbage_chars = frozenset('")\'}）')
|
lines = text.split('\n')
|
cleaned = []
|
for line in lines:
|
stripped = line.strip()
|
if stripped and len(stripped) <= 8 and all(c in _garbage_chars for c in stripped):
|
continue
|
cleaned.append(line)
|
result = '\n'.join(cleaned)
|
# 清理末尾残留的 JSON 闭合符号
|
result = re.sub(r'["\'}\]]+$', '', result.rstrip()).rstrip()
|
return result
|
|
def _merge_tool_calls(self, tool_calls_list, tools_call):
|
"""合并工具调用列表
|
|
Args:
|
tool_calls_list: 已收集的工具调用列表
|
tools_call: 新的工具调用
|
"""
|
for tool_call in tools_call:
|
tool_index = getattr(tool_call, "index", None)
|
if tool_index is None:
|
if tool_call.function.name:
|
# 有 function_name，说明是新的工具调用
|
tool_index = len(tool_calls_list)
|
else:
|
tool_index = len(tool_calls_list) - 1 if tool_calls_list else 0
|
|
# 确保列表有足够的位置
|
if tool_index >= len(tool_calls_list):
|
tool_calls_list.append({"id": "", "name": "", "arguments": ""})
|
|
# 更新工具调用信息
|
if tool_call.id:
|
tool_calls_list[tool_index]["id"] = tool_call.id
|
if tool_call.function.name:
|
tool_calls_list[tool_index]["name"] = tool_call.function.name
|
if tool_call.function.arguments:
|
tool_calls_list[tool_index]["arguments"] += tool_call.function.arguments
|
