Boost up a Helpdesk Chatbot with Dialogflow CX, TFX and Vertex AI
written in collaboration with Gabriele Randelli
All right, you talk, I’ll listen!
Do you have any ideas about Data Science on Google Cloud you would like to see? Please let me know by filling out this form. This would help me on future blog posts =)
Disclaimer
This article assumes some prior knowledge of TFX, Vertex AI and Dialogflow. I would recommend watching these videos from Google Cloud Developer advocacy team for a quick refresh of main concepts:
- What exactly is this TFX thing? (TensorFlow Extended) and How do TFX pipelines work? (TensorFlow Extended) by Robert Crowe
- Introduction to MLOps and Vertex Pipelines and What is Dialogflow CX? by Priyanka Vergadia
Premises
Recently I was discussing with Gabriele Randelli about this idea to integrate Vertex AI with other services of Google Cloud. We both proposed different scenarios (we would like to talk in the coming articles) but then we ended up with that question
Wouldn’t it be awesome to boost up a simple chatbot built with Dialogflow CX with a ML model trained on Vertex AI?
In this article, we’ll illustrate how to formalize a Tensorflow model training in order to run it via TFX on Vertex Pipelines¹. Then we will describe a simple chatbot using Dialogflow CX and we will learn how to integrate the trained model with it in order to provide predictions to users at the moment he/she interacts with the chatbot.
Our scenario
As one of the members of the Data science team I described in the previous article, your company provides a cloud-native application to assist its customers. But, after some time, customers provide bad feedback about long periods of waiting to receive call center assistance.
Because of the huge investment, the business decided to free some further resources and leverage ML capabilities in order to improve that customer care service. In particular, they believe that chatbots and NLP technologies have reached a good level of maturity for building business applications. That’s why they recently hired a conversational scientist and a conversation engineer. But, at the same time, they count on you to build a proof of concept with an AI powered chatbot.
As a KPI, the business would like to improve the containment rate² and consequently reduce waiting time improving overall customer care service. Lucky for all, Google Cloud, the cloud platform you migrated to, can help in targeting their KPI targets thanks to Dialogflow and Vertex AI.
With Dialogflow CX, the conversational team gets a comprehensive platform to design, build and test chatbots and voicebots. Along with all the key features, they appreciate that it has a Visual flow builder and it integrates the latest BERT-based natural language understanding (NLU) models. In this way, they can collaborate and focus on providing the best conversational experience because It will be the model that recognizes intent and context accurately and efficiently in more complex use cases.
At the same time, you need an ML pipeline framework that allows you to easily deploy advanced Tensorflow models on scale. The Tensorflow Extended (TFX for short) platform enables you to deploy automated pipelines in both test and production environments. It provides pre-built components for common steps such as data ingestion, data validation, and training. Every step is clearly defined, it has a clear set of inputs and outputs and it is specifically designed for scalable, high-performance machine learning tasks. And you know what, Vertex AI Pipeline natively supports its TFX SDK. You can define your ML pipelines using Python with TFX, and then execute your pipelines on Vertex AI in a serverless fashion.
Our dataset
With the aim of simulating the scenario, we decide to use the Helpdesk Issue dataset. The dataset is composed of several variables such as the level of seniority and experience of the support person and the category or the type of the issue. The target variable is represented by the resolution time which is the amount of time required to solve a particular issue. Below you can see a view of the Helpdesk issue dataset we used.
When TFX meets Dialogflow CX: Our ML system of the helpdesk chatbot
We started evaluating the architecture and the components we would like to use if someone was required to implement such a system in real-life. We assumed that
- The Tensorflow model would not be of production quality
We really want to focus on understanding if and how a data scientist needs to create a TF preprocessing and training tasks in order to consume it in the corresponding TFX components.
- The TFX pipeline would be as dry as possible
We assume that the ML engineer adopts the framework for the first time and we really want to stay simple in a POC scenario like this one.
- The scenario is not about building a bot.
That’s why we are re-using a Helpdesk Chatbot with Dialogflow & BigQuery ML Dialogflow project. But we adapt the fulfilment code in order to consume predictions generated by a model deployed in an Vertex AI Endpoint.
With that said, let’s dive into details of the ML system we implemented.
Preprocess. Transform, Train. Validate. Push. Repeat.
As we mentioned before, the TFX pipeline we implemented has the key components³:
- An example_gen consumes csv files and generates training and evaluation splits as tf.Example messages based on the output_config we set (⅔ training, ⅓ evaluation).
- A statistics_gen generates featured statistics over both training and serving data which allows us to validate variable distributions along samples and review some computed statistics for data processing and data analysis.
- The schema_gen uses the output of the statistics_gen component to infer a schema i.e. data types for feature values, value ranges, and other properties associated with the dataset⁴.
- The transform allows us to apply missing imputation logic, generate the vocabulary and apply it for categorical imputation and scale numerical variables to have range [0,1]. In order to do that we need to provide a user-defined preprocessing_fn. Here is one of the two components (the other is the trainer) when a data scientist needs to align his experimental code to the format required by the TFX component.
def preprocessing_fn(inputs):
"""tf.transform's callback function for preprocessing inputs.
We assume not using feature store in this example
Args:
inputs: map from feature keys to raw not-yet-transformed features.
Returns:
Map from string feature key to transformed feature operations.
"""
outputs = {}
# generate the vocabulary and apply it for categorical imputation
for key in _CATEGORICAL_VARIABLES:
outputs[key] = \
tft.compute_and_apply_vocabulary(_fill_in_missing(inputs[key]))
# scale numerical variables to have range [0,1]
for key in _NUMERICAL_VARIABLES:
outputs[key] = tft.scale_to_0_1(_fill_in_missing(inputs[key]))- As the previous component, the trainer needs a training module file defined by the user and aligned to the component requirements. For example, in our case, we build a custom Keras model and we use the GenericExecutor in the custom_executor_spec to train it within TFX natively. Then, based on the documentation, a run_fn function is required. Below you can find the one we define in our training module:
# TFX will call this function
def run_fn(fn_args):
"""A function to train the model based on given args.
Args:
fn_args: training args as name/value pairs.
"""
train_batch_size = 100
eval_batch_size = 100
# read the preprocessing data in order to speed up training process
tf_transform_output = \
tft.TFTransformOutput(fn_args.transform_output)
# ingest the transformed data and prepare training and evaluation dataset
train_input_fn = _input_fn(fn_args.train_files,
fn_args.data_accessor, tf_transform_output,
batch_size=train_batch_size) # pylint: disable=g-long-lambda
eval_input_fn = _input_fn(fn_args.eval_files,
fn_args.data_accessor, tf_transform_output,
batch_size=eval_batch_size) # pylint: disable=g-long-lambda
# build the model
model = _build_simple_dnn_regressor()
# define callback for TensorBoard
tensorboard_callback = tf.keras.callbacks.TensorBoard(
log_dir=fn_args.model_run_dir, update_freq='batch')
# fit the model
model.fit(
train_input_fn,
steps_per_epoch=fn_args.train_steps,
validation_data=eval_input_fn,
validation_steps=fn_args.eval_steps,
callbacks=[tensorboard_callback])
# read the feature spec to integrate in the serving function
feature_spec = tf_transform_output.raw_feature_spec()
features_input_signature = {
feature_name:
tf.TensorSpec(
shape=(None, 1), dtype=spec.dtype, name=feature_name
)
for feature_name, spec in feature_spec.items()
if feature_name in _FEATURE_NAMES
}
# define the signature
signatures = {
'serving_default':
_get_serve_tf_fn(model,
tf_transform_output).get_concrete_function(
features_input_signature)
}
# save the model
model.save(fn_args.serving_model_dir, save_format='tf', signatures=signatures)where
- _input_fn generates features and label
- _build_simple_dnn_regressor define feature columns and build the model
- _get_serve_tf_fn parses a serialized tf.Example and applies TFT
Notice that
- During the training, we consume the preprocessed data tf_transform_output to reduce the training time.
- Because we want to serve the model to a Vertex Endpoint, we also provide a custom serving_default signature that will integrate the Transform graph and will allow the model scores correctly.
By the end, the component consumes that module with the training and eval examples, the Protobuf definition of train args and eval args, the schema created by a schema_gen component and the transform graph produced by an upstream transform component and trains the model.
- The evaluator computes model performance metrics with the data from example_gen, the trained model from trainer, and the EvalConfig configuration and returns performance results and validation results. About EvalConfig, we explicit the model_spec to indicate the target variable and the serving signature. We also set the metrics_specs setting a GenericValueThreshold on Keras mean_absolute_error metric. Finally, we leave slicing_specs empty to consider the whole dataset. Once you run it, because the model’s accuracy meets the evaluation criteria, the model is “blessed” aka it can be deployed
- The pusher is the last step of the TFX pipeline. It just checks whether a model has passed validation and exports the model to _serving_model_dir consequently.
Below you can see the Pipeline class I defined with run method that allowed me to run it both on local and on Vertex AI pipelines
class Pipeline:
"""
The Pipeline class generalizes a pipeline definition in order to run it
both on prem and on cloud.
"""
def __init__(self,
runner: str,
pipeline_name: str,
pipeline_root: str,
data_root: str,
modules_path: str,
serving_model_dir: str,
metadata_path: str,
project_id=None,
region=None,
pipeline_definition=None):
self.pipeline = None
self.runner = runner
self.pipeline_name = pipeline_name
self.pipeline_root = pipeline_root
self.data_root = data_root
self.modules_path = modules_path
self.serving_model_dir = serving_model_dir
self.metadata_path = metadata_path
self.transform_filepath = f'{self.modules_path}/{TRANSFORM_MODULE}'
self.trainer_filepath = f'{self.modules_path}/{TRAINER_MODULE}'
self.train_num_steps = 2000
self.eval_num_steps = 1000
self.components = []
if self.runner == 'KubeflowRunner':
self.project_id = project_id
self.region = region
self.pipeline_definition = pipeline_definition
# ExampleGen
example_gen = tfx.components.CsvExampleGen(input_base=self.data_root,
output_config=EXAMPLE_GEN_OUTPUT_CONFIG)
# StatisticsGen
statistics_gen = tfx.components.StatisticsGen(
examples=example_gen.outputs['examples']
)
# SchemaGen
schema_gen = tfx.components.SchemaGen(
statistics=statistics_gen.outputs['statistics'],
infer_feature_shape=True)
# Transform
transform = tfx.components.Transform(
examples=example_gen.outputs['examples'],
schema=schema_gen.outputs['schema'],
module_file=self.transform_filepath)
# Trainer
trainer = tfx.components.Trainer(
module_file=self.trainer_filepath,
examples=transform.outputs['transformed_examples'],
schema=schema_gen.outputs['schema'],
transform_graph=transform.outputs['transform_graph'],
train_args=tfx.proto.TrainArgs(num_steps=self.train_num_steps),
eval_args=tfx.proto.EvalArgs(num_steps=self.eval_num_steps))
# Evaluator
evaluator = tfx.components.Evaluator(
examples=example_gen.outputs['examples'],
model=trainer.outputs['model'],
eval_config=EVALUATOR_CONFIG)
with conditional.Cond(evaluator.outputs['blessing'].future()
[0].custom_property('blessed') == 1):
# Pusher
pusher = tfx.components.Pusher(
model=trainer.outputs['model'],
model_blessing=evaluator.outputs['blessing'],
push_destination=tfx.proto.PushDestination(
filesystem=tfx.proto.PushDestination.Filesystem(
base_directory=self.serving_model_dir)))
# prepare components
self.components = [example_gen,
statistics_gen,
schema_gen,
transform,
trainer,
evaluator,
pusher]
# build pipeline
if self.runner == 'KubeflowRunner':
self.pipeline = tfx.dsl.Pipeline(
pipeline_name=self.pipeline_name,
pipeline_root=self.pipeline_root,
components=self.components)
else:
self.pipeline = tfx.dsl.Pipeline(
pipeline_name=self.pipeline_name,
pipeline_root=self.pipeline_root,
components=self.components,
metadata_connection_config=tfx.orchestration.metadata
.sqlite_metadata_connection_config(self.metadata_path),
beam_pipeline_args=BEAM_PIPELINE_ARGS_RUNNER[self.runner])
def run(self):
if self.runner == 'KubeflowRunner':
self.runner_instance = tfx.orchestration.experimental.KubeflowV2DagRunner(
config=tfx.orchestration.experimental.KubeflowV2DagRunnerConfig(),
output_filename=self.pipeline_definition)
_ = self.runner_instance.run(self.pipeline)
vertex_ai.init(project=self.project_id, location=self.region)
pipeline_job = vertex_ai.PipelineJob(
display_name=self.pipeline_name,
template_path=self.pipeline_definition,
pipeline_root=self.pipeline_root
)
pipeline_job.run(sync=False)
else:
self.runner_instance = tfx.orchestration.LocalDagRunner()
_ = self.runner_instance.run(self.pipeline)And here it is how the execution graph looks like once you submit the pipeline to Vertex AI
Finally, once we train the model, we can easily deploy using the Vertex Python SDK. Then it will be able to score instances passed by any source, even from a Dialogflow CX chatbot.
“Hi, I have a tech problem”
As mentioned previously, the aim of this article is not to design a complex and professional virtual agent, hence its structure has been kept deliberately simple.
Our virtual agent represents a typical customer-facing first level (L1) support agent, whose scope is to collect the customers’ support requests, acquire a basic description about the issue, finally submitting a support ticket and provide an estimation about the resolution time. In particular, within this specific scenario, the category of the issue is what the virtual agent asks for (e.g. performance, billing, technical, authentication).
With Dialogflow CX, we can easily build enterprise-grade conversational agents by designing conversational flows, each composed of multiple conversational rounds. Every utterance of the customer is automatically classified among a list of possible intents, which are defined inside the Dialogflow CX GUI without writing any code.
In our specific example, the agent is structured as a single conversational flow, composed of the following rounds:
- Greetings (virtual agent) and ticket request (customer)
- Collection of the issue information. In particular, the virtual agent asks for the issue category only
- Confirmation of the collected information and ticket submission, as well as reporting of the estimated resolution time.
- Conversation end
This sounds easy but how can the virtual agent infer the estimated resolution time?
In order to fulfill this request, the virtual agent needs to interact with our backend system to get the estimated resolution time for each specific issue and return it to the customer. With Dialogflow CX, this is enabled by webhooks.
Webhooks use the data extracted by Dialogflow CX’s natural language processing to generate dynamic responses, validate collected data, or trigger actions on the backend. In a nutshell, a webhook is a service to trigger HTTPS requests to backend systems.
How is the backend handling the requests coming from Dialogflow CX via webhooks?
A straightforward approach is to implement all the business logic to serve the inference stage with Google’s Cloud Functions, our function as a service (FaaS) framework. A major benefit of this architecture is to ensure a decoupling between the business logic itself and the machine learning service deployed with Vertex AI. Just as an example, in the future we could augment the same Cloud Function to fetch additional information from our CRM system without affecting the ML endpoint.
Out of all the supported programming languages, we opt for Node.js. One of the key reasons for this choice is that Vertex AI offers a very agile Node.js API.
The scaffold of a Cloud Function consists of the following components:
- index.js: the function implementation in Node.js
- package.json: a metadata file wrapping the function dependencies, managed with npm
The package.json notes our two dependencies: Vertex AI, to invoke the prediction of the estimated resolution time, and Dialogflow CX, to parse and manage the request coming from the virtual agent’s webhook:
"dependencies": {
"@google-cloud/dialogflow-cx": "^2.4.0",
"@google-cloud/aiplatform": "^1.12.0"
}As for the function itself, it is worth clarifying few aspects:
- In order to keep the virtual agent simple, as well as not to bother the customer with multiple questions, the virtual agent collects only the issue category (e.g. technical, billing, and so on). All the other features required from the ML endpoint (seniority, priority, experience, …) have been hardcoded. In the future, the best approach will be to fetch these additional features by querying the company’s back end systems
- To submit the request to the prediction endpoint, we rely on helpers to properly format the request payload. Helpers are very useful to get rid of representation transformations, which can be tricky in Node.js
- The most important aspect of the function implementation is the PredictionService in charge of online prediction and explainability requests
Below we report the prediction fragment of the Cloud Function implementation:
async function getPredictionRequest(category) {
// Configure the parent resource
const endpoint = `projects/${projectId}/locations/${location}/endpoints/${endpointId}`;
// Set empty parameters
const parameters = {
structValue: {// TODO
// fields: {category: {stringValue: category}}}
},
};// Set the instance value. For simplicity, we pass one input (category).
// In the real world, you can pass multiple inputs.
// In case of caller inputs, they would be collected from
// Bigtable. In case of ticket inputs, they would be passed by the Dialogflow agent.
const instanceValue = {
seniority: [12],
experience: ['3-Advanced'],
category: [category],
type: ['Request'],
impact: ['4-Critical'],
priority: ['P1']
};
const instance = helpers.toValue(instanceValue);
const instances = [instance];
const request = {
endpoint,
instances,
parameters,
};
return request;
}
async function getResolutionTime(request) {
// Declare the ResolutionTime constant.
var resolutionTime = 1;
// Specifies the location of the api endpoint
const clientOptions = {
apiEndpoint: 'us-central1-aiplatform.googleapis.com',
};
// Instantiates a client
const predictionServiceClient = new PredictionServiceClient(clientOptions);
// Predict request
const [response] = await predictionServiceClient.predict(request);
console.log('Predict custom trained model response');
console.log(`\tDeployed model id : ${response.deployedModelId}`);
const predictions = response.predictions;
console.log('\tPredictions :');
for (const prediction of predictions) {
resolutionTime = JSON.stringify(prediction.listValue.values[0].numberValue);
console.log(`\t\tPrediction : ${resolutionTime}`);
}
return parseInt(resolutionTime);
}
Finally, after you have successfully deployed the fulfillment, the virtual agent passes ticket information to the endpoint, collects the resolution time estimation and provides it to the customer. For better understanding, below you will find a conversation we simulate with Dialogflow CX Simulator.
MLOps Bonus: Automate the model serving with a simple Cloud Build Pipeline
In a real-world scenario, once you train the model with the pipeline and version it to the model registry, it would be time to deploy it. Now, the model deployment requires a number of testing and validation steps that need to be repeated again and again before the model reaches the serving environment. For that reason, you would build a CI/CD pipeline system. With that system, you can validate, test, integrate the model with the serving service and finally deploy them (runtime and model).
With that said, as a bonus part of this article, I implemented a simple example of CI/CD Model Deployment Pipeline with Cloud function and Cloud Build.
In particular, the Cloud Build pipeline will
- Download the model artefact from the Google Cloud Storage
- Build the runtime environment for testing
- Run the prediction unit test
- Run the integration using a temporary Vertex AI Endpoint
- Deploy the model to a target Vertex AI Endpoint
Notice that, for simplicity, I use staging as a production environment. Also it is just a simple example with no-frills. As described in the “Practitioners Guide to MLOps: A framework for continuous delivery and automation of machine learning”, these CI/CD patterns could require a complex progressive delivery process to verify compatibility, latency and throughput which implies several tests too.
Below you can find the get_build function I created to build the Cloud Build’s Model Deployment Pipeline using google-cloud-build Python SDK:
def get_build(project_id, region, bucket, source, model_uri, model_path, serving_signature, model_name, serving_image_uri,
endpoint_name, timeout):
"""
An utility function to get a build object
Args:
project_id: The name of the project
region: The name of the region
bucket: The name of the bucket
source: The GCS uri of the source code
model_uri: The GCS uri of the model
model_path: The path of the model
serving_signature: The default signature to deploy the model
model_name: The model name
serving_image_uri: The serving image uri
endpoint_name: The name of endpoint
timeout: The timeout of cloud build
Returns: build object
"""
build = cloudbuild_v1.Build()
build.source = Source(storage_source=StorageSource(bucket=bucket,
object_=source))
build.steps = [
{"name": 'gcr.io/google.com/cloudsdktool/cloud-sdk',
"args": ['gsutil', '-m', 'cp', '-r', '${_MODEL_URI}', '.'],
"id": 'Download model'},
{"name": 'gcr.io/deeplearning-platform-release/tf-cpu.2-7:latest',
"entrypoint": "pip",
"args": ["install", "-r", "build/deployment/requirements.txt", "--user"],
"id": 'Install requirements',
"wait_for": ['Download model']},
{"name": 'gcr.io/deeplearning-platform-release/tf-cpu.2-7:latest',
"entrypoint": "python",
"args": ["-m", "pytest", 'build/deployment/src/test_model.py'],
"env": ['MODEL_PATH=${_MODEL_PATH}', 'SERVING_SIGNATURE=${_SERVING_SIGNATURE}'],
"id": 'Run unit test',
"wait_for": ['Install requirements']
},
{
"name": 'gcr.io/deeplearning-platform-release/tf-cpu.2-7:latest',
"entrypoint": "python",
"args": ["-m", "pytest", 'build/deployment/src/test_endpoint.py'],
"env": ['PROJECT_ID=${_PROJECT_ID}', 'REGION=${_REGION}',
'MODEL_NAME=${_MODEL_NAME}', 'ARTIFACT_URI=${_MODEL_URI}',
'SERVING_IMAGE_URI=${_SERVING_IMAGE_URI}', 'ENDPOINT_NAME=${_ENDPOINT_NAME}'],
"id": 'Run integration test',
"wait_for": ['Run unit test']
},
{
"name": 'gcr.io/deeplearning-platform-release/tf-cpu.2-7:latest',
"entrypoint": "python",
"args": ['build/deployment/src/deploy.py'],
"env": ['PROJECT_ID=${_PROJECT_ID}', 'REGION=${_REGION}',
'MODEL_NAME=${_MODEL_NAME}', 'ARTIFACT_URI=${_MODEL_URI}',
'SERVING_IMAGE_URI=${_SERVING_IMAGE_URI}', 'ENDPOINT_NAME=${_ENDPOINT_NAME}'],
"id": 'Deploy the validated model',
"wait_for": ['Run integration test']
}
]
build.substitutions = {'_PROJECT_ID': project_id,
'_REGION': region,
'_MODEL_URI': model_uri,
'_MODEL_PATH': model_path,
'_SERVING_SIGNATURE': serving_signature,
'_MODEL_NAME': model_name,
'_SERVING_IMAGE_URI': serving_image_uri,
'_ENDPOINT_NAME': endpoint_name
}
build.timeout = Duration(seconds=timeout)
return buildThen, each time a new model will be built and stored in the target bucket path, that Cloud function will trigger a new build pipeline instance.
Below you can see what you get once the deployment successfully runs.
What’s Next
In this article, we explored how to boost up a simple chatbot built with Dialogflow CX with a ML model trained on Vertex Pipelines.
For learning purposes, we decided to train the model using Tensorflow and operationalize it using the Tensorflow Extended (aka TFX) framework.
We both learned a lot from this experience.
First, we understood how important it is working in hybrid mode where model experimentation and model engineering collaborate in order to deliver a model successfully. About that topic, here you find an interesting article from Google.
Second, I personally enjoyed TFX. I find it consistent and I experienced its potential value in accelerating the MLOps process of Tensorflow models. Also its integration with Vertex AI goes almost for free. So I will continue to experiment with it for sure in order to explore other components and capabilities.
Third, we were not aware of the VertexAI NodeJS SDK. Thanks to this demo, we are now =)
Of course, this is not the end of the story. As you can imagine this is a perfect use case where a solid retraining process could be implemented.
You deploy the virtual agent. It effectively supports customer saving time and improving customer experience. Time you save in collecting requests would affect time on solving particular issues which in turn impact on the model performance. Then consequently a new model training pipeline needs to be triggered. And finally once you get the model, you will have a transition period from its “A” version to the new “B” one.
What I describe so far generates many questions such as
- How would we design a retraining pipeline process?
- How would we implement traffic split in the model serving?
- How would we measure and manage the effects of the Feedback Loop effect?
All interesting challenges that need to be addressed with proper solutions. Don’t they?
For now, I hope you enjoyed the article. If so, clap it or leave comments. And if you’d like to know answers to the above questions, just let us know on LinkedIn or Twitter and let’s talk about it.
Till the next post…
Thank you to Camus Ma and Hussain Chinoy for their great feedback.
References
¹ We decided to use Tensorflow and TFX for learning purposes. Of course the same approach can be generalized with other ML frameworks using KFP.
² The containment rate is the percentage of users who interact with a virtual assistant and leave without speaking to a human agent. A high containment rate is a good because a ‘contained’ call prevents customers from wasting time and saves valuable resources from your customer care service.
³For TFX component documentation: https://www.tensorflow.org/tfx/guide
⁴Notice that you may define the schema manually and add information about your data set. In this way you can define your own data validation policies. For more info about TFDV, https://www.tensorflow.org/tfx/guide/tfdv
⁵The image is a personal reproduction of the Figure 8 in the Practitioners Guide to MLOps: A framework for continuous delivery and automation of machine learning
