Redis

RedisVectorStore

Bases: BasePydanticVectorStore

RedisVectorStore.

RedisVectorStore接受用户定义的模式对象和Redis连接客户端或URL字符串。 模式是可选的，但对于以下情况很有用： - 定义自定义索引名称、键前缀和键分隔符。 - 定义额外的元数据字段以用作查询过滤器。 - 设置字段的自定义规范以改善搜索质量，例如使用哪种向量索引算法。

其他注意事项： - 所有嵌入和文档都存储在Redis中。在查询时，索引使用Redis查询前k个最相似的节点。 - Redis和LlamaIndex期望对于任何模式（默认或自定义），至少有4个必需字段，id、doc_id、text、vector。

Parameters:

Name	Type	Description	Default
redis_url（str，可选）：Redis服务器URL。默认为"redis		//localhost:6379"。	required

Example

from redisvl.schema import IndexSchema from llama_index.vector_stores.redis import RedisVectorStore

使用默认模式

rds = RedisVectorStore(redis_url="redis://localhost:6379")

使用来自字典的自定义模式

schema = IndexSchema.from_dict({ "index": {"name": "my-index", "prefix": "docs"}, "fields": [ {"name": "id", "type": "tag"}, {"name": "doc_id", "type": "tag"}, {"name": "text", "type": "text"}, {"name": "vector", "type": "vector", "attrs": {"dims": 1536, "algorithm": "flat"}} ] }) vector_store = RedisVectorStore( schema=schema, redis_url="redis://localhost:6379" )

Source code in llama_index/vector_stores/redis/base.py

class RedisVectorStore(BasePydanticVectorStore):
    """RedisVectorStore.

    RedisVectorStore接受用户定义的模式对象和Redis连接客户端或URL字符串。 模式是可选的，但对于以下情况很有用：
    - 定义自定义索引名称、键前缀和键分隔符。
    - 定义*额外*的元数据字段以用作查询过滤器。
    - 设置字段的自定义规范以改善搜索质量，例如使用哪种向量索引算法。

    其他注意事项：
    - 所有嵌入和文档都存储在Redis中。在查询时，索引使用Redis查询前k个最相似的节点。
    - Redis和LlamaIndex期望对于任何模式（默认或自定义），至少有4个*必需*字段，`id`、`doc_id`、`text`、`vector`。

    Args:
        schema（IndexSchema，可选）：Redis索引模式对象。
        redis_client（Redis，可选）：Redis客户端连接。
        redis_url（str，可选）：Redis服务器URL。默认为"redis://localhost:6379"。
        overwrite（bool，可选）：如果索引已经存在，是否覆盖它。默认为False。

    Raises:
        ValueError：如果您的Redis服务器未启用搜索或JSON。
        ValueError：如果无法建立Redis连接。
        ValueError：如果提供了无效的模式。

    Example:
        from redisvl.schema import IndexSchema
        from llama_index.vector_stores.redis import RedisVectorStore

        # 使用默认模式
        rds = RedisVectorStore(redis_url="redis://localhost:6379")

        # 使用来自字典的自定义模式
        schema = IndexSchema.from_dict({
            "index": {"name": "my-index", "prefix": "docs"},
            "fields": [
                {"name": "id", "type": "tag"},
                {"name": "doc_id", "type": "tag"},
                {"name": "text", "type": "text"},
                {"name": "vector", "type": "vector", "attrs": {"dims": 1536, "algorithm": "flat"}}
            ]
        })
        vector_store = RedisVectorStore(
            schema=schema,
            redis_url="redis://localhost:6379"
        )"""

    stores_text = True
    stores_node = True
    flat_metadata = False

    _index: SearchIndex = PrivateAttr()
    _overwrite: bool = PrivateAttr()
    _return_fields: List[str] = PrivateAttr()

    def __init__(
        self,
        schema: Optional[IndexSchema] = None,
        redis_client: Optional[Redis] = None,
        redis_url: Optional[str] = None,
        overwrite: bool = False,
        return_fields: Optional[List[str]] = None,
        **kwargs: Any,
    ) -> None:
        # check for indicators of old schema
        self._flag_old_kwargs(**kwargs)

        # Setup schema
        if not schema:
            logger.info("Using default RedisVectorStore schema.")
            schema = RedisVectorStoreSchema()

        self._validate_schema(schema)
        self._return_fields = return_fields or [
            NODE_ID_FIELD_NAME,
            DOC_ID_FIELD_NAME,
            TEXT_FIELD_NAME,
            NODE_CONTENT_FIELD_NAME,
        ]
        self._index = SearchIndex(schema=schema)
        self._overwrite = overwrite

        # Establish redis connection
        if redis_client:
            self._index.set_client(redis_client)
        elif redis_url:
            self._index.connect(redis_url)
        else:
            raise ValueError(
                "Failed to connect to Redis. Must provide a valid redis client or url"
            )

        # Create index
        self.create_index()

        super().__init__()

    def _flag_old_kwargs(self, **kwargs):
        old_kwargs = [
            "index_name",
            "index_prefix",
            "prefix_ending",
            "index_args",
            "metadata_fields",
        ]
        for kwarg in old_kwargs:
            if kwarg in kwargs:
                raise ValueError(
                    f"Deprecated kwarg, {kwarg}, found upon initialization. "
                    "RedisVectorStore now requires an IndexSchema object. "
                    "See the documentation for a complete example: https://docs.llamaindex.ai/en/stable/examples/vector_stores/RedisIndexDemo/"
                )

    def _validate_schema(self, schema: IndexSchema) -> str:
        base_schema = RedisVectorStoreSchema()
        for name, field in base_schema.fields.items():
            if (name not in schema.fields) or (
                not schema.fields[name].type == field.type
            ):
                raise ValueError(
                    f"Required field {name} must be present in the index "
                    f"and of type {schema.fields[name].type}"
                )

    @property
    def client(self) -> "Redis":
        """返回redis客户端实例。"""
        return self._index.client

    @property
    def index_name(self) -> str:
        """根据模式返回索引的名称。"""
        return self._index.name

    @property
    def schema(self) -> IndexSchema:
        """返回索引模式。"""
        return self._index.schema

    def set_return_fields(self, return_fields: List[str]) -> None:
        """更新查询响应的返回字段。"""
        self._return_fields = return_fields

    def index_exists(self) -> bool:
        """检查Redis中是否存在索引。

返回：
    bool：True或False。
"""
        return self._index.exists()

    def create_index(self, overwrite: Optional[bool] = None) -> None:
        """在Redis中创建一个索引。"""
        if overwrite is None:
            overwrite = self._overwrite
        # Create index honoring overwrite policy
        if overwrite:
            self._index.create(overwrite=True, drop=True)
        else:
            self._index.create()

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        """将节点添加到索引中。

Args:
    nodes（List[BaseNode]）：带有嵌入的节点列表

Returns:
    List[str]：已添加到索引的文档的ID列表。

引发：
    ValueError：如果索引已经存在且overwrite为False。
"""
        # Check to see if empty document list was passed
        if len(nodes) == 0:
            return []

        # Now check for the scenario where user is trying to index embeddings that don't align with schema
        embedding_len = len(nodes[0].get_embedding())
        expected_dims = self._index.schema.fields[VECTOR_FIELD_NAME].attrs.dims
        if expected_dims != embedding_len:
            raise ValueError(
                f"Attempting to index embeddings of dim {embedding_len} "
                f"which doesn't match the index schema expectation of {expected_dims}. "
                "Please review the Redis integration example to learn how to customize schema. "
                ""
            )

        data: List[Dict[str, Any]] = []
        for node in nodes:
            embedding = node.get_embedding()
            record = {
                NODE_ID_FIELD_NAME: node.node_id,
                DOC_ID_FIELD_NAME: node.ref_doc_id,
                TEXT_FIELD_NAME: node.get_content(metadata_mode=MetadataMode.NONE),
                VECTOR_FIELD_NAME: array_to_buffer(embedding),
            }
            # parse and append metadata
            additional_metadata = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )
            data.append({**record, **additional_metadata})

        # Load nodes to Redis
        keys = self._index.load(data, id_field=NODE_ID_FIELD_NAME, **add_kwargs)
        logger.info(f"Added {len(keys)} documents to index {self._index.name}")
        return [
            key.strip(self._index.prefix + self._index.key_separator) for key in keys
        ]

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """使用ref_doc_id删除节点。

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        # build a filter to target specific docs by doc ID
        doc_filter = Tag(DOC_ID_FIELD_NAME) == ref_doc_id
        total = self._index.query(CountQuery(doc_filter))
        delete_query = FilterQuery(
            return_fields=[NODE_ID_FIELD_NAME],
            filter_expression=doc_filter,
            num_results=total,
        )
        # fetch docs to delete and flush them
        docs_to_delete = self._index.search(delete_query.query, delete_query.params)
        with self._index.client.pipeline(transaction=False) as pipe:
            for doc in docs_to_delete.docs:
                pipe.delete(doc.id)
            res = pipe.execute()

        logger.info(
            f"Deleted {len(docs_to_delete.docs)} documents from index {self._index.name}"
        )

    def delete_index(self) -> None:
        """删除索引和所有文档。"""
        logger.info(f"Deleting index {self._index.name}")
        self._index.delete(drop=True)

    @staticmethod
    def _to_redis_filter(field: BaseField, filter: MetadataFilter) -> FilterExpression:
        """将标准元数据过滤器转换为特定于Redis的过滤器表达式。

Args:
    field (BaseField): 要进行过滤的字段，必须具有类型属性。
    filter (MetadataFilter): 要应用的过滤器，必须具有操作符和值属性。

Returns:
    FilterExpression: 从输入构造的特定于Redis的过滤器表达式。

引发:
    ValueError: 如果字段类型不受支持，或者如果操作符不受字段类型支持。
"""
        # Check for unsupported field type
        if field.type not in REDIS_LLAMA_FIELD_SPEC:
            raise ValueError(f"Unsupported field type {field.type} for {field.name}")

        field_info = REDIS_LLAMA_FIELD_SPEC[field.type]

        # Check for unsupported operator
        if filter.operator not in field_info["operators"]:
            raise ValueError(
                f"Filter operator {filter.operator} not supported for {field.name} of type {field.type}"
            )

        # Create field instance and apply the operator function
        field_instance = field_info["class"](field.name)
        return field_info["operators"][filter.operator](field_instance, filter.value)

    def _create_redis_filter_expression(
        self, metadata_filters: MetadataFilters
    ) -> FilterExpression:
        """生成一个Redis过滤表达式，作为元数据过滤器的组合。

Args:
    metadata_filters (MetadataFilters): 要使用的元数据过滤器列表。

Returns:
    FilterExpression: 一个Redis过滤表达式。
"""
        filter_expression = FilterExpression("*")
        if metadata_filters:
            if metadata_filters.filters:
                for filter in metadata_filters.filters:
                    # Index must be created with the metadata field in the index schema
                    field = self._index.schema.fields.get(filter.key)
                    if not field:
                        logger.warning(
                            f"{filter.key} field was not included as part of the index schema, and thus cannot be used as a filter condition."
                        )
                        continue
                    # Extract redis filter
                    redis_filter = self._to_redis_filter(field, filter)
                    # Combine with conditional
                    if metadata_filters.condition == "and":
                        filter_expression = filter_expression & redis_filter
                    else:
                        filter_expression = filter_expression | redis_filter
        return filter_expression

    def _to_redis_query(self, query: VectorStoreQuery) -> VectorQuery:
        """从VectorStoreQuery创建一个RedisQuery。"""
        filter_expression = self._create_redis_filter_expression(query.filters)
        return_fields = self._return_fields.copy()
        return VectorQuery(
            vector=query.query_embedding,
            vector_field_name=VECTOR_FIELD_NAME,
            num_results=query.similarity_top_k,
            filter_expression=filter_expression,
            return_fields=return_fields,
        )

    def _extract_node_and_score(self, doc, redis_query: VectorQuery):
        """从文档中提取节点及其分数。"""
        try:
            node = metadata_dict_to_node(
                {NODE_CONTENT_FIELD_NAME: doc[NODE_CONTENT_FIELD_NAME]}
            )
            node.text = doc[TEXT_FIELD_NAME]
        except Exception:
            # Handle legacy metadata format
            node = TextNode(
                text=doc[TEXT_FIELD_NAME],
                id_=doc[NODE_ID_FIELD_NAME],
                embedding=None,
                relationships={
                    NodeRelationship.SOURCE: RelatedNodeInfo(
                        node_id=doc[DOC_ID_FIELD_NAME]
                    )
                },
            )
        score = 1 - float(doc[redis_query.DISTANCE_ID])
        return node, score

    def _process_query_results(
        self, results, redis_query: VectorQuery
    ) -> VectorStoreQueryResult:
        """处理查询结果并返回一个VectorStoreQueryResult。"""
        ids, nodes, scores = [], [], []
        for doc in results:
            node, score = self._extract_node_and_score(doc, redis_query)
            ids.append(doc[NODE_ID_FIELD_NAME])
            nodes.append(node)
            scores.append(score)
        logger.info(f"Found {len(nodes)} results for query with id {ids}")
        return VectorStoreQueryResult(nodes=nodes, ids=ids, similarities=scores)

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询索引。

Args:
    query（VectorStoreQuery）：查询对象

Returns:
    VectorStoreQueryResult：查询结果

引发：
    ValueError：如果query.query_embedding为None。
    redis.exceptions.RedisError：如果查询索引时出现错误。
    redis.exceptions.TimeoutError：如果查询索引超时。
"""
        if not query.query_embedding:
            raise ValueError("Query embedding is required for querying.")

        redis_query = self._to_redis_query(query)
        logger.info(
            f"Querying index {self._index.name} with filters {redis_query.get_filter()}"
        )

        try:
            results = self._index.query(redis_query)
        except RedisTimeoutError as e:
            logger.error(f"Query timed out on {self._index.name}: {e}")
            raise
        except RedisError as e:
            logger.error(f"Error querying {self._index.name}: {e}")
            raise

        return self._process_query_results(results, redis_query)

    def persist(
        self,
        persist_path: Optional[str] = None,
        fs: Optional[fsspec.AbstractFileSystem] = None,
        in_background: bool = True,
    ) -> None:
        """将向量存储持久化到磁盘。

对于Redis，更多的说明请参见这里：https://redis.io/docs/management/persistence/

Args:
    persist_path (str): 持久化向量存储的路径。（不适用）
    in_background (bool, optional): 后台持久化。默认为True。
    fs (fsspec.AbstractFileSystem, optional): 要持久化到的文件系统。（不适用）

抛出:
    redis.exceptions.RedisError: 如果在将索引持久化到磁盘时出错。
"""
        try:
            if in_background:
                logger.info("Saving index to disk in background")
                self._index.client.bgsave()
            else:
                logger.info("Saving index to disk")
                self._index.client.save()

        except RedisError as e:
            logger.error(f"Error saving index to disk: {e}")
            raise

client property

client: Redis

返回redis客户端实例。

index_name property

index_name: str

根据模式返回索引的名称。

schema property

schema: IndexSchema

返回索引模式。

set_return_fields

set_return_fields(return_fields: List[str]) -> None

更新查询响应的返回字段。

Source code in llama_index/vector_stores/redis/base.py

def set_return_fields(self, return_fields: List[str]) -> None:
    """更新查询响应的返回字段。"""
    self._return_fields = return_fields

index_exists

index_exists() -> bool

检查Redis中是否存在索引。

返回： bool：True或False。

Source code in llama_index/vector_stores/redis/base.py

    def index_exists(self) -> bool:
        """检查Redis中是否存在索引。

返回：
    bool：True或False。
"""
        return self._index.exists()

create_index

create_index(overwrite: Optional[bool] = None) -> None

在Redis中创建一个索引。

Source code in llama_index/vector_stores/redis/base.py

def create_index(self, overwrite: Optional[bool] = None) -> None:
    """在Redis中创建一个索引。"""
    if overwrite is None:
        overwrite = self._overwrite
    # Create index honoring overwrite policy
    if overwrite:
        self._index.create(overwrite=True, drop=True)
    else:
        self._index.create()

add

add(nodes: List[BaseNode], **add_kwargs: Any) -> List[str]

将节点添加到索引中。

Returns:

Type	Description
List[str]	List[str]：已添加到索引的文档的ID列表。

引发： ValueError：如果索引已经存在且overwrite为False。

Source code in llama_index/vector_stores/redis/base.py

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        """将节点添加到索引中。

Args:
    nodes（List[BaseNode]）：带有嵌入的节点列表

Returns:
    List[str]：已添加到索引的文档的ID列表。

引发：
    ValueError：如果索引已经存在且overwrite为False。
"""
        # Check to see if empty document list was passed
        if len(nodes) == 0:
            return []

        # Now check for the scenario where user is trying to index embeddings that don't align with schema
        embedding_len = len(nodes[0].get_embedding())
        expected_dims = self._index.schema.fields[VECTOR_FIELD_NAME].attrs.dims
        if expected_dims != embedding_len:
            raise ValueError(
                f"Attempting to index embeddings of dim {embedding_len} "
                f"which doesn't match the index schema expectation of {expected_dims}. "
                "Please review the Redis integration example to learn how to customize schema. "
                ""
            )

        data: List[Dict[str, Any]] = []
        for node in nodes:
            embedding = node.get_embedding()
            record = {
                NODE_ID_FIELD_NAME: node.node_id,
                DOC_ID_FIELD_NAME: node.ref_doc_id,
                TEXT_FIELD_NAME: node.get_content(metadata_mode=MetadataMode.NONE),
                VECTOR_FIELD_NAME: array_to_buffer(embedding),
            }
            # parse and append metadata
            additional_metadata = node_to_metadata_dict(
                node, remove_text=True, flat_metadata=self.flat_metadata
            )
            data.append({**record, **additional_metadata})

        # Load nodes to Redis
        keys = self._index.load(data, id_field=NODE_ID_FIELD_NAME, **add_kwargs)
        logger.info(f"Added {len(keys)} documents to index {self._index.name}")
        return [
            key.strip(self._index.prefix + self._index.key_separator) for key in keys
        ]

delete

delete(ref_doc_id: str, **delete_kwargs: Any) -> None

使用ref_doc_id删除节点。

Source code in llama_index/vector_stores/redis/base.py

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """使用ref_doc_id删除节点。

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        # build a filter to target specific docs by doc ID
        doc_filter = Tag(DOC_ID_FIELD_NAME) == ref_doc_id
        total = self._index.query(CountQuery(doc_filter))
        delete_query = FilterQuery(
            return_fields=[NODE_ID_FIELD_NAME],
            filter_expression=doc_filter,
            num_results=total,
        )
        # fetch docs to delete and flush them
        docs_to_delete = self._index.search(delete_query.query, delete_query.params)
        with self._index.client.pipeline(transaction=False) as pipe:
            for doc in docs_to_delete.docs:
                pipe.delete(doc.id)
            res = pipe.execute()

        logger.info(
            f"Deleted {len(docs_to_delete.docs)} documents from index {self._index.name}"
        )

delete_index

delete_index() -> None

删除索引和所有文档。

Source code in llama_index/vector_stores/redis/base.py

def delete_index(self) -> None:
    """删除索引和所有文档。"""
    logger.info(f"Deleting index {self._index.name}")
    self._index.delete(drop=True)

query

query(
    query: VectorStoreQuery, **kwargs: Any
) -> VectorStoreQueryResult

查询索引。

Returns:

Type	Description
VectorStoreQueryResult	VectorStoreQueryResult：查询结果

引发： ValueError：如果query.query_embedding为None。 redis.exceptions.RedisError：如果查询索引时出现错误。 redis.exceptions.TimeoutError：如果查询索引超时。

Source code in llama_index/vector_stores/redis/base.py

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询索引。

Args:
    query（VectorStoreQuery）：查询对象

Returns:
    VectorStoreQueryResult：查询结果

引发：
    ValueError：如果query.query_embedding为None。
    redis.exceptions.RedisError：如果查询索引时出现错误。
    redis.exceptions.TimeoutError：如果查询索引超时。
"""
        if not query.query_embedding:
            raise ValueError("Query embedding is required for querying.")

        redis_query = self._to_redis_query(query)
        logger.info(
            f"Querying index {self._index.name} with filters {redis_query.get_filter()}"
        )

        try:
            results = self._index.query(redis_query)
        except RedisTimeoutError as e:
            logger.error(f"Query timed out on {self._index.name}: {e}")
            raise
        except RedisError as e:
            logger.error(f"Error querying {self._index.name}: {e}")
            raise

        return self._process_query_results(results, redis_query)

persist

persist(
    persist_path: Optional[str] = None,
    fs: Optional[AbstractFileSystem] = None,
    in_background: bool = True,
) -> None

将向量存储持久化到磁盘。

对于Redis，更多的说明请参见这里：https://redis.io/docs/management/persistence/

Parameters:

Name	Type	Description	Default
persist_path	str	持久化向量存储的路径。（不适用）	None
in_background	bool	后台持久化。默认为True。	True
fs	AbstractFileSystem	要持久化到的文件系统。（不适用）	None

抛出

redis.exceptions.RedisError: 如果在将索引持久化到磁盘时出错。

Source code in llama_index/vector_stores/redis/base.py

    def persist(
        self,
        persist_path: Optional[str] = None,
        fs: Optional[fsspec.AbstractFileSystem] = None,
        in_background: bool = True,
    ) -> None:
        """将向量存储持久化到磁盘。

对于Redis，更多的说明请参见这里：https://redis.io/docs/management/persistence/

Args:
    persist_path (str): 持久化向量存储的路径。（不适用）
    in_background (bool, optional): 后台持久化。默认为True。
    fs (fsspec.AbstractFileSystem, optional): 要持久化到的文件系统。（不适用）

抛出:
    redis.exceptions.RedisError: 如果在将索引持久化到磁盘时出错。
"""
        try:
            if in_background:
                logger.info("Saving index to disk in background")
                self._index.client.bgsave()
            else:
                logger.info("Saving index to disk")
                self._index.client.save()

        except RedisError as e:
            logger.error(f"Error saving index to disk: {e}")
            raise