Debounce optimize() to avoid exceeding too-many-files-open
This commit is contained in:
parent
27b16bdf8d
commit
e08fce84f8
3 changed files with 53 additions and 9 deletions
|
|
@ -6,7 +6,6 @@ from uuid import uuid4
|
||||||
import lancedb
|
import lancedb
|
||||||
from lancedb.pydantic import LanceModel, Vector
|
from lancedb.pydantic import LanceModel, Vector
|
||||||
from pydantic import Field
|
from pydantic import Field
|
||||||
from rich.console import Console
|
|
||||||
|
|
||||||
from haiku.rag.config import Config
|
from haiku.rag.config import Config
|
||||||
from haiku.rag.embeddings import get_embedder
|
from haiku.rag.embeddings import get_embedder
|
||||||
|
|
@ -105,12 +104,9 @@ class Store:
|
||||||
self.settings_table.search().limit(1).to_pydantic(SettingsRecord)
|
self.settings_table.search().limit(1).to_pydantic(SettingsRecord)
|
||||||
)
|
)
|
||||||
if existing_settings:
|
if existing_settings:
|
||||||
console = Console()
|
db_version = self.get_haiku_version() # noqa: F841
|
||||||
db_version = self.get_haiku_version()
|
# XXX Add upgrade logic here similar to SQLite version
|
||||||
# Future: Add upgrade logic here similar to SQLite version
|
|
||||||
console.print(
|
|
||||||
f"[green]LanceDB store initialized (version: {db_version})[/green]"
|
|
||||||
)
|
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,7 @@ from haiku.rag.chunker import chunker
|
||||||
from haiku.rag.embeddings import get_embedder
|
from haiku.rag.embeddings import get_embedder
|
||||||
from haiku.rag.store.engine import DocumentRecord, Store
|
from haiku.rag.store.engine import DocumentRecord, Store
|
||||||
from haiku.rag.store.models.chunk import Chunk
|
from haiku.rag.store.models.chunk import Chunk
|
||||||
|
from haiku.rag.utils import debounce
|
||||||
|
|
||||||
|
|
||||||
class ChunkRepository:
|
class ChunkRepository:
|
||||||
|
|
@ -24,6 +25,14 @@ class ChunkRepository:
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@debounce(1.0)
|
||||||
|
async def _optimize(self) -> None:
|
||||||
|
"""Optimize the chunks table to refresh indexes."""
|
||||||
|
try:
|
||||||
|
self.store.chunks_table.optimize()
|
||||||
|
except RuntimeError:
|
||||||
|
pass
|
||||||
|
|
||||||
async def create(self, entity: Chunk) -> Chunk:
|
async def create(self, entity: Chunk) -> Chunk:
|
||||||
"""Create a chunk in the database."""
|
"""Create a chunk in the database."""
|
||||||
assert entity.document_id, "Chunk must have a document_id to be created"
|
assert entity.document_id, "Chunk must have a document_id to be created"
|
||||||
|
|
@ -49,7 +58,7 @@ class ChunkRepository:
|
||||||
entity.id = chunk_id
|
entity.id = chunk_id
|
||||||
|
|
||||||
# Optimize table after insert to update indexes
|
# Optimize table after insert to update indexes
|
||||||
self.store.chunks_table.optimize()
|
await self._optimize()
|
||||||
return entity
|
return entity
|
||||||
|
|
||||||
async def get_by_id(self, entity_id: str) -> Chunk | None:
|
async def get_by_id(self, entity_id: str) -> Chunk | None:
|
||||||
|
|
@ -88,7 +97,7 @@ class ChunkRepository:
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
# Optimize table after update to refresh indexes
|
# Optimize table after update to refresh indexes
|
||||||
self.store.chunks_table.optimize()
|
await self._optimize()
|
||||||
|
|
||||||
return entity
|
return entity
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,7 @@
|
||||||
|
import asyncio
|
||||||
import sys
|
import sys
|
||||||
|
from collections.abc import Callable
|
||||||
|
from functools import wraps
|
||||||
from importlib import metadata
|
from importlib import metadata
|
||||||
from io import BytesIO
|
from io import BytesIO
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
@ -10,6 +13,42 @@ from docling_core.types.io import DocumentStream
|
||||||
from packaging.version import Version, parse
|
from packaging.version import Version, parse
|
||||||
|
|
||||||
|
|
||||||
|
def debounce(wait: float) -> Callable:
|
||||||
|
"""
|
||||||
|
A decorator to debounce a function, ensuring it is called only after a specified delay
|
||||||
|
and always executes after the last call.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
wait (float): The debounce delay in seconds.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Callable: The decorated function.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def decorator(func: Callable) -> Callable:
|
||||||
|
last_call = None
|
||||||
|
task = None
|
||||||
|
|
||||||
|
@wraps(func)
|
||||||
|
async def debounced(*args, **kwargs):
|
||||||
|
nonlocal last_call, task
|
||||||
|
last_call = asyncio.get_event_loop().time()
|
||||||
|
|
||||||
|
if task:
|
||||||
|
task.cancel()
|
||||||
|
|
||||||
|
async def call_func():
|
||||||
|
await asyncio.sleep(wait)
|
||||||
|
if asyncio.get_event_loop().time() - last_call >= wait: # type: ignore
|
||||||
|
await func(*args, **kwargs)
|
||||||
|
|
||||||
|
task = asyncio.create_task(call_func())
|
||||||
|
|
||||||
|
return debounced
|
||||||
|
|
||||||
|
return decorator
|
||||||
|
|
||||||
|
|
||||||
def get_default_data_dir() -> Path:
|
def get_default_data_dir() -> Path:
|
||||||
"""Get the user data directory for the current system platform.
|
"""Get the user data directory for the current system platform.
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue