Spaces:
Running
Running
| 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 = "" | |
| self._instruction = "" | |
| 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 | |
| def name(self) -> str: | |
| return self._name | |
| def persona(self) -> str: | |
| return self._persona | |
| def persona(self, value: str): | |
| self._persona = value | |
| def instruction(self) -> str: | |
| return self._instruction | |
| def instruction(self, value: str): | |
| self._instruction = value | |
| def task(self) -> str: | |
| return self._task | |
| def task(self, value: str): | |
| self._task = value | |
| def strategy(self) -> Optional[ExecutionStrategy]: | |
| return self._strategy | |
| def strategy(self, strategy_name: str): | |
| """Set the execution strategy by name.""" | |
| self._strategy = StrategyFactory.create_strategy(strategy_name) | |
| def history(self) -> List[Dict[str, str]]: | |
| return self._history | |
| def _build_messages(self, task: Optional[str] = None) -> List[Dict[str, str]]: | |
| """Build the messages list including persona, instruction, and history.""" | |
| messages = [{"role": "system", "content": self.persona}] | |
| if self.instruction: | |
| messages.append({ | |
| "role": "user", | |
| "content": f"Global Instruction: {self.instruction}" | |
| }) | |
| # Add conversation history | |
| messages.extend(self._history) | |
| # Use provided task or stored task | |
| current_task = task if task is not None else self._task | |
| # Apply strategy if set | |
| if self._strategy and current_task: | |
| current_task = self._strategy.build_prompt(current_task, self.instruction) | |
| # Add the current task if it exists | |
| if current_task: | |
| messages.append({"role": "user", "content": current_task}) | |
| return messages | |
| def execute(self, task: Optional[str] = None) -> str: | |
| """Execute a task using the configured LLM.""" | |
| 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 | |
| # Process response through strategy if set | |
| if self._strategy: | |
| response_content = self._strategy.process_response(response_content) | |
| # Store the interaction in history | |
| self._history.append({"role": "user", "content": self._task}) | |
| self._history.append({ | |
| "role": "assistant", | |
| "content": response_content | |
| }) | |
| # Clear the task after execution | |
| self._task = "" | |
| return response_content | |
| except Exception as e: | |
| return f"An error occurred: {str(e)}" | |
| def clear_history(self): | |
| """Clear the conversation history.""" | |
| self._history = [] | |
| def chat(self, user_input: str) -> str: | |
| """Process user input and return the assistant's response.""" | |
| return self.execute(user_input) | |
| def available_strategies(self) -> List[str]: | |
| """Return a list of available strategy names.""" | |
| return StrategyFactory.available_strategies() |