import gradio as gr import requests import os import json class AutonomousEmailAgent: def __init__(self, linkedin_url, company_name, role, word_limit, user_name, email, phone, linkedin): self.linkedin_url = linkedin_url self.company_name = company_name self.role = role self.word_limit = word_limit self.user_name = user_name self.email = email self.phone = phone self.linkedin = linkedin self.bio = None self.skills = [] self.experiences = [] self.company_info = None self.role_description = None self.company_url = None # Reason and Act via LLM: Let the LLM control reasoning and actions dynamically def autonomous_reasoning(self): print("Autonomous Reasoning: Letting the LLM fully reason and act on available data...") reasoning_prompt = f""" You are an autonomous agent responsible for generating a job application email. Here’s the current data: - LinkedIn profile: {self.linkedin_url} - Company Name: {self.company_name} - Role: {self.role} - Candidate's Bio: {self.bio} - Candidate's Skills: {', '.join(self.skills)} - Candidate's Experiences: {', '.join([exp['title'] for exp in self.experiences])} - Company Information: {self.company_info} - Role Description: {self.role_description} Based on this data, decide if it is sufficient to generate the email. If some information is missing or insufficient, respond with: 1. "scrape" to fetch more data from the company website. 2. "generate_email" to proceed with the email generation. 3. "fallback" to use default values. After generating the email, reflect on whether the content aligns with the role and company and whether any improvements are needed. Respond clearly with one of the above options. """ return self.send_request_to_llm(reasoning_prompt) # Send request to Groq Cloud LLM with enhanced debugging and error handling def send_request_to_llm(self, prompt): print("Sending request to Groq Cloud LLM...") api_key = os.getenv("GROQ_API_KEY") if not api_key: print("Error: API key not found. Please set the GROQ_API_KEY environment variable.") return "Error: API key not found." headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } data = { "model": "llama-3.1-70b-versatile", # Model name "messages": [{"role": "user", "content": prompt}] } response = requests.post("https://api.groq.com/openai/v1/chat/completions", headers=headers, json=data) print(f"Status Code: {response.status_code}") if response.status_code == 200: result = response.json() content = "".join([chunk.get("choices", [{}])[0].get("delta", {}).get("content", "") for chunk in result]) print(content) return self.act_on_llm_instructions(content) else: print(f"Error: Unable to connect to Groq Cloud LLM. Status Code: {response.status_code}, Response: {response.text}") return "Error: Unable to generate response." # Function to act on the LLM's structured instructions def act_on_llm_instructions(self, reasoning_output): instruction = reasoning_output.lower().strip() if "scrape" in instruction: self.fetch_company_url() if self.company_url: self.fetch_company_info_with_firecrawl(self.company_url) return self.autonomous_reasoning() elif "generate_email" in instruction: return self.generate_email() elif "fallback" in instruction: print("Action: Using fallback values for missing data.") if not self.company_info: self.company_info = "A leading company in its field." if not self.role_description: self.role_description = f"The role of {self.role} involves leadership and team management." return self.generate_email() else: print("Error: Unrecognized instruction from LLM. Proceeding with available data.") return self.generate_email() # Fetch company URL using SERP API def fetch_company_url(self): serp_api_key = os.getenv("SERP_API_KEY") print(f"Fetching company URL for {self.company_name} using SERP API...") serp_url = f"https://serpapi.com/search.json?q={self.company_name}&api_key={serp_api_key}&num=1" response = requests.get(serp_url) if response.status_code == 200: serp_data = response.json() if 'organic_results' in serp_data and len(serp_data['organic_results']) > 0: self.company_url = serp_data['organic_results'][0]['link'] print(f"Found company URL: {self.company_url}") else: print("No URL found for the company via SERP API.") self.company_url = None else: print(f"Error fetching company URL: {response.status_code}") # Fetch LinkedIn data via Proxycurl def fetch_linkedin_data(self): proxycurl_api_key = os.getenv("PROXYCURL_API_KEY") if not self.linkedin_url: print("Action: No LinkedIn URL provided, using default bio.") self.bio = "A professional with diverse experience." self.skills = ["Adaptable", "Hardworking"] self.experiences = ["Worked across various industries"] else: print("Action: Fetching LinkedIn data via Proxycurl.") headers = {"Authorization": f"Bearer {proxycurl_api_key}"} url = f"https://nubela.co/proxycurl/api/v2/linkedin?url={self.linkedin_url}" response = requests.get(url, headers=headers) if response.status_code == 200: data = response.json() self.bio = data.get("summary", "No bio available") self.skills = data.get("skills", []) self.experiences = data.get("experiences", []) else: print("Error: Unable to fetch LinkedIn profile. Using default bio.") self.bio = "A professional with diverse experience." self.skills = ["Adaptable", "Hardworking"] self.experiences = ["Worked across various industries"] # Fetch company information via Firecrawl API using company URL def fetch_company_info_with_firecrawl(self, company_url): firecrawl_api_key = os.getenv("FIRECRAWL_API_KEY") print(f"Fetching company info for {company_url} using Firecrawl.") headers = {"Authorization": f"Bearer {firecrawl_api_key}"} firecrawl_url = "https://api.firecrawl.dev/v1/scrape" data = {"url": company_url, "patterns": ["description", "about", "careers", "company overview"]} response = requests.post(firecrawl_url, json=data, headers=headers) if response.status_code == 200: firecrawl_data = response.json() self.company_info = firecrawl_data.get("description", "No detailed company info available.") print(f"Company info fetched: {self.company_info}") else: print(f"Error: Unable to fetch company info via Firecrawl. Status code: {response.status_code}") self.company_info = "A leading company in its field." # Final Action: Generate the email using Groq Cloud LLM def generate_email(self): print("Action: Generating the email using Groq Cloud LLM with the gathered information.") prompt = f""" Write a professional job application email applying for the {self.role} position at {self.company_name}. The email should follow the "Start with Why" approach: 1. **Why**: Explain why the candidate is passionate about this role and company. 2. **How**: Highlight the candidate’s skills and experiences. 3. **What**: Provide examples of past achievements. 4. **Call to Action**: Request a meeting or discussion. - LinkedIn bio: {self.bio} - Skills: {', '.join(self.skills)} - Experience: {', '.join([exp['title'] for exp in self.experiences])} - Company information: {self.company_info} Signature: Best regards, {self.user_name} Email: {self.email} Phone: {self.phone} LinkedIn: {self.linkedin} Limit the email to {self.word_limit} words. """ return self.send_request_to_llm(prompt) # Main loop following ReAct pattern def run(self): self.fetch_linkedin_data() return self.autonomous_reasoning() # Gradio UI setup remains the same def gradio_ui(): # Input fields name_input = gr.Textbox(label="Your Name", placeholder="Enter your name") company_input = gr.Textbox(label="Company Name or URL", placeholder="Enter the company name or website URL") role_input = gr.Textbox(label="Role Applying For", placeholder="Enter the role you are applying for") email_input = gr.Textbox(label="Your Email Address", placeholder="Enter your email address") phone_input = gr.Textbox(label="Your Phone Number", placeholder="Enter your phone number") linkedin_input = gr.Textbox(label="Your LinkedIn URL", placeholder="Enter your LinkedIn profile URL") word_limit_slider = gr.Slider(minimum=50, maximum=300, step=10, label="Email Word Limit", value=150) email_output = gr.Textbox(label="Generated Email", placeholder="Your generated email will appear here", lines=10) def create_email(name, company_name, role, email, phone, linkedin_url, word_limit): agent = AutonomousEmailAgent(linkedin_url, company_name, role, word_limit, name, email, phone, linkedin_url) return agent.run() demo = gr.Interface( fn=create_email, inputs=[name_input, company_input, role_input, email_input, phone_input, linkedin_input, word_limit_slider], outputs=[email_output], title="Email Writing AI Agent with ReAct", description="Generate a professional email for a job application using LinkedIn data, company info, and role description.", allow_flagging="never" ) demo.launch() if __name__ == "__main__": gradio_ui()