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

feat: 统一LLM错误处理并添加系统错误回复配置

Browse Source 
在多个LLM提供者中移除try-catch块，将错误处理统一到connection.py的流处理层
添加system_error_response配置项，支持自定义系统错误时的回复内容
在意图识别和流处理中捕获异常时返回配置的错误回复，避免硬编码错误信息

Fixes #2075
This commit is contained in:
huozaimengli
2026-01-25 16:48:01 +08:00
parent 275102f5b7
commit 6ae0af278b
13 changed files with 542 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/config.yaml
+3
View File
@@ -202,6 +202,9 @@ prompt: |
# 默认系统提示词模板文件 # 默认系统提示词模板文件
prompt_template: agent-base-prompt.txt prompt_template: agent-base-prompt.txt
# 系统错误时的回复
system_error_response: "主人，小智现在有点忙，我们稍后再试吧。"
# 结束语prompt # 结束语prompt
end_prompt: end_prompt:
enable: true # 是否开启结束语 enable: true # 是否开启结束语
main/xiaozhi-server/core/connection.py
+59 -36
View File
@@ -40,8 +40,10 @@ from config.logger import setup_logging, build_module_string, create_connection_
from config.manage_api_client import DeviceNotFoundException, DeviceBindException from config.manage_api_client import DeviceNotFoundException, DeviceBindException
from core.utils.prompt_manager import PromptManager from core.utils.prompt_manager import PromptManager
from core.utils.voiceprint_provider import VoiceprintProvider from core.utils.voiceprint_provider import VoiceprintProvider
from core.utils.util import get_system_error_response
from core.utils import textUtils from core.utils import textUtils
TAG = __name__ TAG = __name__
auto_import_modules("plugins_func.functions") auto_import_modules("plugins_func.functions")
@@ -869,46 +871,67 @@ class ConnectionHandler:
content_arguments = "" content_arguments = ""
self.client_abort = False self.client_abort = False
emotion_flag = True emotion_flag = True
for response in llm_responses: try:
if self.client_abort: for response in llm_responses:
break if self.client_abort:
if self.intent_type == "function_call" and functions is not None: break
content, tools_call = response if self.intent_type == "function_call" and functions is not None:
if "content" in response: content, tools_call = response
content = response["content"] if "content" in response:
tools_call = None content = response["content"]
if content is not None and len(content) > 0: tools_call = None
content_arguments += content if content is not None and len(content) > 0:
content_arguments += content
if not tool_call_flag and content_arguments.startswith("<tool_call>"): if not tool_call_flag and content_arguments.startswith("<tool_call>"):
# print("content_arguments", content_arguments) # print("content_arguments", content_arguments)
tool_call_flag = True tool_call_flag = True
if tools_call is not None and len(tools_call) > 0: if tools_call is not None and len(tools_call) > 0:
tool_call_flag = True tool_call_flag = True
self._merge_tool_calls(tool_calls_list, tools_call) self._merge_tool_calls(tool_calls_list, tools_call)
else: else:
content = response content = response
# 在llm回复中获取情绪表情，一轮对话只在开头获取一次 # 在llm回复中获取情绪表情，一轮对话只在开头获取一次
if emotion_flag and content is not None and content.strip(): if emotion_flag and content is not None and content.strip():
asyncio.run_coroutine_threadsafe( asyncio.run_coroutine_threadsafe(
textUtils.get_emotion(self, content), textUtils.get_emotion(self, content),
self.loop, 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=self.sentence_id,
sentence_type=SentenceType.MIDDLE,
content_type=ContentType.TEXT,
content_detail=content,
)
) )
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=self.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=self.sentence_id,
sentence_type=SentenceType.LAST,
content_type=ContentType.TEXT,
content_detail=get_system_error_response(self.config),
)
)
if depth == 0:
self.tts.tts_text_queue.put(
TTSMessageDTO(
sentence_id=self.sentence_id,
sentence_type=SentenceType.LAST,
content_type=ContentType.ACTION,
)
)
self.llm_finish_task = True
return
# 处理function call # 处理function call
if tool_call_flag: if tool_call_flag:
bHasError = False bHasError = False
main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py
+20 -9
View File
@@ -2,11 +2,14 @@ from typing import List, Dict
from ..base import IntentProviderBase from ..base import IntentProviderBase
from plugins_func.functions.play_music import initialize_music_handler from plugins_func.functions.play_music import initialize_music_handler
from config.logger import setup_logging from config.logger import setup_logging
from core.utils.util import get_system_error_response
import re import re
import json import json
import hashlib import hashlib
import time import time
TAG = __name__ TAG = __name__
logger = setup_logging() logger = setup_logging()
@@ -115,12 +118,16 @@ class IntentProvider(IntentProviderBase):
return prompt return prompt
def replyResult(self, text: str, original_text: str): def replyResult(self, text: str, original_text: str):
llm_result = self.llm.response_no_stream( try:
system_prompt=text, llm_result = self.llm.response_no_stream(
user_prompt="请根据以上内容，像人类一样说话的口吻回复用户，要求简洁，请直接返回结果。用户现在说：" system_prompt=text,
+ original_text, user_prompt="请根据以上内容，像人类一样说话的口吻回复用户，要求简洁，请直接返回结果。用户现在说："
) + original_text,
return llm_result )
return llm_result
except Exception as e:
logger.bind(tag=TAG).error(f"Error in generating reply result: {e}")
return get_system_error_response(self.config)
async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str: async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str:
if not self.llm: if not self.llm:
@@ -194,9 +201,13 @@ class IntentProvider(IntentProviderBase):
llm_start_time = time.time() llm_start_time = time.time()
logger.bind(tag=TAG).debug(f"开始LLM意图识别调用, 模型: {model_info}") logger.bind(tag=TAG).debug(f"开始LLM意图识别调用, 模型: {model_info}")
intent = self.llm.response_no_stream( try:
system_prompt=prompt_music, user_prompt=user_prompt intent = self.llm.response_no_stream(
) system_prompt=prompt_music, user_prompt=user_prompt
)
except Exception as e:
logger.bind(tag=TAG).error(f"Error in intent detection LLM call: {e}")
return '{"function_call": {"name": "continue_chat"}}'
# 记录LLM调用完成时间 # 记录LLM调用完成时间
llm_time = time.time() - llm_start_time llm_time = time.time() - llm_start_time
main/xiaozhi-server/core/providers/llm/AliBL/AliBL.py
+68 -73
View File
@@ -21,83 +21,78 @@ class LLMProvider(LLMProviderBase):
check_model_key("AliBLLLM", self.api_key) check_model_key("AliBLLLM", self.api_key)
def response(self, session_id, dialogue): def response(self, session_id, dialogue):
try: # 处理dialogue
# 处理dialogue if self.is_No_prompt:
if self.is_No_prompt: dialogue.pop(0)
dialogue.pop(0)
logger.bind(tag=TAG).debug(
f"【阿里百练API服务】处理后的dialogue: {dialogue}"
)
# 构造调用参数
call_params = {
"api_key": self.api_key,
"app_id": self.app_id,
"session_id": session_id,
"messages": dialogue,
# 开启SDK原生流式
"stream": True,
}
if self.memory_id != False:
# 百练memory需要prompt参数
prompt = dialogue[-1].get("content")
call_params["memory_id"] = self.memory_id
call_params["prompt"] = prompt
logger.bind(tag=TAG).debug(
f"【阿里百练API服务】处理后的prompt: {prompt}"
)
# 可选地设置自定义API基地址（若配置为兼容模式URL则忽略）
if self.base_url and ("/api/" in self.base_url):
dashscope.base_http_api_url = self.base_url
responses = Application.call(**call_params)
# 流式处理（SDK在stream=True时返回可迭代对象；否则返回单次响应对象）
logger.bind(tag=TAG).debug( logger.bind(tag=TAG).debug(
f"【阿里百练API服务】构造参数: {dict(call_params, api_key='***')}" f"【阿里百练API服务】处理后的dialogue: {dialogue}"
) )
last_text = "" # 构造调用参数
try: call_params = {
for resp in responses: "api_key": self.api_key,
if resp.status_code != HTTPStatus.OK: "app_id": self.app_id,
logger.bind(tag=TAG).error( "session_id": session_id,
f"code={resp.status_code}, message={resp.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code" "messages": dialogue,
) # 开启SDK原生流式
continue "stream": True,
current_text = getattr(getattr(resp, "output", None), "text", None) }
if current_text is None: if self.memory_id != False:
continue # 百练memory需要prompt参数
# SDK流式为增量覆盖，计算差量输出 prompt = dialogue[-1].get("content")
if len(current_text) >= len(last_text): call_params["memory_id"] = self.memory_id
delta = current_text[len(last_text):] call_params["prompt"] = prompt
else: logger.bind(tag=TAG).debug(
# 避免偶发回退 f"【阿里百练API服务】处理后的prompt: {prompt}"
delta = current_text )
if delta:
yield delta
last_text = current_text
except TypeError:
# 非流式回落（一次性返回）
if responses.status_code != HTTPStatus.OK:
logger.bind(tag=TAG).error(
f"code={responses.status_code}, message={responses.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
)
yield "【阿里百练API服务响应异常】"
else:
full_text = getattr(getattr(responses, "output", None), "text", "")
logger.bind(tag=TAG).info(
f"【阿里百练API服务】完整响应长度: {len(full_text)}"
)
for i in range(0, len(full_text), self.streaming_chunk_size):
chunk = full_text[i:i + self.streaming_chunk_size]
if chunk:
yield chunk
except Exception as e: # 可选地设置自定义API基地址（若配置为兼容模式URL则忽略）
logger.bind(tag=TAG).error(f"【阿里百练API服务】响应异常: {e}") if self.base_url and ("/api/" in self.base_url):
yield "【LLM服务响应异常】" dashscope.base_http_api_url = self.base_url
responses = Application.call(**call_params)
# 流式处理（SDK在stream=True时返回可迭代对象；否则返回单次响应对象）
logger.bind(tag=TAG).debug(
f"【阿里百练API服务】构造参数: {dict(call_params, api_key='***')}"
)
last_text = ""
try:
for resp in responses:
if resp.status_code != HTTPStatus.OK:
logger.bind(tag=TAG).error(
f"code={resp.status_code}, message={resp.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
)
continue
current_text = getattr(getattr(resp, "output", None), "text", None)
if current_text is None:
continue
# SDK流式为增量覆盖，计算差量输出
if len(current_text) >= len(last_text):
delta = current_text[len(last_text):]
else:
# 避免偶发回退
delta = current_text
if delta:
yield delta
last_text = current_text
except TypeError:
# 非流式回落（一次性返回）
if responses.status_code != HTTPStatus.OK:
logger.bind(tag=TAG).error(
f"code={responses.status_code}, message={responses.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code"
)
yield "【阿里百练API服务响应异常】"
else:
full_text = getattr(getattr(responses, "output", None), "text", "")
logger.bind(tag=TAG).info(
f"【阿里百练API服务】完整响应长度: {len(full_text)}"
)
for i in range(0, len(full_text), self.streaming_chunk_size):
chunk = full_text[i:i + self.streaming_chunk_size]
if chunk:
yield chunk
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
# 阿里百练当前未支持原生的 function call。为保持兼容，这里回退到普通文本流式输出。 # 阿里百练当前未支持原生的 function call。为保持兼容，这里回退到普通文本流式输出。
main/xiaozhi-server/core/providers/llm/base.py
+9 -14
View File
@@ -11,20 +11,15 @@ class LLMProviderBase(ABC):
pass pass
def response_no_stream(self, system_prompt, user_prompt, **kwargs): def response_no_stream(self, system_prompt, user_prompt, **kwargs):
try: # 构造对话格式
# 构造对话格式 dialogue = [
dialogue = [ {"role": "system", "content": system_prompt},
{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}
{"role": "user", "content": user_prompt} ]
] result = ""
result = "" for part in self.response("", dialogue, **kwargs):
for part in self.response("", dialogue, **kwargs): result += part
result += part return result
return result
except Exception as e:
logger.bind(tag=TAG).error(f"Error in Ollama response generation: {e}")
return "【LLM服务响应异常】"
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
""" """
main/xiaozhi-server/core/providers/llm/dify/dify.py
+61 -66
View File
@@ -20,76 +20,71 @@ class LLMProvider(LLMProviderBase):
logger.bind(tag=TAG).error(model_key_msg) logger.bind(tag=TAG).error(model_key_msg)
def response(self, session_id, dialogue, **kwargs): def response(self, session_id, dialogue, **kwargs):
try: # 取最后一条用户消息
# 取最后一条用户消息 last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
last_msg = next(m for m in reversed(dialogue) if m["role"] == "user") conversation_id = self.session_conversation_map.get(session_id)
conversation_id = self.session_conversation_map.get(session_id)
# 发起流式请求 # 发起流式请求
if self.mode == "chat-messages":
request_json = {
"query": last_msg["content"],
"response_mode": "streaming",
"user": session_id,
"inputs": {},
"conversation_id": conversation_id,
}
elif self.mode == "workflows/run":
request_json = {
"inputs": {"query": last_msg["content"]},
"response_mode": "streaming",
"user": session_id,
}
elif self.mode == "completion-messages":
request_json = {
"inputs": {"query": last_msg["content"]},
"response_mode": "streaming",
"user": session_id,
}
with requests.post(
f"{self.base_url}/{self.mode}",
headers={"Authorization": f"Bearer {self.api_key}"},
json=request_json,
stream=True,
) as r:
if self.mode == "chat-messages": if self.mode == "chat-messages":
request_json = { for line in r.iter_lines():
"query": last_msg["content"], if line.startswith(b"data: "):
"response_mode": "streaming", event = json.loads(line[6:])
"user": session_id, # 如果没有找到conversation_id，则获取此次conversation_id
"inputs": {}, if not conversation_id:
"conversation_id": conversation_id, conversation_id = event.get("conversation_id")
} self.session_conversation_map[session_id] = (
conversation_id # 更新映射
)
# 过滤 message_replace 事件，此事件会全量推一次
if event.get("event") != "message_replace" and event.get(
"answer"
):
yield event["answer"]
elif self.mode == "workflows/run": elif self.mode == "workflows/run":
request_json = { for line in r.iter_lines():
"inputs": {"query": last_msg["content"]}, if line.startswith(b"data: "):
"response_mode": "streaming", event = json.loads(line[6:])
"user": session_id, if event.get("event") == "workflow_finished":
} if event["data"]["status"] == "succeeded":
yield event["data"]["outputs"]["answer"]
else:
yield "【服务响应异常】"
elif self.mode == "completion-messages": elif self.mode == "completion-messages":
request_json = { for line in r.iter_lines():
"inputs": {"query": last_msg["content"]}, if line.startswith(b"data: "):
"response_mode": "streaming", event = json.loads(line[6:])
"user": session_id, # 过滤 message_replace 事件，此事件会全量推一次
} if event.get("event") != "message_replace" and event.get(
"answer"
with requests.post( ):
f"{self.base_url}/{self.mode}", yield event["answer"]
headers={"Authorization": f"Bearer {self.api_key}"},
json=request_json,
stream=True,
) as r:
if self.mode == "chat-messages":
for line in r.iter_lines():
if line.startswith(b"data: "):
event = json.loads(line[6:])
# 如果没有找到conversation_id，则获取此次conversation_id
if not conversation_id:
conversation_id = event.get("conversation_id")
self.session_conversation_map[session_id] = (
conversation_id # 更新映射
)
# 过滤 message_replace 事件，此事件会全量推一次
if event.get("event") != "message_replace" and event.get(
"answer"
):
yield event["answer"]
elif self.mode == "workflows/run":
for line in r.iter_lines():
if line.startswith(b"data: "):
event = json.loads(line[6:])
if event.get("event") == "workflow_finished":
if event["data"]["status"] == "succeeded":
yield event["data"]["outputs"]["answer"]
else:
yield "【服务响应异常】"
elif self.mode == "completion-messages":
for line in r.iter_lines():
if line.startswith(b"data: "):
event = json.loads(line[6:])
# 过滤 message_replace 事件，此事件会全量推一次
if event.get("event") != "message_replace" and event.get(
"answer"
):
yield event["answer"]
except Exception as e:
logger.bind(tag=TAG).error(f"Error in response generation: {e}")
yield "【服务响应异常】"
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
if len(dialogue) == 2 and functions is not None and len(functions) > 0: if len(dialogue) == 2 and functions is not None and len(functions) > 0:
main/xiaozhi-server/core/providers/llm/fastgpt/fastgpt.py
+39 -44
View File
@@ -19,53 +19,48 @@ class LLMProvider(LLMProviderBase):
logger.bind(tag=TAG).error(model_key_msg) logger.bind(tag=TAG).error(model_key_msg)
def response(self, session_id, dialogue, **kwargs): def response(self, session_id, dialogue, **kwargs):
try: # 取最后一条用户消息
# 取最后一条用户消息 last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
last_msg = next(m for m in reversed(dialogue) if m["role"] == "user")
# 发起流式请求 # 发起流式请求
with requests.post( with requests.post(
f"{self.base_url}/chat/completions", f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"}, headers={"Authorization": f"Bearer {self.api_key}"},
json={ json={
"stream": True, "stream": True,
"chatId": session_id, "chatId": session_id,
"detail": self.detail, "detail": self.detail,
"variables": self.variables, "variables": self.variables,
"messages": [{"role": "user", "content": last_msg["content"]}], "messages": [{"role": "user", "content": last_msg["content"]}],
}, },
stream=True, stream=True,
) as r: ) as r:
for line in r.iter_lines(): for line in r.iter_lines():
if line: if line:
try: try:
if line.startswith(b"data: "): if line.startswith(b"data: "):
if line[6:].decode("utf-8") == "[DONE]": if line[6:].decode("utf-8") == "[DONE]":
break break
data = json.loads(line[6:]) data = json.loads(line[6:])
if "choices" in data and len(data["choices"]) > 0: if "choices" in data and len(data["choices"]) > 0:
delta = data["choices"][0].get("delta", {}) delta = data["choices"][0].get("delta", {})
if ( if (
delta delta
and "content" in delta and "content" in delta
and delta["content"] is not None and delta["content"] is not None
): ):
content = delta["content"] content = delta["content"]
if "<think>" in content: if "<think>" in content:
continue continue
if "</think>" in content: if "</think>" in content:
continue continue
yield content yield content
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
continue continue
except Exception as e: except Exception as e:
continue continue
except Exception as e:
logger.bind(tag=TAG).error(f"Error in response generation: {e}")
yield "【服务响应异常】"
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
logger.bind(tag=TAG).error( logger.bind(tag=TAG).error(
main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py
+37 -43
View File
@@ -15,55 +15,49 @@ class LLMProvider(LLMProviderBase):
self.api_url = f"{self.base_url}/api/conversation/process" # 拼接完整的 API URL self.api_url = f"{self.base_url}/api/conversation/process" # 拼接完整的 API URL
def response(self, session_id, dialogue, **kwargs): def response(self, session_id, dialogue, **kwargs):
try: # home assistant语音助手自带意图，无需使用xiaozhi ai自带的，只需要把用户说的话传递给home assistant即可
# home assistant语音助手自带意图，无需使用xiaozhi ai自带的，只需要把用户说的话传递给home assistant即可
# 提取最后一个 role 为 'user' 的 content # 提取最后一个 role 为 'user' 的 content
input_text = None input_text = None
if isinstance(dialogue, list): # 确保 dialogue 是一个列表 if isinstance(dialogue, list): # 确保 dialogue 是一个列表
# 逆序遍历，找到最后一个 role 为 'user' 的消息 # 逆序遍历，找到最后一个 role 为 'user' 的消息
for message in reversed(dialogue): for message in reversed(dialogue):
if message.get("role") == "user": # 找到 role 为 'user' 的消息 if message.get("role") == "user": # 找到 role 为 'user' 的消息
input_text = message.get("content", "") input_text = message.get("content", "")
break # 找到后立即退出循环 break # 找到后立即退出循环
# 构造请求数据 # 构造请求数据
payload = { payload = {
"text": input_text, "text": input_text,
"agent_id": self.agent_id, "agent_id": self.agent_id,
"conversation_id": session_id, # 使用 session_id 作为 conversation_id "conversation_id": session_id, # 使用 session_id 作为 conversation_id
} }
# 设置请求头 # 设置请求头
headers = { headers = {
"Authorization": f"Bearer {self.api_key}", "Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json", "Content-Type": "application/json",
} }
# 发起 POST 请求 # 发起 POST 请求
response = requests.post(self.api_url, json=payload, headers=headers) response = requests.post(self.api_url, json=payload, headers=headers)
# 检查请求是否成功 # 检查请求是否成功
response.raise_for_status() response.raise_for_status()
# 解析返回数据 # 解析返回数据
data = response.json() data = response.json()
speech = ( speech = (
data.get("response", {}) data.get("response", {})
.get("speech", {}) .get("speech", {})
.get("plain", {}) .get("plain", {})
.get("speech", "") .get("speech", "")
) )
# 返回生成的内容 # 返回生成的内容
if speech: if speech:
yield speech yield speech
else: else:
logger.bind(tag=TAG).warning("API 返回数据中没有 speech 内容") logger.bind(tag=TAG).warning("API 返回数据中没有 speech 内容")
except RequestException as e:
logger.bind(tag=TAG).error(f"HTTP 请求错误: {e}")
except Exception as e:
logger.bind(tag=TAG).error(f"生成响应时出错: {e}")
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
logger.bind(tag=TAG).error( logger.bind(tag=TAG).error(
main/xiaozhi-server/core/providers/llm/ollama/ollama.py
+118 -128
View File
@@ -25,151 +25,141 @@ class LLMProvider(LLMProviderBase):
self.is_qwen3 = self.model_name and self.model_name.lower().startswith("qwen3") self.is_qwen3 = self.model_name and self.model_name.lower().startswith("qwen3")
def response(self, session_id, dialogue, **kwargs): def response(self, session_id, dialogue, **kwargs):
try: # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
# 如果是qwen3模型，在用户最后一条消息中添加/no_think指令 if self.is_qwen3:
if self.is_qwen3: # 复制对话列表，避免修改原始对话
# 复制对话列表，避免修改原始对话 dialogue_copy = dialogue.copy()
dialogue_copy = dialogue.copy()
# 找到最后一条用户消息 # 找到最后一条用户消息
for i in range(len(dialogue_copy) - 1, -1, -1): for i in range(len(dialogue_copy) - 1, -1, -1):
if dialogue_copy[i]["role"] == "user": if dialogue_copy[i]["role"] == "user":
# 在用户消息前添加/no_think指令 # 在用户消息前添加/no_think指令
dialogue_copy[i]["content"] = ( dialogue_copy[i]["content"] = (
"/no_think " + dialogue_copy[i]["content"] "/no_think " + dialogue_copy[i]["content"]
)
logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
break
# 使用修改后的对话
dialogue = dialogue_copy
responses = self.client.chat.completions.create(
model=self.model_name, messages=dialogue, stream=True
)
is_active = True
# 用于处理跨chunk的标签
buffer = ""
for chunk in responses:
try:
delta = (
chunk.choices[0].delta
if getattr(chunk, "choices", None)
else None
) )
content = delta.content if hasattr(delta, "content") else "" logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
break
if content: # 使用修改后的对话
# 将内容添加到缓冲区 dialogue = dialogue_copy
buffer += content
# 处理缓冲区中的标签 responses = self.client.chat.completions.create(
while "<think>" in buffer and "</think>" in buffer: model=self.model_name, messages=dialogue, stream=True
# 找到完整的<think></think>标签并移除 )
pre = buffer.split("<think>", 1)[0] is_active = True
post = buffer.split("</think>", 1)[1] # 用于处理跨chunk的标签
buffer = pre + post buffer = ""
# 处理只有开始标签的情况 for chunk in responses:
if "<think>" in buffer: try:
is_active = False delta = (
buffer = buffer.split("<think>", 1)[0] chunk.choices[0].delta
if getattr(chunk, "choices", None)
else None
)
content = delta.content if hasattr(delta, "content") else ""
# 处理只有结束标签的情况 if content:
if "</think>" in buffer: # 将内容添加到缓冲区
is_active = True buffer += content
buffer = buffer.split("</think>", 1)[1]
# 如果当前处于活动状态且缓冲区有内容，则输出 # 处理缓冲区中的标签
if is_active and buffer: while "<think>" in buffer and "</think>" in buffer:
yield buffer # 找到完整的<think></think>标签并移除
buffer = "" # 清空缓冲区 pre = buffer.split("<think>", 1)[0]
post = buffer.split("</think>", 1)[1]
buffer = pre + post
except Exception as e: # 处理只有开始标签的情况
logger.bind(tag=TAG).error(f"Error processing chunk: {e}") if "<think>" in buffer:
is_active = False
buffer = buffer.split("<think>", 1)[0]
except Exception as e: # 处理只有结束标签的情况
logger.bind(tag=TAG).error(f"Error in Ollama response generation: {e}") if "</think>" in buffer:
yield "【Ollama服务响应异常】" is_active = True
buffer = buffer.split("</think>", 1)[1]
# 如果当前处于活动状态且缓冲区有内容，则输出
if is_active and buffer:
yield buffer
buffer = "" # 清空缓冲区
except Exception as e:
logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
try: # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
# 如果是qwen3模型，在用户最后一条消息中添加/no_think指令 if self.is_qwen3:
if self.is_qwen3: # 复制对话列表，避免修改原始对话
# 复制对话列表，避免修改原始对话 dialogue_copy = dialogue.copy()
dialogue_copy = dialogue.copy()
# 找到最后一条用户消息 # 找到最后一条用户消息
for i in range(len(dialogue_copy) - 1, -1, -1): for i in range(len(dialogue_copy) - 1, -1, -1):
if dialogue_copy[i]["role"] == "user": if dialogue_copy[i]["role"] == "user":
# 在用户消息前添加/no_think指令 # 在用户消息前添加/no_think指令
dialogue_copy[i]["content"] = ( dialogue_copy[i]["content"] = (
"/no_think " + dialogue_copy[i]["content"] "/no_think " + dialogue_copy[i]["content"]
)
logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
break
# 使用修改后的对话
dialogue = dialogue_copy
stream = self.client.chat.completions.create(
model=self.model_name,
messages=dialogue,
stream=True,
tools=functions,
)
is_active = True
buffer = ""
for chunk in stream:
try:
delta = (
chunk.choices[0].delta
if getattr(chunk, "choices", None)
else None
)
content = delta.content if hasattr(delta, "content") else None
tool_calls = (
delta.tool_calls if hasattr(delta, "tool_calls") else None
) )
logger.bind(tag=TAG).debug(f"为qwen3模型添加/no_think指令")
break
# 如果是工具调用，直接传递 # 使用修改后的对话
if tool_calls: dialogue = dialogue_copy
yield None, tool_calls
continue
# 处理文本内容 stream = self.client.chat.completions.create(
if content: model=self.model_name,
# 将内容添加到缓冲区 messages=dialogue,
buffer += content stream=True,
tools=functions,
)
# 处理缓冲区中的标签 is_active = True
while "<think>" in buffer and "</think>" in buffer: buffer = ""
# 找到完整的<think></think>标签并移除
pre = buffer.split("<think>", 1)[0]
post = buffer.split("</think>", 1)[1]
buffer = pre + post
# 处理只有开始标签的情况 for chunk in stream:
if "<think>" in buffer: try:
is_active = False delta = (
buffer = buffer.split("<think>", 1)[0] chunk.choices[0].delta
if getattr(chunk, "choices", None)
else None
)
content = delta.content if hasattr(delta, "content") else None
tool_calls = (
delta.tool_calls if hasattr(delta, "tool_calls") else None
)
# 处理只有结束标签的情况 # 如果是工具调用，直接传递
if "</think>" in buffer: if tool_calls:
is_active = True yield None, tool_calls
buffer = buffer.split("</think>", 1)[1]
# 如果当前处于活动状态且缓冲区有内容，则输出
if is_active and buffer:
yield buffer, None
buffer = "" # 清空缓冲区
except Exception as e:
logger.bind(tag=TAG).error(f"Error processing function chunk: {e}")
continue continue
except Exception as e: # 处理文本内容
logger.bind(tag=TAG).error(f"Error in Ollama function call: {e}") if content:
yield f"【Ollama服务响应异常: {str(e)}】", None # 将内容添加到缓冲区
buffer += content
# 处理缓冲区中的标签
while "<think>" in buffer and "</think>" in buffer:
# 找到完整的<think></think>标签并移除
pre = buffer.split("<think>", 1)[0]
post = buffer.split("</think>", 1)[1]
buffer = pre + post
# 处理只有开始标签的情况
if "<think>" in buffer:
is_active = False
buffer = buffer.split("<think>", 1)[0]
# 处理只有结束标签的情况
if "</think>" in buffer:
is_active = True
buffer = buffer.split("</think>", 1)[1]
# 如果当前处于活动状态且缓冲区有内容，则输出
if is_active and buffer:
yield buffer, None
buffer = "" # 清空缓冲区
except Exception as e:
logger.bind(tag=TAG).error(f"Error processing function chunk: {e}")
continue
main/xiaozhi-server/core/providers/llm/openai/openai.py
+64 -72
View File
@@ -56,87 +56,79 @@ class LLMProvider(LLMProviderBase):
return dialogue return dialogue
def response(self, session_id, dialogue, **kwargs): def response(self, session_id, dialogue, **kwargs):
try: dialogue = self.normalize_dialogue(dialogue)
dialogue = self.normalize_dialogue(dialogue)
request_params = { request_params = {
"model": self.model_name, "model": self.model_name,
"messages": dialogue, "messages": dialogue,
"stream": True, "stream": True,
} }
# 添加可选参数,只有当参数不为None时才添加 # 添加可选参数,只有当参数不为None时才添加
optional_params = { optional_params = {
"max_tokens": kwargs.get("max_tokens", self.max_tokens), "max_tokens": kwargs.get("max_tokens", self.max_tokens),
"temperature": kwargs.get("temperature", self.temperature), "temperature": kwargs.get("temperature", self.temperature),
"top_p": kwargs.get("top_p", self.top_p), "top_p": kwargs.get("top_p", self.top_p),
"frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty), "frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty),
} }
for key, value in optional_params.items(): for key, value in optional_params.items():
if value is not None: if value is not None:
request_params[key] = value request_params[key] = value
responses = self.client.chat.completions.create(**request_params) # raise ValueError("model_name is required")
responses = self.client.chat.completions.create(**request_params)
is_active = True is_active = True
for chunk in responses: for chunk in responses:
try: try:
delta = chunk.choices[0].delta if getattr(chunk, "choices", None) else None delta = chunk.choices[0].delta if getattr(chunk, "choices", None) else None
content = getattr(delta, "content", "") if delta else "" content = getattr(delta, "content", "") if delta else ""
except IndexError: except IndexError:
content = "" content = ""
if content: if content:
if "<think>" in content: if "<think>" in content:
is_active = False is_active = False
content = content.split("<think>")[0] content = content.split("<think>")[0]
if "</think>" in content: if "</think>" in content:
is_active = True is_active = True
content = content.split("</think>")[-1] content = content.split("</think>")[-1]
if is_active: if is_active:
yield content yield content
except Exception as e:
logger.bind(tag=TAG).error(f"Error in response generation: {e}")
def response_with_functions(self, session_id, dialogue, functions=None, **kwargs): def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
try: dialogue = self.normalize_dialogue(dialogue)
dialogue = self.normalize_dialogue(dialogue)
request_params = { request_params = {
"model": self.model_name, "model": self.model_name,
"messages": dialogue, "messages": dialogue,
"stream": True, "stream": True,
"tools": functions, "tools": functions,
} }
optional_params = { optional_params = {
"max_tokens": kwargs.get("max_tokens", self.max_tokens), "max_tokens": kwargs.get("max_tokens", self.max_tokens),
"temperature": kwargs.get("temperature", self.temperature), "temperature": kwargs.get("temperature", self.temperature),
"top_p": kwargs.get("top_p", self.top_p), "top_p": kwargs.get("top_p", self.top_p),
"frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty), "frequency_penalty": kwargs.get("frequency_penalty", self.frequency_penalty),
} }
for key, value in optional_params.items(): for key, value in optional_params.items():
if value is not None: if value is not None:
request_params[key] = value request_params[key] = value
stream = self.client.chat.completions.create(**request_params) stream = self.client.chat.completions.create(**request_params)
for chunk in stream: for chunk in stream:
if getattr(chunk, "choices", None): if getattr(chunk, "choices", None):
delta = chunk.choices[0].delta delta = chunk.choices[0].delta
content = getattr(delta, "content", "") content = getattr(delta, "content", "")
tool_calls = getattr(delta, "tool_calls", None) tool_calls = getattr(delta, "tool_calls", None)
yield content, tool_calls yield content, tool_calls
elif isinstance(getattr(chunk, "usage", None), CompletionUsage): elif isinstance(getattr(chunk, "usage", None), CompletionUsage):
usage_info = getattr(chunk, "usage", None) usage_info = getattr(chunk, "usage", None)
logger.bind(tag=TAG).info( logger.bind(tag=TAG).info(
f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，" f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，"
f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，" f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，"
f"共计 {getattr(usage_info, 'total_tokens', '未知')}" f"共计 {getattr(usage_info, 'total_tokens', '未知')}"
) )
except Exception as e:
logger.bind(tag=TAG).error(f"Error in function call streaming: {e}")
yield f"【OpenAI服务响应异常: {e}】", None
main/xiaozhi-server/core/providers/llm/xinference/xinference.py
+45 -58
View File
@@ -31,68 +31,55 @@ class LLMProvider(LLMProviderBase):
raise raise
def response(self, session_id, dialogue, **kwargs): def response(self, session_id, dialogue, **kwargs):
try: logger.bind(tag=TAG).debug(
logger.bind(tag=TAG).debug( f"Sending request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}"
f"Sending request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}" )
) responses = self.client.chat.completions.create(
responses = self.client.chat.completions.create( model=self.model_name, messages=dialogue, stream=True
model=self.model_name, messages=dialogue, stream=True )
) is_active = True
is_active = True for chunk in responses:
for chunk in responses: try:
try: delta = (
delta = ( chunk.choices[0].delta
chunk.choices[0].delta if getattr(chunk, "choices", None)
if getattr(chunk, "choices", None) else None
else None )
) content = delta.content if hasattr(delta, "content") else ""
content = delta.content if hasattr(delta, "content") else "" if content:
if content: if "<think>" in content:
if "<think>" in content: is_active = False
is_active = False content = content.split("<think>")[0]
content = content.split("<think>")[0] if "</think>" in content:
if "</think>" in content: is_active = True
is_active = True content = content.split("</think>")[-1]
content = content.split("</think>")[-1] if is_active:
if is_active: yield content
yield content except Exception as e:
except Exception as e: logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
except Exception as e:
logger.bind(tag=TAG).error(f"Error in Xinference response generation: {e}")
yield "【Xinference服务响应异常】"
def response_with_functions(self, session_id, dialogue, functions=None): def response_with_functions(self, session_id, dialogue, functions=None):
try: logger.bind(tag=TAG).debug(
f"Sending function call request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}"
)
if functions:
logger.bind(tag=TAG).debug( logger.bind(tag=TAG).debug(
f"Sending function call request to Xinference with model: {self.model_name}, dialogue length: {len(dialogue)}" f"Function calls enabled with: {[f.get('function', {}).get('name') for f in functions]}"
)
if functions:
logger.bind(tag=TAG).debug(
f"Function calls enabled with: {[f.get('function', {}).get('name') for f in functions]}"
)
stream = self.client.chat.completions.create(
model=self.model_name,
messages=dialogue,
stream=True,
tools=functions,
) )
for chunk in stream: stream = self.client.chat.completions.create(
delta = chunk.choices[0].delta model=self.model_name,
content = delta.content messages=dialogue,
tool_calls = delta.tool_calls stream=True,
tools=functions,
)
if content: for chunk in stream:
yield content, tool_calls delta = chunk.choices[0].delta
elif tool_calls: content = delta.content
yield None, tool_calls tool_calls = delta.tool_calls
except Exception as e: if content:
logger.bind(tag=TAG).error(f"Error in Xinference function call: {e}") yield content, tool_calls
yield { elif tool_calls:
"type": "content", yield None, tool_calls
"content": f"【Xinference服务响应异常: {str(e)}】",
}
main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py
+8 -8
View File
@@ -162,19 +162,19 @@ class MemoryProvider(MemoryProviderBase):
msgStr += f"当前时间：{time_str}" msgStr += f"当前时间：{time_str}"
if self.save_to_file: if self.save_to_file:
result = self.llm.response_no_stream(
short_term_memory_prompt,
msgStr,
max_tokens=2000,
temperature=0.2,
)
json_str = extract_json_data(result)
try: try:
result = self.llm.response_no_stream(
short_term_memory_prompt,
msgStr,
max_tokens=2000,
temperature=0.2,
)
json_str = extract_json_data(result)
json.loads(json_str) # 检查json格式是否正确 json.loads(json_str) # 检查json格式是否正确
self.short_memory = json_str self.short_memory = json_str
self.save_memory_to_file() self.save_memory_to_file()
except Exception as e: except Exception as e:
print("Error:", e) logger.bind(tag=TAG).error(f"Error in saving memory: {e}")
else: else:
# 当save_to_file为False时，调用Java端的聊天记录总结接口 # 当save_to_file为False时，调用Java端的聊天记录总结接口
summary_id = session_id if session_id else self.role_id summary_id = session_id if session_id else self.role_id
main/xiaozhi-server/core/utils/util.py
+11
View File
@@ -573,3 +573,14 @@ def validate_mcp_endpoint(mcp_endpoint: str) -> bool:
return False return False
return True return True
def get_system_error_response(config: dict) -> str:
"""获取系统错误时的回复
Args:
config: 配置字典
Returns:
str: 系统错误时的回复
"""
return config.get("system_error_response", "主人，小智现在有点忙，我们稍后再试吧。")