Source code for langchain_community.retrievers.zep

from __future__ import annotations

from enum import Enum
from typing import TYPE_CHECKING, Any, Dict, List, Optional

from langchain_core.callbacks import (
    AsyncCallbackManagerForRetrieverRun,
    CallbackManagerForRetrieverRun,
)
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from pydantic import model_validator

if TYPE_CHECKING:
    from zep_python.memory import MemorySearchResult



[docs]
class SearchScope(str, Enum):
    """Which documents to search. Messages or Summaries?"""

    messages = "messages"
    """Search chat history messages."""
    summary = "summary"
    """Search chat history summaries."""




[docs]
class SearchType(str, Enum):
    """Enumerator of the types of search to perform."""

    similarity = "similarity"
    """Similarity search."""
    mmr = "mmr"
    """Maximal Marginal Relevance reranking of similarity search."""




[docs]
class ZepRetriever(BaseRetriever):
    """`Zep` MemoryStore Retriever.

    Search your user's long-term chat history with Zep.

    Zep offers both simple semantic search and Maximal Marginal Relevance (MMR)
    reranking of search results.

    Note: You will need to provide the user's `session_id` to use this retriever.

    Args:
        url: URL of your Zep server (required)
        api_key: Your Zep API key (optional)
        session_id: Identifies your user or a user's session (required)
        top_k: Number of documents to return (default: 3, optional)
        search_type: Type of search to perform (similarity / mmr) (default: similarity,
                                                                    optional)
        mmr_lambda: Lambda value for MMR search. Defaults to 0.5 (optional)

    Zep - Fast, scalable building blocks for LLM Apps
    =========
    Zep is an open source platform for productionizing LLM apps. Go from a prototype
    built in LangChain or LlamaIndex, or a custom app, to production in minutes without
    rewriting code.

    For server installation instructions, see:
    https://docs.getzep.com/deployment/quickstart/
    """

    zep_client: Optional[Any] = None
    """Zep client."""
    url: str
    """URL of your Zep server."""
    api_key: Optional[str] = None
    """Your Zep API key."""
    session_id: str
    """Zep session ID."""
    top_k: Optional[int]
    """Number of items to return."""
    search_scope: SearchScope = SearchScope.messages
    """Which documents to search. Messages or Summaries?"""
    search_type: SearchType = SearchType.similarity
    """Type of search to perform (similarity / mmr)"""
    mmr_lambda: Optional[float] = None
    """Lambda value for MMR search."""

    @model_validator(mode="before")
    @classmethod
    def create_client(cls, values: dict) -> Any:
        try:
            from zep_python import ZepClient
        except ImportError:
            raise ImportError(
                "Could not import zep-python package. "
                "Please install it with `pip install zep-python`."
            )
        values["zep_client"] = values.get(
            "zep_client",
            ZepClient(base_url=values["url"], api_key=values.get("api_key")),
        )
        return values

    def _messages_search_result_to_doc(
        self, results: List[MemorySearchResult]
    ) -> List[Document]:
        return [
            Document(
                page_content=r.message.pop("content"),
                metadata={"score": r.dist, **r.message},
            )
            for r in results
            if r.message
        ]

    def _summary_search_result_to_doc(
        self, results: List[MemorySearchResult]
    ) -> List[Document]:
        return [
            Document(
                page_content=r.summary.content,
                metadata={
                    "score": r.dist,
                    "uuid": r.summary.uuid,
                    "created_at": r.summary.created_at,
                    "token_count": r.summary.token_count,
                },
            )
            for r in results
            if r.summary
        ]

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> List[Document]:
        from zep_python.memory import MemorySearchPayload

        if not self.zep_client:
            raise RuntimeError("Zep client not initialized.")

        payload = MemorySearchPayload(
            text=query,
            metadata=metadata,
            search_scope=self.search_scope,
            search_type=self.search_type,
            mmr_lambda=self.mmr_lambda,
        )

        results: List[MemorySearchResult] = self.zep_client.memory.search_memory(
            self.session_id, payload, limit=self.top_k
        )

        if self.search_scope == SearchScope.summary:
            return self._summary_search_result_to_doc(results)

        return self._messages_search_result_to_doc(results)

    async def _aget_relevant_documents(
        self,
        query: str,
        *,
        run_manager: AsyncCallbackManagerForRetrieverRun,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> List[Document]:
        from zep_python.memory import MemorySearchPayload

        if not self.zep_client:
            raise RuntimeError("Zep client not initialized.")

        payload = MemorySearchPayload(
            text=query,
            metadata=metadata,
            search_scope=self.search_scope,
            search_type=self.search_type,
            mmr_lambda=self.mmr_lambda,
        )

        results: List[MemorySearchResult] = await self.zep_client.memory.asearch_memory(
            self.session_id, payload, limit=self.top_k
        )

        if self.search_scope == SearchScope.summary:
            return self._summary_search_result_to_doc(results)

        return self._messages_search_result_to_doc(results)