Google

GoogleVectorStore

Bases: BasePydanticVectorStore

谷歌生成AI向量存储。

目前，它在服务器端计算嵌入向量。

示例

google_vector_store = GoogleVectorStore.from_corpus( corpus_id="my-corpus-id", include_metadata=True, metadata_keys=['file_name', 'creation_date'] ) index = VectorStoreIndex.from_vector_store( vector_store=google_vector_store )

属性

corpus_id: 此向量存储实例将读取和写入的语料库ID。 include_metadata (bool): 指示是否在查询结果中包含自定义元数据。默认为False。 metadata_keys (Optional[List[str]]): 如果include_metadata设置为True，则指定要在查询结果中包含哪些元数据键。如果为None，则包括所有元数据键。默认为None。

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

class GoogleVectorStore(BasePydanticVectorStore):
    """谷歌生成AI向量存储。

    目前，它在服务器端计算嵌入向量。

    示例:
        google_vector_store = GoogleVectorStore.from_corpus(
            corpus_id="my-corpus-id",
            include_metadata=True,
            metadata_keys=['file_name', 'creation_date']
        )
        index = VectorStoreIndex.from_vector_store(
            vector_store=google_vector_store
        )

    属性:
        corpus_id: 此向量存储实例将读取和写入的语料库ID。
        include_metadata (bool): 指示是否在查询结果中包含自定义元数据。默认为False。
        metadata_keys (Optional[List[str]]): 如果include_metadata设置为True，则指定要在查询结果中包含哪些元数据键。如果为None，则包括所有元数据键。默认为None。"""

    # Semantic Retriever stores the document node's text as string and embeds
    # the vectors on the server automatically.
    stores_text: bool = True
    is_embedding_query: bool = False

    # This is not the Google's corpus name but an ID generated in the LlamaIndex
    # world.
    corpus_id: str = Field(frozen=True)
    """此实例向量存储正在使用的语料库ID。"""

    # Configuration options for handling metadata in query results
    include_metadata: bool = False
    metadata_keys: Optional[List[str]] = None

    _client: Any = PrivateAttr()

    def __init__(self, *, client: Any, **kwargs: Any):
        """原始构造函数。

请使用类方法 `from_corpus` 或 `create_corpus`。

Args:
    client: 来自 google.ai.generativelanguage 的低级检索器类。
"""
        try:
            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        super().__init__(**kwargs)

        assert isinstance(client, genai.RetrieverServiceClient)
        self._client = client

    @classmethod
    def from_corpus(
        cls,
        *,
        corpus_id: str,
        include_metadata: bool = False,
        metadata_keys: Optional[List[str]] = None,
    ) -> "GoogleVectorStore":
        """创建一个指向现有语料库的实例。

Args:
    corpus_id (str): Google服务器上现有语料库的ID。
    include_metadata (bool, optional): 指定是否在查询结果中包含自定义元数据。默认为False，表示不包括元数据。
    metadata_keys (Optional[List[str]], optional): 如果include_metadata设置为True，则指定要在查询结果中包含哪些元数据键。如果为None，则包括所有元数据键。默认为None。

Returns:
    指向指定语料库的向量存储实例。

抛出:
    如果找不到这样的语料库，则引发NoSuchCorpusException异常。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.from_corpus(corpus_id={corpus_id})")
        client = genaix.build_semantic_retriever()
        if genaix.get_corpus(corpus_id=corpus_id, client=client) is None:
            raise NoSuchCorpusException(corpus_id=corpus_id)

        return cls(
            corpus_id=corpus_id,
            client=client,
            include_metadata=include_metadata,
            metadata_keys=metadata_keys,
        )

    @classmethod
    def create_corpus(
        cls, *, corpus_id: Optional[str] = None, display_name: Optional[str] = None
    ) -> "GoogleVectorStore":
        """创建一个指向新创建的语料库的实例。

示例：
    store = GoogleVectorStore.create_corpus()
    print(f"创建了ID为{store.corpus_id}的语料库")

    store = GoogleVectorStore.create_corpus(
        display_name="我的第一个语料库"
    )

    store = GoogleVectorStore.create_corpus(
        corpus_id="my-corpus-1",
        display_name="我的第一个语料库"
    )

Args:
    corpus_id：要创建的新语料库的ID。如果未提供，Google服务器将为您提供一个。
    display_name：语料库的标题。如果未提供，Google服务器将为您提供一个。

Returns:
    指向指定语料库的向量存储的实例。

引发：
    如果语料库已经存在或用户达到配额限制，则引发异常。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(
            f"\n\nGoogleVectorStore.create_corpus(new_corpus_id={corpus_id}, new_display_name={display_name})"
        )

        client = genaix.build_semantic_retriever()
        new_corpus_id = corpus_id or str(uuid.uuid4())
        new_corpus = genaix.create_corpus(
            corpus_id=new_corpus_id, display_name=display_name, client=client
        )
        name = genaix.EntityName.from_str(new_corpus.name)
        return cls(corpus_id=name.corpus_id, client=client)

    @classmethod
    def class_name(cls) -> str:
        return "GoogleVectorStore"

    @property
    def client(self) -> Any:
        return self._client

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        """将带有嵌入式的节点添加到向量存储中。

如果节点具有源节点，则将使用源节点的ID创建文档。否则，将使用该语料库的默认文档来容纳节点。

此外，如果源节点具有元数据字段“file_name”，则将其用作文档的标题。如果源节点没有此类字段，则谷歌服务器将为文档分配一个标题。

示例：
    store = GoogleVectorStore.from_corpus(corpus_id="123")
    store.add([
        TextNode(
            text="Hello, my darling",
            relationships={
                NodeRelationship.SOURCE: RelatedNodeInfo(
                    node_id="doc-456",
                    metadata={"file_name": "Title for doc-456"},
                )
            },
        ),
        TextNode(
            text="Goodbye, my baby",
            relationships={
                NodeRelationship.SOURCE: RelatedNodeInfo(
                    node_id="doc-456",
                    metadata={"file_name": "Title for doc-456"},
                )
            },
        ),
    ])

上述代码将创建一个ID为`doc-456`、标题为`Title for doc-456`的文档。该文档将容纳这两个节点。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.add(nodes={nodes})")

        client = cast(genai.RetrieverServiceClient, self.client)

        created_node_ids: List[str] = []
        for nodeGroup in _group_nodes_by_source(nodes):
            source = nodeGroup.source_node
            document_id = source.node_id
            document = genaix.get_document(
                corpus_id=self.corpus_id, document_id=document_id, client=client
            )

            if not document:
                genaix.create_document(
                    corpus_id=self.corpus_id,
                    display_name=source.metadata.get("file_name", None),
                    document_id=document_id,
                    metadata=source.metadata,
                    client=client,
                )

            created_chunks = genaix.batch_create_chunk(
                corpus_id=self.corpus_id,
                document_id=document_id,
                texts=[node.get_content() for node in nodeGroup.nodes],
                metadatas=[node.metadata for node in nodeGroup.nodes],
                client=client,
            )
            created_node_ids.extend([chunk.name for chunk in created_chunks])

        return created_node_ids

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

底层节点和文档将从Google服务器中删除。

Args:
    ref_doc_id: 要删除的文档ID。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.delete(ref_doc_id={ref_doc_id})")

        client = cast(genai.RetrieverServiceClient, self.client)
        genaix.delete_document(
            corpus_id=self.corpus_id, document_id=ref_doc_id, client=client
        )

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询向量存储。

示例：
    store = GoogleVectorStore.from_corpus(corpus_id="123")
    store.query(
        query=VectorStoreQuery(
            query_str="生命的意义是什么？",
            # 仅限具有此作者的节点。
            filters=MetadataFilters(
                filters=[
                    ExactMatchFilter(
                        key="author",
                        value="Arthur Schopenhauer",
                    )
                ]
            ),
            # 仅来自这些文档。如果未提供，则搜索整个语料库。
            doc_ids=["doc-456"],
            similarity_top_k=3,
        )
    )

Args:
    query：参见`llama_index.core.vector_stores.types.VectorStoreQuery`。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.query(query={query})")

        query_str = query.query_str
        if query_str is None:
            raise ValueError("VectorStoreQuery.query_str should not be None.")

        client = cast(genai.RetrieverServiceClient, self.client)

        relevant_chunks: List[genai.RelevantChunk] = []
        if query.doc_ids is None:
            # The chunks from query_corpus should be sorted in reverse order by
            # relevant score.
            relevant_chunks = genaix.query_corpus(
                corpus_id=self.corpus_id,
                query=query_str,
                filter=_convert_filter(query.filters),
                k=query.similarity_top_k,
                client=client,
            )
        else:
            for doc_id in query.doc_ids:
                relevant_chunks.extend(
                    genaix.query_document(
                        corpus_id=self.corpus_id,
                        document_id=doc_id,
                        query=query_str,
                        filter=_convert_filter(query.filters),
                        k=query.similarity_top_k,
                        client=client,
                    )
                )
            # Make sure the chunks are reversed sorted according to relevant
            # scores even across multiple documents.
            relevant_chunks.sort(key=lambda c: c.chunk_relevance_score, reverse=True)

        nodes = []
        include_metadata = self.include_metadata
        metadata_keys = self.metadata_keys
        for chunk in relevant_chunks:
            metadata = {}
            if include_metadata:
                for custom_metadata in chunk.chunk.custom_metadata:
                    # Use getattr to safely extract values
                    value = getattr(custom_metadata, "string_value", None)
                    if (
                        value is None
                    ):  # If string_value is not set, check for numeric_value
                        value = getattr(custom_metadata, "numeric_value", None)
                    # Add to the metadata dictionary only those keys that are present in metadata_keys
                    if value is not None and (
                        metadata_keys is None or custom_metadata.key in metadata_keys
                    ):
                        metadata[custom_metadata.key] = value

            text_node = TextNode(
                text=chunk.chunk.data.string_value,
                id=_extract_chunk_id(chunk.chunk.name),
                metadata=metadata,  # Adding metadata to the node
            )
            nodes.append(text_node)

        return VectorStoreQueryResult(
            nodes=nodes,
            ids=[_extract_chunk_id(chunk.chunk.name) for chunk in relevant_chunks],
            similarities=[chunk.chunk_relevance_score for chunk in relevant_chunks],
        )

corpus_id class-attribute instance-attribute

corpus_id: str = Field(frozen=True)

此实例向量存储正在使用的语料库ID。

from_corpus classmethod

from_corpus(
    *,
    corpus_id: str,
    include_metadata: bool = False,
    metadata_keys: Optional[List[str]] = None
) -> GoogleVectorStore

创建一个指向现有语料库的实例。

Parameters:

Name	Type	Description	Default
corpus_id	str	Google服务器上现有语料库的ID。	required
include_metadata	bool	指定是否在查询结果中包含自定义元数据。默认为False，表示不包括元数据。	False
metadata_keys	Optional[List[str]]	如果include_metadata设置为True，则指定要在查询结果中包含哪些元数据键。如果为None，则包括所有元数据键。默认为None。	None

Returns:

Type	Description
GoogleVectorStore	指向指定语料库的向量存储实例。

抛出

如果找不到这样的语料库，则引发NoSuchCorpusException异常。

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

    @classmethod
    def from_corpus(
        cls,
        *,
        corpus_id: str,
        include_metadata: bool = False,
        metadata_keys: Optional[List[str]] = None,
    ) -> "GoogleVectorStore":
        """创建一个指向现有语料库的实例。

Args:
    corpus_id (str): Google服务器上现有语料库的ID。
    include_metadata (bool, optional): 指定是否在查询结果中包含自定义元数据。默认为False，表示不包括元数据。
    metadata_keys (Optional[List[str]], optional): 如果include_metadata设置为True，则指定要在查询结果中包含哪些元数据键。如果为None，则包括所有元数据键。默认为None。

Returns:
    指向指定语料库的向量存储实例。

抛出:
    如果找不到这样的语料库，则引发NoSuchCorpusException异常。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.from_corpus(corpus_id={corpus_id})")
        client = genaix.build_semantic_retriever()
        if genaix.get_corpus(corpus_id=corpus_id, client=client) is None:
            raise NoSuchCorpusException(corpus_id=corpus_id)

        return cls(
            corpus_id=corpus_id,
            client=client,
            include_metadata=include_metadata,
            metadata_keys=metadata_keys,
        )

create_corpus classmethod

create_corpus(
    *,
    corpus_id: Optional[str] = None,
    display_name: Optional[str] = None
) -> GoogleVectorStore

创建一个指向新创建的语料库的实例。

示例： store = GoogleVectorStore.create_corpus() print(f"创建了ID为{store.corpus_id}的语料库")

store = GoogleVectorStore.create_corpus(
    display_name="我的第一个语料库"
)

store = GoogleVectorStore.create_corpus(
    corpus_id="my-corpus-1",
    display_name="我的第一个语料库"
)

Returns:

Type	Description
GoogleVectorStore	指向指定语料库的向量存储的实例。

引发： 如果语料库已经存在或用户达到配额限制，则引发异常。

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

    @classmethod
    def create_corpus(
        cls, *, corpus_id: Optional[str] = None, display_name: Optional[str] = None
    ) -> "GoogleVectorStore":
        """创建一个指向新创建的语料库的实例。

示例：
    store = GoogleVectorStore.create_corpus()
    print(f"创建了ID为{store.corpus_id}的语料库")

    store = GoogleVectorStore.create_corpus(
        display_name="我的第一个语料库"
    )

    store = GoogleVectorStore.create_corpus(
        corpus_id="my-corpus-1",
        display_name="我的第一个语料库"
    )

Args:
    corpus_id：要创建的新语料库的ID。如果未提供，Google服务器将为您提供一个。
    display_name：语料库的标题。如果未提供，Google服务器将为您提供一个。

Returns:
    指向指定语料库的向量存储的实例。

引发：
    如果语料库已经存在或用户达到配额限制，则引发异常。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(
            f"\n\nGoogleVectorStore.create_corpus(new_corpus_id={corpus_id}, new_display_name={display_name})"
        )

        client = genaix.build_semantic_retriever()
        new_corpus_id = corpus_id or str(uuid.uuid4())
        new_corpus = genaix.create_corpus(
            corpus_id=new_corpus_id, display_name=display_name, client=client
        )
        name = genaix.EntityName.from_str(new_corpus.name)
        return cls(corpus_id=name.corpus_id, client=client)

add

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

将带有嵌入式的节点添加到向量存储中。

如果节点具有源节点，则将使用源节点的ID创建文档。否则，将使用该语料库的默认文档来容纳节点。

此外，如果源节点具有元数据字段“file_name”，则将其用作文档的标题。如果源节点没有此类字段，则谷歌服务器将为文档分配一个标题。

示例： store = GoogleVectorStore.from_corpus(corpus_id="123") store.add([ TextNode( text="Hello, my darling", relationships={ NodeRelationship.SOURCE: RelatedNodeInfo( node_id="doc-456", metadata={"file_name": "Title for doc-456"}, ) }, ), TextNode( text="Goodbye, my baby", relationships={ NodeRelationship.SOURCE: RelatedNodeInfo( node_id="doc-456", metadata={"file_name": "Title for doc-456"}, ) }, ), ])

上述代码将创建一个ID为doc-456、标题为Title for doc-456的文档。该文档将容纳这两个节点。

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

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        """将带有嵌入式的节点添加到向量存储中。

如果节点具有源节点，则将使用源节点的ID创建文档。否则，将使用该语料库的默认文档来容纳节点。

此外，如果源节点具有元数据字段“file_name”，则将其用作文档的标题。如果源节点没有此类字段，则谷歌服务器将为文档分配一个标题。

示例：
    store = GoogleVectorStore.from_corpus(corpus_id="123")
    store.add([
        TextNode(
            text="Hello, my darling",
            relationships={
                NodeRelationship.SOURCE: RelatedNodeInfo(
                    node_id="doc-456",
                    metadata={"file_name": "Title for doc-456"},
                )
            },
        ),
        TextNode(
            text="Goodbye, my baby",
            relationships={
                NodeRelationship.SOURCE: RelatedNodeInfo(
                    node_id="doc-456",
                    metadata={"file_name": "Title for doc-456"},
                )
            },
        ),
    ])

上述代码将创建一个ID为`doc-456`、标题为`Title for doc-456`的文档。该文档将容纳这两个节点。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.add(nodes={nodes})")

        client = cast(genai.RetrieverServiceClient, self.client)

        created_node_ids: List[str] = []
        for nodeGroup in _group_nodes_by_source(nodes):
            source = nodeGroup.source_node
            document_id = source.node_id
            document = genaix.get_document(
                corpus_id=self.corpus_id, document_id=document_id, client=client
            )

            if not document:
                genaix.create_document(
                    corpus_id=self.corpus_id,
                    display_name=source.metadata.get("file_name", None),
                    document_id=document_id,
                    metadata=source.metadata,
                    client=client,
                )

            created_chunks = genaix.batch_create_chunk(
                corpus_id=self.corpus_id,
                document_id=document_id,
                texts=[node.get_content() for node in nodeGroup.nodes],
                metadatas=[node.metadata for node in nodeGroup.nodes],
                client=client,
            )
            created_node_ids.extend([chunk.name for chunk in created_chunks])

        return created_node_ids

delete

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

根据ref_doc_id删除节点。

底层节点和文档将从Google服务器中删除。

Parameters:

Name	Type	Description	Default
ref_doc_id	str	要删除的文档ID。	required

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

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

底层节点和文档将从Google服务器中删除。

Args:
    ref_doc_id: 要删除的文档ID。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.delete(ref_doc_id={ref_doc_id})")

        client = cast(genai.RetrieverServiceClient, self.client)
        genaix.delete_document(
            corpus_id=self.corpus_id, document_id=ref_doc_id, client=client
        )

query

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

查询向量存储。

示例： store = GoogleVectorStore.from_corpus(corpus_id="123") store.query( query=VectorStoreQuery( query_str="生命的意义是什么？", # 仅限具有此作者的节点。 filters=MetadataFilters( filters=[ ExactMatchFilter( key="author", value="Arthur Schopenhauer", ) ] ), # 仅来自这些文档。如果未提供，则搜索整个语料库。 doc_ids=["doc-456"], similarity_top_k=3, ) )

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

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询向量存储。

示例：
    store = GoogleVectorStore.from_corpus(corpus_id="123")
    store.query(
        query=VectorStoreQuery(
            query_str="生命的意义是什么？",
            # 仅限具有此作者的节点。
            filters=MetadataFilters(
                filters=[
                    ExactMatchFilter(
                        key="author",
                        value="Arthur Schopenhauer",
                    )
                ]
            ),
            # 仅来自这些文档。如果未提供，则搜索整个语料库。
            doc_ids=["doc-456"],
            similarity_top_k=3,
        )
    )

Args:
    query：参见`llama_index.core.vector_stores.types.VectorStoreQuery`。
"""
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        _logger.debug(f"\n\nGoogleVectorStore.query(query={query})")

        query_str = query.query_str
        if query_str is None:
            raise ValueError("VectorStoreQuery.query_str should not be None.")

        client = cast(genai.RetrieverServiceClient, self.client)

        relevant_chunks: List[genai.RelevantChunk] = []
        if query.doc_ids is None:
            # The chunks from query_corpus should be sorted in reverse order by
            # relevant score.
            relevant_chunks = genaix.query_corpus(
                corpus_id=self.corpus_id,
                query=query_str,
                filter=_convert_filter(query.filters),
                k=query.similarity_top_k,
                client=client,
            )
        else:
            for doc_id in query.doc_ids:
                relevant_chunks.extend(
                    genaix.query_document(
                        corpus_id=self.corpus_id,
                        document_id=doc_id,
                        query=query_str,
                        filter=_convert_filter(query.filters),
                        k=query.similarity_top_k,
                        client=client,
                    )
                )
            # Make sure the chunks are reversed sorted according to relevant
            # scores even across multiple documents.
            relevant_chunks.sort(key=lambda c: c.chunk_relevance_score, reverse=True)

        nodes = []
        include_metadata = self.include_metadata
        metadata_keys = self.metadata_keys
        for chunk in relevant_chunks:
            metadata = {}
            if include_metadata:
                for custom_metadata in chunk.chunk.custom_metadata:
                    # Use getattr to safely extract values
                    value = getattr(custom_metadata, "string_value", None)
                    if (
                        value is None
                    ):  # If string_value is not set, check for numeric_value
                        value = getattr(custom_metadata, "numeric_value", None)
                    # Add to the metadata dictionary only those keys that are present in metadata_keys
                    if value is not None and (
                        metadata_keys is None or custom_metadata.key in metadata_keys
                    ):
                        metadata[custom_metadata.key] = value

            text_node = TextNode(
                text=chunk.chunk.data.string_value,
                id=_extract_chunk_id(chunk.chunk.name),
                metadata=metadata,  # Adding metadata to the node
            )
            nodes.append(text_node)

        return VectorStoreQueryResult(
            nodes=nodes,
            ids=[_extract_chunk_id(chunk.chunk.name) for chunk in relevant_chunks],
            similarities=[chunk.chunk_relevance_score for chunk in relevant_chunks],
        )