RedisVL 入门
redisvl 是一个功能强大的 Python 库，集成了 CLI，旨在利用 Redis 增强 AI 应用。本指南将引导您完成以下步骤
- 定义
IndexSchema - 准备示例数据集
- 创建
SearchIndex对象 - 测试
rvlCLI 功能 - 加载示例数据
- 构建
VectorQuery对象并执行搜索 - 更新
SearchIndex对象
......等等！
先决条件
- 确保您的 Python 环境中已安装
redisvl。 - 拥有正在运行的 Redis Stack 或 Redis Cloud 实例。
定义 IndexSchema
IndexSchema 维护重要的索引配置和字段定义，以便在 Redis 中启用搜索。为了便于使用，Schema 可以从 Python 字典或 YAML 文件构建。
示例 Schema 创建
考虑一个包含用户信息的数据集，包括 job、age、credit_score，以及一个 3 维的 user_embedding 向量。
您还必须为该数据集决定一个 Redis 索引名称和键前缀。以下是 YAML 和 Dict 格式的示例 Schema 定义。
YAML 定义
version: '0.1.0'
index:
name: user_simple
prefix: user_simple_docs
fields:
- name: user
type: tag
- name: credit_score
type: tag
- name: job
type: text
- name: age
type: numeric
- name: user_embedding
type: vector
attrs:
algorithm: flat
dims: 3
distance_metric: cosine
datatype: float32
将其存储在本地文件中，例如 schema.yaml，以便 RedisVL 使用。
Python 字典
schema = {
"index": {
"name": "user_simple",
"prefix": "user_simple_docs",
},
"fields": [
{"name": "user", "type": "tag"},
{"name": "credit_score", "type": "tag"},
{"name": "job", "type": "text"},
{"name": "age", "type": "numeric"},
{
"name": "user_embedding",
"type": "vector",
"attrs": {
"dims": 3,
"distance_metric": "cosine",
"algorithm": "flat",
"datatype": "float32"
}
}
]
}
示例数据集准备
下面，创建一个包含 user、job、age、credit_score 和 user_embedding 字段的模拟数据集。user_embedding 向量是用于演示的合成示例。
有关创建真实世界嵌入的更多信息，请参阅这篇文章。
import numpy as np
data = [
{
'user': 'john',
'age': 1,
'job': 'engineer',
'credit_score': 'high',
'user_embedding': np.array([0.1, 0.1, 0.5], dtype=np.float32).tobytes()
},
{
'user': 'mary',
'age': 2,
'job': 'doctor',
'credit_score': 'low',
'user_embedding': np.array([0.1, 0.1, 0.5], dtype=np.float32).tobytes()
},
{
'user': 'joe',
'age': 3,
'job': 'dentist',
'credit_score': 'medium',
'user_embedding': np.array([0.9, 0.9, 0.1], dtype=np.float32).tobytes()
}
]
如上所示，示例 user_embedding 向量被转换为字节。使用 NumPy 可以相当容易地实现这一点。
创建 SearchIndex
准备好 Schema 和示例数据集后，创建一个 SearchIndex。
使用您自己的 Redis 连接实例
这适用于您在连接实例上有自定义设置或您的应用程序将共享连接池的场景
from redisvl.index import SearchIndex
from redis import Redis
client = Redis.from_url("redis://localhost:6379")
index = SearchIndex.from_dict(schema, redis_client=client, validate_on_load=True)
让索引管理连接实例
这适用于简单场景
index = SearchIndex.from_dict(schema, redis_url="redis://localhost:6379", validate_on_load=True)
# If you don't specify a client or Redis URL, the index will attempt to
# connect to Redis at the default address "redis://localhost:6379".
<redisvl.index.index.SearchIndex at 0x10faca900>
创建索引
现在我们已经连接到 Redis，需要运行创建命令。
index.create(overwrite=True)
请注意，此时索引没有条目。接下来是数据加载。
使用 rvl CLI 进行检查
使用 rvl CLI 检查创建的索引及其字段
!rvl index listall
10:59:25 [RedisVL] INFO Indices:
10:59:25 [RedisVL] INFO 1. user_simple
!rvl index info -i user_simple
Index Information:
╭──────────────┬────────────────┬──────────────────────┬─────────────────┬────────────╮
│ Index Name │ Storage Type │ Prefixes │ Index Options │ Indexing │
├──────────────┼────────────────┼──────────────────────┼─────────────────┼────────────┤
│ user_simple │ HASH │ ['user_simple_docs'] │ [] │ 0 │
╰──────────────┴────────────────┴──────────────────────┴─────────────────┴────────────╯
Index Fields:
╭────────────────┬────────────────┬─────────┬────────────────┬────────────────┬────────────────┬────────────────┬────────────────┬────────────────┬─────────────────┬────────────────╮
│ Name │ Attribute │ Type │ Field Option │ Option Value │ Field Option │ Option Value │ Field Option │ Option Value │ Field Option │ Option Value │
├────────────────┼────────────────┼─────────┼────────────────┼────────────────┼────────────────┼────────────────┼────────────────┼────────────────┼─────────────────┼────────────────┤
│ user │ user │ TAG │ SEPARATOR │ , │ │ │ │ │ │ │
│ credit_score │ credit_score │ TAG │ SEPARATOR │ , │ │ │ │ │ │ │
│ job │ job │ TEXT │ WEIGHT │ 1 │ │ │ │ │ │ │
│ age │ age │ NUMERIC │ │ │ │ │ │ │ │ │
│ user_embedding │ user_embedding │ VECTOR │ algorithm │ FLAT │ data_type │ FLOAT32 │ dim │ 3 │ distance_metric │ COSINE │
╰────────────────┴────────────────┴─────────┴────────────────┴────────────────┴────────────────┴────────────────┴────────────────┴────────────────┴─────────────────┴────────────────╯
将数据加载到 SearchIndex
将示例数据集加载到 Redis。
加载时验证数据条目
RedisVL 在底层使用 pydantic 验证来确保加载的数据有效并符合您的 Schema。此设置是可选的，可以在 SearchIndex 类中配置。
keys = index.load(data)
print(keys)
['user_simple_docs:01JQ9FEZ4GAAYT9W7BWAF7CV18', 'user_simple_docs:01JQ9FEZ4JCE5FD1D5QY6BAJ0J', 'user_simple_docs:01JQ9FEZ4KF9AZYBKMYNMYBZ5A']
默认情况下，load 将通过结合索引键 prefix 和随机 ULID 创建唯一的 Redis 键。您也可以通过提供直接键或在加载时指向指定的 id_field 来自定义键。
加载无效数据
如果在 SearchIndex 类中将 validate_on_load 设置为 true，这将引发 SchemaValidationError。
# NBVAL_SKIP
keys = index.load([{"user_embedding": True}])
11:00:03 redisvl.index.index ERROR Schema validation error while loading data
Traceback (most recent call last):
File "/Users/tyler.hutcherson/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py", line 204, in _preprocess_and_validate_objects
processed_obj = self._validate(processed_obj)
File "/Users/tyler.hutcherson/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py", line 160, in _validate
return validate_object(self.index_schema, obj)
File "/Users/tyler.hutcherson/Documents/AppliedAI/redis-vl-python/redisvl/schema/validation.py", line 274, in validate_object
validated = model_class.model_validate(flat_obj)
File "/Users/tyler.hutcherson/Library/Caches/pypoetry/virtualenvs/redisvl-VnTEShF2-py3.13/lib/python3.13/site-packages/pydantic/main.py", line 627, in model_validate
return cls.__pydantic_validator__.validate_python(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
obj, strict=strict, from_attributes=from_attributes, context=context
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
pydantic_core._pydantic_core.ValidationError: 1 validation error for user_simple__PydanticModel
user_embedding
Input should be a valid bytes [type=bytes_type, input_value=True, input_type=bool]
For further information visit https://errors.pydantic.dev/2.10/v/bytes_type
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/Users/tyler.hutcherson/Documents/AppliedAI/redis-vl-python/redisvl/index/index.py", line 586, in load
return self._storage.write(
~~~~~~~~~~~~~~~~~~~^
self._redis_client, # type: ignore
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
...<6 lines>...
validate=self._validate_on_load,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)
^
File "/Users/tyler.hutcherson/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py", line 265, in write
prepared_objects = self._preprocess_and_validate_objects(
list(objects), # Convert Iterable to List
...<3 lines>...
validate=validate,
)
File "/Users/tyler.hutcherson/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py", line 211, in _preprocess_and_validate_objects
raise SchemaValidationError(str(e), index=i) from e
redisvl.exceptions.SchemaValidationError: Validation failed for object at index 0: 1 validation error for user_simple__PydanticModel
user_embedding
Input should be a valid bytes [type=bytes_type, input_value=True, input_type=bool]
For further information visit https://errors.pydantic.dev/2.10/v/bytes_type
---------------------------------------------------------------------------
ValidationError Traceback (most recent call last)
File ~/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py:204, in BaseStorage._preprocess_and_validate_objects(self, objects, id_field, keys, preprocess, validate)
203 if validate:
--> 204 processed_obj = self._validate(processed_obj)
206 # Store valid object with its key for writing
File ~/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py:160, in BaseStorage._validate(self, obj)
159 # Pass directly to validation function and let any errors propagate
--> 160 return validate_object(self.index_schema, obj)
File ~/Documents/AppliedAI/redis-vl-python/redisvl/schema/validation.py:274, in validate_object(schema, obj)
273 # Validate against model
--> 274 validated = model_class.model_validate(flat_obj)
275 return validated.model_dump(exclude_none=True)
File ~/Library/Caches/pypoetry/virtualenvs/redisvl-VnTEShF2-py3.13/lib/python3.13/site-packages/pydantic/main.py:627, in BaseModel.model_validate(cls, obj, strict, from_attributes, context)
626 __tracebackhide__ = True
--> 627 return cls.__pydantic_validator__.validate_python(
628 obj, strict=strict, from_attributes=from_attributes, context=context
629 )
ValidationError: 1 validation error for user_simple__PydanticModel
user_embedding
Input should be a valid bytes [type=bytes_type, input_value=True, input_type=bool]
For further information visit https://errors.pydantic.dev/2.10/v/bytes_type
The above exception was the direct cause of the following exception:
SchemaValidationError Traceback (most recent call last)
Cell In[16], line 1
----> 1 keys = index.load([{"user_embedding": True}])
File ~/Documents/AppliedAI/redis-vl-python/redisvl/index/index.py:586, in SearchIndex.load(self, data, id_field, keys, ttl, preprocess, batch_size)
556 """Load objects to the Redis database. Returns the list of keys loaded
557 to Redis.
558
(...)
583 RedisVLError: If there's an error loading data to Redis.
584 """
585 try:
--> 586 return self._storage.write(
587 self._redis_client, # type: ignore
588 objects=data,
589 id_field=id_field,
590 keys=keys,
591 ttl=ttl,
592 preprocess=preprocess,
593 batch_size=batch_size,
594 validate=self._validate_on_load,
595 )
596 except SchemaValidationError:
597 # Pass through validation errors directly
598 logger.exception("Schema validation error while loading data")
File ~/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py:265, in BaseStorage.write(self, redis_client, objects, id_field, keys, ttl, preprocess, batch_size, validate)
262 return []
264 # Pass 1: Preprocess and validate all objects
--> 265 prepared_objects = self._preprocess_and_validate_objects(
266 list(objects), # Convert Iterable to List
267 id_field=id_field,
268 keys=keys,
269 preprocess=preprocess,
270 validate=validate,
271 )
273 # Pass 2: Write all valid objects in batches
274 added_keys = []
File ~/Documents/AppliedAI/redis-vl-python/redisvl/index/storage.py:211, in BaseStorage._preprocess_and_validate_objects(self, objects, id_field, keys, preprocess, validate)
207 prepared_objects.append((key, processed_obj))
209 except ValidationError as e:
210 # Convert Pydantic ValidationError to SchemaValidationError with index context
--> 211 raise SchemaValidationError(str(e), index=i) from e
212 except Exception as e:
213 # Capture other exceptions with context
214 object_id = f"at index {i}"
SchemaValidationError: Validation failed for object at index 0: 1 validation error for user_simple__PydanticModel
user_embedding
Input should be a valid bytes [type=bytes_type, input_value=True, input_type=bool]
For further information visit https://errors.pydantic.dev/2.10/v/bytes_type
使用新数据 Upsert 索引
再次使用 load 方法 Upsert 数据
# Add more data
new_data = [{
'user': 'tyler',
'age': 9,
'job': 'engineer',
'credit_score': 'high',
'user_embedding': np.array([0.1, 0.3, 0.5], dtype=np.float32).tobytes()
}]
keys = index.load(new_data)
print(keys)
['user_simple_docs:01JQ9FHCB1B64GXF6WPK127VZ6']
创建 VectorQuery 对象
接下来，我们将为新填充的索引创建一个向量查询对象。本示例将使用一个简单的向量来演示向量相似度的工作原理。生产环境中的向量通常会比 3 个浮点数大得多，并且通常需要机器学习模型（例如 Huggingface sentence transformers）或嵌入 API（Cohere、OpenAI）。redisvl 提供了一系列向量化器来协助向量创建。
from redisvl.query import VectorQuery
from jupyterutils import result_print
query = VectorQuery(
vector=[0.1, 0.1, 0.5],
vector_field_name="user_embedding",
return_fields=["user", "age", "job", "credit_score", "vector_distance"],
num_results=3
)
执行查询
定义了上面的 VectorQuery 对象后，我们可以使用 query 方法在 SearchIndex 上执行查询。
results = index.query(query)
result_print(results)
| 向量距离 | 用户 | 年龄 | 工作 | 信用评分 |
|---|---|---|---|---|
| 0 | john | 1 | 工程师 | 高 |
| 0 | mary | 2 | 医生 | 低 |
| 0.0566299557686 | tyler | 9 | 工程师 | 高 |
使用异步 Redis 客户端
AsyncSearchIndex 类与异步 Redis python 客户端一起使用，允许异步执行查询、索引创建和数据加载。这是在类生产环境中处理 redisvl 的推荐方式。
schema
{'index': {'name': 'user_simple', 'prefix': 'user_simple_docs'},
'fields': [{'name': 'user', 'type': 'tag'},
{'name': 'credit_score', 'type': 'tag'},
{'name': 'job', 'type': 'text'},
{'name': 'age', 'type': 'numeric'},
{'name': 'user_embedding',
'type': 'vector',
'attrs': {'dims': 3,
'distance_metric': 'cosine',
'algorithm': 'flat',
'datatype': 'float32'}}]}
from redisvl.index import AsyncSearchIndex
from redis.asyncio import Redis
client = Redis.from_url("redis://localhost:6379")
index = AsyncSearchIndex.from_dict(schema, redis_client=client)
# execute the vector query async
results = await index.query(query)
result_print(results)
| 向量距离 | 用户 | 年龄 | 工作 | 信用评分 |
|---|---|---|---|---|
| 0 | john | 1 | 工程师 | 高 |
| 0 | mary | 2 | 医生 | 低 |
| 0.0566299557686 | tyler | 9 | 工程师 | 高 |
更新 Schema
在某些场景下，更新索引 Schema 是有意义的。使用 Redis 和 redisvl，这很容易，因为 Redis 可以在您更改或更新索引配置时保留底层数据。
因此，对于我们的场景，假设我们想以两种方式重新索引此数据
- 通过为
job字段使用Tag类型而不是Text - 通过为
user_embedding字段使用hnsw向量索引而不是flat向量索引
# Modify this schema to have what we want
index.schema.remove_field("job")
index.schema.remove_field("user_embedding")
index.schema.add_fields([
{"name": "job", "type": "tag"},
{
"name": "user_embedding",
"type": "vector",
"attrs": {
"dims": 3,
"distance_metric": "cosine",
"algorithm": "hnsw",
"datatype": "float32"
}
}
])
# Run the index update but keep underlying data in place
await index.create(overwrite=True, drop=False)
11:01:30 redisvl.index.index INFO Index already exists, overwriting.
# Execute the vector query async
results = await index.query(query)
result_print(results)
| 向量距离 | 用户 | 年龄 | 工作 | 信用评分 |
|---|---|---|---|---|
| 0 | john | 1 | 工程师 | 高 |
| 0 | mary | 2 | 医生 | 低 |
| 0.0566299557686 | tyler | 9 | 工程师 | 高 |
检查索引统计信息
使用 rvl CLI 检查索引的统计信息
!rvl stats -i user_simple
Statistics:
╭─────────────────────────────┬─────────────╮
│ Stat Key │ Value │
├─────────────────────────────┼─────────────┤
│ num_docs │ 4 │
│ num_terms │ 0 │
│ max_doc_id │ 4 │
│ num_records │ 20 │
│ percent_indexed │ 1 │
│ hash_indexing_failures │ 0 │
│ number_of_uses │ 2 │
│ bytes_per_record_avg │ 47.8 │
│ doc_table_size_mb │ 0.000423431 │
│ inverted_sz_mb │ 0.000911713 │
│ key_table_size_mb │ 0.000165939 │
│ offset_bits_per_record_avg │ nan │
│ offset_vectors_sz_mb │ 0 │
│ offsets_per_term_avg │ 0 │
│ records_per_doc_avg │ 5 │
│ sortable_values_size_mb │ 0 │
│ total_indexing_time │ 6.529 │
│ total_inverted_index_blocks │ 11 │
│ vector_index_sz_mb │ 0.235947 │
╰─────────────────────────────┴─────────────╯
清理
下面我们将进行清理。首先，您可以使用 .clear() 方法从 Redis 中清除与索引关联的所有数据。这将保留二级索引，以便将来进行插入或更新。
但如果您想清理所有内容，包括索引，只需使用 .delete()，它将默认删除索引和底层数据。
# Clear all data from Redis associated with the index
await index.clear()
4
# Butm the index is still in place
await index.exists()
True
# Remove / delete the index in its entirety
await index.delete()