Coverage for agentos/tools/search_tools.py: 25%
104 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 11:37 +0800
1"""搜索工具 — 文件内容搜索、文件名匹配、代码符号搜索。"""
3from __future__ import annotations
5import fnmatch
6import os
7import re
9from agentos.tools.base import BaseTool, ToolResult
12class GrepTool(BaseTool):
13 """文件内容搜索工具 — 在目录中递归搜索匹配文本。"""
15 name = "grep"
16 description = "在目录中递归搜索文件内容，支持正则表达式，返回匹配路径和行号"
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 }
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)
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}")
45 results = []
46 for root, dirs, files in os.walk(os.path.abspath(directory)):
47 dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("node_modules", "__pycache__", "dist", "build", ".git")]
48 for filename in files:
49 if not fnmatch.fnmatch(filename, file_pattern):
50 continue
51 filepath = os.path.join(root, filename)
52 try:
53 with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
54 for lineno, line in enumerate(f, 1):
55 if regex.search(line):
56 results.append(f"{filepath}:{lineno}: {line.strip()[:200]}")
57 if len(results) >= max_results:
58 return ToolResult.ok(call_id="", output="\n".join(results))
59 except (PermissionError, IsADirectoryError, UnicodeDecodeError):
60 continue
62 return ToolResult.ok(call_id="", output="\n".join(results) if results else "No matches found")
65class FileSearchTool(BaseTool):
66 """文件搜索工具 — 按文件名模式搜索。"""
68 name = "file_search"
69 description = "按文件名模式搜索文件，支持 glob 通配符，返回匹配的文件路径列表"
71 @property
72 def parameters(self) -> dict:
73 return {
74 "type": "object",
75 "properties": {
76 "pattern": {"type": "string", "description": "文件名匹配模式，如 *.py, report*.pdf"},
77 "directory": {"type": "string", "description": "搜索目录，默认当前目录"},
78 "max_results": {"type": "integer", "description": "最大结果数，默认 100"},
79 },
80 "required": ["pattern"],
81 }
83 async def execute(self, arguments: dict, sandbox=None) -> ToolResult:
84 pattern = arguments.get("pattern", "")
85 directory = arguments.get("directory", ".")
86 max_results = arguments.get("max_results", 100)
88 results = []
89 for root, dirs, files in os.walk(os.path.abspath(directory)):
90 dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("node_modules", "__pycache__", "dist", "build", ".git")]
91 for filename in files:
92 if fnmatch.fnmatch(filename, pattern):
93 results.append(os.path.join(root, filename))
94 if len(results) >= max_results:
95 return ToolResult.ok(call_id="", output="\n".join(results))
97 return ToolResult.ok(call_id="", output="\n".join(results) if results else "No files found")
100class CodeSearchTool(BaseTool):
101 """代码符号搜索工具 — 搜索函数/类/导入定义（基于 AST）。"""
103 name = "code_search"
104 description = "在 Python 代码中搜索函数定义、类定义、导入等符号，返回符号名和位置"
106 @property
107 def parameters(self) -> dict:
108 return {
109 "type": "object",
110 "properties": {
111 "query": {"type": "string", "description": "搜索的函数名或类名"},
112 "directory": {"type": "string", "description": "代码目录，默认当前目录"},
113 "symbol_type": {"type": "string", "description": "符号类型：function/class/import/all，默认 all"},
114 "max_results": {"type": "integer", "description": "最大结果数，默认 30"},
115 },
116 "required": ["query"],
117 }
119 async def execute(self, arguments: dict, sandbox=None) -> ToolResult:
120 import ast
122 query = arguments.get("query", "")
123 directory = arguments.get("directory", ".")
124 symbol_type = arguments.get("symbol_type", "all")
125 max_results = arguments.get("max_results", 30)
127 results = []
128 for root, dirs, files in os.walk(os.path.abspath(directory)):
129 dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("node_modules", "__pycache__", "dist", "build", ".git")]
130 for filename in files:
131 if not filename.endswith(".py"):
132 continue
133 filepath = os.path.join(root, filename)
134 try:
135 with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
136 source = f.read()
137 tree = ast.parse(source, filename=filepath)
138 for node in ast.walk(tree):
139 if len(results) >= max_results:
140 break
141 name = None
142 stype = None
143 if isinstance(node, ast.FunctionDef) and symbol_type in ("function", "all"):
144 name, stype = node.name, "function"
145 elif isinstance(node, ast.AsyncFunctionDef) and symbol_type in ("function", "all"):
146 name, stype = node.name, "async_function"
147 elif isinstance(node, ast.ClassDef) and symbol_type in ("class", "all"):
148 name, stype = node.name, "class"
149 elif isinstance(node, ast.Import) and symbol_type in ("import", "all"):
150 for alias in node.names:
151 if query.lower() in alias.name.lower():
152 results.append(f"{filepath}:{node.lineno}: import {alias.name}")
153 elif isinstance(node, ast.ImportFrom) and symbol_type in ("import", "all"):
154 if query.lower() in (node.module or "").lower():
155 results.append(f"{filepath}:{node.lineno}: from {node.module} import ...")
157 if name and stype and query.lower() in name.lower():
158 results.append(f"{filepath}:{node.lineno}: [{stype}] {name}")
159 except (SyntaxError, UnicodeDecodeError, PermissionError):
160 continue
162 return ToolResult.ok(call_id="", output="\n".join(results) if results else "No symbols found")