phospho comes with several built-in jobs that you can use to process your messages: zero-shot evaluation, classification based evaluation, event detection…

But you can also create your own jobs and run them on your messages. This is what we call a custom job.

Creating a custom job function

To create a custom job function, you need to create a function that:

  • takes a lab.Message as input
  • can take additional parameters if needed (they will be passed as JobConfig)
  • returns a lab.JobResult. The lab.JobResult should contain the result of the job function and the type of the result.

For instance, to define a simple job that checks if a message contains a forbidden word, you can create a Job function like this:

from phospho import lab
from typing import List
import re

def my_custom_job(message: lab.Message, forbidden_words: List) -> lab.JobResult:
    For each each message, me will check if the forbidden words are present in the message.
    The function will return a JobResult with a boolean value
    (True if one of the words is present, False otherwise).

    pattern = r'\b(' + '|'.join(re.escape(word) for word in forbidden_words) + r')\b'

    # Use to check if any of the words are in the text
    if, message.content):
        result = True
        result = False

    return lab.JobResult(

Running a custom job

Once you have defined your custom job function, you can create a Job in your workload that will run this job function on your messages.

You need to pass the function in the job_function of the lab.Job object.

In our example:

# Create a workload in our lab
workload = lab.Workload()

# Add our job to the workload
        job_function=my_custom_job, # We add our custom job function here
            forbidden_words=["cat", "dog"]

This workload can then be run on your messages using the async_run method.

await workload.async_run(
        # No forbiden word is present.
            content="I like elephants.",
        # One forbiden word is present.
            content="I love my cat.",

# Let's see the results
for i in range(1, 3):
        f"In message {i}, a forbidden word was detected: {workload.results['message_'+str(i)]['regex_check'].value}"

# In message 1, a forbidden word was detected: False
# In message 2, a forbidden word was detected: True