Langchain

zenml.integrations.langchain special

Initialization of the langchain integration.

LangchainIntegration (Integration)

Definition of langchain integration for ZenML.

Source code in zenml/integrations/langchain/__init__.py

class LangchainIntegration(Integration):
    """Definition of langchain integration for ZenML."""

    NAME = LANGCHAIN
    REQUIREMENTS = [
        "langchain==0.0.325",
        "pyyaml>=6.0.1",
        "tenacity!=8.4.0",  # https://github.com/jd/tenacity/issues/471
    ]
    REQUIREMENTS_IGNORED_ON_UNINSTALL = ["pyyaml","tenacity"]

    @classmethod
    def activate(cls) -> None:
        """Activates the integration."""
        from zenml.integrations.langchain import materializers  # noqa

activate() classmethod

Activates the integration.

Source code in zenml/integrations/langchain/__init__.py

@classmethod
def activate(cls) -> None:
    """Activates the integration."""
    from zenml.integrations.langchain import materializers  # noqa

materializers special

Initialization of the langchain materializer.

document_materializer

Implementation of ZenML's Langchain Document materializer.

LangchainDocumentMaterializer (BaseMaterializer)

Handle Langchain Document objects.

Source code in zenml/integrations/langchain/materializers/document_materializer.py

class LangchainDocumentMaterializer(BaseMaterializer):
    """Handle Langchain Document objects."""

    ASSOCIATED_ARTIFACT_TYPE: ClassVar[ArtifactType] = ArtifactType.DATA
    ASSOCIATED_TYPES: ClassVar[Tuple[Type[Any], ...]] = (Document,)

    def load(self, data_type: Type["Document"]) -> Any:
        """Reads BaseModel from JSON.

        Args:
            data_type: The type of the data to read.

        Returns:
            The data read.
        """
        data_path = os.path.join(self.uri, DEFAULT_FILENAME)
        contents = yaml_utils.read_json(data_path)
        return data_type.parse_raw(contents)

    def save(self, data: "Document") -> None:
        """Serialize a BaseModel to JSON.

        Args:
            data: The data to store.
        """
        data_path = os.path.join(self.uri, DEFAULT_FILENAME)
        yaml_utils.write_json(data_path, data.json())

    def extract_metadata(self, data: Document) -> Dict[str, "MetadataType"]:
        """Extract metadata from the given BaseModel object.

        Args:
            data: The BaseModel object to extract metadata from.

        Returns:
            The extracted metadata as a dictionary.
        """
        return {"schema": data.schema()}

extract_metadata(self, data)

Extract metadata from the given BaseModel object.

Parameters:

Name	Type	Description	Default
data	langchain.docstore.document.Document	The BaseModel object to extract metadata from.	required

Returns:

Type	Description
Dict[str, MetadataType]	The extracted metadata as a dictionary.

Source code in zenml/integrations/langchain/materializers/document_materializer.py

def extract_metadata(self, data: Document) -> Dict[str, "MetadataType"]:
    """Extract metadata from the given BaseModel object.

    Args:
        data: The BaseModel object to extract metadata from.

    Returns:
        The extracted metadata as a dictionary.
    """
    return {"schema": data.schema()}

load(self, data_type)

Reads BaseModel from JSON.

Parameters:

Name	Type	Description	Default
data_type	Type[Document]	The type of the data to read.	required

Returns:

Type	Description
Any	The data read.

Source code in zenml/integrations/langchain/materializers/document_materializer.py

def load(self, data_type: Type["Document"]) -> Any:
    """Reads BaseModel from JSON.

    Args:
        data_type: The type of the data to read.

    Returns:
        The data read.
    """
    data_path = os.path.join(self.uri, DEFAULT_FILENAME)
    contents = yaml_utils.read_json(data_path)
    return data_type.parse_raw(contents)

save(self, data)

Serialize a BaseModel to JSON.

Parameters:

Name	Type	Description	Default
data	Document	The data to store.	required

Source code in zenml/integrations/langchain/materializers/document_materializer.py

def save(self, data: "Document") -> None:
    """Serialize a BaseModel to JSON.

    Args:
        data: The data to store.
    """
    data_path = os.path.join(self.uri, DEFAULT_FILENAME)
    yaml_utils.write_json(data_path, data.json())

openai_embedding_materializer

Implementation of the Langchain OpenAI embedding materializer.

LangchainOpenaiEmbeddingMaterializer (CloudpickleMaterializer)

Handle langchain OpenAI embedding objects.

Source code in zenml/integrations/langchain/materializers/openai_embedding_materializer.py

class LangchainOpenaiEmbeddingMaterializer(CloudpickleMaterializer):
    """Handle langchain OpenAI embedding objects."""

    ASSOCIATED_ARTIFACT_TYPE: ClassVar[ArtifactType] = ArtifactType.MODEL
    ASSOCIATED_TYPES: ClassVar[Tuple[Type[Any], ...]] = (OpenAIEmbeddings,)

vector_store_materializer

Implementation of the langchain vector store materializer.

LangchainVectorStoreMaterializer (CloudpickleMaterializer)

Handle langchain vector store objects.

Source code in zenml/integrations/langchain/materializers/vector_store_materializer.py

class LangchainVectorStoreMaterializer(CloudpickleMaterializer):
    """Handle langchain vector store objects."""

    ASSOCIATED_ARTIFACT_TYPE: ClassVar[ArtifactType] = ArtifactType.DATA
    ASSOCIATED_TYPES: ClassVar[Tuple[Type[Any], ...]] = (VectorStore,)