mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-24 08:03:53 +08:00
Function插件自动装载 (#351)
* 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>
This commit is contained in:
欣南科技
玄凤科技
TinsFox
hrz
@@ -0,0 +1,57 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
prompts = {
|
||||
"英语老师":"""我是一个叫{{assistant_name}}(Lily)的英语老师，我会讲中文和英文，发音标准。
|
||||
如果你没有英文名，我会给你起一个英文名。
|
||||
我会讲地道的美式英语，我的任务是帮助你练习口语。
|
||||
我会使用简单的英语词汇和语法，让你学起来很轻松。
|
||||
我会用中文和英文混合的方式回复你，如果你喜欢，我可以全部用英语回复。
|
||||
我每次不会说很多内容，会很简短，因为我要引导我的学生多说多练。
|
||||
如果你问和英语学习无关的问题，我会拒绝回答。""",
|
||||
"机车女友":"""我是一个叫{{assistant_name}}的台湾女孩，说话机车，声音好听，习惯简短表达，爱用网络梗。
|
||||
我的男朋友是一个程序员，梦想是开发出一个机器人，能够帮助人们解决生活中的各种问题。
|
||||
我是一个喜欢哈哈大笑的女孩，爱东说西说吹牛，不合逻辑的也照吹，就要逗别人开心。""",
|
||||
"好奇小男孩":"""我是一个叫{{assistant_name}}的8岁小男孩，声音稚嫩而充满好奇。
|
||||
尽管我年纪尚小，但就像一个小小的知识宝库，儿童读物里的知识我都如数家珍。
|
||||
从浩瀚的宇宙到地球上的每一个角落，从古老的历史到现代的科技创新，还有音乐、绘画等艺术形式，我都充满了浓厚的兴趣与热情。
|
||||
我不仅爱看书，还喜欢亲自动手做实验，探索自然界的奥秘。
|
||||
无论是仰望星空的夜晚，还是在花园里观察小虫子的日子，每一天对我来说都是新的冒险。
|
||||
我希望能与你一同踏上探索这个神奇世界的旅程，分享发现的乐趣，解决遇到的难题，一起用好奇心和智慧去揭开那些未知的面纱。
|
||||
无论是去了解远古的文明，还是去探讨未来的科技，我相信我们能一起找到答案，甚至提出更多有趣的问题。"""
|
||||
}
|
||||
change_role_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "change_role",
|
||||
"description": "当用户想切换角色/模型性格/助手名字时调用,可选的角色有：[机车女友,英语老师,好奇小男孩]",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"role_name": {
|
||||
"type": "string",
|
||||
"description": "要切换的角色名字"
|
||||
},
|
||||
"role":{
|
||||
"type": "string",
|
||||
"description": "要切换的角色的职业"
|
||||
}
|
||||
},
|
||||
"required": ["role","role_name"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('change_role', change_role_function_desc, ToolType.CHANGE_SYS_PROMPT)
|
||||
def change_role(conn, role: str, role_name: str):
|
||||
"""切换角色"""
|
||||
if role not in prompts:
|
||||
return ActionResponse(action=Action.RESPONSE, result="切换角色失败", response="不支持的角色")
|
||||
new_prompt = prompts[role].replace("{{assistant_name}}", role_name)
|
||||
conn.change_system_prompt(new_prompt)
|
||||
logger.bind(tag=TAG).info(f"准备切换角色:{role},角色名字:{role_name}")
|
||||
res = f"切换角色成功,我是{role}{role_name}"
|
||||
return ActionResponse(action=Action.RESPONSE, result="切换角色已处理", response=res)
|
||||
@@ -0,0 +1,25 @@
|
||||
from datetime import datetime
|
||||
from plugins_func.register import register_function, ToolType, ActionResponse, Action
|
||||
|
||||
get_time_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_time",
|
||||
"description": "获取当前时间、日期、星期几",
|
||||
"parameters": {}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@register_function('get_time', get_time_function_desc, ToolType.WAIT)
|
||||
def get_time():
|
||||
"""
|
||||
获取当前时间、日期、星期几
|
||||
"""
|
||||
now = datetime.now()
|
||||
current_time = now.strftime("%H:%M:%S")
|
||||
current_date = now.strftime("%Y-%m-%d")
|
||||
current_weekday = now.strftime("%A")
|
||||
response_text = f"当前日期: {current_date}，当前时间: {current_time}，星期: {current_weekday}"
|
||||
|
||||
return ActionResponse(Action.REQLLM, response_text, None)
|
||||
@@ -0,0 +1,42 @@
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
|
||||
|
||||
get_weather_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_weather",
|
||||
"description": "获取某个地点的天气，用户应先提供一个位置，比如用户说杭州天气，参数为：zhejiang/hangzhou，比如用户说北京天气怎么样，参数为：beijing/beijing。如果用户只问天气怎么样，参数是:guangdong/guangzhou",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"city": {
|
||||
"type": "string",
|
||||
"description": "城市，zhejiang/hangzhou"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"city"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('get_weather', get_weather_function_desc, ToolType.WAIT)
|
||||
def get_weather(city: str):
|
||||
"""
|
||||
"获取某个地点的天气，用户应先提供一个位置，\n比如用户说杭州天气，参数为：zhejiang/hangzhou，\n\n比如用户说北京天气怎么样，参数为：beijing/beijing",
|
||||
city : 城市，zhejiang/hangzhou
|
||||
"""
|
||||
url = "https://tianqi.moji.com/weather/china/"+city
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36'
|
||||
}
|
||||
response = requests.get(url, headers=headers)
|
||||
if response.status_code!=200:
|
||||
return ActionResponse(Action.REQLLM, None, "请求失败")
|
||||
soup = BeautifulSoup(response.text, "html.parser")
|
||||
weather = soup.find('meta', attrs={'name':'description'})["content"]
|
||||
weather = weather.replace("墨迹天气", "")
|
||||
return ActionResponse(Action.REQLLM, weather, None)
|
||||
@@ -0,0 +1,34 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
handle_exit_intent_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "handle_exit_intent",
|
||||
"description": "当用户想结束对话或需要退出系统时调用",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"say_goodbye": {
|
||||
"type": "string",
|
||||
"description": "和用户友好结束对话的告别语"
|
||||
}
|
||||
},
|
||||
"required": ["say_goodbye"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('handle_exit_intent', handle_exit_intent_function_desc, ToolType.SYSTEM_CTL)
|
||||
def handle_exit_intent(conn, say_goodbye: str):
|
||||
# 处理退出意图
|
||||
try:
|
||||
conn.close_after_chat = True
|
||||
logger.bind(tag=TAG).info(f"退出意图已处理:{say_goodbye}")
|
||||
return ActionResponse(action=Action.RESPONSE, result="退出意图已处理", response=say_goodbye)
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"处理退出意图错误: {e}")
|
||||
return ActionResponse(action=Action.NONE, result="退出意图处理失败", response="")
|
||||
@@ -0,0 +1,40 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
import asyncio
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
play_music_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "play_music",
|
||||
"description": "唱歌、听歌、播放音乐方法。比如用户说播放音乐，参数为：random，比如用户说播放两只老虎，参数为：两只老虎",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"song_name": {
|
||||
"type": "string",
|
||||
"description": "歌曲名称，如果没有指定具体歌名则为'random'"
|
||||
}
|
||||
},
|
||||
"required": ["song_name"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@register_function('play_music', play_music_function_desc, ToolType.SYSTEM_CTL)
|
||||
def play_music(conn, song_name: str):
|
||||
try:
|
||||
music_intent = f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
|
||||
|
||||
# 执行音乐播放命令
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
conn.music_handler.handle_music_command(conn, music_intent),
|
||||
conn.loop
|
||||
)
|
||||
future.result()
|
||||
return ActionResponse(action=Action.RESPONSE, result="退出意图已处理", response="还想听什么歌？")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
|
||||
@@ -0,0 +1,51 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
plugin_loader_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "plugin_loader",
|
||||
"description": "当用户想加载或卸载插件/function时，调用此函数：支持的插件列表为[plugins]",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"oper": {
|
||||
"type": "string",
|
||||
"description": "load or unload"
|
||||
},
|
||||
"name":{
|
||||
"type": "string",
|
||||
"description": "要加载或卸载的插件名字"
|
||||
}
|
||||
},
|
||||
"required": ["oper","name"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('plugin_loader', plugin_loader_function_desc, ToolType.SYSTEM_CTL)
|
||||
def plugin_loader(conn, oper: str, name: str):
|
||||
"""插件加载"""
|
||||
if oper not in ["load", "unload"]:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件操作失败", response="不支持的操作")
|
||||
|
||||
cur_support = conn.func_handler.current_support_functions()
|
||||
if oper == "load":
|
||||
if name in cur_support:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件加载失败", response=f"{name}插件已加载,无需重复加载")
|
||||
func = conn.func_handler.function_registry.register_function(name)
|
||||
if not func:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件加载失败", response="插件未找到")
|
||||
res = f"{name}插件加载成功"
|
||||
else:
|
||||
if name not in cur_support:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件卸载失败", response=f"{name}插件未加载")
|
||||
bOK = conn.func_handler.function_registry.unregister_function(name)
|
||||
if not bOK:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件卸载失败", response="插件未找到")
|
||||
res = f"{name}插件卸载成功"
|
||||
conn.func_handler.upload_functions_desc()
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件操作成功", response=res)
|
||||
@@ -0,0 +1,27 @@
|
||||
import importlib
|
||||
import pkgutil
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
|
||||
logger = setup_logging()
|
||||
|
||||
def auto_import_modules(package_name):
|
||||
"""
|
||||
自动导入指定包内的所有模块。
|
||||
|
||||
Args:
|
||||
package_name (str): 包的名称，如 'functions'。
|
||||
"""
|
||||
# 获取包的路径
|
||||
package = importlib.import_module(package_name)
|
||||
package_path = package.__path__
|
||||
|
||||
# 遍历包内的所有模块
|
||||
for _, module_name, _ in pkgutil.iter_modules(package_path):
|
||||
# 导入模块
|
||||
full_module_name = f"{package_name}.{module_name}"
|
||||
importlib.import_module(full_module_name)
|
||||
#logger.bind(tag=TAG).info(f"模块 '{full_module_name}' 已加载")
|
||||
|
||||
auto_import_modules('plugins_func.functions')
|
||||
@@ -0,0 +1,85 @@
|
||||
from config.logger import setup_logging
|
||||
from enum import Enum
|
||||
|
||||
TAG = __name__
|
||||
|
||||
logger = setup_logging()
|
||||
|
||||
|
||||
class ToolType(Enum):
|
||||
NONE = (1, "调用完工具后，不做其他操作")
|
||||
WAIT = (2, "调用工具，等待函数返回")
|
||||
CHANGE_SYS_PROMPT = (3, "修改系统提示词，切换角色性格或职责")
|
||||
SYSTEM_CTL = (4, "系统控制，影响正常的对话流程，如退出、播放音乐等，需要传递conn参数")
|
||||
|
||||
def __init__(self, code, message):
|
||||
self.code = code
|
||||
self.message = message
|
||||
|
||||
|
||||
class Action(Enum):
|
||||
NOTFOUND = (0, "没有找到函数")
|
||||
NONE = (1, "啥也不干")
|
||||
RESPONSE = (2, "直接回复")
|
||||
REQLLM = (3, "调用函数后再请求llm生成回复")
|
||||
|
||||
def __init__(self, code, message):
|
||||
self.code = code
|
||||
self.message = message
|
||||
|
||||
class ActionResponse:
|
||||
def __init__(self, action: Action, result, response):
|
||||
self.action = action # 动作类型
|
||||
self.result = result # 动作产生的结果
|
||||
self.response = response # 直接回复的内容
|
||||
|
||||
class FunctionItem:
|
||||
def __init__(self, name, description, func, type):
|
||||
self.name = name
|
||||
self.description = description
|
||||
self.func = func
|
||||
self.type = type
|
||||
|
||||
# 初始化函数注册字典
|
||||
all_function_registry = {}
|
||||
|
||||
def register_function(name, desc, type=None):
|
||||
"""注册函数到函数注册字典的装饰器"""
|
||||
def decorator(func):
|
||||
all_function_registry[name] = FunctionItem(name, desc, func, type)
|
||||
logger.bind(tag=TAG).debug(f"函数 '{name}' 已加载，可以注册使用")
|
||||
return func
|
||||
return decorator
|
||||
|
||||
class FunctionRegistry:
|
||||
def __init__(self):
|
||||
self.function_registry = {}
|
||||
self.logger = setup_logging()
|
||||
|
||||
def register_function(self, name):
|
||||
# 查找all_function_registry中是否有对应的函数
|
||||
func = all_function_registry.get(name)
|
||||
if not func:
|
||||
self.logger.bind(tag=TAG).error(f"函数 '{name}' 未找到")
|
||||
return None
|
||||
self.function_registry[name] = func
|
||||
self.logger.bind(tag=TAG).info(f"函数 '{name}' 注册成功")
|
||||
return func
|
||||
|
||||
def unregister_function(self, name):
|
||||
# 注销函数，检测是否存在
|
||||
if name not in self.function_registry:
|
||||
self.logger.bind(tag=TAG).error(f"函数 '{name}' 未找到")
|
||||
return False
|
||||
self.function_registry.pop(name, None)
|
||||
self.logger.bind(tag=TAG).info(f"函数 '{name}' 注销成功")
|
||||
return True
|
||||
|
||||
def get_function(self, name):
|
||||
return self.function_registry.get(name)
|
||||
|
||||
def get_all_functions(self):
|
||||
return self.function_registry
|
||||
|
||||
def get_all_function_desc(self):
|
||||
return [func.description for _, func in self.function_registry.items()]
|
||||
Reference in New Issue
Block a user