AWS MemoryDB向量检索结合JSON字段过滤失效问题咨询

需求可行性确认 #

AWS MemoryDB 7.1（启用RedisSearch）完全支持向量相似检索结合JSON元数据前置过滤，无需手动后置处理。不管是标签型字段（如tenant）还是数值型字段，都能在索引阶段将嵌套JSON字段提取为可检索的索引字段，检索时结合KNN向量查询完成过滤。

当前tenant过滤失效的原因与修复 #

你当前的问题出在查询字符串的语法顺序和格式错误：

• 错误地将过滤条件放在了KNN语法的箭头前，导致过滤逻辑未被正确解析；
• Tag字段的查询语法无需多余符号。

修正后的检索代码 #

ef_runtime = 512
offset = page_number * page_size
total_k = offset + page_size
tenant='abc'

# 正确语法：先写过滤条件，再用=>连接KNN检索逻辑
query_string = f'@tenant:{tenant} => [KNN {total_k} @vector $query_vec AS vector_score] => {{$EF_RUNTIME: {ef_runtime}}}'
q = (Query(query_string)
  .dialect(2)
  .sort_by('vector_score', asc=True)
  .return_fields('vector_score', 'metadata')
  .paging(offset, page_size))

params_dict = { "query_vec": user_vector_bytes }

results = redis_client.ft('my_index').search(q, params_dict)

额外验证点 #

确认你的数据存储格式是Hash类型，且metadata字段是标准JSON字符串，例如：

redis_client.hset("item:1", mapping={
    "vector": user_vector_bytes,
    "metadata": '{"tenant": "abc", "price": 99.9, "category": "electronics"}'
})

数值字段过滤的实现方案 #

如果需要基于JSON元数据中的数值字段（如metadata.price）过滤，只需在索引阶段添加对应的数值型字段索引，检索时结合范围过滤语法即可。

1. 更新索引schema #

from redis.commands.search.field import VectorField, TextField, TagField, NumericField

schema = [
  VectorField(
    "vector",                           
    "HNSW",
    {
      "TYPE": "FLOAT32",
      "DIM": 32,
      "DISTANCE_METRIC": "COSINE"
    }
  ),
  TextField("metadata"),
  TagField("$.metadata.tenant", as_name="tenant"),
  # 添加数值字段索引，提取metadata中的price字段
  NumericField("$.metadata.price", as_name="price")
]

definition = IndexDefinition(
  prefix=["item:"],
  index_type=IndexType.HASH
)

# 注意：若已有索引，需先删除再重建（或用ALTER INDEX添加字段）
redis_client.ft('my_index').create_index(schema, definition=definition)

2. 结合数值过滤的向量检索示例 #

比如检索tenant=abc且price<=100的物品，同时做向量KNN：

query_string = f'@tenant:{tenant} @price:[0 100] => [KNN {total_k} @vector $query_vec AS vector_score] => {{$EF_RUNTIME: {ef_runtime}}}'
q = (Query(query_string)
  .dialect(2)
  .sort_by('vector_score', asc=True)
  .return_fields('vector_score', 'metadata')
  .paging(offset, page_size))

results = redis_client.ft('my_index').search(q, params_dict)

内容的提问来源于stack exchange，提问作者vellamatic