app.py

from fastapi import FastAPI, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi import HTTPException
from pydantic import BaseModel
import openai
from dotenv import load_dotenv
import os
from deepgram import DeepgramClient, PrerecordedOptions
from typing import List, Dict
from fastapi.responses import PlainTextResponse
import re
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from openai import AzureOpenAI

# Load environment variables from .env file
load_dotenv()

# # Configure OpenAI for Azure
# openai.api_type = "azure"
# openai.api_base = "https://amplifai-openai.openai.azure.com/"
# # openai.api_version = "2023-07-01-preview"
# openai.api_version = "2024-08-01-preview"
# openai.api_key = os.getenv("OPENAI_API_KEY")

client = AzureOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),  
    api_version="2024-05-01-preview",
    azure_endpoint = "https://amplifai-openai.openai.azure.com/"
)

deployment_name='gpt-4'

DEEPGRAM_API_KEY = os.getenv("DEEPGRAM_API_KEY")
if not DEEPGRAM_API_KEY:
    raise Exception("Deepgram API key not found in environment variables")

app = FastAPI()

class Prompt(BaseModel):
    text: str

class GenerateResponse(BaseModel):
    text: str
class GenerateResponse2(BaseModel):
    text: dict
class GenerateResponse3(BaseModel):
    text: List[Dict[str, str]]
class GenerateResponse4(BaseModel):
    text: list

# #Prepare the chat prompt 
# chat_prompt = [
#     {
#         "role": "system",
#         "content": [
#             {
#                 "type": "text",
#                 "text": "The data is a pre-recorded conversation between a call center agent and a customer. Identify the following: keywords/metrics about the conversation, a summary of the conversation including details. Output should be in the format Keywords/Metrics: ,\n Summary:. \n Data: "
#             }
#         ]
#     }
# ] 

# #Summarize function
# def summarize_text(text: str):
#         response = client.chat.completions.create(
#             model = deployment_name,
#             #engine="AmplifAI-Chat",
#             messages="The data is a pre-recorded conversation between a call center agent and a customer. Identify the following: keywords/metrics about the conversation, a summary of the conversation including details. Output should be in the format Keywords/Metrics: ,\n Summary:. \n Data: " + text,
#             temperature=0.7,
#             max_tokens=750,
#             top_p=1.0,
#             frequency_penalty=0.0,
#             presence_penalty=0.0
#         )
#         if response.choices:
#             # Extract the text from the response
#             summary = response.choices[0].text.strip()
            
#             # Use regular expressions to find the keywords/metrics and summary sections
#             keywords_match = re.search(r"Keywords/Metrics:(.*?)(\n|$)", summary, re.S)
#             summary_match = re.search(r"Summary:(.*?)(\n|$)", summary, re.S)
            
#             # Extract the content if matches are found
#             keywords = keywords_match.group(1).strip() if keywords_match else "No keywords found"
#             conversation_summary = summary_match.group(1).strip() if summary_match else "No summary found"
            
#             # Return a clean version with just the keywords and summary
#             return GenerateResponse(text=f"Keywords/Metrics: {keywords}\nSummary: {conversation_summary}")
        
#         else:
#             return "No response from the API."
#         # if response.choices:
#         #     summary = response.choices[0].text.strip()
#         #     return GenerateResponse(text=summary)
#         # else:
#         #     return "No response from the API."


#Summarize function
def summarize_text(text: str):
        response = client.chat.completions.create(
            model = deployment_name,
            #engine="AmplifAI-Chat",
            messages=[
                {"role": "system", "content": "You are an AI assistant that summarizes conversations and extracts keywords/metrics."},
                {"role": "user", "content": f"The data is a pre-recorded conversation between a call center agent and a customer. Identify the following: \n\n1. Keywords/Metrics about the conversation \n2. A summary of the conversation including details (max 4 lines). \n\nOutput format: \nKeywords/Metrics: [Extracted Keywords]\nSummary: [Generated Summary]\n\nData: {text}"}
            ],
            temperature=0.7,
            max_tokens=150,
            top_p=1.0,
            frequency_penalty=0.0,
            presence_penalty=0.0
        )
        if response.choices:
            # Extract the text from the response
            # summary = response.choices[0].text.strip()
            summary = response.choices[0].message.content.strip()
            
            # Use regular expressions to find the keywords/metrics and summary sections
            keywords_match = re.search(r"Keywords/Metrics:(.*?)(\n|$)", summary, re.S)
            summary_match = re.search(r"Summary:(.*?)(\n|$)", summary, re.S)
            
            # Extract the content if matches are found
            keywords = keywords_match.group(1).strip() if keywords_match else "No keywords found"
            conversation_summary = summary_match.group(1).strip() if summary_match else "No summary found"
            
            # Return a clean version with just the keywords and summary
            return GenerateResponse(text=f"Keywords/Metrics: {keywords}\nSummary: {conversation_summary}")
        
        else:
            return "No response from the API."
        # if response.choices:
        #     summary = response.choices[0].text.strip()
        #     return GenerateResponse(text=summary)
        # else:
        #     return "No response from the API."

# QA Automation function
def qa_automation(text: str):
    questions = [
        "Did the agent verify the identity of the caller clearly? Provide Details.",
        #"Did the agent resolve the Primary reason of the call? Provide Details.",
        "Was the agent courteous? Provide Details.",
        # "Did the agent get a promise to pay? Provide Details."
        "Rate the satisfaction of the customer on a scale of 10. 0 being worse and 10 being best.",
        #"Rate the customer effort on a scale of 0-10. 0 being the most effort spent by the customer and 10 being least effort:",
        "How could the agent have made the call easier for the customer in upto 3 simple steps?"
    ]
    answers = []
    for question in questions:
        response = client.chat.completions.create(
            model = deployment_name,
            messages=[
                {"role": "system", "content": "You are an AI assistant evaluating a customer service call based on quality assurance criteria. Keep responses concise."},
                {"role": "user", "content": f"Question: {question}\nTranscript: {text}\nAnswer with a brief Yes/No and a short reason (max 2 sentences):"}
            ],
            max_tokens=100,
            temperature=0.5
        )
        if response.choices:
            answer = response.choices[0].message.content.strip()
            
            # Split the answer into the main answer (Yes/No) and the details
            if "Yes" in answer:
                main_answer, details = answer.split("Yes", 1)
                main_answer += "Yes"
            elif "No" in answer:
                main_answer, details = answer.split("No", 1)
                main_answer += "No"
            else:
                main_answer = answer
                details = "No additional details provided."
    
            # Format the answer with the "Comments:" tag in bold and on a new line
            # formatted_answer = f"**{question}** {main_answer}\n\n**Comments:** {details}" - change made on 8/27/2024
            # Change is to attain answers as a list of dictionaries (json) instead of a formatted string
            answer_dict = {
                "question": question,
                "answer": main_answer.strip(),
                "comments": details.strip()
            }
            answers.append(answer_dict) # Previously, we appended formatted_answer to 'answers'
        # return GenerateResponse(text = ("\n\n".join(answers)))
    return GenerateResponse3(text=answers)


def check_agent_steps(text: str):
    categories = {
        "Introduction": [
            "Greet the customer politely.",
            "Identify yourself and your company."        
        ],
        "Purpose of the Call": [
            "State the purpose of the call clearly."
        ],
        # "Account Review": [
        #     "Provide details of outstanding debt, the amount, due date, and charges or fees."
        # ],
        "Listen to the Customer": [
            "Give the customer an opportunity to explain their situation.",
            "Show empathy and understanding."
        ],
        # "Payment Discussion": [
        #     "Ask the customer about their ability to pay the outstanding amount."
        # ],
        # "Confirmation": [
        #     "Confirm the agreed-upon payment plan or next steps.",
        #     "Provide details on how the payment can be made (e.g., online, phone, mail)."
        # ],
        "Documentation": [
            "Offer to send a written confirmation of any agreements made during the call."
        ],
        "Closing": [
            "Thank the customer for their time and cooperation."
        ]
    }

    # Initialize a dictionary to store the results
    # results = {category: {} for category in categories}

    # # Iterate through each category and its questions
    # for category, questions in categories.items():
    #     for question in questions:
    #         # print("Checking question:", question)
    #         response = openai.Completion.create(
    #             engine="AmplifAI-Chat",
    #             prompt=f"Based on the transcript, did the agent follow the step: {question}? Provide a 'Yes' or 'No' answer.\nTranscript: {text}\nAnswer: ",
    #             temperature=0.2,
    #             max_tokens=10,
    #             top_p=1.0,
    #             frequency_penalty=0.0,
    #             presence_penalty=0.0,
    #             stop=["\n"]
    #         )
    #         answer = response.choices[0].text.strip()
    #         # emoji_answer = '✅' if answer == "Yes" else '❌'
    #         # print("Answer:", answer)
    #         results[category][question] = answer

    results = []

    for category, questions in categories.items():
        cat_map = {
            "title": category,
            "responses": []
        }
        for question in questions:
            response = client.chat.completions.create(
                model = deployment_name,
                messages=[
                    {"role": "system", "content": "You are an AI assistant evaluating call center agent performance. Keep answers concise."},
                    {"role": "user", "content": f"Based on the transcript, did the agent follow this step: '{question}'?\nTranscript: {text}\nAnswer with 'Right' or 'Wrong' only:"}
                ],
                temperature=0.2,
                max_tokens=10,
                top_p=1.0,
                frequency_penalty=0.0,
                presence_penalty=0.0,
                stop=["\n"]
            )
            answer = response.choices[0].message.content.strip()
            response_map = {
                "lable": question,
                "icon": answer
            }
            cat_map["responses"].append(response_map)
        results.append(cat_map)
        
    return GenerateResponse4(text=results)

# def custom_qa_automation(text: str, custom_question: str):
#     # Ensure the custom question ends with "Provide Details."
#     if not custom_question.strip().endswith("Provide Details."):
#         custom_question = custom_question.strip() + " Provide Details."

#     response = openai.Completion.create(
#         engine="AmplifAI-Chat",
#         prompt=f"Respons with a Yes/No followed by a reason. Question: {custom_question}\nTranscript: {text}\nAnswer:",
#         temperature=0.2,
#         max_tokens=100,
#         top_p=1.0,
#         frequency_penalty=0.0,
#         presence_penalty=0.0,
#         stop=["\n"]
#     )
#     answer = response.choices[0].text.strip()
#     # Split the answer into the main answer (Yes/No) and the details
#     if "Yes" in answer:
#         main_answer, details = answer.split("Yes", 1)
#         main_answer += "Yes"
#     elif "No" in answer:
#         main_answer, details = answer.split("No", 1)
#         main_answer += "No"
#     else:
#         main_answer = answer
#         details = "No additional details provided."

#     # Format the answer with the "Comments:" tag in bold and on a new line
#     formatted_answer = f"<p><b>Response:</b> {main_answer}</p><p><b>Comment:</b> {details}</p>" #f"**{custom_question}** {main_answer}\n\n**Comments:** {details}"

#     #return GenerateResponse(text=formatted_answer)
#     return formatted_answer # Returning string directly instead of being placed within the GenerateResponse class

def custom_qa_automation(text: str, custom_question: str):
    # Ensure the custom question ends with "Provide Details."
    if not custom_question.strip().endswith("Provide Details."):
        custom_question = custom_question.strip() + " Provide Details."

        response = client.chat.completions.create(
            model = deployment_name,
            messages=[
                {"role": "system", "content": "You are an AI assistant evaluating a customer service call. Keep responses concise."},
                {"role": "user", "content": f"Question: {custom_question}\nTranscript: {text}\nAnswer with a brief Yes/No and a short reason (max 1 sentence):"}
            ],
            temperature=0.2,
            max_tokens=100,
            top_p=1.0,
            frequency_penalty=0.0,
            presence_penalty=0.0,
            stop=["\n"]
        )
        answer = response.choices[0].message.content.strip()
        # Split the answer into the main answer (Yes/No) and the details
        if "Yes" in answer:
            main_answer, details = answer.split("Yes", 1)
            main_answer += "Yes"
        elif "No" in answer:
            main_answer, details = answer.split("No", 1)
            main_answer += "No"
        else:
            main_answer = answer
            details = "No additional details provided."
    
        # Format the answer with the "Comments:" tag in bold and on a new line
        formatted_answer = f"<p><b>Response:</b> {main_answer}</p><p><b>Comment:</b> {details}</p>" #f"**{custom_question}** {main_answer}\n\n**Comments:** {details}"

    #return GenerateResponse(text=formatted_answer)
    return formatted_answer # Returning string directly instead of being placed within the GenerateResponse class

# Transcription function
def transcription(file):
    try:
        deepgram = DeepgramClient(DEEPGRAM_API_KEY)
        if isinstance(file, str):  # File from folder
            with open(file, "rb") as f:
                buffer_data = f.read()
        else:  # Uploaded file
            buffer_data = file.file.read()

        payload = {"buffer": buffer_data}
        options = PrerecordedOptions(model="nova-2", smart_format=True, diarize=True, redact = ['SSN'])
        response = deepgram.listen.prerecorded.v("1").transcribe_file(payload, options)
        return response.results.channels[0].alternatives[0].paragraphs.transcript
    except Exception as e:
        print(f"Exception: {e}")
        return None

def transcription_content(file):
    try:
        deepgram = DeepgramClient(DEEPGRAM_API_KEY)
        if isinstance(file, str):  # File from folder
            with open(file, "rb") as f:
                buffer_data = f.read()
        else:  # Uploaded file
            buffer_data = file.file.read()

        payload = {"buffer": buffer_data}
        options = PrerecordedOptions(model="nova-2", smart_format=True, diarize=True)
        response = deepgram.listen.prerecorded.v("1").transcribe_file(payload, options)
        return response.results.channels[0].alternatives[0].transcript
    except Exception as e:
        print(f"Exception: {e}")
        return None



ALLOWED_EXTENSIONS = {".mp3", ".wav", ".wma"}

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Middleware to handle large request body size (e.g., up to 50MB)
class LargeRequestMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_body_size = request.headers.get('content-length')
        if request_body_size and int(request_body_size) > 50 * 1024 * 1024:  # 50MB limit
            return Response("Request body too large", status_code=413)
        response = await call_next(request)
        return response

# Add LargeRequestMiddleware to app
app.add_middleware(LargeRequestMiddleware)

@app.get("/", tags=["Home"])
def api_home():
    return {'detail': 'Welcome to FastAPI TextGen Tutorial!'}

@app.post("/api/summarize", summary="Generate text from prompt", tags=["Generate"], response_model=GenerateResponse)
def inference(input_prompt: Prompt):
    return summarize_text(text=input_prompt.text)

@app.post("/api/qautomation", summary="Generate text from prompt", tags=["Generate"], response_model=GenerateResponse3)
def inference(input_prompt: Prompt):
    return qa_automation(text=input_prompt.text)

@app.post("/api/verify-call-flow", summary="Generate text from prompt", tags=["Generate"], response_model=GenerateResponse4)
def inference(input_prompt: Prompt):
    return check_agent_steps(text=input_prompt.text)

# @app.post("/api/custom-qa", summary="Generate text from prompt", tags=["Generate"], response_model=GenerateResponse)
# def inference(input_prompt: Prompt, question: Prompt):
#     return custom_qa_automation(text=input_prompt.text, custom_question=question.text)

@app.post("/api/custom-qa", summary="Generate text from prompt", response_class=PlainTextResponse) # Attempt to only return HTML string instead of json output
def inference(input_prompt: Prompt, question: Prompt):
    return custom_qa_automation(text=input_prompt.text, custom_question=question.text)


@app.post("/api/transcribe-display", summary="Transcribe audio file", tags=["Transcription"])
async def transcribe_audio_route(file: UploadFile = File(...)):
    # Check if the file extension is allowed
    file_extension = os.path.splitext(file.filename)[1].lower()
    if file_extension not in ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail=f"File format not supported. Allowed formats: {', '.join(ALLOWED_EXTENSIONS)}")

    transcription_result = transcription(file)
    if transcription_result:
        return {"transcription": transcription_result}
    else:
        return {"error": "Transcription failed"}

@app.post("/api/transcribe-content", summary="Transcribe audio file", tags=["Transcription"])
async def transcribe_audio_route(file: UploadFile = File(...)):
    # Check if the file extension is allowed
    file_extension = os.path.splitext(file.filename)[1].lower()
    if file_extension not in ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail=f"File format not supported. Allowed formats: {', '.join(ALLOWED_EXTENSIONS)}")

    transcription_result = transcription_content(file)
    if transcription_result:
        return {"transcription": transcription_result}
    else:
        return {"error": "Transcription failed"}