Job Search Agent Langraph

1# Job Search Agent - Apify Actor Project Outline
2
3Based on the existing codebase and the Apify templates, I'll outline a project structure for converting the current job search functionality into an Apify actor using LangGraph. This approach maintains the core functionality while leveraging Apify's infrastructure.
4
5## Project Structure
6
7```
8job-search-agent/
9├── .actor/                      # Apify actor configuration
10│   ├── actor.json               # Actor metadata
11│   └── input_schema.json        # Input schema definition
12├── src/
13│   ├── agents/                  # Agent definitions
14│   │   ├── __init__.py
15│   │   ├── manager.py           # Orchestration agent
16│   │   ├── resume_parser.py     # Resume parsing agent
17│   │   ├── job_scraper.py       # Job search agent
18│   │   ├── relevance_scorer.py  # Scoring and matching agent
19│   │   └── output_formatter.py  # Results formatting agent
20│   ├── tools/                   # Tool definitions
21│   │   ├── __init__.py
22│   │   ├── apify_tools.py       # LinkedIn/Indeed search tools
23│   │   └── utility_tools.py     # Helper tools
24│   ├── models/                  # Data models
25│   │   ├── __init__.py
26│   │   ├── resume.py            # Resume data structure
27│   │   └── job.py               # Job listing data structure
28│   ├── utils/                   # Utility functions
29│   │   ├── __init__.py
30│   │   ├── text_processing.py   # Text cleaning utilities
31│   │   └── scoring.py           # Scoring algorithms
32│   ├── config.py                # Configuration settings
33│   └── main.py                  # Entry point
34├── .gitignore
35├── README.md
36└── requirements.txt             # Dependencies
37```
38
39## Workflow Logic
40
41The job search workflow will follow these steps:
42
431. **Input Processing**
44   - Parse user input (resume text, location preferences, work mode)
45   - Validate and prepare for processing
46
472. **Resume Parsing**
48   - Extract essential information from resume
49   - Categorize skills, experience, education
50   - Identify key search terms
51
523. **Job Search**
53   - Use Apify actors for LinkedIn and Indeed searches
54   - Apply search parameters from resume
55   - Collect raw job listings
56
574. **Relevance Scoring**
58   - Score each job based on match criteria
59   - Calculate component scores (skills, experience, location)
60   - Generate overall match percentage
61
625. **Results Formatting**
63   - Structure results with match details
64   - Sort by relevance score
65   - Add match explanations
66
676. **Output Generation**
68   - Save to Apify dataset
69   - Generate summary statistics
70   - Create downloadable results
71
72## LangGraph Implementation
73
74Using the [python-langgraph template](https://github.com/apify/actor-templates/tree/master/templates/python-langgraph), we'll implement a graph-based workflow:
75
76```python
77# Simplified graph structure
78graph = StateGraph(nodes=[
79    Node("resume_parser", resume_parser_agent),
80    Node("job_scraper", job_scraper_agent),
81    Node("relevance_scorer", relevance_scorer_agent),
82    Node("output_formatter", output_formatter_agent)
83])
84
85# Define transitions
86graph.add_edge("resume_parser", "job_scraper")
87graph.add_edge("job_scraper", "relevance_scorer")
88graph.add_edge("relevance_scorer", "output_formatter")
89```
90
91## Scoring Logic
92
93The relevance scoring will use a weighted approach:
94
95```
96Overall Score = (Skills Match × 0.40) + 
97                (Experience Match × 0.25) + 
98                (Location Match × 0.20) + 
99                (Company/Role Fit × 0.15)
100```
101
102Each component will be scored from 0-100:
103
1041. **Skills Match (40%)**
105   - Required skills present in resume
106   - Skill level/proficiency match
107   - Technology stack alignment
108
1092. **Experience Match (25%)**
110   - Years of experience match
111   - Role similarity
112   - Industry relevance
113
1143. **Location Match (20%)**
115   - Exact location match
116   - Remote/hybrid preference match
117   - Commute distance (if applicable)
118
1194. **Company/Role Fit (15%)**
120   - Company size preference
121   - Industry alignment
122   - Career growth potential
123
124## Input Schema
125
126```json
127{
128  "title": "Job Search Input",
129  "type": "object",
130  "schemaVersion": 1,
131  "properties": {
132    "resumeText": {
133      "title": "Resume Text",
134      "type": "string",
135      "description": "Full text of the resume",
136      "editor": "textarea"
137    },
138    "locationPreference": {
139      "title": "Location Preference",
140      "type": "string",
141      "description": "Preferred job location",
142      "editor": "textfield"
143    },
144    "workModePreference": {
145      "title": "Work Mode",
146      "type": "string",
147      "description": "Preferred work mode (Remote, Hybrid, On-site)",
148      "editor": "select",
149      "enum": ["Remote", "Hybrid", "On-site", "Any"]
150    },
151    "searchRadius": {
152      "title": "Search Radius (miles)",
153      "type": "integer",
154      "description": "Maximum distance from preferred location",
155      "default": 25
156    },
157    "minSalary": {
158      "title": "Minimum Salary",
159      "type": "integer",
160      "description": "Minimum acceptable salary",
161      "default": 0
162    }
163  },
164  "required": ["resumeText"]
165}
166```
167
168## Output Format
169
170```json
171{
172  "query": {
173    "resumeSummary": "...",
174    "searchParameters": {
175      "location": "...",
176      "workMode": "...",
177      "searchRadius": 25
178    }
179  },
180  "results": [
181    {
182      "position": "Software Engineer",
183      "company": "Example Corp",
184      "location": "New York, NY",
185      "workMode": "Hybrid",
186      "salary": "$120,000 - $150,000",
187      "matchScore": 87,
188      "matchDetails": {
189        "skillsMatch": 90,
190        "experienceMatch": 85,
191        "locationMatch": 75,
192        "companyFitMatch": 95
193      },
194      "matchExplanation": "Strong match on required skills (Python, React). Experience level aligns well with 5+ years requirement. Location is within commuting distance. Company culture emphasizes work-life balance.",
195      "keyRequirements": ["Python", "React", "5+ years experience"],
196      "applicationUrl": "https://example.com/jobs/123"
197    }
198  ],
199  "statistics": {
200    "totalJobsFound": 120,
201    "averageMatchScore": 72,
202    "topSkillsRequested": ["Python", "JavaScript", "AWS"],
203    "salaryRange": {
204      "min": 80000,
205      "max": 180000,
206      "average": 125000
207    }
208  }
209}
210```
211
212## Implementation Approach
213
2141. **Start with the python-langgraph template**
215   - This provides the LangGraph structure needed for agent workflow
216   - Already includes Apify SDK integration
217
2182. **Implement agents as LangGraph nodes**
219   - Each agent will be a node in the graph
220   - Define clear input/output contracts
221
2223. **Use Apify actors as tools**
223   - LinkedIn and Indeed scrapers as external tools
224   - Integrate with LangChain tool calling format
225
2264. **Implement scoring logic in Python**
227   - Create utility functions for match calculations
228   - Use weighted scoring approach
229
2305. **Store results in Apify dataset**
231   - Structured job matches with scores
232   - Include match explanations and statistics
233
234## Deployment and Execution
235
2361. **Local Development**
237   - Use Apify CLI for local testing
238   - Run with `apify run` command
239
2402. **Apify Platform Deployment**
241   - Push to Apify with `apify push`
242   - Configure environment variables
243
2443. **Execution**
245   - Run via Apify UI or API
246   - Monitor execution in Apify console
247   - Download results as JSON
248
249This project structure maintains the core functionality of the existing job search implementation while leveraging Apify's infrastructure and the LangGraph framework for agent orchestration.
250
251[Source: Python LangGraph template](https://github.com/apify/actor-templates/tree/master/templates/python-langgraph)

1# Feel free to add your Python dependencies below. For formatting guidelines, see:
2# https://pip.pypa.io/en/latest/reference/requirements-file-format/
3
4# Apify SDK
5apify==1.1.3
6apify-client==1.4.0
7
8# LangGraph and related packages
9langgraph==0.0.20
10
11# OpenAI
12openai>=1.10.0
13
14# Utilities
15python-dotenv==1.0.0
16pydantic==1.10.8
17tenacity==8.2.3
18
19# Web scraping
20beautifulsoup4==4.12.2
21httpx==0.24.1

1"""
2AI Job Search Agent - Apify Actor
3
4This actor analyzes a resume and finds matching job opportunities from LinkedIn and Indeed.
5"""
6
7import os
8import json
9import logging
10from typing import Dict, Any
11from datetime import datetime
12import asyncio
13import sys
14
15# Add the src directory to the Python path
16sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
17
18from apify import Actor
19from openai import OpenAI
20from dotenv import load_dotenv
21from apify_client import ApifyClient
22
23# Import settings
24from src.config.settings import (
25    OPENAI_MODEL,
26    ACTOR_IDS,
27    FINAL_SCORING_WEIGHTS
28)
29
30# Import agents
31from src.agents.manager import ManagerAgent
32from src.agents.resume_parser import ResumeParserAgent
33from src.agents.relevance_scorer import RelevanceScorerAgent
34from src.agents.notification import NotificationAgent
35from src.models.schema import JobSearchState
36
37# Configure logging
38logging.basicConfig(
39    level=logging.INFO,
40    format='%(asctime)s [%(levelname)s] [%(name)s] %(message)s',
41    handlers=[logging.StreamHandler()]
42)
43logger = logging.getLogger(__name__)
44
45# Load environment variables
46load_dotenv()
47
48# Event names and prices
49EVENTS = {
50    'resume_parse': 0.10,      # Resume parsing
51    'job_score': 0.02,         # Per job scoring
52    'results_summary': 0.10    # Final summary
53}
54
55class JobSearchActor:
56    def __init__(self, apify_client: ApifyClient):
57        self.apify_client = apify_client
58        self.manager = ManagerAgent(apify_client)
59        self.resume_parser = ResumeParserAgent(apify_client)
60        self.relevance_scorer = RelevanceScorerAgent(apify_client)
61        self.notification = NotificationAgent(apify_client)
62        
63    async def process_step(self, state: Dict[str, Any]) -> Dict[str, Any]:
64        """Process a single step in the workflow based on current phase."""
65        current_phase = state.get("current_phase", "initialize")
66        
67        try:
68            if current_phase == "parse_resume":
69                logger.info("Parsing resume...")
70                state = await self.resume_parser.process(state)
71                # Log charge for resume parsing
72                await self._log_charge('resume_parse')
73                
74            elif current_phase == "search_jobs":
75                logger.info("Searching for jobs...")
76                # Use Apify's LinkedIn Jobs Scraper
77                run_input = {
78                    "keywords": state["resume_data"]["desired_role"],
79                    "location": state["resume_data"]["location_preference"],
80                    "maxItems": 50,
81                    "timeoutSecs": 120,
82                    "memoryMbytes": 1024
83                }
84                
85                run = await self.apify_client.actor(ACTOR_IDS["linkedin_scraper"]).call(run_input=run_input)
86                dataset_items = await self.apify_client.dataset(run["defaultDatasetId"]).list_items().items()
87                
88                if not dataset_items:
89                    logger.warning("No jobs found in search results")
90                    state["error_log"] = state.get("error_log", []) + ["No jobs found during search"]
91                else:
92                    logger.info(f"Found {len(dataset_items)} jobs")
93                    state["job_listings"] = dataset_items
94                
95                state["scraping_complete"] = True
96                
97            elif current_phase == "score_jobs":
98                logger.info("Scoring jobs...")
99                state = await self.relevance_scorer.process(state)
100                # Log charge for job scoring
101                job_count = len(state.get("scored_listings", []))
102                for _ in range(job_count):
103                    await self._log_charge('job_score')
104                
105            elif current_phase == "send_notifications":
106                logger.info("Generating results summary...")
107                state = await self.notification.process(state)
108                # Log charge for results summary
109                await self._log_charge('results_summary')
110                
111        except Exception as e:
112            logger.error(f"Error in {current_phase}: {str(e)}")
113            state["error_log"] = state.get("error_log", []) + [f"Error in {current_phase}: {str(e)}"]
114            
115        return state
116
117    async def _log_charge(self, event_type: str):
118        """Log a charge event."""
119        if event_type in EVENTS:
120            await self.apify_client.key_value_store().set_record(
121                f"CHARGE_{datetime.now().isoformat()}",
122                {
123                    "event": event_type,
124                    "amount": EVENTS[event_type],
125                    "timestamp": datetime.now().isoformat()
126                }
127            )
128
129    async def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
130        """Run the job search workflow."""
131        # Initialize state
132        state = JobSearchState(
133            resume_text=input_data.get("resumeText", ""),
134            resume_parsed=False,
135            scraping_complete=False,
136            current_phase="initialize",
137            next_step="parse_resume"
138        ).dict()
139        
140        # Process workflow until completion
141        while state.get("status") not in ["completed", "failed"]:
142            # Let manager determine next step
143            state = await self.manager.process(state)
144            
145            if state.get("status") in ["completed", "failed"]:
146                break
147                
148            # Process the current step
149            state = await self.process_step(state)
150            
151            # Add checkpoint to dataset
152            await self.apify_client.key_value_store().set_record(
153                "STATE_CHECKPOINT",
154                state
155            )
156        
157        return state
158
159async def main():
160    """Main entry point for the actor."""
161    async with Actor:
162        # Initialize Apify client
163        apify_client = ApifyClient()
164        
165        try:
166            # Get input
167            actor_input = await Actor.get_input() or {}
168            
169            # Initialize and run job search
170            job_search = JobSearchActor(apify_client)
171            result = await job_search.run(actor_input)
172            
173            # Save output
174            await Actor.push_data(result)
175            
176        except Exception as e:
177            logger.error(f"Actor failed: {str(e)}")
178            raise
179
180if __name__ == "__main__":
181    asyncio.run(main())

1import asyncio
2
3from .main import main
4
5# Execute the Actor entry point.
6asyncio.run(main())

1"""Configuration settings for the job search actor."""
2
3import os
4from typing import Dict
5from dotenv import load_dotenv
6
7# Load environment variables
8load_dotenv()
9
10# API Keys - these should be set as environment variables
11APIFY_API_TOKEN = os.getenv("APIFY_API_TOKEN")
12OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
13
14# OpenAI Model Configuration
15OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
16
17# Apify Actor IDs
18ACTOR_IDS: Dict[str, str] = {
19    "linkedin_scraper": "krandiash/linkedin-jobs-scraper",
20    "indeed_scraper": "epctex/indeed-scraper"
21}
22
23# Job Search Settings
24JOB_SEARCH_CONFIG = {
25    "initial_results_per_source": 5,    # Reduced for testing
26    "detail_fetch_threshold": 0.6,      # Minimum initial score to fetch details
27    "max_details_to_fetch": 5,          # Reduced for testing
28    "max_days_old": 30                  # Maximum age of job postings
29}
30
31# Scoring weights for initial filtering
32INITIAL_SCORING_WEIGHTS = {
33    "title_match": 0.4,
34    "location_match": 0.3,
35    "company_relevance": 0.2,
36    "posting_date": 0.1
37}
38
39# Detailed scoring weights
40FINAL_SCORING_WEIGHTS = {
41    "position_match": 0.25,
42    "skills_experience": 0.40,
43    "location": 0.20,
44    "company": 0.15,
45    "salary": 0.0,
46    "benefits": 0.0
47}
48
49# System settings
50RETRY_ATTEMPTS = 3
51RETRY_DELAY = 5  # seconds

1"""Configuration module for job search actor."""
2
3# Config package initialization
4from .settings import (
5    APIFY_API_TOKEN,
6    OPENAI_API_KEY,
7    OPENAI_MODEL,
8    ACTOR_IDS,
9    JOB_SEARCH_CONFIG,
10    INITIAL_SCORING_WEIGHTS,
11    FINAL_SCORING_WEIGHTS,
12    RETRY_ATTEMPTS,
13    RETRY_DELAY
14)
15
16__all__ = [
17    'APIFY_API_TOKEN',
18    'OPENAI_API_KEY',
19    'OPENAI_MODEL',
20    'ACTOR_IDS',
21    'JOB_SEARCH_CONFIG',
22    'INITIAL_SCORING_WEIGHTS',
23    'FINAL_SCORING_WEIGHTS',
24    'RETRY_ATTEMPTS',
25    'RETRY_DELAY'
26]

1from typing import Any, Dict, Optional
2from apify_client import ApifyClient
3from abc import ABC, abstractmethod
4import logging
5
6# Import the retry utility using absolute import
7from src.utils.retry import with_retry
8
9logger = logging.getLogger(__name__)
10
11class BaseAgent(ABC):
12    def __init__(self, apify_client: ApifyClient, actor_id: Optional[str] = None):
13        self.apify_client = apify_client
14        self.actor_id = actor_id
15        self.system_prompt = self._get_system_prompt()
16
17    @abstractmethod
18    def _get_system_prompt(self) -> str:
19        """Return the system prompt for this agent."""
20        raise NotImplementedError("Subclasses must implement _get_system_prompt")
21
22    @abstractmethod
23    async def process(self, state: Dict[str, Any]) -> Dict[str, Any]:
24        """Process the current state and return updated state."""
25        raise NotImplementedError("Subclasses must implement process")
26
27    @with_retry()
28    def run_actor(self, actor_id: Optional[str] = None, input_data: Dict[str, Any] = None) -> Dict[str, Any]:
29        """Run the associated Apify actor if one exists."""
30        actor_id = actor_id or self.actor_id
31        if not actor_id:
32            raise ValueError("No actor_id specified for this agent")
33        
34        try:
35            run = self.apify_client.actor(actor_id).call(run_input=input_data)
36            dataset_items = self.apify_client.dataset(run["defaultDatasetId"]).list_items().items
37            return dataset_items
38        except Exception as e:
39            # Log the error and re-raise
40            error_msg = f"Error running actor {actor_id}: {str(e)}"
41            logger.error(error_msg)
42            raise
43
44    async def _safe_api_call(self, func, *args, **kwargs):
45        """Safely make an API call with error handling."""
46        try:
47            return await func(*args, **kwargs)
48        except Exception as e:
49            logger.error(f"API call failed: {str(e)}")
50            raise

1from typing import Dict, Any, List
2from langchain_openai import ChatOpenAI
3from langchain_core.messages import SystemMessage, HumanMessage
4from .base import BaseAgent
5from ..config.settings import OPENAI_API_KEY, OPENAI_MODEL
6
7class FeedbackRefinerAgent(BaseAgent):
8    def __init__(self, apify_client):
9        super().__init__(apify_client)
10        self.llm = ChatOpenAI(
11            api_key=OPENAI_API_KEY,
12            model=OPENAI_MODEL,
13            temperature=0.2
14        )
15
16    def _get_system_prompt(self) -> str:
17        return """
18        Feedback Refiner Agent responsible for:
19        1. Processing user feedback on job matches
20        2. Refining search parameters
21        3. Adjusting scoring weights based on preferences
22        4. Improving match quality over time
23        """
24
25    def _analyze_feedback(self, feedback: Dict[str, Any]) -> Dict[str, Any]:
26        """Analyze user feedback to determine parameter adjustments."""
27        adjustments = {
28            "search_params": {},
29            "scoring_weights": {},
30            "filters": {}
31        }
32
33        # Process location preferences
34        if feedback.get("location_feedback"):
35            location_pref = feedback["location_feedback"]
36            adjustments["search_params"]["location"] = location_pref
37            if "remote" in location_pref.lower():
38                adjustments["filters"]["remote_only"] = True
39
40        # Process role preferences
41        if feedback.get("role_feedback"):
42            role_pref = feedback["role_feedback"]
43            adjustments["search_params"]["keywords"] = role_pref
44            
45        # Process experience level preferences
46        if feedback.get("experience_feedback"):
47            exp_pref = feedback["experience_feedback"]
48            adjustments["search_params"]["experience_level"] = exp_pref
49
50        # Process salary preferences
51        if feedback.get("salary_feedback"):
52            salary_pref = feedback["salary_feedback"]
53            adjustments["filters"]["salary_min"] = salary_pref.get("min")
54            adjustments["filters"]["salary_max"] = salary_pref.get("max")
55
56        # Process company preferences
57        if feedback.get("company_preferences"):
58            company_prefs = feedback["company_preferences"]
59            adjustments["filters"]["preferred_companies"] = company_prefs
60
61        return adjustments
62
63    def _refine_search_parameters(self, current_params: Dict[str, Any], adjustments: Dict[str, Any]) -> Dict[str, Any]:
64        """Refine search parameters based on feedback analysis."""
65        refined_params = current_params.copy()
66        
67        # Update search parameters
68        if "search_params" in adjustments:
69            refined_params.update(adjustments["search_params"])
70        
71        # Update filters
72        if "filters" in adjustments:
73            if "filters" not in refined_params:
74                refined_params["filters"] = {}
75            refined_params["filters"].update(adjustments["filters"])
76        
77        # Update scoring weights if provided
78        if "scoring_weights" in adjustments:
79            if "scoring_weights" not in refined_params:
80                refined_params["scoring_weights"] = {}
81            refined_params["scoring_weights"].update(adjustments["scoring_weights"])
82        
83        return refined_params
84
85    async def process(self, state: Dict[str, Any]) -> Dict[str, Any]:
86        """Process user feedback and refine search parameters."""
87        feedback = state.get("feedback", {})
88        current_params = state.get("search_parameters", {})
89        
90        if not feedback:
91            return state
92        
93        # Analyze feedback
94        adjustments = self._analyze_feedback(feedback)
95        
96        # Refine parameters
97        refined_params = self._refine_search_parameters(current_params, adjustments)
98        
99        # Update state
100        state["search_parameters"] = refined_params
101        state["feedback_processed"] = True
102        state["parameter_adjustments"] = adjustments
103        
104        return state

1from typing import Any, Dict, List
2from datetime import datetime, timedelta
3import asyncio
4from .base import BaseAgent
5from ..config.settings import (
6    ACTOR_IDS,
7    JOB_SEARCH_CONFIG,
8    INITIAL_SCORING_WEIGHTS
9)
10from ..models.schema import ResumeData
11
12class JobScraperAgent(BaseAgent):
13    def __init__(self, apify_client):
14        super().__init__(apify_client)
15        self.linkedin_search_actor = ACTOR_IDS["linkedin_jobs_search"]
16        self.linkedin_detail_actor = ACTOR_IDS["linkedin_job_detail"]
17        self.indeed_scraper = ACTOR_IDS["indeed_scraper"]
18
19    def _get_system_prompt(self) -> str:
20        return """
21        Job Scraping Agent responsible for:
22        1. Initial job search from multiple sources
23        2. Preliminary filtering and scoring
24        3. Detailed information gathering for promising matches
25        """
26
27    def _search_linkedin_jobs(self, search_params: Dict[str, Any]) -> List[Dict]:
28        """Perform initial LinkedIn job search."""
29        search_input = {
30            "keywords": search_params["keywords"],
31            "location": search_params["location"],
32            "limit": JOB_SEARCH_CONFIG["initial_results_per_source"]
33        }
34        
35        return self.run_actor(
36            actor_id=self.linkedin_search_actor,
37            input_data=search_input
38        )
39
40    def _search_indeed_jobs(self, search_params: Dict[str, Any]) -> List[Dict]:
41        """Perform initial Indeed job search."""
42        search_input = {
43            "keyword": search_params["keywords"],
44            "location": search_params["location"],
45            "maxResults": JOB_SEARCH_CONFIG["initial_results_per_source"]
46        }
47        
48        return self.run_actor(
49            actor_id=self.indeed_scraper,
50            input_data=search_input
51        )
52
53    def _calculate_initial_score(self, job: Dict, resume_data: ResumeData) -> float:
54        """Calculate preliminary score for a job listing."""
55        scores = {
56            "title_match": self._score_title_match(job["title"], resume_data.desired_role),
57            "location_match": self._score_location_match(job["location"], resume_data.location_preference),
58            "company_relevance": self._score_company_relevance(job["company"], resume_data.industry_experience),
59            "posting_date": self._score_posting_date(job.get("posted_date", datetime.now()))
60        }
61        
62        return sum(
63            scores[key] * INITIAL_SCORING_WEIGHTS[key]
64            for key in INITIAL_SCORING_WEIGHTS
65        )
66
67    def _fetch_job_details(self, job_urls: List[str]) -> List[Dict]:
68        """Fetch detailed information for promising job listings."""
69        detailed_jobs = []
70        for url in job_urls:
71            if "linkedin.com" in url:
72                job_details = self.run_actor(
73                    actor_id=self.linkedin_detail_actor,
74                    input_data={"url": url}
75                )
76                detailed_jobs.extend(job_details)
77            # Add similar handling for Indeed URLs if needed
78        
79        return detailed_jobs
80
81    async def process(self, state: Dict[str, Any]) -> Dict[str, Any]:
82        """Process the job search workflow."""
83        resume_data = state["resume_data"]
84        work_mode_preference = state["feedback"].get("work_mode_preference", "Any")
85        
86        # Prepare search parameters
87        search_params = {
88            "keywords": resume_data.desired_role,
89            "location": resume_data.location_preference
90        }
91        
92        # Add work mode to keywords if specified
93        if work_mode_preference and work_mode_preference != "Any":
94            search_params["keywords"] += f" {work_mode_preference}"
95        
96        # Initial job search from multiple sources
97        linkedin_jobs = self._search_linkedin_jobs(search_params)
98        indeed_jobs = self._search_indeed_jobs(search_params)
99        
100        # Combine and score initial results
101        all_jobs = []
102        for job in linkedin_jobs + indeed_jobs:
103            # Normalize job data structure
104            normalized_job = self._normalize_job_data(job)
105            
106            # Calculate initial score
107            score = self._calculate_initial_score(normalized_job, resume_data)
108            normalized_job["initial_score"] = score
109            all_jobs.append(normalized_job)
110        
111        # Sort by score and filter
112        promising_jobs = sorted(
113            [j for j in all_jobs if j["initial_score"] >= JOB_SEARCH_CONFIG["detail_fetch_threshold"]],
114            key=lambda x: x["initial_score"],
115            reverse=True
116        )[:JOB_SEARCH_CONFIG["max_details_to_fetch"]]
117        
118        # Fetch detailed information for promising jobs
119        job_urls = [job["url"] for job in promising_jobs if "url" in job]
120        detailed_jobs = self._fetch_job_details(job_urls)
121        
122        # If we couldn't get detailed jobs, use the promising jobs
123        if not detailed_jobs:
124            detailed_jobs = promising_jobs
125        
126        # Update state
127        state["jobs_scraped"] = True
128        state["job_listings"] = detailed_jobs
129        
130        return state
131
132    def _normalize_job_data(self, job: Dict[str, Any]) -> Dict[str, Any]:
133        """Normalize job data from different sources to a common format."""
134        normalized = {
135            "job_id": job.get("id", job.get("jobId", str(hash(str(job))))),
136            "title": job.get("title", job.get("name", "Unknown Position")),
137            "company": job.get("company", job.get("companyName", "Unknown Company")),
138            "location": job.get("location", job.get("place", "Unknown Location")),
139            "url": job.get("url", job.get("link", job.get("applicationLink", ""))),
140            "source": "LinkedIn" if "linkedin" in job.get("url", "") else "Indeed"
141        }
142        
143        # Handle posted date
144        posted_date = job.get("postedDate", job.get("date", job.get("listedAt", None)))
145        if isinstance(posted_date, str):
146            try:
147                normalized["posted_date"] = datetime.fromisoformat(posted_date.replace('Z', '+00:00'))
148            except (ValueError, TypeError):
149                normalized["posted_date"] = datetime.now() - timedelta(days=7)  # Default to 1 week ago
150        else:
151            normalized["posted_date"] = datetime.now() - timedelta(days=7)
152        
153        return normalized
154
155    def _score_title_match(self, job_title: str, desired_role: str) -> float:
156        """Score how well the job title matches the desired role."""
157        # Implement fuzzy matching or keyword matching logic
158        job_title = job_title.lower()
159        desired_role = desired_role.lower()
160        
161        if job_title == desired_role:
162            return 1.0
163        elif desired_role in job_title or job_title in desired_role:
164            return 0.8
165        # Add more sophisticated matching logic here
166        return 0.4
167
168    def _score_location_match(self, job_location: str, preferred_location: str) -> float:
169        """Score how well the job location matches preferences."""
170        # Implement location matching logic
171        if not job_location or not preferred_location:
172            return 0.5
173            
174        job_location = job_location.lower()
175        preferred_location = preferred_location.lower()
176        
177        if "remote" in job_location:
178            return 1.0
179        elif preferred_location in job_location:
180            return 1.0
181        # Add more sophisticated location matching logic here
182        return 0.5
183
184    def _score_company_relevance(self, company: str, industry_experience: List[str]) -> float:
185        """Score company relevance based on industry experience."""
186        # Implement company/industry matching logic
187        # This could be enhanced with company industry data
188        return 0.7  # Default score, improve with better matching logic
189
190    def _score_posting_date(self, posted_date: datetime) -> float:
191        """Score job based on how recently it was posted."""
192        days_old = (datetime.now() - posted_date).days
193        if days_old <= 7:
194            return 1.0
195        elif days_old <= 14:
196            return 0.8
197        elif days_old <= 21:
198            return 0.6
199        else:
200            return 0.4