diff --git a/python/semantic_kernel/connectors/redis.py b/python/semantic_kernel/connectors/redis.py
index 575624895aca..006b1b9d97a2 100644
--- a/python/semantic_kernel/connectors/redis.py
+++ b/python/semantic_kernel/connectors/redis.py
@@ -277,9 +277,12 @@ async def ensure_collection_exists(self, **kwargs) -> None:
                 return
             raise VectorStoreOperationException("Invalid index type supplied.")
         fields = _definition_to_redis_fields(self.definition, self.collection_type)
-        index_definition = IndexDefinition(
-            prefix=f"{self.collection_name}:", index_type=INDEX_TYPE_MAP[self.collection_type]
-        )
+        if self.prefix_collection_name_to_key_names:
+            index_definition = IndexDefinition(
+                prefix=[f"{self.collection_name}:"], index_type=INDEX_TYPE_MAP[self.collection_type]
+            )
+        else:
+            index_definition = IndexDefinition(index_type=INDEX_TYPE_MAP[self.collection_type])
         await self.redis_database.ft(self.collection_name).create_index(fields, definition=index_definition, **kwargs)
 
     @override
@@ -706,7 +709,7 @@ def _add_key(self, key: TKey, record: dict[str, Any]) -> dict[str, Any]:
 
     @override
     async def _inner_delete(self, keys: Sequence[str], **kwargs: Any) -> None:
-        await asyncio.gather(*[self.redis_database.json().delete(key, **kwargs) for key in keys])
+        await asyncio.gather(*[self.redis_database.json().delete(self._get_redis_key(key), **kwargs) for key in keys])
 
     @override
     def _serialize_dicts_to_store_models(
diff --git a/python/tests/unit/connectors/memory/test_redis_store.py b/python/tests/unit/connectors/memory/test_redis_store.py
index e779ad945a97..05e8457a5003 100644
--- a/python/tests/unit/connectors/memory/test_redis_store.py
+++ b/python/tests/unit/connectors/memory/test_redis_store.py
@@ -306,3 +306,28 @@ 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_create_index_respects_prefix_flag(
+    collection_hash, collection_with_prefix_hash, mock_ensure_collection_exists
+):
+    from redis.commands.search.index_definition import IndexDefinition
+
+    # Without prefix flag: IndexDefinition should NOT contain the collection prefix
+    await collection_hash.ensure_collection_exists()
+    index_def = mock_ensure_collection_exists.call_args.kwargs["definition"]
+    assert isinstance(index_def, IndexDefinition)
+    assert "test:" not in index_def.args
+
+    mock_ensure_collection_exists.reset_mock()
+
+    # With prefix flag: IndexDefinition should include ["test:"] as prefix
+    await collection_with_prefix_hash.ensure_collection_exists()
+    index_def = mock_ensure_collection_exists.call_args.kwargs["definition"]
+    assert isinstance(index_def, IndexDefinition)
+    assert "test:" in index_def.args
+
+
+async def test_delete_json_with_prefix(collection_with_prefix_json, mock_delete_json):
+    await collection_with_prefix_json._inner_delete(["id1"])
+    mock_delete_json.assert_called_once_with("test:id1")