ccai-insights-sample-data/synthetic-convo-insights.py

# generates synthetic conversation data utilizing Gemini 1.5 Flash
# includes fields for metadata
# formatted for successful import into CCAI Insights

import argparse
import datetime
import json
import random
import vertexai
import ast
import os
import time
import re
from google.oauth2 import service_account
from google.cloud import storage
from vertexai.preview.language_models import TextGenerationModel
from google.cloud import aiplatform
from google.api_core import retry
from google.api_core.exceptions import ServiceUnavailable
from vertexai.generative_models import GenerativeModel, GenerationConfig, Part, FinishReason
from collections import deque
from fuzzywuzzy import fuzz

parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument(
    "num_call_logs",
    nargs="?",
    default=10000, #UPDATE WITH NUMBER OF OUTPUT FILES YOU NEED
    type=int,
    help="Number of call log files to generate",
)
args = parser.parse_args()
config = vars(args)

NUM_CALL_LOG_FILES = config["num_call_logs"]

# Authentication
SERVICE_ACCOUNT_KEY_FILE = 'XXX.json'  # REPLACE WITH YOUR ACTUAL SERVICE ACCOUNT KEY FILE PATH
credentials = service_account.Credentials.from_service_account_file(SERVICE_ACCOUNT_KEY_FILE)

# Project and Location Setup (Ensure they match your GCP project)
PROJECT_ID = '[Project ID]'  # REPLACE WITH YOUR PROJECT ID
LOCATION = 'us-central1'  # REPLACE AS NEEDED WITH YOUR CHOSEN REGION

# Vertex AI Initialization
vertexai.init(project=PROJECT_ID, location=LOCATION, credentials=credentials)

# Create the Storage Client
storage_client = storage.Client(project=PROJECT_ID, credentials=credentials)

# Use Gemini 1.5 Flash model, change config variables as needed
model = GenerativeModel(model_name="gemini-1.5-flash-001")
generation_config = GenerationConfig(
    temperature=0.5,
    top_p=1.0,
    top_k=32,
    candidate_count=1,
    max_output_tokens=8192,
)

def generate_lists():
    # Prompts for Gemini to generate lists (CHANGE PROMPTS AS NEEDED BASED ON YOUR CUSTOMER)
    service_prompt = """I work for Ulta Beauty. Can you list out 5 common Ulta Beauty related issues customers may call customer support about? Feel free to use the https://www.ulta.com/ website for inspiration on common products or services or to learn more about the company. Make sure to be very specific about the exact product or service a customer is calling Ulta Beauty about. Your output should mirror this structure. 
    Example output: ["Service1","Service2","Service3","Service4","Service5"].Do not generate anything else other than the list"""
    problem_prompt = """List 5 common problems Ulta Beauty customers might have with their experience at Ulta Beauty. Be as specific and detailed as possible with customer responses, product types, issues, etc. Make sure to be very specific about the exact product, service, or issue a customer is calling about. Here is a website that give examples on common complaints Ulta Beauty customers may have: https://www.reddit.com/r/Ulta/. Here is another website that give examples on common complaints Ulta Beauty customers may have: https://www.bbb.org/us/il/bolingbrook/profile/retail-stores/ulta-beauty-0654-27005363/customer-reviews.  Your output should mirror this structure. 
    Example output: ["Problem1","Problem2","Problem3","Problem4","Problem5"].Do not generate anything else other than the list"""
    greeting_prompt = """List 5 different realistic ways an Ulta Beauty customer service representative might greet a customer on the phone, ensuring each greeting uses a UNIQUE and DISTINCT agent name. Include both a greeting and an offer to assist, mirroring Ulta Beauty's professional style. Replace '[Agent Name]' with a diverse range of realistic first names (both female and male). The agent names should be diverse, representing various genders and cultural backgrounds.
    Example output: ["Hello, thank you for calling Ulta Beauty. My name is Jose. How may I help you today?", "Welcome to Ulta Beauty! This is Ahmad. How can I assist you?", "Good morning/afternoon/evening! Thank you for contacting Ulta Beauty. This is David. What can I do for you today?","Hello, this is Ava with Ulta Beauty. How can I assist help you beautify your day?","It's a pleasure to assist you at Ulta Beauty. My name is Priya. How may I help you today?"]
    Do not generate anything else other than the list."""
    agent_name_prompt = """Generate a list of 50 DISTINCT and culturally diverse names that could be used for Ulta Beauty customer service representatives. Include names that are uncommon or less frequent, representing a wide range of ethnicities and genders. Ensure no repetition or similarity. 
    Example output: ["Jessica", "Priya", "Zaid", "Diego", "Luis"]."""
    closing_prompt = """List 5 different ways an Ulta Beauty customer service representative might end a call with a customer.Your output should mirror this structure. 
    Example output: ["Closing greeting1","Closing greeting2","Closing greeting3","Closing greeting4","Closing greeting5"].Do not generate anything else other than the list"""
    closing_response_prompt = """List 5 different ways a customer might respond to an Ulta Beauty customer service representative ending a call based off the last closing prompt message from the agent.
    Example output: ["Thank you for your help. Have a great day!", "You're welcome. Goodbye.", "No problem. Thanks for your time."]
    Do not generate anything else other than the list."""
    
    # Get responses from Gemini and clean up the output for CCAI Insights formatting
    services = model.generate_content(service_prompt)
    services_text = services.text.strip()[1:-1].replace('"', '').split(",")

    problems = model.generate_content(problem_prompt)
    problems_text = problems.text.strip()[1:-1].replace('"', '').replace("\n", "").split(",")

    greetings = model.generate_content(greeting_prompt)
    greetings_text = greetings.text.strip()[1:-1].replace('"', '').split(",")

    # Filter out unprofessional greetings and ensure Ulta Beauty is mentioned in the initial agent greeting
    greetings_text = list(
        filter(
            lambda g: "ulta beauty" in g.lower() and any(keyword in g.lower() for keyword in ["help", "assist", "welcome"]),
            greetings_text
        )
    )

    agent_names = model.generate_content(agent_name_prompt)
    agent_names_text = agent_names.text.strip()[1:-1].replace('"', '').split(",")

    closing_remarks = model.generate_content(closing_prompt)
    closing_remarks_text = closing_remarks.text.strip()[1:-1].replace('"', '').split(",")
    
    closing_responses = model.generate_content(closing_response_prompt)
    closing_responses_text = closing_responses.text.strip()[1:-1].replace('"', '').replace('-', '').split(",")

    #Parse with ast.literal_eval
    return services_text, problems_text, greetings_text, closing_remarks_text, closing_responses_text, agent_names_text

def generate_log(services, problems, greetings, closing_remarks, closing_responses, agent_names_buffer, max_retries=3, max_regeneration_retries=3):
    global shuffled_agent_names
    service = random.choice(services)
    problem_description = random.choice(problems)

    # Generate timestamps
     # Specify the desired start and end years, along with month/day ranges
    start_year = 2024    # Change to desired start year
    start_month = 1      # Change to desired start month
    start_day = 1       # Change to desired start day
    end_year = 2024      # Change to desired end year
    end_month = 9       # Change to desired end month
    end_day = 3         # Change to desired end day
    
    # Generate random timestamps within the specified range
    def generate_random_timestamp():
        start_date = datetime.datetime(year=start_year, month=start_month, day=start_day)
        end_date = datetime.datetime(year=end_year, month=end_month, day=end_day, hour=23, minute=59, second=59)
    
        random_datetime = start_date + random.random() * (end_date - start_date)
        return int(random_datetime.timestamp() * 1000000)
    
    # Generate timestamp and response delay
    timestamp = generate_random_timestamp()
    response_delay = random.randint(5000000, 10000000)  # 5 to 10 seconds for all roles

    # Increased regeneration threshold (adjust as needed)
    regeneration_threshold = 3  # Lowered threshold for more frequent regeneration

    if len(agent_names_buffer) < 2:
        for _ in range(max_regeneration_retries):
            print("Regenerating agent names and greetings for more variety...")
            services, problems, greetings, closing_remarks, closing, new_agent_names = generate_lists()  # Regenerate greetings as well
            agent_names_buffer.extend(new_agent_names)
            if len(agent_names_buffer) >= 2:
                break
        else:
            print("Max regeneration retries reached. Skipping this call log.")
            return None

    customer_behavior = random.choice(["polite and patient", "frustrated and impatient", "angry and demanding", "confused and unsure"])

    # Generate a natural problem statement
    problem_statement_prompt = f"""
    Rewrite this issue into a natural statement a customer would say to describe their problem with their {service}: "{problem_description}" at the BEGINNING of their call. Make sure the customer provides context to their issue.
    """

    for retry_count in range(max_retries):
        try:
            problem_statement_response = model.generate_content(problem_statement_prompt)

            # Check for safety filter blocks in any candidate
            for candidate in problem_statement_response.candidates:
                if candidate.finish_reason == "STOP_REASON_SAFETY":
                    raise Exception("Safety filter triggered. Retrying...")

            customer_statement = problem_statement_response.text.strip()

            # --- Generate Metadata (but don't extract agent name yet) ---
            call_id = random.randint(1000, 999999)
            #language_code = "en-US"
            #call_type = random.choice(["inbound", "outbound"])
            channel = random.choice(["phone", "chat"])  # Randomly chooses phone or chat
            #agent_group = random.choice(["Tier 1 Support", "Billing"])
            agent_experience = random.choice(["junior", "senior", "manager", "supervisor", "trainee"])
            agent_location = random.choice(["US", "India", "EMEA"])
            customer_id = random.randint(100, 9999)
            customer_sentiment = "positive" if customer_behavior == "polite and patient" else "negative"  # Example inference
            customer_region = random.choice(["East Coast", "West Coast", "Central"])



            prompt_template = f"""
            Create a customer support transcript where an Ulta Beauty agent helps a customer with their {service}.
            The conversation starts with the agent's greeting.
            Adhere strictly to this format:
            Agent: {random.choice(greetings)}
            Customer: {customer_statement}
            Agent: [Agent's response acknowledging the problem and starting troubleshooting]
            Customer: [Customer's response to the troubleshooting steps]
            Agent: [Further troubleshooting or resolution steps]
            ... (continue the back-and-forth as needed)
            Agent: [Resolution of the issue or escalation]
            Agent: {random.choice(closing_remarks)}
            Customer: [Customer's natural response acknowledging resolution and ending the call]

            Additional instructions:

            *   Use "{random.choice(greetings)}" for the agent's greeting.
            *   Use "{random.choice(closing_remarks)}" for the agent's closing remark.
            *   The conversation MUST include troubleshooting steps and a resolution.
            *   Focus on a single core issue the customer is experiencing
            *   The customer is "{customer_behavior}"
            """

            print(prompt_template)

            response = model.generate_content(prompt_template, generation_config=generation_config)

            transcript = response.text

            # Check for safety filter blocks in any candidate (not just the first one)
            for candidate in response.candidates:
                if candidate.finish_reason == "STOP_REASON_SAFETY":
                    raise Exception("Safety filter triggered. Retrying...")

            # Enhanced Transcript Parsing with Logic to Prevent Unnatural Endings
            entries = []
            current_speaker = None
            customer_said_no = False
            short_customer_response = False
            agent_asked_anything_else = False
            last_agent_line = ""

            for line in transcript.splitlines():
                line = line.strip()
                if line.lower().startswith("customer") or line.lower().startswith("agent"):
                    if line.lower().startswith("customer"):
                        entries.append({"role": "CUSTOMER", "text": line[8:].strip(), "user_id": 1})
                        if line.lower().strip() in ["no", "no thanks", "that's all", "that's it", "i'm good", "nothing else","okay"]:
                            customer_said_no = True
                        if len(line.lower().strip()) <= 3:
                            short_customer_response = True
                    elif line.lower().startswith("agent"):
                        last_agent_line = line[5:].strip()
                        if "anything else" in last_agent_line.lower():
                            agent_asked_anything_else = True
                        
                        # Condition to skip the "anything else" response after customer says no
                        if not (customer_said_no and "anything else" in last_agent_line.lower()):
                            entries.append({"role": "AGENT", "text": line[5:].strip(), "user_id": 2})

            # Check if the first agent entry has a proper greeting (with fuzzy matching)
            if entries and entries[0]["role"] == "AGENT":
                agent_greeting = entries[0]["text"].lower()
            
                # Define a list of acceptable greeting keywords/phrases
                greeting_keywords = ["help", "assist", "welcome", "hello", "hi", "good morning", "good afternoon", "good evening"]
            
                # Check if any of the greeting keywords/phrases have a high similarity score with the agent's greeting
                has_valid_greeting = any(
                    fuzz.partial_ratio(keyword, agent_greeting) > 80
                    for keyword in greeting_keywords
                )
            
                if not has_valid_greeting:
                    print("Agent's greeting is missing or incomplete (fuzzy matching). Retrying...")
                    return generate_log(services, problems, greetings, closing_remarks, closing_responses, agent_names)
            
            # Additional check for the first few agent turns
            for i, entry in enumerate(entries[:3]):  # Check the first 3 agent turns
                if entry["role"] == "AGENT":
                    if any(product.lower() in entry["text"].lower() for product in services):
                        print("Agent assumed the product too early. Retrying...")
                        return generate_log(services, problems, greetings, closing_remarks, closing_responses, agent_names)

            # Retry Conditions (consolidated for readability)
            if any((
                agent_asked_anything_else and last_agent_line == entries[-1]['text'],
                customer_said_no and "anything else" in last_agent_line.lower(),
                short_customer_response,
                not entries  # Check for blank output
            )):
                retry_reason = (
                    "Customer didn't answer 'anything else?'"
                    if agent_asked_anything_else
                    and last_agent_line == entries[-1]["text"]
                    else "Agent asked again after customer said no"
                    if customer_said_no and "anything else" in last_agent_line.lower()
                    else "Customer response too short"
                    if short_customer_response
                    else "Blank output"
                )
                print(f"{retry_reason}. Retrying...")
                return generate_log(
                    services,
                    problems,
                    greetings,
                    closing_remarks,
                    closing_responses,
                    agent_names,
                )

            # Add timestamps
            for i, entry in enumerate(entries):
                entry["start_timestamp_usec"] = timestamp + response_delay * i



            # *** Extract agent name AFTER populating entries ***
            agent_name_found = False
            for entry in entries:
                if entry["role"] == "AGENT":
                    # Try to find the agent name using the regular expression
                    match = re.search(
                        r"(?:my name is|this is|i'm)\s+([\w\s]+)",
                        entry["text"],
                        re.IGNORECASE,
                    )
                    if match:
                        agent_name = match.group(1).strip()
                        agent_name_found = True
                        break  # Stop after finding the agent name

            if not agent_name_found:
                # Handle the case where no agent name is found in the transcript
                print("No agent greeting found in the transcript. Retrying...")
                return generate_log(
                    services,
                    problems,
                    greetings,
                    closing_remarks,
                    closing_responses,
                    agent_names_buffer,
                )



            # Generate a unique Agent ID (you can customize this logic)
            agent_id = random.randint(1000, 9999)
            
            # Replace any remaining placeholders
            #for entry in entries:
             #   if "[agent name]" in entry["text"].lower():
              #      entry["text"] = entry["text"].replace("[agent name]", agent_name_for_transcript)

            
            # Add agent metadata to each agent entry
            for entry in entries:
                if entry["role"] == "AGENT":
                    entry["agent_name"] = agent_name  # Use the extracted agent_name
                    entry["agent_id"] = agent_id

            
            # --- Create metadata dictionary ---
            metadata = {
                "call_id": call_id,
                #"language_code": language_code,
                #"call_type": call_type,
                "channel": channel,
                #"agent_group": agent_group,
                "agent_experience": agent_experience,
                "agent_location": agent_location,
                "customer_id": customer_id,
                "customer_sentiment": customer_sentiment,
                "customer_region": customer_region,
                "agent_id": agent_id,
                # ... add more metadata as needed
            }


            call_log = {
                "entries": entries,
                "metadata": metadata,
            }  # Include metadata in the call_log
            json_object = json.dumps(call_log, indent=4)
            return json_object  # Return the generated JSON if successful

        except Exception as e:
            print(f"Error generating log (attempt {retry_count + 1}/{max_retries}): {e}")
            if retry_count < max_retries - 1:
                time.sleep(2 ** retry_count)
            else:
                print("Max retries reached. Skipping this call log.")
                return None


# Main execution loop

services, problems, greetings, closing_remarks, closing, agent_names = generate_lists()
random.shuffle(agent_names)  # Shuffle the entire list once at the beginning

# Create a circular buffer (deque) from the shuffled agent names
agent_names_buffer = deque(agent_names)

# Main execution loop
for i in range(NUM_CALL_LOG_FILES):
    json_object = generate_log(services, problems, greetings, closing_remarks, closing, agent_names_buffer)

    if json_object is not None:  # Only save if generation was successful
        # Upload to GCS with overwrite prevention
        bucket = storage_client.bucket("ulta-enriched4") #UPDATE WITH YOUR GCS BUCKET
        base_filename = f"gem_chat_{i}_np.json" #CHANGE FILE NAME AS NEEDED
        filename = base_filename
        counter = 1

        blob = bucket.blob(filename)
        while blob.exists():  # Check for existing blob
            filename = f"{os.path.splitext(base_filename)[0]}_{counter}.json"
            blob = bucket.blob(filename)  # Update the blob reference
            counter += 1

         # Retry mechanism for blob upload
        @retry.Retry(predicate=retry.if_exception_type(ServiceUnavailable), deadline=60)
        def upload_blob():
            blob.upload_from_string(json_object)

        try:
            upload_blob()
            print(f"Uploaded {filename} to GCS bucket ulta-enriched4") #UPDATE WITH YOUR GCS BUCKET
        except Exception as e:  # Catch any remaining errors
            print(f"Upload failed after retries: {e}")
    else:
        print(f"Skipping call log {i} due to an error.")