HyderabadIndiaChapter_ChatbotInterviewPreparation/application/main.py

import asyncio
from database.chroma import get_chroma_collection, get_relevant_qa
from database.duckdb import get_sample_questions
from agents.response_evaluator import ResponseEvaluator
from agents.question_generator import QuestionGenerator
from multiprocessing import Pool
from stt_tts.stt_tts import STT_TTS


def generate_questions(skill_tested: str, sample_questions: list) -> str:
    """
    Generate questions based on the skill to be tested

    Parameters:
    -----------
    sample_questions: List of samples questions if any
    skill_tested: List of skills that needs to be tested

    Returns:
    -----------
    List of questions that are generated by the question generator
    """
    question_generator = QuestionGenerator()
    question_generator_question = question_generator.generate_question(
        skill_tested, sample_questions
    )
    return question_generator_question


async def process_questions(questions: list, skills_to_test: str) -> list:
    """
    Async function to process questions based on the skills_to_test

    Parameters:
    -----------
    questions: List of questions to be processed
    skills_to_test: List of skills that needs to be tested

    Returns:
    -----------
    List of feedback along with questions and user responses
    """
    tasks = []
    db = get_chroma_collection("question_embeddings_v2")
    for question, skill_to_test in zip(questions, skills_to_test):
        print("Question: ", question)
        stt_tts = STT_TTS()
        question_audio = stt_tts.text_to_speech(question)
        print("Question audio: ", question_audio)
        try:
            relevant_qa = get_relevant_qa(db, question, skill_question_map[skill_to_test][1])
        except:
            print("No relavant questions found for this question generated by LLM")
            relevant_qa = []

        # print(relevant_qa)
        user_response_audio = input("Enter the audio wav filepath of response for the question:")
        user_response = stt_tts.speech_to_text(user_response_audio)

        # Call the async function while continuing to the next question
        evaluator = ResponseEvaluator()
        task = asyncio.create_task(
            evaluator.evaluate_response(question, user_response, skill_to_test, relevant_qa)
        )
        tasks.append(task)

    results = await asyncio.gather(*tasks)

    return results


if __name__ == "__main__":
    skills_to_test = ["social", "speaking", "management", "technical"]
    company = "Salesforce"
    profile = "Business Development Manager"
    skill_question_map = {}
    for skill_to_test in skills_to_test:
        skill_question_map[skill_to_test] = get_sample_questions(company, profile, skill_to_test, 5)
    print(skill_question_map)
    sample_questions_list = [(k, v[0]) for k, v in skill_question_map.items()]
    # Creating a process Pool to generate all questions in parallel
    with Pool(4) as p:
        questions = p.starmap(generate_questions, sample_questions_list)
    # print(questions)
    feedback = asyncio.run(process_questions(questions, skills_to_test))
    print(feedback)