Deployed pipelines

[stefannica]

Describe changes

This PR introduces pipeline deployment (i.e. serving a pipeline as a HTTP endpoint) capability to ZenML.

Deployer Stack Components

A new Deployer stack component type is introduced with 3 different implementations / flavors:

• docker: manages pipeline deployments as locally running Docker containers
• gcp: manages pipeline deployments as GCP Cloud Run instances
• aws: manages pipeline deployments as AWS App Runner instances

Add a Deployer to your stack to unlock the ability to deploy pipelines and snapshots:

Example:

zenml deployer register docker -f docker
zenml stack register docker-deployer -o default -a default -D docker --set
zenml pipeline deploy --name my-endpoint my_module.my_pipeline 
zenml pipeline snapshot deploy --name my-endpoint my-snapshot 

curl -X POST "http://localhost:8000/invoke"   -H "Content-Type: application/json"   -d '{"parameters": {"city": "Munich"}}'

or

zenml pipeline endpoint invoke my-endpoint --param-one=value-one --param-two=value-two

Deployments

Introducing Deployments as entities resulting from deploying pipelines. These are representations of the HTTP servers used to run pipelines in online / inference mode. They can be managed independently of pipelines and the active stack.

Examples:

zenml deployment list
zenml deployment describe my-endpoint
zenml deployment invoke my-endpoint --param-one=value-one --param-two=value-two
zenml deployment logs -f my-endpoint
zenml deployment deprovision my-endpoint

Pipeline Deployment devX

It is possible to deploy any existing pipeline without any modifications. However, the utility comes from having parameterized pipeline steps:

@pipeline
def weather_agent_pipeline(city: str = "London") -> str:
    weather_data = get_weather(city=city)
    result = analyze_weather_with_llm(weather_data=weather_data, city=city)
    return result

The pipeline steps that expect input parameters are those that dictate the signature of the HTTP endpoint:

curl -X POST "http://localhost:8000/invoke"   -H "Content-Type: application/json"   -d '{"parameters": {"city": "Munich"}}'

Two new pipeline-level hooks are supported for initialization and cleanup. These hooks are only executed once per deployment, during server startup and shutdown. The initialization hook may also take input parameters and return a value which is preserved as the global state:

def init_hook(**kwargs: Any) ->  openai.OpenAI:
    return  openai.OpenAI(**config)

@step
def analyze_weather_with_llm(weather_data: Dict[str, float], city: str) -> str:
    step_context = get_step_context()
    llm = step_context.pipeline_state
    ...

@pipeline(
    on_init=init_hook,
    on_init_kwargs={"url"="openapi.org/..."}
)
def weather_agent_pipeline(city: str = "London") -> str:
    weather_data = get_weather(city=city)
    result = analyze_weather_with_llm(weather_data=weather_data, city=city)
    return result

Pre-requisites

Please ensure you have done the following:

• I have read the CONTRIBUTING.md document.
• I have added tests to cover my changes.
• I have based my new branch on develop and the open PR is targeting develop. If your branch wasn't based on develop read Contribution guide on rebasing branch to develop.
• IMPORTANT: I made sure that my changes are reflected properly in the following resources:
  • ZenML Docs
  • Dashboard: Needs to be communicated to the frontend team.
  • Templates: Might need adjustments (that are not reflected in the template tests) in case of non-breaking changes and deprecations.
  • Projects: Depending on the version dependencies, different projects might get affected.

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Other (add details above)