MeiliSearch as a pure document store, without its
(experimental) vector-store functionality.
We aim to use MeiliSearch for fast lexical search.
Note that what we call "Collection" in Langroid is referred to as
"Index" in MeiliSearch. Each data-store has its own terminology,
but for uniformity we use the Langroid terminology here.
MeiliSearch(config=MeiliSearchConfig())
Bases: VectorStore
Source code in langroid/vector_store/meilisearch.py
| def __init__(self, config: MeiliSearchConfig = MeiliSearchConfig()):
super().__init__(config)
self.config: MeiliSearchConfig = config
self.host = config.host
self.port = config.port
load_dotenv()
self.key = os.getenv("MEILISEARCH_API_KEY") or "masterKey"
self.url = os.getenv("MEILISEARCH_API_URL") or f"http://{self.host}:{self.port}"
if config.cloud and None in [self.key, self.url]:
logger.warning(
f"""MEILISEARCH_API_KEY, MEILISEARCH_API_URL env variable must be set
to use MeiliSearch in cloud mode. Please set these values
in your .env file. Switching to local MeiliSearch at
{self.url}
"""
)
config.cloud = False
self.client: Callable[[], meilisearch.AsyncClient] = lambda: (
meilisearch.AsyncClient(url=self.url, api_key=self.key)
)
# Note: Only create collection if a non-null collection name is provided.
# This is useful to delay creation of db until we have a suitable
# collection name (e.g. we could get it from the url or folder path).
if config.collection_name is not None:
self.create_collection(
config.collection_name, replace=config.replace_collection
)
|
clear_empty_collections()
All collections are treated as non-empty in MeiliSearch, so this is a
no-op
Source code in langroid/vector_store/meilisearch.py
| def clear_empty_collections(self) -> int:
"""All collections are treated as non-empty in MeiliSearch, so this is a
no-op"""
return 0
|
clear_all_collections(really=False, prefix='')
Delete all indices whose names start with prefix
Source code in langroid/vector_store/meilisearch.py
| def clear_all_collections(self, really: bool = False, prefix: str = "") -> int:
"""Delete all indices whose names start with `prefix`"""
if not really:
logger.warning("Not deleting all collections, set really=True to confirm")
return 0
coll_names = [c for c in self.list_collections() if c.startswith(prefix)]
deletes = asyncio.run(self._async_delete_indices(coll_names))
n_deletes = sum(deletes)
logger.warning(f"Deleted {n_deletes} indices in MeiliSearch")
return n_deletes
|
list_collections(empty=False)
Returns:
| Type |
Description |
List[str]
|
List of index names stored. We treat any existing index as non-empty.
|
Source code in langroid/vector_store/meilisearch.py
| def list_collections(self, empty: bool = False) -> List[str]:
"""
Returns:
List of index names stored. We treat any existing index as non-empty.
"""
indexes = asyncio.run(self._async_get_indexes())
if len(indexes) == 0:
return []
else:
return [ind.uid for ind in indexes]
|
create_collection(collection_name, replace=False)
Create a collection with the given name, optionally replacing an existing
collection if replace is True.
Args:
collection_name (str): Name of the collection to create.
replace (bool): Whether to replace an existing collection
with the same name. Defaults to False.
Source code in langroid/vector_store/meilisearch.py
| def create_collection(self, collection_name: str, replace: bool = False) -> None:
"""
Create a collection with the given name, optionally replacing an existing
collection if `replace` is True.
Args:
collection_name (str): Name of the collection to create.
replace (bool): Whether to replace an existing collection
with the same name. Defaults to False.
"""
self.config.collection_name = collection_name
collections = self.list_collections()
if collection_name in collections:
logger.warning(
f"MeiliSearch Non-empty Index {collection_name} already exists"
)
if not replace:
logger.warning("Not replacing collection")
return
else:
logger.warning("Recreating fresh collection")
asyncio.run(self._async_delete_index(collection_name))
asyncio.run(self._async_create_index(collection_name))
collection_info = asyncio.run(self._async_get_index(collection_name))
if settings.debug:
level = logger.getEffectiveLevel()
logger.setLevel(logging.INFO)
logger.info(collection_info)
logger.setLevel(level)
|