Jaguar

JaguarReader

Bases: BaseReader

Jaguar 读取器。 从现有的持久化的Jaguar存储中检索文档。

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

class JaguarReader(BaseReader):
    """Jaguar 读取器。
    从现有的持久化的Jaguar存储中检索文档。"""

    def __init__(
        self,
        pod: str,
        store: str,
        vector_index: str,
        vector_type: str,
        vector_dimension: int,
        url: str,
    ):
        """JaguarReader的构造函数。

Args:
    pod: pod的名称（数据库）
    store: pod中向量存储的名称
    vector_index: 存储的向量索引的名称
    vector_type: 向量索引的类型
    vector_dimension: 向量索引的维度
    url: jaguar http服务器的终端URL
"""
        self._pod = pod
        self._store = store
        self._vector_index = vector_index
        self._vector_type = vector_type
        self._vector_dimension = vector_dimension
        self._jag = JaguarHttpClient(url)
        self._token = ""

    def login(
        self,
        jaguar_api_key: Optional[str] = "",
    ) -> bool:
        """使用 jaguar_api_key 登录到 jaguar 服务器，或者让 self._jag 找到一个密钥。

Args:
    可选的 jaguar_api_key (str): 用户到 jaguardb 服务器的 API 密钥。
    如果未提供，将从环境变量 JAGUAR_API_KEY 或文件 $HOME/.jagrc 中读取 jaguar API 密钥。
Returns:
    如果成功则返回 True；如果不成功则返回 False。
"""
        if jaguar_api_key == "":
            jaguar_api_key = self._jag.getApiKey()
        self._jaguar_api_key = jaguar_api_key
        self._token = self._jag.login(jaguar_api_key)
        if self._token == "":
            return False
        return True

    def logout(self) -> None:
        """从jaguar服务器注销以清理资源。

Args:无参数
Returns:无
"""
        self._jag.logout(self._token)

    def load_data(
        self,
        embedding: Optional[List[float]] = None,
        k: int = 10,
        metadata_fields: Optional[List[str]] = None,
        where: Optional[str] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """从jaguar向量存储中加载数据。

Args:
    embedding: 用于向量的浮点数列表。如果给定，返回topk相似的文档。
    k: 要返回的结果数量。
    where: "a = '100' or ( b > 100 and c < 200 )"
           如果未提供embedding，则在metadata_fields中查找列的值和文本值。
    metadata_fields: 可选[List[str]] 要加载的元数据字段列表，除了文本文档之外。

Returns:
    文档列表
"""
        if embedding is not None:
            return self._load_similar_data(
                embedding=embedding,
                k=k,
                metadata_fields=metadata_fields,
                where=where,
                **kwargs,
            )
        else:
            return self._load_store_data(
                k=k, metadata_fields=metadata_fields, where=where, **kwargs
            )

    def _load_similar_data(
        self,
        embedding: List[float],
        k: int = 10,
        metadata_fields: Optional[List[str]] = None,
        where: Optional[str] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """从jaguar存储中通过相似性搜索加载数据。"""
        ### args is additional search conditions, such as time decay
        args = kwargs.get("args", None)
        fetch_k = kwargs.get("fetch_k", -1)

        vcol = self._vector_index
        vtype = self._vector_type
        str_embeddings = [str(f) for f in embedding]
        qv_comma = ",".join(str_embeddings)
        podstore = self._pod + "." + self._store
        q = (
            "select similarity("
            + vcol
            + ",'"
            + qv_comma
            + "','topk="
            + str(k)
            + ",fetch_k="
            + str(fetch_k)
            + ",type="
            + vtype
        )
        q += ",with_score,with_text"
        if args is not None:
            q += "," + args

        if metadata_fields is not None:
            x = "&".join(metadata_fields)
            q += ",metadata=" + x

        q += "') from " + podstore

        if where is not None:
            q += " where " + where

        jarr = self.run(q)
        if jarr is None:
            return []

        docs = []
        for js in jarr:
            score = js["score"]
            text = js["text"]
            zid = js["zid"]

            md = {}
            md["zid"] = zid
            md["score"] = score
            if metadata_fields is not None:
                for m in metadata_fields:
                    md[m] = js[m]

            doc = Document(
                id_=zid,
                text=text,
                metadata=md,
            )
            docs.append(doc)

        return docs

    def _load_store_data(
        self,
        k: int = 10,
        metadata_fields: Optional[List[str]] = None,
        where: Optional[str] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """从jaguar存储中加载一定数量的文档。"""
        vcol = self._vector_index
        podstore = self._pod + "." + self._store
        txtcol = vcol + ":text"

        sel_str = "zid," + txtcol
        if metadata_fields is not None:
            sel_str += "," + ",".join(metadata_fields)

        q = "select " + sel_str
        q += " from " + podstore

        if where is not None:
            q += " where " + where
        q += " limit " + str(k)

        jarr = self.run(q)
        if jarr is None:
            return []

        docs = []
        for ds in jarr:
            js = json.loads(ds)
            text = js[txtcol]
            zid = js["zid"]

            md = {}
            md["zid"] = zid
            if metadata_fields is not None:
                for m in metadata_fields:
                    md[m] = js[m]

            doc = Document(
                id_=zid,
                text=text,
                metadata=md,
            )
            docs.append(doc)

        return docs

    def run(self, query: str) -> dict:
        """在jaguardb中运行任何查询语句。

Args:
    query (str): 要发送到jaguardb的查询语句
Returns:
    无效令牌时返回None，或者
    json结果字符串
"""
        if self._token == "":
            return {}

        resp = self._jag.post(query, self._token, False)
        txt = resp.text
        try:
            return json.loads(txt)
        except Exception:
            return {}

    def prt(self, msg: str) -> None:
        nows = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open("/tmp/debugjaguarrdr.log", "a") as file:
            print(f"{nows} msg={msg}", file=file, flush=True)

login

login(jaguar_api_key: Optional[str] = '') -> bool

使用 jaguar_api_key 登录到 jaguar 服务器，或者让 self._jag 找到一个密钥。

Parameters:

Name	Type	Description	Default
可选的	jaguar_api_key (str	用户到 jaguardb 服务器的 API 密钥。	required

Returns: 如果成功则返回 True；如果不成功则返回 False。

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

    def login(
        self,
        jaguar_api_key: Optional[str] = "",
    ) -> bool:
        """使用 jaguar_api_key 登录到 jaguar 服务器，或者让 self._jag 找到一个密钥。

Args:
    可选的 jaguar_api_key (str): 用户到 jaguardb 服务器的 API 密钥。
    如果未提供，将从环境变量 JAGUAR_API_KEY 或文件 $HOME/.jagrc 中读取 jaguar API 密钥。
Returns:
    如果成功则返回 True；如果不成功则返回 False。
"""
        if jaguar_api_key == "":
            jaguar_api_key = self._jag.getApiKey()
        self._jaguar_api_key = jaguar_api_key
        self._token = self._jag.login(jaguar_api_key)
        if self._token == "":
            return False
        return True

logout

logout() -> None

从jaguar服务器注销以清理资源。

Args:无参数 Returns:无

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

    def logout(self) -> None:
        """从jaguar服务器注销以清理资源。

Args:无参数
Returns:无
"""
        self._jag.logout(self._token)

load_data

load_data(
    embedding: Optional[List[float]] = None,
    k: int = 10,
    metadata_fields: Optional[List[str]] = None,
    where: Optional[str] = None,
    **kwargs: Any
) -> List[Document]

从jaguar向量存储中加载数据。

Parameters:

Name	Type	Description	Default
embedding	Optional[List[float]]	用于向量的浮点数列表。如果给定，返回topk相似的文档。	None
k	int	要返回的结果数量。	10
where	Optional[str]	"a = '100' or ( b > 100 and c < 200 )" 如果未提供embedding，则在metadata_fields中查找列的值和文本值。	None
metadata_fields	Optional[List[str]]	可选[List[str]] 要加载的元数据字段列表，除了文本文档之外。	None

Returns:

Type	Description
List[Document]	文档列表

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

    def load_data(
        self,
        embedding: Optional[List[float]] = None,
        k: int = 10,
        metadata_fields: Optional[List[str]] = None,
        where: Optional[str] = None,
        **kwargs: Any,
    ) -> List[Document]:
        """从jaguar向量存储中加载数据。

Args:
    embedding: 用于向量的浮点数列表。如果给定，返回topk相似的文档。
    k: 要返回的结果数量。
    where: "a = '100' or ( b > 100 and c < 200 )"
           如果未提供embedding，则在metadata_fields中查找列的值和文本值。
    metadata_fields: 可选[List[str]] 要加载的元数据字段列表，除了文本文档之外。

Returns:
    文档列表
"""
        if embedding is not None:
            return self._load_similar_data(
                embedding=embedding,
                k=k,
                metadata_fields=metadata_fields,
                where=where,
                **kwargs,
            )
        else:
            return self._load_store_data(
                k=k, metadata_fields=metadata_fields, where=where, **kwargs
            )

run

run(query: str) -> dict

在jaguardb中运行任何查询语句。

Parameters:

Name	Type	Description	Default
query	str	要发送到jaguardb的查询语句	required

Returns: 无效令牌时返回None，或者 json结果字符串

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

    def run(self, query: str) -> dict:
        """在jaguardb中运行任何查询语句。

Args:
    query (str): 要发送到jaguardb的查询语句
Returns:
    无效令牌时返回None，或者
    json结果字符串
"""
        if self._token == "":
            return {}

        resp = self._jag.post(query, self._token, False)
        txt = resp.text
        try:
            return json.loads(txt)
        except Exception:
            return {}