use_cases/archive/call_center/notebooks/call_center.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Call center case - Post-call transcription and analysis with Azure OpenAI Service\n",
    "\n",
    "In this exercise, we will perform sentiment analysis and summerization using call center transcriptions. We will transribe the customer recording to text, then use OpenAI to detect sentiment. We also use OpenAI to summerize long text into a few sentences for further analysis."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dotenv import load_dotenv\n",
    "from pathlib import Path\n",
    "env_path = Path('../../../.env') # Change with your .env file\n",
    "load_dotenv(dotenv_path=env_path,override=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# from azure.ai.textanalytics import TextAnalyticsClient\n",
    "# from azure.core.credentials import AzureKeyCredential\n",
    "import azure.cognitiveservices.speech as speechsdk\n",
    "\n",
    "import json, os\n",
    "import string\n",
    "import time\n",
    "import wave\n",
    "\n",
    "import openai\n",
    "import re\n",
    "import requests\n",
    "import sys\n",
    "from num2words import num2words\n",
    "import os\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from openai.embeddings_utils import get_embedding, cosine_similarity\n",
    "from transformers import GPT2TokenizerFast\n",
    "\n",
    "openai.api_type = \"azure\"\n",
    "openai.api_key = os.getenv('OPENAI_API_KEY') \n",
    "openai.api_base = os.getenv('OPENAI_API_BASE') \n",
    "openai.api_version = \"2022-06-01-preview\"\n",
    "\n",
    "SPEECH_KEY = os.environ[\"SPEECH_API_KEY\"]\n",
    "\n",
    "COMPLETIONS_MODEL = os.environ[\"COMPLETIONS_MODEL\"]\n",
    "\n",
    "def recognize_speech_from_file(filename):\n",
    "    # Set up the subscription info for the Speech Service:\n",
    "    # Replace with your own subscription key and service region (e.g., \"westus\").\n",
    "    speech_key = SPEECH_KEY\n",
    "    service_region = \"westeurope\"\n",
    "\n",
    "    speech_config = speechsdk.SpeechConfig(subscription=speech_key, region=service_region)\n",
    "    audio_config = speechsdk.audio.AudioConfig(filename=filename)\n",
    "    # Creates a speech recognizer using a file as audio input, also specify the speech language\n",
    "    speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config,  audio_config=audio_config)\n",
    "\n",
    "    global done \n",
    "    done = False\n",
    "    global recognized_text_list \n",
    "    recognized_text_list=[]\n",
    "    def stop_cb(evt: speechsdk.SessionEventArgs):\n",
    "        \"\"\"callback that signals to stop continuous recognition upon receiving an event `evt`\"\"\"\n",
    "        print('CLOSING on {}'.format(evt))\n",
    "        global done\n",
    "        done = True\n",
    "\n",
    "    def recognize_cb(evt: speechsdk.SpeechRecognitionEventArgs):\n",
    "        \"\"\"callback for recognizing the recognized text\"\"\"\n",
    "        global recognized_text_list\n",
    "        recognized_text_list.append(evt.result.text)\n",
    "        # print('RECOGNIZED: {}'.format(evt.result.text))\n",
    "\n",
    "    # Connect callbacks to the events fired by the speech recognizer\n",
    "    # speech_recognizer.recognizing.connect(lambda evt: print('RECOGNIZING: {}'.format(evt)))\n",
    "    speech_recognizer.recognized.connect(recognize_cb)\n",
    "    speech_recognizer.session_started.connect(lambda evt: print('STT SESSION STARTED: {}'.format(evt)))\n",
    "    speech_recognizer.session_stopped.connect(lambda evt: print('STT SESSION STOPPED {}'.format(evt)))\n",
    "    # speech_recognizer.canceled.connect(lambda evt: print('CANCELED {}'.format(evt)))\n",
    "    # stop continuous recognition on either session stopped or canceled events\n",
    "    speech_recognizer.session_stopped.connect(stop_cb)\n",
    "    # speech_recognizer.canceled.connect(stop_cb)\n",
    "\n",
    "    # Start continuous speech recognition\n",
    "    speech_recognizer.start_continuous_recognition()\n",
    "    while not done:\n",
    "        time.sleep(.5)\n",
    "\n",
    "    speech_recognizer.stop_continuous_recognition()\n",
    "\n",
    "    return recognized_text_list"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Sentiment Analysis\n",
    "### Transcribe Customer Call to Text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "text = recognize_speech_from_file(\"../data/good_review.wav\")\n",
    "print(text)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Promot for sentiment analysis\n",
    "Use natural language to instruct OpenAI to detect customer's sentiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prompt = f\"Detect whether customer is positive or negative.  Just say positive or negative.\\n\\n{' '.join(text)}\"\n",
    "\n",
    "openai.Completion.create(\n",
    "    prompt=prompt,\n",
    "    temperature=0,\n",
    "    max_tokens=300,\n",
    "    engine=COMPLETIONS_MODEL\n",
    ")[\"choices\"][0][\"text\"].strip(\" \\n\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use a negative example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "text = recognize_speech_from_file(\"../data/bad_review.wav\")\n",
    "print(text)\n",
    "prompt = f\"Detect whether customer is positive or negative. Just say positive or negative.\\n\\n{' '.join(text)}\"\n",
    "\n",
    "openai.Completion.create(\n",
    "    prompt=prompt,\n",
    "    temperature=0,\n",
    "    max_tokens=300,\n",
    "    engine=COMPLETIONS_MODEL\n",
    ")[\"choices\"][0][\"text\"].strip(\" \\n\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Summarization\n",
    "\n",
    "Use OpenAI to summerize customer message"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "text = recognize_speech_from_file(\"../data/good_review.wav\")\n",
    "print(text)\n",
    "prompt = f\"Summerize the following text.\\n\\n{' '.join(text)}\"\n",
    "\n",
    "openai.Completion.create(\n",
    "    prompt=prompt,\n",
    "    temperature=0,\n",
    "    max_tokens=300,\n",
    "    engine=COMPLETIONS_MODEL\n",
    ")[\"choices\"][0][\"text\"].strip(\" \\n\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "azureml_py310_sdkv2",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.3"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "2139c70ac98f3202d028164a545621647e07f47fd6f5d8ac55cf952bf7c15ed1"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}