import time import uuid from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait import click import requests from requests import get from uuid import uuid4 from re import findall from requests.exceptions import RequestException from curl_cffi.requests import get, RequestsError import g4f from random import randint from PIL import Image import io import re import json import yaml from ..AIutel import Optimizers from ..AIutel import Conversation from ..AIutel import AwesomePrompts, sanitize_stream from ..AIbase import Provider, AsyncProvider from Helpingai_T2 import Perplexity from webscout import exceptions from typing import Any, AsyncGenerator, Dict import logging import httpx #----------------------------------------------------------OpenAI----------------------------------- class OPENAI(Provider): def __init__( self, api_key: str, is_conversation: bool = True, max_tokens: int = 600, temperature: float = 1, presence_penalty: int = 0, frequency_penalty: int = 0, top_p: float = 1, model: str = "gpt-3.5-turbo", timeout: int = 30, intro: str = None, filepath: str = None, update_file: bool = True, proxies: dict = {}, history_offset: int = 10250, act: str = None, ): """Instantiates OPENAI Args: api_key (key): OpenAI's API key. is_conversation (bool, optional): Flag for chatting conversationally. Defaults to True. max_tokens (int, optional): Maximum number of tokens to be generated upon completion. Defaults to 600. temperature (float, optional): Charge of the generated text's randomness. Defaults to 1. presence_penalty (int, optional): Chances of topic being repeated. Defaults to 0. frequency_penalty (int, optional): Chances of word being repeated. Defaults to 0. top_p (float, optional): Sampling threshold during inference time. Defaults to 0.999. model (str, optional): LLM model name. Defaults to "gpt-3.5-turbo". timeout (int, optional): Http request timeout. Defaults to 30. intro (str, optional): Conversation introductory prompt. Defaults to None. filepath (str, optional): Path to file containing conversation history. Defaults to None. update_file (bool, optional): Add new prompts and responses to the file. Defaults to True. proxies (dict, optional): Http request proxies. Defaults to {}. history_offset (int, optional): Limit conversation history to this number of last texts. Defaults to 10250. act (str|int, optional): Awesome prompt key or index. (Used as intro). Defaults to None. """ self.is_conversation = is_conversation self.max_tokens_to_sample = max_tokens self.api_key = api_key self.model = model self.temperature = temperature self.presence_penalty = presence_penalty self.frequency_penalty = frequency_penalty self.top_p = top_p self.chat_endpoint = "https://api.openai.com/v1/chat/completions" self.stream_chunk_size = 64 self.timeout = timeout self.last_response = {} self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", } self.__available_optimizers = ( method for method in dir(Optimizers) if callable(getattr(Optimizers, method)) and not method.startswith("__") ) self.session.headers.update(self.headers) Conversation.intro = ( AwesomePrompts().get_act( act, raise_not_found=True, default=None, case_insensitive=True ) if act else intro or Conversation.intro ) self.conversation = Conversation( is_conversation, self.max_tokens_to_sample, filepath, update_file ) self.conversation.history_offset = history_offset self.session.proxies = proxies def ask( self, prompt: str, stream: bool = False, raw: bool = False, optimizer: str = None, conversationally: bool = False, ) -> dict: """Chat with AI Args: prompt (str): Prompt to be send. stream (bool, optional): Flag for streaming response. Defaults to False. raw (bool, optional): Stream back raw response as received. Defaults to False. optimizer (str, optional): Prompt optimizer name - `[code, shell_command]`. Defaults to None. conversationally (bool, optional): Chat conversationally when using optimizer. Defaults to False. Returns: dict : {} ```json { "id": "chatcmpl-TaREJpBZsRVQFRFic1wIA7Q7XfnaD", "object": "chat.completion", "created": 1704623244, "model": "gpt-3.5-turbo", "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 }, "choices": [ { "message": { "role": "assistant", "content": "Hello! How can I assist you today?" }, "finish_reason": "stop", "index": 0 } ] } ``` """ conversation_prompt = self.conversation.gen_complete_prompt(prompt) if optimizer: if optimizer in self.__available_optimizers: conversation_prompt = getattr(Optimizers, optimizer)( conversation_prompt if conversationally else prompt ) else: raise exceptions.FailedToGenerateResponseError( f"Optimizer is not one of {self.__available_optimizers}" ) self.session.headers.update(self.headers) payload = { "frequency_penalty": self.frequency_penalty, "messages": [{"content": conversation_prompt, "role": "user"}], "model": self.model, "presence_penalty": self.presence_penalty, "stream": stream, "temperature": self.temperature, "top_p": self.top_p, } def for_stream(): response = self.session.post( self.chat_endpoint, json=payload, stream=True, timeout=self.timeout ) if not response.ok: raise exceptions.FailedToGenerateResponseError( f"Failed to generate response - ({response.status_code}, {response.reason}) - {response.text}" ) message_load = "" for value in response.iter_lines( decode_unicode=True, delimiter="" if raw else "data:", chunk_size=self.stream_chunk_size, ): try: resp = json.loads(value) incomplete_message = self.get_message(resp) if incomplete_message: message_load += incomplete_message resp["choices"][0]["delta"]["content"] = message_load self.last_response.update(resp) yield value if raw else resp elif raw: yield value except json.decoder.JSONDecodeError: pass self.conversation.update_chat_history( prompt, self.get_message(self.last_response) ) def for_non_stream(): response = self.session.post( self.chat_endpoint, json=payload, stream=False, timeout=self.timeout ) if ( not response.ok or not response.headers.get("Content-Type", "") == "application/json" ): raise exceptions.FailedToGenerateResponseError( f"Failed to generate response - ({response.status_code}, {response.reason}) - {response.text}" ) resp = response.json() self.last_response.update(resp) self.conversation.update_chat_history( prompt, self.get_message(self.last_response) ) return resp return for_stream() if stream else for_non_stream() def chat( self, prompt: str, stream: bool = False, optimizer: str = None, conversationally: bool = False, ) -> str: """Generate response `str` Args: prompt (str): Prompt to be send. stream (bool, optional): Flag for streaming response. Defaults to False. optimizer (str, optional): Prompt optimizer name - `[code, shell_command]`. Defaults to None. conversationally (bool, optional): Chat conversationally when using optimizer. Defaults to False. Returns: str: Response generated """ def for_stream(): for response in self.ask( prompt, True, optimizer=optimizer, conversationally=conversationally ): yield self.get_message(response) def for_non_stream(): return self.get_message( self.ask( prompt, False, optimizer=optimizer, conversationally=conversationally, ) ) return for_stream() if stream else for_non_stream() def get_message(self, response: dict) -> str: """Retrieves message only from response Args: response (dict): Response generated by `self.ask` Returns: str: Message extracted """ assert isinstance(response, dict), "Response should be of dict data-type only" try: if response["choices"][0].get("delta"): return response["choices"][0]["delta"]["content"] return response["choices"][0]["message"]["content"] except KeyError: return "" class AsyncOPENAI(AsyncProvider): def __init__( self, api_key: str, is_conversation: bool = True, max_tokens: int = 600, temperature: float = 1, presence_penalty: int = 0, frequency_penalty: int = 0, top_p: float = 1, model: str = "gpt-3.5-turbo", timeout: int = 30, intro: str = None, filepath: str = None, update_file: bool = True, proxies: dict = {}, history_offset: int = 10250, act: str = None, ): """Instantiates OPENAI Args: api_key (key): OpenAI's API key. is_conversation (bool, optional): Flag for chatting conversationally. Defaults to True. max_tokens (int, optional): Maximum number of tokens to be generated upon completion. Defaults to 600. temperature (float, optional): Charge of the generated text's randomness. Defaults to 1. presence_penalty (int, optional): Chances of topic being repeated. Defaults to 0. frequency_penalty (int, optional): Chances of word being repeated. Defaults to 0. top_p (float, optional): Sampling threshold during inference time. Defaults to 0.999. model (str, optional): LLM model name. Defaults to "gpt-3.5-turbo". timeout (int, optional): Http request timeout. Defaults to 30. intro (str, optional): Conversation introductory prompt. Defaults to None. filepath (str, optional): Path to file containing conversation history. Defaults to None. update_file (bool, optional): Add new prompts and responses to the file. Defaults to True. proxies (dict, optional): Http request proxies. Defaults to {}. history_offset (int, optional): Limit conversation history to this number of last texts. Defaults to 10250. act (str|int, optional): Awesome prompt key or index. (Used as intro). Defaults to None. """ self.is_conversation = is_conversation self.max_tokens_to_sample = max_tokens self.api_key = api_key self.model = model self.temperature = temperature self.presence_penalty = presence_penalty self.frequency_penalty = frequency_penalty self.top_p = top_p self.chat_endpoint = "https://api.openai.com/v1/chat/completions" self.stream_chunk_size = 64 self.timeout = timeout self.last_response = {} self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", } self.__available_optimizers = ( method for method in dir(Optimizers) if callable(getattr(Optimizers, method)) and not method.startswith("__") ) Conversation.intro = ( AwesomePrompts().get_act( act, raise_not_found=True, default=None, case_insensitive=True ) if act else intro or Conversation.intro ) self.conversation = Conversation( is_conversation, self.max_tokens_to_sample, filepath, update_file ) self.conversation.history_offset = history_offset self.session = httpx.AsyncClient( headers=self.headers, proxies=proxies, ) async def ask( self, prompt: str, stream: bool = False, raw: bool = False, optimizer: str = None, conversationally: bool = False, ) -> dict | AsyncGenerator: """Chat with AI asynchronously. Args: prompt (str): Prompt to be send. stream (bool, optional): Flag for streaming response. Defaults to False. raw (bool, optional): Stream back raw response as received. Defaults to False. optimizer (str, optional): Prompt optimizer name - `[code, shell_command]`. Defaults to None. conversationally (bool, optional): Chat conversationally when using optimizer. Defaults to False. Returns: dict|AsyncGenerator : ai content. ```json { "id": "chatcmpl-TaREJpBZsRVQFRFic1wIA7Q7XfnaD", "object": "chat.completion", "created": 1704623244, "model": "gpt-3.5-turbo", "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 }, "choices": [ { "message": { "role": "assistant", "content": "Hello! How can I assist you today?" }, "finish_reason": "stop", "index": 0 } ] } ``` """ conversation_prompt = self.conversation.gen_complete_prompt(prompt) if optimizer: if optimizer in self.__available_optimizers: conversation_prompt = getattr(Optimizers, optimizer)( conversation_prompt if conversationally else prompt ) else: raise Exception( f"Optimizer is not one of {self.__available_optimizers}" ) payload = { "frequency_penalty": self.frequency_penalty, "messages": [{"content": conversation_prompt, "role": "user"}], "model": self.model, "presence_penalty": self.presence_penalty, "stream": stream, "temperature": self.temperature, "top_p": self.top_p, } async def for_stream(): async with self.session.stream( "POST", self.chat_endpoint, json=payload, timeout=self.timeout ) as response: if not response.is_success: raise Exception( f"Failed to generate response - ({response.status_code}, {response.reason_phrase})" ) message_load = "" async for value in response.aiter_lines(): try: resp = sanitize_stream(value) incomplete_message = await self.get_message(resp) if incomplete_message: message_load += incomplete_message resp["choices"][0]["delta"]["content"] = message_load self.last_response.update(resp) yield value if raw else resp elif raw: yield value except json.decoder.JSONDecodeError: pass self.conversation.update_chat_history( prompt, await self.get_message(self.last_response) ) async def for_non_stream(): response = httpx.post( self.chat_endpoint, json=payload, timeout=self.timeout, headers=self.headers, ) if ( not response.is_success or not response.headers.get("Content-Type", "") == "application/json" ): raise Exception( f"Failed to generate response - ({response.status_code}, {response.reason_phrase})" ) resp = response.json() self.last_response.update(resp) self.conversation.update_chat_history( prompt, await self.get_message(self.last_response) ) return resp return for_stream() if stream else await for_non_stream() async def chat( self, prompt: str, stream: bool = False, optimizer: str = None, conversationally: bool = False, ) -> str | AsyncGenerator: """Generate response `str` asynchronously. Args: prompt (str): Prompt to be send. stream (bool, optional): Flag for streaming response. Defaults to False. optimizer (str, optional): Prompt optimizer name - `[code, shell_command]`. Defaults to None. conversationally (bool, optional): Chat conversationally when using optimizer. Defaults to False. Returns: str|AsyncGenerator: Response generated """ async def for_stream(): async_ask = await self.ask( prompt, True, optimizer=optimizer, conversationally=conversationally ) async for response in async_ask: yield await self.get_message(response) async def for_non_stream(): return await self.get_message( await self.ask( prompt, False, optimizer=optimizer, conversationally=conversationally, ) ) return for_stream() if stream else await for_non_stream() async def get_message(self, response: dict) -> str: """Retrieves message only from response asynchronously. Args: response (dict): Response generated by `self.ask` Returns: str: Message extracted """ assert isinstance(response, dict), "Response should be of dict data-type only" try: if response["choices"][0].get("delta"): return response["choices"][0]["delta"]["content"] return response["choices"][0]["message"]["content"] except KeyError: return ""