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参数",
|
)
|
IOT_CTL = (5, "IOT设备控制，需要传递conn参数")
|
MCP_CLIENT = (6, "MCP客户端")
|
|
def __init__(self, code, message):
|
self.code = code
|
self.message = message
|
|
|
class Action(Enum):
|
ERROR = (-1, "错误")
|
NOTFOUND = (0, "没有找到函数")
|
NONE = (1, "啥也不干")
|
RESPONSE = (2, "直接回复")
|
REQLLM = (3, "调用函数后再请求llm生成回复")
|
RECORD = (4, "记录工具调用到对话历史，不调用LLM")
|
|
def __init__(self, code, message):
|
self.code = code
|
self.message = message
|
|
|
class ActionResponse:
|
def __init__(self, action: Action, result=None, response=None):
|
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
|
|
|
class DeviceTypeRegistry:
|
"""设备类型注册表，用于管理IOT设备类型及其函数"""
|
|
def __init__(self):
|
self.type_functions = {} # type_signature -> {func_name: FunctionItem}
|
|
def generate_device_type_id(self, descriptor):
|
"""通过设备能力描述生成类型ID"""
|
properties = sorted(descriptor["properties"].keys())
|
methods = sorted(descriptor["methods"].keys())
|
# 使用属性和方法的组合作为设备类型的唯一标识
|
type_signature = (
|
f"{descriptor['name']}:{','.join(properties)}:{','.join(methods)}"
|
)
|
return type_signature
|
|
def get_device_functions(self, type_id):
|
"""获取设备类型对应的所有函数"""
|
return self.type_functions.get(type_id, {})
|
|
def register_device_type(self, type_id, functions):
|
"""注册设备类型及其函数"""
|
if type_id not in self.type_functions:
|
self.type_functions[type_id] = functions
|
|
|
# 初始化函数注册字典
|
all_function_registry = {}
|
# 模块名 -> 函数名列表的映射，用于将模块级别的插件名展开为具体的函数名
|
module_func_map = {}
|
|
|
def register_function(name, desc, type=None):
|
"""注册函数到函数注册字典的装饰器"""
|
|
def decorator(func):
|
all_function_registry[name] = FunctionItem(name, desc, func, type)
|
# 记录模块名到函数名的映射，用于 expand 模块级别的插件配置
|
module_name = func.__module__.split(".")[-1]
|
module_func_map.setdefault(module_name, []).append(name)
|
logger.bind(tag=TAG).debug(f"函数 '{name}' 已加载，可以注册使用")
|
return func
|
|
return decorator
|
|
|
def register_device_function(name, desc, type=None):
|
"""注册设备级别的函数到函数注册字典的装饰器"""
|
|
def decorator(func):
|
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, func_item=None):
|
# 如果提供了func_item，直接注册
|
if func_item:
|
self.function_registry[name] = func_item
|
self.logger.bind(tag=TAG).debug(f"函数 '{name}' 直接注册成功")
|
return func_item
|
|
# 否则从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).debug(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()]
|
