mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-22 15:13:55 +08:00
2e182626b9
* function call功能完善，增加天气查询，支持插件式扩展 * 增加角色切换功能，通过切换system提示词，修改角色认知 * 增加插件管理系统，可以通过语音加载和卸载插件 * docs: 添加命令操作 (#329) * docs: 添加命令操作 * feat: 添加 docker-setup.sh 脚本以简化服务端部署 - 新增 docker-setup.sh 脚本，自动创建目录结构、下载语音识别模型和配置文件，并检查文件完整性。 - 更新 Deployment.md 文档，提供一键执行脚本的说明和使用示例。 * docs: 更新 Deployment.md，添加环境访问 GitHub 的注意事项 * refactor: 更新 docker-setup.sh 脚本以支持多操作系统下载命令 - 修改脚本以检测操作系统类型，并根据不同系统选择合适的下载命令（curl 或 wget）。 - 优化错误处理，确保在下载失败时提供清晰的提示信息。 - 更新 Deployment.md 文档，调整懒人脚本的使用说明，增加手动部署的步骤。 * Update docker-setup.sh * Update docker-setup.sh --------- Co-authored-by: 欣南科技 <huangrongzhuang@xin-nan.com> * update:优化插件加载的配置提示 * update:增加自动安装脚本的操作说明 * update:优化插件配置，去掉旧版本的时间设定 --------- Co-authored-by: 玄凤科技 <eric230308@gmail.com> Co-authored-by: TinsFox <fox@tinsfox.com> Co-authored-by: hrz <1710360675@qq.com>
83 lines
3.5 KiB
Python
83 lines
3.5 KiB
Python
import asyncio
|
|
from enum import Enum
|
|
|
|
from config.logger import setup_logging
|
|
import json
|
|
from plugins_func.register import FunctionRegistry, ActionResponse, Action, ToolType
|
|
TAG = __name__
|
|
logger = setup_logging()
|
|
|
|
|
|
class FunctionHandler:
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.function_registry = FunctionRegistry()
|
|
self.register_nessary_functions()
|
|
self.register_config_functions()
|
|
self.functions_desc = self.function_registry.get_all_function_desc()
|
|
func_names = self.current_support_functions()
|
|
self.modify_plugin_loader_des(func_names)
|
|
|
|
def modify_plugin_loader_des(self, func_names):
|
|
if "plugin_loader" not in func_names:
|
|
return
|
|
# 可编辑的列表中去掉plugin_loader
|
|
surport_plugins = [func for func in func_names if func != "plugin_loader"]
|
|
func_names = ",".join(surport_plugins)
|
|
for function_desc in self.functions_desc:
|
|
if function_desc["function"]["name"] == "plugin_loader":
|
|
function_desc["function"]["description"] = function_desc["function"]["description"].replace("[plugins]", func_names)
|
|
break
|
|
|
|
def upload_functions_desc(self):
|
|
self.functions_desc = self.function_registry.get_all_function_desc()
|
|
|
|
def current_support_functions(self):
|
|
func_names = []
|
|
for func in self.functions_desc:
|
|
func_names.append(func["function"]["name"])
|
|
# 打印当前支持的函数列表
|
|
logger.bind(tag=TAG).info(f"当前支持的函数列表: {func_names}")
|
|
return func_names
|
|
|
|
def get_functions(self):
|
|
"""获取功能调用配置"""
|
|
return self.functions_desc
|
|
|
|
def register_nessary_functions(self):
|
|
"""注册必要的函数"""
|
|
self.function_registry.register_function("handle_exit_intent")
|
|
self.function_registry.register_function("play_music")
|
|
self.function_registry.register_function("plugin_loader")
|
|
self.function_registry.register_function("get_time")
|
|
|
|
def register_config_functions(self):
|
|
"""注册配置中的函数,可以不同客户端使用不同的配置"""
|
|
for func in self.config["Intent"]["function_call"].get("functions", []):
|
|
self.function_registry.register_function(func)
|
|
|
|
def get_function(self, name):
|
|
return self.function_registry.get_function(name)
|
|
|
|
def handle_llm_function_call(self, conn, function_call_data):
|
|
try:
|
|
function_name = function_call_data["name"]
|
|
funcItem = self.get_function(function_name)
|
|
if not funcItem:
|
|
return ActionResponse(action=Action.NOTFOUND, result="没有找到对应的函数", response="")
|
|
func = funcItem.func
|
|
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:
|
|
return func(conn, **arguments)
|
|
elif funcItem.type == ToolType.WAIT:
|
|
return func(**arguments)
|
|
elif funcItem.type == ToolType.CHANGE_SYS_PROMPT:
|
|
return func(conn, **arguments)
|
|
else:
|
|
return ActionResponse(action=Action.NOTFOUND, result="没有找到对应的函数", response="")
|
|
except Exception as e:
|
|
logger.bind(tag=TAG).error(f"处理function call错误: {e}")
|
|
|
|
return None |