tool_factory.py 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  1. import os
  2. import sys
  3. import importlib
  4. import inspect
  5. from pathlib import Path
  6. from typing import List
  7. from langchain.tools import BaseTool
  8. def get_all_tools() -> List[BaseTool]:
  9. """
  10. 自动发现并返回所有工具 - 修复检查逻辑
  11. """
  12. tools = []
  13. print("🛠️ 开始自动发现工具...")
  14. # 获取项目根目录
  15. project_root = Path(__file__).parent.parent
  16. tools_dir = Path(__file__).parent
  17. # 将项目根目录添加到Python路径
  18. if str(project_root) not in sys.path:
  19. sys.path.insert(0, str(project_root))
  20. # 扫描工具文件
  21. tool_files = []
  22. for file_path in tools_dir.glob("*_tools.py"):
  23. if file_path.is_file():
  24. module_name = file_path.stem
  25. tool_files.append(module_name)
  26. print(f"📦 发现工具文件: {module_name}")
  27. if not tool_files:
  28. tool_files = ["knowledge_tools", "sale_tools"]
  29. print("⚠️ 使用默认工具列表")
  30. for module_name in tool_files:
  31. try:
  32. # 导入模块
  33. full_module_path = f"tools.{module_name}"
  34. module = importlib.import_module(full_module_path)
  35. # print(f"✅ 加载模块: {module_name}")
  36. # 查找工具 - 使用更全面的方法
  37. tool_count = 0
  38. # 方法1: 检查模块的所有属性
  39. for attr_name in dir(module):
  40. if attr_name.startswith("_"):
  41. continue
  42. attr = getattr(module, attr_name)
  43. # # 详细调试信息
  44. # print(f" 🔍 检查 {attr_name}:")
  45. # print(f" 类型: {type(attr)}")
  46. # 检查是否是BaseTool实例
  47. if isinstance(attr, BaseTool):
  48. tools.append(attr)
  49. tool_count += 1
  50. # print(f" ✅ 发现BaseTool工具: {getattr(attr, 'name', attr_name)}")
  51. continue
  52. # 检查是否是函数且具有工具属性
  53. if callable(attr):
  54. # print(f" 是否有name属性: {hasattr(attr, 'name')}")
  55. # if hasattr(attr, "name"):
  56. # print(f" name值: {getattr(attr, 'name', '无')}")
  57. # print(f" 是否有description属性: {hasattr(attr, 'description')}")
  58. # if hasattr(attr, "description"):
  59. # print(
  60. # f" description前50字: {getattr(attr, 'description', '')[:50]}"
  61. # )
  62. # 检查是否是工具函数
  63. if is_tool_function(attr):
  64. tools.append(attr)
  65. tool_count += 1
  66. # print(f" ✅ 发现工具函数: {attr_name}")
  67. # 方法2: 检查模块的全局变量
  68. print(f" 🔍 检查模块全局变量...")
  69. for name, value in module.__dict__.items():
  70. if name.startswith("_"):
  71. continue
  72. if isinstance(value, BaseTool):
  73. if value not in tools: # 避免重复添加
  74. tools.append(value)
  75. tool_count += 1
  76. # print(
  77. # f" ✅ 从全局变量发现BaseTool工具: {getattr(value, 'name', name)}"
  78. # )
  79. if tool_count == 0:
  80. # print(f" ⚠️ 模块 {module_name} 中未发现工具")
  81. # 尝试手动创建工具
  82. manual_tools = create_tools_manually(module_name, module)
  83. if manual_tools:
  84. tools.extend(manual_tools)
  85. tool_count = len(manual_tools)
  86. # print(f" 🔧 手动创建了 {tool_count} 个工具")
  87. else:
  88. print(f" 📊 模块 {module_name} 中发现 {tool_count} 个工具")
  89. except Exception as e:
  90. print(f"❌ 加载模块 {module_name} 失败: {e}")
  91. print(f"🎯 总共发现 {len(tools)} 个工具")
  92. # # 打印工具详情
  93. # for i, tool in enumerate(tools):
  94. # tool_name = getattr(tool, "name", f"tool_{i+1}")
  95. # tool_desc = getattr(tool, "description", "无描述")
  96. # print(f" {i+1}. {tool_name}: {tool_desc[:50]}...")
  97. return tools
  98. def create_tools_manually(module_name: str, module) -> List[BaseTool]:
  99. """手动创建工具 - 针对@tool装饰器的问题"""
  100. tools = []
  101. if module_name == "knowledge_tools":
  102. # 手动导入并创建知识库工具
  103. try:
  104. from langchain.tools import tool
  105. # 检查模块中是否有工具函数
  106. if hasattr(module, "get_knowledge_list"):
  107. func = getattr(module, "get_knowledge_list")
  108. if callable(func):
  109. # 使用@tool装饰器重新创建工具
  110. tool_instance = tool(func)
  111. tools.append(tool_instance)
  112. print(f" 🔧 手动创建工具: get_knowledge_list")
  113. if hasattr(module, "get_knowledge_content"):
  114. func = getattr(module, "get_knowledge_content")
  115. if callable(func):
  116. tool_instance = tool(func)
  117. tools.append(tool_instance)
  118. print(f" 🔧 手动创建工具: get_knowledge_content")
  119. except Exception as e:
  120. print(f" ❌ 手动创建知识库工具失败: {e}")
  121. elif module_name == "sale_tools":
  122. # 手动创建销售工具
  123. try:
  124. from langchain.tools import tool
  125. if hasattr(module, "get_sale_amt"):
  126. func = getattr(module, "get_sale_amt")
  127. if callable(func):
  128. tool_instance = tool(func)
  129. tools.append(tool_instance)
  130. print(f" 🔧 手动创建工具: get_sale_amt")
  131. except Exception as e:
  132. print(f" ❌ 手动创建销售工具失败: {e}")
  133. return tools
  134. def is_tool_instance(obj) -> bool:
  135. """检查是否是工具实例"""
  136. if not callable(obj):
  137. return False
  138. # 检查是否是BaseTool实例
  139. if isinstance(obj, BaseTool):
  140. return True
  141. # 检查是否有工具的标准属性
  142. tool_attrs = ["name", "description"]
  143. if all(hasattr(obj, attr) for attr in tool_attrs):
  144. return True
  145. return False
  146. def is_tool_function(obj) -> bool:
  147. """检查是否是工具函数"""
  148. if not callable(obj):
  149. return False
  150. # 检查是否被@tool装饰
  151. if hasattr(obj, "_is_tool") and getattr(obj, "_is_tool", False):
  152. return True
  153. # 检查是否有tool属性
  154. if hasattr(obj, "tool"):
  155. return True
  156. # 检查是否是函数且具有工具属性
  157. if callable(obj) and hasattr(obj, "name") and hasattr(obj, "description"):
  158. return True
  159. return False
  160. # 测试函数
  161. def test_tool_detection():
  162. """测试工具检测"""
  163. print("🧪 测试工具检测...")
  164. # 导入一个模块测试
  165. import tools.knowledge_tools as kt
  166. print(f"模块: {kt}")
  167. for attr_name in dir(kt):
  168. if not attr_name.startswith("_"):
  169. attr = getattr(kt, attr_name)
  170. print(f"\n🔍 检查 {attr_name}:")
  171. print(f" 类型: {type(attr)}")
  172. print(f" 可调用: {callable(attr)}")
  173. if callable(attr):
  174. print(f" 是否有name属性: {hasattr(attr, 'name')}")
  175. if hasattr(attr, "name"):
  176. print(f" name值: {getattr(attr, 'name', '无')}")
  177. print(f" 是否有description属性: {hasattr(attr, 'description')}")
  178. if hasattr(attr, "description"):
  179. print(
  180. f" description前50字: {getattr(attr, 'description', '')[:50]}"
  181. )
  182. print(f" 是否被@tool装饰: {is_tool_function(attr)}")
  183. print(f" 是否是工具实例: {is_tool_instance(attr)}")