LongjoeAgent/tools/tool_factory.py

import os
import sys
import importlib
import inspect
from pathlib import Path
from typing import List
from langchain.tools import BaseTool


def get_all_tools() -> List[BaseTool]:
    """
    自动发现并返回所有工具 - 修复检查逻辑
    """
    tools = []

    print("🛠️ 开始自动发现工具...")

    # 获取项目根目录
    project_root = Path(__file__).parent.parent
    tools_dir = Path(__file__).parent

    # 将项目根目录添加到Python路径
    if str(project_root) not in sys.path:
        sys.path.insert(0, str(project_root))
    # 导入配置以检查知识库开关
    from config.settings import settings

    # 扫描工具文件
    tool_files = []
    # 模式1: 编译前的.py文件
    for file_path in tools_dir.glob("*_tools.py"):
        if file_path.is_file():
            module_name = file_path.stem
            # 如果知识库被禁用，跳过知识库工具
            if module_name == "knowledge_tools" and not settings.KNOWLEDGE_BASE_ENABLED:
                print("知识库功能已禁用，跳过知识库工具")
                continue
            tool_files.append(module_name)
            print(f"发现工具文件: {module_name}")

    # 模式2: 编译后的.pyd文件
    for file_path in tools_dir.glob("*_tools.cp*.pyd"):
        if file_path.is_file():
            # 从文件名中提取模块名，如: ware_tools.cp313-win_amd64.pyd -> ware_tools
            module_name = file_path.stem.split(".")[0]
            # 如果知识库被禁用，跳过知识库工具
            if module_name == "knowledge_tools" and not settings.KNOWLEDGE_BASE_ENABLED:
                print("知识库功能已禁用，跳过知识库工具")
                continue
            if module_name not in tool_files:  # 避免重复添加
                tool_files.append(module_name)
                print(f"发现编译后工具文件: {module_name}")

    if not tool_files:
        # 如果知识库被禁用，从默认列表中移除知识库工具
        default_tools = ["sale_tools", "ware_tools", "price_tools", "money_tools"]
        if settings.KNOWLEDGE_BASE_ENABLED:
            default_tools.insert(0, "knowledge_tools")
        tool_files = default_tools
        print("使用默认工具列表")

    for module_name in tool_files:
        try:
            # 导入模块
            full_module_path = f"tools.{module_name}"
            module = importlib.import_module(full_module_path)
            # print(f"加载模块: {module_name}")

            # 查找工具 - 使用更全面的方法
            tool_count = 0

            # 方法1: 检查模块的所有属性
            for attr_name in dir(module):
                if attr_name.startswith("_"):
                    continue

                attr = getattr(module, attr_name)

                # # 详细调试信息
                # print(f"  检查 {attr_name}:")
                # print(f"    类型: {type(attr)}")

                # 检查是否是BaseTool实例
                if isinstance(attr, BaseTool):
                    tools.append(attr)
                    tool_count += 1
                    # print(f"  发现BaseTool工具: {getattr(attr, 'name', attr_name)}")
                    continue

                # 检查是否是函数且具有工具属性
                if callable(attr):
                    # print(f"    是否有name属性: {hasattr(attr, 'name')}")
                    # if hasattr(attr, "name"):
                    #     print(f"    name值: {getattr(attr, 'name', '无')}")
                    # print(f"    是否有description属性: {hasattr(attr, 'description')}")
                    # if hasattr(attr, "description"):
                    #     print(
                    #         f"    description前50字: {getattr(attr, 'description', '')[:50]}"
                    #     )

                    # 检查是否是工具函数
                    if is_tool_function(attr):
                        tools.append(attr)
                        tool_count += 1
                        # print(f"  发现工具函数: {attr_name}")

            # 方法2: 检查模块的全局变量
            print(f"  🔍 检查模块全局变量...")
            for name, value in module.__dict__.items():
                if name.startswith("_"):
                    continue

                if isinstance(value, BaseTool):
                    if value not in tools:  # 避免重复添加
                        tools.append(value)
                        tool_count += 1
                        # print(
                        #     f"  从全局变量发现BaseTool工具: {getattr(value, 'name', name)}"
                        # )

            if tool_count == 0:
                # print(f"  模块 {module_name} 中未发现工具")
                # 尝试手动创建工具
                manual_tools = create_tools_manually(module_name, module)
                if manual_tools:
                    tools.extend(manual_tools)
                    tool_count = len(manual_tools)
                    # print(f"  🔧 手动创建了 {tool_count} 个工具")
            else:
                print(f"  模块 {module_name} 中发现 {tool_count} 个工具")

        except Exception as e:
            print(f"加载模块 {module_name} 失败: {e}")

    print(f"总共发现 {len(tools)} 个工具")

    # # 打印工具详情
    # for i, tool in enumerate(tools):
    #     tool_name = getattr(tool, "name", f"tool_{i+1}")
    #     tool_desc = getattr(tool, "description", "无描述")
    #     print(f"  {i+1}. {tool_name}: {tool_desc[:50]}...")

    return tools


def create_tools_manually(module_name: str, module) -> List[BaseTool]:
    """手动创建工具 - 针对@tool装饰器的问题"""
    tools = []

    if module_name == "knowledge_tools":
        # 手动导入并创建知识库工具
        try:
            from langchain.tools import tool

            # 检查模块中是否有工具函数
            if hasattr(module, "get_knowledge_list"):
                func = getattr(module, "get_knowledge_list")
                if callable(func):
                    # 使用@tool装饰器重新创建工具
                    tool_instance = tool(func)
                    tools.append(tool_instance)
                    print(f"  手动创建工具: get_knowledge_list")

            if hasattr(module, "get_knowledge_content"):
                func = getattr(module, "get_knowledge_content")
                if callable(func):
                    tool_instance = tool(func)
                    tools.append(tool_instance)
                    print(f"  手动创建工具: get_knowledge_content")

        except Exception as e:
            print(f"  手动创建知识库工具失败: {e}")

    elif module_name == "sale_tools":
        # 手动创建销售工具
        try:
            from langchain.tools import tool

            if hasattr(module, "get_sale_amt"):
                func = getattr(module, "get_sale_amt")
                if callable(func):
                    tool_instance = tool(func)
                    tools.append(tool_instance)
                    print(f"  手动创建工具: get_sale_amt")

        except Exception as e:
            print(f"  手动创建销售工具失败: {e}")

    return tools


def is_tool_instance(obj) -> bool:
    """检查是否是工具实例"""
    if not callable(obj):
        return False

    # 检查是否是BaseTool实例
    if isinstance(obj, BaseTool):
        return True

    # 检查是否有工具的标准属性
    tool_attrs = ["name", "description"]
    if all(hasattr(obj, attr) for attr in tool_attrs):
        return True

    return False


def is_tool_function(obj) -> bool:
    """检查是否是工具函数"""
    if not callable(obj):
        return False

    # 检查是否被@tool装饰
    if hasattr(obj, "_is_tool") and getattr(obj, "_is_tool", False):
        return True

    # 检查是否有tool属性
    if hasattr(obj, "tool"):
        return True

    # 检查是否是函数且具有工具属性
    if callable(obj) and hasattr(obj, "name") and hasattr(obj, "description"):
        return True

    return False


# 测试函数
def test_tool_detection():
    """测试工具检测"""
    print("测试工具检测...")

    # 导入一个模块测试
    import tools.knowledge_tools as kt

    print(f"模块: {kt}")

    for attr_name in dir(kt):
        if not attr_name.startswith("_"):
            attr = getattr(kt, attr_name)
            print(f"\n检查 {attr_name}:")
            print(f"  类型: {type(attr)}")
            print(f"  可调用: {callable(attr)}")

            if callable(attr):
                print(f"  是否有name属性: {hasattr(attr, 'name')}")
                if hasattr(attr, "name"):
                    print(f"  name值: {getattr(attr, 'name', '无')}")

                print(f"  是否有description属性: {hasattr(attr, 'description')}")
                if hasattr(attr, "description"):
                    print(
                        f"  description前50字: {getattr(attr, 'description', '')[:50]}"
                    )

                print(f"  是否被@tool装饰: {is_tool_function(attr)}")
                print(f"  是否是工具实例: {is_tool_instance(attr)}")