Underground-Digital/Workflow-Engine

1from abc import ABC, abstractmethod2from collections.abc import Sequence3from typing import Any, Optional4 5from pydantic import BaseModel, Field6 7 8class Document(BaseModel):9    """Class for storing a piece of text and associated metadata."""10 11    page_content: str12 13    vector: Optional[list[float]] = None14 15    """Arbitrary metadata about the page content (e.g., source, relationships to other16        documents, etc.).17    """18    metadata: Optional[dict] = Field(default_factory=dict)19 20    provider: Optional[str] = "dify"21 22 23class BaseDocumentTransformer(ABC):24    """Abstract base class for document transformation systems.25 26    A document transformation system takes a sequence of Documents and returns a27    sequence of transformed Documents.28 29    Example:30        .. code-block:: python31 32            class EmbeddingsRedundantFilter(BaseDocumentTransformer, BaseModel):33                embeddings: Embeddings34                similarity_fn: Callable = cosine_similarity35                similarity_threshold: float = 0.9536 37                class Config:38                    arbitrary_types_allowed = True39 40                def transform_documents(41                    self, documents: Sequence[Document], **kwargs: Any42                ) -> Sequence[Document]:43                    stateful_documents = get_stateful_documents(documents)44                    embedded_documents = _get_embeddings_from_stateful_docs(45                        self.embeddings, stateful_documents46                    )47                    included_idxs = _filter_similar_embeddings(48                        embedded_documents, self.similarity_fn, self.similarity_threshold49                    )50                    return [stateful_documents[i] for i in sorted(included_idxs)]51 52                async def atransform_documents(53                    self, documents: Sequence[Document], **kwargs: Any54                ) -> Sequence[Document]:55                    raise NotImplementedError56 57    """58 59    @abstractmethod60    def transform_documents(self, documents: Sequence[Document], **kwargs: Any) -> Sequence[Document]:61        """Transform a list of documents.62 63        Args:64            documents: A sequence of Documents to be transformed.65 66        Returns:67            A list of transformed Documents.68        """69 70    @abstractmethod71    async def atransform_documents(self, documents: Sequence[Document], **kwargs: Any) -> Sequence[Document]:72        """Asynchronously transform a list of documents.73 74        Args:75            documents: A sequence of Documents to be transformed.76 77        Returns:78            A list of transformed Documents.79        """80