diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 41ac96b7..cae6b3bc 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -139,6 +139,16 @@ plugins: - ".p3" refresh_time: 300 # 刷新音乐列表的时间间隔,单位为秒 +# 声纹识别配置 +voiceprint: + # 声纹接口地址 + url: + # 说话人配置:speaker_id,名称,描述 + speakers: + - "test1,张三,张三是一个程序员" + - "test2,李四,李四是一个产品经理" + - "test3,王五,王五是一个设计师" + # ##################################################################################### # ################################以下是角色模型配置###################################### diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index fdfd36fa..5c740aa8 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -631,8 +631,33 @@ class ConnectionHandler: self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") self.llm_finish_task = False + # 检查是否是JSON格式的消息(包含说话人信息) + enhanced_query = query + try: + if query.strip().startswith('{') and query.strip().endswith('}'): + data = json.loads(query) + if 'speaker' in data and 'content' in data: + # 直接使用JSON格式,不重新格式化 + enhanced_query = query + self.logger.bind(tag=TAG).info(f"识别到说话人: {data['speaker']}") + else: + # 如果有说话人信息但不是JSON格式,按原逻辑处理 + if hasattr(self, 'current_speaker') and self.current_speaker: + enhanced_query = f"[说话人: {self.current_speaker}] {query}" + self.logger.bind(tag=TAG).info(f"识别到说话人: {self.current_speaker}") + else: + # 如果有说话人信息但不是JSON格式,按原逻辑处理 + if hasattr(self, 'current_speaker') and self.current_speaker: + enhanced_query = f"[说话人: {self.current_speaker}] {query}" + self.logger.bind(tag=TAG).info(f"识别到说话人: {self.current_speaker}") + except json.JSONDecodeError: + # JSON解析失败,按原逻辑处理 + if hasattr(self, 'current_speaker') and self.current_speaker: + enhanced_query = f"[说话人: {self.current_speaker}] {query}" + self.logger.bind(tag=TAG).info(f"识别到说话人: {self.current_speaker}") + if not tool_call: - self.dialogue.put(Message(role="user", content=query)) + self.dialogue.put(Message(role="user", content=enhanced_query)) # Define intent functions functions = None @@ -645,7 +670,7 @@ class ConnectionHandler: memory_str = None if self.memory is not None: future = asyncio.run_coroutine_threadsafe( - self.memory.query_memory(query), self.loop + self.memory.query_memory(enhanced_query), self.loop ) memory_str = future.result() diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 465b081a..e6f39632 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -4,6 +4,7 @@ from core.utils.output_counter import check_device_output_limit from core.handle.abortHandle import handleAbortMessage import time import asyncio +import json from core.handle.sendAudioHandle import SentenceType from core.utils.util import audio_to_data @@ -38,6 +39,31 @@ async def resume_vad_detection(conn): async def startToChat(conn, text): + # 检查输入是否是JSON格式(包含说话人信息) + speaker_name = None + actual_text = text + + try: + # 尝试解析JSON格式的输入 + if text.strip().startswith('{') and text.strip().endswith('}'): + data = json.loads(text) + if 'speaker' in data and 'content' in data: + speaker_name = data['speaker'] + actual_text = data['content'] + conn.logger.bind(tag=TAG).info(f"解析到说话人信息: {speaker_name}") + + # 直接使用JSON格式的文本,不解析 + actual_text = text + except (json.JSONDecodeError, KeyError): + # 如果解析失败,继续使用原始文本 + pass + + # 保存说话人信息到连接对象 + if speaker_name: + conn.current_speaker = speaker_name + else: + conn.current_speaker = None + if conn.need_bind: await check_bind_device(conn) return @@ -52,16 +78,16 @@ async def startToChat(conn, text): if conn.client_is_speaking: await handleAbortMessage(conn) - # 首先进行意图分析 - intent_handled = await handle_user_intent(conn, text) + # 首先进行意图分析,使用实际文本内容 + intent_handled = await handle_user_intent(conn, actual_text) if intent_handled: # 如果意图已被处理,不再进行聊天 return - # 意图未被处理,继续常规聊天流程 - await send_stt_message(conn, text) - conn.executor.submit(conn.chat, text) + # 意图未被处理,继续常规聊天流程,使用实际文本内容 + await send_stt_message(conn, actual_text) + conn.executor.submit(conn.chat, actual_text) async def no_voice_close_connect(conn, have_voice): diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index eb9f565f..486e6d90 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -76,7 +76,6 @@ async def sendAudio(conn, audios, pre_buffer=True): frame_duration = 60 # 帧时长(毫秒),匹配 Opus 编码 start_time = time.perf_counter() play_position = 0 - last_reset_time = time.perf_counter() # 记录最后的重置时间 # 仅当第一句话时执行预缓冲 if pre_buffer: @@ -137,7 +136,21 @@ async def send_stt_message(conn, text): return """发送 STT 状态消息""" - stt_text = get_string_no_punctuation_or_emoji(text) + + # 解析JSON格式,提取实际的用户说话内容 + display_text = text + try: + # 尝试解析JSON格式 + if text.strip().startswith('{') and text.strip().endswith('}'): + parsed_data = json.loads(text) + if isinstance(parsed_data, dict) and "content" in parsed_data: + # 如果是包含说话人信息的JSON格式,只显示content部分 + display_text = parsed_data["content"] + except (json.JSONDecodeError, TypeError): + # 如果不是JSON格式,直接使用原始文本 + display_text = text + + stt_text = get_string_no_punctuation_or_emoji(display_text) await conn.websocket.send( json.dumps({"type": "stt", "text": stt_text, "session_id": conn.session_id}) ) diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 71098a20..ef9fa01e 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -1,19 +1,23 @@ import os import wave -import copy import uuid import queue import asyncio import traceback import threading import opuslib_next +import json +import io +import time +import concurrent.futures from abc import ABC, abstractmethod from config.logger import setup_logging -from typing import Optional, Tuple, List +from typing import Optional, Tuple, List, Dict, Any from core.handle.receiveAudioHandle import startToChat from core.handle.reportHandle import enqueue_asr_report from core.utils.util import remove_punctuation_and_length from core.handle.receiveAudioHandle import handleAudioMessage +from core.utils.voiceprint_provider import VoiceprintProvider TAG = __name__ logger = setup_logging() @@ -21,13 +25,16 @@ logger = setup_logging() class ASRProviderBase(ABC): def __init__(self): - pass + self.voiceprint_provider = None + + def init_voiceprint(self, voiceprint_config: dict): + """初始化声纹识别""" + if voiceprint_config: + self.voiceprint_provider = VoiceprintProvider(voiceprint_config) + logger.bind(tag=TAG).info("声纹识别模块已初始化") # 打开音频通道 - # 这里默认是非流式的处理方式 - # 流式处理方式请在子类中重写 async def open_audio_channels(self, conn): - # tts 消化线程 conn.asr_priority_thread = threading.Thread( target=self.asr_text_priority_thread, args=(conn,), daemon=True ) @@ -52,41 +59,173 @@ class ASRProviderBase(ABC): continue # 接收音频 - # 这里默认是非流式的处理方式 - # 流式处理方式请在子类中重写 async def receive_audio(self, conn, audio, audio_have_voice): if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": have_voice = audio_have_voice else: have_voice = conn.client_have_voice - # 如果本次没有声音,本段也没声音,就把声音丢弃了 + conn.asr_audio.append(audio) - if have_voice == False and conn.client_have_voice == False: + if not have_voice and not conn.client_have_voice: conn.asr_audio = conn.asr_audio[-10:] return - # 如果本段有声音,且已经停止了 if conn.client_voice_stop: - asr_audio_task = copy.deepcopy(conn.asr_audio) + asr_audio_task = conn.asr_audio.copy() conn.asr_audio.clear() - - # 音频太短了,无法识别 conn.reset_vad_states() + if len(asr_audio_task) > 15: await self.handle_voice_stop(conn, asr_audio_task) # 处理语音停止 - async def handle_voice_stop(self, conn, asr_audio_task): - raw_text, _ = await self.speech_to_text( - asr_audio_task, conn.session_id, conn.audio_format - ) # 确保ASR模块返回原始文本 - conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}") - text_len, _ = remove_punctuation_and_length(raw_text) - self.stop_ws_connection() - if text_len > 0: - # 使用自定义模块进行上报 - await startToChat(conn, raw_text) - enqueue_asr_report(conn, raw_text, asr_audio_task) + async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]): + """并行处理ASR和声纹识别""" + try: + total_start_time = time.monotonic() + + # 准备音频数据 + if conn.audio_format == "pcm": + pcm_data = asr_audio_task + else: + pcm_data = self.decode_opus(asr_audio_task) + + combined_pcm_data = b"".join(pcm_data) + + # 预先准备WAV数据 + wav_data = None + if self.voiceprint_provider and combined_pcm_data: + wav_data = self._pcm_to_wav(combined_pcm_data) + + + # 定义ASR任务 + def run_asr(): + start_time = time.monotonic() + try: + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete( + self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) + ) + end_time = time.monotonic() + logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s") + return result + finally: + loop.close() + except Exception as e: + end_time = time.monotonic() + logger.bind(tag=TAG).error(f"ASR失败: {e}") + return ("", None) + + # 定义声纹识别任务 + def run_voiceprint(): + if not wav_data: + return None + start_time = time.monotonic() + try: + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete( + self.voiceprint_provider.identify_speaker(wav_data, conn.session_id) + ) + return result + finally: + loop.close() + except Exception as e: + logger.bind(tag=TAG).error(f"声纹识别失败: {e}") + return None + + # 使用线程池执行器并行运行 + parallel_start_time = time.monotonic() + + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: + asr_future = thread_executor.submit(run_asr) + + if self.voiceprint_provider and wav_data: + voiceprint_future = thread_executor.submit(run_voiceprint) + + # 等待两个线程都完成 + asr_result = asr_future.result(timeout=15) + voiceprint_result = voiceprint_future.result(timeout=15) + + results = {"asr": asr_result, "voiceprint": voiceprint_result} + else: + asr_result = asr_future.result(timeout=15) + results = {"asr": asr_result, "voiceprint": None} + + parallel_execution_time = time.monotonic() - parallel_start_time + + # 处理结果 + raw_text, file_path = results.get("asr", ("", None)) + speaker_name = results.get("voiceprint", None) + + # 记录识别结果 + if raw_text: + logger.bind(tag=TAG).info(f"识别文本: {raw_text}") + if speaker_name: + logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") + + # 性能监控 + total_time = time.monotonic() - total_start_time + logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s") + + # 检查文本长度 + text_len, _ = remove_punctuation_and_length(raw_text) + self.stop_ws_connection() + + if text_len > 0: + # 构建包含说话人信息的JSON字符串 + enhanced_text = self._build_enhanced_text(raw_text, speaker_name) + + # 使用自定义模块进行上报 + await startToChat(conn, enhanced_text) + enqueue_asr_report(conn, enhanced_text, asr_audio_task) + + except Exception as e: + logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") + import traceback + logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}") + + def _build_enhanced_text(self, text: str, speaker_name: Optional[str]) -> str: + """构建包含说话人信息的文本""" + if speaker_name and speaker_name.strip(): + return json.dumps({ + "speaker": speaker_name, + "content": text + }, ensure_ascii=False) + else: + return text + + def _pcm_to_wav(self, pcm_data: bytes) -> bytes: + """将PCM数据转换为WAV格式""" + if len(pcm_data) == 0: + logger.bind(tag=TAG).warning("PCM数据为空,无法转换WAV") + return b"" + + # 确保数据长度是偶数(16位音频) + if len(pcm_data) % 2 != 0: + pcm_data = pcm_data[:-1] + + # 创建WAV文件头 + wav_buffer = io.BytesIO() + try: + with wave.open(wav_buffer, 'wb') as wav_file: + wav_file.setnchannels(1) # 单声道 + wav_file.setsampwidth(2) # 16位 + wav_file.setframerate(16000) # 16kHz采样率 + wav_file.writeframes(pcm_data) + + wav_buffer.seek(0) + wav_data = wav_buffer.read() + + return wav_data + except Exception as e: + logger.bind(tag=TAG).error(f"WAV转换失败: {e}") + return b"" def stop_ws_connection(self): pass @@ -113,27 +252,29 @@ class ASRProviderBase(ABC): pass @staticmethod - def decode_opus(opus_data: List[bytes]) -> bytes: + def decode_opus(opus_data: List[bytes]) -> List[bytes]: """将Opus音频数据解码为PCM数据""" try: - decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道 + decoder = opuslib_next.Decoder(16000, 1) pcm_data = [] - buffer_size = 960 # 每次处理960个采样点 - - for opus_packet in opus_data: + buffer_size = 960 # 每次处理960个采样点 (60ms at 16kHz) + + for i, opus_packet in enumerate(opus_data): try: - # 使用较小的缓冲区大小进行处理 + if not opus_packet or len(opus_packet) == 0: + continue + pcm_frame = decoder.decode(opus_packet, buffer_size) - if pcm_frame: + if pcm_frame and len(pcm_frame) > 0: pcm_data.append(pcm_frame) + except opuslib_next.OpusError as e: - logger.bind(tag=TAG).warning(f"Opus解码错误,跳过当前数据包: {e}") - continue + logger.bind(tag=TAG).warning(f"Opus解码错误,跳过数据包 {i}: {e}") except Exception as e: - logger.bind(tag=TAG).error(f"音频处理错误: {e}", exc_info=True) - continue - + logger.bind(tag=TAG).error(f"音频处理错误,数据包 {i}: {e}") + return pcm_data + except Exception as e: - logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}", exc_info=True) + logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}") return [] diff --git a/main/xiaozhi-server/core/utils/dialogue.py b/main/xiaozhi-server/core/utils/dialogue.py index 15eb06f3..41cffca0 100644 --- a/main/xiaozhi-server/core/utils/dialogue.py +++ b/main/xiaozhi-server/core/utils/dialogue.py @@ -2,6 +2,7 @@ import uuid import re from typing import List, Dict from datetime import datetime +from config.settings import load_config class Message: @@ -46,10 +47,9 @@ class Dialogue: dialogue.append({"role": m.role, "content": m.content}) def get_llm_dialogue(self) -> List[Dict[str, str]]: - dialogue = [] - for m in self.dialogue: - self.getMessages(m, dialogue) - return dialogue + # 直接调用get_llm_dialogue_with_memory,传入None作为memory_str + # 这样确保说话人功能在所有调用路径下都生效 + return self.get_llm_dialogue_with_memory(None) def update_system_message(self, new_content: str): """更新或添加系统消息""" @@ -63,10 +63,7 @@ class Dialogue: def get_llm_dialogue_with_memory( self, memory_str: str = None ) -> List[Dict[str, str]]: - if memory_str is None or len(memory_str) == 0: - return self.get_llm_dialogue() - - # 构建带记忆的对话 + # 构建对话 dialogue = [] # 添加系统提示和记忆 @@ -75,6 +72,33 @@ class Dialogue: ) if system_message: + # 基础系统提示 + enhanced_system_prompt = system_message.content + + # 添加说话人个性化描述 + try: + config = load_config() + voiceprint_config = config.get("voiceprint", {}) + speakers = voiceprint_config.get("speakers", []) + + if speakers: + enhanced_system_prompt += "\n\n" + for speaker_str in speakers: + try: + parts = speaker_str.split(",", 2) + if len(parts) >= 2: + speaker_id = parts[0].strip() + name = parts[1].strip() + # 如果描述为空,则为"" + description = parts[2].strip() if len(parts) >= 3 else "" + enhanced_system_prompt += f"\n- {name}:{description}" + except: + pass + enhanced_system_prompt += "\n\n" + except: + # 配置读取失败时忽略错误,不影响其他功能 + pass + # 使用正则表达式匹配 标签,不管中间有什么内容 enhanced_system_prompt = re.sub( r".*?", diff --git a/main/xiaozhi-server/core/utils/modules_initialize.py b/main/xiaozhi-server/core/utils/modules_initialize.py index f2e3968e..a8db1630 100644 --- a/main/xiaozhi-server/core/utils/modules_initialize.py +++ b/main/xiaozhi-server/core/utils/modules_initialize.py @@ -125,4 +125,13 @@ def initialize_asr(config): config["ASR"][select_asr_module], str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"), ) + + # 初始化声纹识别功能 + voiceprint_config = config.get("voiceprint") + if voiceprint_config and voiceprint_config.get("url") and voiceprint_config.get("speakers"): + new_asr.init_voiceprint(voiceprint_config) + logger.bind(tag=TAG).info("ASR模块声纹识别功能已启用") + else: + logger.bind(tag=TAG).info("ASR模块声纹识别功能已禁用") + return new_asr diff --git a/main/xiaozhi-server/core/utils/voiceprint_provider.py b/main/xiaozhi-server/core/utils/voiceprint_provider.py new file mode 100644 index 00000000..deec20c8 --- /dev/null +++ b/main/xiaozhi-server/core/utils/voiceprint_provider.py @@ -0,0 +1,133 @@ +import asyncio +import json +import time +import aiohttp +from urllib.parse import urlparse, parse_qs +from typing import Optional, Dict +from config.logger import setup_logging + +TAG = __name__ +logger = setup_logging() + + +class VoiceprintProvider: + """声纹识别服务提供者""" + + def __init__(self, config: dict): + self.original_url = config.get("url", "") + self.speakers = config.get("speakers", []) + self.speaker_map = self._parse_speakers() + + # 解析API地址和密钥 + self.api_url = None + self.api_key = None + self.speaker_ids = [] + + if not self.original_url: + logger.bind(tag=TAG).warning("声纹识别URL未配置,声纹识别将被禁用") + self.enabled = False + else: + # 解析URL和key + parsed_url = urlparse(self.original_url) + base_url = f"{parsed_url.scheme}://{parsed_url.netloc}" + + # 从查询参数中提取key + query_params = parse_qs(parsed_url.query) + self.api_key = query_params.get('key', [''])[0] + + if not self.api_key: + logger.bind(tag=TAG).error("URL中未找到key参数,声纹识别将被禁用") + self.enabled = False + else: + # 构造identify接口地址 + self.api_url = f"{base_url}/voiceprint/identify" + + # 提取speaker_ids + for speaker_str in self.speakers: + try: + parts = speaker_str.split(",", 2) + if len(parts) >= 1: + speaker_id = parts[0].strip() + self.speaker_ids.append(speaker_id) + except Exception: + continue + + # 检查是否有有效的说话人配置 + if not self.speaker_ids: + logger.bind(tag=TAG).warning("未配置有效的说话人,声纹识别将被禁用") + self.enabled = False + else: + self.enabled = True + logger.bind(tag=TAG).info(f"声纹识别已配置: API={self.api_url}, 说话人={len(self.speaker_ids)}个") + + def _parse_speakers(self) -> Dict[str, Dict[str, str]]: + """解析说话人配置""" + speaker_map = {} + for speaker_str in self.speakers: + try: + parts = speaker_str.split(",", 2) + if len(parts) >= 3: + speaker_id, name, description = parts[0].strip(), parts[1].strip(), parts[2].strip() + speaker_map[speaker_id] = { + "name": name, + "description": description + } + except Exception as e: + logger.bind(tag=TAG).warning(f"解析说话人配置失败: {speaker_str}, 错误: {e}") + return speaker_map + + async def identify_speaker(self, audio_data: bytes, session_id: str) -> Optional[str]: + """识别说话人""" + if not self.enabled or not self.api_url or not self.api_key: + logger.bind(tag=TAG).debug("声纹识别功能已禁用或未配置,跳过识别") + return None + + try: + api_start_time = time.monotonic() + + # 准备请求头 + headers = { + 'Authorization': f'Bearer {self.api_key}', + 'Accept': 'application/json' + } + + # 准备multipart/form-data数据 + data = aiohttp.FormData() + data.add_field('speaker_ids', ','.join(self.speaker_ids)) + data.add_field('file', audio_data, filename='audio.wav', content_type='audio/wav') + + timeout = aiohttp.ClientTimeout(total=10) + + # 网络请求 + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post(self.api_url, headers=headers, data=data) as response: + + if response.status == 200: + result = await response.json() + speaker_id = result.get("speaker_id") + score = result.get("score", 0) + total_elapsed_time = time.monotonic() - api_start_time + + logger.bind(tag=TAG).info(f"声纹识别耗时: {total_elapsed_time:.3f}s") + + # 置信度检查 + if score < 0.5: + logger.bind(tag=TAG).warning(f"声纹识别置信度较低: {score:.3f}") + + if speaker_id and speaker_id in self.speaker_map: + result_name = self.speaker_map[speaker_id]["name"] + return result_name + else: + return None + else: + logger.bind(tag=TAG).error(f"声纹识别API错误: HTTP {response.status}") + return None + + except asyncio.TimeoutError: + elapsed = time.monotonic() - api_start_time + logger.bind(tag=TAG).error(f"声纹识别超时: {elapsed:.3f}s") + return None + except Exception as e: + elapsed = time.monotonic() - api_start_time + logger.bind(tag=TAG).error(f"声纹识别失败: {e}") + return None