Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-26 17:13:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

merge main

Browse Source
This commit is contained in:
lizhongxiang
2025-03-13 14:24:18 +08:00
parent 0e8edfe002 a8f06f5718
commit 91bdc05a38
52 changed files with 1013 additions and 1184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/connection.py
+85 -49
View File
@@ -11,7 +11,7 @@ import websockets
from typing import Dict, Any
from core.utils.dialogue import Message, Dialogue
from core.handle.textHandle import handleTextMessage
from core.utils.util import get_string_no_punctuation_or_emoji
from core.utils.util import get_string_no_punctuation_or_emoji, extract_json_from_string
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from core.handle.sendAudioHandle import sendAudioMessage, sendAudioMessageStream
from core.handle.receiveAudioHandle import handleAudioMessage
@@ -100,8 +100,6 @@ class ConnectionHandler:
if self.config["selected_module"]["Intent"] == 'function_call':
self.use_function_call_mode = True
self.logger.bind(tag=TAG).info(f"use_function_call_mode:{self.use_function_call_mode}")
async def handle_connection(self, ws):
try:
# 获取并验证headers
@@ -330,8 +328,7 @@ class ConnectionHandler:
response_message = []
processed_chars = 0 # 跟踪已处理的字符位置
function_call_data = None # 存储function call数据
try:
start_time = time.time()
@@ -339,7 +336,7 @@ class ConnectionHandler:
future = asyncio.run_coroutine_threadsafe(self.memory.query_memory(query), self.loop)
memory_str = future.result()
# self.logger.bind(tag=TAG).info(f"记忆内容: {memory_str}")
#self.logger.bind(tag=TAG).info(f"对话记录: {self.dialogue.get_llm_dialogue_with_memory(memory_str)}")
# 使用支持functions的streaming接口
llm_responses = self.llm.response_with_functions(
@@ -355,48 +352,61 @@ class ConnectionHandler:
text_index = 0
# 处理流式响应
tool_call_flag = False
function_name = None
function_id = None
function_arguments = ""
content_arguments = ""
for response in llm_responses:
if response["type"] == "content":
content = response["content"]
response_message.append(content)
content, tools_call = response
if content is not None and len(content)>0:
if len(response_message)<=0 and content=="```":
tool_call_flag = True
if self.client_abort:
break
if tools_call is not None:
tool_call_flag = True
if tools_call[0].id is not None:
function_id = tools_call[0].id
if tools_call[0].function.name is not None:
function_name = tools_call[0].function.name
if tools_call[0].function.arguments is not None:
function_arguments += tools_call[0].function.arguments
end_time = time.time()
self.logger.bind(tag=TAG).debug(f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}")
if content is not None and len(content) > 0:
if tool_call_flag:
content_arguments+=content
else:
response_message.append(content)
# 处理文本分段和TTS逻辑
# 合并当前全部文本并处理未分割部分
full_text = "".join(response_message)
current_text = full_text[processed_chars:] # 从未处理的位置开始
if self.client_abort:
break
# 查找最后一个有效标点
punctuations = ("。", "？", "！", "；", "：")
last_punct_pos = -1
for punct in punctuations:
pos = current_text.rfind(punct)
if pos > last_punct_pos:
last_punct_pos = pos
end_time = time.time()
self.logger.bind(tag=TAG).debug(f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}")
# 找到分割点则处理
if last_punct_pos != -1:
segment_text_raw = current_text[:last_punct_pos + 1]
segment_text = get_string_no_punctuation_or_emoji(segment_text_raw)
if segment_text:
text_index += 1
self.recode_first_last_text(segment_text, text_index)
future = self.executor.submit(self.speak_and_play, segment_text, text_index)
self.tts_queue.put(future)
processed_chars += len(segment_text_raw) # 更新已处理字符位置
# 处理文本分段和TTS逻辑
# 合并当前全部文本并处理未分割部分
full_text = "".join(response_message)
current_text = full_text[processed_chars:] # 从未处理的位置开始
elif response["type"] == "function_call":
# Extract function call data
function_call_data = {
"name": response["function_call"]["function"]["name"],
"arguments": response["function_call"]["function"]["arguments"]
}
self.logger.bind(tag=TAG).info(f"Function call detected: {function_call_data}")
# 查找最后一个有效标点
punctuations = ("。", "？", "！", "；", "：")
last_punct_pos = -1
for punct in punctuations:
pos = current_text.rfind(punct)
if pos > last_punct_pos:
last_punct_pos = pos
# 找到分割点则处理
if last_punct_pos != -1:
segment_text_raw = current_text[:last_punct_pos + 1]
segment_text = get_string_no_punctuation_or_emoji(segment_text_raw)
if segment_text:
text_index += 1
self.recode_first_last_text(segment_text, text_index)
future = self.executor.submit(self.speak_and_play, segment_text, text_index)
self.tts_queue.put(future)
processed_chars += len(segment_text_raw) # 更新已处理字符位置
# 处理最后剩余的文本
full_text = "".join(response_message)
@@ -410,23 +420,49 @@ class ConnectionHandler:
self.tts_queue.put(future)
# 存储对话内容
self.dialogue.put(Message(role="assistant", content="".join(response_message)))
if len(response_message)>0:
self.dialogue.put(Message(role="assistant", content="".join(response_message)))
# 处理function call
if function_call_data:
if tool_call_flag:
if function_id is None:
a = extract_json_from_string(content_arguments)
if a is not None:
content_arguments_json = json.loads(a)
function_name = content_arguments_json["function_name"]
function_arguments = json.dumps(content_arguments_json["args"], ensure_ascii=False)
function_id = str(uuid.uuid4().hex)
else:
return []
function_arguments = json.loads(function_arguments)
self.logger.bind(tag=TAG).info(f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}")
function_call_data = {
"name": function_name,
"id": function_id,
"arguments": function_arguments
}
result = handle_llm_function_call(self, function_call_data)
if result.action == Action.RESPONSE:
text = result.response
text_index += 1
self.recode_first_last_text(text, text_index)
future = self.executor.submit(self.speak_and_play, text, text_index)
self.tts_queue.put(future)
self._handle_function_result(result, function_call_data, text_index+1)
self.llm_finish_task = True
self.logger.bind(tag=TAG).debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False))
return True
def _handle_function_result(self, result, function_call_data, text_index):
if result.action == Action.RESPONSE: # 直接回复前端
text = result.response
self.recode_first_last_text(text, text_index)
future = self.executor.submit(self.speak_and_play, text, text_index)
self.tts_queue.put(future)
self.dialogue.put(Message(role="assistant", content=text))
if result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
text = result.response
if result.action == Action.NOTFOUND:
text = result.response
def _tts_priority_thread(self):
if self.tts_stream:
self._tts_priority_thread_stream()
main/xiaozhi-server/core/handle/intentHandler.py
+5 -8
View File
@@ -2,6 +2,7 @@ from config.logger import setup_logging
import json
from core.handle.sendAudioHandle import send_stt_message
from core.utils.dialogue import Message
from core.utils.util import remove_punctuation_and_length
from config.functionCallConfig import FunctionCallConfig
import asyncio
from enum import Enum
@@ -67,7 +68,7 @@ def handle_llm_function_call(conn, function_call_data):
except Exception as e:
logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
else:
return ActionResponse(action=Action.NOTFOUND, result="没有找到对应的函数", response="")
return ActionResponse(action=Action.NOTFOUND, result="没有找到对应的函数", response="没有找到对应的函数处理相对于的功能呢，你可以需要添加预设的对应函数处理呢")
except Exception as e:
logger.bind(tag=TAG).error(f"处理function call错误: {e}")
@@ -93,8 +94,6 @@ async def handle_user_intent(conn, text):
# 使用支持function calling的聊天方法,不再进行意图分析
return False
logger.bind(tag=TAG).info(f"分析用户意图: {text}")
# 使用LLM进行意图分析
intent = await analyze_intent_with_llm(conn, text)
@@ -107,6 +106,7 @@ async def handle_user_intent(conn, text):
async def check_direct_exit(conn, text):
"""检查是否有明确的退出命令"""
_, text = remove_punctuation_and_length(text)
cmd_exit = conn.cmd_exit
for cmd in cmd_exit:
if text == cmd:
@@ -122,13 +122,10 @@ async def analyze_intent_with_llm(conn, text):
logger.bind(tag=TAG).warning("意图识别服务未初始化")
return None
# 创建对话历史记录
# 对话历史记录
dialogue = conn.dialogue
dialogue.put(Message(role="user", content=text))
try:
intent_result = await conn.intent.detect_intent(dialogue.dialogue)
logger.bind(tag=TAG).info(f"意图识别结果: {intent_result}")
intent_result = await conn.intent.detect_intent(conn, dialogue.dialogue, text)
# 尝试解析JSON结果
try:
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+3 -2
View File
@@ -67,8 +67,9 @@ async def no_voice_close_connect(conn):
else:
no_voice_time = time.time() * 1000 - conn.client_no_voice_last_time
close_connection_no_voice_time = conn.config.get("close_connection_no_voice_time", 120)
if no_voice_time > 1000 * close_connection_no_voice_time:
if not conn.close_after_chat and no_voice_time > 1000 * close_connection_no_voice_time:
conn.close_after_chat = True
conn.client_abort = False
conn.asr_server_receive = False
prompt = "时间过得真快，我都好久没说话了。请你用十个字左右话跟我告别，以“再见”或“拜拜”为结尾"
prompt = "请你以“时间过得真快”未来头，用富有感情、依依不舍的话来结束这场对话吧。"
await startToChat(conn, prompt)
main/xiaozhi-server/core/handle/sendAudioHandle.py
+17 -8
View File
@@ -75,28 +75,37 @@ async def sendAudioMessage(conn, audios, text, text_index=0):
logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
await send_tts_message(conn, "sentence_start", text)
# 初始化流控参数
frame_duration = 60 # 毫秒
start_time = time.perf_counter() # 使用高精度计时器
play_position = 0 # 已播放的时长（毫秒）
# 流控参数优化
original_frame_duration = 60 # 原始帧时长（毫秒）
adjusted_frame_duration = int(original_frame_duration * 0.8) # 缩短20%
total_frames = len(audios) # 获取总帧数
compensation = total_frames * (original_frame_duration - adjusted_frame_duration) / 1000 # 补偿时间（秒）
start_time = time.perf_counter()
play_position = 0 # 已播放时长（毫秒）
for opus_packet in audios:
if conn.client_abort:
return
# 计算当前包的预期发送时间
# 计算带加速因子的预期时间
expected_time = start_time + (play_position / 1000)
current_time = time.perf_counter()
# 等待直到预期时间
# 流控等待（使用加速后的帧时长）
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
# 发送音频包
await conn.websocket.send(opus_packet)
play_position += frame_duration # 更新播放位置
play_position += adjusted_frame_duration # 使用调整后的帧时长
# 补偿因加速损失的时长
if compensation > 0:
await asyncio.sleep(compensation)
await send_tts_message(conn, "sentence_end", text)
# 发送结束消息（如果是最后一个文本）
if conn.llm_finish_task and text_index == conn.tts_last_text_index:
await send_tts_message(conn, 'stop', None)
main/xiaozhi-server/core/providers/intent/base.py
+3 -2
View File
@@ -5,6 +5,7 @@ from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
class IntentProviderBase(ABC):
def __init__(self, config):
self.config = config
@@ -17,9 +18,9 @@ class IntentProviderBase(ABC):
def set_llm(self, llm):
self.llm = llm
logger.bind(tag=TAG).debug("Set LLM for intent provider")
@abstractmethod
async def detect_intent(self, dialogue_history: List[Dict]) -> str:
async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str:
"""
检测用户最后一句话的意图
Args:
main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py
+69 -21
View File
@@ -1,7 +1,7 @@
from typing import List, Dict
from ..base import IntentProviderBase
from config.logger import setup_logging
import re
TAG = __name__
logger = setup_logging()
@@ -20,42 +20,90 @@ class IntentProvider(IntentProviderBase):
格式化后的系统提示词
"""
intent_list = []
"""
"continue_chat": "1.继续聊天, 除了播放音乐和结束聊天的时候的选项, 比如日常的聊天和问候, 对话等",
"end_chat": "2.结束聊天, 用户发来如再见之类的表示结束的话, 不想再进行对话的时候",
"play_music": "3.播放音乐, 用户希望你可以播放音乐, 只用于播放音乐的意图"
"""
for key, value in self.intent_options.items():
if key == "play_music":
intent_list.append(f"{value} [歌名]")
intent_list.append("3.播放音乐, 用户希望你可以播放音乐, 只用于播放音乐的意图")
elif key == "end_chat":
intent_list.append("2.结束聊天, 用户发来如再见之类的表示结束的话, 不想再进行对话的时候")
elif key == "continue_chat":
intent_list.append("1.继续聊天, 除了播放音乐和结束聊天的时候的选项, 比如日常的聊天和问候, 对话等")
else:
intent_list.append(value)
# "如果是唱歌、听歌、播放音乐，请指定歌名，格式为'播放音乐 [识别出的歌名]'。\n"
# "如果听不出具体歌名，可以返回'随机播放音乐'。\n"
# "只需要返回意图结果的json，不要解释。"
# "返回格式如下：\n"
prompt = (
"你是一个意图识别助手。你需要根据和用户的对话记录，重点分析用户的最后一句话，判断用户意图属于以下哪一类：\n"
f"{', '.join(intent_list)}\n"
"如果是唱歌、听歌、播放音乐，请指定歌名，格式为'播放音乐 [识别出的歌名]'。\n"
"如果听不出具体歌名，可以返回'随机播放音乐'。\n"
"只需要返回意图结果的json，不要解释。"
"返回格式如下：\n"
"{intent: '用户意图'}"
"你是一个意图识别助手。你需要根据和用户的对话记录，重点分析用户的最后一句话，判断用户意图属于以下哪一类(使用<start>和<end>标志)：\n"
"<start>"
f"{', '.join(intent_list)}"
"<end>\n"
"你需要按照以下的步骤处理用户的对话"
"1. 思考出对话的意图是哪一类的"
"2. 属于1和2的意图, 直接返回，返回格式如下：\n"
"{intent: '用户意图'}\n"
"3. 属于3的意图，则继续分析用户希望播放的音乐\n"
"4. 如果无法识别出具体歌名，可以返回'随机播放音乐'\n"
"{intent: '播放音乐 [获取的音乐名字]'}\n"
"下面是几个处理的示例(思考的内容不返回, 只返回json部分, 无额外的内容)\n"
"```"
"用户: 你今天怎么样?\n"
"思考(不返回): 用户发来的数据是一个问候语，属于继续聊天的意图, 是种类1, 种类1的需求是直接返回\n"
"返回结果: {intent: '继续聊天'}\n"
"```"
"用户: 我今天有点累了, 我们明天再聊吧\n"
"思考(不返回): 用户表达了今天不想继续对话，属于结束聊天的意图, 是种类2, 种类2的需求是直接返回\n"
"返回结果: {intent: '结束聊天'}\n"
"```"
"用户: 我今天有点累了, 我们明天再聊吧\n"
"思考(不返回): 用户表达了今天不想继续对话，属于结束聊天的意图, 是种类2, 种类2的需求是直接返回\n"
"返回结果: {intent: '结束聊天'}\n"
"```"
"用户: 你可以播放一首中秋月给我听吗\n"
"思考(不返回): 用户表达了想听音乐的续签，属于播放音乐的意图, 是种类3, 种类3的需求需要继续判断播放的音乐, 这里用户希望的歌曲名明确给出是中秋月\n"
"返回结果: {intent: '播放音乐 [中秋月]'}\n"
"```"
"你现在可以使用的音乐的名称如下(使用<start>和<end>标志):\n"
)
return prompt
async def detect_intent(self, dialogue_history: List[Dict]) -> str:
async def detect_intent(self, conn, dialogue_history: List[Dict], text:str) -> str:
if not self.llm:
raise ValueError("LLM provider not set")
# 构建用户最后一句话的提示
msgStr = ""
for msg in dialogue_history:
if msg.role == "user":
msgStr += f"User: {msg.content}\n"
elif msg.role== "assistant":
msgStr += f"Assistant: {msg.content}\n"
user_prompt = f"请分析用户的意图：\n{msgStr}"
# 只使用最后两句即可
if len(dialogue_history) >= 2:
# 保证最少有两句话的时候处理
msgStr += f"{dialogue_history[-2].role}: {dialogue_history[-2].content}\n"
msgStr += f"{dialogue_history[-1].role}: {dialogue_history[-1].content}\n"
msgStr += f"User: {text}\n"
user_prompt = f"当前的对话如下：\n{msgStr}"
prompt_music = f"{self.promot}\n<start>{conn.music_handler.music_files}\n<end>"
logger.bind(tag=TAG).debug(f"User prompt: {prompt_music}")
# 使用LLM进行意图识别
intent = self.llm.response_no_stream(
system_prompt=self.promot,
system_prompt=prompt_music,
user_prompt=user_prompt
)
# 使用正则表达式提取大括号中的内容
# 使用正则表达式提取 {} 中的内容
match = re.search(r'\{.*?\}', intent)
if match:
result = match.group(0) # 获取匹配到的内容（包含 {}）
print(result) # 输出：{intent: '播放音乐 [中秋月]'}
intent = result
else:
intent = "{intent: '继续聊天'}"
logger.bind(tag=TAG).info(f"Detected intent: {intent}")
return intent.strip()
return intent.strip()
main/xiaozhi-server/core/providers/intent/nointent/nointent.py
+3 -1
View File
@@ -5,12 +5,14 @@ from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
class IntentProvider(IntentProviderBase):
async def detect_intent(self, dialogue_history: List[Dict]) -> str:
async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str:
"""
默认的意图识别实现，始终返回继续聊天
Args:
dialogue_history: 对话历史记录列表
text: 本次对话记录
Returns:
固定返回"继续聊天"
"""
main/xiaozhi-server/core/providers/llm/coze/coze.py
+14 -1
View File
@@ -11,19 +11,31 @@ from cozepy import Coze, TokenAuth, Message, ChatStatus, MessageContentType, Cha
TAG = __name__
logger = setup_logging()
class LLMProvider(LLMProviderBase):
def __init__(self, config):
self.personal_access_token = config.get("personal_access_token")
self.bot_id = config.get("bot_id")
self.user_id = config.get("user_id")
self.session_conversation_map = {} # 存储session_id和conversation_id的映射
def response(self, session_id, dialogue):
coze_api_token = self.personal_access_token
coze_api_base = COZE_CN_BASE_URL
last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
coze = Coze(auth=TokenAuth(token=coze_api_token), base_url=coze_api_base)
conversation_id = self.session_conversation_map.get(session_id)
# 如果没有找到conversation_id，则创建新的对话
if not conversation_id:
conversation = coze.conversations.create(
messages=[
]
)
conversation_id = conversation.id
self.session_conversation_map[session_id] = conversation_id # 更新映射
for event in coze.chat.stream(
bot_id=self.bot_id,
@@ -31,6 +43,7 @@ class LLMProvider(LLMProviderBase):
additional_messages=[
Message.build_user_question_text(last_msg["content"]),
],
conversation_id=conversation_id,
):
if event.event == ChatEventType.CONVERSATION_MESSAGE_DELTA:
print(event.message.content, end="", flush=True)
main/xiaozhi-server/core/providers/llm/ollama/ollama.py
+1 -23
View File
@@ -51,30 +51,8 @@ class LLMProvider(LLMProviderBase):
tools=functions,
)
current_function_call = None
current_content = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
current_content += delta.content
yield {"type": "content", "content": delta.content}
if delta.tool_calls:
tool_call = delta.tool_calls[0]
# Handle the function call data using proper attribute access
if not current_function_call:
current_function_call = {
"function": {
"name": tool_call.function.name,
"arguments": tool_call.function.arguments
}
}
if current_function_call:
logger.bind(tag=TAG).debug(f"ollama Function call detected: {current_function_call}")
yield {"type": "function_call", "function_call": current_function_call}
yield chunk.choices[0].delta.content, chunk.choices[0].delta.tool_calls
except Exception as e:
logger.bind(tag=TAG).error(f"Error in Ollama function call: {e}")
main/xiaozhi-server/core/providers/llm/openai/openai.py
+1 -23
View File
@@ -53,30 +53,8 @@ class LLMProvider(LLMProviderBase):
tools=functions,
)
current_function_call = None
current_content = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
current_content += delta.content
yield {"type": "content", "content": delta.content}
if delta.tool_calls:
tool_call = delta.tool_calls[0]
# Handle the function call data using proper attribute access
if not current_function_call:
current_function_call = {
"function": {
"name": tool_call.function.name,
"arguments": tool_call.function.arguments
}
}
if current_function_call:
logger.bind(tag=TAG).debug(f"openai Function call detected: {current_function_call}")
yield {"type": "function_call", "function_call": current_function_call}
yield chunk.choices[0].delta.content, chunk.choices[0].delta.tool_calls
except Exception as e:
self.logger.bind(tag=TAG).error(f"Error in function call streaming: {e}")
main/xiaozhi-server/core/utils/util.py
+9
View File
@@ -4,6 +4,7 @@ import yaml
import socket
import subprocess
import logging
import re
def get_project_dir():
@@ -119,3 +120,11 @@ def check_ffmpeg_installed():
error_msg += "1、按照项目的安装文档，正确进入conda环境\n"
error_msg += "2、查阅安装文档，如何在conda环境中安装ffmpeg\n"
raise ValueError(error_msg)
def extract_json_from_string(input_string):
"""提取字符串中的 JSON 部分"""
pattern = r'(\{.*\})'
match = re.search(pattern, input_string)
if match:
return match.group(1) # 返回提取的 JSON 字符串
return None
main/xiaozhi-server/core/websocket_server.py
+2
View File
@@ -65,6 +65,8 @@ class WebSocketServer:
server_config = self.config["server"]
host = server_config["ip"]
port = server_config["port"]
selected_module = self.config.get("selected_module")
self.logger.bind(tag=TAG).info(f"selected_module: {selected_module}")
self.logger.bind(tag=TAG).info("Server is running at ws://{}:{}", get_local_ip(), port)
self.logger.bind(tag=TAG).info("=======上面的地址是websocket协议地址，请勿用浏览器访问=======")