Sequence_Generator / app_copy.py
ak0601's picture
Upload 2 files
366097c verified
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
@app.post("/generate-email-templates", response_model=GenerateTemplatesResponse)
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)}")
@app.get("/health")
async def health_check():
"""Health check endpoint (no authentication required)"""
return {"status": "healthy", "langchain_available": LANGCHAIN_AVAILABLE}
@app.get("/health/secure")
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
}
@app.get("/")
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)