apunkt/MetaGPT
1
0
Fork 0
mirror of https://github.com/FoundationAgents/MetaGPT.git synced 2026-07-05 16:02:14 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: search_index_repo path or filename

Browse source
This commit is contained in:
莘权 马 2024-09-06 15:35:47 +08:00
parent 95bf4c3e22
commit 6a57cb5e0a
3 changed files with 24 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

10
metagpt/tools/libs/editor.py
View file

@ -18,6 +18,7 @@ from metagpt.logs import logger
from metagpt.tools.libs.index_repo import OTHER_TYPE, IndexRepo
from metagpt.tools.libs.linter import Linter
from metagpt.tools.tool_registry import register_tool
from metagpt.utils.common import list_files
from metagpt.utils.file import File
from metagpt.utils.report import EditorReporter
@ -866,7 +867,7 @@ class Editor(BaseModel):
return path
@staticmethod
async def search_index_repo(query: str, files_or_paths: List[Union[str, Path]]) -> List[str]:
async def search_index_repo(query: str, file_or_path: Union[str, Path]) -> List[str]:
"""Searches the index repository for a given query across specified files or paths.
This method classifies the provided files or paths, performing a search on each cluster
@ -875,13 +876,16 @@ class Editor(BaseModel):
Args:
query (str): The search query string to look for in the indexed files.
files_or_paths (List[Union[str, Path]]): A list of file paths or names to search within.
file_or_path (Union[str, Path]): A path or a filename to search within.
Returns:
List[str]: A list of search results as strings, containing the text from the merged results
and any direct results from other files.
"""
clusters, roots = IndexRepo.classify_path(files_or_paths)
if not file_or_path or not Path(file_or_path).exists():
raise ValueError(f'"{str(file_or_path)}" not exists')
files = [file_or_path] if not Path(file_or_path).is_dir() else list_files(file_or_path)
clusters, roots = IndexRepo.classify_path(files)
futures = []
others = set()
for persist_path, filenames in clusters.items():

12
metagpt/tools/libs/index_repo.py
View file

@ -85,7 +85,9 @@ class IndexRepo(BaseModel):
"""
encoding = tiktoken.get_encoding("cl100k_base")
result: List[Union[NodeWithScore, TextScore]] = []
filenames, _ = await self._filter(filenames)
filenames, excludes = await self._filter(filenames)
if not filenames:
raise ValueError(f"Unsupported file types: {[str(i) for i in excludes]}")
filter_filenames = set()
meta = await self._read_meta()
for i in filenames:
@ -269,7 +271,7 @@ class IndexRepo(BaseModel):
List[NodeWithScore]: A list of nodes with scores matching the query.
"""
if not Path(self.persist_path).exists():
return []
raise ValueError(f"IndexRepo {Path(self.persist_path).name} not exists.")
engine = SimpleEngine.from_index(
index_config=FAISSIndexConfig(persist_path=self.persist_path), retriever_configs=[FAISSRetrieverConfig()]
)
@ -293,11 +295,11 @@ class IndexRepo(BaseModel):
return old_fp != fp
@staticmethod
def classify_path(files_or_paths: List[Union[str, Path]]) -> Tuple[Dict[str, Set[Path]], Dict[str, str]]:
def classify_path(files: List[Union[str, Path]]) -> Tuple[Dict[str, Set[Path]], Dict[str, str]]:
"""Classify a list of file paths or Path objects into different categories.
Args:
files_or_paths (List[Union[str, Path]]): A list of file paths or Path objects to be classified.
files (List[Union[str, Path]]): A list of file paths or Path objects to be classified.
Returns:
Tuple[Dict[str, Set[Path]], Dict[str, str]]:
@ -311,7 +313,7 @@ class IndexRepo(BaseModel):
clusters = {}
roots = {}
for i in files_or_paths:
for i in files:
path = Path(i).absolute()
path_type = OTHER_TYPE
for type_, pattern in mappings.items():