Tutorials

Moderation for Trust and Safety

In platforms and communities powered by user-generated content (e.g., Snapchat, Tiktok, Meta, Imgur, etc.), visual content is uploaded at high volumes and shared rapidly. Unwanted visual content poses a great risk in this context if it goes undetected, degrading both individual user experiences and platform perception.

Using coactive, visual content moderation can be automated and customized to your individual trust and safety guidelines.

For example, you can use coactive to identify problematic trends in real-time (e.g., a violent video going viral) and stop unwanted visual content at the source.

Real-time monitoring

coactive allows users to easily perform real-time monitoring at scale through millions of unstructured visual assets. In this notebook we'll demonstrate how you can use coactive's python SDK to perform real-time classification and analytics for content moderation.

1. Import dependencies

First, we import any necesary dependencies. Note that you will first need to install the coactive SDK package to your local python environment.

# From python core and other common packages
import os
from PIL import Image
from IPython.display import HTML
import pandas as pd

# From coactive SDK package
import coactive
from coactive.apis import ClassificationApi, QueryApi
from coactive.model.classification_request import ClassificationRequest

from coactive.model.query import Query
from coactive.model.query_request import QueryRequest

2. Authenticate

Next, we load the authentication and environment variables necessary for calling coactive's APIs. Note that these variables are environment specific and will be provided by Coactive.

from auth_credentials import COACTIVE_HOST, CLIENT_ID, CLIENT_SECRET
from env_variables import CLIENT_EMBEDDING_ID

access_token = f"{CLIENT_ID}:{CLIENT_SECRET}"
configuration = coactive.Configuration(
    host = COACTIVE_HOST,
    access_token = access_token,
)

3. Real-time analytics

coactive's real-time analytics engine allows you to perform classification using our Classification API. Additionally, you can query your unstructured visual data using SQL using our Query API. More specifically, coactive:

  • provides a structured view of your visual data (i.e. rows = visual asset, columns = metadata) giving you the ability to run analytical SQL queries
  • can generate metadata on demand for your visual assets using a library of visual concepts
  • gives you full control over these visual concepts used in classification and queries, allowing you to seamlessly create and update these concepts as your tasks or visual data change over time

Below, we highlight the core functions of our real-time analytics engine to monitor a dataset of user-generated visual assets for problematic content.

We define a custom functions to make it simple to run multiple API requests using the SDK and to visually validate the results using this notebook. These custom functions can be found below.

import time
from typing import List

from helpers import get_images_df_from_api_response, image_formatter

def image_classification(image_queries: List[str]) -> None:
    '''
    Example function that displays the results of calling the Classification API.
    '''

    # Call ClassificationAPI
    with coactive.ApiClient(configuration) as api_client:

        try:
            classification_request = ClassificationRequest(
              embedding_id=CLIENT_EMBEDDING_ID,
              paths=image_queries,
            )
            api_response = ClassificationApi(api_client).classify_assets(
                classification_request)
        except coactive.ApiException as e:
            print("Exception when calling ClassificationApi: %s\n" % e)

    # Display results
    df = get_images_df_from_api_response(image_queries, api_response)
    display(HTML(df.to_html(formatters={'image': image_formatter},
                            escape=False,
                            index=False)))


def run_visual_sql_query(query: str) -> pd.DataFrame:
    '''
    Example function that runs a SQL query using the Query API.
    '''

    # Call QueryAPI to start query
    with coactive.ApiClient(configuration) as api_client:
        try:
            api_response = QueryApi(api_client).execute_query(
                QueryRequest(query=query, embedding_id=CLIENT_EMBEDDING_ID))
            api_response = api_response.to_dict()
        except coactive.ApiException as e:
            print("Exception when calling QueryApi: %s\n" % e)
    query_id = api_response['query_id']
    status = api_response['status']

    # Check query status until it is complete
    while status != 'Complete':
        # Wait
        time.sleep(5)

        # Check query status until complete
        with coactive.ApiClient(configuration) as api_client:
            try:
                query_response = QueryApi(api_client).get_query_by_id(query_id)
                query_response = query_response.to_dict()
            except coactive.ApiException as e:
                print("Exception when calling QueryApi: %s\n" % e)
        status = query_response['status']

    # Display results
    run_time = query_response['end_dt'] - query_response['created_dt'] 
    print(f'Query run time (hrs:min:sec) = {run_time}')
    return pd.DataFrame([row['data'] for row in query_response['results']['data']])

We define a custom functions to make it simple to run multiple API requests using the SDK and to visually validate the results using this notebook. These custom functions can be found below.