File size: 2,767 Bytes
48c9ae6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 |
import os
from typing import List, Dict, Optional
from openai import OpenAI
from strategy import StrategyFactory, ExecutionStrategy
class Agent:
def __init__(self, name: str):
self._name = name
self._persona = "You are an AI assistant specializing in Learning & Development recommendations."
self._instruction = "Provide clear, well-reasoned recommendations based on performance, skill gaps, and feedback."
self._task = ""
self._api_key = os.getenv('OPENAI_API_KEY', '')
self._model = "gpt-4o-mini"
self._history: List[Dict[str, str]] = []
self._strategy: Optional[ExecutionStrategy] = None
@property
def name(self) -> str:
return self._name
@property
def strategy(self) -> Optional[ExecutionStrategy]:
return self._strategy
@strategy.setter
def strategy(self, strategy_name: str):
self._strategy = StrategyFactory.create_strategy(strategy_name)
def _build_messages(self, task: Optional[str] = None) -> List[Dict[str, str]]:
messages = [{"role": "system", "content": self._persona}]
messages.append({"role": "user", "content": self._instruction})
# Add conversation history
messages.extend(self._history)
current_task = task or self._task
if self._strategy and current_task:
current_task = self._strategy.build_prompt(current_task, self._instruction)
if current_task:
messages.append({"role": "user", "content": current_task})
return messages
def execute(self, task: Optional[str] = None) -> str:
if task is not None:
self._task = task
if not self._api_key:
return "API key not found. Please set the OPENAI_API_KEY environment variable."
if not self._task:
return "No task specified. Please provide a task to execute."
client = OpenAI(api_key=self._api_key)
messages = self._build_messages()
try:
response = client.chat.completions.create(
model=self._model,
messages=messages
)
response_content = response.choices[0].message.content
if self._strategy:
response_content = self._strategy.process_response(response_content)
self._history.append({"role": "user", "content": self._task})
self._history.append({"role": "assistant", "content": response_content})
self._task = ""
return response_content
except Exception as e:
return f"An error occurred: {str(e)}"
def available_strategies(self) -> List[str]:
return StrategyFactory.available_strategies()
|