adai/MemoryBear
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tool system): tool system development

Browse Source
This commit is contained in:
谢俊男
2025-12-20 16:03:06 +08:00
parent f838b11d97
commit b00d6e37e3
2 changed files with 179 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files

356
api/app/core/workflow/executor.py
View File

@@ -5,7 +5,7 @@
""" """
import logging import logging
import uuid # import uuid
import datetime import datetime
from typing import Any from typing import Any
@@ -16,10 +16,11 @@ from langgraph.graph.state import CompiledStateGraph
from app.core.workflow.expression_evaluator import evaluate_condition from app.core.workflow.expression_evaluator import evaluate_condition
from app.core.workflow.nodes import WorkflowState, NodeFactory from app.core.workflow.nodes import WorkflowState, NodeFactory
from app.core.workflow.nodes.enums import NodeType from app.core.workflow.nodes.enums import NodeType
from app.core.tools.registry import ToolRegistry # from app.core.tools.registry import ToolRegistry
from app.core.tools.executor import ToolExecutor # from app.core.tools.executor import ToolExecutor
from app.core.tools.langchain_adapter import LangchainAdapter # from app.core.tools.langchain_adapter import LangchainAdapter
TOOL_MANAGEMENT_AVAILABLE = True # TOOL_MANAGEMENT_AVAILABLE = True
# from app.db import get_db
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -466,176 +467,175 @@ async def execute_workflow_stream(
# ==================== 工具管理系统集成 ==================== # ==================== 工具管理系统集成 ====================
def get_workflow_tools(workspace_id: str, user_id: str) -> list: # def get_workflow_tools(workspace_id: str, user_id: str) -> list:
"""获取工作流可用的工具列表 # """获取工作流可用的工具列表
#
Args: # Args:
workspace_id: 工作空间ID # workspace_id: 工作空间ID
user_id: 用户ID # user_id: 用户ID
#
Returns: # Returns:
可用工具列表 # 可用工具列表
""" # """
if not TOOL_MANAGEMENT_AVAILABLE: # if not TOOL_MANAGEMENT_AVAILABLE:
logger.warning("工具管理系统不可用") # logger.warning("工具管理系统不可用")
return [] # return []
#
try: # try:
from sqlalchemy.orm import Session # db = next(get_db())
db = next(get_db()) #
# # 创建工具注册表
# 创建工具注册表 # registry = ToolRegistry(db)
registry = ToolRegistry(db) #
# # 注册内置工具类
# 注册内置工具类 # from app.core.tools.builtin import (
from app.core.tools.builtin import ( # DateTimeTool, JsonTool, BaiduSearchTool, MinerUTool, TextInTool
DateTimeTool, JsonTool, BaiduSearchTool, MinerUTool, TextInTool # )
) # registry.register_tool_class(DateTimeTool)
registry.register_tool_class(DateTimeTool) # registry.register_tool_class(JsonTool)
registry.register_tool_class(JsonTool) # registry.register_tool_class(BaiduSearchTool)
registry.register_tool_class(BaiduSearchTool) # registry.register_tool_class(MinerUTool)
registry.register_tool_class(MinerUTool) # registry.register_tool_class(TextInTool)
registry.register_tool_class(TextInTool) #
# # 获取活跃的工具
# 获取活跃的工具 # import uuid
import uuid # tools = registry.list_tools(workspace_id=uuid.UUID(workspace_id))
tools = registry.list_tools(workspace_id=uuid.UUID(workspace_id)) # active_tools = [tool for tool in tools if tool.status.value == "active"]
active_tools = [tool for tool in tools if tool.status.value == "active"] #
# # 转换为Langchain工具
# 转换为Langchain工具 # langchain_tools = []
langchain_tools = [] # for tool_info in active_tools:
for tool_info in active_tools: # try:
try: # tool_instance = registry.get_tool(tool_info.id)
tool_instance = registry.get_tool(tool_info.id) # if tool_instance:
if tool_instance: # langchain_tool = LangchainAdapter.convert_tool(tool_instance)
langchain_tool = LangchainAdapter.convert_tool(tool_instance) # langchain_tools.append(langchain_tool)
langchain_tools.append(langchain_tool) # except Exception as e:
except Exception as e: # logger.error(f"转换工具失败: {tool_info.name}, 错误: {e}")
logger.error(f"转换工具失败: {tool_info.name}, 错误: {e}") #
# logger.info(f"为工作流获取了 {len(langchain_tools)} 个工具")
logger.info(f"为工作流获取了 {len(langchain_tools)} 个工具") # return langchain_tools
return langchain_tools #
# except Exception as e:
except Exception as e: # logger.error(f"获取工作流工具失败: {e}")
logger.error(f"获取工作流工具失败: {e}") # return []
return [] #
#
# class ToolWorkflowNode:
class ToolWorkflowNode: # """工具工作流节点 - 在工作流中执行工具"""
"""工具工作流节点 - 在工作流中执行工具""" #
# def __init__(self, node_config: dict, workflow_config: dict):
def __init__(self, node_config: dict, workflow_config: dict): # """初始化工具节点
"""初始化工具节点 #
# Args:
Args: # node_config: 节点配置
node_config: 节点配置 # workflow_config: 工作流配置
workflow_config: 工作流配置 # """
""" # self.node_config = node_config
self.node_config = node_config # self.workflow_config = workflow_config
self.workflow_config = workflow_config # self.tool_id = node_config.get("tool_id")
self.tool_id = node_config.get("tool_id") # self.tool_parameters = node_config.get("parameters", {})
self.tool_parameters = node_config.get("parameters", {}) #
# async def run(self, state: WorkflowState) -> WorkflowState:
async def run(self, state: WorkflowState) -> WorkflowState: # """执行工具节点"""
"""执行工具节点""" # if not TOOL_MANAGEMENT_AVAILABLE:
if not TOOL_MANAGEMENT_AVAILABLE: # logger.error("工具管理系统不可用")
logger.error("工具管理系统不可用") # state["error"] = "工具管理系统不可用"
state["error"] = "工具管理系统不可用" # return state
return state #
# try:
try: # from sqlalchemy.orm import Session
from sqlalchemy.orm import Session # db = next(get_db())
db = next(get_db()) #
# # 创建工具执行器
# 创建工具执行器 # registry = ToolRegistry(db)
registry = ToolRegistry(db) # executor = ToolExecutor(db, registry)
executor = ToolExecutor(db, registry) #
# # 准备参数（支持变量替换）
# 准备参数（支持变量替换） # parameters = self._prepare_parameters(state)
parameters = self._prepare_parameters(state) #
# # 执行工具
# 执行工具 # result = await executor.execute_tool(
result = await executor.execute_tool( # tool_id=self.tool_id,
tool_id=self.tool_id, # parameters=parameters,
parameters=parameters, # user_id=uuid.UUID(state["user_id"]),
user_id=uuid.UUID(state["user_id"]), # workspace_id=uuid.UUID(state["workspace_id"])
workspace_id=uuid.UUID(state["workspace_id"]) # )
) #
# # 更新状态
# 更新状态 # node_id = self.node_config.get("id")
node_id = self.node_config.get("id") # if result.success:
if result.success: # state["node_outputs"][node_id] = {
state["node_outputs"][node_id] = { # "type": "tool",
"type": "tool", # "tool_id": self.tool_id,
"tool_id": self.tool_id, # "output": result.data,
"output": result.data, # "execution_time": result.execution_time,
"execution_time": result.execution_time, # "token_usage": result.token_usage
"token_usage": result.token_usage # }
} #
# # 更新运行时变量
# 更新运行时变量 # if isinstance(result.data, dict):
if isinstance(result.data, dict): # for key, value in result.data.items():
for key, value in result.data.items(): # state["runtime_vars"][f"{node_id}.{key}"] = value
state["runtime_vars"][f"{node_id}.{key}"] = value # else:
else: # state["runtime_vars"][f"{node_id}.result"] = result.data
state["runtime_vars"][f"{node_id}.result"] = result.data # else:
else: # state["error"] = result.error
state["error"] = result.error # state["error_node"] = node_id
state["error_node"] = node_id # state["node_outputs"][node_id] = {
state["node_outputs"][node_id] = { # "type": "tool",
"type": "tool", # "tool_id": self.tool_id,
"tool_id": self.tool_id, # "error": result.error,
"error": result.error, # "execution_time": result.execution_time
"execution_time": result.execution_time # }
} #
# return state
return state #
# except Exception as e:
except Exception as e: # logger.error(f"工具节点执行失败: {e}")
logger.error(f"工具节点执行失败: {e}") # state["error"] = str(e)
state["error"] = str(e) # state["error_node"] = self.node_config.get("id")
state["error_node"] = self.node_config.get("id") # return state
return state #
# def _prepare_parameters(self, state: WorkflowState) -> dict:
def _prepare_parameters(self, state: WorkflowState) -> dict: # """准备工具参数（支持变量替换）"""
"""准备工具参数（支持变量替换）""" # parameters = {}
parameters = {} #
# for key, value in self.tool_parameters.items():
for key, value in self.tool_parameters.items(): # if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
if isinstance(value, str) and value.startswith("${") and value.endswith("}"): # # 变量替换
# 变量替换 # var_path = value[2:-1]
var_path = value[2:-1] #
# # 支持多层级变量访问，如 ${sys.message} 或 ${node1.result}
# 支持多层级变量访问，如 ${sys.message} 或 ${node1.result} # if "." in var_path:
if "." in var_path: # parts = var_path.split(".")
parts = var_path.split(".") # current = state.get("variables", {})
current = state.get("variables", {}) #
# for part in parts:
for part in parts: # if isinstance(current, dict) and part in current:
if isinstance(current, dict) and part in current: # current = current[part]
current = current[part] # else:
else: # # 尝试从运行时变量获取
# 尝试从运行时变量获取 # runtime_key = ".".join(parts)
runtime_key = ".".join(parts) # current = state.get("runtime_vars", {}).get(runtime_key, value)
current = state.get("runtime_vars", {}).get(runtime_key, value) # break
break #
# parameters[key] = current
parameters[key] = current # else:
else: # # 简单变量
# 简单变量 # variables = state.get("variables", {})
variables = state.get("variables", {}) # parameters[key] = variables.get(var_path, value)
parameters[key] = variables.get(var_path, value) # else:
else: # parameters[key] = value
parameters[key] = value #
# return parameters
return parameters #
#
# # 注册工具节点到NodeFactory（如果存在）
# 注册工具节点到NodeFactory（如果存在） # try:
try: # from app.core.workflow.nodes import NodeFactory
from app.core.workflow.nodes import NodeFactory # if hasattr(NodeFactory, 'register_node_type'):
if hasattr(NodeFactory, 'register_node_type'): # NodeFactory.register_node_type("tool", ToolWorkflowNode)
NodeFactory.register_node_type("tool", ToolWorkflowNode) # logger.info("工具节点已注册到工作流系统")
logger.info("工具节点已注册到工作流系统") # except Exception as e:
except Exception as e: # logger.warning(f"注册工具节点失败: {e}")
logger.warning(f"注册工具节点失败: {e}")

219
api/app/services/agent_tools.py
View File

@@ -13,10 +13,6 @@ from app.core.exceptions import BusinessException, ResourceNotFoundException
from app.core.error_codes import BizCode from app.core.error_codes import BizCode
from app.core.logging_config import get_business_logger from app.core.logging_config import get_business_logger
from app.repositories import workspace_repository, knowledge_repository from app.repositories import workspace_repository, knowledge_repository
from app.core.tools.registry import ToolRegistry
from app.core.tools.executor import ToolExecutor
from app.core.tools.langchain_adapter import LangchainAdapter
TOOL_MANAGEMENT_AVAILABLE = True
logger = get_business_logger() logger = get_business_logger()
@@ -333,217 +329,4 @@ def create_agent_invocation_tool(
) )
return f"调用 Agent 失败: {str(e)}" return f"调用 Agent 失败: {str(e)}"
return invoke_agent return invoke_agent
def get_available_tools_for_agent(
db: Session,
workspace_id: uuid.UUID,
agent_id: Optional[uuid.UUID] = None
) -> List[Dict[str, Any]]:
"""获取Agent可用的工具列表
Args:
db: 数据库会话
workspace_id: 工作空间ID
agent_id: Agent ID（可选）
Returns:
可用工具列表
"""
if not TOOL_MANAGEMENT_AVAILABLE:
logger.warning("工具管理系统不可用")
return []
try:
# 创建工具注册表
registry = ToolRegistry(db)
# 获取工具列表
tools = registry.list_tools(workspace_id=workspace_id)
# 转换为Agent可用的格式
available_tools = []
for tool_info in tools:
if tool_info.status.value == "active":
available_tools.append({
"id": tool_info.id,
"name": tool_info.name,
"description": tool_info.description,
"type": tool_info.tool_type.value,
"version": tool_info.version,
"tags": tool_info.tags,
"parameters": [
{
"name": param.name,
"type": param.type.value,
"description": param.description,
"required": param.required,
"default": param.default
}
for param in tool_info.parameters
]
})
logger.info(f"为Agent获取到 {len(available_tools)} 个可用工具")
return available_tools
except Exception as e:
logger.error(f"获取Agent可用工具失败: {e}")
return []
def create_langchain_tools_for_agent(
db: Session,
workspace_id: uuid.UUID,
agent_id: Optional[uuid.UUID] = None
) -> List[Any]:
"""为Agent创建Langchain兼容的工具列表
Args:
db: 数据库会话
workspace_id: 工作空间ID
agent_id: Agent ID（可选）
Returns:
Langchain工具列表
"""
if not TOOL_MANAGEMENT_AVAILABLE:
logger.warning("工具管理系统不可用")
return []
try:
# 创建工具注册表
registry = ToolRegistry(db)
# 注册内置工具类
from app.core.tools.builtin import (
DateTimeTool, JsonTool, BaiduSearchTool, MinerUTool, TextInTool
)
registry.register_tool_class(DateTimeTool)
registry.register_tool_class(JsonTool)
registry.register_tool_class(BaiduSearchTool)
registry.register_tool_class(MinerUTool)
registry.register_tool_class(TextInTool)
# 获取活跃的工具
tools = registry.list_tools(workspace_id=workspace_id)
active_tools = [tool for tool in tools if tool.status.value == "active"]
# 转换为Langchain工具
langchain_tools = []
for tool_info in active_tools:
try:
tool_instance = registry.get_tool(tool_info.id)
if tool_instance:
langchain_tool = LangchainAdapter.convert_tool(tool_instance)
langchain_tools.append(langchain_tool)
except Exception as e:
logger.error(f"转换工具失败: {tool_info.name}, 错误: {e}")
logger.info(f"为Agent创建了 {len(langchain_tools)} 个Langchain工具")
return langchain_tools
except Exception as e:
logger.error(f"创建Agent Langchain工具失败: {e}")
return []
class ToolExecutionInput(BaseModel):
"""工具执行输入参数"""
tool_id: str = Field(..., description="工具ID")
parameters: Dict[str, Any] = Field(default_factory=dict, description="工具参数")
timeout: Optional[float] = Field(None, description="超时时间（秒）")
def create_tool_execution_tool(
db: Session,
workspace_id: uuid.UUID,
user_id: uuid.UUID
):
"""创建工具执行工具
Args:
db: 数据库会话
workspace_id: 工作空间ID
user_id: 用户ID
Returns:
工具执行工具
"""
if not TOOL_MANAGEMENT_AVAILABLE:
logger.warning("工具管理系统不可用")
return None
@tool(args_schema=ToolExecutionInput)
async def execute_tool(
tool_id: str,
parameters: Dict[str, Any] = None,
timeout: Optional[float] = None
) -> str:
"""执行指定的工具。当需要使用系统中的工具来完成特定任务时使用。
Args:
tool_id: 工具ID（通过工具列表获取）
parameters: 工具参数（根据工具要求提供）
timeout: 超时时间（秒，可选）
Returns:
工具执行结果
"""
try:
# 创建工具执行器
registry = ToolRegistry(db)
executor = ToolExecutor(db, registry)
# 执行工具
result = await executor.execute_tool(
tool_id=tool_id,
parameters=parameters or {},
user_id=user_id,
workspace_id=workspace_id,
timeout=timeout
)
if result.success:
# 格式化成功结果
if isinstance(result.data, str):
return result.data
else:
import json
return json.dumps(result.data, ensure_ascii=False, indent=2)
else:
return f"工具执行失败: {result.error}"
except Exception as e:
logger.error(f"工具执行异常: {tool_id}, 错误: {e}")
return f"工具执行异常: {str(e)}"
return execute_tool
def get_tool_management_tools(
db: Session,
workspace_id: uuid.UUID,
user_id: uuid.UUID
) -> List[Any]:
"""获取工具管理相关的工具
Args:
db: 数据库会话
workspace_id: 工作空间ID
user_id: 用户ID
Returns:
工具管理工具列表
"""
if not TOOL_MANAGEMENT_AVAILABLE:
return []
tools = []
# 添加工具执行工具
execution_tool = create_tool_execution_tool(db, workspace_id, user_id)
if execution_tool:
tools.append(execution_tool)
return tools