Coverage for agentos/tools/search_tools.py: 22%

1"""搜索工具 — 文件内容搜索、文件名匹配、代码符号搜索。""" 

2 

3from __future__ import annotations 

4 

5import fnmatch 

6import os 

7import re 

8 

9from agentos.tools.base import BaseTool, ToolResult 

10 

11 

12class GrepTool(BaseTool): 

13 """文件内容搜索工具 — 在目录中递归搜索匹配文本。""" 

14 

15 name = "grep" 

16 description = "在目录中递归搜索文件内容，支持正则表达式，返回匹配路径和行号" 

17 

18 @property 

19 def parameters(self) -> dict: 

20 return { 

21 "type": "object", 

22 "properties": { 

23 "pattern": {"type": "string", "description": "搜索的文本或正则表达式"}, 

24 "directory": {"type": "string", "description": "搜索目录，默认当前目录"}, 

25 "file_pattern": {"type": "string", "description": "文件名匹配模式，如 *.py"}, 

26 "max_results": {"type": "integer", "description": "最大结果数，默认 50"}, 

27 "case_sensitive": {"type": "boolean", "description": "是否区分大小写，默认 true"}, 

28 }, 

29 "required": ["pattern"], 

30 } 

31 

32 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

33 pattern = arguments.get("pattern", "") 

34 directory = arguments.get("directory", ".") 

35 file_pattern = arguments.get("file_pattern", "*") 

36 max_results = arguments.get("max_results", 50) 

37 case_sensitive = arguments.get("case_sensitive", True) 

38 

39 flags = 0 if case_sensitive else re.IGNORECASE 

40 try: 

41 regex = re.compile(pattern, flags) 

42 except re.error as e: 

43 return ToolResult.fail(call_id="", error=f"Invalid regex: {e}") 

44 

45 results = [] 

46 for root, dirs, files in os.walk(os.path.abspath(directory)): 

47 dirs[:] = [ 

48 d 

49 for d in dirs 

50 if not d.startswith(".") 

51 and d not in ("node_modules", "__pycache__", "dist", "build", ".git") 

52 ] 

53 for filename in files: 

54 if not fnmatch.fnmatch(filename, file_pattern): 

55 continue 

56 filepath = os.path.join(root, filename) 

57 try: 

58 with open(filepath, encoding="utf-8", errors="ignore") as f: 

59 for lineno, line in enumerate(f, 1): 

60 if regex.search(line): 

61 results.append(f"{filepath}:{lineno}: {line.strip()[:200]}") 

62 if len(results) >= max_results: 

63 return ToolResult.ok(call_id="", output="\n".join(results)) 

64 except (PermissionError, IsADirectoryError, UnicodeDecodeError): 

65 continue 

66 

67 return ToolResult.ok( 

68 call_id="", output="\n".join(results) if results else "No matches found" 

69 ) 

70 

71 

72class FileSearchTool(BaseTool): 

73 """文件搜索工具 — 按文件名模式搜索。""" 

74 

75 name = "file_search" 

76 description = "按文件名模式搜索文件，支持 glob 通配符，返回匹配的文件路径列表" 

77 

78 @property 

79 def parameters(self) -> dict: 

80 return { 

81 "type": "object", 

82 "properties": { 

83 "pattern": { 

84 "type": "string", 

85 "description": "文件名匹配模式，如 *.py, report*.pdf", 

86 }, 

87 "directory": {"type": "string", "description": "搜索目录，默认当前目录"}, 

88 "max_results": {"type": "integer", "description": "最大结果数，默认 100"}, 

89 }, 

90 "required": ["pattern"], 

91 } 

92 

93 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

94 pattern = arguments.get("pattern", "") 

95 directory = arguments.get("directory", ".") 

96 max_results = arguments.get("max_results", 100) 

97 

98 results = [] 

99 for root, dirs, files in os.walk(os.path.abspath(directory)): 

100 dirs[:] = [ 

101 d 

102 for d in dirs 

103 if not d.startswith(".") 

104 and d not in ("node_modules", "__pycache__", "dist", "build", ".git") 

105 ] 

106 for filename in files: 

107 if fnmatch.fnmatch(filename, pattern): 

108 results.append(os.path.join(root, filename)) 

109 if len(results) >= max_results: 

110 return ToolResult.ok(call_id="", output="\n".join(results)) 

111 

112 return ToolResult.ok(call_id="", output="\n".join(results) if results else "No files found") 

113 

114 

115class CodeSearchTool(BaseTool): 

116 """代码符号搜索工具 — 搜索函数/类/导入定义（基于 AST）。""" 

117 

118 name = "code_search" 

119 description = "在 Python 代码中搜索函数定义、类定义、导入等符号，返回符号名和位置" 

120 

121 @property 

122 def parameters(self) -> dict: 

123 return { 

124 "type": "object", 

125 "properties": { 

126 "query": {"type": "string", "description": "搜索的函数名或类名"}, 

127 "directory": {"type": "string", "description": "代码目录，默认当前目录"}, 

128 "symbol_type": { 

129 "type": "string", 

130 "description": "符号类型：function/class/import/all，默认 all", 

131 }, 

132 "max_results": {"type": "integer", "description": "最大结果数，默认 30"}, 

133 }, 

134 "required": ["query"], 

135 } 

136 

137 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

138 import ast 

139 

140 query = arguments.get("query", "") 

141 directory = arguments.get("directory", ".") 

142 symbol_type = arguments.get("symbol_type", "all") 

143 max_results = arguments.get("max_results", 30) 

144 

145 results = [] 

146 for root, dirs, files in os.walk(os.path.abspath(directory)): 

147 dirs[:] = [ 

148 d 

149 for d in dirs 

150 if not d.startswith(".") 

151 and d not in ("node_modules", "__pycache__", "dist", "build", ".git") 

152 ] 

153 for filename in files: 

154 if not filename.endswith(".py"): 

155 continue 

156 filepath = os.path.join(root, filename) 

157 try: 

158 with open(filepath, encoding="utf-8", errors="ignore") as f: 

159 source = f.read() 

160 tree = ast.parse(source, filename=filepath) 

161 for node in ast.walk(tree): 

162 if len(results) >= max_results: 

163 break 

164 name = None 

165 stype = None 

166 if isinstance(node, ast.FunctionDef) and symbol_type in ("function", "all"): 

167 name, stype = node.name, "function" 

168 elif isinstance(node, ast.AsyncFunctionDef) and symbol_type in ( 

169 "function", 

170 "all", 

171 ): 

172 name, stype = node.name, "async_function" 

173 elif isinstance(node, ast.ClassDef) and symbol_type in ("class", "all"): 

174 name, stype = node.name, "class" 

175 elif isinstance(node, ast.Import) and symbol_type in ("import", "all"): 

176 for alias in node.names: 

177 if query.lower() in alias.name.lower(): 

178 results.append(f"{filepath}:{node.lineno}: import {alias.name}") 

179 elif isinstance(node, ast.ImportFrom) and symbol_type in ("import", "all"): 

180 if query.lower() in (node.module or "").lower(): 

181 results.append( 

182 f"{filepath}:{node.lineno}: from {node.module} import ..." 

183 ) 

184 

185 if name and stype and query.lower() in name.lower(): 

186 results.append(f"{filepath}:{node.lineno}: [{stype}] {name}") 

187 except (SyntaxError, UnicodeDecodeError, PermissionError): 

188 continue 

189 

190 return ToolResult.ok( 

191 call_id="", output="\n".join(results) if results else "No symbols found" 

192 )