feat(vector_stores): add OpenSearch k-NN search as a vector store provider

This commit is contained in:
mateo-berri 2026-10-20 19:42:28 -07:00
parent 5e189c7267
commit 37e2b86d42
10 changed files with 553 additions and 12 deletions

View file

View file

@ -0,0 +1,279 @@
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Final, NoReturn
from urllib.parse import quote, urlsplit
import httpx
from pydantic import ConfigDict, Field, TypeAdapter, ValidationError
from litellm.exceptions import BadRequestError
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.vector_store.transformation import (
BaseQueryEmbeddingVectorStoreConfig,
VectorStoreEmbeddingExecutor,
)
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import EmbeddingResponse
from litellm.types.vector_stores import (
BaseVectorStoreAuthCredentials,
VectorStoreCreateOptionalRequestParams,
VectorStoreIndexEndpoints,
VectorStoreResultContent,
VectorStoreSearchOptionalRequestParams,
VectorStoreSearchResponse,
VectorStoreSearchResult,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
DEFAULT_EMBEDDING_FIELD: Final = "embedding"
DEFAULT_TEXT_FIELD: Final = "text"
DEFAULT_MAX_NUM_RESULTS: Final = 10
MAX_MAX_NUM_RESULTS: Final = 50
_PARAM_PREFIX: Final = "opensearch_"
_QUERY_VECTOR: Final = TypeAdapter(tuple[float, ...])
_SEARCH_ONLY_MESSAGE: Final = (
"The OpenSearch vector store is search-only. Create the k-NN index in OpenSearch, "
"then register it here with the index name as vector_store_id."
)
def _bad_request(message: str) -> BadRequestError:
return BadRequestError(message=message, model=None, llm_provider="opensearch")
class _OpenSearchParams(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
litellm_embedding_model: str | None = None
opensearch_embedding_field: str | None = None
opensearch_text_field: str | None = None
@property
def embedding_field(self) -> str:
return self.opensearch_embedding_field or DEFAULT_EMBEDDING_FIELD
@property
def text_field(self) -> str:
return self.opensearch_text_field or DEFAULT_TEXT_FIELD
def require_embedding_model(self) -> str:
if not self.litellm_embedding_model:
raise _bad_request(
"litellm_embedding_model is required for the OpenSearch vector store. It must be the model that "
f"produced the vectors in '{self.embedding_field}', for example openai/text-embedding-3-small."
)
return self.litellm_embedding_model
class _SearchOptions(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
max_num_results: int | None = None
filters: Mapping[str, object] | None = None
ranking_options: Mapping[str, object] | None = None
rewrite_query: bool | None = None
@property
def unsupported(self) -> tuple[str, ...]:
requested: Final = (
("filters", self.filters),
("ranking_options", self.ranking_options),
("rewrite_query", self.rewrite_query),
)
return tuple(name for name, value in requested if value)
@property
def k(self) -> int:
if self.max_num_results is None:
return DEFAULT_MAX_NUM_RESULTS
if not 1 <= self.max_num_results <= MAX_MAX_NUM_RESULTS:
raise _bad_request(
f"max_num_results must be between 1 and {MAX_MAX_NUM_RESULTS}, got {self.max_num_results}"
)
return self.max_num_results
_KNOWN_PARAMS: Final = frozenset(name for name in _OpenSearchParams.model_fields if name.startswith(_PARAM_PREFIX))
class _LoggedSearch(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True, extra="ignore")
query: str | tuple[str, ...] = ""
litellm_params: _OpenSearchParams = _OpenSearchParams()
@property
def search_query(self) -> str:
return BaseQueryEmbeddingVectorStoreConfig.query_text(self.query)
class _Hit(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
id: str = Field(alias="_id")
score: float | None = Field(default=None, alias="_score")
source: Mapping[str, object] = Field(default_factory=dict, alias="_source")
class _Hits(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
hits: Sequence[_Hit]
class _SearchResponse(LiteLLMBaseModel):
model_config = ConfigDict(frozen=True)
hits: _Hits
def _search_result(hit: _Hit, params: _OpenSearchParams) -> VectorStoreSearchResult:
text: Final = hit.source.get(params.text_field)
filename: Final = hit.source.get("filename")
return VectorStoreSearchResult(
score=hit.score,
content=[VectorStoreResultContent(type="text", text=text)] if isinstance(text, str) else [],
file_id=hit.id,
filename=filename if isinstance(filename, str) else None,
attributes={
key: value for key, value in hit.source.items() if key not in (params.text_field, params.embedding_field)
},
)
class OpenSearchVectorStoreConfig(BaseQueryEmbeddingVectorStoreConfig):
def get_auth_credentials(self, litellm_params: Mapping[str, object]) -> BaseVectorStoreAuthCredentials:
return BaseVectorStoreAuthCredentials()
def get_vector_store_endpoints_by_type(self) -> VectorStoreIndexEndpoints:
return VectorStoreIndexEndpoints(read=[], write=[])
def validate_environment(
self, headers: Mapping[str, object], litellm_params: GenericLiteLLMParams | None
) -> dict[str, object]: # mutable-ok: the shared HTTP handler requires writable headers
return {**headers, "Content-Type": "application/json"}
def get_complete_url(self, api_base: str | None, litellm_params: Mapping[str, object]) -> str:
parsed: Final = urlsplit(api_base or "")
if parsed.scheme not in ("http", "https") or not parsed.hostname:
raise _bad_request(
"api_base must be the OpenSearch endpoint URL, for example https://<opensearch-host>:9200."
)
return (api_base or "").rstrip("/")
@staticmethod
def _params(
litellm_params: Mapping[str, object],
optional_params: VectorStoreSearchOptionalRequestParams,
extra_body: Mapping[str, object] | None,
) -> tuple[_OpenSearchParams, int]:
unknown: Final = sorted(
key for key in litellm_params if key.startswith(_PARAM_PREFIX) and key not in _KNOWN_PARAMS
)
if unknown:
raise _bad_request(
f"Unknown OpenSearch vector store parameter(s): {', '.join(unknown)}. "
f"Supported: {', '.join(sorted(_KNOWN_PARAMS))}."
)
if extra_body:
raise _bad_request("The OpenSearch vector store does not support extra_body.")
try:
options: Final = _SearchOptions.model_validate(optional_params)
except ValidationError as error:
raise _bad_request(f"Invalid vector store search parameters: {error}") from None
if options.unsupported:
raise _bad_request(
f"The OpenSearch vector store does not support {', '.join(options.unsupported)} yet. "
"Remove it from the request."
)
try:
return _OpenSearchParams.model_validate(litellm_params), options.k
except ValidationError:
raise _bad_request(
"opensearch_embedding_field, opensearch_text_field, and litellm_embedding_model must be strings."
) from None
@staticmethod
def _search_request(
vector_store_id: str,
embedding: EmbeddingResponse,
k: int,
params: _OpenSearchParams,
api_base: str,
) -> tuple[str, dict[str, object]]: # mutable-ok: the provider contract returns a writable JSON request body
vector: Final = _QUERY_VECTOR.validate_python(embedding.data[0]["embedding"]) # pyright: ignore[reportUnknownMemberType] # EmbeddingResponse.data is an untyped list, the vector is validated here
return (
f"{api_base}/{quote(vector_store_id, safe='')}/_search",
{
"size": k,
"query": {"knn": {params.embedding_field: {"vector": vector, "k": k}}},
"_source": {"excludes": [params.embedding_field]},
},
)
def transform_search_vector_store_request(
self,
vector_store_id: str,
query: str | Sequence[str],
vector_store_search_optional_params: VectorStoreSearchOptionalRequestParams,
api_base: str,
litellm_logging_obj: "LiteLLMLoggingObj",
litellm_params: Mapping[str, object],
extra_body: Mapping[str, object] | None = None,
embedding_executor: VectorStoreEmbeddingExecutor | None = None,
) -> tuple[str, dict[str, object]]: # mutable-ok: the provider contract returns a writable JSON request body
params, k = self._params(litellm_params, vector_store_search_optional_params, extra_body)
embedding: Final = self.query_embedding_executor(embedding_executor, None).embed(
params.require_embedding_model(), self.query_text(query), self.query_embedding_configuration(litellm_params)
)
return self._search_request(vector_store_id, embedding, k, params, api_base)
async def atransform_search_vector_store_request(
self,
vector_store_id: str,
query: str | Sequence[str],
vector_store_search_optional_params: VectorStoreSearchOptionalRequestParams,
api_base: str,
litellm_logging_obj: "LiteLLMLoggingObj",
litellm_params: Mapping[str, object],
extra_body: Mapping[str, object] | None = None,
embedding_executor: VectorStoreEmbeddingExecutor | None = None,
) -> tuple[str, dict[str, object]]: # mutable-ok: the provider contract returns a writable JSON request body
params, k = self._params(litellm_params, vector_store_search_optional_params, extra_body)
embedding: Final = await self.query_embedding_executor(embedding_executor, None).aembed(
params.require_embedding_model(), self.query_text(query), self.query_embedding_configuration(litellm_params)
)
return self._search_request(vector_store_id, embedding, k, params, api_base)
def transform_search_vector_store_response(
self, response: httpx.Response, litellm_logging_obj: "LiteLLMLoggingObj"
) -> VectorStoreSearchResponse:
if not response.is_success:
raise BaseLLMException(status_code=response.status_code, message=response.text, headers=response.headers)
try:
hits: Final = _SearchResponse.model_validate_json(response.content).hits.hits
except ValidationError:
raise BaseLLMException(
status_code=502,
message="OpenSearch answered the search without a hits.hits list.",
headers=response.headers,
) from None
logged: Final = _LoggedSearch.model_validate(litellm_logging_obj.model_call_details)
return VectorStoreSearchResponse(
object="vector_store.search_results.page",
search_query=logged.search_query,
data=[_search_result(hit, logged.litellm_params) for hit in hits],
)
def validate_create_vector_store(self) -> NoReturn:
raise _bad_request(_SEARCH_ONLY_MESSAGE)
def transform_create_vector_store_request(
self, vector_store_create_optional_params: VectorStoreCreateOptionalRequestParams, api_base: str
) -> NoReturn:
raise _bad_request(_SEARCH_ONLY_MESSAGE)
def transform_create_vector_store_response(self, response: httpx.Response) -> NoReturn:
raise _bad_request(_SEARCH_ONLY_MESSAGE)

View file

@ -4507,6 +4507,7 @@ class LlmProviders(str, Enum):
S3_VECTORS = "s3_vectors"
VALKEY = "valkey"
MONGODB = "mongodb"
OPENSEARCH = "opensearch"
HELICONE = "helicone"
HYPERBOLIC = "hyperbolic"
RECRAFT = "recraft"

View file

@ -10189,6 +10189,22 @@ class ProviderConfigManager:
return BedrockVectorStore.get_initialized_custom_logger()
return None
@staticmethod
def _get_vertex_ai_vector_stores_config(api_type: str | None) -> BaseVectorStoreConfig | None:
if api_type == "rag_api" or api_type is None:
from litellm.llms.vertex_ai.vector_stores.rag_api.transformation import (
VertexVectorStoreConfig,
)
return VertexVectorStoreConfig()
if api_type == "search_api":
from litellm.llms.vertex_ai.vector_stores.search_api.transformation import (
VertexSearchAPIVectorStoreConfig,
)
return VertexSearchAPIVectorStoreConfig()
return None
@staticmethod
def get_provider_vector_stores_config(
provider: LlmProviders,
@ -10210,18 +10226,7 @@ class ProviderConfigManager:
return AzureOpenAIVectorStoreConfig()
elif litellm.LlmProviders.VERTEX_AI == provider:
if api_type == "rag_api" or api_type is None: # default to rag_api
from litellm.llms.vertex_ai.vector_stores.rag_api.transformation import (
VertexVectorStoreConfig,
)
return VertexVectorStoreConfig()
elif api_type == "search_api":
from litellm.llms.vertex_ai.vector_stores.search_api.transformation import (
VertexSearchAPIVectorStoreConfig,
)
return VertexSearchAPIVectorStoreConfig()
return ProviderConfigManager._get_vertex_ai_vector_stores_config(api_type)
elif litellm.LlmProviders.BEDROCK == provider:
from litellm.llms.bedrock.vector_stores.transformation import (
BedrockVectorStoreConfig,
@ -10276,6 +10281,12 @@ class ProviderConfigManager:
)
return MongoDBVectorStoreConfig()
elif litellm.LlmProviders.OPENSEARCH == provider:
from litellm.llms.opensearch.vector_stores.transformation import (
OpenSearchVectorStoreConfig,
)
return OpenSearchVectorStoreConfig()
return None
@staticmethod

View file

@ -3098,6 +3098,13 @@
"vector_stores_search": true
}
},
"opensearch": {
"display_name": "Amazon OpenSearch (`opensearch`)",
"url": "https://docs.litellm.ai/docs/providers/opensearch_vector_stores",
"endpoints": {
"vector_stores_search": true
}
},
"helicone": {
"display_name": "Helicone (`helicone`)",
"url": "https://docs.litellm.ai/docs/providers/helicone",

View file

View file

@ -0,0 +1,242 @@
import json
from collections.abc import Callable, Mapping
from datetime import datetime, timezone
from typing import Final
from unittest.mock import MagicMock
import httpx
import openai
import pytest
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.llms.opensearch.vector_stores.transformation import OpenSearchVectorStoreConfig
from litellm.types.utils import EmbeddingResponse
from litellm.types.vector_stores import VectorStoreSearchResponse
QUERY: Final = "How many days a year can I work abroad?"
QUERY_VECTOR: Final = [0.1, 0.2, 0.3]
OPENSEARCH: Final = "https://opensearch.internal:9200"
BASE_PARAMS: Final = {"api_base": OPENSEARCH, "litellm_embedding_model": "embedding-alias"}
class RecordingEmbeddingExecutor:
def __init__(self) -> None:
self.call: Final = MagicMock(return_value=EmbeddingResponse(data=[{"embedding": QUERY_VECTOR}]))
def embed(self, model: str, query: str, configuration: Mapping[str, object]) -> EmbeddingResponse:
return self.call(model, query, configuration)
async def aembed(self, model: str, query: str, configuration: Mapping[str, object]) -> EmbeddingResponse:
return self.call(model, query, configuration)
def _sent(respond: MagicMock) -> tuple[httpx.Request, ...]:
return tuple(call.args[0] for call in respond.call_args_list)
def _opensearch_hits(text_field: str, embedding_field: str) -> Mapping[str, object]:
return {
"took": 4,
"timed_out": False,
"hits": {
"total": {"value": 2, "relation": "eq"},
"max_score": 0.581,
"hits": [
{
"_index": "litellm-qa-docs",
"_id": "1%3A0%3Aw6ftW6EB",
"_score": 0.581,
"_source": {
text_field: "Employees may work abroad for at most 37 calendar days per year.",
embedding_field: [0.5, 0.5, 0.5],
"filename": "hr-remote-work-policy.md",
"category": "hr",
},
},
{"_index": "litellm-qa-docs", "_id": "doc-2", "_score": 0.42, "_source": {"category": "facilities"}},
],
},
}
EXPECTED_RESULTS: Final = [
{
"score": 0.581,
"content": [{"type": "text", "text": "Employees may work abroad for at most 37 calendar days per year."}],
"file_id": "1%3A0%3Aw6ftW6EB",
"filename": "hr-remote-work-policy.md",
"attributes": {"filename": "hr-remote-work-policy.md", "category": "hr"},
},
{"score": 0.42, "content": [], "file_id": "doc-2", "filename": None, "attributes": {"category": "facilities"}},
]
async def _search(
asynchronous: bool,
respond: Callable[[httpx.Request], httpx.Response],
executor: RecordingEmbeddingExecutor,
kwargs: Mapping[str, object],
) -> VectorStoreSearchResponse:
request_kwargs: Final = {
**BASE_PARAMS,
"vector_store_id": "litellm-qa-docs",
"query": QUERY,
"custom_llm_provider": "opensearch",
"_direct_vector_store_embedding_executor": executor,
**kwargs,
}
if asynchronous:
return await litellm.vector_stores.asearch(
**request_kwargs, client=AsyncHTTPHandler(transport=httpx.MockTransport(respond))
)
with httpx.Client(transport=httpx.MockTransport(respond)) as transport:
return litellm.vector_stores.search(**request_kwargs, client=HTTPHandler(client=transport))
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize(
"fields",
[{}, {"opensearch_embedding_field": "passage_vector", "opensearch_text_field": "passage"}],
)
@pytest.mark.parametrize("api_base", [OPENSEARCH, "http://localhost:9200/", "https://search.docs.example/os/"])
@pytest.mark.asyncio
async def test_search_sends_an_unsigned_knn_query_and_returns_an_openai_results_page(
asynchronous: bool, fields: Mapping[str, str], api_base: str
) -> None:
executor: Final = RecordingEmbeddingExecutor()
embedding_field: Final = fields.get("opensearch_embedding_field", "embedding")
text_field: Final = fields.get("opensearch_text_field", "text")
respond: Final = MagicMock(return_value=httpx.Response(200, json=_opensearch_hits(text_field, embedding_field)))
response: Final = await _search(
asynchronous,
respond,
executor,
{**fields, "api_base": api_base, "max_num_results": 3},
)
assert response == {"object": "vector_store.search_results.page", "search_query": QUERY, "data": EXPECTED_RESULTS}
sent: Final = _sent(respond)
assert len(sent) == 1
assert str(sent[0].url) == f"{api_base.rstrip('/')}/litellm-qa-docs/_search"
assert json.loads(sent[0].content) == {
"size": 3,
"query": {"knn": {embedding_field: {"vector": QUERY_VECTOR, "k": 3}}},
"_source": {"excludes": [embedding_field]},
}
assert sent[0].headers["content-type"] == "application/json"
assert "authorization" not in sent[0].headers
executor.call.assert_called_once_with("embedding-alias", QUERY, {})
@pytest.mark.asyncio
async def test_search_defaults_to_ten_results_like_openai() -> None:
respond: Final = MagicMock(return_value=httpx.Response(200, json={"hits": {"hits": []}}))
response: Final = await _search(True, respond, RecordingEmbeddingExecutor(), {})
assert response == {"object": "vector_store.search_results.page", "search_query": QUERY, "data": []}
body: Final = json.loads(_sent(respond)[0].content)
assert (body["size"], body["query"]["knn"]["embedding"]["k"]) == (10, 10)
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.asyncio
async def test_a_list_query_is_embedded_and_echoed_as_one_string(asynchronous: bool) -> None:
executor: Final = RecordingEmbeddingExecutor()
respond: Final = MagicMock(return_value=httpx.Response(200, json={"hits": {"hits": []}}))
response: Final = await _search(asynchronous, respond, executor, {"query": ["remote work", "abroad"]})
assert response["search_query"] == "remote work abroad"
executor.call.assert_called_once_with("embedding-alias", "remote work abroad", {})
@pytest.mark.parametrize("asynchronous", [False, True])
@pytest.mark.parametrize("status", [403, 404, 429, 500])
@pytest.mark.asyncio
async def test_opensearch_error_keeps_its_status(asynchronous: bool, status: int) -> None:
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(status, json={"status": status, "error": {"reason": "upstream refused", "type": "x"}})
with pytest.raises(openai.APIStatusError) as error:
await _search(asynchronous, respond, RecordingEmbeddingExecutor(), {})
assert error.value.status_code == status
@pytest.mark.parametrize(
"overrides,message",
[
({"filters": {"type": "eq", "key": "category", "value": "hr"}}, "filters"),
({"ranking_options": {"score_threshold": 0.5}}, "ranking_options"),
({"rewrite_query": True}, "rewrite_query"),
({"max_num_results": 0}, "max_num_results"),
({"max_num_results": 51}, "max_num_results"),
({"max_num_results": "ten"}, "Invalid vector store search parameters"),
({"litellm_embedding_model": None}, "litellm_embedding_model"),
({"opensearch_text_feild": "passage"}, "opensearch_text_feild"),
({"opensearch_text_field": 7}, "opensearch_text_field"),
({"extra_body": {"min_score": 0.5}}, "extra_body"),
({"api_base": None}, "api_base"),
({"api_base": "opensearch.internal:9200"}, "api_base"),
],
)
@pytest.mark.asyncio
async def test_unsupported_search_is_rejected_before_embedding_or_any_request(
overrides: Mapping[str, object], message: str
) -> None:
executor: Final = RecordingEmbeddingExecutor()
respond: Final = MagicMock(return_value=httpx.Response(200, json={"hits": {"hits": []}}))
with pytest.raises(litellm.BadRequestError, match=message) as error:
await _search(True, respond, executor, overrides)
assert error.value.status_code == 400
respond.assert_not_called()
executor.call.assert_not_called()
@pytest.mark.asyncio
async def test_no_op_values_of_unsupported_params_are_accepted() -> None:
def respond(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, json=_opensearch_hits("text", "embedding"))
response: Final = await _search(
True, respond, RecordingEmbeddingExecutor(), {"filters": {}, "ranking_options": {}, "rewrite_query": False}
)
assert response["data"] == EXPECTED_RESULTS
def test_create_is_rejected_because_the_store_is_search_only() -> None:
with pytest.raises(litellm.BadRequestError, match="search-only"):
litellm.vector_stores.create(custom_llm_provider="opensearch")
@pytest.mark.parametrize(
"response,status",
[
(httpx.Response(403, json={"status": 403, "error": {"type": "Forbidden"}}), 403),
(httpx.Response(200, json={"took": 1}), 502),
(httpx.Response(200, text="<html>gateway</html>"), 502),
],
)
def test_response_without_search_hits_is_an_error_not_an_empty_page(response: httpx.Response, status: int) -> None:
logging: Final = Logging(
model="",
messages=[],
stream=False,
call_type="vector_store_search",
start_time=datetime(2026, 10, 20, tzinfo=timezone.utc),
litellm_call_id="opensearch-test",
function_id="opensearch-test",
)
with pytest.raises(BaseLLMException) as error:
OpenSearchVectorStoreConfig().transform_search_vector_store_response(response, logging)
assert error.value.status_code == status

View file

@ -523,6 +523,7 @@ ADD_MODEL_UNLISTED_PROVIDERS: Final = frozenset(
"mongodb",
"nano-gpt",
"neosantara",
"opensearch",
"parasail",
"pinstripes",
"poe",