Spaces:
Sleeping
Sleeping
import os | |
import json | |
from typing import List, Dict, Any, Optional | |
from pydantic import BaseModel, Field | |
from fastapi import FastAPI, HTTPException, Depends, Header | |
from fastapi.responses import JSONResponse | |
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials | |
from enum import Enum | |
from dotenv import load_dotenv | |
load_dotenv() | |
# LangChain and OpenAI imports | |
try: | |
from langchain_openai import ChatOpenAI | |
from langchain.prompts import ChatPromptTemplate | |
from langchain.output_parsers import PydanticOutputParser | |
from langchain_core.pydantic_v1 import BaseModel as LangChainBaseModel, Field as LangChainField | |
LANGCHAIN_AVAILABLE = True | |
except ImportError: | |
LANGCHAIN_AVAILABLE = False | |
print("Warning: LangChain not available. Install with: pip install langchain langchain-openai") | |
# Security configuration | |
API_KEY = os.getenv("API_KEY") | |
security = HTTPBearer() | |
# Initialize FastAPI app | |
app = FastAPI(title="Job Candidate Email Template Generator", version="1.0.0") | |
# Security dependency | |
async def verify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)): | |
"""Verify the API key from the Authorization header""" | |
if credentials.credentials != API_KEY: | |
raise HTTPException( | |
status_code=401, | |
detail="Invalid API key. Please provide a valid API key in the Authorization header.", | |
headers={"WWW-Authenticate": "Bearer"}, | |
) | |
return credentials.credentials | |
# Enums for sequence types | |
class SequenceType(str, Enum): | |
INITIAL_OUTREACH = "initial_outreach" | |
FOLLOW_UP = "follow_up" | |
REPLY_TO_RESPONSE = "reply_to_response" | |
class ConversationMessage(BaseModel): | |
sender: str = Field(..., description="Who sent the message (candidate/recruiter)") | |
content: str = Field(..., description="Message content") | |
timestamp: Optional[str] = Field(None, description="Message timestamp") | |
# Data Models | |
class EmailTemplate(BaseModel): | |
subject: Optional[str] = Field(None, description="Email subject line (only for initial outreach)") | |
body: str = Field(..., description="Email body content with HTML formatting") | |
variant: str = Field(..., description="Template variant (A, B, or C)") | |
sequence_type: str = Field(..., description="Type of email sequence") | |
class GenerateTemplatesRequest(BaseModel): | |
job_description: Optional[str] = Field(None, description="Job description to generate email templates for") | |
company_name: Optional[str] = Field(None, description="Company name (optional)") | |
role_title: Optional[str] = Field(None, description="Job title/role (optional)") | |
salary_range: Optional[str] = Field(None, description="Salary range (optional)") | |
location: Optional[str] = Field(None, description="Job location (optional)") | |
tech_stack: Optional[str] = Field(None, description="Technology stack (optional)") | |
sequence_type: SequenceType = Field(SequenceType.INITIAL_OUTREACH, description="Type of email sequence to generate") | |
conversation_history: Optional[List[ConversationMessage]] = Field(None, description="Previous conversation messages for reply generation") | |
candidate_response: Optional[str] = Field(None, description="Candidate's response for follow-up generation") | |
days_since_last_contact: Optional[int] = Field(None, description="Days since last contact (for follow-ups)") | |
class GenerateTemplatesResponse(BaseModel): | |
success: bool = Field(..., description="Whether the generation was successful") | |
templates: List[EmailTemplate] = Field(..., description="Generated email templates") | |
message: str = Field(..., description="Response message") | |
# LangChain Pydantic models for structured output | |
class EmailTemplateStructured(LangChainBaseModel): | |
"""Single email template for LangChain structured output""" | |
subject: Optional[str] = LangChainField(None, description="Engaging email subject line (only for initial outreach)") | |
body: str = LangChainField(description="HTML formatted email body with <br>, <p>, <strong> tags and {{first_name}} placeholder") | |
class EmailTemplatesStructured(LangChainBaseModel): | |
"""All three email templates for LangChain structured output""" | |
template_a: EmailTemplateStructured = LangChainField(description="Direct and professional approach template") | |
template_b: EmailTemplateStructured = LangChainField(description="Casual and conversational approach template") | |
template_c: EmailTemplateStructured = LangChainField(description="Value-focused approach highlighting benefits template") | |
async def generate_email_templates_with_llm(request: GenerateTemplatesRequest) -> List[EmailTemplate]: | |
"""Generate email templates using LangChain and OpenAI based on job description and sequence type""" | |
if not LANGCHAIN_AVAILABLE: | |
raise HTTPException(status_code=500, detail="LangChain not available. Please install langchain and langchain-openai") | |
try: | |
# Initialize OpenAI client | |
openai_api_key = os.getenv("OPENAI_API_KEY") | |
if not openai_api_key: | |
raise HTTPException(status_code=500, detail="OPENAI_API_KEY not set in environment variables") | |
# Initialize LLM with structured output | |
llm = ChatOpenAI( | |
model="gpt-4o-mini", | |
temperature=0.7, | |
openai_api_key=openai_api_key | |
) | |
# Configure LLM to use structured output | |
structured_llm = llm.with_structured_output(EmailTemplatesStructured) | |
# Get appropriate system prompt based on sequence type | |
system_prompt = get_system_prompt_for_sequence_type(request.sequence_type) | |
# Build context based on sequence type | |
context = build_context_for_sequence_type(request) | |
# Create the prompt template | |
prompt_template = ChatPromptTemplate.from_messages([ | |
("system", system_prompt), | |
("human", "Generate email templates for this situation:\n\n{context}") | |
]) | |
# Create the chain | |
chain = prompt_template | structured_llm | |
# Generate the structured output | |
result = await chain.ainvoke({"context": context}) | |
print(f"Structured output received successfully for {request.sequence_type}") | |
# Convert structured output to EmailTemplate objects | |
templates = [ | |
EmailTemplate( | |
subject=result.template_a.subject, | |
body=result.template_a.body, | |
variant="A", | |
sequence_type=request.sequence_type.value | |
), | |
EmailTemplate( | |
subject=result.template_b.subject, | |
body=result.template_b.body, | |
variant="B", | |
sequence_type=request.sequence_type.value | |
), | |
EmailTemplate( | |
subject=result.template_c.subject, | |
body=result.template_c.body, | |
variant="C", | |
sequence_type=request.sequence_type.value | |
) | |
] | |
return templates | |
except Exception as e: | |
print(f"Error generating templates with LLM: {str(e)}") | |
# Fallback to basic templates if LLM fails | |
return create_fallback_templates(request) | |
def get_system_prompt_for_sequence_type(sequence_type: SequenceType) -> str: | |
"""Get appropriate system prompt based on sequence type""" | |
base_prompt = """You are an expert recruitment email template generator, your name is Ali Taghikhani, CEO of SRN. Create 3 different email templates (A, B, C) for recruitment communication. | |
Each template should be professional, personalized, and follow this structure: | |
- Body: HTML-formatted email with proper <br> tags and <p> tags | |
- Include placeholders like {{first_name}} for personalization | |
- Professional but friendly tone | |
- Clear call-to-action | |
- Focus on building interest and trust | |
Template Guidelines: | |
1. Template A: Direct and professional approach - straight to the point, formal but friendly | |
2. Template B: More casual and conversational approach - friendly, informal, relatable | |
3. Template C: Value-focused approach highlighting benefits - emphasize growth, culture, perks | |
Use this sample structure as inspiration but create unique variations: | |
- Start with personalized greeting using {{first_name}} | |
- Brief introduction and reason for reaching out | |
- Key details about the role and company | |
- Clear call-to-action | |
- Professional sign-off | |
Make sure each template has a distinctly different tone and approach while maintaining professionalism.""" | |
if sequence_type == SequenceType.INITIAL_OUTREACH: | |
return base_prompt + """ | |
SPECIFIC GUIDELINES FOR INITIAL OUTREACH: | |
- This is the first contact with the candidate | |
- Include engaging subject lines for each template | |
- Focus on introducing the opportunity and building initial interest | |
- Don't mention salary range unless specifically provided | |
- Emphasize the role, company culture, and growth opportunities | |
- Make it easy for them to respond with their CV and interest""" | |
elif sequence_type == SequenceType.FOLLOW_UP: | |
return base_prompt + """ | |
SPECIFIC GUIDELINES FOR FOLLOW-UP EMAILS: | |
- This is a follow-up to a previous outreach that didn't get a response | |
- Do not include subject lines (these are reply emails) keep that empty | |
- Be polite and not pushy - acknowledge they might be busy | |
- Reference the previous contact and the opportunity | |
- Offer additional value or information | |
- Give them an easy way to respond or opt out | |
- Consider the timing (mention if it's been a few days/weeks) | |
- Keep it brief and respectful""" | |
elif sequence_type == SequenceType.REPLY_TO_RESPONSE: | |
return base_prompt + """ | |
SPECIFIC GUIDELINES FOR REPLYING TO CANDIDATE RESPONSES: | |
- This is a reply to a candidate who has responded to your outreach | |
- Do NOT include subject lines (these are reply emails) keep that empty | |
- Acknowledge their response and show enthusiasm | |
- Address any questions or concerns they raised | |
- Provide next steps in the process | |
- Be responsive to their level of interest | |
- Maintain the conversation flow naturally | |
- If they're interested, guide them to the next step | |
- If they're not interested, thank them politely and keep the door open""" | |
return base_prompt | |
def build_context_for_sequence_type(request: GenerateTemplatesRequest) -> str: | |
"""Build appropriate context based on sequence type""" | |
# Base job information | |
context_parts = [] | |
if request.job_description: | |
context_parts.append(f"Job Description: {request.job_description}") | |
else: | |
context_parts.append("Job Description: General recruitment opportunity") | |
if request.company_name: | |
context_parts.append(f"Company: {request.company_name}") | |
if request.role_title: | |
context_parts.append(f"Role: {request.role_title}") | |
if request.salary_range: | |
context_parts.append(f"Salary: {request.salary_range}") | |
if request.location: | |
context_parts.append(f"Location: {request.location}") | |
if request.tech_stack: | |
context_parts.append(f"Tech Stack: {request.tech_stack}") | |
# Add sequence-specific context | |
if request.sequence_type == SequenceType.FOLLOW_UP: | |
if request.days_since_last_contact: | |
context_parts.append(f"Days since last contact: {request.days_since_last_contact}") | |
if request.candidate_response: | |
context_parts.append(f"Previous candidate response: {request.candidate_response}") | |
context_parts.append("This is a follow-up email to a candidate who hasn't responded to the initial outreach.") | |
elif request.sequence_type == SequenceType.REPLY_TO_RESPONSE: | |
if request.conversation_history: | |
context_parts.append("Conversation History:") | |
for msg in request.conversation_history: | |
context_parts.append(f"- {msg.sender}: {msg.content}") | |
if request.candidate_response: | |
context_parts.append(f"Latest candidate response: {request.candidate_response}") | |
context_parts.append("This is a reply to a candidate who has responded to your outreach.") | |
return "\n".join(context_parts) | |
def create_fallback_templates(request: GenerateTemplatesRequest) -> List[EmailTemplate]: | |
"""Create fallback email templates when LLM fails""" | |
# Build job info string | |
job_info_parts = [] | |
if request.job_description: | |
job_info_parts.append(request.job_description) | |
else: | |
job_info_parts.append("exciting opportunity") | |
if request.company_name: | |
job_info_parts.insert(0, f"Company: {request.company_name}") | |
if request.role_title: | |
job_info_parts.insert(1, f"Role: {request.role_title}") | |
if request.salary_range: | |
job_info_parts.append(f"Salary: {request.salary_range}") | |
if request.location: | |
job_info_parts.append(f"Location: {request.location}") | |
if request.tech_stack: | |
job_info_parts.append(f"Tech Stack: {request.tech_stack}") | |
job_info = "<br>".join(job_info_parts) | |
# Create templates based on sequence type | |
templates = [] | |
if request.sequence_type == SequenceType.INITIAL_OUTREACH: | |
# Template A: Direct approach | |
templates.append(EmailTemplate( | |
subject="Exciting Opportunity: Senior Developer Position Available", | |
body=f"""<p>Hi {{{{first_name}}}},</p> | |
<p>I'm reaching out because I have an opportunity that aligns perfectly with your experience and expertise.</p> | |
<p>{job_info}</p> | |
<p>If this sounds interesting to you, I'd love to discuss the details further. Please send me your updated CV and salary expectations, and I'll fast-track your application to the hiring team.</p> | |
<p>Best regards,<br> | |
[Your Name]<br> | |
[Your Title]</p>""", | |
variant="A", | |
sequence_type=request.sequence_type.value | |
)) | |
# Template B: Casual approach | |
templates.append(EmailTemplate( | |
subject="Quick question about your next career move 🚀", | |
body=f"""<p>Hey {{{{first_name}}}},</p> | |
<p>Hope you're having a great day! I came across your profile and thought you might be interested in something exciting.</p> | |
<p>{job_info}</p> | |
<p>The team is amazing, and they're looking for someone just like you. Want to chat about it? Just shoot me your CV and let me know what you're looking for in your next role.</p> | |
<p>Cheers,<br> | |
[Your Name]</p>""", | |
variant="B", | |
sequence_type=request.sequence_type.value | |
)) | |
# Template C: Value-focused approach | |
templates.append(EmailTemplate( | |
subject="Transform Your Career: Join a Leading Tech Team", | |
body=f"""<p>Hi {{{{first_name}}}},</p> | |
<p>I'm reaching out with an opportunity that offers exceptional growth potential and the chance to work with cutting-edge technologies.</p> | |
<p>{job_info}</p> | |
<p><strong>Why this role stands out:</strong><br> | |
• Work with the latest technologies<br> | |
• Competitive compensation and benefits<br> | |
• Flexible work arrangements<br> | |
• Clear career progression path</p> | |
<p>If you're ready to take your career to the next level, I'd love to share more details. Please send me your resume and salary requirements.</p> | |
<p>Looking forward to connecting,<br> | |
[Your Name]<br> | |
[Your Title]</p>""", | |
variant="C", | |
sequence_type=request.sequence_type.value | |
)) | |
elif request.sequence_type == SequenceType.FOLLOW_UP: | |
days_text = f" {request.days_since_last_contact} days ago" if request.days_since_last_contact else "" | |
# Template A: Polite follow-up | |
templates.append(EmailTemplate( | |
subject="", | |
body=f"""<p>Hi {{{{first_name}}}},</p> | |
<p>I hope this email finds you well. I wanted to follow up on the opportunity I reached out about{days_text}.</p> | |
<p>{job_info}</p> | |
<p>I understand you're likely busy, but I wanted to make sure you had all the information you need. If you're interested, I'd be happy to discuss this further. If not, no worries at all - just let me know either way!</p> | |
<p>Best regards,<br> | |
[Your Name]</p>""", | |
variant="A", | |
sequence_type=request.sequence_type.value | |
)) | |
# Template B: Casual follow-up | |
templates.append(EmailTemplate( | |
subject="", | |
body=f"""<p>Hey {{{{first_name}}}},</p> | |
<p>Just wanted to check in and see if you had a chance to look at the opportunity I mentioned{days_text}.</p> | |
<p>{job_info}</p> | |
<p>No pressure at all - just wanted to make sure you didn't miss it! Let me know if you're interested or if you have any questions.</p> | |
<p>Cheers,<br> | |
[Your Name]</p>""", | |
variant="B", | |
sequence_type=request.sequence_type.value | |
)) | |
# Template C: Value-added follow-up | |
templates.append(EmailTemplate( | |
subject="", | |
body=f"""<p>Hi {{{{first_name}}}},</p> | |
<p>I hope you're doing well. I wanted to follow up on the opportunity I shared{days_text} and provide some additional context.</p> | |
<p>{job_info}</p> | |
<p><strong>What's new:</strong><br> | |
• The team is growing rapidly<br> | |
• New exciting projects are starting soon<br> | |
• Flexible work arrangements available</p> | |
<p>If this sounds interesting, I'd love to discuss it further. If not, I completely understand!</p> | |
<p>Best regards,<br> | |
[Your Name]</p>""", | |
variant="C", | |
sequence_type=request.sequence_type.value | |
)) | |
elif request.sequence_type == SequenceType.REPLY_TO_RESPONSE: | |
# Template A: Professional reply | |
templates.append(EmailTemplate( | |
subject="", | |
body=f"""<p>Hi {{{{first_name}}}},</p> | |
<p>Thank you for your response! I'm excited to hear about your interest.</p> | |
<p>{job_info}</p> | |
<p>Next steps would be to schedule a brief call to discuss the role in more detail and answer any questions you might have. When would be a good time for you?</p> | |
<p>Looking forward to our conversation,<br> | |
[Your Name]</p>""", | |
variant="A", | |
sequence_type=request.sequence_type.value | |
)) | |
# Template B: Friendly reply | |
templates.append(EmailTemplate( | |
subject="", | |
body=f"""<p>Hey {{{{first_name}}}},</p> | |
<p>Awesome! Thanks for getting back to me. I'm glad you're interested.</p> | |
<p>{job_info}</p> | |
<p>Let's set up a quick chat to go over the details and see if it's a good fit. What's your schedule like this week?</p> | |
<p>Talk soon,<br> | |
[Your Name]</p>""", | |
variant="B", | |
sequence_type=request.sequence_type.value | |
)) | |
# Template C: Detailed reply | |
templates.append(EmailTemplate( | |
subject="", | |
body=f"""<p>Hi {{{{first_name}}}},</p> | |
<p>Excellent! Thank you for your interest. I'm looking forward to discussing this opportunity with you.</p> | |
<p>{job_info}</p> | |
<p><strong>What happens next:</strong><br> | |
• Brief 15-minute call to discuss the role<br> | |
• Technical assessment (if interested)<br> | |
• Team interview<br> | |
• Offer discussion</p> | |
<p>When would be convenient for you to have our initial call?</p> | |
<p>Best regards,<br> | |
[Your Name]</p>""", | |
variant="C", | |
sequence_type=request.sequence_type.value | |
)) | |
return templates | |
async def generate_email_templates(request: GenerateTemplatesRequest, api_key: str = Depends(verify_api_key)): | |
"""Generate 3 variants of email templates for job candidates | |
This endpoint generates professional email templates for different types of recruitment communication | |
based on the provided parameters and sequence type. | |
Parameters: | |
- job_description: Detailed description of the job/role (optional, defaults to general opportunity) | |
- company_name: Name of the company (optional) | |
- role_title: Job title/role (optional) | |
- salary_range: Salary range (optional) | |
- location: Job location (optional) | |
- tech_stack: Technology stack (optional) | |
- sequence_type: Type of email sequence (initial_outreach, follow_up, reply_to_response) - defaults to initial_outreach | |
- conversation_history: Previous conversation messages for reply generation (optional) | |
- candidate_response: Candidate's response for follow-up generation (optional) | |
- days_since_last_contact: Days since last contact for follow-ups (optional) | |
Returns: | |
- 3 email templates (A, B, C) with different approaches and tones based on sequence type | |
""" | |
try: | |
# Generate templates using LLM | |
templates = await generate_email_templates_with_llm(request) | |
return GenerateTemplatesResponse( | |
success=True, | |
templates=templates, | |
message=f"Successfully generated {len(templates)} email templates" | |
) | |
except HTTPException: | |
# Re-raise HTTP exceptions | |
raise | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=f"Unexpected error: {str(e)}") | |
async def health_check(): | |
"""Health check endpoint (no authentication required)""" | |
return {"status": "healthy", "langchain_available": LANGCHAIN_AVAILABLE} | |
async def health_check_secure(api_key: str = Depends(verify_api_key)): | |
"""Secure health check endpoint (requires API key)""" | |
return { | |
"status": "healthy", | |
"langchain_available": LANGCHAIN_AVAILABLE, | |
"api_key_valid": True | |
} | |
async def root(): | |
"""Root endpoint with API information""" | |
return { | |
"message": "Job Candidate Email Template Generator API", | |
"version": "1.0.0", | |
"authentication": { | |
"type": "Bearer Token", | |
"required": "Yes (for /generate-email-templates and /health/secure)", | |
"header": "Authorization: Bearer <api_key>" | |
}, | |
"endpoints": { | |
"generate_templates": "/generate-email-templates (requires API key)", | |
"health": "/health (no auth required)", | |
"health_secure": "/health/secure (requires API key)" | |
}, | |
"usage": { | |
"POST /generate-email-templates": "Generate 3 email template variants for different recruitment scenarios" | |
}, | |
"sequence_types": { | |
"initial_outreach": "First contact with potential candidates", | |
"follow_up": "Follow-up emails for candidates who haven't responded", | |
"reply_to_response": "Replies to candidates who have responded to outreach" | |
} | |
} | |
# Example usage for testing | |
if __name__ == "__main__": | |
import uvicorn | |
# Example requests for testing different sequence types | |
example_requests = { | |
"minimal_request": { | |
"sequence_type": "initial_outreach" | |
}, | |
"initial_outreach": { | |
"job_description": "We're looking for a Senior Full-Stack Developer to join our growing team.", | |
"company_name": "TechCorp", | |
"role_title": "Senior Full-Stack Developer", | |
"salary_range": "€80,000 - €100,000", | |
"location": "Remote (EU timezone)", | |
"tech_stack": "React, Node.js, PostgreSQL, AWS", | |
"sequence_type": "initial_outreach" | |
}, | |
"follow_up": { | |
"job_description": "We're looking for a Senior Full-Stack Developer to join our growing team.", | |
"company_name": "TechCorp", | |
"role_title": "Senior Full-Stack Developer", | |
"salary_range": "€80,000 - €100,000", | |
"location": "Remote (EU timezone)", | |
"tech_stack": "React, Node.js, PostgreSQL, AWS", | |
"sequence_type": "follow_up", | |
"days_since_last_contact": 7 | |
}, | |
"reply_to_response": { | |
"job_description": "We're looking for a Senior Full-Stack Developer to join our growing team.", | |
"company_name": "TechCorp", | |
"role_title": "Senior Full-Stack Developer", | |
"salary_range": "€80,000 - €100,000", | |
"location": "Remote (EU timezone)", | |
"tech_stack": "React, Node.js, PostgreSQL, AWS", | |
"sequence_type": "reply_to_response", | |
"conversation_history": [ | |
{ | |
"sender": "recruiter", | |
"content": "Hi John, I have an exciting opportunity for a Senior Full-Stack Developer role at TechCorp." | |
}, | |
{ | |
"sender": "candidate", | |
"content": "Hi Ali, thanks for reaching out! I'm definitely interested. Could you tell me more about the team and the tech stack?" | |
} | |
], | |
"candidate_response": "Hi Ali, thanks for reaching out! I'm definitely interested. Could you tell me more about the team and the tech stack?" | |
} | |
} | |
print("Example requests for different sequence types:") | |
for seq_type, request in example_requests.items(): | |
print(f"\n{seq_type.upper()}:") | |
print(json.dumps(request, indent=2)) | |
uvicorn.run(app, host="0.0.0.0", port=8000) |