Merge branch 'feature/served-pipelines' of https://github.com/zenml-io/zenml into feature/served-pipelines

Author: safoinme

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.

src/zenml/client.py

@@ -3393,6 +3393,8 @@ def list_snapshots(
         schedule_id: Optional[Union[str, UUID]] = None,
         source_snapshot_id: Optional[Union[str, UUID]] = None,
         runnable: Optional[bool] = None,
+        deployable: Optional[bool] = None,
+        deployed: Optional[bool] = None,
         tag: Optional[str] = None,
         tags: Optional[List[str]] = None,
         hydrate: bool = False,
@@ -3418,6 +3420,8 @@ def list_snapshots(
             schedule_id: The ID of the schedule to filter by.
             source_snapshot_id: The ID of the source snapshot to filter by.
             runnable: Whether the snapshot is runnable.
+            deployable: Whether the snapshot is deployable.
+            deployed: Whether the snapshot is deployed.
             tag: Filter by tag.
             tags: Filter by tags.
             hydrate: Flag deciding whether to hydrate the output model(s)
@@ -3444,6 +3448,8 @@ def list_snapshots(
             schedule_id=schedule_id,
             source_snapshot_id=source_snapshot_id,
             runnable=runnable,
+            deployable=deployable,
+            deployed=deployed,
             tag=tag,
             tags=tags,
         )
@@ -3745,6 +3751,9 @@ def list_deployments(
         status: Optional[DeploymentStatus] = None,
         url: Optional[str] = None,
         user: Optional[Union[UUID, str]] = None,
+        pipeline: Optional[Union[UUID, str]] = None,
+        tag: Optional[str] = None,
+        tags: Optional[List[str]] = None,
         hydrate: bool = False,
     ) -> Page[DeploymentResponse]:
         """List deployments.
@@ -3764,6 +3773,9 @@ def list_deployments(
             status: The status of the deployment to filter by.
             url: The url of the deployment to filter by.
             user: Filter by user name/ID.
+            pipeline: Filter by pipeline name/ID.
+            tag: Tag to filter by.
+            tags: Tags to filter by.
             hydrate: Flag deciding whether to hydrate the output model(s)
                 by including metadata fields in the response.
 
@@ -3786,6 +3798,9 @@ def list_deployments(
                 deployer_id=deployer_id,
                 status=status,
                 url=url,
+                pipeline=pipeline,
+                tag=tag,
+                tags=tags,
             ),
             hydrate=hydrate,
         )
@@ -4632,6 +4647,7 @@ def list_pipeline_runs(
         hydrate: bool = False,
         include_full_metadata: bool = False,
         triggered_by_step_run_id: Optional[Union[UUID, str]] = None,
+        triggered_by_deployment_id: Optional[Union[UUID, str]] = None,
     ) -> Page[PipelineRunResponse]:
         """List all pipeline runs.
 
@@ -4678,6 +4694,8 @@ def list_pipeline_runs(
                 the response.
             triggered_by_step_run_id: The ID of the step run that triggered
                 the pipeline run.
+            triggered_by_deployment_id: The ID of the deployment that triggered
+                the pipeline run.
 
         Returns:
             A page with Pipeline Runs fitting the filter description
@@ -4719,6 +4737,7 @@ def list_pipeline_runs(
             in_progress=in_progress,
             templatable=templatable,
             triggered_by_step_run_id=triggered_by_step_run_id,
+            triggered_by_deployment_id=triggered_by_deployment_id,
         )
         return self.zen_store.list_runs(
             runs_filter_model=runs_filter_model,

src/zenml/deployers/base_deployer.py

@@ -174,7 +174,7 @@ def _check_deployment_inputs_outputs(
 
         Raises:
             DeploymentProvisionError: if the deployment has no compiled schemas
-            for the pipeline inputs and outputs.
+                for the pipeline inputs and outputs.
         """
         if (
             not snapshot.pipeline_spec

src/zenml/deployers/containerized_deployer.py

@@ -19,7 +19,9 @@
     Set,
 )
 
+import zenml
 from zenml.config.build_configuration import BuildConfiguration
+from zenml.config.global_config import GlobalConfiguration
 from zenml.constants import (
     DEPLOYER_DOCKER_IMAGE_KEY,
 )
@@ -69,6 +71,12 @@ def requirements(self) -> Set[str]:
         """
         requirements = super().requirements
         requirements.update(self.CONTAINER_REQUIREMENTS)
+
+        if self.config.is_local and GlobalConfiguration().uses_sql_store:
+            # If we're directly connected to a DB, we need to install the
+            # `local` extra in the Docker image to include the DB dependencies.
+            requirements.add(f"'zenml[local]=={zenml.__version__}'")
+
         return requirements
 
     def get_docker_builds(

src/zenml/deployers/docker/docker_deployer.py

@@ -54,12 +54,10 @@
 )
 from zenml.deployers.server.entrypoint_configuration import (
     AUTH_KEY_OPTION,
+    DEPLOYMENT_ID_OPTION,
     PORT_OPTION,
     DeploymentEntrypointConfiguration,
 )
-from zenml.entrypoints.base_entrypoint_configuration import (
-    SNAPSHOT_ID_OPTION,
-)
 from zenml.enums import DeploymentStatus, StackComponentType
 from zenml.logger import get_logger
 from zenml.models import (
@@ -251,7 +249,6 @@ def _get_container_operational_state(
             state.url = "http://localhost"
             if metadata.port:
                 state.url += f":{metadata.port}"
-            # TODO: check if the deployment is healthy.
 
         return state
 
@@ -304,7 +301,7 @@ def do_provision_deployment(
         entrypoint = DeploymentEntrypointConfiguration.get_entrypoint_command()
 
         entrypoint_kwargs = {
-            SNAPSHOT_ID_OPTION: snapshot.id,
+            DEPLOYMENT_ID_OPTION: deployment.id,
             PORT_OPTION: 8000,
         }
         if deployment.auth_key:

src/zenml/deployers/server/app.py

@@ -27,7 +27,6 @@
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-from pydantic import BaseModel
 
 from zenml.deployers.server.models import (
     ExecutionMetrics,
@@ -65,24 +64,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # Startup
     logger.info("🚀 Starting ZenML Pipeline Serving service...")
 
-    snapshot_id = os.getenv("ZENML_SNAPSHOT_ID")
-    if not snapshot_id:
-        raise ValueError("ZENML_SNAPSHOT_ID environment variable is required")
+    deployment_id = os.getenv("ZENML_DEPLOYMENT_ID")
+    if not deployment_id:
+        raise ValueError(
+            "ZENML_DEPLOYMENT_ID environment variable is required"
+        )
 
     try:
         global _service
-        _service = PipelineDeploymentService(snapshot_id)
+        _service = PipelineDeploymentService(deployment_id)
         _service.initialize()
-        # params model is available.
-        try:
-            params_model = _service.params_model
-            if isinstance(params_model, type) and issubclass(
-                params_model, BaseModel
-            ):
-                app.include_router(_build_invoke_router(_service))
-        except Exception:
-            # Skip router installation if parameter model is not ready
-            pass
+        app.include_router(_build_invoke_router(_service))
         logger.info("✅ Pipeline deployment service initialized successfully")
     except Exception as e:
         logger.error(f"❌ Failed to initialize: {e}")
@@ -107,7 +99,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
 # Create FastAPI application with OpenAPI security scheme
 app = FastAPI(
-    title="ZenML Pipeline Deployment",
+    title=f"ZenML Pipeline Deployment {os.getenv('ZENML_DEPLOYMENT_ID')}",
     description="deploy ZenML pipelines as FastAPI endpoints",
     version="0.2.0",
     lifespan=lifespan,
@@ -346,8 +338,8 @@ def runtime_error_handler(request: Request, exc: RuntimeError) -> JSONResponse:
 
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        "--snapshot_id",
-        default=os.getenv("ZENML_SNAPSHOT_ID"),
+        "--deployment_id",
+        default=os.getenv("ZENML_DEPLOYMENT_ID"),
         help="Pipeline snapshot ID",
     )
     parser.add_argument(
@@ -371,8 +363,8 @@ def runtime_error_handler(request: Request, exc: RuntimeError) -> JSONResponse:
     )
     args = parser.parse_args()
 
-    if args.snapshot_id:
-        os.environ["ZENML_SNAPSHOT_ID"] = args.snapshot_id
+    if args.deployment_id:
+        os.environ["ZENML_DEPLOYMENT_ID"] = args.deployment_id
     if args.auth_key:
         os.environ["ZENML_DEPLOYMENT_AUTH_KEY"] = args.auth_key