Skip to main content

Create, Get, Update, Delete

Manage your Mesh Workflows


Workflows is a useful Clarifai's feature that allows you to combine multiple models and carry out different operations. With workflows, you can create a powerful multimodal system that meets various use cases in a single API call—instead of relying only on one model.

You can use Clarifai's built-in models or your own custom models.

info

The initialization code used in the following examples is outlined in detail on the client installation page.

Create

In this example, we'll create a simple custom workflow that first extracts text from an image and then translates the extracted text to Spanish.

We'll connect the following two models to achieve our objective:

We'll specify the IDs of the models and their versions—since a model can have several versions.

########################################################################################
# In this section, we set the user authentication, app ID, and the details of the new
# custom workflow we want to create. Change these strings to run your own example.
########################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to create your own custom workflow
WORKFLOW_ID = 'my-custom-workflow'
NODE_ID_1 = 'optical-character-recognizer'
MODEL_ID_1 = 'ocr-scene-english-paddleocr'
MODEL_VERSION_ID_1 = '40dbb2c9cde44a27af226782e7157006'

NODE_ID_2 = 'text-to-text'
MODEL_ID_2 = 'text-translation-english-spanish'
MODEL_VERSION_ID_2 = '643f30558de34013aff72b0e21f244f5'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

post_workflows_response = stub.PostWorkflows(
service_pb2.PostWorkflowsRequest(
user_app_id=userDataObject,
workflows=[
resources_pb2.Workflow(
id=WORKFLOW_ID,
nodes=[
resources_pb2.WorkflowNode(
id=NODE_ID_1,
model=resources_pb2.Model(
id=MODEL_ID_1,
model_version=resources_pb2.ModelVersion(
id=MODEL_VERSION_ID_1
)
)
),
resources_pb2.WorkflowNode(
id=NODE_ID_2,
model=resources_pb2.Model(
id=MODEL_ID_2,
model_version=resources_pb2.ModelVersion(
id=MODEL_VERSION_ID_2
)
),
node_inputs=[
resources_pb2.NodeInput(node_id=NODE_ID_1)
]
),
]
)
]
),
metadata=metadata
)

if post_workflows_response.status.code != status_code_pb2.SUCCESS:
print(post_workflows_response.status)
raise Exception("Post workflows failed, status: " + post_workflows_response.status.description)

Workflow Predict

After creating the workflow, let's now use it to extract texts from this image and translate them into Spanish.

The response will contain the predictions each model in the workflow returns for the input.

tip

If you want to make a predict call with an external workflow that is outside the scope of your app, you need to use a PAT while specifying the app_id and the user_id associated with the workflow you want to use.

#############################################################################
# In this section, we set the user authentication, app ID, workflow ID, and
# image URL. Change these strings to run your own example.
##############################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to make your own predictions
WORKFLOW_ID = 'my-custom-workflow'
IMAGE_URL = 'https://samples.clarifai.com/featured-models/ocr-woman-holding-sold-sign.jpg'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

post_workflow_results_response = stub.PostWorkflowResults(
service_pb2.PostWorkflowResultsRequest(
user_app_id=userDataObject,
workflow_id=WORKFLOW_ID,
inputs=[
resources_pb2.Input(
data=resources_pb2.Data(
image=resources_pb2.Image(
url=IMAGE_URL
)
)
)
]
),
metadata=metadata
)
if post_workflow_results_response.status.code != status_code_pb2.SUCCESS:
print(post_workflow_results_response.status)
raise Exception("Post workflow results failed, status: " + post_workflow_results_response.status.description)

# We'll get one WorkflowResult for each input we used above. Because of one input, we have here one WorkflowResult
results = post_workflow_results_response.results[0]

# Each model we have in the workflow will produce its output
for output in results.outputs:
model = output.model
print("Output for the model: `%s`" % model.id)
i = 0
while(i < len(output.data.regions)):
print(output.data.regions[i].data.text.raw)
i += 1
Code Output Example
Output for the model: `ocr-scene-english-paddleocr`
SOLD
We can SELL YOURS too!
Output for the model: `text-translation-english-spanish`
VENDIDO
¡Nosotros también podemos venderla!

Get

Get all Workflows in an App

You can return all custom workflows in your app.

tip

If you want to get a list of the workflows not within the scope of your app, you need to use your PAT while specifying the user_id of their owner and the app_id of the application that you’re accessing. For example, to get Clarifai's workflows in the main app, you need to use your PAT while specifying Clarifai's user_id as "clarifai" and app_id as "main" in the request.

###################################################################
# In this section, we set the user authentication and app ID.
# Change these strings to run your own example.
###################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

list_workflows_response = stub.ListWorkflows(
service_pb2.ListWorkflowsRequest(
user_app_id=userDataObject
),
metadata=metadata
)

if list_workflows_response.status.code != status_code_pb2.SUCCESS:
print(list_workflows_response.status)
raise Exception("List workflows failed, status: " + list_workflows_response.status.description)

for workflow in list_workflows_response.workflows:
print(f"The workflow {workflow.id} consists of these models:")
for workflow_node in workflow.nodes:
model = workflow_node.model
print(model.id)
print()

Get a Workflow by a Specific ID

You can return information about a specific workflow.

############################################################################
# In this section, we set the user authentication, app ID, and workflow ID.
# Change these strings to run your own example.
#############################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change this to your own workflow ID
WORKFLOW_ID = 'my-custom-workflow'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

get_workflow_response = stub.GetWorkflow(
service_pb2.GetWorkflowRequest(
user_app_id=userDataObject,
workflow_id=WORKFLOW_ID
),
metadata=metadata
)

if get_workflow_response.status.code != status_code_pb2.SUCCESS:
print(get_workflow_response.status)
raise Exception("Get workflow failed, status: " + get_workflow_response.status.description)

workflow = get_workflow_response.workflow
print(f"The workflow consists of these models:")
for workflow_node in workflow.nodes:
model = workflow_node.model
print(model.id)

Update

Patch Workflow

You can change a workflow; that is, change the models of which the workflow consists.

The possible actions are overwrite, merge, and remove.

###################################################################################
# In this section, we set the user authentication, app ID, and the details of the
# workflow we want to update. Change these strings to run your own example.
###################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to update your own workflow
WORKFLOW_ID = 'my-custom-workflow'
NODE_ID_1 = 'audio-to-text'
MODEL_ID_1 = 'asr-wav2vec2-base-960h-english'
MODEL_VERSION_ID_1 = 'f4deae70a473492a8e2f9b7bb1dbee85'

NODE_ID_2 = 'text-summarization'
MODEL_ID_2 = 'text-summarization-english-distilbart-cnn-12-6'
MODEL_VERSION_ID_2 = '8279cec2221a4b1d9db774470940aebd'

NODE_ID_3 = 'english-to-french'
MODEL_ID_3 = 'translation-english-to-french-text'
MODEL_VERSION_ID_3 = 'c65a4a51c2b646fca5f0e4bf1ff200d7'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

patch_workflows_response = stub.PatchWorkflows(
service_pb2.PatchWorkflowsRequest(
user_app_id=userDataObject,
action="overwrite",
workflows=[
resources_pb2.Workflow(
id=WORKFLOW_ID,
nodes=[
resources_pb2.WorkflowNode(
id=NODE_ID_1,
model=resources_pb2.Model(
id=MODEL_ID_1,
model_version=resources_pb2.ModelVersion(
id=MODEL_VERSION_ID_1
)
)
),
resources_pb2.WorkflowNode(
id=NODE_ID_2,
model=resources_pb2.Model(
id=MODEL_ID_2,
model_version=resources_pb2.ModelVersion(
id=MODEL_VERSION_ID_2
)
),
node_inputs=[
resources_pb2.NodeInput(node_id=NODE_ID_1)
]
),
resources_pb2.WorkflowNode(
id=NODE_ID_3,
model=resources_pb2.Model(
id=MODEL_ID_3,
model_version=resources_pb2.ModelVersion(
id=MODEL_VERSION_ID_3
)
),
node_inputs=[
resources_pb2.NodeInput(node_id=NODE_ID_2)
]
),
]
)
]
),
metadata=metadata
)

if patch_workflows_response.status.code != status_code_pb2.SUCCESS:
print(patch_workflows_response.status)
raise Exception("Patch workflows failed, status: " + patch_workflows_response.status.description)

Delete

Delete Workflow by ID

You can delete a specific workflow.

###################################################################################
# In this section, we set the user authentication, app ID, and the ID of the
# workflow we want to delete. Change these strings to run your own example.
###################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change this to delete your own workflow
WORKFLOW_ID = 'my-custom-workflow'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

delete_workflow_response = stub.DeleteWorkflow(
service_pb2.DeleteWorkflowRequest(
user_app_id=userDataObject,
workflow_id=WORKFLOW_ID
),
metadata=metadata
)

if delete_workflow_response.status.code != status_code_pb2.SUCCESS:
print(delete_workflow_response.status)
raise Exception("Delete workflow failed, status: " + delete_workflow_response.status.description)

Delete all Workflows

You can delete all custom workflows.

tip

Instead of delete_all, you can specify a list of workflow IDs to be deleted, using the ids field.

###################################################################
# In this section, we set the user authentication and app ID.
# Change these strings to run your own example.
###################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID) # The userDataObject is required when using a PAT

delete_workflows_response = stub.DeleteWorkflows(
service_pb2.DeleteWorkflowsRequest(
user_app_id=userDataObject,
delete_all=True
# ids = ['workflow_id_1', 'workflow_id_2'] # Specify a list of workflow IDs to be deleted

),
metadata=metadata
)

if delete_workflows_response.status.code != status_code_pb2.SUCCESS:
print(delete_workflows_response.status)
raise Exception("Delete workflows failed, status: " + delete_workflows_response.status.description)