pratikshahp/HR-Learning-and-Development-Recommendation-System
0
1import os2from typing import List, Dict, Optional3from openai import OpenAI4from strategy import StrategyFactory, ExecutionStrategy5 6class Agent:7 def __init__(self, name: str):8 self._name = name9 self._persona = "You are an AI assistant specializing in Learning & Development recommendations."10 self._instruction = "Provide clear, well-reasoned recommendations based on performance, skill gaps, and feedback."11 self._task = ""12 self._api_key = os.getenv('OPENAI_API_KEY', '')13 self._model = "gpt-4o-mini"14 self._history: List[Dict[str, str]] = []15 self._strategy: Optional[ExecutionStrategy] = None16 17 @property18 def name(self) -> str:19 return self._name20 21 @property22 def strategy(self) -> Optional[ExecutionStrategy]:23 return self._strategy24 25 @strategy.setter26 def strategy(self, strategy_name: str):27 self._strategy = StrategyFactory.create_strategy(strategy_name)28 29 def _build_messages(self, task: Optional[str] = None) -> List[Dict[str, str]]:30 messages = [{"role": "system", "content": self._persona}]31 messages.append({"role": "user", "content": self._instruction})32 33 # Add conversation history34 messages.extend(self._history)35 current_task = task or self._task36 37 if self._strategy and current_task:38 current_task = self._strategy.build_prompt(current_task, self._instruction)39 40 if current_task:41 messages.append({"role": "user", "content": current_task})42 43 return messages44 45 def execute(self, task: Optional[str] = None) -> str:46 if task is not None:47 self._task = task48 49 if not self._api_key:50 return "API key not found. Please set the OPENAI_API_KEY environment variable."51 52 if not self._task:53 return "No task specified. Please provide a task to execute."54 55 client = OpenAI(api_key=self._api_key)56 messages = self._build_messages()57 58 try:59 response = client.chat.completions.create(60 model=self._model,61 messages=messages62 )63 response_content = response.choices[0].message.content64 65 if self._strategy:66 response_content = self._strategy.process_response(response_content)67 68 self._history.append({"role": "user", "content": self._task})69 self._history.append({"role": "assistant", "content": response_content})70 71 self._task = ""72 return response_content73 except Exception as e:74 return f"An error occurred: {str(e)}"75 76 def available_strategies(self) -> List[str]:77 return StrategyFactory.available_strategies()78 