update:优化代码

This commit is contained in:
hrz
2025-11-08 16:00:37 +08:00
parent d09fa0c1f6
commit c02f2105d7
4 changed files with 141 additions and 105 deletions
@@ -1,6 +1,6 @@
"""服务端插件工具执行器"""
from typing import Dict, Any, Optional, Tuple
from typing import Dict, Any
from ..base import ToolType, ToolDefinition, ToolExecutor
from plugins_func.register import all_function_registry, Action, ActionResponse
@@ -11,53 +11,12 @@ class ServerPluginExecutor(ToolExecutor):
def __init__(self, conn):
self.conn = conn
self.config = conn.config
# 存储知识库工具名称到真实插件名称的映射
self._knowledge_base_mapping: Dict[str, str] = {}
def _parse_knowledge_base_config(self, config_name: str) -> Optional[str]:
"""
解析知识库配置名称,提取真实的插件名称
Args:
config_name: 配置名称,格式为 xzKnowledgeBase_<plugin_name>_<index>
Returns:
真实的插件名称,如果不是知识库配置则返回 None
Example:
"xzKnowledgeBase_search_from_ragflow_0" -> "search_from_ragflow"
"xzKnowledgeBase_search_from_ragflow_1" -> "search_from_ragflow"
"""
if not config_name.startswith("xzKnowledgeBase_"):
return None
# 移除前缀
name_without_prefix = config_name[len("xzKnowledgeBase_"):]
# 找到最后一个下划线的位置
last_underscore_index = name_without_prefix.rfind("_")
if last_underscore_index == -1:
return None
# 提取真实插件名称(从开头到最后一个下划线之前)
real_plugin_name = name_without_prefix[:last_underscore_index]
return real_plugin_name
async def execute(
self, conn, tool_name: str, arguments: Dict[str, Any]
) -> ActionResponse:
"""执行服务端插件工具"""
# 检查是否是知识库工具调用
real_plugin_name = self._knowledge_base_mapping.get(tool_name)
if real_plugin_name:
# 使用真实的插件名称获取函数
func_item = all_function_registry.get(real_plugin_name)
else:
# 普通插件调用
func_item = all_function_registry.get(tool_name)
func_item = all_function_registry.get(tool_name)
if not func_item:
return ActionResponse(
action=Action.NOTFOUND, response=f"插件函数 {tool_name} 不存在"
@@ -112,51 +71,27 @@ class ServerPluginExecutor(ToolExecutor):
for func_name in all_required_functions:
func_item = all_function_registry.get(func_name)
if func_item:
# 从函数注册中获取描述
fun_description = (
self.config.get("plugins", {})
.get(func_name, {})
.get("description", "")
)
if fun_description is not None and len(fun_description) > 0:
if "function" in func_item.description and isinstance(
func_item.description["function"], dict
):
func_item.description["function"][
"description"
] = fun_description
tools[func_name] = ToolDefinition(
name=func_name,
description=func_item.description,
tool_type=ToolType.SERVER_PLUGIN,
)
# 处理知识库配置
plugins_config = self.config.get("plugins", {})
for config_name, config_value in plugins_config.items():
# 检查是否是知识库配置
real_plugin_name = self._parse_knowledge_base_config(config_name)
if real_plugin_name:
# 获取真实的插件函数
func_item = all_function_registry.get(real_plugin_name)
if func_item and isinstance(config_value, dict):
# 从配置中获取自定义的 name 和 description
custom_name = config_value.get("name", "")
custom_description = config_value.get("description", "")
# 创建动态的工具名称(使用配置名称去掉前缀部分作为工具名)
tool_name = config_name[len("xzKnowledgeBase_"):]
# 复制原始函数描述并修改
custom_func_desc = func_item.description.copy()
if "function" in custom_func_desc:
custom_func_desc["function"] = custom_func_desc["function"].copy()
custom_func_desc["function"]["name"] = tool_name
custom_func_desc["function"]["description"] = custom_description
# 注册工具
tools[tool_name] = ToolDefinition(
name=tool_name,
description=custom_func_desc,
tool_type=ToolType.SERVER_PLUGIN,
)
# 保存映射关系
self._knowledge_base_mapping[tool_name] = real_plugin_name
return tools
def has_tool(self, tool_name: str) -> bool:
"""检查是否有指定的服务端插件工具"""
# 检查是否是知识库工具
if tool_name in self._knowledge_base_mapping:
return True
# 检查是否是普通工具
return tool_name in all_function_registry