Typesense

TypesenseVectorStore

Bases: BasePydanticVectorStore

Typesense向量存储。

在这个向量存储中，嵌入和文档都存储在Typesense索引中。

在查询时，索引使用Typesense查询前k个最相似的节点。

Parameters:

Name	Type	Description	Default
client	Any	Typesense客户端	required
tokenizer	Optional[Callable[[str], List]]	分词器函数。	None

示例

pip install llama-index-vector-stores-typesense

from llama_index.vector_stores.typesense import TypesenseVectorStore
from typesense import Client

# 注册Typesense并获取API密钥
typesense_client = Client(
    {
        "api_key": "your_api_key_here",
        "nodes": [{"host": "localhost", "port": "8108", "protocol": "http"}],
        "connection_timeout_seconds": 2,
    }
)

# 创建TypesenseVectorStore的实例
vector_store = TypesenseVectorStore(typesense_client)

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

class TypesenseVectorStore(BasePydanticVectorStore):
    """Typesense向量存储。

在这个向量存储中，嵌入和文档都存储在Typesense索引中。

在查询时，索引使用Typesense查询前k个最相似的节点。

Args:
    client (Any): Typesense客户端
    tokenizer (Optional[Callable[[str], List]]): 分词器函数。

示例:
    `pip install llama-index-vector-stores-typesense`

    ```python
    from llama_index.vector_stores.typesense import TypesenseVectorStore
    from typesense import Client

    # 注册Typesense并获取API密钥
    typesense_client = Client(
        {
            "api_key": "your_api_key_here",
            "nodes": [{"host": "localhost", "port": "8108", "protocol": "http"}],
            "connection_timeout_seconds": 2,
        }
    )

    # 创建TypesenseVectorStore的实例
    vector_store = TypesenseVectorStore(typesense_client)
    ```"""

    stores_text: bool = True
    is_embedding_query: bool = False
    flat_metadata: bool = False

    _tokenizer: Callable[[str], List] = PrivateAttr()
    _text_key: str = PrivateAttr()
    _collection_name: str = PrivateAttr()
    _collection: Any = PrivateAttr()
    _batch_size: int = PrivateAttr()
    _metadata_key: str = PrivateAttr()
    _client: typesense.Client = PrivateAttr()

    def __init__(
        self,
        client: Any,
        tokenizer: Optional[Callable[[str], List]] = None,
        text_key: str = DEFAULT_TEXT_KEY,
        collection_name: str = DEFAULT_COLLECTION_NAME,
        batch_size: int = DEFAULT_BATCH_SIZE,
        metadata_key: str = DEFAULT_METADATA_KEY,
        **kwargs: Any,
    ) -> None:
        """初始化参数。"""
        super().__init__()

        if client is not None:
            if not isinstance(client, typesense.Client):
                raise ValueError(
                    f"client should be an instance of typesense.Client, "
                    f"got {type(client)}"
                )
            self._client = cast(typesense.Client, client)
        self._tokenizer = tokenizer or get_tokenizer()
        self._text_key = text_key
        self._collection_name = collection_name
        self._collection = self._client.collections[self._collection_name]
        self._batch_size = batch_size
        self._metadata_key = metadata_key

    @classmethod
    def class_name(cls) -> str:
        """类名。"""
        return "TypesenseVectorStore"

    @property
    def client(self) -> Any:
        """返回Typesense客户端。"""
        return self._client

    @property
    def collection(self) -> Any:
        """返回Typesense集合。"""
        return self._collection

    def _create_collection(self, num_dim: int) -> None:
        fields = [
            {"name": "vec", "type": "float[]", "num_dim": num_dim},
            {"name": f"{self._text_key}", "type": "string"},
            {"name": ".*", "type": "auto"},
        ]
        self._client.collections.create(
            {"name": self._collection_name, "fields": fields}
        )

    def _create_upsert_docs(self, nodes: List[BaseNode]) -> List[dict]:
        upsert_docs = []
        for node in nodes:
            doc = {
                "id": node.node_id,
                "vec": node.get_embedding(),
                f"{self._text_key}": node.get_content(metadata_mode=MetadataMode.NONE),
                "ref_doc_id": node.ref_doc_id,
                f"{self._metadata_key}": node_to_metadata_dict(
                    node, remove_text=True, flat_metadata=self.flat_metadata
                ),
            }
            upsert_docs.append(doc)

        return upsert_docs

    @staticmethod
    def _to_typesense_filter(standard_filters: MetadataFilters) -> str:
        """从标准数据类转换为typesense过滤器字典。"""
        for filter in standard_filters.legacy_filters():
            if filter.key == "filter_by":
                return str(filter.value)

        return ""

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

Args:
    节点: List[BaseNode]: 带有嵌入的节点列表
"""
        docs = self._create_upsert_docs(nodes)

        try:
            collection = cast(Collection, self.collection)
            collection.documents.import_(
                docs, {"action": "upsert"}, batch_size=self._batch_size
            )
        except ObjectNotFound:
            # Create the collection if it doesn't already exist
            num_dim = len(nodes[0].get_embedding())
            self._create_collection(num_dim)
            collection.documents.import_(
                docs, {"action": "upsert"}, batch_size=self._batch_size
            )

        return [node.node_id for node in nodes]

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

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        collection = cast(Collection, self.collection)
        collection.documents.delete({"filter_by": f"ref_doc_id:={ref_doc_id}"})

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询Typesense索引以获取最相似的前k个节点。

Args:
    query（VectorStoreQuery）：向量存储查询对象。
"""
        if query.filters:
            typesense_filter = self._to_typesense_filter(query.filters)
        else:
            typesense_filter = ""

        if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
            if query.query_embedding:
                embedded_query = [str(x) for x in query.query_embedding]
                search_requests = {
                    "searches": [
                        {
                            "collection": self._collection_name,
                            "q": "*",
                            "vector_query": f'vec:([{",".join(embedded_query)}],'
                            + f"k:{query.similarity_top_k})",
                            "filter_by": typesense_filter,
                        }
                    ]
                }
            else:
                raise ValueError("Vector search requires a query embedding")
        if query.mode is VectorStoreQueryMode.TEXT_SEARCH:
            if query.query_str:
                search_requests = {
                    "searches": [
                        {
                            "collection": self._collection_name,
                            "q": query.query_str,
                            "query_by": self._text_key,
                            "filter_by": typesense_filter,
                        }
                    ]
                }
            else:
                raise ValueError("Text search requires a query string")
        response = self._client.multi_search.perform(search_requests, {})

        top_k_nodes = []
        top_k_ids = []
        top_k_scores = None
        if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
            top_k_scores = []

        for hit in response["results"][0]["hits"]:
            document = hit["document"]
            id = document["id"]
            text = document[self._text_key]

            # Note that typesense distances range from 0 to 2, \
            # where 0 is most similar and 2 is most dissimilar
            if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
                score = hit["vector_distance"]

            try:
                node = metadata_dict_to_node(document[self._metadata_key])
                node.text = text
            except Exception:
                extra_info, node_info, relationships = legacy_metadata_dict_to_node(
                    document[self._metadata_key], text_key=self._text_key
                )
                node = TextNode(
                    text=text,
                    id_=id,
                    metadata=extra_info,
                    start_chart_idx=node_info.get("start", None),
                    end_chart_idx=node_info.get("end", None),
                    relationships=relationships,
                )

            top_k_ids.append(id)
            top_k_nodes.append(node)
            if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
                top_k_scores.append(score)

        return VectorStoreQueryResult(
            nodes=top_k_nodes, similarities=top_k_scores, ids=top_k_ids
        )

client property

client: Any

返回Typesense客户端。

collection property

collection: Any

返回Typesense集合。

class_name classmethod

class_name() -> str

类名。

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

@classmethod
def class_name(cls) -> str:
    """类名。"""
    return "TypesenseVectorStore"

add

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

将节点添加到索引中。

Parameters:

Name	Type	Description	Default
节点		List[BaseNode]: 带有嵌入的节点列表	required

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

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

Args:
    节点: List[BaseNode]: 带有嵌入的节点列表
"""
        docs = self._create_upsert_docs(nodes)

        try:
            collection = cast(Collection, self.collection)
            collection.documents.import_(
                docs, {"action": "upsert"}, batch_size=self._batch_size
            )
        except ObjectNotFound:
            # Create the collection if it doesn't already exist
            num_dim = len(nodes[0].get_embedding())
            self._create_collection(num_dim)
            collection.documents.import_(
                docs, {"action": "upsert"}, batch_size=self._batch_size
            )

        return [node.node_id for node in nodes]

delete

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

使用ref_doc_id删除节点。

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

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

Args:
    ref_doc_id（str）：要删除的文档的doc_id。
"""
        collection = cast(Collection, self.collection)
        collection.documents.delete({"filter_by": f"ref_doc_id:={ref_doc_id}"})

query

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

查询Typesense索引以获取最相似的前k个节点。

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

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """查询Typesense索引以获取最相似的前k个节点。

Args:
    query（VectorStoreQuery）：向量存储查询对象。
"""
        if query.filters:
            typesense_filter = self._to_typesense_filter(query.filters)
        else:
            typesense_filter = ""

        if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
            if query.query_embedding:
                embedded_query = [str(x) for x in query.query_embedding]
                search_requests = {
                    "searches": [
                        {
                            "collection": self._collection_name,
                            "q": "*",
                            "vector_query": f'vec:([{",".join(embedded_query)}],'
                            + f"k:{query.similarity_top_k})",
                            "filter_by": typesense_filter,
                        }
                    ]
                }
            else:
                raise ValueError("Vector search requires a query embedding")
        if query.mode is VectorStoreQueryMode.TEXT_SEARCH:
            if query.query_str:
                search_requests = {
                    "searches": [
                        {
                            "collection": self._collection_name,
                            "q": query.query_str,
                            "query_by": self._text_key,
                            "filter_by": typesense_filter,
                        }
                    ]
                }
            else:
                raise ValueError("Text search requires a query string")
        response = self._client.multi_search.perform(search_requests, {})

        top_k_nodes = []
        top_k_ids = []
        top_k_scores = None
        if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
            top_k_scores = []

        for hit in response["results"][0]["hits"]:
            document = hit["document"]
            id = document["id"]
            text = document[self._text_key]

            # Note that typesense distances range from 0 to 2, \
            # where 0 is most similar and 2 is most dissimilar
            if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
                score = hit["vector_distance"]

            try:
                node = metadata_dict_to_node(document[self._metadata_key])
                node.text = text
            except Exception:
                extra_info, node_info, relationships = legacy_metadata_dict_to_node(
                    document[self._metadata_key], text_key=self._text_key
                )
                node = TextNode(
                    text=text,
                    id_=id,
                    metadata=extra_info,
                    start_chart_idx=node_info.get("start", None),
                    end_chart_idx=node_info.get("end", None),
                    relationships=relationships,
                )

            top_k_ids.append(id)
            top_k_nodes.append(node)
            if query.mode is not VectorStoreQueryMode.TEXT_SEARCH:
                top_k_scores.append(score)

        return VectorStoreQueryResult(
            nodes=top_k_nodes, similarities=top_k_scores, ids=top_k_ids
        )