dosubot[bot]
commented
1 week ago Hey @kobiche! 👋 I'm here to help you with any bugs, questions, or contributions you have. I'm a bot designed to assist you while you wait for a human maintainer.
Yes, the OpensearchVectorClient in LlamaIndex runs exclusively using async code. This is evident from the use of asynchronous methods and the AsyncOpenSearch client throughout the implementation. For example, the _bulk_ingest_embeddings method is defined as an asynchronous function, and the client is initialized using AsyncOpenSearch [1].
To adjust the code in LlamaIndex to offer both sync and async possibilities, you can create synchronous counterparts for the asynchronous methods. Here is an example of how you can modify the OpensearchVectorClient class to include both sync and async methods:
class OpensearchVectorClient:
# Existing async methods...
def _get_sync_opensearch_client(self, opensearch_url: str, **kwargs: Any) -> Any:
"""Get Sync OpenSearch client from the opensearch_url, otherwise raise error."""
try:
from opensearchpy import OpenSearch
client = OpenSearch(opensearch_url, **kwargs)
except ValueError as e:
raise ValueError(
f"OpenSearch client string provided is not in proper format. "
f"Got error: {e} "
)
return client
def _bulk_ingest_embeddings_sync(
self,
client: Any,
index_name: str,
embeddings: List[List[float]],
texts: Iterable[str],
metadatas: Optional[List[dict]] = None,
ids: Optional[List[str]] = None,
vector_field: str = "embedding",
text_field: str = "content",
mapping: Optional[Dict] = None,
max_chunk_bytes: Optional[int] = 1 * 1024 * 1024,
is_aoss: bool = False,
) -> List[str]:
"""Sync Bulk Ingest Embeddings into given index."""
if not mapping:
mapping = {}
from opensearchpy.helpers import bulk
not_found_error = self._import_not_found_error()
requests = []
return_ids = []
mapping = mapping
try:
client.indices.get(index=index_name)
except not_found_error:
client.indices.create(index=index_name, body=mapping)
for i, text in enumerate(texts):
metadata = metadatas[i] if metadatas else {}
_id = ids[i] if ids else str(uuid.uuid4())
request = {
"_op_type": "index",
"_index": index_name,
vector_field: embeddings[i],
text_field: text,
"metadata": metadata,
}
if is_aoss:
request["id"] = _id
else:
request["_id"] = _id
requests.append(request)
return_ids.append(_id)
bulk(client, requests, max_chunk_bytes=max_chunk_bytes)
if not is_aoss:
client.indices.refresh(index=index_name)
return return_ids
def add_sync(
self,
nodes: List[BaseNode],
**add_kwargs: Any,
) -> List[str]:
"""
Add nodes to index synchronously.
Args:
nodes: List[BaseNode]: list of nodes with embeddings.
"""
return self._bulk_ingest_embeddings_sync(
self._os_client, self._index, [node.embedding for node in nodes],
[node.text for node in nodes], [node.metadata for node in nodes],
[node.id for node in nodes], self._embedding_field, self._text_field
)This example shows how to add synchronous methods to the OpensearchVectorClient class. You can follow a similar approach to add synchronous versions of other asynchronous methods as needed [1].
Additionally, there are existing synchronous methods in the OpensearchVectorClient class that can be used as a reference for creating new synchronous methods. Specifically, the __init__ method uses synchronous calls to the OpenSearch client by running asynchronous methods within the event loop using event_loop.run_until_complete [1].
To continue talking to Dosu, mention @dosu.
logan-markewich
GeorgeDittmar
kobiche
Question Validation
Question
Like others (see #10719) I had some issues running my code using the OS-VectorClient as database. I have seen in the code this only runs using async code. Why? If I check the code in LangChain, they give sync and async possibilities. Can we adjust the code in LlamaIndex to offer both possibilities?