| """ |
| PIOE LLM Client Abstraction Layer |
| |
| Supports Gemini (default) and OpenAI as providers. |
| """ |
| from abc import ABC, abstractmethod |
| from typing import Optional |
| import json |
|
|
| from ..config import get_settings |
|
|
|
|
| class BaseLLMClient(ABC): |
| """Abstract base class for LLM providers.""" |
| |
| @abstractmethod |
| def classify(self, text: str) -> dict: |
| """Classify opportunity text into category and domain.""" |
| pass |
| |
| @abstractmethod |
| def summarize(self, text: str, max_length: int = 150) -> str: |
| """Generate concise summary of opportunity.""" |
| pass |
| |
| @abstractmethod |
| def recommend_action(self, opportunity: dict) -> dict: |
| """Recommend action based on opportunity context.""" |
| pass |
| |
| @abstractmethod |
| def extract_metadata(self, text: str) -> dict: |
| """Extract structured metadata (deadline, location, reward, etc.).""" |
| pass |
|
|
|
|
| class GeminiClient(BaseLLMClient): |
| """Google Gemini implementation.""" |
| |
| def __init__(self, api_key: str): |
| import google.generativeai as genai |
| genai.configure(api_key=api_key) |
| self.model = genai.GenerativeModel('gemini-2.5-flash') |
| |
| def _generate(self, prompt: str, as_json: bool = False) -> str: |
| """Generate response from Gemini.""" |
| response = self.model.generate_content(prompt) |
| return response.text |
| |
| def classify(self, text: str) -> dict: |
| """Classify opportunity into category and domain.""" |
| prompt = f"""Analyze this opportunity and classify it. Return JSON only. |
| |
| TEXT: {text[:2000]} |
| |
| Return this exact JSON structure: |
| {{ |
| "category": "one of: scholarship, fellowship, internship, job, research, hackathon, competition, grant, conference, open_source, investment, weak_signal, other", |
| "domain": "one of: ai, computer_vision, robotics, finance, crypto, academia, mixed", |
| "confidence": 0.0 to 1.0 |
| }}""" |
| |
| try: |
| result = self._generate(prompt) |
| |
| start = result.find('{') |
| end = result.rfind('}') + 1 |
| if start != -1 and end > start: |
| return json.loads(result[start:end]) |
| except Exception as e: |
| print(f"Classification error: {e}") |
| |
| return {"category": "other", "domain": "mixed", "confidence": 0.0} |
| |
| def summarize(self, text: str, max_length: int = 150) -> str: |
| """Generate concise summary.""" |
| prompt = f"""Summarize this opportunity in {max_length} characters or less. |
| Focus on: what it is, who it's for, and deadline if any. |
| |
| TEXT: {text[:2000]} |
| |
| Return only the summary, no quotes or labels.""" |
| |
| try: |
| return self._generate(prompt).strip()[:max_length] |
| except Exception as e: |
| print(f"Summary error: {e}") |
| return text[:max_length] |
| |
| def recommend_action(self, opportunity: dict) -> dict: |
| """ |
| PIOE 2.0 Enhanced Action Guidance. |
| Returns comprehensive recommendations for how to approach the opportunity. |
| """ |
| prompt = f"""You are an expert career and opportunity advisor. Analyze this opportunity and provide detailed action guidance. |
| |
| OPPORTUNITY DETAILS: |
| - Title: {opportunity.get('title', '')} |
| - Category: {opportunity.get('category', '')} |
| - Domain: {opportunity.get('domain', '')} |
| - Deadline: {opportunity.get('deadline', 'No deadline specified')} |
| - Description: {opportunity.get('raw_text', '')[:1500]} |
| - ROI Score: {opportunity.get('roi_score', 'N/A')} |
| - Competition Level: {opportunity.get('competition_level', 'N/A')} |
| - Region: {opportunity.get('region', 'global')} |
| |
| USER CONTEXT: |
| - Location: Nigeria, Africa |
| - Interests: AI, Computer Vision, Robotics, Web3 |
| - Status: Student/Early Career |
| |
| Provide strategic action guidance. Return JSON only: |
| {{ |
| "primary_action": "one of: apply_now, apply_prepared, track, save_for_later, deep_research, network_first, skip", |
| "urgency": "one of: immediate, this_week, this_month, whenever, expired", |
| "timing_status": "one of: early, optimal, late, unknown", |
| |
| "skills_to_highlight": ["skill1", "skill2", "skill3"], |
| "portfolio_pieces": ["project type 1", "project type 2"], |
| |
| "preparation_steps": [ |
| "step 1", |
| "step 2", |
| "step 3" |
| ], |
| |
| "networking_tips": "who to contact or how to stand out (1 sentence)", |
| "differentiation_angle": "what unique angle to take (1 sentence)", |
| |
| "success_probability": 0.0 to 1.0, |
| "time_investment_hours": estimated hours to apply well, |
| "risk_level": "low, medium, or high", |
| |
| "why": "brief strategic reasoning (max 100 chars)", |
| "red_flags": ["any concerns"] or [] |
| }}""" |
| |
| try: |
| result = self._generate(prompt) |
| start = result.find('{') |
| end = result.rfind('}') + 1 |
| if start != -1 and end > start: |
| parsed = json.loads(result[start:end]) |
| |
| return { |
| "primary_action": parsed.get("primary_action", "save_for_later"), |
| "urgency": parsed.get("urgency", "whenever"), |
| "timing_status": parsed.get("timing_status", "unknown"), |
| "skills_to_highlight": parsed.get("skills_to_highlight", []), |
| "portfolio_pieces": parsed.get("portfolio_pieces", []), |
| "preparation_steps": parsed.get("preparation_steps", []), |
| "networking_tips": parsed.get("networking_tips", ""), |
| "differentiation_angle": parsed.get("differentiation_angle", ""), |
| "success_probability": parsed.get("success_probability", 0.3), |
| "time_investment_hours": parsed.get("time_investment_hours", 10), |
| "risk_level": parsed.get("risk_level", "medium"), |
| "why": parsed.get("why", "Review and decide"), |
| "red_flags": parsed.get("red_flags", []), |
| } |
| except Exception as e: |
| print(f"Action guidance error: {e}") |
| |
| |
| return { |
| "primary_action": "save_for_later", |
| "urgency": "whenever", |
| "timing_status": "unknown", |
| "skills_to_highlight": [], |
| "portfolio_pieces": [], |
| "preparation_steps": ["Review the opportunity details", "Assess fit with your goals"], |
| "networking_tips": "", |
| "differentiation_angle": "", |
| "success_probability": 0.3, |
| "time_investment_hours": 10, |
| "risk_level": "medium", |
| "why": "Needs manual review", |
| "red_flags": [], |
| } |
| |
| def extract_metadata(self, text: str) -> dict: |
| """Extract structured metadata from text.""" |
| prompt = f"""Extract metadata from this opportunity text. Return JSON only. |
| |
| TEXT: {text[:2000]} |
| |
| Return this structure (use null for missing info): |
| {{ |
| "deadline": "YYYY-MM-DD or null", |
| "location": "location or 'remote' or null", |
| "reward": "amount or null", |
| "organization": "org name or null", |
| "requirements": ["skill1", "skill2"] or [], |
| "url": "application url or null" |
| }}""" |
| |
| try: |
| result = self._generate(prompt) |
| start = result.find('{') |
| end = result.rfind('}') + 1 |
| if start != -1 and end > start: |
| return json.loads(result[start:end]) |
| except Exception as e: |
| print(f"Metadata extraction error: {e}") |
| |
| return {} |
|
|
|
|
| class OpenAIClient(BaseLLMClient): |
| """OpenAI implementation (fallback).""" |
| |
| def __init__(self, api_key: str): |
| from openai import OpenAI |
| self.client = OpenAI(api_key=api_key) |
| self.model = "gpt-3.5-turbo" |
| |
| def _generate(self, prompt: str) -> str: |
| """Generate response from OpenAI.""" |
| response = self.client.chat.completions.create( |
| model=self.model, |
| messages=[{"role": "user", "content": prompt}], |
| temperature=0.3 |
| ) |
| return response.choices[0].message.content |
| |
| def classify(self, text: str) -> dict: |
| """Classify opportunity - same logic as Gemini.""" |
| prompt = f"""Classify this opportunity. Return JSON only with keys: category, domain, confidence. |
| Categories: scholarship, fellowship, internship, job, research, hackathon, competition, grant, conference, open_source, investment, weak_signal, other |
| Domains: ai, computer_vision, robotics, finance, crypto, academia, mixed |
| |
| TEXT: {text[:2000]}""" |
| |
| try: |
| result = self._generate(prompt) |
| start = result.find('{') |
| end = result.rfind('}') + 1 |
| if start != -1 and end > start: |
| return json.loads(result[start:end]) |
| except Exception: |
| pass |
| return {"category": "other", "domain": "mixed", "confidence": 0.0} |
| |
| def summarize(self, text: str, max_length: int = 150) -> str: |
| prompt = f"Summarize in {max_length} chars: {text[:2000]}" |
| try: |
| return self._generate(prompt).strip()[:max_length] |
| except Exception: |
| return text[:max_length] |
| |
| def recommend_action(self, opportunity: dict) -> dict: |
| return {"action": "save", "reason": "Review later", "urgency": "low"} |
| |
| def extract_metadata(self, text: str) -> dict: |
| return {} |
|
|
|
|
| class LLMClient: |
| """ |
| Factory class that provides the configured LLM client. |
| Uses Gemini by default, falls back to OpenAI if configured. |
| """ |
| |
| _instance: Optional[BaseLLMClient] = None |
| |
| @classmethod |
| def get_client(cls) -> BaseLLMClient: |
| """Get or create the LLM client instance.""" |
| if cls._instance is None: |
| settings = get_settings() |
| |
| if settings.ai_provider == "gemini" and settings.gemini_api_key: |
| cls._instance = GeminiClient(settings.gemini_api_key) |
| elif settings.openai_api_key: |
| cls._instance = OpenAIClient(settings.openai_api_key) |
| else: |
| |
| cls._instance = MockLLMClient() |
| |
| return cls._instance |
|
|
|
|
| class MockLLMClient(BaseLLMClient): |
| """Mock client for development without API keys. PIOE 2.0 compatible.""" |
| |
| def classify(self, text: str) -> dict: |
| |
| text_lower = text.lower() |
| |
| if any(kw in text_lower for kw in ["scholarship", "fellowship", "grant"]): |
| return {"category": "scholarship", "domain": "academia", "confidence": 0.7} |
| elif any(kw in text_lower for kw in ["hackathon", "competition", "challenge"]): |
| return {"category": "hackathon", "domain": "ai", "confidence": 0.7} |
| elif any(kw in text_lower for kw in ["internship", "intern"]): |
| return {"category": "internship", "domain": "mixed", "confidence": 0.7} |
| elif any(kw in text_lower for kw in ["job", "hiring", "position"]): |
| return {"category": "job", "domain": "mixed", "confidence": 0.7} |
| elif any(kw in text_lower for kw in ["bounty", "ecosystem", "solana", "ethereum"]): |
| return {"category": "bounty", "domain": "crypto", "confidence": 0.7} |
| elif any(kw in text_lower for kw in ["pitch", "demo day", "accelerator"]): |
| return {"category": "pitch_event", "domain": "mixed", "confidence": 0.7} |
| elif any(kw in text_lower for kw in ["collaborat", "partner", "looking for"]): |
| return {"category": "collaboration", "domain": "mixed", "confidence": 0.6} |
| |
| return {"category": "other", "domain": "mixed", "confidence": 0.3} |
| |
| def summarize(self, text: str, max_length: int = 150) -> str: |
| return text[:max_length] |
| |
| def recommend_action(self, opportunity: dict) -> dict: |
| """PIOE 2.0 action guidance - rule-based fallback.""" |
| category = opportunity.get("category", "other") |
| |
| |
| action_map = { |
| "hackathon": ("apply_now", "this_week", ["Python", "ML/AI"], ["Previous hackathon project"]), |
| "grant": ("apply_prepared", "this_month", ["Technical writing", "Project planning"], ["Open source contributions"]), |
| "ecosystem_grant": ("apply_prepared", "this_month", ["Solidity/Rust", "Web3"], ["DApp or smart contract"]), |
| "internship": ("apply_now", "this_week", ["Relevant coursework", "Projects"], ["GitHub portfolio"]), |
| "scholarship": ("apply_prepared", "this_month", ["Academic excellence", "Leadership"], ["Research paper or thesis"]), |
| "bounty": ("apply_now", "immediate", ["Specific tech stack"], ["Related code samples"]), |
| "pitch_event": ("apply_prepared", "this_month", ["Presentation", "Business model"], ["Pitch deck", "Demo video"]), |
| "collaboration": ("network_first", "whenever", ["Domain expertise"], ["Relevant projects"]), |
| } |
| |
| action, urgency, skills, portfolio = action_map.get( |
| category, |
| ("save_for_later", "whenever", [], []) |
| ) |
| |
| return { |
| "primary_action": action, |
| "urgency": urgency, |
| "timing_status": "unknown", |
| "skills_to_highlight": skills, |
| "portfolio_pieces": portfolio, |
| "preparation_steps": [ |
| "Review the opportunity requirements", |
| "Prepare relevant materials", |
| "Submit before deadline" |
| ], |
| "networking_tips": "Research the organization and connect with past participants", |
| "differentiation_angle": "Highlight unique projects and Africa/Nigeria perspective", |
| "success_probability": 0.3, |
| "time_investment_hours": 10, |
| "risk_level": "medium", |
| "why": f"Standard approach for {category}", |
| "red_flags": [], |
| } |
| |
| def extract_metadata(self, text: str) -> dict: |
| return {} |
|
|
|
|
