import requests
import json
from typing import *
from webscout.AIutel import Optimizers
from webscout.AIutel import Conversation
from webscout.AIutel import AwesomePrompts
from webscout.AIbase import Provider
from webscout import exceptions
class TypeGPT(Provider):
"""
A class to interact with the TypeGPT.net API. Improved to match webscout standards.
"""
url = "https://chat.typegpt.net"
working = True
supports_message_history = True
models = [
# OpenAI Models
"gpt-3.5-turbo",
"gpt-3.5-turbo-202201",
"gpt-4o",
"gpt-4o-2024-05-13",
"o1-preview",
# Claude Models
"claude",
"claude-3-5-sonnet",
"claude-sonnet-3.5",
"claude-3-5-sonnet-20240620",
# Meta/LLaMA Models
"@cf/meta/llama-2-7b-chat-fp16",
"@cf/meta/llama-2-7b-chat-int8",
"@cf/meta/llama-3-8b-instruct",
"@cf/meta/llama-3.1-8b-instruct",
"@cf/meta-llama/llama-2-7b-chat-hf-lora",
"llama-3.1-405b",
"llama-3.1-70b",
"llama-3.1-8b",
"meta-llama/Llama-2-7b-chat-hf",
"meta-llama/Llama-3.1-70B-Instruct",
"meta-llama/Llama-3.1-8B-Instruct",
"meta-llama/Llama-3.2-11B-Vision-Instruct",
"meta-llama/Llama-3.2-1B-Instruct",
"meta-llama/Llama-3.2-3B-Instruct",
"meta-llama/Llama-3.2-90B-Vision-Instruct",
"meta-llama/Llama-Guard-3-8B",
"meta-llama/Meta-Llama-3-70B-Instruct",
"meta-llama/Meta-Llama-3-8B-Instruct",
"meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
"meta-llama/Meta-Llama-3.1-8B-Instruct",
"meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
# Mistral Models
"mistral",
"mistral-large",
"@cf/mistral/mistral-7b-instruct-v0.1",
"@cf/mistral/mistral-7b-instruct-v0.2-lora",
"@hf/mistralai/mistral-7b-instruct-v0.2",
"mistralai/Mistral-7B-Instruct-v0.2",
"mistralai/Mistral-7B-Instruct-v0.3",
"mistralai/Mixtral-8x22B-Instruct-v0.1",
"mistralai/Mixtral-8x7B-Instruct-v0.1",
# Qwen Models
"@cf/qwen/qwen1.5-0.5b-chat",
"@cf/qwen/qwen1.5-1.8b-chat",
"@cf/qwen/qwen1.5-7b-chat-awq",
"@cf/qwen/qwen1.5-14b-chat-awq",
"Qwen/Qwen2.5-3B-Instruct",
"Qwen/Qwen2.5-72B-Instruct",
"Qwen/Qwen2.5-Coder-32B-Instruct",
# Google/Gemini Models
"@cf/google/gemma-2b-it-lora",
"@cf/google/gemma-7b-it-lora",
"@hf/google/gemma-7b-it",
"google/gemma-1.1-2b-it",
"google/gemma-1.1-7b-it",
"gemini-pro",
"gemini-1.5-pro",
"gemini-1.5-pro-latest",
"gemini-1.5-flash",
# Cohere Models
"c4ai-aya-23-35b",
"c4ai-aya-23-8b",
"command",
"command-light",
"command-light-nightly",
"command-nightly",
"command-r",
"command-r-08-2024",
"command-r-plus",
"command-r-plus-08-2024",
"rerank-english-v2.0",
"rerank-english-v3.0",
"rerank-multilingual-v2.0",
"rerank-multilingual-v3.0",
# Microsoft Models
"@cf/microsoft/phi-2",
"microsoft/DialoGPT-medium",
"microsoft/Phi-3-medium-4k-instruct",
"microsoft/Phi-3-mini-4k-instruct",
"microsoft/Phi-3.5-mini-instruct",
"microsoft/WizardLM-2-8x22B",
# Yi Models
"01-ai/Yi-1.5-34B-Chat",
"01-ai/Yi-34B-Chat",
# Specialized Models and Tools
"@cf/deepseek-ai/deepseek-math-7b-base",
"@cf/deepseek-ai/deepseek-math-7b-instruct",
"@cf/defog/sqlcoder-7b-2",
"@cf/openchat/openchat-3.5-0106",
"@cf/thebloke/discolm-german-7b-v1-awq",
"@cf/tiiuae/falcon-7b-instruct",
"@cf/tinyllama/tinyllama-1.1b-chat-v1.0",
"@hf/nexusflow/starling-lm-7b-beta",
"@hf/nousresearch/hermes-2-pro-mistral-7b",
"@hf/thebloke/deepseek-coder-6.7b-base-awq",
"@hf/thebloke/deepseek-coder-6.7b-instruct-awq",
"@hf/thebloke/llama-2-13b-chat-awq",
"@hf/thebloke/llamaguard-7b-awq",
"@hf/thebloke/neural-chat-7b-v3-1-awq",
"@hf/thebloke/openhermes-2.5-mistral-7b-awq",
"@hf/thebloke/zephyr-7b-beta-awq",
"AndroidDeveloper",
"AngularJSAgent",
"AzureAgent",
"BitbucketAgent",
"DigitalOceanAgent",
"DockerAgent",
"ElectronAgent",
"ErlangAgent",
"FastAPIAgent",
"FirebaseAgent",
"FlaskAgent",
"FlutterAgent",
"GitAgent",
"GitlabAgent",
"GoAgent",
"GodotAgent",
"GoogleCloudAgent",
"HTMLAgent",
"HerokuAgent",
"ImageGeneration",
"JavaAgent",
"JavaScriptAgent",
"MongoDBAgent",
"Next.jsAgent",
"PyTorchAgent",
"PythonAgent",
"ReactAgent",
"RepoMap",
"SwiftDeveloper",
"XcodeAgent",
"YoutubeAgent",
"blackboxai",
"blackboxai-pro",
"builderAgent",
"dify",
"flux",
"openchat/openchat-3.6-8b",
"rtist",
"searchgpt",
"sur",
"sur-mistral",
"unity"
]
def __init__(
self,
is_conversation: bool = True,
max_tokens: int = 4000, # Set a reasonable default
timeout: int = 30,
intro: str = None,
filepath: str = None,
update_file: bool = True,
proxies: dict = {},
history_offset: int = 10250,
act: str = None,
model: str = "claude-3-5-sonnet-20240620",
system_prompt: str = "You are a helpful assistant.",
temperature: float = 0.5,
presence_penalty: int = 0,
frequency_penalty: int = 0,
top_p: float = 1,
):
"""Initializes the TypeGPT API client."""
if model not in self.models:
raise ValueError(f"Invalid model: {model}. Choose from: {', '.join(self.models)}")
self.session = requests.Session()
self.is_conversation = is_conversation
self.max_tokens_to_sample = max_tokens
self.api_endpoint = "https://chat.typegpt.net/api/openai/v1/chat/completions"
self.timeout = timeout
self.last_response = {}
self.last_response_status_code = None # Added line for status code
self.model = model
self.system_prompt = system_prompt
self.temperature = temperature
self.presence_penalty = presence_penalty
self.frequency_penalty = frequency_penalty
self.top_p = top_p
self.headers = {
"authority": "chat.typegpt.net",
"accept": "application/json, text/event-stream",
"accept-language": "en-US,en;q=0.9",
"content-type": "application/json",
"origin": "https://chat.typegpt.net",
"referer": "https://chat.typegpt.net/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
}
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.proxies = proxies
def ask(
self,
prompt: str,
stream: bool = False,
raw: bool = False,
optimizer: str = None,
conversationally: bool = False,
) -> Dict[str, Any] | Generator:
"""Sends a prompt to the TypeGPT.net API and returns the response."""
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}"
)
payload = {
"messages": [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": conversation_prompt}
],
"stream": stream,
"model": self.model,
"temperature": self.temperature,
"presence_penalty": self.presence_penalty,
"frequency_penalty": self.frequency_penalty,
"top_p": self.top_p,
"max_tokens": self.max_tokens_to_sample,
}
def for_stream():
response = self.session.post(
self.api_endpoint, headers=self.headers, json=payload, stream=True, timeout=self.timeout
)
self.last_response_status_code = response.status_code # Capture status code
if not response.ok:
raise exceptions.FailedToGenerateResponseError(
f"Failed to generate response - ({response.status_code}, {response.reason}) - {response.text}"
)
message_load = ""
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data: "):
line = line[6:] # Remove "data: " prefix
# Skip [DONE] message
if line.strip() == "[DONE]":
break
try:
data = json.loads(line)
# Extract and yield only new content
if 'choices' in data and len(data['choices']) > 0:
delta = data['choices'][0].get('delta', {})
if 'content' in delta:
new_content = delta['content']
message_load += new_content
# Yield only the new content
yield dict(text=new_content) if not raw else new_content
self.last_response = dict(text=message_load)
except json.JSONDecodeError:
continue
self.conversation.update_chat_history(prompt, self.get_message(self.last_response))
def for_non_stream():
response = self.session.post(self.api_endpoint, headers=self.headers, json=payload)
self.last_response_status_code = response.status_code # Capture status code
if not response.ok:
raise exceptions.FailedToGenerateResponseError(
f"Request failed - {response.status_code}: {response.text}"
)
self.last_response = response.json()
self.conversation.update_chat_history(prompt, self.get_message(self.last_response))
return self.last_response
return for_stream() if stream else for_non_stream()
def chat(
self,
prompt: str,
stream: bool = False,
optimizer: str = None,
conversationally: bool = False,
) -> str | Generator[str, None, None]:
"""Generate response `str` or stream."""
if stream:
gen = self.ask(
prompt, stream=True, optimizer=optimizer, conversationally=conversationally
)
for chunk in gen:
yield self.get_message(chunk) # Extract text from streamed chunks
else:
return self.get_message(self.ask(prompt, stream=False, optimizer=optimizer, conversationally=conversationally))
def get_message(self, response: Dict[str, Any]) -> str:
"""Retrieves message from response."""
if isinstance(response, str): # Handle raw responses
return response
elif isinstance(response, dict):
assert isinstance(response, dict), "Response should be of dict data-type only"
return response.get("text", "") # Extract text from dictionary response
else:
raise TypeError("Invalid response type. Expected str or dict.")
if __name__ == "__main__":
from rich import print
from rich.progress import Progress, BarColumn, TextColumn, TimeRemainingColumn, SpinnerColumn
from rich.console import Console
from rich.table import Table
import concurrent.futures
def make_api_call(thread_number, results):
ai = TypeGPT()
try:
ai.ask("Test message", stream=False)
status_code = ai.last_response_status_code
results[thread_number] = status_code
except Exception as e:
results[thread_number] = str(e)
results = {}
total_requests = 100
console = Console()
print("[bold magenta]Starting API Load Test with 100 simultaneous requests...[/bold magenta]\n")
with Progress(
SpinnerColumn(),
"[progress.description]{task.description}",
BarColumn(bar_width=None),
"[progress.percentage]{task.percentage:>3.0f}%",
TimeRemainingColumn(),
console=console,
) as progress:
task = progress.add_task("[cyan]Sending API Requests...", total=total_requests)
with concurrent.futures.ThreadPoolExecutor(max_workers=total_requests) as executor:
futures = {
executor.submit(make_api_call, i, results): i for i in range(total_requests)
}
for future in concurrent.futures.as_completed(futures):
progress.update(task, advance=1)
progress.stop()
# Process and display the results
successful_calls = sum(1 for status in results.values() if status == 200)
failed_calls = total_requests - successful_calls
print("\n[bold magenta]API Load Test Results:[/bold magenta]\n")
print(f"[bold green]Successful calls: {successful_calls}")
print(f"[bold red]Failed calls: {failed_calls}\n")
# Create a table to display detailed results
table = Table(show_header=True, header_style="bold blue")
table.add_column("Thread Number", justify="right", style="dim")
table.add_column("Status", style="bold")
for thread_number, status in results.items():
if status == 200:
table.add_row(f"{thread_number}", f"[green]Success[/green]")
else:
table.add_row(f"{thread_number}", f"[red]Failed ({status})[/red]")
print(table)