Merge branch 'xinnan-tech:main' into DifyBug

This commit is contained in:
Jiao Haoyang
2025-03-18 17:09:51 +08:00
committed by GitHub
20 changed files with 490 additions and 157 deletions
+33 -23
View File
@@ -197,7 +197,7 @@ class ConnectionHandler:
self.dialogue.put(Message(role="system", content=self.prompt))
self.func_handler = FunctionHandler(self.config)
def change_system_prompt(self, prompt):
self.prompt = prompt
# 找到原来的role==system,替换原来的系统提示
@@ -303,7 +303,7 @@ class ConnectionHandler:
self.logger.bind(tag=TAG).debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False))
return True
def chat_with_function_calling(self, query, tool_call = False):
def chat_with_function_calling(self, query, tool_call=False):
self.logger.bind(tag=TAG).debug(f"Chat with function calling start: {query}")
"""Chat with function calling for intent detection using streaming"""
if self.isNeedAuth():
@@ -311,7 +311,7 @@ class ConnectionHandler:
future = asyncio.run_coroutine_threadsafe(self._check_and_broadcast_auth_code(), self.loop)
future.result()
return True
if not tool_call:
self.dialogue.put(Message(role="user", content=query))
@@ -320,7 +320,7 @@ class ConnectionHandler:
response_message = []
processed_chars = 0 # 跟踪已处理的字符位置
try:
start_time = time.time()
@@ -328,7 +328,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"对话记录: {self.dialogue.get_llm_dialogue_with_memory(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(
@@ -351,8 +351,8 @@ class ConnectionHandler:
content_arguments = ""
for response in llm_responses:
content, tools_call = response
if content is not None and len(content)>0:
if len(response_message)<=0 and (content=="```" or "<tool_call>" in content):
if content is not None and len(content) > 0:
if len(response_message) <= 0 and (content == "```" or "<tool_call>" in content):
tool_call_flag = True
if tools_call is not None:
@@ -366,7 +366,7 @@ class ConnectionHandler:
if content is not None and len(content) > 0:
if tool_call_flag:
content_arguments+=content
content_arguments += content
else:
response_message.append(content)
@@ -422,16 +422,17 @@ class ConnectionHandler:
else:
function_arguments = json.loads(function_arguments)
if not bHasError:
self.logger.bind(tag=TAG).info(f"function_name={function_name}, function_id={function_id}, function_arguments={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 = self.func_handler.handle_llm_function_call(self, function_call_data)
self._handle_function_result(result, function_call_data, text_index+1)
self._handle_function_result(result, function_call_data, text_index + 1)
# 处理最后剩余的文本
# 处理最后剩余的文本
full_text = "".join(response_message)
remaining_text = full_text[processed_chars:]
if remaining_text:
@@ -443,7 +444,7 @@ class ConnectionHandler:
self.tts_queue.put(future)
# 存储对话内容
if len(response_message)>0:
if len(response_message) > 0:
self.dialogue.put(Message(role="assistant", content="".join(response_message)))
self.llm_finish_task = True
@@ -452,31 +453,40 @@ class ConnectionHandler:
return True
def _handle_function_result(self, result, function_call_data, text_index):
if result.action == Action.RESPONSE: # 直接回复前端
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生成回复
elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
text = result.result
if text is not None and len(text) > 0:
function_id = function_call_data["id"]
function_name = function_call_data["name"]
function_arguments = function_call_data["arguments"]
self.dialogue.put(Message(role='assistant',
tool_calls=[{"id": function_id,
"function": {"arguments": function_arguments,"name": function_name},
"type": 'function',
"index": 0}]))
tool_calls=[{"id": function_id,
"function": {"arguments": function_arguments,
"name": function_name},
"type": 'function',
"index": 0}]))
self.dialogue.put(Message(role="tool", tool_call_id=function_id, content=text))
self.chat_with_function_calling(text, tool_call=True)
if result.action == Action.NOTFOUND:
text = result.response
elif result.action == Action.NOTFOUND:
text = result.result
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))
else:
text = result.result
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))
def _tts_priority_thread(self):
while not self.stop_event.is_set():
@@ -50,6 +50,7 @@ class FunctionHandler:
self.function_registry.register_function("play_music")
self.function_registry.register_function("plugin_loader")
self.function_registry.register_function("get_time")
self.function_registry.register_function("raise_and_lower_the_volume")
def register_config_functions(self):
"""注册配置中的函数,可以不同客户端使用不同的配置"""
@@ -69,7 +70,7 @@ class FunctionHandler:
arguments = function_call_data["arguments"]
arguments = json.loads(arguments) if arguments else {}
logger.bind(tag=TAG).info(f"调用函数: {function_name}, 参数: {arguments}")
if funcItem.type == ToolType.SYSTEM_CTL:
if funcItem.type == ToolType.SYSTEM_CTL or funcItem.type == ToolType.IOT_CTL:
return func(conn, **arguments)
elif funcItem.type == ToolType.WAIT:
return func(**arguments)
+273 -97
View File
@@ -1,24 +1,151 @@
import json
import asyncio
from config.logger import setup_logging
from plugins_func.register import device_type_registry, register_function, ActionResponse, Action, ToolType
TAG = __name__
logger = setup_logging()
def wrap_async_function(async_func):
"""包装异步函数为同步函数"""
def wrapper(*args, **kwargs):
try:
# 获取连接对象(第一个参数)
conn = args[0]
if not hasattr(conn, 'loop'):
logger.bind(tag=TAG).error("Connection对象没有loop属性")
return ActionResponse(Action.ERROR, "Connection对象没有loop属性",
"执行操作时出错: Connection对象没有loop属性")
# 使用conn对象中的事件循环
loop = conn.loop
# 在conn的事件循环中运行异步函数
future = asyncio.run_coroutine_threadsafe(async_func(*args, **kwargs), loop)
# 等待结果返回
return future.result()
except Exception as e:
logger.bind(tag=TAG).error(f"运行异步函数时出错: {e}")
return ActionResponse(Action.ERROR, str(e), f"执行操作时出错: {e}")
return wrapper
def create_iot_function(device_name, method_name, method_info):
"""
根据IOT设备描述生成通用的控制函数
"""
async def iot_control_function(conn, response_success=None, response_failure=None, **params):
try:
# 打印响应参数
logger.bind(tag=TAG).info(
f"控制函数接收到的响应参数: success='{response_success}', failure='{response_failure}'")
# 发送控制命令
await send_iot_conn(conn, device_name, method_name, params)
# 等待一小段时间让状态更新
await asyncio.sleep(0.1)
# 生成结果信息
result = f"{device_name}{method_name}操作执行成功"
# 根据方法名尝试自动更新状态
await update_state_by_method(conn, device_name, method_name, params)
# 处理响应中可能的占位符
response = response_success
# 替换{value}占位符
for param_name, param_value in params.items():
# 先尝试直接替换参数值
if "{" + param_name + "}" in response:
response = response.replace("{" + param_name + "}", str(param_value))
# 如果有{value}占位符,用相关参数替换
if "{value}" in response:
response = response.replace("{value}", str(param_value))
break
return ActionResponse(Action.RESPONSE, result, response)
except Exception as e:
logger.bind(tag=TAG).error(f"执行{device_name}{method_name}操作失败: {e}")
# 操作失败时使用大模型提供的失败响应
response = response_failure
return ActionResponse(Action.ERROR, str(e), response)
return wrap_async_function(iot_control_function)
def create_iot_query_function(device_name, prop_name, prop_info):
"""
根据IOT设备属性创建查询函数
"""
async def iot_query_function(conn, response_success=None, response_failure=None):
try:
# 打印响应参数
logger.bind(tag=TAG).info(
f"查询函数接收到的响应参数: success='{response_success}', failure='{response_failure}'")
value = await get_iot_status(conn, device_name, prop_name)
# 查询成功,生成结果
if value is not None:
# 使用大模型提供的成功响应,并替换其中的占位符
response = response_success.replace("{value}", str(value))
return ActionResponse(Action.RESPONSE, str(value), response)
else:
# 查询失败,使用大模型提供的失败响应
response = response_failure
return ActionResponse(Action.ERROR, f"属性{prop_name}不存在", response)
except Exception as e:
logger.bind(tag=TAG).error(f"查询{device_name}{prop_name}时出错: {e}")
# 查询出错时使用大模型提供的失败响应
response = response_failure
return ActionResponse(Action.ERROR, str(e), response)
return wrap_async_function(iot_query_function)
async def update_state_by_method(conn, device_name, method_name, params):
"""根据方法和参数自动更新设备状态"""
try:
# 规则1: 方法名为TurnOn,设置power为True
if method_name == "TurnOn":
await set_iot_status(conn, device_name, "power", True)
# 规则2: 方法名为TurnOff,设置power为False
elif method_name == "TurnOff":
await set_iot_status(conn, device_name, "power", False)
# 规则3: Set开头的方法,尝试更新对应参数
elif method_name.startswith("Set"):
# 从参数中找到可能的状态值
for param_name, param_value in params.items():
# 尝试更新对应名称的属性
await set_iot_status(conn, device_name, param_name, param_value)
# 其他方法,尝试直接从参数更新状态
else:
for param_name, param_value in params.items():
# 检查设备是否有此属性
status = await get_iot_status(conn, device_name, param_name)
if status is not None: # 属性存在
await set_iot_status(conn, device_name, param_name, param_value)
except Exception as e:
logger.bind(tag=TAG).warning(f"自动更新状态失败: {e}")
class IotDescriptor:
"""
A class to represent an IoT descriptor.
Attributes:
----------
name : str
The name of the IoT descriptor.
description : str
A brief description of the IoT descriptor.
properties : dict
A dictionary containing properties of the IoT descriptor.
methods : dict
A dictionary containing methods of the IoT descriptor.
-------
"""
def __init__(self, name, description, properties, methods):
@@ -29,17 +156,7 @@ class IotDescriptor:
# 根据描述创建属性
for key, value in properties.items():
# "volume":{"description":"当前音量 值","type":"number"}
"""
等价于
{
'name': 名字,
'description': 描述,
'value': 0
}
"""
# setattr(self, key, {}) # 创建一个空字典, 名字是属性名
property_item = globals()[key] = {} # 创建一个空字典, 名字是属性名
property_item = globals()[key] = {}
property_item['name'] = key
property_item["description"] = value["description"]
if value["type"] == "number":
@@ -52,23 +169,10 @@ class IotDescriptor:
# 根据描述创建方法
for key, value in methods.items():
# "SetVolume": {"description":"设置音量","parameters":{"volume":{"description":"0到100之间的整数","type":"number"}}}
"""
等价于
SetVolume = {
`description`: 描述,
`volume`: {
`description`: 描述,
`value`: 0
}
}
"""
# setattr(self, key, {}) # 创建一个空字典, 名字是方法名
method = globals()[key] = {} # 创建一个空字典, 名字是方法名
method = globals()[key] = {}
method["description"] = value["description"]
method['name'] = key
for k, v in value["parameters"].items():
# 不同的参数解析
method[k] = {}
method[k]["description"] = v["description"]
if v["type"] == "number":
@@ -77,58 +181,136 @@ class IotDescriptor:
method[k]["value"] = False
else:
method[k]["value"] = ""
self.methods.append(method)
async def handleIotDescriptors(conn, descriptors):
"""
处理物联网描述
示例: [{
"name":"Speaker",
"description":"当前 AI 机器人的扬声器",
"properties":{
"volume":{"description":"当前音量 值","type":"number"} 可以有boolean, number, string三种类型
},
"methods":{
"SetVolume":{
"description":"设置音量","parameters":{"volume":{"description":"0到100之间的整数","type":"number"}}
def register_device_type(descriptor):
"""注册设备类型及其功能"""
device_name = descriptor["name"]
type_id = device_type_registry.generate_device_type_id(descriptor)
# 如果该类型已注册,直接返回类型ID
if type_id in device_type_registry.type_functions:
return type_id
functions = {}
# 为每个属性创建查询函数
for prop_name, prop_info in descriptor["properties"].items():
func_name = f"get_{device_name.lower()}_{prop_name.lower()}"
func_desc = {
"type": "function",
"function": {
"name": func_name,
"description": f"查询{descriptor['description']}{prop_info['description']}",
"parameters": {
"type": "object",
"properties": {
"response_success": {
"type": "string",
"description": f"查询成功时的友好回复,必须使用{{value}}作为占位符表示查询到的值"
},
"response_failure": {
"type": "string",
"description": f"查询失败时的友好回复,例如:'无法获取{device_name}{prop_info['description']}'"
}
},
"required": ["response_success", "response_failure"]
}
}
}
}]
descriptors: 描述列表
"""
query_func = create_iot_query_function(device_name, prop_name, prop_info)
decorated_func = register_function(func_name, func_desc, ToolType.IOT_CTL)(query_func)
functions[func_name] = decorated_func
# 为每个方法创建控制函数
for method_name, method_info in descriptor["methods"].items():
func_name = f"{device_name.lower()}_{method_name.lower()}"
# 创建参数字典,添加原有参数
parameters = {
param_name: {
"type": param_info["type"],
"description": param_info["description"]
}
for param_name, param_info in method_info["parameters"].items()
}
# 添加响应参数
parameters.update({
"response_success": {
"type": "string",
"description": "操作成功时的友好回复,关于该设备的操作结果,设备名称尽量使用description中的名称"
},
"response_failure": {
"type": "string",
"description": "操作失败时的友好回复,关于该设备的操作结果,设备名称尽量使用description中的名称"
}
})
# 构建必须参数列表(原有参数 + 响应参数)
required_params = list(method_info["parameters"].keys())
required_params.extend(["response_success", "response_failure"])
func_desc = {
"type": "function",
"function": {
"name": func_name,
"description": f"{descriptor['description']} - {method_info['description']}",
"parameters": {
"type": "object",
"properties": parameters,
"required": required_params
}
}
}
control_func = create_iot_function(device_name, method_name, method_info)
decorated_func = register_function(func_name, func_desc, ToolType.IOT_CTL)(control_func)
functions[func_name] = decorated_func
device_type_registry.register_device_type(type_id, functions)
return type_id
# 用于接受前端设备推送的搜索iot描述
async def handleIotDescriptors(conn, descriptors):
"""处理物联网描述"""
functions_changed = False
for descriptor in descriptors:
# 创建IOT设备描述符
iot_descriptor = IotDescriptor(descriptor["name"], descriptor["description"], descriptor["properties"],
descriptor["methods"])
conn.iot_descriptors[descriptor["name"]] = iot_descriptor
# 暂时从配置文件中设置音量,后期通过意图识别控制音量
default_iot_volume = 100
if "iot" in conn.config:
default_iot_volume = conn.config["iot"]["Speaker"]["volume"]
logger.bind(tag=TAG).info(f"服务端设置音量为{default_iot_volume}")
await send_iot_conn(conn, "Speaker", "SetVolume", {"volume": default_iot_volume})
if conn.use_function_call_mode:
# 注册或获取设备类型
type_id = register_device_type(descriptor)
device_functions = device_type_registry.get_device_functions(type_id)
# 在连接级注册设备函数
if hasattr(conn, 'func_handler'):
for func_name in device_functions:
conn.func_handler.function_registry.register_function(func_name)
logger.bind(tag=TAG).info(f"注册IOT函数到function handler: {func_name}")
functions_changed = True
# 如果注册了新函数,更新function描述列表
if functions_changed and hasattr(conn, 'func_handler'):
conn.func_handler.upload_functions_desc()
func_names = conn.func_handler.current_support_functions()
logger.bind(tag=TAG).info(f"设备类型: {type_id}")
logger.bind(tag=TAG).info(f"更新function描述列表完成,当前支持的函数: {func_names}")
async def handleIotStatus(conn, states):
"""
处理物联网状态
示例: [{
"name":"Speaker",
"state":{
"volume":100
}
}]
states: 状态列表
"""
"""处理物联网状态"""
for state in states:
for key, value in conn.iot_descriptors.items():
if key == state["name"]:
for property_item in value.properties:
# properties为字典列表, 记录各种属性
for k, v in state["state"].items():
# state为字典, 记录各种属性的值, 是需要记录的信息
if property_item["name"] == k:
# 检查一下属性是不是相同的
if type(v) != type(property_item["value"]):
logger.bind(tag=TAG).error(f"属性{property_item['name']}的值类型不匹配")
break
@@ -138,41 +320,35 @@ async def handleIotStatus(conn, states):
break
break
async def get_iot_status(conn, name, property_name):
"""
获取物联网状态
name: 设备名称 "Speaker"
property_name: 属性名称 "volume"
返回值: 属性值, 实际的属性有int, bool和str三种类型
"""
"""获取物联网状态"""
for key, value in conn.iot_descriptors.items():
if key == name:
for property_item in value.properties:
if property_item["name"] == property_name:
return property_item["value"]
logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
return None
async def send_iot_conn(conn, name, method_name, parameters):
"""
发送物联网指令
name: 设备名称 "Speaker"
method: 方法 "SetVolume"
parameters: 参数, 是一个字典 {"volume": 100}
发送示例:
{
"type": "iot",
"commands": [
{
"name" : "Speaker",
"method": "SetVolume",
"parameters": {
"volume": 100
}
}
]
}
"""
async def set_iot_status(conn, name, property_name, value):
"""设置物联网状态"""
for key, iot_descriptor in conn.iot_descriptors.items():
if key == name:
for property_item in iot_descriptor.properties:
if property_item["name"] == property_name:
if type(value) != type(property_item["value"]):
logger.bind(tag=TAG).error(f"属性{property_item['name']}的值类型不匹配")
return
property_item["value"] = value
logger.bind(tag=TAG).info(f"物联网状态更新: {name} , {property_name} = {value}")
return
logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
async def send_iot_conn(conn, name, method_name, parameters):
"""发送物联网指令"""
for key, value in conn.iot_descriptors.items():
if key == name:
# 找到了设备
@@ -2,7 +2,7 @@ from config.logger import setup_logging
import json
from core.handle.abortHandle import handleAbortMessage
from core.handle.helloHandle import handleHelloMessage
from core.handle.receiveAudioHandle import startToChat
from core.handle.receiveAudioHandle import startToChat, handleAudioMessage
from core.handle.iotHandle import handleIotDescriptors, handleIotStatus
TAG = __name__
@@ -24,13 +24,15 @@ async def handleTextMessage(conn, message):
elif msg_json["type"] == "listen":
if "mode" in msg_json:
conn.client_listen_mode = msg_json["mode"]
logger.bind(tag=TAG).debug(f"客户端拾音模式:{conn.client_listen_mode}")
logger.bind(tag=TAG).debug(f"客户端拾音模式:{conn. client_listen_mode}")
if msg_json["state"] == "start":
conn.client_have_voice = True
conn.client_voice_stop = False
elif msg_json["state"] == "stop":
conn.client_have_voice = True
conn.client_voice_stop = True
if len(conn.asr_audio) > 0:
await handleAudioMessage(conn, b'')
elif msg_json["state"] == "detect":
conn.asr_server_receive = False
conn.client_have_voice = False
@@ -13,7 +13,7 @@ logger = setup_logging()
class TTSProviderBase(ABC):
def __init__(self, config, delete_audio_file):
self.delete_audio_file = delete_audio_file
self.output_file = config.get("output_file")
self.output_file = config.get("output_dir")
@abstractmethod
def generate_filename(self):
@@ -15,7 +15,7 @@ class TTSProvider(TTSProviderBase):
self.headers = config.get("headers", {})
self.params = config.get("params")
self.format = config.get("format", "wav")
self.output_file = config.get("output_file", "tmp/")
self.output_file = config.get("output_dir", "tmp/")
def generate_filename(self):
return os.path.join(self.output_file, f"tts-{datetime.now().date()}@{uuid.uuid4().hex}.{self.format}")
@@ -22,7 +22,7 @@ class TTSProvider(TTSProviderBase):
self.cut_punc = config.get("cut_punc","")
self.speed = config.get("speed", 1.0)
self.inp_refs = config.get("inp_refs",[])
self.sample_steps = config.get("inp_refs",32)
self.sample_steps = config.get("sample_steps",32)
self.if_sr = config.get("if_sr",False)
@@ -14,7 +14,7 @@ class TTSProvider(TTSProviderBase):
self.voice = config.get("voice", "alloy")
self.response_format = "wav"
self.speed = config.get("speed", 1.0)
self.output_file = config.get("output_file", "tmp/")
self.output_file = config.get("output_dir", "tmp/")
check_model_key("TTS", self.api_key)
def generate_filename(self, extension=".wav"):
@@ -17,7 +17,7 @@ class TTSProvider(TTSProviderBase):
self.volume_change_dB = config.get("volume_change_dB", 0)
self.speed_factor = config.get("speed_factor", 1)
self.stream = config.get("stream", False)
self.output_file = config.get("output_file")
self.output_file = config.get("output_dir")
self.pitch_factor = config.get("pitch_factor", 0)
self.format = config.get("format", "mp3")
self.emotion = config.get("emotion", 1)