Singlestore

SingleStoreReader

Bases: BaseReader

SingleStore读取器。

Source code in llama_index/readers/singlestore/base.py

class SingleStoreReader(BaseReader):
    """SingleStore读取器。

    Args:
        scheme（str）：数据库方案。
        host（str）：数据库主机。
        port（str）：数据库端口。
        user（str）：数据库用户。
        password（str）：数据库密码。
        dbname（str）：数据库名称。
        table_name（str）：表名称。
        content_field（str）：内容字段。
        vector_field（str）：向量字段。"""

    def __init__(
        self,
        scheme: str,
        host: str,
        port: str,
        user: str,
        password: str,
        dbname: str,
        table_name: str,
        content_field: str = "text",
        vector_field: str = "embedding",
    ):
        """使用参数进行初始化。"""
        self.scheme = scheme
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.dbname = dbname
        self.table_name = table_name
        self.content_field = content_field
        self.vector_field = vector_field

        try:
            import pymysql

            pymysql.install_as_MySQLdb()
        except ImportError:
            pass

        self.DatabaseReader = DatabaseReader
        self.reader = self.DatabaseReader(
            scheme=self.scheme,
            host=self.host,
            port=self.port,
            user=self.user,
            password=self.password,
            dbname=self.dbname,
        )

    def load_data(self, search_embedding: str, top_k: int = 5) -> List[Document]:
        """从SingleStore加载数据。

Args:
    search_embedding (str): 要搜索的嵌入。
    top_k (int): 要返回的结果数量。

Returns:
    List[Document]: 文档列表。
"""
        query = f"""
        SELECT {self.content_field}, DOT_PRODUCT_F64({self.vector_field}, JSON_ARRAY_PACK_F64(\'{search_embedding}\')) AS score
        FROM {self.table_name}
        ORDER BY score
        DESC LIMIT {top_k}
        """

        return self.reader.load_data(query=query)

load_data

load_data(
    search_embedding: str, top_k: int = 5
) -> List[Document]

从SingleStore加载数据。

Parameters:

Name	Type	Description	Default
search_embedding	str	要搜索的嵌入。	required
top_k	int	要返回的结果数量。	5

Returns:

Type	Description
List[Document]	List[Document]: 文档列表。

Source code in llama_index/readers/singlestore/base.py

    def load_data(self, search_embedding: str, top_k: int = 5) -> List[Document]:
        """从SingleStore加载数据。

Args:
    search_embedding (str): 要搜索的嵌入。
    top_k (int): 要返回的结果数量。

Returns:
    List[Document]: 文档列表。
"""
        query = f"""
        SELECT {self.content_field}, DOT_PRODUCT_F64({self.vector_field}, JSON_ARRAY_PACK_F64(\'{search_embedding}\')) AS score
        FROM {self.table_name}
        ORDER BY score
        DESC LIMIT {top_k}
        """

        return self.reader.load_data(query=query)