Job Search Agent

# ExaAI for job search
from exa_py import Exa

# LangChain for agent orchestration
from langchain_nebius import ChatNebius
from langchain_core.messages import SystemMessage, HumanMessage

# Memori for contextual memory
from memori import Memori, create_memory_tool

# Memory system is initialized in the Streamlit app
memory_system = Memori(
    database_connect="sqlite:///memori.db",
    auto_ingest=True,
    conscious_ingest=True,
    verbose=False,
)
memory_system.enable()
memory_tool = create_memory_tool(memory_system)

from pydantic import BaseModel, Field
from typing import Optional

class JobSearchConfig(BaseModel):
    job_title: str = Field(..., min_length=1)
    location: Optional[str] = None
    work_style: Optional[str] = Field(None, pattern="^(Remote|Hybrid|Onsite|Any)$")
    num_jobs: int = Field(default=5, ge=1, le=20)

class JobListing(BaseModel):
    title: str
    company: str
    location: str
    work_style: str
    url: str
    description: str
    salary: Optional[str] = None

from exa_py import Exa
from datetime import datetime, timedelta
import os

def search_jobs_with_exa(config: JobSearchConfig) -> list:
    """Search for jobs using ExaAI from multiple sources."""
    # Initialize ExaAI client
    exa_client = Exa(api_key=os.getenv("EXA_API_KEY"))
    
    # Build search query
    search_query = build_search_query(config)
    
    # Calculate date filter (last 7 days)
    seven_days_ago = (datetime.now() - timedelta(days=7)).isoformat()
    
    # Job board domains organized in groups
    job_domain_groups = [
        ["indeed.com", "glassdoor.com"],
        ["monster.com", "ziprecruiter.com"],
        ["careerbuilder.com", "jobs.com"],
        ["linkedin.com"],
    ]
    
    all_results = []
    seen_urls = set()
    
    # Search across different domain groups
    for domain_group in job_domain_groups:
        results = exa_client.search_and_contents(
            query=search_query,
            num_results=max(5, config.num_jobs // 2),
            include_domains=domain_group,
            text=True,
            type="auto",
            start_published_date=seven_days_ago,
        )
        
        # Add unique results
        for result in results.results:
            if result.url not in seen_urls:
                seen_urls.add(result.url)
                all_results.append(result)
        
        # Stop if we have enough results
        if len(all_results) >= config.num_jobs * 2:
            break
    
    # Process results into job listings
    job_listings = []
    for result in all_results:
        job_data = {
            "title": result.title or "Job Opening",
            "company": extract_company(result.text or ""),
            "location": config.location or "Location not specified",
            "work_style": config.work_style or "Any",
            "url": result.url,
            "description": result.text if result.text else "No description available",
            "salary": extract_salary(result.text or ""),
        }
        job_listings.append(job_data)
    
    return job_listings[:config.num_jobs]

def build_search_query(config: JobSearchConfig) -> str:
    """Build a search query from config."""
    query_parts = [config.job_title]
    
    if config.location:
        query_parts.append(config.location)
    
    if config.work_style and config.work_style != "Any":
        query_parts.append(config.work_style)
    
    query_parts.extend(["job opening", "hiring"])
    return " ".join(query_parts)

from langchain_nebius import ChatNebius
from langchain_core.messages import SystemMessage, HumanMessage

JOB_EXTRACTION_SYSTEM_PROMPT = """
You are an expert job search assistant. Extract job information from web search results.

When processing job listings:
1. Extract the job title accurately
2. Identify the company name from the content
3. Determine the location if mentioned
4. Identify work style (Remote, Hybrid, Onsite)
5. Extract salary information if available
6. Create a concise description (max 500 characters)

Be precise and only extract clearly stated information.
"""

def extract_company(text: str, llm=None) -> str:
    """Extract company name using LLM if available, otherwise regex."""
    if llm:
        try:
            prompt = f"""
            Extract the company name from this job description. 
            Return only the company name, nothing else.
            
            Job description:
            {text[:1000]}
            
            Company name:
            """
            response = llm.invoke([
                SystemMessage(content=JOB_EXTRACTION_SYSTEM_PROMPT),
                HumanMessage(content=prompt),
            ])
            company = response.content.strip()
            if company and len(company) < 100 and company.lower() not in ["none", "not specified"]:
                return company
        except Exception:
            pass  # Fall back to regex
    
    # Regex fallback
    import re
    patterns = [
        r"at\s+([A-Z][a-zA-Z\s&]+?)(?:\s|,|\.|$)",
        r"([A-Z][a-zA-Z\s&]+?)\s+is\s+hiring",
    ]
    
    for pattern in patterns:
        match = re.search(pattern, text[:300], re.IGNORECASE)
        if match:
            company = match.group(1).strip()
            if len(company) > 2 and len(company) < 50:
                return company
    
    return "Company not specified"

def extract_salary(text: str) -> Optional[str]:
    """Extract salary information from job description."""
    import re
    patterns = [
        r"\$(\d{1,3}(?:,\d{3})*(?:k|K)?)\s*-\s*\$(\d{1,3}(?:,\d{3})*(?:k|K)?)",
        r"(\d{1,3}(?:,\d{3})*)\s*-\s*(\d{1,3}(?:,\d{3})*)\s*(?:USD|per year|annually)",
    ]
    
    for pattern in patterns:
        match = re.search(pattern, text[:500])
        if match:
            return match.group(0).strip()
    
    return None

import pdfplumber
from typing import Dict, Any

def parse_resume(resume_file) -> Dict[str, Any]:
    """Extract text from resume PDF."""
    with pdfplumber.open(resume_file) as pdf:
        text = ""
        for page in pdf.pages:
            text += page.extract_text() + "\n"
    
    return {
        "raw_text": text,
        "file_name": resume_file.name,
        "page_count": len(pdf.pages)
    }

def store_resume_in_memory(memory_system, resume_data: Dict[str, Any]):
    """Store resume details in Memori."""
    memory_system.record_conversation(
        user_input=f"I uploaded my resume: {resume_data['file_name']}",
        ai_output=f"I've analyzed your resume. It contains {resume_data['page_count']} pages. I'll use this information to help you find relevant jobs and suggest resume improvements.",
        metadata={
            "type": "resume_upload",
            "file_name": resume_data["file_name"],
            "resume_text": resume_data["raw_text"][:1000],  # Store excerpt
        }
    )

def store_job_listing_in_memory(memory_system, job: JobListing, search_config: JobSearchConfig):
    """Store individual job listing in Memori for matching."""
    memory_system.record_conversation(
        user_input=f"Found job: {job.title} at {job.company}",
        ai_output=f"Job listing stored: {job.title} at {job.company} in {job.location}. {job.description[:200]}...",
        metadata={
            "type": "job_listing",
            "job_title": job.title,
            "company": job.company,
            "location": job.location,
            "work_style": job.work_style,
            "url": job.url,
            "search_query": search_config.job_title,
        }
    )

def ask_memory_about_jobs(memory_tool, question: str) -> str:
    """Query Memori about job searches and resume matching."""
    try:
        result = memory_tool.execute(query=question)
        return result if result else "No relevant information found in memory."
    except Exception as e:
        return f"Error querying memory: {str(e)}"

# Example queries:
# "Which job am I best suited for with my resume?"
# "What can I add to my resume to make it fit for Software Engineer at Google?"
# "What jobs did I search for?"
# "What companies did I find for software engineer positions?"

import streamlit as st

st.title("🔍 Job Search Agent with Memory")

# Initialize Memori
if "memory_system" not in st.session_state:
    st.session_state.memory_system = Memori(
        database_connect="sqlite:///memori.db",
        auto_ingest=True,
        conscious_ingest=True,
    )
    st.session_state.memory_system.enable()
    st.session_state.memory_tool = create_memory_tool(st.session_state.memory_system)

# Tab layout
tab1, tab2, tab3 = st.tabs(["🔍 Job Search", "📝 Resume", "🧠 Memory"])

with tab1:
    st.header("Job Search")
    
    job_title = st.text_input("Job Title", placeholder="Software Engineer")
    location = st.text_input("Location (optional)", placeholder="San Francisco")
    work_style = st.selectbox("Work Style", ["Any", "Remote", "Hybrid", "Onsite"])
    num_jobs = st.slider("Number of Jobs", 1, 20, 5)
    
    if st.button("Search Jobs"):
        config = JobSearchConfig(
            job_title=job_title,
            location=location,
            work_style=work_style,
            num_jobs=num_jobs
        )
        
        with st.spinner("Searching for jobs..."):
            jobs = search_jobs_with_exa(config)
            
            # Display jobs
            for job in jobs:
                with st.expander(f"{job.title} - {job.company}"):
                    st.write(f"**Location:** {job.location}")
                    st.write(f"**Work Style:** {job.work_style}")
                    if job.salary:
                        st.write(f"**Salary:** {job.salary}")
                    st.write(f"**Description:** {job.description[:500]}...")
                    st.link_button("Apply Now", job.url)
                
                # Store in memory
                store_job_listing_in_memory(
                    st.session_state.memory_system, 
                    job, 
                    config
                )

with tab2:
    st.header("Upload Resume")
    
    resume_file = st.file_uploader("Upload your resume (PDF)", type=["pdf"])
    
    if resume_file:
        with st.spinner("Processing resume..."):
            resume_data = parse_resume(resume_file)
            store_resume_in_memory(st.session_state.memory_system, resume_data)
            st.success("✅ Resume uploaded and stored in memory!")

with tab3:
    st.header("Ask About Your Job Search")
    
    question = st.chat_input("Ask me about your searches or resume matching...")
    
    if question:
        with st.spinner("Searching memory..."):
            answer = ask_memory_about_jobs(st.session_state.memory_tool, question)
            st.markdown(answer)

# Resume matching
"Which job am I best suited for with my resume?"

# Resume improvement
"What can I add to my resume to make it fit for Software Engineer at Google?"

# Search history
"What jobs did I search for?"

# Company tracking
"What companies did I find for software engineer positions?"

# Skill gaps
"What skills am I missing for the Google job?"

git clone https://github.com/Arindam200/awesome-ai-apps.git
cd memory_agents/job_search_agent
uv sync

EXA_API_KEY=your_exa_api_key_here
NEBIUS_API_KEY=your_nebius_api_key_here

uv run streamlit run app.py