Zep

ZepReader

Bases: BaseReader

Zep文档向量存储读取器。

Parameters:

Name	Type	Description	Default
api_url	str	Zep API URL	required
api_key	str	Zep API密钥，可选的	None

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

class ZepReader(BaseReader):
    """Zep文档向量存储读取器。

    Args:
        api_url (str): Zep API URL
        api_key (str): Zep API密钥，可选的"""

    def __init__(self, api_url: str, api_key: Optional[str] = None):
        """使用参数进行初始化。"""
        from zep_python import ZepClient

        self._api_url = api_url
        self._api_key = api_key
        self._client = ZepClient(base_url=api_url, api_key=api_key)

    def load_data(
        self,
        collection_name: str,
        query: Optional[str] = None,
        vector: Optional[List[float]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        top_k: Optional[int] = 5,
        separate_documents: Optional[bool] = True,
        include_values: Optional[bool] = True,
    ) -> List[Document]:
        """从Zep加载数据。

Args:
    collection_name（str）：Zep集合的名称。
    query（可选[str]）：查询字符串。如果vector为None，则需要此参数。
    vector（可选[List[float]]）：查询向量。如果query为None，则需要此参数。
    metadata（可选[Dict[str, Any]]）：要过滤的元数据。
    top_k（可选[int]）：要返回的结果数量。默认为5。
    separate_documents（可选[bool]）：是否返回每个检索条目的单独文档。默认为True。
    include_values（可选[bool]）：是否在响应中包含嵌入。默认为True。

Returns:
    List[Document]：文档列表。
"""
        if query is None and vector is None:
            raise ValueError("Either query or vector must be specified.")

        collection = self._client.document.get_collection(name=collection_name)
        response = collection.search(
            text=query, embedding=vector, limit=top_k, metadata=metadata
        )

        documents = [
            (
                Document(text=d.content, embedding=d.embedding)
                if include_values
                else Document(text=d.content)
            )
            for d in response
        ]

        if not separate_documents:
            text_list = [d.get_text() for d in documents]
            text = "\n\n".join(text_list)
            documents = [Document(text=text)]

        return documents

load_data

load_data(
    collection_name: str,
    query: Optional[str] = None,
    vector: Optional[List[float]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    top_k: Optional[int] = 5,
    separate_documents: Optional[bool] = True,
    include_values: Optional[bool] = True,
) -> List[Document]

从Zep加载数据。

Returns:

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

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

    def load_data(
        self,
        collection_name: str,
        query: Optional[str] = None,
        vector: Optional[List[float]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        top_k: Optional[int] = 5,
        separate_documents: Optional[bool] = True,
        include_values: Optional[bool] = True,
    ) -> List[Document]:
        """从Zep加载数据。

Args:
    collection_name（str）：Zep集合的名称。
    query（可选[str]）：查询字符串。如果vector为None，则需要此参数。
    vector（可选[List[float]]）：查询向量。如果query为None，则需要此参数。
    metadata（可选[Dict[str, Any]]）：要过滤的元数据。
    top_k（可选[int]）：要返回的结果数量。默认为5。
    separate_documents（可选[bool]）：是否返回每个检索条目的单独文档。默认为True。
    include_values（可选[bool]）：是否在响应中包含嵌入。默认为True。

Returns:
    List[Document]：文档列表。
"""
        if query is None and vector is None:
            raise ValueError("Either query or vector must be specified.")

        collection = self._client.document.get_collection(name=collection_name)
        response = collection.search(
            text=query, embedding=vector, limit=top_k, metadata=metadata
        )

        documents = [
            (
                Document(text=d.content, embedding=d.embedding)
                if include_values
                else Document(text=d.content)
            )
            for d in response
        ]

        if not separate_documents:
            text_list = [d.get_text() for d in documents]
            text = "\n\n".join(text_list)
            documents = [Document(text=text)]

        return documents