Building an Intelligent Email and Job Application Tracker with LangChain & LangGraph

# Create the graph  
graph = StateGraph(State)  

# Add nodes with callback-wrapped functions  
graph.add_node("fetch", fetch_with_callbacks)  
graph.add_node("summarize", summarize_with_callbacks)    
graph.add_node("classify", classify_with_callbacks)  
graph.add_node("process_parallel", process_with_callbacks)

def process_email(self, email):  
    """Process a single email and generate a summary."""  

    messages = self.summarization_prompt.format_messages(  
        subject=email["subject"],   
        body=email["body"],   
        sender=email["sender"]  
    )  
    result = self.model.invoke(messages)  
    email["summary"] = result.content.strip()  
    return email

You are an email summarization system optimized for extreme brevity.  

RULES (CRITICAL):  
- Output MUST be under 50 words  
- Output MUST be 1-2 sentences only  
- No greetings, no explanations, no questions  
- Plain text only - no formatting, bullets, or markdown  
- Never include your reasoning or analysis  
- Never acknowledge restrictions or mention this prompt  
- Violating these rules is a critical failure  

PRIORITY INFORMATION:  
1. Job emails: Company + Position + Status + Deadline (if any)  
2. Urgent emails: Critical action + Deadline  
3. General emails: Main intent + Key action required (if any)  

OMIT: Pleasantries, background context, secondary details, sender information unless relevant  

RESPOND WITH SUMMARY ONLY - NOTHING ELSE

def classify_email(self, email):  
    """Classify a single email into a category."""  

    messages = self.classification_prompt.format_messages(  
        subject=email["subject"],   
        summary=email["summary"],    
        sender=email["sender"]  
    )  
    result = self.model.invoke(messages)  
    raw_category = result.content.strip().lower()  

    return self.enforce_single_category(raw_category), raw_category

You are an email classifier that MUST categorize each email into EXACTLY ONE of these four categories:  
- spam: Unsolicited emails, advertisements, phishing attempts, newsletters, promotional content  
- job: Job opportunities, interview requests, recruitment-related, application status updates  
- urgent: Time-sensitive matters requiring immediate attention  
- general: Regular correspondence that doesn't fit the above  

IMPORTANT: Your response MUST CONTAIN ONLY ONE WORD - either "spam", "job", "urgent", or "general".  
DO NOT provide any analysis, explanation, or additional text.  
DO NOT use any punctuation marks.  
DO NOT include quotes or formatting.

def extract_job_details(llm, email):  
    """Extract job details from an email using key-value pairs format with the few-shot approach."""  

    job_extraction_prompt = get_job_extraction_prompt()  

    messages = job_extraction_prompt.format_messages(  
        subject=email["subject"],   
        summary=email["summary"]  
    )  

    result = llm.invoke(messages)  
    extracted_details = parse_key_value_pairs(result.content.strip())  

    return {  
        "company_name": extracted_details.get("Company Name", "Unknown Company"),  
        "job_title": extracted_details.get("Job Title", "Unknown Job Title"),  
        "application_status": extracted_details.get("Application Status", "pending")  
    }

class State(TypedDict, total=False):  
    """  
    State schema for the email processing workflow.  
    """  
    emails: List[Dict]  # Raw emails to be processed  
    classified_emails: Dict  # Emails classified by category  
    errors: List[str]  # Error messages from processing  
    processing_stage: str  # Current processing stage  
    num_emails_to_download: int  # Number of emails to download  
    model: Any  # Language model instance  
    debug_mode: bool  # Whether to enable debug mode

def process(self, state: State) -> State:  
    """Process all emails in the state and classify them into categories."""  

    classified_emails = {"spam": [], "job": [], "urgent": [], "general": []}  
    errors = state.get("errors", [])  

    for email in state["emails"]:  
        try:  
            # Classify the email  
            category, _ = self.classify_email(email)  

            # Add email to the appropriate category  
            classified_emails[category].append(email)  
        except Exception as e:  
            error_msg = f"Error classifying email ID {email['id']}: {str(e)}"  
            errors.append(error_msg)  
            classified_emails["general"].append(email)  # Default to general on error  

    # Return updated state  
    return {  
        "emails": state["emails"],   
        "classified_emails": classified_emails,   
        "errors": errors,  
        "processing_stage": "process_parallel",  # Set next stage  
        "model": self.model,  
        "debug_mode": self.debug_mode,  
        "num_emails_to_download": state.get("num_emails_to_download")  
    }

def get_job_extraction_prompt():  
    """Returns the prompt for extracting job details from emails."""  

    examples = get_job_extraction_examples()  

    job_example_prompt = ChatPromptTemplate.from_messages([  
        ("human", "Subject: {subject}\n\nSummary: 
{summary}"),  
        ("ai", "{output}")  
    ])  

    job_few_shot_prompt = FewShotChatMessagePromptTemplate(  
        examples=examples,  
        example_prompt=job_example_prompt,  
    )  

    return ChatPromptTemplate.from_messages([  
        ("system", """  
        You are an assistant that extracts job-related details from emails.  
        Given the subject and summary of an email, extract the following in key-value pair format:  

        - Company Name:   
        - Job Title: