Merge pull request #3920 from zenml-io/feature/served-pipelines

Deployed pipelines

Author: stefannica

docs/book/getting-started/core-concepts.md

@@ -60,7 +60,7 @@ As seen in the image, a step might use the outputs from a previous step and thus
 
 Pipelines and steps are defined in code using Python _decorators_ or _classes_. This is where the core business logic and value of your work live, and you will spend most of your time defining these two things.
 
-Even though pipelines are simple Python functions, you are only allowed to call steps within this function. The inputs for steps called within a pipeline can either be the outputs of previous steps or alternatively, you can pass in values directly (as long as they're JSON-serializable).
+Even though pipelines are simple Python functions, you are only allowed to call steps within this function. The inputs for steps called within a pipeline can either be the outputs of previous steps or alternatively, you can pass in values directly or map them onto pipeline parameters (as long as they're JSON-serializable). Similarly, you can return values from a pipeline that are step outputs as long as they are JSON-serializable.
 
 ```python
 from zenml import pipeline
@@ -71,19 +71,19 @@ def my_pipeline():
     step_2(input_one="hello", input_two=output_step_one)
 
 @pipeline
-def agent_evaluation_pipeline():
+def agent_evaluation_pipeline(query: str = "What is machine learning?") -> str:
     """An AI agent evaluation pipeline."""
     prompt = "You are a helpful assistant. Please answer: {query}"
-    test_query = "What is machine learning?"
-    evaluation_result = evaluate_agent_response(prompt, test_query)
+    evaluation_result = evaluate_agent_response(prompt, query)
+    return evaluation_result
 ```
 
 Executing the Pipeline is as easy as calling the function that you decorated with the `@pipeline` decorator.
 
 ```python
 if __name__ == "__main__":
     my_pipeline()
-    agent_evaluation_pipeline()
+    agent_evaluation_pipeline(query="What is an LLM?")
 ```
 
 #### Artifacts
@@ -118,9 +118,11 @@ Once you have implemented your workflow by using the concepts described above, y
 
 #### Stacks & Components
 
-When you want to execute a pipeline run with ZenML, **Stacks** come into play. A **Stack** is a collection of **stack components**, where each component represents the respective configuration regarding a particular function in your MLOps pipeline, such as orchestration systems, artifact repositories, and model deployment platforms.
+When you want to execute a pipeline run with ZenML, **Stacks** come into play. A **Stack** is a collection of **stack components**, where each component represents the respective configuration regarding a particular function in your MLOps pipeline, such as pipeline orchestration or deployment systems, artifact repositories and container registries.
 
-For instance, if you take a close look at the default local stack of ZenML, you will see two components that are **required** in every stack in ZenML, namely an _orchestrator_ and an _artifact store_.
+Pipelines can be executed in two ways: in **batch mode** (traditional execution through an orchestrator) or in **online mode** (long-running HTTP servers that can be invoked via REST API calls). Deploying pipelines for online mode execution allows you to serve your ML workflows as real-time endpoints, making them accessible for live inference and interactive use cases.
+
+For instance, if you take a close look at the default local stack of ZenML, you will see two components that are **required** in every stack in ZenML, namely an _orchestrator_ and an _artifact store_. Additional components like _deployers_ can be added to enable specific functionality such as deploying pipelines as HTTP endpoints.
 
 ![ZenML running code on the Local Stack.](../.gitbook/assets/02_pipeline_local_stack.png)
 
@@ -130,16 +132,30 @@ Keep in mind that each one of these components is built on top of base abstracti
 
 #### Orchestrator
 
-An **Orchestrator** is a workhorse that coordinates all the steps to run in a pipeline. Since pipelines can be set up with complex combinations of steps with various asynchronous dependencies between them, the orchestrator acts as the component that decides what steps to run and when to run them.
+An **Orchestrator** is a workhorse that coordinates all the steps to run in a pipeline in batch mode. Since pipelines can be set up with complex combinations of steps with various asynchronous dependencies between them, the orchestrator acts as the component that decides what steps to run and when to run them.
 
 ZenML comes with a default _local orchestrator_ designed to run on your local machine. This is useful, especially during the exploration phase of your project. You don't have to rent a cloud instance just to try out basic things.
 
+#### Deployer
+
+A **Deployer** is a stack component that manages the deployment of pipelines as long-running HTTP servers useful for online mode execution. Unlike orchestrators that execute pipelines in batch mode, deployers can create and manage persistent services that wrap your pipeline in a web application, usually containerized, allowing it to be invoked through HTTP requests.
+
+ZenML comes with a _Docker deployer_ that can run deployments on your local machine as Docker containers, making it easy to test and develop real-time pipeline endpoints before moving to production infrastructure.
+
+#### Pipeline Run
+
+A **Pipeline Run** is a record of a pipeline execution. When you run a pipeline using an orchestrator, a pipeline run is created tracking information about the execution such as the status, the artifacts and metadata produced by the pipeline and all its steps. When a pipeline is deployed for online mode execution, a pipeline run is similarly created for every HTTP request made to it.
+
 #### Artifact Store
 
 An **Artifact Store** is a component that houses all data that passes through the pipeline as inputs and outputs. Each artifact that gets stored in the artifact store is tracked and versioned and this allows for extremely useful features like data caching, which speeds up your workflows.
 
 Similar to the orchestrator, ZenML comes with a default _local artifact store_ designed to run on your local machine. This is useful, especially during the exploration phase of your project. You don't have to set up a cloud storage system to try out basic things.
 
+#### Deployment
+
+A **Deployment** is a running instance of a pipeline deployed as an HTTP endpoint. When you deploy a pipeline using a deployer, it becomes a long-running service that can be invoked through REST API calls. Each HTTP request to a deployment triggers a new pipeline run, creating the same artifacts and metadata tracking as traditional batch pipeline executions. This enables real-time inference, interactive ML workflows, and seamless integration with web applications and external services.
+
 #### Flavor
 
 ZenML provides a dedicated base abstraction for each stack component type. These abstractions are used to develop solutions, called **Flavors**, tailored to specific use cases/tools. With ZenML installed, you get access to a variety of built-in and integrated Flavors for each component type, but users can also leverage the base abstractions to create their own custom flavors.

docs/book/how-to/steps-pipelines/advanced_features.md

@@ -628,8 +628,8 @@ This is particularly useful for steps that interact with external services or re
 Hooks allow you to execute custom code at specific points in the pipeline or step lifecycle:
 
 ```python
-def success_hook(step_name, step_output):
-    print(f"Step {step_name} completed successfully with output: {step_output}")
+def success_hook():
+    print(f"Step completed successfully")
 
 def failure_hook(exception: BaseException):
     print(f"Step failed with error: {str(exception)}")
@@ -639,6 +639,11 @@ def my_step():
     return 42
 ```
 
+The following conventions apply to hooks:
+
+* the success hook takes no arguments
+* the failure hook optionally takes a single `BaseException` typed argument
+
 You can also define hooks at the pipeline level to apply to all steps:
 
 ```python

docs/book/toc.md

@@ -55,6 +55,7 @@
 * [Templates](how-to/templates/templates.md)
 * [Dashboard](how-to/dashboard/dashboard-features.md)
 
+
 ## Reference
 
 * [Community & content](reference/community-and-content.md)

pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "distro>=1.6.0,<2.0.0",
     "docker~=7.1.0",
     "gitpython>=3.1.18,<4.0.0",
+    "jsonref",
     "packaging>=24.1",
     "psutil>=5.0.0",
     "pydantic>=2.0,<=2.11.9",
@@ -368,5 +369,6 @@ module = [
     "numba.*",
     "uvloop.*",
     "litellm",
+    "jsonref",
 ]
 ignore_missing_imports = true

scripts/install-zenml-dev.sh

@@ -2,6 +2,40 @@
 
 INTEGRATIONS=no
 PIP_ARGS=
+UPGRADE_ALL=no
+
+show_help() {
+    cat << EOF
+Usage: $0 [OPTIONS]
+
+Install ZenML in development mode with optional integrations.
+
+OPTIONS:
+    -i, --integrations yes|no    Install integrations (default: no)
+    -s, --system                 Install packages system-wide instead of in virtual environment
+    -u, --upgrade-all           Uninstall existing ZenML, clear caches, and install latest versions
+    -h, --help                  Show this help message
+
+EXAMPLES:
+    # Basic installation
+    $0
+    
+    # Install with integrations
+    $0 --integrations yes
+    
+    # Force reinstall with latest versions of all dependencies
+    $0 --upgrade-all --integrations yes
+    
+    # System-wide installation with latest versions
+    $0 --system --upgrade-all
+
+NOTES:
+    - The --upgrade-all flag will uninstall existing ZenML installation and clear all caches
+    - This ensures you get the latest compatible versions of all dependencies
+    - Use this when you want to refresh your environment with the newest packages
+
+EOF
+}
 
 parse_args () {
     while [ $# -gt 0 ]; do
@@ -15,8 +49,17 @@ parse_args () {
                 PIP_ARGS="--system"
                 shift # past argument
                 ;;
+            -u|--upgrade-all)
+                UPGRADE_ALL="yes"
+                shift # past argument
+                ;;
+            -h|--help)
+                show_help
+                exit 0
+                ;;
             -*|--*)
                 echo "Unknown option $1"
+                show_help
                 exit 1
                 ;;
             *)
@@ -26,12 +69,39 @@ parse_args () {
     done
 }
 
+clean_and_uninstall() {
+    echo "🧹 Cleaning existing ZenML installation and clearing caches..."
+    
+    # Uninstall ZenML (if installed) and clear pip cache
+    uv pip uninstall $PIP_ARGS zenml || true
+    
+    # Clear uv cache to ensure fresh downloads
+    uv cache clean || true
+    
+    # Clear pip cache as well (in case pip was used previously)
+    python -m pip cache purge 2>/dev/null || true
+    
+    echo "✅ Cleanup completed"
+}
+
 install_zenml() {
+    echo "📦 Installing ZenML in editable mode..."
+    
+    # Build upgrade arguments based on UPGRADE_ALL flag
+    upgrade_args=""
+    if [ "$UPGRADE_ALL" = "yes" ]; then
+        upgrade_args="--upgrade --force-reinstall"
+        echo "🔄 Using --upgrade --force-reinstall to get latest versions"
+    fi
+    
     # install ZenML in editable mode
-    uv pip install $PIP_ARGS -e ".[server,templates,terraform,secrets-aws,secrets-gcp,secrets-azure,secrets-hashicorp,s3fs,gcsfs,adlfs,dev,connectors-aws,connectors-gcp,connectors-azure,azureml,sagemaker,vertex]"
+    uv pip install $PIP_ARGS $upgrade_args -e ".[server,templates,terraform,secrets-aws,secrets-gcp,secrets-azure,secrets-hashicorp,s3fs,gcsfs,adlfs,dev,connectors-aws,connectors-gcp,connectors-azure,azureml,sagemaker,vertex]"
+    
+    echo "✅ ZenML installation completed"
 }
 
 install_integrations() {
+    echo "🔌 Installing ZenML integrations..."
 
     # figure out the python version
     python_version=$(python -c "import sys; print('.'.join(map(str, sys.version_info[:2])))")
@@ -54,18 +124,37 @@ install_integrations() {
         --output-file integration-requirements.txt \
         $ignore_integrations_args
 
-    # pin pyyaml>=6.0.1
-    echo "" >> integration-requirements.txt
-    echo "pyyaml>=6.0.1" >> integration-requirements.txt
-    echo "pyopenssl" >> integration-requirements.txt
-    echo "typing-extensions" >> integration-requirements.txt
+    # Handle package pins based on upgrade mode
+    if [ "$UPGRADE_ALL" = "yes" ]; then
+        echo "🔄 Using latest versions for integration dependencies"
+        # When upgrading, use minimum versions to allow latest compatible
+        echo "" >> integration-requirements.txt
+        echo "pyyaml>=6.0.1" >> integration-requirements.txt
+        echo "pyopenssl" >> integration-requirements.txt
+        echo "typing-extensions" >> integration-requirements.txt
+        echo "maison<2" >> integration-requirements.txt
+    else
+        # Original behavior with specific pins
+        echo "" >> integration-requirements.txt
+        echo "pyyaml>=6.0.1" >> integration-requirements.txt
+        echo "pyopenssl" >> integration-requirements.txt
+        echo "typing-extensions" >> integration-requirements.txt
+        echo "maison<2" >> integration-requirements.txt
+    fi
+    
     echo "-e .[server,templates,terraform,secrets-aws,secrets-gcp,secrets-azure,secrets-hashicorp,s3fs,gcsfs,adlfs,dev,connectors-aws,connectors-gcp,connectors-azure,azureml,sagemaker,vertex]" >> integration-requirements.txt
 
-    # workaround to make yamlfix work
-    echo "maison<2" >> integration-requirements.txt
+    # Build upgrade arguments based on UPGRADE_ALL flag
+    upgrade_args=""
+    if [ "$UPGRADE_ALL" = "yes" ]; then
+        upgrade_args="--upgrade --force-reinstall"
+        echo "🔄 Using --upgrade --force-reinstall for integration dependencies"
+    fi
 
-    uv pip install $PIP_ARGS -r integration-requirements.txt
+    uv pip install $PIP_ARGS $upgrade_args -r integration-requirements.txt
     rm integration-requirements.txt
+    
+    echo "✅ Integration installation completed"
 
     # https://github.com/Kludex/python-multipart/pull/166
     # There is an install conflict between multipart and python_multipart
@@ -83,7 +172,14 @@ export ZENML_ANALYTICS_OPT_IN=false
 
 parse_args "$@"
 
-python -m pip install --upgrade wheel pip uv
+# Clean and upgrade tooling packages if upgrading all
+if [ "$UPGRADE_ALL" = "yes" ]; then
+    echo "🚀 Upgrading all dependencies to latest versions..."
+    clean_and_uninstall
+    python -m pip install --upgrade --force-reinstall wheel pip uv
+else
+    python -m pip install --upgrade wheel pip uv
+fi
 
 install_zenml
 

src/zenml/analytics/enums.py

@@ -91,3 +91,9 @@ class AnalyticsEvent(str, Enum):
 
     # Server Settings
     SERVER_SETTINGS_UPDATED = "Server Settings Updated"
+
+    # Deployment
+    DEPLOY_PIPELINE = "Pipeline deployed"
+    CREATE_DEPLOYMENT = "Deployment created"
+    STOP_DEPLOYMENT = "Deployment stopped"
+    DELETE_DEPLOYMENT = "Deployment deleted"

src/zenml/artifact_stores/base_artifact_store.py

@@ -106,6 +106,10 @@ def _validate_path(self, path: str) -> None:
             IllegalOperationError: If the path is a local file and the server
                 is not configured to allow local file access.
         """
+        # Skip validation for memory:// URIs used in serving mode
+        if path.startswith("memory://"):
+            return
+
         if not self.allow_local_file_access and not io_utils.is_remote(path):
             raise IllegalOperationError(
                 "Files in a local artifact store cannot be accessed from the "
@@ -139,6 +143,11 @@ def _sanitize_potential_path(self, potential_path: Any) -> Any:
             # Neither string nor bytes, this is not a path
             return potential_path
 
+        # Preserve special in-memory scheme used by serving mode as-is
+        # to avoid treating it as a local filesystem path.
+        if isinstance(path, str) and path.startswith("memory://"):
+            return path
+
         if io_utils.is_remote(path):
             # If we have a remote path, replace windows path separators with
             # slashes

src/zenml/artifacts/utils.py

@@ -152,7 +152,21 @@ def _store_artifact_data_and_prepare_request(
         Artifact version request for the artifact data that was stored.
     """
     artifact_store = Client().active_stack.artifact_store
-    artifact_store.makedirs(uri)
+
+    # Detect in-memory materializer to avoid touching the artifact store.
+    # Local import to minimize import-time dependencies.
+    from zenml.materializers.in_memory_materializer import (
+        InMemoryMaterializer,
+    )
+
+    is_in_memory = issubclass(materializer_class, InMemoryMaterializer)
+
+    if not is_in_memory:
+        artifact_store.makedirs(uri)
+    else:
+        # Ensure URI clearly indicates in-memory storage and not the artifact store
+        if not uri.startswith("memory://"):
+            uri = f"memory://custom_artifacts/{name}/{uuid4()}"
 
     materializer = materializer_class(uri=uri, artifact_store=artifact_store)
     materializer.uri = materializer.uri.replace("\\", "/")
@@ -190,7 +204,7 @@ def _store_artifact_data_and_prepare_request(
         data_type=source_utils.resolve(data_type),
         content_hash=content_hash,
         project=Client().active_project.id,
-        artifact_store_id=artifact_store.id,
+        artifact_store_id=None if is_in_memory else artifact_store.id,
         visualizations=visualizations,
         has_custom_name=has_custom_name,
         save_type=save_type,

src/zenml/cli/__init__.py

@@ -2506,6 +2506,7 @@ def my_pipeline(...):
 from zenml.cli.base import *  # noqa
 from zenml.cli.code_repository import *  # noqa
 from zenml.cli.config import *  # noqa
+from zenml.cli.deployment import *  # noqa
 from zenml.cli.downgrade import *  # noqa
 from zenml.cli.feature import *  # noqa
 from zenml.cli.integration import *  # noqa