From Calls to Data Analytics: Unleashing the Power of Amazon Transcribe in Your Call Center

Image taken from Splash

Amazon Transcribe is a well-known speech-to-text (speech recognition) transcription service that enables businesses to transcribe call recordings and extract valuable insights regarding sentiment, call characteristics, call summarisation, and call tagging. You can transcribe media in real time (streaming) or you can transcribe media files located in an Amazon S3 bucket (batch).

For a comprehensive understanding of Amazon Transcribe’s features, supported languages, pricing details, and more, a thorough description is available on the AWS website.

In this article, I aim to guide you through the technical setup of Amazon Transcribe, empowering you to seamlessly integrate it into your call operations to transcribe call recordings and obtain post call analytics insights through a batch approach.

The first step is to import the needed packages and to define key input variables for your code setup.

# import packages
import pandas as pd
import boto3
import requests as r
from retry import retry
from loguru import logger

# define main variables
boto_session = boto3.Session(profile_name="<your credentials profile name>")
# This is a boto3 session object which establishes connection to AWS
boto_iam = boto_session.resource("iam")
# A boto3 resource object which provides access to  IAM-related operations
TRANSCRIBE = boto_session.client("transcribe")
# A boto3 client object for the Amazon Transcribe service 
DATA_ACCESS_ROLE = boto_iam.Role("<your iam role name>")
# An IAM Role object grants necessary permissions to Transcribe to access data.

Now we’ll initiate transcription jobs for the call recordings stored in your S3 bucket using the API call TRANSCRIBE.start_call_analytics_job.

Note that:

• df_calls_jobs: A DataFrame outlining information about the calls
• job_name: The unique name you'll use to identify and retrieve results for each job.
• job_url: The S3 URI pinpointing the location of the corresponding call recording within the bucket.

@retry(tries=5)
def start_transcribe_analytics_jobs(df, TRANSCRIBE, DATA_ACCESS_ROLE):
   """
    Initiates call analytics jobs using the Amazon Transcribe analytics API.

    Args:
    - df (pd.DataFrame): DataFrame containing job information with columns 'job_name' and 'job_url'.
    - TRANSCRIBE (boto3.client): Boto3 client for the Amazon Transcribe service.
    - DATA_ACCESS_ROLE (IAM.Role): IAM role providing data access permissions for the analytics job
     """
    def job_func(row):
        job_name = row['job_name']
        job_url = row['job_url']
        try:
            TRANSCRIBE.start_call_analytics_job(
                CallAnalyticsJobName=job_name,
                Media={'MediaFileUri': job_url},
                DataAccessRoleArn=DATA_ACCESS_ROLE.arn,
                ChannelDefinitions=[{'ChannelId': 0, 'ParticipantRole': 'AGENT'},
                                    {'ChannelId': 1, 'ParticipantRole': 'CUSTOMER'}])
        except:
            pass
        return row
    df = df.apply(job_func, axis=1)
    return df

 # apply the function
 df_calls_jobs = start_transcribe_analytics_jobs(
        df_calls_jobs, TRANSCRIBE, DATA_ACCESS_ROLE)
 logger.info("The processing of transcribe call analytics jobs with number of calls: {}".format(
        len(df_calls_jobs)))

Now, we will define another function which calls the API get_call_analytics_job to retrieve the job response output using the job name.

@retry(tries=5)
def get_transcribe_analytics_job_output(df, TRANSCRIBE):
    """
    Retrieves and parses the output of call analytics jobs. 

    Args:
    - df (pd.DataFrame): DataFrame containing job information with the 'job_name' column.
    - TRANSCRIBE (boto3.client): Boto3 client for the Amazon Transcribe service.

    Returns:
    pd.DataFrame: Updated DataFrame with job response output.
    """
    def job_func(row):
        job_name = row['job_name']
        try:
            row['job_response_output'] = r.get(
                TRANSCRIBE.get_call_analytics_job(CallAnalyticsJobName=job_name)['CallAnalyticsJob']['Transcript'][
                    'TranscriptFileUri']).json()
        except:
            row['job_response_output'] = None
        return row
    df = df.apply(job_func, axis=1)
    incomplete_jobs = df['job_response_output'].isnull().sum()
    logger.info(f"Number of incomplete jobs: {incomplete_jobs}")
    return df

# apply the function 
df_calls_jobs_result = get_transcribe_analytics_job_output(
    df_calls_jobs, TRANSCRIBE)

Following that, we'll create functions designed to parse the job response outputs, allowing us to extract the full text transcript, analytics attributes related to sentiment for both agents and customers, call characteristics like talk time, interruption time, and silence time, as well as summarisation features highlighting issues, outcomes, and detected action items within the call. For a comprehensive overview of the diverse analytics attributes, please refer to this link.

We will start with the function to get the transcripts from the job output.

def analytics_call_transcript(row):
   """
    Parses the transcript datapoint from the job output of call analytics.

    Args:
    - row (pd.Series): DataFrame row containing the 'job_response_output' column.

    Returns:
    pd.Series: Updated row with the parsed transcript added as the 'transcript' column
    """
    try:
        for i in row['job_response_output']:
            transcript_dict = i.get('Transcript', {})
            if transcript_dict:
                content_list = [dict_result.get('Content', '')
                            for dict_result in transcript_dict]
                transcript = [
                f"{speakers[i % len(speakers)]}: {text}" for i, text in enumerate(content_list)]
            row['transcript'] = ', '.join(transcript)
    except:
        row['transcript'] = None
    return row

def transcript_call_public_fnt(tmp) -> pd.DataFrame:
    """
    Processes a DataFrame containing job response outputs and generates transcripts using the 'analytics_call_transcript' function
    Args:
    - tmp (pd.DataFrame): DataFrame containing job response outputs.

    Returns:
    pd.DataFrame: DataFrame with transcripts extracted from the job response outputs.
    """
    results_df = pd.DataFrame()
    for i in range(0, len(tmp)):
        try:
            row_df = tmp.loc[[i], :].copy()
            row_df_results = analytics_call_transcript(row_df)
            results_df = pd.concat([results_df, row_df_results])
        except:
            pass
    results_df = results_df.reset_index(drop=True)
    empty_transcripts = results_df['transcript'].isnull().sum()
    logger.info(f"Number of empty transcripts: {empty_transcripts}")
    return results_df

# apply the public transcript function 
df_calls_analytics_ouput = transcript_call_public_fnt(
    df_calls_analytics_ouput)

Then, we will have a function which extracts the job output related to sentiment, call characteristics, and matched categories.

def parse_call_analytics_output(df) -> pd.DataFrame:

    """
    This function parse each data attribute related to sentiment and call traits 
    from the job output with try/except approach to account for missing values. 
    """
    try:
        df['overall_sentiment_customer'] = df['job_response_output']['ConversationCharacteristics']['Sentiment']['OverallSentiment']['CUSTOMER']
    except Exception as e:
        df['overall_sentiment_customer'] = None
    try:
        df['overall_sentiment_agent'] = df['job_response_output']['ConversationCharacteristics']['Sentiment']['OverallSentiment']['AGENT']
    except Exception as e:
        df['overall_sentiment_agent'] = None
    try:
        df['sentiment_scores_agent_per_quarter'] = [d['Score'] for d in df['job_response_output']
                                                    ['ConversationCharacteristics']['Sentiment']['SentimentByPeriod']['QUARTER']['AGENT']]
    except Exception as e:
        df['sentiment_scores_agent_per_quarter'] = None
    try:
        df['sentiment_scores_customer_per_quarter'] = [d['Score'] for d in df['job_response_output']
                                                       ['ConversationCharacteristics']['Sentiment']['SentimentByPeriod']['QUARTER']['CUSTOMER']]
    except Exception as e:
        df['sentiment_scores_customer_per_quarter'] = None
    try:
        df['non_talk_time_sec'] = (
            df['job_response_output']['ConversationCharacteristics']['NonTalkTime']['TotalTimeMillis'])/1000
    except Exception as e:
        df['non_talk_time_sec'] = None
    try:
        df['interupted_time_agent_sec'] = (df['job_response_output']['ConversationCharacteristics']
                                           ['Interruptions']['InterruptionsByInterrupter']['AGENT'][0]['DurationMillis'])/1000
    except Exception as e:
        df['interupted_time_agent_sec'] = None
    try:
        df['interupted_time_customer_sec'] = (df['job_response_output']['ConversationCharacteristics']
                                              ['Interruptions']['InterruptionsByInterrupter']['CUSTOMER'][0]['DurationMillis'])/1000
    except Exception as e:
        df['interupted_time_customer_sec'] = None
    try:
        df['talk_speed_words_per_min_agent'] = (
            df['job_response_output']['ConversationCharacteristics']['TalkSpeed']['DetailsByParticipant']['AGENT']['AverageWordsPerMinute'])
    except Exception as e:
        df['talk_speed_words_per_min_agent'] = None
    try:
        df['talk_speed_words_per_min_customer'] = (
            df['job_response_output']['ConversationCharacteristics']['TalkSpeed']['DetailsByParticipant']['CUSTOMER']['AverageWordsPerMinute'])
    except Exception as e:
        df['talk_speed_words_per_min_customer'] = None
    try:
        df['talk_time_agent_sec'] = (df['job_response_output']['ConversationCharacteristics']
                                     ['TalkTime']['DetailsByParticipant']['AGENT']['TotalTimeMillis'])/1000
    except Exception as e:
        df['talk_time_agent_sec'] = None
    try:
        df['talk_time_customer_sec'] = (df['job_response_output']['ConversationCharacteristics']
                                        ['TalkTime']['DetailsByParticipant']['CUSTOMER']['TotalTimeMillis'])/1000
    except Exception as e:
        df['talk_time_customer_sec'] = None
    try:
        df['matched_categories'] = (df['job_response_output']['Categories']
                                    ['MatchedCategories'])
    except Exception as e:
        df['matched_categories'] = None
    return df


# apply the above function
df_calls_analytics_ouput = df_calls_jobs_result.apply(
    parse_call_analytics_output, axis=1)

Next, we will have a function to get the summary features data points: issues detected, outcomes detected, and action items detected in the call.

def analytics_call_summarization(row):
  """
    Parses summarization datapoints from the job response output.
    Args:
    - row (pd.Series): DataFrame row containing the 'job_response_output' column.

    Returns:
    pd.Series: Updated row with the parsed summarization data.
    """
    key_list = []
    content_list = []
    key_cols = ['IssuesDetected', 'OutcomesDetected', 'ActionItemsDetected']
    df_summary = pd.DataFrame()  
    try:
        for i in row['job_response_output']:
            transcript = i['Transcript']  
            for dict_result in transcript:
                keys = dict_result.keys()  
                for k in keys:
                    if k in key_cols:  
                        key_list.append(k)
                        content_list.append(dict_result['Content'])
        if key_list:
            data = {name: [] for name in key_list}
            for i, val in enumerate(content_list):
                data[key_list[i % len(key_list)]].append(val)
            df_summary = pd.DataFrame(data)
    except Exception as e:
        df_summary = pd.DataFrame(columns=key_cols)
    for col in key_cols:
        if col not in df_summary.columns:
            df_summary[col] = None
    row.index = [0]
    row = pd.concat([row, df_summary], axis=1)
    return row

def summary_call_public_fnt(tmp) -> pd.DataFrame:
    """
    Processes a DataFrame containing job response outputs and generates summarization data using the 'analytics_call_summarization' function.
    Args:
    - tmp (pd.DataFrame): DataFrame containing job response outputs.
    Returns:
    pd.DataFrame: DataFrame with summarization data extracted from the job response outputs.
    """
    results_df = pd.DataFrame()
    for i in range(0, len(tmp)):
        row_df = tmp.iloc[[i], :].copy()
        row_df_results = analytics_call_summarization(row_df)
        results_df = pd.concat([results_df, row_df_results], axis=0)
        results_df = results_df.reset_index(drop=True)
    return results_df

# apply the above summary_call_public_fnt
df_calls_analytics_ouput_final = summary_call_public_fnt(
    df_calls_analytics_ouput)

Finally, we managed to complete the technical setup to utilise Amazon Transcribe to get analytical insights on the call recordings and thereby boost customer experience and satisfaction.

I hope you enjoyed reading the article. Make sure to follow me for more captivating reads on topics spanning Data Science, Machine Learning, Data Engineering and AWS.