import os import requests import gradio as gr from openai import OpenAI import logging # Configure logging logging.basicConfig(level=logging.INFO) # Fetch API keys from environment variables OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") PROXYCURL_API_KEY = os.getenv("PROXYCURL_API_KEY") FIRECRAWL_API_KEY = os.getenv("FIRECRAWL_API_KEY") # Function to fetch LinkedIn data using the Proxycurl API def fetch_linkedin_data(linkedin_url): api_key = os.getenv("PROXYCURL_API_KEY") headers = {'Authorization': f'Bearer {api_key}'} api_endpoint = 'https://nubela.co/proxycurl/api/v2/linkedin' logging.info("Fetching LinkedIn data...") response = requests.get(api_endpoint, params={'url': linkedin_url}, headers=headers, timeout=10) # Adding a timeout for safety if response.status_code == 200: logging.info("LinkedIn data fetched successfully.") return response.json() else: logging.error(f"Error fetching LinkedIn data: {response.text}") return {"error": f"Error fetching LinkedIn data: {response.text}"} # Function to fetch company information using Firecrawl API def fetch_company_info(company_url): api_key = os.getenv("FIRECRAWL_API_KEY") headers = { 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json' } api_endpoint = 'https://api.firecrawl.dev/v1/crawl' data = { "url": company_url, "limit": 100, "scrapeOptions": { "formats": ["markdown", "html"] } } logging.info("Fetching company information...") response = requests.post(api_endpoint, json=data, headers=headers, timeout=15) # Adding a timeout for safety if response.status_code == 200: logging.info("Company information fetched successfully.") return response.json() else: logging.error(f"Error fetching company information: {response.text}") return {"error": f"Error fetching company information: {response.text}"} # Function to structure the email using the "Start with Why" model def structure_email(user_data, linkedin_info, company_info): why = f"I am passionate about {company_info.get('mission', 'your mission')} because it aligns with my experience as {linkedin_info.get('current_role', 'a professional')}." how = f"My skills in {user_data['role']} match the requirements and goals of your organization." what = f"I can bring my experience in {linkedin_info.get('skills', 'relevant skills')} to help achieve {company_info.get('goal', 'your company goals')}." structured_input = f"{why}\n\n{how}\n\n{what}" return structured_input # Function to generate email content using Nvidia Nemotron LLM (non-streaming for simplicity) def generate_email_content(api_key, prompt): client = OpenAI( base_url="https://integrate.api.nvidia.com/v1", api_key=api_key ) logging.info("Generating email content...") try: response = client.chat.completions.create( model="nvidia/llama-3.1-nemotron-70b-instruct", messages=[ {"role": "user", "content": prompt} ], temperature=0.5, top_p=1, max_tokens=1024, stream=False # Disable streaming for simplicity ) email_content = response['choices'][0]['message']['content'] logging.info("Email content generated successfully.") return email_content except Exception as e: logging.error(f"Error generating email content: {e}") return "Error generating email content." # Function to validate the generated email for professional tone and completeness def validate_email(email_content): return "Why" in email_content and "How" in email_content and "What" in email_content # Custom Agent class to simulate behavior similar to OpenAI's Swarm framework class Agent: def __init__(self, name, instructions, user_data): self.name = name self.instructions = instructions self.user_data = user_data def act(self): if self.name == "Data Collection Agent": linkedin_info = fetch_linkedin_data(self.user_data['linkedin_url']) company_info = fetch_company_info(self.user_data['company_url']) return linkedin_info, company_info elif self.name == "Email Generation Agent": user_data = self.user_data['user_data'] linkedin_info = self.user_data['linkedin_info'] company_info = self.user_data['company_info'] prompt = structure_email(user_data, linkedin_info, company_info) email_content = generate_email_content(OPENAI_API_KEY, prompt) return email_content # Simulated Swarm class to manage agents class Swarm: def __init__(self): self.agents = [] def add_agent(self, agent): self.agents.append(agent) def run(self): for agent in self.agents: if agent.name == "Data Collection Agent": linkedin_info, company_info = agent.act() if "error" in linkedin_info or "error" in company_info: return "Error fetching data. Please check the LinkedIn and company URLs." return linkedin_info, company_info # Function that integrates the agents and manages iterations def run_agent(name, email, phone, linkedin_url, company_url, role): user_data = { "name": name, "email": email, "phone": phone, "linkedin_url": linkedin_url, "company_url": company_url, "role": role } # Create a Swarm and add the Data Collection Agent email_swarm = Swarm() data_collection_agent = Agent("Data Collection Agent", "Collect user inputs and relevant data", user_data) email_swarm.add_agent(data_collection_agent) # Get data from the Data Collection Agent linkedin_info, company_info = email_swarm.run() if isinstance(linkedin_info, str): # If an error message is returned return linkedin_info # Create a structured dictionary for the Email Generation Agent agent_data = { "user_data": user_data, "linkedin_info": linkedin_info, "company_info": company_info } # Pass the collected data to the Email Generation Agent email_agent = Agent("Email Generation Agent", "Generate the email content", agent_data) email_content = email_agent.act() # Validate and refine the email using a ReAct pattern with a maximum of 3 iterations for i in range(3): if validate_email(email_content): return email_content else: refined_prompt = f"Refine: {structure_email(user_data, linkedin_info, company_info)}" email_content = generate_email_content(OPENAI_API_KEY, refined_prompt) return "Unable to generate a valid email after 3 attempts." # Set up the Gradio interface final_interface = gr.Interface( fn=run_agent, inputs=[ gr.Textbox(label="Name"), gr.Textbox(label="Email"), gr.Textbox(label="Phone Number"), gr.Textbox(label="LinkedIn Profile URL"), gr.Textbox(label="Company URL or Name"), gr.Textbox(label="Role Being Applied For") ], outputs="text", title="Email Writing AI Agent", description="Autonomously generate a professional email tailored to the job application." ) if __name__ == "__main__": final_interface.launch()