Spaces:
Running
Running
| import os | |
| from openai import OpenAI | |
| from typing import List, Optional | |
| from dotenv import load_dotenv, find_dotenv | |
| from ..utils import log | |
| class GPTClient: | |
| def __init__(self, model: str, api_key: Optional[str] = None): | |
| self.model = model | |
| self._init_environment(api_key) | |
| self.histories = list() | |
| self.token_usages = dict() | |
| self.__first_turn = True | |
| def _init_environment(self, api_key: Optional[str] = None) -> None: | |
| if not api_key: | |
| dotenv_path = find_dotenv(usecwd=True) | |
| load_dotenv(dotenv_path, override=True) | |
| api_key = os.environ.get("OPENAI_API_KEY", None) | |
| self.client = OpenAI(api_key=api_key) | |
| def reset_history(self, verbose: bool = True) -> None: | |
| self.__first_turn = True | |
| self.histories = list() | |
| self.token_usages = dict() | |
| if verbose: | |
| log('Conversation history has been reset.', color=True) | |
| def __make_payload(self, user_prompt: str) -> List[dict]: | |
| return [{"role": "user", "content": [{"type": "text", "text": user_prompt}]}] | |
| def __call__(self, | |
| user_prompt: str, | |
| system_prompt: Optional[str] = None, | |
| using_multi_turn: bool = True, | |
| greeting: Optional[str] = None, | |
| verbose: bool = True, | |
| **kwargs) -> str: | |
| try: | |
| if not using_multi_turn: | |
| self.reset_history(verbose) | |
| if self.__first_turn: | |
| if system_prompt: | |
| self.histories.append({"role": "system", "content": [{"type": "text", "text": system_prompt}]}) | |
| if greeting: | |
| self.histories.append({"role": "assistant", "content": [{"type": "text", "text": greeting}]}) | |
| self.__first_turn = False | |
| self.histories += self.__make_payload(user_prompt) | |
| # kwargs.pop('seed', None) # passed by agents but not valid for all models | |
| kwargs.pop('verbose', None) # internal flag | |
| # gpt 5-series reasoning models: use minimum reasoning effort | |
| kwargs['reasoning_effort'] = 'none' | |
| response = self.client.chat.completions.create( | |
| model=self.model, | |
| messages=self.histories, | |
| **kwargs | |
| ) | |
| assistant_msg = response.choices[0].message | |
| self.histories.append({"role": assistant_msg.role, "content": [{"type": "text", "text": assistant_msg.content}]}) | |
| if response.usage: | |
| self.token_usages.setdefault("prompt_tokens", []).append(response.usage.prompt_tokens) | |
| self.token_usages.setdefault("completion_tokens", []).append(response.usage.completion_tokens) | |
| self.token_usages.setdefault("total_tokens", []).append(response.usage.total_tokens) | |
| self.token_usages.setdefault("reasoning_tokens", []).append(response.usage.completion_tokens_details.reasoning_tokens) | |
| return assistant_msg.content | |
| except Exception as e: | |
| raise e | |