final_professional_persona_agent.py

import os

import json

import requests

from groq import Groq

from PyPDF2 import PdfReader

import gradio as gr

from openai import OpenAI

from dotenv import load_dotenv

load_dotenv(override=True)

# Get API keys with explicit error handling
def get_required_env_var(var_name):
    value = os.getenv(var_name)
    if not value:
        raise ValueError(f"{var_name} environment variable is not set. Please set it in your Hugging Face Space secrets.")
    return value

try:
    # Get required API keys
    groq_api_key = get_required_env_var('GROQ_API_KEY')
    openai_api_key = get_required_env_var('OPENAI_API_KEY')
    
    # Optional API keys
    pushover_user_key = os.getenv('PUSHOVER_USER_KEY')
    pushover_api_key = os.getenv('PUSHOVER_API_KEY')
    pushover_url = 'https://api.pushover.net/1/messages.json'

    # Initialize clients
    groq_client = Groq(api_key=groq_api_key)
    openai_client = OpenAI(api_key=openai_api_key)
    
except Exception as e:
    print(f"Error during initialization: {str(e)}")
    raise

def push_message(message):

    print(f"Pushing message: {message}")

    payload = {

        "user": pushover_user_key,

        "token": pushover_api_key,

        "message": message

    }
    response = requests.post(pushover_url, data=payload)

    return response.json()

def register_user_details(email, name="not provided", notes="not provided"):

    push_message(f"Registering interest from {name} with {email} and notes {notes}")

    return {"Registered": "Ok"}

def register_unknown_question(question):

    print(f"Registering {question} which was asked and really challenging even for me to answer")

    return {"Registered": "Ok"}

register_user_details_json = {


                            "name" : "register_user_details",

                            "description" : " Use this tool to register the user who is interested to be in touch provided the email address",

                            "parameters" : {

                                            "email" : {

                                                       "type" : "string",

                                                       "description" :  "The email address of this user"
                                            },

                                            "name" : {

                                                      "type" : "string",

                                                      "description" : "The name of the user, if they provided it"
                                            },

                                            "notes" : {

                                                       "type" : "string",

                                                       "description" : "Any interesting information about the conversation that is worth keeping a note of to enhance context"
                                            }

                            },

                            "required" : ["email"],

                            "additionalProperties" : False


}

register_unknown_question_json = {


                                  "name" : "register_unknown_question",

                                  "description" : "Use this tool to register any question that you could not answer as you were not 100% confident.",

                                  "parameters" : {


                                                  "type" : "object",

                                                  "properties" : {


                                                                  "question"  : {

                                                                                 "type" : "string",

                                                                                 "description" : "The question which was challenging to answer"
                                                                  },

                                                  },

                                                  "required" : ["question"],

                                                  "additionalProperties" : False
                                  }

}

llm_tools = [{"type" : "function", "function" : register_user_details_json},

             {"type" : "function", "function" : register_unknown_question_json}]


class FaiazAhmed:

     def __init__(self):

      self.groq = groq_client
    
      self.name = "Faiaz Ahmed"
    
      # Initialize with default values
      self.linkedin_profile = ""
      self.professional_summary = ""

      # Debug: Print current working directory and list files
      print(f"Current working directory: {os.getcwd()}")
      print("Contents of current directory:")
      print(os.listdir("."))
      print("\nContents of FaiazDocs directory:")
      try:
          print(os.listdir("./FaiazDocs"))
      except Exception as e:
          print(f"Error listing FaiazDocs directory: {str(e)}")

      try:
          # Try to read LinkedIn profile PDF
          pdf_path = "./FaiazDocs/LinkedIn_Profile_PDF.pdf"
          print(f"\nAttempting to read PDF from: {pdf_path}")
          print(f"File exists: {os.path.exists(pdf_path)}")
          pdf_reader = PdfReader(pdf_path)
          for page in pdf_reader.pages:
              text = page.extract_text()
              if text:
                  self.linkedin_profile += text
      except FileNotFoundError as e:
          print(f"Error reading LinkedIn profile: {str(e)}")
          self.linkedin_profile = "Experienced software engineer with expertise in AI and machine learning."
      except Exception as e:
          print(f"Unexpected error reading PDF: {str(e)}")
          self.linkedin_profile = "Experienced software engineer with expertise in AI and machine learning."

      try:
          # Try to read summary file
          summary_path = "./FaiazDocs/summary.txt"
          print(f"\nAttempting to read summary from: {summary_path}")
          print(f"File exists: {os.path.exists(summary_path)}")
          with open(summary_path, "r", encoding="utf-8") as f:
              self.professional_summary = f.read()
      except FileNotFoundError as e:
          print(f"Error reading summary: {str(e)}")
          self.professional_summary = "Passionate about AI and software development with a focus on creating innovative solutions."
      except Exception as e:
          print(f"Unexpected error reading summary: {str(e)}")
          self.professional_summary = "Passionate about AI and software development with a focus on creating innovative solutions."

     def utilize_tools(self, tools):

         results = []

         for tool in tools:

            tool_name = tool.function.name

            tool_arguments = json.loads(tool.function.arguments)

         print(f"Tool loaded: {tool_name}", flush = True)

         if tool_name == "register_user_details":

          result = register_user_details(**tool_arguments)

          results.append({"role" : "tool", "content" : json.dumps(result), "tool_call_id" : tool.id})

         elif tool_name == "register_unknown_question":

             result = register_unknown_question(**tool_arguments)

             results.append({"role" : "tool", "content" : json.dumps(result), "tool_call_id" : tool.id})

         return results             
    

     def agent_prompt(self):

       agent_prompt = f"""

       You are taking up the identity as {self.name} and your job is to answer questions to users who comes across {self.name}'s professional profile relating to {self.name}'s background, experience and skills.

       Be as faithful, convincing and genuine as possible. Assume that you are being interviewed by a recruiter or a hiring manager who is specifically hiring for Tech roles with you being provided with self.{self.name}'s professional summary and linkedin profile, prove them why you are different from other potential candidates and why they should hire you.

       If you unable give an answer to a specific query to a question, use your register_unknown_question tool to register the question that you could not answer, even if it's something trivial and unrelated to career.

       While discussion, if the user is taking more interest, then try your ultimate best to steer the user towards getting in touch with {self.name} via email, ask for their email and register it using your register_user_details tool.

"""

       agent_prompt += f"Professional Summary of {self.name} : {self.professional_summary}"

       agent_prompt += f"LinkedIn Profile of {self.name} : {self.linkedin_profile}"

       agent_prompt += f"With this provided professional context, chat and interact with a friendly, professional tone with the user, keeping in mind that you are {self.name}"

       return agent_prompt

     def initialize_chat(self, message, history=None):
        
        if history is None:
            
            history = []
            
        filtered_history = []

        for msg in history:
            
            if isinstance(msg, dict):
                
                role = msg.get("role")

                content = msg.get("content")

                if role in ["user", "assistant"] and isinstance(content, str):
                    
                    filtered_history.append({"role": role, "content": content})

        messages = [{"role": "system", "content": self.agent_prompt()}] + filtered_history + [{"role": "user", "content": message}]

        done = False

        while not done:
            
            chatbot_response = groq_client.chat.completions.create(
               
                model='llama-3.3-70b-versatile',

                messages=messages,

                tools=llm_tools
            )

            finish_reason = chatbot_response.choices[0].finish_reason

            if finish_reason == "tool_calls":
                
                message = chatbot_response.choices[0].message

                tool_calls = message.tool_calls

                result = self.utilize_tools(tool_calls)

                messages.append(message)

                messages.extend(result)
            else:
                done = True

        return chatbot_response.choices[0].message.content

#def main():
 
    # Test the setup
  #  push_message("System initialized successfully!")
    
    # Create a simple Gradio interface
   # with gr.Blocks() as demo:
        #gr.Markdown("# Professional Persona Agent")
        #with gr.Row():
           # email = gr.Textbox(label="Email")
            #name = gr.Textbox(label="Name (optional)")
            #notes = gr.Textbox(label="Notes (optional)")
        #submit_btn = gr.Button("Register Interest")
        #output = gr.JSON()
        
        #submit_btn.click(
            #fn=register_user_details,
            #inputs=[email, name, notes],
            #outputs=output
        #)
    
    #demo.launch()

if __name__ == "__main__":
    
    faiaz = FaiazAhmed()

    gr.ChatInterface(faiaz.initialize_chat, type = "messages").launch(debug = True)