danvics/haiku.rag
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement upsert and delete in monitor

Browse source
This commit is contained in:
Yiorgis Gozadinos 2025-06-20 09:28:44 +02:00
parent ee57bc0a82
commit 51e7f990a0
No known key found for this signature in database
2 changed files with 39 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
src/haiku/rag/app.py
View file

@ -94,13 +94,14 @@ class HaikuRAGApp:
async def serve(self, transport: str | None = None): async def serve(self, transport: str | None = None):
"""Start the MCP server.""" """Start the MCP server."""
monitor = FileWatcher(paths=Config.MONITOR_DIRECTORIES) async with HaikuRAG(self.db_path) as client:
asyncio.create_task(monitor.observe()) monitor = FileWatcher(paths=Config.MONITOR_DIRECTORIES, client=client)
server = create_mcp_server(self.db_path) asyncio.create_task(monitor.observe())
server = create_mcp_server(self.db_path)
if transport == "stdio": if transport == "stdio":
await server.run_stdio_async() await server.run_stdio_async()
elif transport == "sse": elif transport == "sse":
await server.run_sse_async("sse") await server.run_sse_async("sse")
else: else:
await server.run_http_async("streamable-http") await server.run_http_async("streamable-http")

36
src/haiku/rag/monitor.py
View file

@ -2,6 +2,7 @@ from pathlib import Path
from watchfiles import Change, DefaultFilter, awatch from watchfiles import Change, DefaultFilter, awatch
from haiku.rag.client import HaikuRAG
from haiku.rag.logging import get_logger from haiku.rag.logging import get_logger
from haiku.rag.reader import FileReader from haiku.rag.reader import FileReader
from haiku.rag.store.models.document import Document from haiku.rag.store.models.document import Document
@ -19,8 +20,9 @@ class FileFilter(DefaultFilter):
class FileWatcher: class FileWatcher:
def __init__(self, paths: list[Path]): def __init__(self, paths: list[Path], client: HaikuRAG):
self.paths = paths self.paths = paths
self.client = client
async def observe(self): async def observe(self):
logger.info(f"Watching files in {self.paths}") logger.info(f"Watching files in {self.paths}")
@ -43,10 +45,30 @@ class FileWatcher:
if f.is_file() and f.suffix in FileReader.extensions: if f.is_file() and f.suffix in FileReader.extensions:
await self._upsert_document(f) await self._upsert_document(f)
async def _delete_document(self, file: Path):
logger.info(f"Deleting document from {file}")
pass
async def _upsert_document(self, file: Path) -> Document | None: async def _upsert_document(self, file: Path) -> Document | None:
logger.info(f"Updating document from {file}") try:
pass uri = file.as_uri()
existing_doc = await self.client.get_document_by_uri(uri)
print(uri)
if existing_doc:
doc = await self.client.create_document_from_source(str(file))
logger.info(f"Updated document {existing_doc.id} from {file}")
return doc
else:
doc = await self.client.create_document_from_source(str(file))
logger.info(f"Created new document {doc.id} from {file}")
return doc
except Exception as e:
logger.error(f"Failed to upsert document from {file}: {e}")
return None
async def _delete_document(self, file: Path):
try:
uri = file.as_uri()
existing_doc = await self.client.get_document_by_uri(uri)
if existing_doc and existing_doc.id:
await self.client.delete_document(existing_doc.id)
logger.info(f"Deleted document {existing_doc.id} for {file}")
except Exception as e:
logger.error(f"Failed to delete document for {file}: {e}")