Skip to content

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,)