fix(python/redis): restore vector search broken by redisvl>=0.5 API change and KeyError on include_vectors=False

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
 

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