泽普

Zep阅读器 #

基类：EventBaseReader

Zep文档向量存储读取器。

参数：

名称	类型	描述	默认
`api_url`	`str`	Zep API 网址	required
`api_key`	`str`	Zep API密钥，可选	`None`

workflows/handler.py 中的源代码llama_index/readers/zep/base.py

class ZepReader(BaseReader):
    """
    Zep document vector store reader.

    Args:
        api_url (str): Zep API URL
        api_key (str): Zep API key, optional

    """

    def __init__(self, api_url: str, api_key: Optional[str] = None):
        """Initialize with parameters."""
        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]:
        """
        Load data from Zep.

        Args:
            collection_name (str): Name of the Zep collection.
            query (Optional[str]): Query string. Required if vector is None.
            vector (Optional[List[float]]): Query vector. Required if query is None.
            metadata (Optional[Dict[str, Any]]): Metadata to filter on.
            top_k (Optional[int]): Number of results to return. Defaults to 5.
            separate_documents (Optional[bool]): Whether to return separate
                documents per retrieved entry. Defaults to True.
            include_values (Optional[bool]): Whether to include the embedding in
                the response. Defaults to True.

        Returns:
            List[Document]: A list of documents.

        """
        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[文档]

从 Zep 加载数据。

参数：

名称	类型	描述	默认
`collection_name`	`str`	Zep 集合的名称。	required
`query`	`Optional[str]`	查询字符串。如果向量为 None 则必须提供。	`None`
`vector`	`Optional[List[float]]`	查询向量。如果查询为空则必需。	`None`
`metadata`	`Optional[Dict[str, Any]]`	用于筛选的元数据。	`None`
`top_k`	`Optional[int]`	返回的结果数量。默认为5。	`5`
`separate_documents`	`Optional[bool]`	是否对每个检索到的条目返回单独的文档。默认为 True。	`True`
`include_values`	`Optional[bool]`	是否在响应中包含嵌入向量。默认为 True。	`True`

返回：

类型	描述
`List[文档]`	List[Document]: 文档列表。

workflows/handler.py 中的源代码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]:
    """
    Load data from Zep.

    Args:
        collection_name (str): Name of the Zep collection.
        query (Optional[str]): Query string. Required if vector is None.
        vector (Optional[List[float]]): Query vector. Required if query is None.
        metadata (Optional[Dict[str, Any]]): Metadata to filter on.
        top_k (Optional[int]): Number of results to return. Defaults to 5.
        separate_documents (Optional[bool]): Whether to return separate
            documents per retrieved entry. Defaults to True.
        include_values (Optional[bool]): Whether to include the embedding in
            the response. Defaults to True.

    Returns:
        List[Document]: A list of documents.

    """
    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

选项：成员：- ZepReader