Vector

VectorIndexRetriever

Bases: BaseRetriever

向量索引检索器。

Parameters:

Name	Type	Description	Default
index	VectorStoreIndex	向量存储索引。	required
similarity_top_k	int	返回前k个相似结果的数量。	DEFAULT_SIMILARITY_TOP_K
vector_store_query_mode	str	向量存储查询模式 有关支持的模式的完整列表，请参阅VectorStoreQueryMode的参考资料。	DEFAULT
filters	Optional[MetadataFilters]	元数据过滤器，默认为None	None
alpha	float	稀疏/密集检索的权重，仅用于混合查询模式。	None
doc_ids	Optional[List[str]]	限制搜索的文档列表。	None
vector_store_kwargs	dict	传递给向量存储的额外特定kwargs，在查询时传递。	required

Source code in llama_index/core/indices/vector_store/retrievers/retriever.py

class VectorIndexRetriever(BaseRetriever):
    """向量索引检索器。

    Args:
        index (VectorStoreIndex): 向量存储索引。
        similarity_top_k (int): 返回前k个相似结果的数量。
        vector_store_query_mode (str): 向量存储查询模式
            有关支持的模式的完整列表，请参阅VectorStoreQueryMode的参考资料。
        filters (Optional[MetadataFilters]): 元数据过滤器，默认为None
        alpha (float): 稀疏/密集检索的权重，仅用于混合查询模式。
        doc_ids (Optional[List[str]]): 限制搜索的文档列表。
        vector_store_kwargs (dict): 传递给向量存储的额外特定kwargs，在查询时传递。"""

    def __init__(
        self,
        index: VectorStoreIndex,
        similarity_top_k: int = DEFAULT_SIMILARITY_TOP_K,
        vector_store_query_mode: VectorStoreQueryMode = VectorStoreQueryMode.DEFAULT,
        filters: Optional[MetadataFilters] = None,
        alpha: Optional[float] = None,
        node_ids: Optional[List[str]] = None,
        doc_ids: Optional[List[str]] = None,
        sparse_top_k: Optional[int] = None,
        callback_manager: Optional[CallbackManager] = None,
        object_map: Optional[dict] = None,
        embed_model: Optional[BaseEmbedding] = None,
        verbose: bool = False,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        self._index = index
        self._vector_store = self._index.vector_store
        self._embed_model = embed_model or self._index._embed_model
        self._docstore = self._index.docstore

        self._similarity_top_k = similarity_top_k
        self._vector_store_query_mode = VectorStoreQueryMode(vector_store_query_mode)
        self._alpha = alpha
        self._node_ids = node_ids
        self._doc_ids = doc_ids
        self._filters = filters
        self._sparse_top_k = sparse_top_k
        self._kwargs: Dict[str, Any] = kwargs.get("vector_store_kwargs", {})

        callback_manager = callback_manager or CallbackManager()
        super().__init__(
            callback_manager=callback_manager,
            object_map=object_map,
            verbose=verbose,
        )

    @property
    def similarity_top_k(self) -> int:
        """返回相似度前k名。"""
        return self._similarity_top_k

    @similarity_top_k.setter
    def similarity_top_k(self, similarity_top_k: int) -> None:
        """设置相似度的前k个值。"""
        self._similarity_top_k = similarity_top_k

    @dispatcher.span
    def _retrieve(
        self,
        query_bundle: QueryBundle,
    ) -> List[NodeWithScore]:
        if self._vector_store.is_embedding_query:
            if query_bundle.embedding is None and len(query_bundle.embedding_strs) > 0:
                query_bundle.embedding = (
                    self._embed_model.get_agg_embedding_from_queries(
                        query_bundle.embedding_strs
                    )
                )
        return self._get_nodes_with_embeddings(query_bundle)

    @dispatcher.span
    async def _aretrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
        embedding = query_bundle.embedding
        if self._vector_store.is_embedding_query:
            if query_bundle.embedding is None and len(query_bundle.embedding_strs) > 0:
                embed_model = self._embed_model
                embedding = await embed_model.aget_agg_embedding_from_queries(
                    query_bundle.embedding_strs
                )
        return await self._aget_nodes_with_embeddings(
            QueryBundle(query_str=query_bundle.query_str, embedding=embedding)
        )

    def _build_vector_store_query(
        self, query_bundle_with_embeddings: QueryBundle
    ) -> VectorStoreQuery:
        return VectorStoreQuery(
            query_embedding=query_bundle_with_embeddings.embedding,
            similarity_top_k=self._similarity_top_k,
            node_ids=self._node_ids,
            doc_ids=self._doc_ids,
            query_str=query_bundle_with_embeddings.query_str,
            mode=self._vector_store_query_mode,
            alpha=self._alpha,
            filters=self._filters,
            sparse_top_k=self._sparse_top_k,
        )

    def _build_node_list_from_query_result(
        self, query_result: VectorStoreQueryResult
    ) -> List[NodeWithScore]:
        if query_result.nodes is None:
            # NOTE: vector store does not keep text and returns node indices.
            # Need to recover all nodes from docstore
            if query_result.ids is None:
                raise ValueError(
                    "Vector store query result should return at "
                    "least one of nodes or ids."
                )
            assert isinstance(self._index.index_struct, IndexDict)
            node_ids = [
                self._index.index_struct.nodes_dict[idx] for idx in query_result.ids
            ]
            nodes = self._docstore.get_nodes(node_ids)
            query_result.nodes = nodes
        else:
            # NOTE: vector store keeps text, returns nodes.
            # Only need to recover image or index nodes from docstore
            for i in range(len(query_result.nodes)):
                source_node = query_result.nodes[i].source_node
                if (not self._vector_store.stores_text) or (
                    source_node is not None and source_node.node_type != ObjectType.TEXT
                ):
                    node_id = query_result.nodes[i].node_id
                    if self._docstore.document_exists(node_id):
                        query_result.nodes[i] = self._docstore.get_node(
                            node_id
                        )  # type: ignore[index]

        log_vector_store_query_result(query_result)

        node_with_scores: List[NodeWithScore] = []
        for ind, node in enumerate(query_result.nodes):
            score: Optional[float] = None
            if query_result.similarities is not None:
                score = query_result.similarities[ind]
            node_with_scores.append(NodeWithScore(node=node, score=score))

        return node_with_scores

    def _get_nodes_with_embeddings(
        self, query_bundle_with_embeddings: QueryBundle
    ) -> List[NodeWithScore]:
        query = self._build_vector_store_query(query_bundle_with_embeddings)
        query_result = self._vector_store.query(query, **self._kwargs)
        return self._build_node_list_from_query_result(query_result)

    async def _aget_nodes_with_embeddings(
        self, query_bundle_with_embeddings: QueryBundle
    ) -> List[NodeWithScore]:
        query = self._build_vector_store_query(query_bundle_with_embeddings)
        query_result = await self._vector_store.aquery(query, **self._kwargs)
        return self._build_node_list_from_query_result(query_result)

similarity_top_k property writable

similarity_top_k: int

返回相似度前k名。

VectorIndexAutoRetriever

Bases: BaseAutoRetriever

向量存储自动检索器。

用于向量存储索引的检索器，使用LLM自动设置向量存储查询参数。

Parameters:

Name	Type	Description	Default
index	VectorStoreIndex	向量存储索引	required
vector_store_info	VectorStoreInfo	关于向量存储内容和支持的元数据过滤器的附加信息。自然语言描述由LLM用于自动设置向量存储查询参数。	required
prompt_template_str	Optional[str]	LLM的自定义提示模板字符串。如果为None，则使用默认模板字符串。	None
service_context	Optional[ServiceContext]	包含对LLM引用的服务上下文。如果为None，则默认使用索引的服务上下文。	None
similarity_top_k	int	返回前k个相似结果的数量。	DEFAULT_SIMILARITY_TOP_K
empty_query_top_k	Optional[int]	如果推断的查询字符串为空（仅使用元数据过滤器），则返回前k个结果的数量。可以设置为None，这将使用similarity_top_k代替。默认设置为10。	10
max_top_k	int	允许的最大top_k值。LLM或similarity_top_k设置将被限制为此值。	10
vector_store_query_mode	str	向量存储查询模式。有关支持的模式的完整列表，请参阅VectorStoreQueryMode的参考。	DEFAULT
default_empty_query_vector	Optional[List[float]]	默认的空查询向量。默认为None。如果不是None，则将此向量用作查询向量（如果查询为空）。	None
callback_manager	Optional[CallbackManager]	回调管理器	None
verbose	bool	详细模式	False

Source code in llama_index/core/indices/vector_store/retrievers/auto_retriever/auto_retriever.py

class VectorIndexAutoRetriever(BaseAutoRetriever):
    """    向量存储自动检索器。

    用于向量存储索引的检索器，使用LLM自动设置向量存储查询参数。

    Args:
        index (VectorStoreIndex): 向量存储索引
        vector_store_info (VectorStoreInfo): 关于向量存储内容和支持的元数据过滤器的附加信息。自然语言描述由LLM用于自动设置向量存储查询参数。
        prompt_template_str: LLM的自定义提示模板字符串。如果为None，则使用默认模板字符串。
        service_context: 包含对LLM引用的服务上下文。如果为None，则默认使用索引的服务上下文。
        similarity_top_k (int): 返回前k个相似结果的数量。
        empty_query_top_k (Optional[int]): 如果推断的查询字符串为空（仅使用元数据过滤器），则返回前k个结果的数量。可以设置为None，这将使用similarity_top_k代替。默认设置为10。
        max_top_k (int): 允许的最大top_k值。LLM或similarity_top_k设置将被限制为此值。
        vector_store_query_mode (str): 向量存储查询模式。有关支持的模式的完整列表，请参阅VectorStoreQueryMode的参考。
        default_empty_query_vector (Optional[List[float]]): 默认的空查询向量。默认为None。如果不是None，则将此向量用作查询向量（如果查询为空）。
        callback_manager (Optional[CallbackManager]): 回调管理器
        verbose (bool): 详细模式"""

    def __init__(
        self,
        index: VectorStoreIndex,
        vector_store_info: VectorStoreInfo,
        llm: Optional[LLM] = None,
        prompt_template_str: Optional[str] = None,
        max_top_k: int = 10,
        similarity_top_k: int = DEFAULT_SIMILARITY_TOP_K,
        empty_query_top_k: Optional[int] = 10,
        vector_store_query_mode: VectorStoreQueryMode = VectorStoreQueryMode.DEFAULT,
        default_empty_query_vector: Optional[List[float]] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        extra_filters: Optional[MetadataFilters] = None,
        object_map: Optional[dict] = None,
        objects: Optional[List[IndexNode]] = None,
        # deprecated
        service_context: Optional[ServiceContext] = None,
        **kwargs: Any,
    ) -> None:
        self._index = index
        self._vector_store_info = vector_store_info
        self._default_empty_query_vector = default_empty_query_vector

        service_context = service_context or self._index.service_context
        self._llm = llm or llm_from_settings_or_context(Settings, service_context)
        callback_manager = (
            callback_manager
            or callback_manager_from_settings_or_context(Settings, service_context)
        )

        # prompt
        prompt_template_str = (
            prompt_template_str or DEFAULT_VECTOR_STORE_QUERY_PROMPT_TMPL
        )
        self._output_parser = VectorStoreQueryOutputParser()
        self._prompt = PromptTemplate(template=prompt_template_str)

        # additional config
        self._max_top_k = max_top_k
        self._similarity_top_k = similarity_top_k
        self._empty_query_top_k = empty_query_top_k
        self._vector_store_query_mode = vector_store_query_mode
        # if extra_filters is OR condition, we don't support that yet
        if extra_filters is not None and extra_filters.condition == FilterCondition.OR:
            raise ValueError("extra_filters cannot be OR condition")
        self._extra_filters = extra_filters or MetadataFilters(filters=[])
        self._kwargs = kwargs
        super().__init__(
            callback_manager=callback_manager,
            object_map=object_map or self._index._object_map,
            objects=objects,
            verbose=verbose,
        )

    def _get_prompts(self) -> PromptDictType:
        """获取提示。"""
        return {
            "prompt": self._prompt,
        }

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """获取提示模块。"""
        if "prompt" in prompts:
            self._prompt = prompts["prompt"]

    def _get_query_bundle(self, query: str) -> QueryBundle:
        """获取查询包。"""
        if not query and self._default_empty_query_vector is not None:
            return QueryBundle(
                query_str="",
                embedding=self._default_empty_query_vector,
            )
        else:
            return QueryBundle(query_str=query)

    def _parse_generated_spec(
        self, output: str, query_bundle: QueryBundle
    ) -> BaseModel:
        """解析生成的规范。"""
        try:
            structured_output = cast(
                StructuredOutput, self._output_parser.parse(output)
            )
            query_spec = cast(VectorStoreQuerySpec, structured_output.parsed_output)
        except OutputParserException:
            _logger.warning("Failed to parse query spec, using defaults as fallback.")
            query_spec = VectorStoreQuerySpec(
                query=query_bundle.query_str,
                filters=[],
                top_k=None,
            )

        return query_spec

    def generate_retrieval_spec(
        self, query_bundle: QueryBundle, **kwargs: Any
    ) -> BaseModel:
        # prepare input
        info_str = self._vector_store_info.json(indent=4)
        schema_str = VectorStoreQuerySpec.schema_json(indent=4)

        # call LLM
        output = self._llm.predict(
            self._prompt,
            schema_str=schema_str,
            info_str=info_str,
            query_str=query_bundle.query_str,
        )

        # parse output
        return self._parse_generated_spec(output, query_bundle)

    async def agenerate_retrieval_spec(
        self, query_bundle: QueryBundle, **kwargs: Any
    ) -> BaseModel:
        # prepare input
        info_str = self._vector_store_info.json(indent=4)
        schema_str = VectorStoreQuerySpec.schema_json(indent=4)

        # call LLM
        output = await self._llm.apredict(
            self._prompt,
            schema_str=schema_str,
            info_str=info_str,
            query_str=query_bundle.query_str,
        )

        # parse output
        return self._parse_generated_spec(output, query_bundle)

    def _build_retriever_from_spec(
        self, spec: VectorStoreQuerySpec
    ) -> Tuple[BaseRetriever, QueryBundle]:
        # construct new query bundle from query_spec
        # insert 0 vector if query is empty and default_empty_query_vector is not None
        new_query_bundle = self._get_query_bundle(spec.query)

        _logger.info(f"Using query str: {spec.query}")
        filter_list = [
            (filter.key, filter.operator.value, filter.value) for filter in spec.filters
        ]
        _logger.info(f"Using filters: {filter_list}")
        if self._verbose:
            print(f"Using query str: {spec.query}")
            print(f"Using filters: {filter_list}")

        # define similarity_top_k
        # if query is specified, then use similarity_top_k
        # if query is blank, then use empty_query_top_k
        if spec.query or self._empty_query_top_k is None:
            similarity_top_k = self._similarity_top_k
        else:
            similarity_top_k = self._empty_query_top_k

        # if query_spec.top_k is specified, then use it
        # as long as below max_top_k and similarity_top_k
        if spec.top_k is not None:
            similarity_top_k = min(spec.top_k, self._max_top_k, similarity_top_k)

        _logger.info(f"Using top_k: {similarity_top_k}")

        # avoid passing empty filters to retriever
        if len(spec.filters) + len(self._extra_filters.filters) == 0:
            filters = None
        else:
            filters = MetadataFilters(
                filters=[*spec.filters, *self._extra_filters.filters]
            )

        return (
            VectorIndexRetriever(
                self._index,
                filters=filters,
                similarity_top_k=similarity_top_k,
                vector_store_query_mode=self._vector_store_query_mode,
                object_map=self.object_map,
                verbose=self._verbose,
                **self._kwargs,
            ),
            new_query_bundle,
        )