File size: 2,182 Bytes
7e2f74d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | from tools.base_tool import BaseTool
from memory.vector_store import VectorStore
from typing import Any
import logging
logger = logging.getLogger("file_reader_tool")
class FileReaderTool(BaseTool):
def __init__(self, vector_store: VectorStore):
self.store = vector_store
@property
def name(self) -> str:
return "file_reader"
@property
def description(self) -> str:
return "Reads the content of a specific source code file by retrieving its chunks. Inputs: repo_id (str), path (str)."
def execute(self, **kwargs) -> Any:
repo_id = kwargs.get("repo_id")
path = kwargs.get("path")
if not repo_id or not path:
return {"error": "Missing required parameters: repo_id and path."}
try:
collection = self.store.get_collection(repo_id)
results = collection.get(where={"path": path})
if not results or not results.get("documents"):
return {"error": f"File '{path}' not found in knowledge index."}
docs = results["documents"]
metas = results["metadatas"]
# Reconstruct sorting by chunk index
chunks = []
for doc, meta in zip(docs, metas):
idx = meta.get("chunk_index", 0)
chunks.append((idx, doc))
chunks.sort(key=lambda x: x[0])
# Rebuild file and remove the file header prefix
content_builder = []
for idx, content in chunks:
if "\n\n" in content and content.startswith("File: "):
parts = content.split("\n\n", 1)
content_builder.append(parts[1])
else:
content_builder.append(content)
return {
"path": path,
"content": "".join(content_builder),
"chunks_found": len(chunks)
}
except Exception as e:
logger.error(f"Error executing file_reader tool: {e}")
return {"error": f"Failed to retrieve file contents: {str(e)}"}
|