diff --git a/python/semantic_kernel/connectors/redis.py b/python/semantic_kernel/connectors/redis.py index 575624895aca..5f32e98d16d7 100644 --- a/python/semantic_kernel/connectors/redis.py +++ b/python/semantic_kernel/connectors/redis.py @@ -3,6 +3,7 @@ import ast import asyncio import contextlib +import inspect import json import logging import sys @@ -17,6 +18,7 @@ from redis.commands.search.field import Field as RedisField from redis.commands.search.field import NumericField, TagField, TextField, VectorField from redis.commands.search.index_definition import IndexDefinition, IndexType +from redisvl.index import AsyncSearchIndex from redisvl.index.index import process_results from redisvl.query.filter import FilterExpression, Num, Tag, Text from redisvl.query.query import BaseQuery, VectorQuery @@ -166,6 +168,36 @@ def _definition_to_redis_fields( return fields +async def _process_search_results( + results: Any, + query: BaseQuery, + collection_name: str, + redis_database: Redis, + collection_type: RedisCollectionTypes, +) -> Any: + """Process RedisVL results across its old and current APIs.""" + parameters = list(inspect.signature(process_results).parameters.values()) + if len(parameters) < 3: + raise VectorSearchExecutionException( + "Unsupported redisvl process_results() signature." + ) + + third_parameter = parameters[2].name + if third_parameter == "storage_type": + return process_results(results, query, STORAGE_TYPE_MAP[collection_type]) + + if third_parameter == "schema": + index = await AsyncSearchIndex.from_existing( + name=collection_name, + redis_client=redis_database, + ) + return process_results(results, query, index.schema) + + raise VectorSearchExecutionException( + f"Unsupported redisvl process_results() parameter: {third_parameter}." + ) + + @release_candidate class RedisSettings(KernelBaseSettings): """Redis model settings. @@ -321,7 +353,20 @@ async def _inner_search( results = await self.redis_database.ft(self.collection_name).search( # type: ignore query=query.query, query_params=query.params ) - processed = process_results(results, query, STORAGE_TYPE_MAP[self.collection_type]) + try: + processed = await _process_search_results( + results, + query, + self.collection_name, + self.redis_database, + self.collection_type, + ) + except VectorSearchExecutionException: + raise + except Exception as exc: + raise VectorSearchExecutionException( + f"An error occurred during the search: {exc}" + ) from exc return KernelSearchResults( results=self._get_vector_search_results_from_results(desync_list(processed)), total_count=results.total, @@ -616,8 +661,15 @@ def _deserialize_store_models_to_dicts( case FieldTypes.KEY: rec[field.name] = self._unget_redis_key(rec[field.name]) case "vector": - dtype = DATATYPE_MAP_VECTOR[field.type_ or "default"] - rec[field.name] = buffer_to_array(rec[field.name], dtype) + # When include_vectors=False (the default for search), the vector + # field is not returned by Redis and will be absent from `rec`. + # Guard against KeyError before attempting to decode the buffer. + storage_name = field.storage_name or field.name + if storage_name in rec: + dtype = DATATYPE_MAP_VECTOR[field.type_ or "default"] + rec[field.name] = buffer_to_array(rec[storage_name], dtype) + else: + rec[field.name] = None results.append(rec) return results diff --git a/python/tests/unit/connectors/memory/test_redis_store.py b/python/tests/unit/connectors/memory/test_redis_store.py index e779ad945a97..8d7c556c6f99 100644 --- a/python/tests/unit/connectors/memory/test_redis_store.py +++ b/python/tests/unit/connectors/memory/test_redis_store.py @@ -2,6 +2,8 @@ from unittest.mock import AsyncMock, patch +import semantic_kernel.connectors.redis as redis_module + import numpy as np from pytest import fixture, mark, raises from redis.asyncio.client import Redis @@ -306,3 +308,72 @@ async def test_create_index_manual(collection_hash, mock_ensure_collection_exist async def test_create_index_fail(collection_hash, mock_ensure_collection_exists): with raises(VectorStoreOperationException, match="Invalid index type supplied."): await collection_hash.ensure_collection_exists(index_definition="index_definition", fields="fields") + + +async def test_process_search_results_with_legacy_redisvl_api(): + results = object() + query = object() + + def legacy_process_results(results_arg, query_arg, storage_type): + assert results_arg is results + assert query_arg is query + assert storage_type is redis_module.StorageType.HASH + return [{"id": "legacy"}] + + with patch.object( + redis_module, "process_results", new=legacy_process_results + ): + processed = await redis_module._process_search_results( + results, + query, + "test", + object(), + redis_module.RedisCollectionTypes.HASHSET, + ) + + assert processed == [{"id": "legacy"}] + + +async def test_process_search_results_with_current_redisvl_api(): + results = object() + query = object() + schema = object() + index = type("Index", (), {"schema": schema})() + redis_database = object() + + def current_process_results(results_arg, query_arg, schema_arg): + assert results_arg is results + assert query_arg is query + assert schema_arg is schema + return [{"id": "current"}] + + with ( + patch.object( + redis_module, "process_results", new=current_process_results + ), + patch.object( + redis_module.AsyncSearchIndex, + "from_existing", + new=AsyncMock(return_value=index), + ) as from_existing, + ): + processed = await redis_module._process_search_results( + results, + query, + "test", + redis_database, + redis_module.RedisCollectionTypes.HASHSET, + ) + + from_existing.assert_awaited_once_with( + name="test", redis_client=redis_database + ) + assert processed == [{"id": "current"}] + + +def test_hash_deserialization_handles_omitted_vector(collection_hash): + records = collection_hash._deserialize_store_models_to_dicts( + [{"id": "id1", "content": "content"}] + ) + + assert records[0]["vector"] is None