USER
user: mira esto es una estructura de una función para openwebui "from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import base64
import re
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def detect_file_type(file_content: bytes) -> str:
if file_content.startswith(b"%PDF"):
return "PDF"
elif file_content.startswith(b"\xD0\xCF\x11\xE0\xA1\xB1\x1A\xE1"):
return "DOC"
elif file_content.startswith(b"PK\x03\x04"):
return "DOCX"
elif all(
0x20 <= byte <= 0x7E or byte in (0x09, 0x0A, 0x0D)
for byte in file_content[:1024]
):
return "TXT"
elif b"," in file_content[:1024] and b"\n" in file_content[:1024]:
return "CSV"
else:
return "UNKNOWN"
def detect_image_type(image_data: str) -> str:
if image_data.startswith("data:image/jpeg"):
return "JPEG"
elif image_data.startswith("data:image/png"):
return "PNG"
elif image_data.startswith("data:image/gif"):
return "GIF"
else:
return "UNKNOWN"
def format_prompt(messages: Messages) -> List[Dict[str, Any]]:
formatted_messages = []
for m in messages:
role = m.get("role", "")
content = m.get("content", "")
logger.debug(f"Processing message: role={role}, content type={type(content)}")
if isinstance(content, list):
text_parts = []
for item in content:
if isinstance(item, str):
text_parts.append(item)
elif isinstance(item, dict):
if item.get("type") == "image_url":
image_url = item.get("image_url", {}).get("url", "")
if image_url:
text_parts.append(f"[IMAGE: URL={image_url}]")
elif item.get("type") == "image":
image_data = item.get("image", "")
if image_data:
image_type = detect_image_type(image_data)
# Extraer la parte de base64 de la cadena de datos
base64_data = re.sub(
r"^data:image/\w+;base64,", "", image_data
)
# Limitar el tamaño de la imagen a 10MB
max_size = 10 * 1024 * 1024 # 10MB en bytes
if len(base64.b64decode(base64_data)) > max_size:
text_parts.append(
f"[IMAGE: Type={image_type}, Size=Too Large (Max 10MB)]"
)
else:
text_parts.append(
f"[IMAGE: Type={image_type}]\n{image_data}"
)
elif item.get("type") == "file":
file_content = base64.b64decode(item.get("file_base64", ""))
file_type = detect_file_type(file_content)
if file_type == "PDF":
text_parts.append("[FILE: PDF]")
elif file_type in ["DOC", "DOCX"]:
text_parts.append("[FILE: WORD]")
elif file_type == "CSV":
text_parts.append("[FILE: CSV]")
elif file_type == "TXT":
try:
text_content = file_content.decode("utf-8")
text_parts.append(f"[FILE: TXT]\n{text_content}")
except UnicodeDecodeError:
text_parts.append("[FILE: TXT (unable to decode)]")
else:
text_parts.append("[FILE: UNKNOWN]")
formatted_content = " ".join(text_parts)
elif isinstance(content, str):
if content.startswith("data:image"):
image_type = detect_image_type(content)
# Extraer la parte de base64 de la cadena de datos
base64_data = re.sub(r"^data:image/\w+;base64,", "", content)
# Limitar el tamaño de la imagen a 10MB
max_size = 10 * 1024 * 1024 # 10MB en bytes
if len(base64.b64decode(base64_data)) > max_size:
formatted_content = (
f"[IMAGE: Type={image_type}, Size=Too Large (Max 10MB)]"
)
else:
formatted_content = f"[IMAGE: Type={image_type}]\n{content}"
else:
formatted_content = content
else:
logger.warning(f"Tipo de contenido no esperado: {type(content)}")
formatted_content = str(content)
formatted_messages.append({"role": role, "content": formatted_content})
return formatted_messages
class UnlimitedAI(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://api.voids.top/v1/chat/completions"
models_url = "https://api.voids.top/v1/models"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "gpt-4o-mini-free"
models = [
"gpt-4o-mini-free",
"gpt-4o-mini",
"gpt-4o-free",
"gpt-4-turbo-2024-04-09",
"gpt-4o-2024-08-06",
"grok-2",
"grok-2-mini",
"claude-3-opus-20240229",
"claude-3-opus-20240229-gcp",
"claude-3-sonnet-20240229",
"claude-3-5-sonnet-20240620",
"claude-3-haiku-20240307",
"claude-2.1",
"gemini-1.5-flash-exp-0827",
"gemini-1.5-pro-exp-0827",
]
model_aliases = {}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
headers = {
"Content-Type": "application/json",
}
formatted_messages = format_prompt(messages)
data = {
"model": model,
"messages": formatted_messages,
"stream": True, # Solicitar respuesta en streaming
}
logger.debug(f"Sending request to {cls.url} with data: {data}")
async with aiohttp.ClientSession(headers=headers) as session:
try:
async with session.post(cls.url, json=data, proxy=proxy) as response:
logger.debug(f"Received response with status: {response.status}")
if response.status != 200:
error_text = await response.text()
logger.error(f"Error response: {error_text}")
yield f"Error: {response.status}, {error_text}"
return
async for line in response.content:
if line:
try:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
json_data = json.loads(line[6:])
if "choices" in json_data and json_data["choices"]:
content = json_data["choices"][0]["delta"].get(
"content", ""
)
if content:
if raw:
yield json.dumps(json_data)
else:
yield content
except json.JSONDecodeError:
logger.warning(f"Failed to decode JSON: {line}")
except Exception as e:
logger.error(f"Error processing line: {str(e)}")
except aiohttp.ClientError as e:
logger.error(f"Error en la solicitud a UnlimitedAI: {str(e)}")
yield f"Error: No se pudo completar la solicitud a UnlimitedAI. {str(e)}"
except Exception as e:
logger.error(f"Error inesperado: {str(e)}", exc_info=True)
yield f"Error inesperado: {str(e)}"
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = UnlimitedAI()
def pipes(self):
return [
{
"id": f"unlimitedai/{model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
logger.debug(f"Pipe called with: model={model}, raw={raw}")
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple y una imagen
test_body = {
"model": "unlimitedai/gpt_4o_mini_free",
"messages": [
{"role": "system", "content": "Eres un asistente útil, en español."},
{
"role": "user",
"content": [
"What can you tell me about this image?",
{
"type": "image",
"image": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAACklEQVR4nGMAAQAABQABDQottAAAAABJRU5ErkJggg==",
},
],
},
],
"raw": False,
}
print("\nProbando con el modelo 'gpt-4o-mini-free':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())"
assistant: Entiendo. Esta es una implementación detallada de un proveedor de IA llamado UnlimitedAI para OpenWebUI. Aquí hay un resumen de las partes principales:
1. Se definen clases y funciones auxiliares para manejar diferentes tipos de contenido (texto, imágenes, archivos) y formatear los mensajes.
2. La clase `UnlimitedAI` es el núcleo del proveedor, que implementa la lógica para hacer solicitudes a la API de UnlimitedAI.
3. La clase `Pipe` actúa como una interfaz entre OpenWebUI y el proveedor UnlimitedAI. Define los modelos disponibles y maneja las solicitudes de generación de texto.
4. El código incluye manejo de errores y logging para facilitar la depuración.
5. Soporta streaming de respuestas, lo que permite recibir y procesar la salida del modelo de manera incremental.
6. Incluye soporte para varios modelos de IA, incluyendo variantes de GPT-4, Claude, y Gemini.
7. Puede manejar entradas multimodales, incluyendo texto e imágenes.
8. Al final, hay una sección de prueba para verificar el funcionamiento del código.
Este código parece estar bien estructurado y preparado para integrarse con OpenWebUI, proporcionando acceso a varios modelos de IA a través de la API de UnlimitedAI.
user: Puedes crearme una igual con estos datos "from __future__ import annotations
import json
import uuid
from aiohttp import ClientSession, ClientTimeout, ClientResponseError
from ..typing import AsyncResult, Messages
from .base_provider import AsyncGeneratorProvider, ProviderModelMixin
from .helper import format_prompt
from ..image import ImageResponse
class AmigoChat(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://amigochat.io/chat/"
chat_api_endpoint = "https://api.amigochat.io/v1/chat/completions"
image_api_endpoint = "https://api.amigochat.io/v1/images/generations"
working = True
supports_gpt_4 = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = 'gpt-4o-mini'
chat_models = [
'gpt-4o',
default_model,
'o1-preview',
'o1-mini',
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo',
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo',
'claude-3-sonnet-20240229',
'gemini-1.5-pro',
]
image_models = [
'flux-pro/v1.1',
'flux-realism',
'flux-pro',
'dalle-e-3',
]
models = [*chat_models, *image_models]
model_aliases = {
"o1": "o1-preview",
"llama-3.1-405b": "meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo",
"llama-3.2-90b": "meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo",
"claude-3.5-sonnet": "claude-3-sonnet-20240229",
"gemini-pro": "gemini-1.5-pro",
"flux-pro": "flux-pro/v1.1",
"dalle-3": "dalle-e-3",
}
persona_ids = {
'gpt-4o': "gpt",
'gpt-4o-mini': "amigo",
'o1-preview': "openai-o-one",
'o1-mini': "openai-o-one-mini",
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo': "llama-three-point-one",
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo': "llama-3-2",
'claude-3-sonnet-20240229': "claude",
'gemini-1.5-pro': "gemini-1-5-pro",
'flux-pro/v1.1': "flux-1-1-pro",
'flux-realism': "flux-realism",
'flux-pro': "flux-pro",
'dalle-e-3': "dalle-three",
}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_chat_model if model in cls.chat_models else cls.default_image_model
@classmethod
def get_personaId(cls, model: str) -> str:
return cls.persona_ids[model]
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
stream: bool = False,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
device_uuid = str(uuid.uuid4())
max_retries = 3
retry_count = 0
while retry_count < max_retries:
try:
headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": "Bearer",
"cache-control": "no-cache",
"content-type": "application/json",
"origin": cls.url,
"pragma": "no-cache",
"priority": "u=1, i",
"referer": f"{cls.url}/",
"sec-ch-ua": '"Chromium";v="129", "Not=A?Brand";v="8"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Linux"',
"user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
"x-device-language": "en-US",
"x-device-platform": "web",
"x-device-uuid": device_uuid,
"x-device-version": "1.0.32"
}
async with ClientSession(headers=headers) as session:
if model in cls.chat_models:
# Chat completion
data = {
"messages": [{"role": m["role"], "content": m["content"]} for m in messages],
"model": model,
"personaId": cls.get_personaId(model),
"frequency_penalty": 0,
"max_tokens": 4000,
"presence_penalty": 0,
"stream": stream,
"temperature": 0.5,
"top_p": 0.95
}
timeout = ClientTimeout(total=300) # 5 minutes timeout
async with session.post(cls.chat_api_endpoint, json=data, proxy=proxy, timeout=timeout) as response:
if response.status not in (200, 201):
error_text = await response.text()
raise Exception(f"Error {response.status}: {error_text}")
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
if line == 'data: [DONE]':
break
try:
chunk = json.loads(line[6:]) # Remove 'data: ' prefix
if 'choices' in chunk and len(chunk['choices']) > 0:
choice = chunk['choices'][0]
if 'delta' in choice:
content = choice['delta'].get('content')
elif 'text' in choice:
content = choice['text']
else:
content = None
if content:
yield content
except json.JSONDecodeError:
pass
else:
# Image generation
prompt = messages[0]['content']
data = {
"prompt": prompt,
"model": model,
"personaId": cls.get_personaId(model)
}
async with session.post(cls.image_api_endpoint, json=data, proxy=proxy) as response:
response.raise_for_status()
response_data = await response.json()
if "data" in response_data:
image_urls = []
for item in response_data["data"]:
if "url" in item:
image_url = item["url"]
image_urls.append(image_url)
if image_urls:
yield ImageResponse(image_urls, prompt)
else:
yield None
break
except (ClientResponseError, Exception) as e:
retry_count += 1
if retry_count >= max_retries:
raise e
device_uuid = str(uuid.uuid4())
" recuerda que tiene que tener la misma estructura para que se pueda instalar en openwebui
assistant: Entendido. Aquí tienes una versión adaptada del código para AmigoChat que sigue la misma estructura que el ejemplo anterior de UnlimitedAI, para que sea compatible con OpenWebUI:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import uuid
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_prompt(messages: Messages) -> List[Dict[str, Any]]:
return [{"role": m["role"], "content": m["content"]} for m in messages]
class AmigoChat(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://amigochat.io/chat/"
chat_api_endpoint = "https://api.amigochat.io/v1/chat/completions"
image_api_endpoint = "https://api.amigochat.io/v1/images/generations"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = 'gpt-4o-mini'
chat_models = [
'gpt-4o',
default_model,
'o1-preview',
'o1-mini',
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo',
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo',
'claude-3-sonnet-20240229',
'gemini-1.5-pro',
]
image_models = [
'flux-pro/v1.1',
'flux-realism',
'flux-pro',
'dalle-e-3',
]
models = [*chat_models, *image_models]
model_aliases = {
"o1": "o1-preview",
"llama-3.1-405b": "meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo",
"llama-3.2-90b": "meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo",
"claude-3.5-sonnet": "claude-3-sonnet-20240229",
"gemini-pro": "gemini-1.5-pro",
"flux-pro": "flux-pro/v1.1",
"dalle-3": "dalle-e-3",
}
persona_ids = {
'gpt-4o': "gpt",
'gpt-4o-mini': "amigo",
'o1-preview': "openai-o-one",
'o1-mini': "openai-o-one-mini",
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo': "llama-three-point-one",
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo': "llama-3-2",
'claude-3-sonnet-20240229': "claude",
'gemini-1.5-pro': "gemini-1-5-pro",
'flux-pro/v1.1': "flux-1-1-pro",
'flux-realism': "flux-realism",
'flux-pro': "flux-pro",
'dalle-e-3': "dalle-three",
}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
def get_personaId(cls, model: str) -> str:
return cls.persona_ids[model]
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
device_uuid = str(uuid.uuid4())
max_retries = 3
retry_count = 0
while retry_count < max_retries:
try:
headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": "Bearer",
"cache-control": "no-cache",
"content-type": "application/json",
"origin": cls.url,
"pragma": "no-cache",
"priority": "u=1, i",
"referer": f"{cls.url}/",
"sec-ch-ua": '"Chromium";v="129", "Not=A?Brand";v="8"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Linux"',
"user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
"x-device-language": "en-US",
"x-device-platform": "web",
"x-device-uuid": device_uuid,
"x-device-version": "1.0.32"
}
async with aiohttp.ClientSession(headers=headers) as session:
if model in cls.chat_models:
data = {
"messages": format_prompt(messages),
"model": model,
"personaId": cls.get_personaId(model),
"frequency_penalty": 0,
"max_tokens": 4000,
"presence_penalty": 0,
"stream": True,
"temperature": 0.5,
"top_p": 0.95
}
async with session.post(cls.chat_api_endpoint, json=data, proxy=proxy) as response:
if response.status not in (200, 201):
error_text = await response.text()
yield f"Error: {response.status}, {error_text}"
return
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
if line == 'data: [DONE]':
break
try:
chunk = json.loads(line[6:])
if 'choices' in chunk and len(chunk['choices']) > 0:
choice = chunk['choices'][0]
if 'delta' in choice:
content = choice['delta'].get('content')
elif 'text' in choice:
content = choice['text']
else:
content = None
if content:
if raw:
yield json.dumps(chunk)
else:
yield content
except json.JSONDecodeError:
logger.warning(f"Failed to decode JSON: {line}")
else:
prompt = messages[0]['content']
data = {
"prompt": prompt,
"model": model,
"personaId": cls.get_personaId(model)
}
async with session.post(cls.image_api_endpoint, json=data, proxy=proxy) as response:
response.raise_for_status()
response_data = await response.json()
if "data" in response_data:
image_urls = [item["url"] for item in response_data["data"] if "url" in item]
if image_urls:
yield json.dumps({"image_urls": image_urls, "prompt": prompt})
else:
yield "No image URLs found in the response."
break
except Exception as e:
retry_count += 1
if retry_count >= max_retries:
logger.error(f"Error in AmigoChat request: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
device_uuid = str(uuid.uuid4())
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = AmigoChat()
def pipes(self):
return [
{
"id": f"amigochat/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
logger.debug(f"Pipe called with: model={model}, raw={raw}")
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "amigochat/gpt_4o_mini",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
"raw": False,
}
print("\nProbando con el modelo 'gpt-4o-mini':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Este código sigue la misma estructura que el ejemplo de UnlimitedAI y debería ser compatible con OpenWebUI. Incluye la clase `AmigoChat` que implementa la lógica específica del proveedor, y la clase `Pipe` que actúa como interfaz para OpenWebUI. También se han incluido las funciones de logging y manejo de errores, así como una sección de prueba al final del archivo.
user: perfecto ! ahora creame otra aplicacion con estos datos perfecto ahora créame uno con este 'import os
import json
import random
import hashlib
import requests
from ...typing import sha256, Dict, get_type_hints
url = 'https://deepai.org '
model = ['gpt-3.5-turbo']
supports_stream = True
needs_auth = False
def _create_completion(model: str, messages: list, stream: bool, **kwargs):
def md5(text: str) -> str:
return hashlib.md5(text.encode()).hexdigest()[::-1]
def get_api_key(user_agent: str) -> str:
part1 = str(random.randint(0, 10**11))
part2 = md5(user_agent + md5(user_agent + md5(user_agent + part1 + "x")))
return f"tryit-{part1}-{part2}"
user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
headers = {
"api-key": get_api_key(user_agent),
"user-agent": user_agent
}
files = {
"chat_style": (None, "chat"),
"chatHistory": (None, json.dumps(messages))
}
r = requests.post("https://api.deepai.org/chat_response", headers=headers, files=files, stream=True)
for chunk in r.iter_content(chunk_size=None):
r.raise_for_status()
yield chunk.decode()
params = f'g4f.Providers.{os.path.basename(file )[:-3]} supports: ' +
'(%s)' % ', '.join(
[f"{name}: {get_type_hints(_create_completion)[name].name }" for name in _create_completion.code .co_varnames[:_create_completion.code .co_argcount]])
assistant: Entendido. Aquí tienes una implementación adaptada para el proveedor DeepAI, siguiendo la misma estructura que los ejemplos anteriores para que sea compatible con OpenWebUI:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import random
import hashlib
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def md5(text: str) -> str:
return hashlib.md5(text.encode()).hexdigest()[::-1]
def get_api_key(user_agent: str) -> str:
part1 = str(random.randint(0, 10**11))
part2 = md5(user_agent + md5(user_agent + md5(user_agent + part1 + "x")))
return f"tryit-{part1}-{part2}"
class DeepAI(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://deepai.org"
api_endpoint = "https://api.deepai.org/chat_response"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = 'gpt-3.5-turbo'
models = ['gpt-3.5-turbo']
model_aliases = {}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs
) -> AsyncResult:
user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
headers = {
"api-key": get_api_key(user_agent),
"user-agent": user_agent
}
data = {
"chat_style": "chat",
"chatHistory": json.dumps(messages)
}
async with aiohttp.ClientSession() as session:
try:
async with session.post(cls.api_endpoint, headers=headers, data=data, proxy=proxy) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error response: {error_text}")
yield f"Error: {response.status}, {error_text}"
return
async for chunk in response.content:
if chunk:
try:
decoded_chunk = chunk.decode()
if raw:
yield json.dumps({"content": decoded_chunk})
else:
yield decoded_chunk
except Exception as e:
logger.error(f"Error processing chunk: {str(e)}")
except aiohttp.ClientError as e:
logger.error(f"Error en la solicitud a DeepAI: {str(e)}")
yield f"Error: No se pudo completar la solicitud a DeepAI. {str(e)}"
except Exception as e:
logger.error(f"Error inesperado: {str(e)}", exc_info=True)
yield f"Error inesperado: {str(e)}"
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = DeepAI()
def pipes(self):
return [
{
"id": f"deepai/{model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
logger.debug(f"Pipe called with: model={model}, raw={raw}")
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "deepai/gpt_3_5_turbo",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
"raw": False,
}
print("\nProbando con el modelo 'gpt-3.5-turbo':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Esta implementación sigue la misma estructura que los ejemplos anteriores y debería ser compatible con OpenWebUI. Incluye:
1. La clase `DeepAI` que implementa la lógica específica del proveedor.
2. Funciones auxiliares como `md5` y `get_api_key` para generar la clave API necesaria.
3. La clase `Pipe` que actúa como interfaz para OpenWebUI.
4. Manejo de errores y logging.
5. Una sección de prueba al final del archivo.
Este código utiliza la API de DeepAI para generar respuestas de chat, utilizando el modelo gpt-3.5-turbo. La implementación soporta streaming de respuestas y maneja tanto el formato raw como el formato de texto plano para la salida.
user: perfecto! ahora vamos hacer uno más complicado, esto es de donde nos vamos a basar "from __future__ import annotations import os import time import random import string import threading import asyncio import base64 import aiohttp import queue from typing import Union, AsyncIterator, Iterator from ..providers.base_provider import AsyncGeneratorProvider from ..image import ImageResponse, to_image, to_data_uri from ..typing import Messages, ImageType from ..providers.types import BaseProvider, ProviderType, FinishReason from ..providers.conversation import BaseConversation from ..image import ImageResponse as ImageProviderResponse from ..errors import NoImageResponseError from .stubs import ChatCompletion, ChatCompletionChunk, Image, ImagesResponse from .image_models import ImageModels from .types import IterResponse, ImageProvider from .types import Client as BaseClient from .service import get_model_and_provider, get_last_provider from .helper import find_stop, filter_json, filter_none from ..models import ModelUtils from ..Provider import IterListProvider # Helper function to convert an async generator to a synchronous iterator def to_sync_iter(async_gen: AsyncIterator) -> Iterator: q = queue.Queue() loop = asyncio.new_event_loop() done = object() def _run(): asyncio.set_event_loop(loop) async def iterate(): try: async for item in async_gen: q.put(item) finally: q.put(done) loop.run_until_complete(iterate()) loop.close() threading.Thread(target=_run).start() while True: item = q.get() if item is done: break yield item # Helper function to convert a synchronous iterator to an async iterator async def to_async_iterator(iterator): for item in iterator: yield item # Synchronous iter_response function def iter_response( response: Union[Iterator[str], AsyncIterator[str]], stream: bool, response_format: dict = None, max_tokens: int = None, stop: list = None ) -> Iterator[Union[ChatCompletion, ChatCompletionChunk]]: content = "" finish_reason = None completion_id = ''.join(random.choices(string.ascii_letters + string.digits, k=28)) idx = 0 if hasattr(response, '__aiter__'): # It's an async iterator, wrap it into a sync iterator response = to_sync_iter(response) for chunk in response: if isinstance(chunk, FinishReason): finish_reason = chunk.reason break elif isinstance(chunk, BaseConversation): yield chunk continue content += str(chunk) if max_tokens is not None and idx + 1 >= max_tokens: finish_reason = "length" first, content, chunk = find_stop(stop, content, chunk if stream else None) if first != -1: finish_reason = "stop" if stream: yield ChatCompletionChunk(chunk, None, completion_id, int(time.time())) if finish_reason is not None: break idx += 1 finish_reason = "stop" if finish_reason is None else finish_reason if stream: yield ChatCompletionChunk(None, finish_reason, completion_id, int(time.time())) else: if response_format is not None and "type" in response_format: if response_format["type"] == "json_object": content = filter_json(content) yield ChatCompletion(content, finish_reason, completion_id, int(time.time())) # Synchronous iter_append_model_and_provider function def iter_append_model_and_provider(response: Iterator) -> Iterator: last_provider = None for chunk in response: last_provider = get_last_provider(True) if last_provider is None else last_provider chunk.model = last_provider.get("model") chunk.provider = last_provider.get("name") yield chunk class Client(BaseClient): def __init__( self, provider: ProviderType = None, image_provider: ImageProvider = None, **kwargs ) -> None: super().__init__(**kwargs) self.chat: Chat = Chat(self, provider) self._images: Images = Images(self, image_provider) @property def images(self) -> Images: return self._images async def async_images(self) -> Images: return self._images class Completions: def __init__(self, client: Client, provider: ProviderType = None): self.client: Client = client self.provider: ProviderType = provider def create( self, messages: Messages, model: str, provider: ProviderType = None, stream: bool = False, proxy: str = None, response_format: dict = None, max_tokens: int = None, stop: Union[list[str], str] = None, api_key: str = None, ignored: list[str] = None, ignore_working: bool = False, ignore_stream: bool = False, **kwargs ) -> Union[ChatCompletion, Iterator[ChatCompletionChunk]]: model, provider = get_model_and_provider( model, self.provider if provider is None else provider, stream, ignored, ignore_working, ignore_stream, ) stop = [stop] if isinstance(stop, str) else stop if asyncio.iscoroutinefunction(provider.create_completion): # Run the asynchronous function in an event loop response = asyncio.run(provider.create_completion( model, messages, stream=stream, **filter_none( proxy=self.client.get_proxy() if proxy is None else proxy, max_tokens=max_tokens, stop=stop, api_key=self.client.api_key if api_key is None else api_key ), **kwargs )) else: response = provider.create_completion( model, messages, stream=stream, **filter_none( proxy=self.client.get_proxy() if proxy is None else proxy, max_tokens=max_tokens, stop=stop, api_key=self.client.api_key if api_key is None else api_key ), **kwargs ) if stream: if hasattr(response, '__aiter__'): # It's an async generator, wrap it into a sync iterator response = to_sync_iter(response) # Now 'response' is an iterator response = iter_response(response, stream, response_format, max_tokens, stop) response = iter_append_model_and_provider(response) return response else: if hasattr(response, '__aiter__'): # If response is an async generator, collect it into a list response = list(to_sync_iter(response)) response = iter_response(response, stream, response_format, max_tokens, stop) response = iter_append_model_and_provider(response) return next(response) async def async_create( self, messages: Messages, model: str, provider: ProviderType = None, stream: bool = False, proxy: str = None, response_format: dict = None, max_tokens: int = None, stop: Union[list[str], str] = None, api_key: str = None, ignored: list[str] = None, ignore_working: bool = False, ignore_stream: bool = False, **kwargs ) -> Union[ChatCompletion, AsyncIterator[ChatCompletionChunk]]: model, provider = get_model_and_provider( model, self.provider if provider is None else provider, stream, ignored, ignore_working, ignore_stream, ) stop = [stop] if isinstance(stop, str) else stop if asyncio.iscoroutinefunction(provider.create_completion): response = await provider.create_completion( model, messages, stream=stream, **filter_none( proxy=self.client.get_proxy() if proxy is None else proxy, max_tokens=max_tokens, stop=stop, api_key=self.client.api_key if api_key is None else api_key ), **kwargs ) else: response = provider.create_completion( model, messages, stream=stream, **filter_none( proxy=self.client.get_proxy() if proxy is None else proxy, max_tokens=max_tokens, stop=stop, api_key=self.client.api_key if api_key is None else api_key ), **kwargs ) # Removed 'await' here since 'async_iter_response' returns an async generator response = async_iter_response(response, stream, response_format, max_tokens, stop) response = async_iter_append_model_and_provider(response) if stream: return response else: async for result in response: return result class Chat: completions: Completions def __init__(self, client: Client, provider: ProviderType = None): self.completions = Completions(client, provider) # Asynchronous versions of the helper functions async def async_iter_response( response: Union[AsyncIterator[str], Iterator[str]], stream: bool, response_format: dict = None, max_tokens: int = None, stop: list = None ) -> AsyncIterator[Union[ChatCompletion, ChatCompletionChunk]]: content = "" finish_reason = None completion_id = ''.join(random.choices(string.ascii_letters + string.digits, k=28)) idx = 0 if not hasattr(response, '__aiter__'): response = to_async_iterator(response) async for chunk in response: if isinstance(chunk, FinishReason): finish_reason = chunk.reason break elif isinstance(chunk, BaseConversation): yield chunk continue content += str(chunk) if max_tokens is not None and idx + 1 >= max_tokens: finish_reason = "length" first, content, chunk = find_stop(stop, content, chunk if stream else None) if first != -1: finish_reason = "stop" if stream: yield ChatCompletionChunk(chunk, None, completion_id, int(time.time())) if finish_reason is not None: break idx += 1 finish_reason = "stop" if finish_reason is None else finish_reason if stream: yield ChatCompletionChunk(None, finish_reason, completion_id, int(time.time())) else: if response_format is not None and "type" in response_format: if response_format["type"] == "json_object": content = filter_json(content) yield ChatCompletion(content, finish_reason, completion_id, int(time.time())) async def async_iter_append_model_and_provider(response: AsyncIterator) -> AsyncIterator: last_provider = None if not hasattr(response, '__aiter__'): response = to_async_iterator(response) async for chunk in response: last_provider = get_last_provider(True) if last_provider is None else last_provider chunk.model = last_provider.get("model") chunk.provider = last_provider.get("name") yield chunk async def iter_image_response(response: AsyncIterator) -> Union[ImagesResponse, None]: response_list = [] async for chunk in response: if isinstance(chunk, ImageProviderResponse): response_list.extend(chunk.get_list()) elif isinstance(chunk, str): response_list.append(chunk) if response_list: return ImagesResponse([Image(image) for image in response_list]) return None async def create_image(client: Client, provider: ProviderType, prompt: str, model: str = "", **kwargs) -> AsyncIterator: if isinstance(provider, type) and provider.__name__ == "You": kwargs["chat_mode"] = "create" else: prompt = f"create an image with: {prompt}" if asyncio.iscoroutinefunction(provider.create_completion): response = await provider.create_completion( model, [{"role": "user", "content": prompt}], stream=True, proxy=client.get_proxy(), **kwargs ) else: response = provider.create_completion( model, [{"role": "user", "content": prompt}], stream=True, proxy=client.get_proxy(), **kwargs ) # Wrap synchronous iterator into async iterator if necessary if not hasattr(response, '__aiter__'): response = to_async_iterator(response) return response class Image: def __init__(self, url: str = None, b64_json: str = None): self.url = url self.b64_json = b64_json def __repr__(self): return f"Image(url={self.url}, b64_json={'<base64 data>' if self.b64_json else None})" class ImagesResponse: def __init__(self, data: list[Image]): self.data = data def __repr__(self): return f"ImagesResponse(data={self.data})" class Images: def __init__(self, client: 'Client', provider: 'ImageProvider' = None): self.client: 'Client' = client self.provider: 'ImageProvider' = provider self.models: ImageModels = ImageModels(client) def generate(self, prompt: str, model: str = None, response_format: str = "url", **kwargs) -> ImagesResponse: """ Synchronous generate method that runs the async_generate method in an event loop. """ return asyncio.run(self.async_generate(prompt, model, response_format=response_format, **kwargs)) async def async_generate(self, prompt: str, model: str = None, response_format: str = "url", **kwargs) -> ImagesResponse: provider = self.models.get(model, self.provider) if provider is None: raise ValueError(f"Unknown model: {model}") if isinstance(provider, IterListProvider): if provider.providers: provider = provider.providers[0] else: raise ValueError(f"IterListProvider for model {model} has no providers") if isinstance(provider, type) and issubclass(provider, AsyncGeneratorProvider): messages = [{"role": "user", "content": prompt}] async for response in provider.create_async_generator(model, messages, **kwargs): if isinstance(response, ImageResponse): return await self._process_image_response(response, response_format) elif isinstance(response, str): image_response = ImageResponse([response], prompt) return await self._process_image_response(image_response, response_format) elif hasattr(provider, 'create'): if asyncio.iscoroutinefunction(provider.create): response = await provider.create(prompt) else: response = provider.create(prompt) if isinstance(response, ImageResponse): return await self._process_image_response(response, response_format) elif isinstance(response, str): image_response = ImageResponse([response], prompt) return await self._process_image_response(image_response, response_format) else: raise ValueError(f"Provider {provider} does not support image generation") raise NoImageResponseError(f"Unexpected response type: {type(response)}") async def _process_image_response(self, response: ImageResponse, response_format: str) -> ImagesResponse: processed_images = [] for image_data in response.get_list(): if image_data.startswith('http://') or image_data.startswith('https://'): if response_format == "url": processed_images.append(Image(url=image_data)) elif response_format == "b64_json": # Fetch the image data and convert it to base64 image_content = await self._fetch_image(image_data) b64_json = base64.b64encode(image_content).decode('utf-8') processed_images.append(Image(b64_json=b64_json)) else: # Assume image_data is base64 data or binary if response_format == "url": if image_data.startswith('data:image'): # Remove the data URL scheme and get the base64 data header, base64_data = image_data.split(',', 1) else: base64_data = image_data # Decode the base64 data image_data_bytes = base64.b64decode(base64_data) # Convert bytes to an image image = to_image(image_data_bytes) file_name = self._save_image(image) processed_images.append(Image(url=file_name)) elif response_format == "b64_json": if isinstance(image_data, bytes): b64_json = base64.b64encode(image_data).decode('utf-8') else: b64_json = image_data # If already base64-encoded string processed_images.append(Image(b64_json=b64_json)) return ImagesResponse(processed_images) async def _fetch_image(self, url: str) -> bytes: # Asynchronously fetch image data from the URL async with aiohttp.ClientSession() as session: async with session.get(url) as resp: if resp.status == 200: return await resp.read() else: raise Exception(f"Failed to fetch image from {url}, status code {resp.status}") def _save_image(self, image: 'PILImage') -> str: os.makedirs('generated_images', exist_ok=True) file_name = f"generated_images/image_{int(time.time())}_{random.randint(0, 10000)}.png" image.save(file_name) return file_name async def create_variation(self, image: Union[str, bytes], model: str = None, response_format: str = "url", **kwargs): # Existing implementation, adjust if you want to support b64_json here as well pass" from .NexraBing import NexraBing from .NexraBlackbox import NexraBlackbox from .NexraChatGPT import NexraChatGPT from .NexraChatGPT4o import NexraChatGPT4o from .NexraChatGptV2 import NexraChatGptV2 from .NexraChatGptWeb import NexraChatGptWeb from .NexraDallE import NexraDallE from .NexraDallE2 import NexraDallE2 from .NexraEmi import NexraEmi from .NexraFluxPro import NexraFluxPro from .NexraGeminiPro import NexraGeminiPro from .NexraMidjourney import NexraMidjourney from .NexraProdiaAI import NexraProdiaAI from .NexraQwen import NexraQwen from .NexraSD15 import NexraSD15 from .NexraSDLora import NexraSDLora from .NexraSDTurbo import NexraSDTurbo" from __future__ import annotations
import json
import requests
from ...typing import CreateResult, Messages
from ..base_provider import ProviderModelMixin, AbstractProvider
from ..helper import format_prompt
class NexraChatGPT4o(AbstractProvider, ProviderModelMixin):
label = "Nexra ChatGPT4o"
url = "https://nexra.aryahcr.cc/documentation/chatgpt/en"
api_endpoint = "https://nexra.aryahcr.cc/api/chat/complements"
working = True
supports_stream = True
default_model = "gpt-4o"
models = [default_model]
@classmethod
def get_model(cls, model: str) -> str:
return cls.default_model
@classmethod
def create_completion(
cls,
model: str,
messages: Messages,
stream: bool,
proxy: str = None,
markdown: bool = False,
**kwargs
) -> CreateResult:
model = cls.get_model(model)
headers = {
'Content-Type': 'application/json'
}
data = {
"messages": [
{
"role": "user",
"content": format_prompt(messages)
}
],
"stream": stream,
"markdown": markdown,
"model": model
}
response = requests.post(cls.api_endpoint, headers=headers, json=data, stream=stream)
if stream:
return cls.process_streaming_response(response)
else:
return cls.process_non_streaming_response(response)
@classmethod
def process_non_streaming_response(cls, response):
if response.status_code == 200:
try:
content = response.text.lstrip('')
data = json.loads(content)
return data.get('message', '')
except json.JSONDecodeError:
return "Error: Unable to decode JSON response"
else:
return f"Error: {response.status_code}"
@classmethod
def process_streaming_response(cls, response):
full_message = ""
for line in response.iter_lines(decode_unicode=True):
if line:
try:
line = line.lstrip('')
data = json.loads(line)
if data.get('finish'):
break
message = data.get('message', '')
if message and message != full_message:
yield message[len(full_message):]
full_message = message
except json.JSONDecodeError:
pass
assistant: Entendido. Aquí tienes una implementación adaptada para el proveedor NexraChatGPT4o, siguiendo la misma estructura que los ejemplos anteriores para que sea compatible con OpenWebUI:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class NexraChatGPT4o(AsyncGeneratorProvider, ProviderModelMixin):
label = "Nexra ChatGPT4o"
url = "https://nexra.aryahcr.cc/documentation/chatgpt/en"
api_endpoint = "https://nexra.aryahcr.cc/api/chat/complements"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "gpt-4o"
models = [default_model]
model_aliases = {}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
markdown: bool = False,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
headers = {
'Content-Type': 'application/json'
}
data = {
"messages": [
{
"role": "user",
"content": format_prompt(messages)
}
],
"stream": True,
"markdown": markdown,
"model": model
}
async with aiohttp.ClientSession() as session:
try:
async with session.post(cls.api_endpoint, headers=headers, json=data, proxy=proxy) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Error response: {error_text}")
yield f"Error: {response.status}, {error_text}"
return
full_message = ""
async for line in response.content:
if line:
try:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
line = line[6:] # Remove 'data: ' prefix
data = json.loads(line)
if data.get('finish'):
break
message = data.get('message', '')
if message and message != full_message:
new_content = message[len(full_message):]
if raw:
yield json.dumps({"content": new_content})
else:
yield new_content
full_message = message
except json.JSONDecodeError:
logger.warning(f"Failed to decode JSON: {line}")
except Exception as e:
logger.error(f"Error processing line: {str(e)}")
except aiohttp.ClientError as e:
logger.error(f"Error en la solicitud a NexraChatGPT4o: {str(e)}")
yield f"Error: No se pudo completar la solicitud a NexraChatGPT4o. {str(e)}"
except Exception as e:
logger.error(f"Error inesperado: {str(e)}", exc_info=True)
yield f"Error inesperado: {str(e)}"
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = NexraChatGPT4o()
def pipes(self):
return [
{
"id": f"nexrachatgpt4o/{model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
markdown = body.get("markdown", False)
logger.debug(f"Pipe called with: model={model}, raw={raw}, markdown={markdown}")
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, markdown=markdown
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "nexrachatgpt4o/gpt_4o",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
"raw": False,
"markdown": False,
}
print("\nProbando con el modelo 'gpt-4o':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Esta implementación sigue la misma estructura que los ejemplos anteriores y debería ser compatible con OpenWebUI. Incluye:
1. La clase `NexraChatGPT4o` que implementa la lógica específica del proveedor.
2. La función `format_prompt` para formatear los mensajes.
3. La clase `Pipe` que actúa como interfaz para OpenWebUI.
4. Manejo de errores y logging.
5. Soporte para streaming de respuestas.
6. Una sección de prueba al final del archivo.
Este código utiliza la API de NexraChatGPT4o para generar respuestas de chat, utilizando el modelo gpt-4o. La implementación soporta streaming de respuestas y maneja tanto el formato raw como el formato de texto plano para la salida. También incluye la opción de generar respuestas en formato Markdown.
user: tenemos este error "Error: 403, <!DOCTYPE html> <!--[if lt IE 7]> <html class="no-js ie6 oldie" lang="en-US"> <![endif]--> <!--[if IE 7]> <html class="no-js ie7 oldie" lang="en-US"> <![endif]--> <!--[if IE 8]> <html class="no-js ie8 oldie" lang="en-US"> <![endif]--> <!--[if gt IE 8]><!--> <html class="no-js" lang="en-US"> <!--<![endif]--> <head> <title>Attention Required! | Cloudflare</title> <meta charset="UTF-8" /> <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" /> <meta http-equiv="X-UA-Compatible" content="IE=Edge" /> <meta name="robots" content="noindex, nofollow" /> <meta name="viewport" content="width=device-width,initial-scale=1" /> <link rel="stylesheet" id="cf_styles-css" href="/cdn-cgi/styles/cf.errors.css" /> <!--[if lt IE 9]><link rel="stylesheet" id='cf_styles-ie-css' href="/cdn-cgi/styles/cf.errors.ie.css" /><![endif]--> <style>body{margin:0;padding:0}</style> <!--[if gte IE 10]><!--> <script> if (!navigator.cookieEnabled) { window.addEventListener('DOMContentLoaded', function () { var cookieEl = document.getElementById('cookie-alert'); cookieEl.style.display = 'block'; }) } </script> <!--<![endif]--> </head> <body> <div id="cf-wrapper"> <div class="cf-alert cf-alert-error cf-cookie-error" id="cookie-alert" data-translate="enable_cookies">Please enable cookies.</div> <div id="cf-error-details" class="cf-error-details-wrapper"> <div class="cf-wrapper cf-header cf-error-overview"> <h1 data-translate="block_headline">Sorry, you have been blocked</h1> <h2 class="cf-subheadline"><span data-translate="unable_to_access">You are unable to access</span> aryahcr.cc</h2> </div><!-- /.header --> <div class="cf-section cf-highlight"> <div class="cf-wrapper"> <div class="cf-screenshot-container cf-screenshot-full"> <span class="cf-no-screenshot error"></span> </div> </div> </div><!-- /.captcha-container --> <div class="cf-section cf-wrapper"> <div class="cf-columns two"> <div class="cf-column"> <h2 data-translate="blocked_why_headline">Why have I been blocked?</h2> <p data-translate="blocked_why_detail">This website is using a security service to protect itself from online attacks. The action you just performed triggered the security solution. There are several actions that could trigger this block including submitting a certain word or phrase, a SQL command or malformed data.</p> </div> <div class="cf-column"> <h2 data-translate="blocked_resolve_headline">What can I do to resolve this?</h2> <p data-translate="blocked_resolve_detail">You can email the site owner to let them know you were blocked. Please include what you were doing when this page came up and the Cloudflare Ray ID found at the bottom of this page.</p> </div> </div> </div><!-- /.section --> <div class="cf-error-footer cf-wrapper w-240 lg:w-full py-10 sm:py-4 sm:px-8 mx-auto text-center sm:text-left border-solid border-0 border-t border-gray-300"> <p class="text-13"> <span class="cf-footer-item sm:block sm:mb-1">Cloudflare Ray ID: <strong class="font-semibold">8d91bbb98a62cfda</strong></span> <span class="cf-footer-separator sm:hidden">•</span> <span id="cf-footer-item-ip" class="cf-footer-item hidden sm:block sm:mb-1"> Your IP: <button type="button" id="cf-footer-ip-reveal" class="cf-footer-ip-reveal-btn">Click to reveal</button> <span class="hidden" id="cf-footer-ip">85.56.128.100</span> <span class="cf-footer-separator sm:hidden">•</span> </span> <span class="cf-footer-item sm:block sm:mb-1"><span>Performance & security by</span> <a rel="noopener noreferrer" href="https://www.cloudflare.com/5xx-error-landing" id="brand_link" target="_blank">Cloudflare</a></span> </p> <script>(function(){function d(){var b=a.getElementById("cf-footer-item-ip"),c=a.getElementById("cf-footer-ip-reveal");b&&"classList"in b&&(b.classList.remove("hidden"),c.addEventListener("click",function(){c.classList.add("hidden");a.getElementById("cf-footer-ip").classList.remove("hidden")}))}var a=document;document.addEventListener&&a.addEventListener("DOMContentLoaded",d)})();</script> </div><!-- /.error-footer --> </div><!-- /#cf-error-details --> </div><!-- /#cf-wrapper --> <script> window._cf_translation = {}; </script> </body> </html> " pero en el otro programa no lo hace quieres que te muestre las carpetas del otro programa para ver mejor si puedes ver como lo hacen para hacerlo igua?
assistant: Entiendo. El error 403 que estás recibiendo sugiere que Cloudflare está bloqueando las solicitudes, probablemente porque detecta que están viniendo de un script en lugar de un navegador web normal.
Para resolver esto, necesitaremos imitar más de cerca el comportamiento de un navegador web. Esto generalmente implica:
1. Usar headers más completos, similares a los que envía un navegador.
2. Posiblemente manejar cookies y seguir redirecciones.
3. Tal vez incluso resolver desafíos de JavaScript de Cloudflare.
Sí, por favor, muéstrame las carpetas del otro programa que funciona correctamente. Eso nos ayudará a entender cómo están manejando estas solicitudes y qué diferencias pueden existir. Con esa información, podremos ajustar nuestra implementación para que sea más similar y posiblemente evitar el bloqueo de Cloudflare.
user: esto es la carpeta principal " Directorio: F:\gpt4free-0.3.3.1 Mode LastWriteTime Length Name ---- ------------- ------ ---- d----- 18/10/2024 10:48 .github d----- 18/10/2024 10:48 docker d----- 18/10/2024 10:48 docs d----- 18/10/2024 10:48 etc d----- 18/10/2024 10:50 g4f d----- 19/10/2024 21:52 generated_images d----- 18/10/2024 10:48 har_and_cookies d----- 18/10/2024 10:48 models d----- 18/10/2024 10:48 projects -a---- 17/10/2024 17:56 65 .gitattributes -a---- 17/10/2024 17:56 815 .gitignore -a---- 17/10/2024 17:56 349 .gitpod.yml -a---- 17/10/2024 17:56 5220 CODE_OF_CONDUCT.md -a---- 17/10/2024 17:56 502 CONTRIBUTING.md -a---- 17/10/2024 17:56 308 docker-compose.yml -a---- 17/10/2024 17:56 3903 LEGAL_NOTICE.md -a---- 17/10/2024 17:56 35148 LICENSE -a---- 17/10/2024 17:56 189 MANIFEST.in -a---- 17/10/2024 17:56 39636 README.md -a---- 17/10/2024 17:56 80 requirements-min.txt -a---- 17/10/2024 17:56 261 requirements.txt -a---- 17/10/2024 17:56 306 SECURITY.md -a---- 17/10/2024 17:56 3690 setup.py" los modelos se encuentran aquí "Directorio: F:\gpt4free-0.3.3.1\g4f Mode LastWriteTime Length Name ---- ------------- ------ ---- d----- 18/10/2024 10:50 api d----- 18/10/2024 10:50 client d----- 18/10/2024 10:50 gui d----- 18/10/2024 10:48 local d----- 18/10/2024 10:50 locals d----- 18/10/2024 10:50 Provider d----- 18/10/2024 10:50 providers d----- 18/10/2024 10:50 requests d----- 18/10/2024 10:50 __pycache__ -a---- 17/10/2024 17:56 2575 cli.py -a---- 17/10/2024 17:56 6403 cookies.py -a---- 17/10/2024 17:56 169 debug.py -a---- 17/10/2024 17:56 768 errors.py -a---- 17/10/2024 17:56 9308 image.py -a---- 17/10/2024 17:56 22038 models.py -a---- 17/10/2024 17:56 2938 stubs.py -a---- 17/10/2024 17:56 923 typing.py -a---- 17/10/2024 17:56 3855 version.py -a---- 17/10/2024 17:56 9951 webdriver.py -a---- 17/10/2024 17:56 6848 __init__.py " aqui donde estan los proveidores que usamos "Directorio: F:\gpt4free-0.3.3.1\g4f\Provider Mode LastWriteTime Length Name ---- ------------- ------ ---- d----- 18/10/2024 10:50 bing d----- 18/10/2024 10:50 deprecated d----- 18/10/2024 10:48 gigachat_crt d----- 18/10/2024 10:50 needs_auth d----- 18/10/2024 10:50 nexra d----- 18/10/2024 10:48 npm d----- 18/10/2024 10:50 openai d----- 18/10/2024 10:50 selenium d----- 18/10/2024 10:50 you d----- 18/10/2024 10:50 __pycache__ -a---- 17/10/2024 17:56 2552 AI365VIP.py -a---- 17/10/2024 17:56 2382 Ai4Chat.py -a---- 17/10/2024 17:56 2743 AIChatFree.py -a---- 17/10/2024 17:56 2335 AiChatOnline.py -a---- 17/10/2024 17:56 4532 AiChats.py -a---- 17/10/2024 17:56 2584 AiMathGPT.py -a---- 17/10/2024 17:56 8634 Airforce.py -a---- 17/10/2024 17:56 4891 AIUncensored.py -a---- 17/10/2024 17:56 2786 Allyfy.py -a---- 17/10/2024 17:56 7854 AmigoChat.py -a---- 17/10/2024 17:56 1702 Aura.py -a---- 17/10/2024 17:56 194 base_provider.py -a---- 17/10/2024 17:56 21427 Bing.py -a---- 17/10/2024 17:56 1956 BingCreateImages.py -a---- 17/10/2024 17:56 13237 Blackbox.py -a---- 17/10/2024 17:56 2713 ChatGot.py -a---- 17/10/2024 17:56 8062 ChatGpt.py -a---- 17/10/2024 17:56 3192 Chatgpt4o.py -a---- 17/10/2024 17:56 3000 Chatgpt4Online.py -a---- 17/10/2024 17:56 2943 ChatGptEs.py -a---- 17/10/2024 17:56 4138 ChatgptFree.py -a---- 17/10/2024 17:56 2878 ChatHub.py -a---- 17/10/2024 17:56 2611 ChatifyAI.py -a---- 17/10/2024 17:56 6913 Cloudflare.py -a---- 17/10/2024 17:56 3115 DarkAI.py -a---- 17/10/2024 17:56 3956 DDG.py -a---- 17/10/2024 17:56 2006 DeepInfra.py -a---- 17/10/2024 17:56 5785 DeepInfraChat.py -a---- 17/10/2024 17:56 3032 DeepInfraImage.py -a---- 17/10/2024 17:56 2623 Editee.py -a---- 17/10/2024 17:56 3819 FlowGpt.py -a---- 17/10/2024 17:56 2809 Free2GPT.py -a---- 17/10/2024 17:56 3857 FreeChatgpt.py -a---- 17/10/2024 17:56 2298 FreeGpt.py -a---- 17/10/2024 17:56 4112 FreeNetfly.py -a---- 17/10/2024 17:56 4345 GeminiPro.py -a---- 17/10/2024 17:56 3922 GigaChat.py -a---- 17/10/2024 17:56 2292 GPROChat.py -a---- 17/10/2024 17:56 111 helper.py -a---- 17/10/2024 17:56 5538 HuggingChat.py -a---- 17/10/2024 17:56 4071 HuggingFace.py -a---- 17/10/2024 17:56 3041 Koala.py -a---- 17/10/2024 17:56 10740 Liaobots.py -a---- 17/10/2024 17:56 1203 Local.py -a---- 17/10/2024 17:56 2972 MagickPen.py -a---- 17/10/2024 17:56 10473 MetaAI.py -a---- 17/10/2024 17:56 669 MetaAIAccount.py -a---- 17/10/2024 17:56 2244 Nexra.py -a---- 17/10/2024 17:56 1164 Ollama.py -a---- 17/10/2024 17:56 3987 PerplexityLabs.py -a---- 17/10/2024 17:56 2418 Pi.py -a---- 17/10/2024 17:56 1800 Pizzagpt.py -a---- 17/10/2024 17:56 6474 Prodia.py -a---- 17/10/2024 17:56 6011 Reka.py -a---- 17/10/2024 17:56 3449 Replicate.py -a---- 17/10/2024 17:56 5856 ReplicateHome.py -a---- 17/10/2024 17:56 6160 RubiksAI.py -a---- 17/10/2024 17:56 2699 TeachAnything.py -a---- 17/10/2024 17:56 2753 Upstage.py -a---- 17/10/2024 17:56 2077 WhiteRabbitNeo.py -a---- 17/10/2024 17:56 8494 You.py -a---- 17/10/2024 17:56 3287 __init__.py" nosotros estamos usando este " Directorio: F:\gpt4free-0.3.3.1\g4f\Provider\nexra Mode LastWriteTime Length Name ---- ------------- ------ ---- d----- 18/10/2024 10:50 __pycache__ -a---- 17/10/2024 17:56 3535 NexraBing.py -a---- 17/10/2024 17:56 3501 NexraBlackbox.py -a---- 17/10/2024 17:56 3105 NexraChatGPT.py -a---- 17/10/2024 17:56 2590 NexraChatGPT4o.py -a---- 17/10/2024 17:56 3344 NexraChatGptV2.py -a---- 17/10/2024 17:56 2163 NexraChatGptWeb.py -a---- 17/10/2024 17:56 2235 NexraDallE.py -a---- 17/10/2024 17:56 2456 NexraDallE2.py -a---- 17/10/2024 17:56 2249 NexraDalleMini.py -a---- 17/10/2024 17:56 2225 NexraEmi.py -a---- 17/10/2024 17:56 2456 NexraFluxPro.py -a---- 17/10/2024 17:56 2302 NexraGeminiPro.py -a---- 17/10/2024 17:56 3321 NexraLLaMA31.py -a---- 17/10/2024 17:56 2254 NexraMidjourney.py -a---- 17/10/2024 17:56 6483 NexraProdiaAI.py -a---- 17/10/2024 17:56 3128 NexraQwen.py -a---- 17/10/2024 17:56 2393 NexraSD15.py -a---- 17/10/2024 17:56 2575 NexraSD21.py -a---- 17/10/2024 17:56 2491 NexraSDLora.py -a---- 17/10/2024 17:56 2494 NexraSDTurbo.py -a---- 17/10/2024 17:56 1016 __init__.py" y en concreto hemos empezado con este "NexraChatGPT4o.py" ves algo que te pueda ayudar, si hay alguna carpeta que quieras ver me lo dices
assistant: Gracias por proporcionar esta información detallada sobre la estructura del proyecto. Basándome en lo que has compartido, parece que el proyecto está utilizando una estructura más compleja y posiblemente algunas técnicas adicionales para manejar las solicitudes a los proveedores de API.
Vamos a examinar el archivo `NexraChatGPT4o.py` en la carpeta `nexra`. ¿Podrías mostrarme el contenido de ese archivo? Es probable que contenga detalles importantes sobre cómo se están manejando las solicitudes para evitar el bloqueo de Cloudflare.
Además, sería útil ver el contenido de algunos archivos auxiliares:
1. El contenido de `F:\gpt4free-0.3.3.1\g4f\Provider\base_provider.py`
2. El contenido de `F:\gpt4free-0.3.3.1\g4f\Provider\helper.py`
3. El contenido de `F:\gpt4free-0.3.3.1\g4f\requests\__init__.py` (si existe)
Estos archivos probablemente contienen funciones y clases base que se utilizan en todos los proveedores, incluido el manejo de solicitudes HTTP y posiblemente algunas técnicas para evadir la detección de bots.
Una vez que tenga esta información, podré sugerir modificaciones más precisas a nuestra implementación para que sea más similar a la que está funcionando correctamente.
user: claude manten la misma estructura como en el ejemplo ya que sigue el mismo error en el otro programa el modelo contesta perfectamente
assistant: Entiendo. Vamos a mantener la estructura lo más similar posible al ejemplo original que me mostraste. Aquí tienes una versión que mantiene la estructura original pero adaptada para funcionar con OpenWebUI:
```python
from __future__ import annotations
import re
import json
import aiohttp
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
headers = {
'authority': 'chatgptfree.ai',
'accept': '*/*',
'accept-language': 'en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3',
'origin': 'https://chatgptfree.ai',
'referer': 'https://chatgptfree.ai/chat/',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
}
async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
if not cls._nonce:
async with session.get(f"{cls.url}/", proxy=proxy) as response:
response_text = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response_text)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response_text)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, proxy=proxy) as response:
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield content
except json.JSONDecodeError:
continue
elif line:
buffer += line
if buffer:
try:
json_response = json.loads(buffer)
if 'data' in json_response:
yield json_response['data']
except json.JSONDecodeError:
print(f"Failed to decode final JSON. Buffer content: {buffer}")
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in [ChatgptFree.default_model, *ChatgptFree.model_aliases.keys()]
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
yield chunk
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "chatgptfree/gpt_4o_mini",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
}
print("\nProbando con el modelo 'gpt-4o-mini':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Este código mantiene la estructura original lo más posible, incluyendo:
1. Las clases `AsyncGeneratorProvider` y `ProviderModelMixin`.
2. La estructura de la clase `ChatgptFree` con sus métodos y atributos.
3. La lógica para obtener el nonce y post_id.
4. El manejo de cookies y proxy.
Cambios mínimos:
1. Se usa `aiohttp` en lugar de `StreamSession` para mantener la compatibilidad con asyncio.
2. Se ha ajustado ligeramente la estructura de `Pipe` para trabajar con OpenWebUI.
Para usar este código en OpenWebUI:
1. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
2. Asegúrate de que el archivo esté en la ubicación correcta donde OpenWebUI pueda acceder a él.
3. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Si aún persiste el error, podría ser útil:
1. Verificar que todas las dependencias necesarias estén instaladas (especialmente `aiohttp`).
2. Revisar los logs de OpenWebUI para ver si hay algún error específico.
3. Asegurarse de que OpenWebUI esté configurado correctamente para usar este proveedor.
Si el problema continúa, por favor proporciona cualquier mensaje de error adicional que puedas ver en los logs o la consola. Esto nos ayudará a identificar y resolver cualquier problema restante.
user: nada algo estas haciendo mal por que sigue igual ¡Uh oh! Hubo un problema al conectarse a GPT-4O-MINI-2024-07-18. network error recuerda que solo hay un modelo no dos mira "¡Uh oh! Hubo un problema al conectarse a este ultimo/GPT-4O-MINI-2024-07-18. network error" mantente fiel from __future__ import annotations
import re
import json
import asyncio
from ..requests import StreamSession, raise_for_status
from ..typing import Messages, AsyncGenerator
from .base_provider import AsyncGeneratorProvider, ProviderModelMixin
from .helper import format_prompt
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
headers = {
'authority': 'chatgptfree.ai',
'accept': '*/*',
'accept-language': 'en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3',
'origin': 'https://chatgptfree.ai',
'referer': 'https://chatgptfree.ai/chat/',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
}
async with StreamSession(
headers=headers,
cookies=cookies,
impersonate="chrome",
proxies={"all": proxy},
timeout=timeout
) as session:
if not cls._nonce:
async with session.get(f"{cls.url}/") as response:
await raise_for_status(response)
response = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, cookies=cookies) as response:
await raise_for_status(response)
buffer = ""
async for line in response.iter_lines():
line = line.decode('utf-8').strip()
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield content
except json.JSONDecodeError:
continue
elif line:
buffer += line
if buffer:
try:
json_response = json.loads(buffer)
if 'data' in json_response:
yield json_response['data']
except json.JSONDecodeError:
print(f"Failed to decode final JSON. Buffer content: {buffer}")
assistant: Entiendo. Me disculpo por la confusión anterior. Vamos a mantener la estructura lo más fiel posible al código original que proporcionaste. Aquí tienes una versión adaptada que debería funcionar con OpenWebUI mientras mantiene la estructura original:
```python
from __future__ import annotations
import re
import json
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Simulando las importaciones que no tenemos
class StreamSession:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
async def get(self, url):
# Simular una respuesta
class Response:
async def text(self):
return '<div data-post-id="12345" data-nonce="abcdef123456"></div>'
return Response()
async def post(self, url, data=None, cookies=None):
# Simular una respuesta
class Response:
async def iter_lines(self):
yield b'data: {"choices":[{"delta":{"content":"Hello"}}]}'
yield b'choices":[{"delta":{"content":" World"}}]}'
yield b'data: [DONE]'
return Response()
async def raise_for_status(response):
# Simulación de la función
pass
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
headers = {
'authority': 'chatgptfree.ai',
'accept': '*/*',
'accept-language': 'en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3',
'origin': 'https://chatgptfree.ai',
'referer': 'https://chatgptfree.ai/chat/',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
}
async with StreamSession(
headers=headers,
cookies=cookies,
impersonate="chrome",
proxies={"all": proxy},
timeout=timeout
) as session:
if not cls._nonce:
async with session.get(f"{cls.url}/") as response:
await raise_for_status(response)
response = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, cookies=cookies) as response:
await raise_for_status(response)
buffer = ""
async for line in response.iter_lines():
line = line.decode('utf-8').strip()
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield content
except json.JSONDecodeError:
continue
elif line:
buffer += line
if buffer:
try:
json_response = json.loads(buffer)
if 'data' in json_response:
yield json_response['data']
except json.JSONDecodeError:
print(f"Failed to decode final JSON. Buffer content: {buffer}")
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{ChatgptFree.default_model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{ChatgptFree.default_model.upper()}",
}
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
yield chunk
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
```
Cambios principales:
1. Se mantiene la estructura original de `ChatgptFree` casi intacta.
2. Se ha simplificado `pipes()` para que solo devuelva el modelo predeterminado, ya que mencionaste que solo hay un modelo.
3. Se han simulado las clases y funciones que no tenemos disponibles (`StreamSession` y `raise_for_status`).
4. La clase `Pipe` se ha ajustado mínimamente para trabajar con OpenWebUI.
Para usar este código en OpenWebUI:
1. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
2. Asegúrate de que el archivo esté en la ubicación correcta donde OpenWebUI pueda acceder a él.
3. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Si aún persiste el error de conexión, podría ser debido a:
1. La falta de implementación real de `StreamSession` y `raise_for_status`. Necesitarás reemplazar estas simulaciones con las implementaciones reales de tu proyecto.
2. Problemas de red o configuración en OpenWebUI.
Para depurar más a fondo, podrías añadir más logs en el código, especialmente en la función `create_async_generator`, para ver exactamente dónde está fallando la conexión.
user: oye el error sigue "¡Uh oh! Hubo un problema al conectarse a GPT-4O-MINI-2024-07-18. network error
assistant: Entiendo que el error persiste. Vamos a intentar una estrategia diferente. En lugar de simular las clases y funciones que no tenemos, vamos a usar `aiohttp`, que es una biblioteca estándar para hacer solicitudes HTTP asíncronas. Esto debería resolver los problemas de red que estamos experimentando.
Aquí tienes una versión actualizada del código que utiliza `aiohttp` y mantiene la estructura lo más fiel posible al original:
```python
from __future__ import annotations
import re
import json
import aiohttp
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
headers = {
'authority': 'chatgptfree.ai',
'accept': '*/*',
'accept-language': 'en,fr-FR;q=0.9,fr;q=0.8,es-ES;q=0.7,es;q=0.6,en-US;q=0.5,am;q=0.4,de;q=0.3',
'origin': 'https://chatgptfree.ai',
'referer': 'https://chatgptfree.ai/chat/',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36',
}
async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
if not cls._nonce:
async with session.get(f"{cls.url}/", proxy=proxy) as response:
response.raise_for_status()
response_text = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response_text)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response_text)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, proxy=proxy) as response:
response.raise_for_status()
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield content
except json.JSONDecodeError:
continue
elif line:
buffer += line
if buffer:
try:
json_response = json.loads(buffer)
if 'data' in json_response:
yield json_response['data']
except json.JSONDecodeError:
print(f"Failed to decode final JSON. Buffer content: {buffer}")
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{ChatgptFree.default_model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{ChatgptFree.default_model.upper()}",
}
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
try:
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
yield chunk
except Exception as e:
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "chatgptfree/gpt_4o_mini_2024_07_18",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
}
print("\nProbando con el modelo 'gpt-4o-mini-2024-07-18':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Cambios principales:
1. Se ha reemplazado `StreamSession` con `aiohttp.ClientSession`.
2. Se ha implementado el manejo de errores en la función `pipe` para capturar y reportar cualquier excepción.
3. Se ha simplificado `pipes()` para que solo devuelva el modelo predeterminado.
4. Se ha añadido una función de prueba al final del archivo.
Para usar este código en OpenWebUI:
1. Asegúrate de tener instalado `aiohttp`. Si no lo tienes, puedes instalarlo con `pip install aiohttp`.
2. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
3. Coloca el archivo en la ubicación correcta donde OpenWebUI pueda acceder a él.
4. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Si aún persiste el error de conexión, podría ser útil:
1. Verificar que no haya problemas de red o firewall que estén bloqueando las conexiones.
2. Asegurarte de que OpenWebUI tiene los permisos necesarios para hacer solicitudes de red.
3. Probar el código de manera independiente (usando la función de prueba al final del archivo) para ver si el problema está en el código o en la integración con OpenWebUI.
Si el problema persiste, por favor proporciona cualquier mensaje de error adicional o logs que puedas obtener de OpenWebUI. Esto nos ayudará a identificar el problema con más precisión.
user: ahora error Error: 403, message='Forbidden', url='https://chatgptfree.ai/ '
assistant: Entiendo. El error 403 Forbidden sugiere que el servidor está rechazando nuestra solicitud, posiblemente debido a algún tipo de protección contra bots o scraping. Vamos a intentar algunas estrategias para superar esta limitación:
1. Añadir más headers para simular un navegador real.
2. Implementar un sistema de reintentos con intervalos aleatorios.
3. Usar un User-Agent aleatorio en cada intento.
Aquí tienes una versión actualizada del código que implementa estas estrategias:
```python
from __future__ import annotations
import re
import json
import aiohttp
import random
import asyncio
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/109.0'
]
headers = {
'authority': 'chatgptfree.ai',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-language': 'en-US,en;q=0.9',
'cache-control': 'max-age=0',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': random.choice(user_agents),
}
async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
for attempt in range(3): # Try up to 3 times
try:
if not cls._nonce:
async with session.get(f"{cls.url}/", proxy=proxy, timeout=timeout) as response:
if response.status == 403:
print(f"Attempt {attempt + 1}: 403 Forbidden. Retrying...")
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
continue
response.raise_for_status()
response_text = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response_text)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response_text)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, proxy=proxy, timeout=timeout) as response:
response.raise_for_status()
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith(':
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield content
except json.JSONDecodeError:
continue
elif line:
buffer += line
if buffer:
try:
json_response = json.loads(buffer)
if 'data' in json_response:
yield json_response['data']
except json.JSONDecodeError:
print(f"Failed to decode final JSON. Buffer content: {buffer}")
break # If we get here, we've successfully completed the request
except aiohttp.ClientError as e:
print(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == 2: # This was our last attempt
yield f"Error: Failed to connect after 3 attempts. Last error: {str(e)}"
else:
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{ChatgptFree.default_model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{ChatgptFree.default_model.upper()}",
}
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
try:
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
yield chunk
except Exception as e:
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "chatgptfree/gpt_4o_mini_2024_07_18",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
}
print("\nProbando con el modelo 'gpt-4o-mini-2024-07-18':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Cambios principales:
1. Se ha añadido una lista de User-Agents y se selecciona uno aleatorio para cada solicitud.
2. Se han ampliado los headers para simular mejor un navegador real.
3. Se ha implementado un sistema de reintentos (hasta 3 intentos) con intervalos aleatorios entre cada intento.
4. Se ha añadido más manejo de errores y logging para ayudar en la depuración.
Para usar este código en OpenWebUI:
1. Asegúrate de tener instalado `aiohttp`. Si no lo tienes, puedes instalarlo con `pip install aiohttp`.
2. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
3. Coloca el archivo en la ubicación correcta donde OpenWebUI pueda acceder a él.
4. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Si aún persiste el error 403, podría ser que el sitio web esté implementando medidas más estrictas contra el scraping. En ese caso, podrías necesitar:
1. Implementar un sistema de rotación de IP si es posible.
2. Considerar el uso de un servicio de proxy o VPN.
3. Verificar si el sitio web tiene una API oficial que puedas usar en su lugar.
Si el problema persiste, por favor proporciona cualquier mensaje de error adicional o logs que puedas obtener. Esto nos ayudará a identificar el problema con más precisión.
user: claude ahora no se puede instalar por este error Cannot parse: 106:47: if line.startswith(': recuerda que las correcciones siempre completas
assistant: Entiendo, disculpa por el error. Tienes razón, siempre debo proporcionar las correcciones completas. Aquí tienes el código corregido y completo:
```python
from __future__ import annotations
import re
import json
import aiohttp
import random
import asyncio
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/109.0'
]
headers = {
'authority': 'chatgptfree.ai',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-language': 'en-US,en;q=0.9',
'cache-control': 'max-age=0',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': random.choice(user_agents),
}
async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
for attempt in range(3): # Try up to 3 times
try:
if not cls._nonce:
async with session.get(f"{cls.url}/", proxy=proxy, timeout=timeout) as response:
if response.status == 403:
print(f"Attempt {attempt + 1}: 403 Forbidden. Retrying...")
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
continue
response.raise_for_status()
response_text = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response_text)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response_text)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, proxy=proxy, timeout=timeout) as response:
response.raise_for_status()
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
yield content
except json.JSONDecodeError:
continue
elif line:
buffer += line
if buffer:
try:
json_response = json.loads(buffer)
if 'data' in json_response:
yield json_response['data']
except json.JSONDecodeError:
print(f"Failed to decode final JSON. Buffer content: {buffer}")
break # If we get here, we've successfully completed the request
except aiohttp.ClientError as e:
print(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == 2: # This was our last attempt
yield f"Error: Failed to connect after 3 attempts. Last error: {str(e)}"
else:
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{ChatgptFree.default_model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{ChatgptFree.default_model.upper()}",
}
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
try:
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
yield chunk
except Exception as e:
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "chatgptfree/gpt_4o_mini_2024_07_18",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
}
print("\nProbando con el modelo 'gpt-4o-mini-2024-07-18':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
En esta versión corregida, he arreglado el error en la línea 106. El problema estaba en la condición `if line.startswith(':` que estaba incompleta. La he corregido a `if line.startswith('data: ')`.
Este código debería funcionar correctamente ahora. Incluye todas las mejoras mencionadas anteriormente:
1. Uso de User-Agents aleatorios.
2. Headers más completos para simular un navegador real.
3. Sistema de reintentos con intervalos aleatorios.
4. Mejor manejo de errores y logging.
Para usar este código en OpenWebUI:
1. Asegúrate de tener instalado `aiohttp`. Si no lo tienes, puedes instalarlo con `pip install aiohttp`.
2. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
3. Coloca el archivo en la ubicación correcta donde OpenWebUI pueda acceder a él.
4. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Si encuentras algún otro problema o necesitas más ajustes, por favor házmelo saber.
user: ahora en el chat no aparece texto si el modelo a contestado no lo se porque no hay texto de respuesta en el chat
assistant: Entiendo el problema. Si no hay texto de respuesta visible en el chat, es posible que el generador no esté produciendo ningún contenido o que el contenido no se esté pasando correctamente a la interfaz de OpenWebUI. Vamos a modificar el código para asegurarnos de que estamos generando y enviando contenido, y añadiremos más logging para ayudar a diagnosticar el problema. Aquí tienes una versión actualizada del código:
```python
from __future__ import annotations
import re
import json
import aiohttp
import random
import asyncio
import logging
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/109.0'
]
headers = {
'authority': 'chatgptfree.ai',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-language': 'en-US,en;q=0.9',
'cache-control': 'max-age=0',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': random.choice(user_agents),
}
async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
for attempt in range(3): # Try up to 3 times
try:
if not cls._nonce:
async with session.get(f"{cls.url}/", proxy=proxy, timeout=timeout) as response:
if response.status == 403:
logger.warning(f"Attempt {attempt + 1}: 403 Forbidden. Retrying...")
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
continue
response.raise_for_status()
response_text = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response_text)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response_text)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, proxy=proxy, timeout=timeout) as response:
response.raise_for_status()
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
logger.debug(f"Received line: {line}")
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
logger.info("Received [DONE] signal")
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
logger.info(f"Yielding content: {content}")
yield content
except json.JSONDecodeError:
logger.warning(f"Failed to decode JSON: {data}")
continue
elif line:
buffer += line
if buffer:
logger.debug(f"Processing final buffer: {buffer}")
try:
json_response = json.loads(buffer)
if 'data' in json_response:
logger.info(f"Yielding final data: {json_response['data']}")
yield json_response['data']
else:
logger.warning(f"No 'data' key in final JSON response: {json_response}")
except json.JSONDecodeError:
logger.error(f"Failed to decode final JSON. Buffer content: {buffer}")
logger.info("Request completed successfully")
break # If we get here, we've successfully completed the request
except aiohttp.ClientError as e:
logger.error(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == 2: # This was our last attempt
error_message = f"Error: Failed to connect after 3 attempts. Last error: {str(e)}"
logger.error(error_message)
yield error_message
else:
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{ChatgptFree.default_model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{ChatgptFree.default_model.upper()}",
}
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
logger.info(f"Starting pipe for model: {model}")
try:
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
logger.debug(f"Pipe yielding chunk: {chunk}")
yield chunk
except Exception as e:
error_message = f"Error: {str(e)}"
logger.error(error_message)
yield error_message
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "chatgptfree/gpt_4o_mini_2024_07_18",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
}
print("\nProbando con el modelo 'gpt-4o-mini-2024-07-18':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Cambios principales:
1. Se ha añadido logging detallado en varios puntos del código para ayudar a diagnosticar el problema.
2. Se ha asegurado que cada vez que se genera contenido, se registra en los logs.
3. Se han añadido más mensajes de error y advertencia para capturar posibles problemas.
Para usar este código:
1. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
2. Asegúrate de que tienes instalado `aiohttp` y `pydantic`.
3. Coloca el archivo en la ubicación correcta donde OpenWebUI pueda acceder a él.
4. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Después de implementar estos cambios, por favor ejecuta el código y observa los logs. Deberías ver mensajes detallados sobre lo que está sucediendo en cada paso del proceso. Si no hay contenido siendo generado, los logs nos ayudarán a identificar en qué punto está fallando el proceso.
Si aún no ves ninguna respuesta en el chat después de estos cambios, por favor proporciona los logs o cualquier mensaje de error que veas. Esto nos ayudará a identificar si el problema está en la generación de contenido, en la comunicación con el servicio de ChatGPT Free, o en cómo OpenWebUI está manejando las respuestas.
user: sigue igual no hay su texto de respuesta en el chat
assistant: Entiendo que el problema persiste. Vamos a intentar un enfoque diferente. Es posible que el formato de la respuesta no sea exactamente lo que OpenWebUI espera. Vamos a modificar el código para que se ajuste más estrechamente al formato que OpenWebUI probablemente está esperando. Aquí tienes una versión actualizada:
```python
from __future__ import annotations
import re
import json
import aiohttp
import random
import asyncio
import logging
from typing import List, Dict, Any, AsyncGenerator
from pydantic import BaseModel
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
Messages = List[Dict[str, Any]]
def format_prompt(messages: Messages) -> str:
return "\n".join([f"{m['role']}: {m['content']}" for m in messages])
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class ChatgptFree(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://chatgptfree.ai"
supports_gpt_4 = True
working = True
_post_id = None
_nonce = None
default_model = 'gpt-4o-mini-2024-07-18'
model_aliases = {
"gpt-4o-mini": "gpt-4o-mini-2024-07-18",
}
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
timeout: int = 120,
cookies: dict = None,
**kwargs
) -> AsyncGenerator[str, None]:
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/109.0'
]
headers = {
'authority': 'chatgptfree.ai',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'accept-language': 'en-US,en;q=0.9',
'cache-control': 'max-age=0',
'sec-ch-ua': '"Chromium";v="118", "Google Chrome";v="118", "Not=A?Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': random.choice(user_agents),
}
async with aiohttp.ClientSession(headers=headers, cookies=cookies) as session:
for attempt in range(3): # Try up to 3 times
try:
if not cls._nonce:
async with session.get(f"{cls.url}/", proxy=proxy, timeout=timeout) as response:
if response.status == 403:
logger.warning(f"Attempt {attempt + 1}: 403 Forbidden. Retrying...")
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
continue
response.raise_for_status()
response_text = await response.text()
result = re.search(r'data-post-id="([0-9]+)"', response_text)
if not result:
raise RuntimeError("No post id found")
cls._post_id = result.group(1)
result = re.search(r'data-nonce="(.*?)"', response_text)
if result:
cls._nonce = result.group(1)
else:
raise RuntimeError("No nonce found")
prompt = format_prompt(messages)
data = {
"_wpnonce": cls._nonce,
"post_id": cls._post_id,
"url": cls.url,
"action": "wpaicg_chat_shortcode_message",
"message": prompt,
"bot_id": "0"
}
async with session.post(f"{cls.url}/wp-admin/admin-ajax.php", data=data, proxy=proxy, timeout=timeout) as response:
response.raise_for_status()
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
logger.debug(f"Received line: {line}")
if line.startswith('data: '):
data = line[6:]
if data == '[DONE]':
logger.info("Received [DONE] signal")
break
try:
json_data = json.loads(data)
content = json_data['choices'][0]['delta'].get('content', '')
if content:
logger.info(f"Yielding content: {content}")
yield json.dumps({"content": content})
except json.JSONDecodeError:
logger.warning(f"Failed to decode JSON: {data}")
continue
elif line:
buffer += line
if buffer:
logger.debug(f"Processing final buffer: {buffer}")
try:
json_response = json.loads(buffer)
if 'data' in json_response:
logger.info(f"Yielding final son_response['data']}")
yield json.dumps({"content": json_response['data']})
else:
logger.warning(f"No 'data' key in final JSON response: {json_response}")
except json.JSONDecodeError:
logger.error(f"Failed to decode final JSON. Buffer content: {buffer}")
logger.info("Request completed successfully")
break # If we get here, we've successfully completed the request
except aiohttp.ClientError as e:
logger.error(f"Attempt {attempt + 1} failed: {str(e)}")
if attempt == 2: # This was our last attempt
error_message = f"Error: Failed to connect after 3 attempts. Last error: {str(e)}"
logger.error(error_message)
yield json.dumps({"content": error_message})
else:
await asyncio.sleep(random.uniform(1, 3)) # Random delay between 1 and 3 seconds
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = ChatgptFree()
def pipes(self):
return [
{
"id": f"chatgptfree/{ChatgptFree.default_model.lower().replace('-', '_')}",
"name": f"{self.valves.NAME_PREFIX}{ChatgptFree.default_model.upper()}",
}
]
async def pipe(self, body: dict) -> AsyncGenerator[str, None]:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
proxy = body.get("proxy")
timeout = body.get("timeout", 120)
cookies = body.get("cookies")
logger.info(f"Starting pipe for model: {model}")
try:
async for chunk in self.provider.create_async_generator(
model,
messages,
proxy=proxy,
timeout=timeout,
cookies=cookies
):
logger.debug(f"Pipe yielding chunk: {chunk}")
yield chunk
except Exception as e:
error_message = f"Error: {str(e)}"
logger.error(error_message)
yield json.dumps({"content": error_message})
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "chatgptfree/gpt_4o_mini_2024_07_18",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
}
print("\nProbando con el modelo 'gpt-4o-mini-2024-07-18':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Cambios principales:
1. Ahora, cada vez que generamos contenido, lo envolvemos en un objeto JSON con una clave "content". Esto es más consistente con lo que OpenWebUI probablemente espera.
2. Hemos modificado la forma en que manejamos los errores para que también se envíen como objetos JSON con una clave "content".
3. Hemos mantenido el logging detallado para ayudar con la depuración.
Para usar este código:
1. Guarda este código en un archivo, por ejemplo `chatgptfree_provider.py`.
2. Asegúrate de que tienes instalado `aiohttp` y `pydantic`.
3. Coloca el archivo en la ubicación correcta donde OpenWebUI pueda acceder a él.
4. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Después de implementar estos cambios, por favor intenta nuevamente en OpenWebUI. Si aún no ves ninguna respuesta en el chat, por favor proporciona la siguiente información:
1. ¿Hay algún mensaje de error visible en la interfaz de OpenWebUI?
2. ¿Puedes ver los logs de OpenWebUI? Si es así, ¿qué mensajes aparecen cuando intentas usar este proveedor?
3. Si es posible, ejecuta el script de manera independiente (usando la función `test()` al final del archivo) y comparte la salida.
Esta información nos ayudará a entender mejor dónde podría estar ocurriendo el problema y cómo resolverlo.
user: ahora cambiando a otro código "from __future__ import annotations from typing import AsyncGenerator, List, Dict, Any, Optional from pydantic import BaseModel import aiohttp import json import logging import asyncio import tiktoken # Configurar logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) # Definiciones de tipos AsyncResult = AsyncGenerator[str, None] Messages = List[Dict[str, Any]] class AsyncGeneratorProvider: pass class ProviderModelMixin: @classmethod def get_model(cls, model: str) -> str: return cls.model_aliases.get(model, model) def format_messages(messages: Messages) -> List[Dict[str, Any]]: MESSAGE_MAX_CHARS = 1000 - 10 formatted = [] for message in messages: role = message["role"] content = message["content"] while content: chunk = content[:MESSAGE_MAX_CHARS] content = content[MESSAGE_MAX_CHARS:] formatted.append({"role": role, "content": chunk}) if content: role = "user" if role == "assistant" else "assistant" formatted.append( {"role": role, "content": "[system: continuing message]"} ) return formatted def num_tokens_from_string(string: str, model: str = "cl100k_base") -> int: encoding = tiktoken.get_encoding(model) num_tokens = len(encoding.encode(string)) return num_tokens class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin): api_url = "https://api.airforce/chat/completions" working = True supports_stream = True supports_system_message = True supports_message_history = True default_model = "claude-3-5-sonnet-20240620" models = [ "claude-3-5-sonnet-20240620", "claude-3-opus-20240229", "chatgpt-4o-latest", "gpt-4o", "gpt-4", "WizardLM-2-8x22B", "llama-3.1-405b-turbo", "llama-3.1-70b-turbo", ] model_aliases = {} @classmethod def get_model(cls, model: str) -> str: if model in cls.models: return model elif model in cls.model_aliases: return cls.model_aliases[model] else: return cls.default_model @classmethod async def create_async_generator( cls, model: str, messages: Messages, raw: bool = False, proxy: str = None, **kwargs, ) -> AsyncResult: model = cls.get_model(model) formatted_messages = format_messages(messages) headers = { "Accept": "application/json", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en-US,en;q=0.9", "Authorization": "Bearer missing api key", "Origin": "https://llmplayground.net", "Referer": "https://llmplayground.net/", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36", "Content-Type": "application/json", } max_attempts = 5 attempt = 0 full_response = "" while attempt < max_attempts: data = { "messages": formatted_messages, "model": model, "max_tokens": kwargs.get("max_tokens", 4000), "temperature": kwargs.get("temperature", 0.7), "top_p": 0.9, "stream": True, } async with aiohttp.ClientSession(headers=headers) as session: async with session.post( cls.api_url, json=data, proxy=proxy ) as response: if response.status != 200: error_text = await response.text() yield f"Error: {response.status}, {error_text}" return last_chunk_time = asyncio.get_event_loop().time() timeout = 30.0 chunk_response = "" async for line in response.content: line = line.decode("utf-8").strip() if line.startswith("data: "): line = line[6:] try: data = json.loads(line) content = data["choices"][0]["delta"].get("content") if content: chunk_response += content if raw: yield json.dumps(data) else: yield content last_chunk_time = asyncio.get_event_loop().time() except json.JSONDecodeError: continue if asyncio.get_event_loop().time() - last_chunk_time > timeout: break full_response += chunk_response formatted_messages.append({"role": "assistant", "content": chunk_response}) formatted_messages.append( {"role": "user", "content": "Continue from where you left off."} ) if chunk_response.strip().endswith((".", "!", "?")): break attempt += 1 logger.info(f"Full response length: {len(full_response)}") logger.info(f"Number of attempts: {attempt + 1}") class Pipe: class Valves(BaseModel): NAME_PREFIX: str = "" def __init__(self): self.type = "manifold" self.valves = self.Valves() self.provider = RocksProvider() def pipes(self): return [ { "id": f"rocks/{model.lower().replace('-', '_').replace('/', '_')}", "name": f"{self.valves.NAME_PREFIX}{model.upper()}", } for model in self.provider.models ] async def pipe(self, body: dict) -> AsyncResult: model = body["model"].split("/")[-1].replace("_", "-") messages = body["messages"] raw = body.get("raw", False) proxy = body.get("proxy") max_tokens = body.get("max_tokens", 4000) logger.debug( f"Pipe called with: model={model}, raw={raw}, max_tokens={max_tokens}" ) try: async for chunk in self.provider.create_async_generator( model, messages, raw=raw, proxy=proxy, max_tokens=max_tokens ): yield chunk except Exception as e: logger.error(f"Error in pipe: {str(e)}", exc_info=True) yield f"Error: {str(e)}" # Instancia de Pipe para ser utilizada por OpenWebUI pipe = Pipe() # Función de prueba if __name__ == "__main__": import asyncio async def test(): # Probar el método pipes models = pipe.pipes() print("Modelos disponibles:", models) # Probar el método pipe con un modelo de chat test_body_chat = { "model": "rocks/claude_3_5_sonnet_20240620", "messages": [ { "role": "user", "content": "Explícame detalladamente qué modelo de IA eres y cuáles son tus capacidades.", }, ], "raw": False, "max_tokens": 8000, } print(f"\nProbando con el modelo de chat '{test_body_chat['model']}':") async for chunk in pipe.pipe(test_body_chat): print(chunk, end="", flush=True) print("\nPrueba de chat completada.") asyncio.run(test()) " resulta que el modelo no termina la contesta cio por culpa de si te fijas alfinal del mensaje te lo dice que pasa! puedes arreglarlo de alguna manera "Entendido. Te proporcionaré el código para el archivo .tpl y un archivo PHP separado para manejar la configuración del número de WhatsApp. Aquí tienes:
1. **footer.tpl** (o donde quieras colocar el botón flotante):
```html
{* Botón flotante de WhatsApp *}
<a href="https://wa.me/{$whatsapp_number}" target="_blank" class="whatsapp-float">
<img src="{$urls.img_dir}whatsapp-icon.png" alt="WhatsApp">
</a>
<style>
.whatsapp-float {
position: fixed;
width: 60px;
height: 60px;
bottom: 20px;
right: 20px;
background-color: #25d366;
color: #FFF;
border-radius: 50%;
text-align: center;
font-size: 30px;
box-shadow: 2px 2px 3px #999;
z-index: 100;
}
.whatsapp-float img {
width: 100%;
height: auto;
}
.whatsapp-float:hover {
background-color: #128C7E;
transition-duration: .3s;
}
</style>
```
2. **WhatsApp.php** (este sería un módulo simple para manejar la configuración):
```php
<?php
if (!defined('_PS_VERSION_')) {
exit;
}
class WhatsApp extends Module
{
public function __construct()
{
$this->name = 'whatsapp';
$this->tab = 'front_office_features';
$this->version = '1.0.0';
$this->author = 'Tu Nombre';
$this->need_instance = 0;
$this->ps_versions_compliancy = array('min' => '1.7', 'max' => _PS_VERSION_);One message exceeds the 1000chars per message limit. Join our discord for more: [https://discord.com/invite/q55gsH8z5F](https://discord.com/invite/q55gsH8z5F)One message exceeds the 1000chars per message limit. Join our discord for more: [https://discord.com/invite/q55gsH8z5F](https://discord.com/invite/q55gsH8z5F)One message exceeds the 1000chars per message limit. Join our discord for more: [https://discord.com/invite/q55gsH8z5F](https://discord.com/invite/q55gsH8z5F)One message exceeds the 1000chars per message limit. Join our discord for more: [https://discord.com/invite/q55gsH8z5F](https://discord.com/invite/q55gsH8z5F)
assistant: Entiendo el problema. El modelo está limitado a 1000 caracteres por mensaje y no está manejando correctamente los mensajes más largos. Vamos a modificar el código para manejar mejor esta limitación. Aquí tienes una versión actualizada del `RocksProvider` que debería resolver este problema:
```python
class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin):
api_url = "https://api.airforce/chat/completions"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "claude-3-5-sonnet-20240620"
models = [
"claude-3-5-sonnet-20240620",
"claude-3-opus-20240229",
"chatgpt-4o-latest",
"gpt-4o",
"gpt-4",
"WizardLM-2-8x22B",
"llama-3.1-405b-turbo",
"llama-3.1-70b-turbo",
]
model_aliases = {}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
model = cls.get_model(model)
formatted_messages = format_messages(messages)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Authorization": "Bearer missing api key",
"Origin": "https://llmplayground.net",
"Referer": "https://llmplayground.net/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36",
"Content-Type": "application/json",
}
max_attempts = 5
attempt = 0
full_response = ""
while attempt < max_attempts:
data = {
"messages": formatted_messages,
"model": model,
"max_tokens": kwargs.get("max_tokens", 4000),
"temperature": kwargs.get("temperature", 0.7),
"top_p": 0.9,
"stream": True,
}
async with aiohttp.ClientSession(headers=headers) as session:
async with session.post(
cls.api_url, json=data, proxy=proxy
) as response:
if response.status != 200:
error_text = await response.text()
yield f"Error: {response.status}, {error_text}"
return
last_chunk_time = asyncio.get_event_loop().time()
timeout = 30.0
chunk_response = ""
async for line in response.content:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
line = line[6:]
try:
data = json.loads(line)
content = data["choices"][0]["delta"].get("content")
if content:
chunk_response += content
if raw:
yield json.dumps(data)
else:
yield content
last_chunk_time = asyncio.get_event_loop().time()
except json.JSONDecodeError:
continue
if asyncio.get_event_loop().time() - last_chunk_time > timeout:
break
full_response += chunk_response
formatted_messages.append({"role": "assistant", "content": chunk_response})
# Check if the response is complete
if chunk_response.strip().endswith((".", "!", "?")) or len(full_response) >= kwargs.get("max_tokens", 4000):
break
# If not complete, continue the conversation
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
attempt += 1
logger.info(f"Full response length: {len(full_response)}")
logger.info(f"Number of attempts: {attempt + 1}")
```
Los cambios principales son:
1. Hemos modificado la condición de salida del bucle principal. Ahora, además de verificar si la respuesta termina con un punto, signo de exclamación o interrogación, también verificamos si la longitud total de la respuesta ha alcanzado el límite de tokens especificado.
2. Si la respuesta no está completa, en lugar de pedirle que continúe desde donde lo dejó, ahora le pedimos explícitamente que continúe su respuesta anterior. Esto debería ayudar a mantener la coherencia en la respuesta.
3. Hemos mantenido el límite de 5 intentos para evitar bucles infinitos.
Esta modificación debería permitir que el modelo complete su respuesta de manera más efectiva, incluso cuando se trata de respuestas largas que superan el límite de 1000 caracteres por mensaje.
Para implementar estos cambios:
1. Reemplaza la clase `RocksProvider` existente con esta nueva versión.
2. Asegúrate de que la función `format_messages` sigue siendo la misma que en tu código original, ya que esta función es crucial para manejar los mensajes largos.
Con estos cambios, el modelo debería ser capaz de completar sus respuestas de manera más efectiva, incluso cuando son largas. Si aún experimentas problemas, por favor házmelo saber y podemos hacer ajustes adicionales.
user: pero dame la correccion completa
assistant: Entiendo, te proporcionaré la corrección completa del código. Aquí tienes la versión actualizada completa del archivo:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import asyncio
import tiktoken
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_messages(messages: Messages) -> List[Dict[str, Any]]:
MESSAGE_MAX_CHARS = 1000 - 10
formatted = []
for message in messages:
role = message["role"]
content = message["content"]
while content:
chunk = content[:MESSAGE_MAX_CHARS]
content = content[MESSAGE_MAX_CHARS:]
formatted.append({"role": role, "content": chunk})
if content:
role = "user" if role == "assistant" else "assistant"
formatted.append(
{"role": role, "content": "[system: continuing message]"}
)
return formatted
def num_tokens_from_string(string: str, model: str = "cl100k_base") -> int:
encoding = tiktoken.get_encoding(model)
num_tokens = len(encoding.encode(string))
return num_tokens
class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin):
api_url = "https://api.airforce/chat/completions"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "claude-3-5-sonnet-20240620"
models = [
"claude-3-5-sonnet-20240620",
"claude-3-opus-20240229",
"chatgpt-4o-latest",
"gpt-4o",
"gpt-4",
"WizardLM-2-8x22B",
"llama-3.1-405b-turbo",
"llama-3.1-70b-turbo",
]
model_aliases = {}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
model = cls.get_model(model)
formatted_messages = format_messages(messages)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Authorization": "Bearer missing api key",
"Origin": "https://llmplayground.net",
"Referer": "https://llmplayground.net/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36",
"Content-Type": "application/json",
}
max_attempts = 5
attempt = 0
full_response = ""
while attempt < max_attempts:
data = {
"messages": formatted_messages,
"model": model,
"max_tokens": kwargs.get("max_tokens", 4000),
"temperature": kwargs.get("temperature", 0.7),
"top_p": 0.9,
"stream": True,
}
async with aiohttp.ClientSession(headers=headers) as session:
async with session.post(
cls.api_url, json=data, proxy=proxy
) as response:
if response.status != 200:
error_text = await response.text()
yield f"Error: {response.status}, {error_text}"
return
last_chunk_time = asyncio.get_event_loop().time()
timeout = 30.0
chunk_response = ""
async for line in response.content:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
line = line[6:]
try:
data = json.loads(line)
content = data["choices"][0]["delta"].get("content")
if content:
chunk_response += content
if raw:
yield json.dumps(data)
else:
yield content
last_chunk_time = asyncio.get_event_loop().time()
except json.JSONDecodeError:
continue
if asyncio.get_event_loop().time() - last_chunk_time > timeout:
break
full_response += chunk_response
formatted_messages.append({"role": "assistant", "content": chunk_response})
# Check if the response is complete
if chunk_response.strip().endswith((".", "!", "?")) or len(full_response) >= kwargs.get("max_tokens", 4000):
break
# If not complete, continue the conversation
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
attempt += 1
logger.info(f"Full response length: {len(full_response)}")
logger.info(f"Number of attempts: {attempt + 1}")
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = RocksProvider()
def pipes(self):
return [
{
"id": f"rocks/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
max_tokens = body.get("max_tokens", 4000)
logger.debug(
f"Pipe called with: model={model}, raw={raw}, max_tokens={max_tokens}"
)
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, max_tokens=max_tokens
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un modelo de chat
test_body_chat = {
"model": "rocks/claude_3_5_sonnet_20240620",
"messages": [
{
"role": "user",
"content": "Explícame detalladamente qué modelo de IA eres y cuáles son tus capacidades.",
},
],
"raw": False,
"max_tokens": 8000,
}
print(f"\nProbando con el modelo de chat '{test_body_chat['model']}':")
async for chunk in pipe.pipe(test_body_chat):
print(chunk, end="", flush=True)
print("\nPrueba de chat completada.")
asyncio.run(test())
```
Los cambios principales se han realizado en la clase `RocksProvider`, específicamente en el método `create_async_generator`. Estos cambios incluyen:
1. Una condición de salida mejorada para el bucle principal que verifica si la respuesta está completa o si ha alcanzado el límite de tokens.
2. Una modificación en cómo se continúa la conversación si la respuesta no está completa.
3. Un mejor manejo de los intentos y la longitud total de la respuesta.
El resto del código se mantiene prácticamente igual, incluyendo las funciones de ayuda, la clase `Pipe`, y la función de prueba.
Esta versión debería manejar mejor las respuestas largas y evitar que se corten abruptamente. Si aún experimentas problemas o necesitas más ajustes, por favor házmelo saber.
user: no es la solucion correcta Error: 500, <!doctype html>
<html lang=en>
<title>500 Internal Server Error</title>
<h1>Internal Server Error</h1>
<p>The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.</p>
<script>(function(){function c(){var b=a.contentDocument||a.contentWindow.document;if(b){var d=b.createElement('script');d.innerHTML="window.__CF$cv$params={r:'8d9e31d7fff6cfe6',t:'MTczMDE1Mjc4OC4wMDAwMDA='};var a=document.createElement('script');a.nonce='';a.src='/cdn-cgi/challenge-platform/scripts/jsd/main.js';document.getElementsByTagName('head')[0].appendChild(a);";b.getElementsByTagName('head')[0].appendChild(d)}}if(document.body){var a=document.createElement('iframe');a.height=1;a.width=1;a.style.position='absolute';a.style.top=0;a.style.left=0;a.style.border='none';a.style.visibility='hidden';document.body.appendChild(a);if('loading'!==document.readyState)c();else if(window.addEventListener)document.addEventListener('DOMContentLoaded',c);else{var e=document.onreadystatechange||function(){};document.onreadystatechange=function(b){e(b);'loading'!==document.readyState&&(document.onreadystatechange=e,c())}}}})();</script>
assistant: Entiendo, parece que el error 500 está siendo causado por el servidor al que estamos intentando acceder. Este tipo de error generalmente indica un problema en el lado del servidor, no en nuestro código. Sin embargo, podemos intentar algunas estrategias para manejar mejor este tipo de situaciones. Aquí tienes una versión actualizada del código que intenta manejar este tipo de errores de una manera más robusta:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import asyncio
import tiktoken
import random
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_messages(messages: Messages) -> List[Dict[str, Any]]:
MESSAGE_MAX_CHARS = 1000 - 10
formatted = []
for message in messages:
role = message["role"]
content = message["content"]
while content:
chunk = content[:MESSAGE_MAX_CHARS]
content = content[MESSAGE_MAX_CHARS:]
formatted.append({"role": role, "content": chunk})
if content:
role = "user" if role == "assistant" else "assistant"
formatted.append(
{"role": role, "content": "[system: continuing message]"}
)
return formatted
def num_tokens_from_string(string: str, model: str = "cl100k_base") -> int:
encoding = tiktoken.get_encoding(model)
num_tokens = len(encoding.encode(string))
return num_tokens
class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin):
api_url = "https://api.airforce/chat/completions"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "claude-3-5-sonnet-20240620"
models = [
"claude-3-5-sonnet-20240620",
"claude-3-opus-20240229",
"chatgpt-4o-latest",
"gpt-4o",
"gpt-4",
"WizardLM-2-8x22B",
"llama-3.1-405b-turbo",
"llama-3.1-70b-turbo",
]
model_aliases = {}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
model = cls.get_model(model)
formatted_messages = format_messages(messages)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Authorization": "Bearer missing api key",
"Origin": "https://llmplayground.net",
"Referer": "https://llmplayground.net/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36",
"Content-Type": "application/json",
}
max_attempts = 5
attempt = 0
full_response = ""
while attempt < max_attempts:
try:
data = {
"messages": formatted_messages,
"model": model,
"max_tokens": kwargs.get("max_tokens", 4000),
"temperature": kwargs.get("temperature", 0.7),
"top_p": 0.9,
"stream": True,
}
async with aiohttp.ClientSession(headers=headers) as session:
async with session.post(
cls.api_url, json=data, proxy=proxy, timeout=30
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Server responded with status {response.status}: {error_text}")
if response.status == 500:
attempt += 1
if attempt < max_attempts:
wait_time = random.uniform(1, 5)
logger.info(f"Retrying in {wait_time:.2f} seconds...")
await asyncio.sleep(wait_time)
continue
yield f"Error: {response.status}, {error_text}"
return
last_chunk_time = asyncio.get_event_loop().time()
timeout = 30.0
chunk_response = ""
async for line in response.content:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
line = line[6:]
try:
data = json.loads(line)
content = data["choices"][0]["delta"].get("content")
if content:
chunk_response += content
if raw:
yield json.dumps(data)
else:
yield content
last_chunk_time = asyncio.get_event_loop().time()
except json.JSONDecodeError:
continue
if asyncio.get_event_loop().time() - last_chunk_time > timeout:
break
full_response += chunk_response
if chunk_response.strip().endswith((".", "!", "?")) or len(full_response) >= kwargs.get("max_tokens", 4000):
break
formatted_messages.append({"role": "assistant", "content": chunk_response})
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
attempt += 1
except aiohttp.ClientError as e:
logger.error(f"Network error occurred: {str(e)}")
attempt += 1
if attempt < max_attempts:
wait_time = random.uniform(1, 5)
logger.info(f"Retrying in {wait_time:.2f} seconds...")
await asyncio.sleep(wait_time)
else:
yield f"Error: Failed to connect after {max_attempts} attempts. Last error: {str(e)}"
return
logger.info(f"Full response length: {len(full_response)}")
logger.info(f"Number of attempts: {attempt}")
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = RocksProvider()
def pipes(self):
return [
{
"id": f"rocks/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
max_tokens = body.get("max_tokens", 4000)
logger.debug(
f"Pipe called with: model={model}, raw={raw}, max_tokens={max_tokens}"
)
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, max_tokens=max_tokens
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un modelo de chat
test_body_chat = {
"model": "rocks/claude_3_5_sonnet_20240620",
"messages": [
{
"role": "user",
"content": "Explícame detalladamente qué modelo de IA eres y cuáles son tus capacidades.",
},
],
"raw": False,
"max_tokens": 8000,
}
print(f"\nProbando con el modelo de chat '{test_body_chat['model']}':")
async for chunk in pipe.pipe(test_body_chat):
print(chunk, end="", flush=True)
print("\nPrueba de chat completada.")
asyncio.run(test())
```
Los cambios principales en esta versión son:
1. Se ha añadido un sistema de reintentos más robusto para manejar errores 500.
2. Se ha implementado un tiempo de espera aleatorio entre reintentos para evitar sobrecargar el servidor.
3. Se ha mejorado el manejo de errores de red y timeouts.
4. Se han añadido más logs para ayudar en la depuración.
Esta versión debería ser más resistente a errores temporales del servidor. Sin embargo, si el error 500 persiste después de varios intentos, es probable que haya un problema más serio en el lado del servidor que está fuera de nuestro control.
Si el problema persiste, podrías considerar:
1. Verificar si la URL del API es correcta y si el servicio está funcionando normalmente.
2. Comprobar si necesitas algún tipo de autenticación adicional para acceder al API.
3. Contactar con los administradores del servicio para informar sobre el error persistente.
Si necesitas más ajustes o tienes más información sobre el error, por favor házmelo saber.
user: sigue igual!! vuelve al que funcionaba al primero que te e dado y vamos a cambiar de estrategia vamos, como el modelo está limitado a 1000 podemos hacer que cuando llegue a 900 pare y volvemos a mandar como un continuar para que siga hasta acabar es decir dividir
assistant: Entiendo. Volvamos al código original y apliquemos la estrategia que sugieres. Vamos a modificar el código para que divida la respuesta en chunks de aproximadamente 900 caracteres y luego continúe con la siguiente parte. Aquí tienes la versión modificada del código original:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import asyncio
import tiktoken
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_messages(messages: Messages) -> List[Dict[str, Any]]:
MESSAGE_MAX_CHARS = 1000 - 10
formatted = []
for message in messages:
role = message["role"]
content = message["content"]
while content:
chunk = content[:MESSAGE_MAX_CHARS]
content = content[MESSAGE_MAX_CHARS:]
formatted.append({"role": role, "content": chunk})
if content:
role = "user" if role == "assistant" else "assistant"
formatted.append(
{"role": role, "content": "[system: continuing message]"}
)
return formatted
def num_tokens_from_string(string: str, model: str = "cl100k_base") -> int:
encoding = tiktoken.get_encoding(model)
num_tokens = len(encoding.encode(string))
return num_tokens
class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin):
api_url = "https://api.airforce/chat/completions"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "claude-3-5-sonnet-20240620"
models = [
"claude-3-5-sonnet-20240620",
"claude-3-opus-20240229",
"chatgpt-4o-latest",
"gpt-4o",
"gpt-4",
"WizardLM-2-8x22B",
"llama-3.1-405b-turbo",
"llama-3.1-70b-turbo",
]
model_aliases = {}
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
model = cls.get_model(model)
formatted_messages = format_messages(messages)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Authorization": "Bearer missing api key",
"Origin": "https://llmplayground.net",
"Referer": "https://llmplayground.net/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36",
"Content-Type": "application/json",
}
max_attempts = 5
attempt = 0
full_response = ""
while attempt < max_attempts:
data = {
"messages": formatted_messages,
"model": model,
"max_tokens": min(900, kwargs.get("max_tokens", 4000)),
"temperature": kwargs.get("temperature", 0.7),
"top_p": 0.9,
"stream": True,
}
async with aiohttp.ClientSession(headers=headers) as session:
async with session.post(
cls.api_url, json=data, proxy=proxy
) as response:
if response.status != 200:
error_text = await response.text()
yield f"Error: {response.status}, {error_text}"
return
last_chunk_time = asyncio.get_event_loop().time()
timeout = 30.0
chunk_response = ""
async for line in response.content:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
line = line[6:]
try:
data = json.loads(line)
content = data["choices"][0]["delta"].get("content")
if content:
chunk_response += content
if raw:
yield json.dumps(data)
else:
yield content
last_chunk_time = asyncio.get_event_loop().time()
except json.JSONDecodeError:
continue
if asyncio.get_event_loop().time() - last_chunk_time > timeout:
break
full_response += chunk_response
formatted_messages.append({"role": "assistant", "content": chunk_response})
# Check if we've reached the desired total length
if len(full_response) >= kwargs.get("max_tokens", 4000):
break
# If not complete, continue the conversation
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
attempt += 1
logger.info(f"Full response length: {len(full_response)}")
logger.info(f"Number of attempts: {attempt + 1}")
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = RocksProvider()
def pipes(self):
return [
{
"id": f"rocks/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
max_tokens = body.get("max_tokens", 4000)
logger.debug(
f"Pipe called with: model={model}, raw={raw}, max_tokens={max_tokens}"
)
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, max_tokens=max_tokens
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un modelo de chat
test_body_chat = {
"model": "rocks/claude_3_5_sonnet_20240620",
"messages": [
{
"role": "user",
"content": "Explícame detalladamente qué modelo de IA eres y cuáles son tus capacidades.",
},
],
"raw": False,
"max_tokens": 8000,
}
print(f"\nProbando con el modelo de chat '{test_body_chat['model']}':")
async for chunk in pipe.pipe(test_body_chat):
print(chunk, end="", flush=True)
print("\nPrueba de chat completada.")
asyncio.run(test())
```
Los cambios principales en esta versión son:
1. En el método `create_async_generator`, hemos modificado el parámetro `max_tokens` en la solicitud para que sea como máximo 900:
```python
"max_tokens": min(900, kwargs.get("max_tokens", 4000)),
```
2. Después de cada respuesta, verificamos si hemos alcanzado el límite total de tokens solicitado:
```python
if len(full_response) >= kwargs.get("max_tokens", 4000):
break
```
3. Si no hemos alcanzado el límite, añadimos un mensaje para continuar la respuesta:
```python
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
```
4. El bucle `while` seguirá ejecutándose hasta que se alcance el límite de tokens o se alcance el número máximo de intentos.
Esta estrategia debería permitir que el modelo genere respuestas más largas divididas en chunks de aproximadamente 900 caracteres cada uno. El modelo continuará su respuesta en cada iteración hasta que se alcance el límite total de tokens o se complete la respuesta.
Si aún experimentas problemas o necesitas más ajustes, por favor házmelo saber.
user: ahora el servidor nos ha bloqueado <!--[if lt IE 7]> <html class="no-js ie6 oldie" lang="en-US"> <![endif]-->
<!--[if IE 7]> <html class="no-js ie7 oldie" lang="en-US"> <![endif]-->
<!--[if IE 8]> <html class="no-js ie8 oldie" lang="en-US"> <![endif]-->
<!--[if gt IE 8]><!--> <html class="no-js" lang="en-US"> <!--<![endif]-->
<head>
<title>api.airforce | 502: Bad gateway</title>
<meta charset="UTF-8" />
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
<meta http-equiv="X-UA-Compatible" content="IE=Edge" />
<meta name="robots" content="noindex, nofollow" />
<meta name="viewport" content="width=device-width,initial-scale=1" />
<link rel="stylesheet" id="cf_styles-css" href="/cdn-cgi/styles/main.css" />
</head>
<body>
<div id="cf-wrapper">
<div id="cf-error-details" class="p-0">
<header class="mx-auto pt-10 lg:pt-6 lg:px-8 w-240 lg:w-full mb-8">
<h1 class="inline-block sm:block sm:mb-2 font-light text-60 lg:text-4xl text-black-dark leading-tight mr-2">
<span class="inline-block">Bad gateway</span>
<span class="code-label">Error code 502</span>
</h1>
<div>
Visit <a href="https://www.cloudflare.com/5xx-error-landing?utm_source=errorcode_502&utm_campaign=api.airforce" target="_blank" rel="noopener noreferrer">cloudflare.com</a> for more information.
</div>
<div class="mt-3">2024-10-28 22:02:39 UTC</div>
</header>
<div class="my-8 bg-gradient-gray">
<div class="w-240 lg:w-full mx-auto">
<div class="clearfix md:px-8">
<div id="cf-browser-status" class=" relative w-1/3 md:w-full py-15 md:p-0 md:py-8 md:text-left md:border-solid md:border-0 md:border-b md:border-gray-400 overflow-hidden float-left md:float-none text-center">
<div class="relative mb-10 md:m-0">
<span class="cf-icon-browser block md:hidden h-20 bg-center bg-no-repeat"></span>
<span class="cf-icon-ok w-12 h-12 absolute left-1/2 md:left-auto md:right-0 md:top-0 -ml-6 -bottom-4"></span> </div>
<span class="md:block w-full truncate">You</span>
<h3 class="md:inline-block mt-3 md:mt-0 text-2xl text-gray-600 font-light leading-1.3">
Browser </h3>
<span class="leading-1.3 text-2xl text-green-success">Working</span>
</div>
<div id="cf-cloudflare-status" class=" relative w-1/3 md:w-full py-15 md:p-0 md:py-8 md:text-left md:border-solid md:border-0 md:border-b md:border-gray-400 overflow-hidden float-left md:float-none text-center">
<div class="relative mb-10 md:m-0">
<a href="https://www.cloudflare.com/5xx-error-landing?utm_source=errorcode_502&utm_campaign=api.airforce" target="_blank" rel="noopener noreferrer">
<span class="cf-icon-cloud block md:hidden h-20 bg-center bg-no-repeat"></span>
<span class="cf-icon-ok w-12 h-12 absolute left-1/2 md:left-auto md:right-0 md:top-0 -ml-6 -bottom-4"></span>
</a>
</div>
<span class="md:block w-full truncate">Paris</span>
<h3 class="md:inline-block mt-3 md:mt-0 text-2xl text-gray-600 font-light leading-1.3">
<a href="https://www.cloudflare.com/5xx-error-landing?utm_source=errorcode_502&utm_campaign=api.airforce" target="_blank" rel="noopener noreferrer">
Cloudflare
</a>
</h3>
<span class="leading-1.3 text-2xl text-green-success">Working</span>
</div>
<div id="cf-host-status" class="cf-error-source relative w-1/3 md:w-full py-15 md:p-0 md:py-8 md:text-left md:border-solid md:border-0 md:border-b md:border-gray-400 overflow-hidden float-left md:float-none text-center">
<div class="relative mb-10 md:m-0">
<span class="cf-icon-server block md:hidden h-20 bg-center bg-no-repeat"></span>
<span class="cf-icon-error w-12 h-12 absolute left-1/2 md:left-auto md:right-0 md:top-0 -ml-6 -bottom-4"></span> </div>
<span class="md:block w-full truncate">api.airforce</span>
<h3 class="md:inline-block mt-3 md:mt-0 text-2xl text-gray-600 font-light leading-1.3">
Host </h3>
<span class="leading-1.3 text-2xl text-red-error">Error</span>
</div>
</div>
</div>
</div>
<div class="w-240 lg:w-full mx-auto mb-8 lg:px-8">
<div class="clearfix">
<div class="w-1/2 md:w-full float-left pr-6 md:pb-10 md:pr-0 leading-relaxed">
<h2 class="text-3xl font-normal leading-1.3 mb-4">What happened?</h2>
<p>The web server reported a bad gateway error.</p>
</div>
<div class="w-1/2 md:w-full float-left leading-relaxed">
<h2 class="text-3xl font-normal leading-1.3 mb-4">What can I do?</h2>
<p class="mb-6">Please try again in a few minutes.</p>
</div>
</div>
</div>
<div class="cf-error-footer cf-wrapper w-240 lg:w-full py-10 sm:py-4 sm:px-8 mx-auto text-center sm:text-left border-solid border-0 border-t border-gray-300"> <p class="text-13">
<span class="cf-footer-item sm:block sm:mb-1">Cloudflare Ray ID: <strong class="font-semibold">8d9e361c9d2900ca</strong></span>
<span class="cf-footer-separator sm:hidden">•</span>
<span id="cf-footer-item-ip" class="cf-footer-item hidden sm:block sm:mb-1">
Your IP:
<button type="button" id="cf-footer-ip-reveal" class="cf-footer-ip-reveal-btn">Click to reveal</button>
<span class="hidden" id="cf-footer-ip">85.56.128.100</span>
<span class="cf-footer-separator sm:hidden">•</span>
</span>
<span class="cf-footer-item sm:block sm:mb-1"><span>Performance & security by</span> <a rel="noopener noreferrer" href="https://www.cloudflare.com/5xx-error-landing?utm_source=errorcode_502&utm_campaign=api.airforce" id="brand_link" target="_blank">Cloudflare</a></span>
</p>
<script>(function(){function d(){var b=a.getElementById("cf-footer-item-ip"),c=a.getElementById("cf-footer-ip-reveal");b&&"classList"in b&&(b.classList.remove("hidden"),c.addEventListener("click",function(){c.classList.add("hidden");a.getElementById("cf-footer-ip").classList.remove("hidden")}))}var a=document;document.addEventListener&&a.addEventListener("DOMContentLoaded",d)})();</script>
</div><!-- /.error-footer -->
</div></div>
</body>
</html>
assistant: Entiendo que estamos enfrentando un bloqueo del servidor. Este tipo de respuesta (Error 502: Bad Gateway) generalmente indica un problema en el lado del servidor, pero también puede ser causado por un bloqueo debido a demasiadas solicitudes o algún tipo de protección contra bots.
Dado que no podemos cambiar el comportamiento del servidor, vamos a intentar una estrategia diferente para hacer que nuestro código sea más resistente a estos problemas. Aquí hay algunas modificaciones que podemos hacer:
1. Implementar un sistema de reintentos con retroceso exponencial.
2. Rotar User-Agents para parecer menos como un bot.
3. Añadir delays aleatorios entre solicitudes.
4. Implementar un sistema de caché simple para evitar solicitudes repetidas.
Aquí tienes una versión modificada del código que implementa estas estrategias:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import asyncio
import tiktoken
import random
import time
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_messages(messages: Messages) -> List[Dict[str, Any]]:
MESSAGE_MAX_CHARS = 1000 - 10
formatted = []
for message in messages:
role = message["role"]
content = message["content"]
while content:
chunk = content[:MESSAGE_MAX_CHARS]
content = content[MESSAGE_MAX_CHARS:]
formatted.append({"role": role, "content": chunk})
if content:
role = "user" if role == "assistant" else "assistant"
formatted.append(
{"role": role, "content": "[system: continuing message]"}
)
return formatted
def num_tokens_from_string(string: str, model: str = "cl100k_base") -> int:
encoding = tiktoken.get_encoding(model)
num_tokens = len(encoding.encode(string))
return num_tokens
class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin):
api_url = "https://api.airforce/chat/completions"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "claude-3-5-sonnet-20240620"
models = [
"claude-3-5-sonnet-20240620",
"claude-3-opus-20240229",
"chatgpt-4o-latest",
"gpt-4o",
"gpt-4",
"WizardLM-2-8x22B",
"llama-3.1-405b-turbo",
"llama-3.1-70b-turbo",
]
model_aliases = {}
user_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0.3 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36",
]
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
model = cls.get_model(model)
formatted_messages = format_messages(messages)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Authorization": "Bearer missing api key",
"Origin": "https://llmplayground.net",
"Referer": "https://llmplayground.net/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": random.choice(cls.user_agents),
"Content-Type": "application/json",
}
max_retries = 5
base_delay = 1
for attempt in range(max_retries):
try:
data = {
"messages": formatted_messages,
"model": model,
"max_tokens": min(900, kwargs.get("max_tokens", 4000)),
"temperature": kwargs.get("temperature", 0.7),
"top_p": 0.9,
"stream": True,
}
async with aiohttp.ClientSession(headers=headers) as session:
async with session.post(
cls.api_url, json=data, proxy=proxy, timeout=30
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Server responded with status {response.status}: {error_text}")
if response.status in [429, 502, 503, 504]:
delay = (base_delay * 2 ** attempt) + random.uniform(0, 1)
logger.info(f"Retrying in {delay:.2f} seconds...")
await asyncio.sleep(delay)
continue
yield f"Error: {response.status}, {error_text}"
return
chunk_response = ""
async for line in response.content:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
line = line[6:]
try:
data = json.loads(line)
content = data["choices"][0]["delta"].get("content")
if content:
chunk_response += content
if raw:
yield json.dumps(data)
else:
yield content
except json.JSONDecodeError:
continue
formatted_messages.append({"role": "assistant", "content": chunk_response})
if len(chunk_response) < 850: # Si la respuesta es corta, asumimos que está completa
break
# Si no está completa, continuamos la conversación
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
# Añadir un pequeño delay aleatorio entre solicitudes
await asyncio.sleep(random.uniform(1, 3))
except aiohttp.ClientError as e:
logger.error(f"Network error occurred: {str(e)}")
if attempt == max_retries - 1:
yield f"Error: Failed to connect after {max_retries} attempts. Last error: {str(e)}"
return
delay = (base_delay * 2 ** attempt) + random.uniform(0, 1)
logger.info(f"Retrying in {delay:.2f} seconds...")
await asyncio.sleep(delay)
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = RocksProvider()
def pipes(self):
return [
{
"id": f"rocks/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
max_tokens = body.get("max_tokens", 4000)
logger.debug(
f"Pipe called with: model={model}, raw={raw}, max_tokens={max_tokens}"
)
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, max_tokens=max_tokens
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un modelo de chat
test_body_chat = {
"model": "rocks/claude_3_5_sonnet_20240620",
"messages": [
{
"role": "user",
"content": "Explícame detalladamente qué modelo de IA eres y cuáles son tus capacidades.",
},
],
"raw": False,
"max_tokens": 8000,
}
print(f"\nProbando con el modelo de chat '{test_body_chat['model']}':")
async for chunk in pipe.pipe(test_body_chat):
print(chunk, end="", flush=True)
print("\nPrueba de chat completada.")
asyncio.run(test())
```
Los cambios principales incluyen:
1. Un sistema de reintentos con retroceso exponencial para manejar errores de servidor.
2. Rotación de User-Agents para parecer menos como un bot.
3. Delays aleatorios entre solicitudes para evitar ser detectado como un bot.
4. Una lógica mejorada para manejar respuestas largas, dividiendo la solicitud en múltiples llamadas si es necesario.
Esta versión del código debería ser más resistente a los problemas de bloqueo del servidor. Sin embargo, si el servidor está implementando medidas muy estrictas contra bots o scraping, es posible que aún experimentemos algunos problemas.
Si después de estos cambios sigues experimentando bloqueos consistentes, podría ser necesario considerar otras alternativas, como usar una API oficial si está disponible, o buscar otros proveedores de servicios similares que sean más amigables con el uso programático.
user: ahora ni contesta hace 10 minutos que le e mandado un mensaje y esta alli activo pero sin contestar nada
assistant: Entiendo que el modelo no está respondiendo en absoluto ahora. Este comportamiento podría indicar un problema más serio con el servicio o un bloqueo más estricto. Vamos a implementar algunas estrategias adicionales para manejar esta situación:
1. Implementar un timeout más corto para cada solicitud individual.
2. Añadir un mecanismo para cancelar la operación si no hay respuesta después de un cierto tiempo.
3. Implementar un sistema de "heartbeat" para verificar si el modelo está respondiendo.
Aquí tienes una versión actualizada del código que implementa estas estrategias:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import asyncio
import tiktoken
import random
import time
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
def format_messages(messages: Messages) -> List[Dict[str, Any]]:
MESSAGE_MAX_CHARS = 1000 - 10
formatted = []
for message in messages:
role = message["role"]
content = message["content"]
while content:
chunk = content[:MESSAGE_MAX_CHARS]
content = content[MESSAGE_MAX_CHARS:]
formatted.append({"role": role, "content": chunk})
if content:
role = "user" if role == "assistant" else "assistant"
formatted.append(
{"role": role, "content": "[system: continuing message]"}
)
return formatted
def num_tokens_from_string(string: str, model: str = "cl100k_base") -> int:
encoding = tiktoken.get_encoding(model)
num_tokens = len(encoding.encode(string))
return num_tokens
class RocksProvider(AsyncGeneratorProvider, ProviderModelMixin):
api_url = "https://api.airforce/chat/completions"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = "claude-3-5-sonnet-20240620"
models = [
"claude-3-5-sonnet-20240620",
"claude-3-opus-20240229",
"chatgpt-4o-latest",
"gpt-4o",
"gpt-4",
"WizardLM-2-8x22B",
"llama-3.1-405b-turbo",
"llama-3.1-70b-turbo",
]
model_aliases = {}
user_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0.3 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36",
]
@classmethod
def get_model(cls, model: str) -> str:
if model in cls.models:
return model
elif model in cls.model_aliases:
return cls.model_aliases[model]
else:
return cls.default_model
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
raw: bool = False,
proxy: str = None,
**kwargs,
) -> AsyncResult:
model = cls.get_model(model)
formatted_messages = format_messages(messages)
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Authorization": "Bearer missing api key",
"Origin": "https://llmplayground.net",
"Referer": "https://llmplayground.net/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": random.choice(cls.user_agents),
"Content-Type": "application/json",
}
max_retries = 5
base_delay = 1
overall_timeout = 300 # 5 minutos de timeout total
request_timeout = 30 # 30 segundos de timeout por solicitud
start_time = time.time()
for attempt in range(max_retries):
if time.time() - start_time > overall_timeout:
yield "Error: Overall timeout reached. The model is not responding."
return
try:
data = {
"messages": formatted_messages,
"model": model,
"max_tokens": min(900, kwargs.get("max_tokens", 4000)),
"temperature": kwargs.get("temperature", 0.7),
"top_p": 0.9,
"stream": True,
}
async with aiohttp.ClientSession(headers=headers) as session:
async with session.post(
cls.api_url, json=data, proxy=proxy, timeout=request_timeout
) as response:
if response.status != 200:
error_text = await response.text()
logger.error(f"Server responded with status {response.status}: {error_text}")
if response.status in [429, 502, 503, 504]:
delay = (base_delay * 2 ** attempt) + random.uniform(0, 1)
logger.info(f"Retrying in {delay:.2f} seconds...")
await asyncio.sleep(delay)
continue
yield f"Error: {response.status}, {error_text}"
return
chunk_response = ""
last_update_time = time.time()
async for line in response.content:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
line = line[6:]
try:
data = json.loads(line)
content = data["choices"][0]["delta"].get("content")
if content:
chunk_response += content
if raw:
yield json.dumps(data)
else:
yield content
last_update_time = time.time()
except json.JSONDecodeError:
continue
if time.time() - last_update_time > request_timeout:
logger.warning("No response from model for too long. Restarting request.")
break
if chunk_response:
formatted_messages.append({"role": "assistant", "content": chunk_response})
if len(chunk_response) < 850: # Si la respuesta es corta, asumimos que está completa
break
# Si no está completa, continuamos la conversación
formatted_messages.append(
{"role": "user", "content": "Please continue your previous response."}
)
else:
logger.warning("Received empty response. Retrying.")
# Añadir un pequeño delay aleatorio entre solicitudes
await asyncio.sleep(random.uniform(1, 3))
except asyncio.TimeoutError:
logger.error("Request timed out")
if attempt == max_retries - 1:
yield "Error: Requests consistently timing out. The model might be unresponsive."
return
except aiohttp.ClientError as e:
logger.error(f"Network error occurred: {str(e)}")
if attempt == max_retries - 1:
yield f"Error: Failed to connect after {max_retries} attempts. Last error: {str(e)}"
return
delay = (base_delay * 2 ** attempt) + random.uniform(0, 1)
logger.info(f"Retrying in {delay:.2f} seconds...")
await asyncio.sleep(delay)
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = RocksProvider()
def pipes(self):
return [
{
"id": f"rocks/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
max_tokens = body.get("max_tokens", 4000)
logger.debug(
f"Pipe called with: model={model}, raw={raw}, max_tokens={max_tokens}"
)
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, max_tokens=max_tokens
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un modelo de chat
test_body_chat = {
"model": "rocks/claude_3_5_sonnet_20240620",
"messages": [
{
"role": "user",
"content": "Explícame detalladamente qué modelo de IA eres y cuáles son tus capacidades.",
},
],
"raw": False,
"max_tokens": 8000,
}
print(f"\nProbando con el modelo de chat '{test_body_chat['model']}':")
async for chunk in pipe.pipe(test_body_chat):
print(chunk, end="", flush=True)
print("\nPrueba de chat completada.")
asyncio.run(test())
```
Los cambios principales en esta versión incluyen:
1. Un timeout global de 5 minutos para toda la operación.
2. Un timeout de 30 segundos para cada solicitud individual.
3. Un mecanismo para detectar si el modelo ha dejado de responder durante una solicitud.
4. Manejo mejorado de timeouts y errores de red.
5. Logging más detallado para ayudar en la depuración.
Esta versión debería ser más robusta en situaciones donde el modelo no responde o tarda mucho en responder. Si después de un tiempo razonable (5 minutos en este caso) no se recibe una respuesta completa, el código generará un mensaje de error indicando que el modelo no está respondiendo.
Si aún así no recibes ninguna respuesta, es posible que el servicio esté experimentando problemas más serios o que haya implementado medidas de bloqueo más estrictas. En ese caso, podrías considerar:
1. Verificar si el servicio está operativo a través de otros medios (por ejemplo, intentando acceder directamente desde un navegador).
2. Contactar a los administradores del servicio para averiguar si hay problemas conocidos o si han implementado nuevas restricciones.
3. Considerar usar un servicio alternativo si este continúa siendo inaccesible o no responde.
user: perfecto ahora creame otra aplicaccion con estos nuevos datos "from __future__ import annotations
import json
import uuid
from ..typing import AsyncResult, Messages
from .base_provider import AsyncGeneratorProvider, ProviderModelMixin
from ..image import ImageResponse
from ..requests import StreamSession, raise_for_status
from ..errors import ResponseStatusError
class AmigoChat(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://amigochat.io/chat/"
chat_api_endpoint = "https://api.amigochat.io/v1/chat/completions"
image_api_endpoint = "https://api.amigochat.io/v1/images/generations"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = 'gpt-4o-mini'
chat_models = [
'gpt-4o',
default_model,
'o1-preview',
'o1-mini',
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo',
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo',
'claude-3-sonnet-20240229',
'gemini-1.5-pro',
]
image_models = [
'flux-pro/v1.1',
'flux-realism',
'flux-pro',
'dalle-e-3',
]
models = [*chat_models, *image_models]
model_aliases = {
"o1": "o1-preview",
"llama-3.1-405b": "meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo",
"llama-3.2-90b": "meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo",
"claude-3.5-sonnet": "claude-3-sonnet-20240229",
"gemini-pro": "gemini-1.5-pro",
"flux-pro": "flux-pro/v1.1",
"dalle-3": "dalle-e-3",
}
persona_ids = {
'gpt-4o': "gpt",
'gpt-4o-mini': "amigo",
'o1-preview': "openai-o-one",
'o1-mini': "openai-o-one-mini",
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo': "llama-three-point-one",
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo': "llama-3-2",
'claude-3-sonnet-20240229': "claude",
'gemini-1.5-pro': "gemini-1-5-pro",
'flux-pro/v1.1': "flux-1-1-pro",
'flux-realism': "flux-realism",
'flux-pro': "flux-pro",
'dalle-e-3': "dalle-three",
}
@classmethod
def get_personaId(cls, model: str) -> str:
return cls.persona_ids[model]
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
stream: bool = False,
timeout: int = 300,
frequency_penalty: float = 0,
max_tokens: int = 4000,
presence_penalty: float = 0,
temperature: float = 0.5,
top_p: float = 0.95,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
device_uuid = str(uuid.uuid4())
max_retries = 3
retry_count = 0
while retry_count < max_retries:
try:
headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": "Bearer",
"cache-control": "no-cache",
"content-type": "application/json",
"origin": cls.url,
"pragma": "no-cache",
"priority": "u=1, i",
"referer": f"{cls.url}/",
"sec-ch-ua": '"Chromium";v="129", "Not=A?Brand";v="8"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"Linux"',
"user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
"x-device-language": "en-US",
"x-device-platform": "web",
"x-device-uuid": device_uuid,
"x-device-version": "1.0.41"
}
async with StreamSession(headers=headers, proxy=proxy) as session:
if model not in cls.image_models:
data = {
"messages": messages,
"model": model,
"personaId": cls.get_personaId(model),
"frequency_penalty": frequency_penalty,
"max_tokens": max_tokens,
"presence_penalty": presence_penalty,
"stream": stream,
"temperature": temperature,
"top_p": top_p
}
async with session.post(cls.chat_api_endpoint, json=data, timeout=timeout) as response:
await raise_for_status(response)
async for line in response.iter_lines():
line = line.decode('utf-8').strip()
if line.startswith('data: '):
if line == 'data: [DONE]':
break
try:
chunk = json.loads(line[6:]) # Remove 'data: ' prefix
if 'choices' in chunk and len(chunk['choices']) > 0:
choice = chunk['choices'][0]
if 'delta' in choice:
content = choice['delta'].get('content')
elif 'text' in choice:
content = choice['text']
else:
content = None
if content:
yield content
except json.JSONDecodeError:
pass
else:
# Image generation
prompt = messages[-1]['content']
data = {
"prompt": prompt,
"model": model,
"personaId": cls.get_personaId(model)
}
async with session.post(cls.image_api_endpoint, json=data) as response:
await raise_for_status(response)
response_data = await response.json()
if "data" in response_data:
image_urls = []
for item in response_data["data"]:
if "url" in item:
image_url = item["url"]
image_urls.append(image_url)
if image_urls:
yield ImageResponse(image_urls, prompt)
else:
yield None
break
except (ResponseStatusError, Exception) as e:
retry_count += 1
if retry_count >= max_retries:
raise e
device_uuid = str(uuid.uuid4())
assistant: Entendido. Vamos a actualizar el código de ejemplo que proporcionaste para que sea compatible dentro de la estructura del modelo usado para OpenWebUI. A continuación, te presento la implementación adaptada, siguiendo la estructura general de las aplicaciones para OpenWebUI:
```python
from __future__ import annotations
from typing import AsyncGenerator, List, Dict, Any, Optional
from pydantic import BaseModel
import aiohttp
import json
import logging
import asyncio
import uuid
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
AsyncResult = AsyncGenerator[str, None]
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class AmigoChat(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://amigochat.io/chat/"
chat_api_endpoint = "https://api.amigochat.io/v1/chat/completions"
image_api_endpoint = "https://api.amigochat.io/v1/images/generations"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = 'gpt-4o-mini'
chat_models = [
'gpt-4o',
default_model,
'o1-preview',
'o1-mini',
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo',
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo',
'claude-3-sonnet-20240229',
'gemini-1.5-pro',
]
image_models = [
'flux-pro/v1.1',
'flux-realism',
'flux-pro',
'dalle-e-3',
]
models = [*chat_models, *image_models]
model_aliases = {
"o1": "o1-preview",
"llama-3.1-405b": "meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo",
"llama-3.2-90b": "meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo",
"claude-3.5-sonnet": "claude-3-sonnet-20240229",
"gemini-pro": "gemini-1.5-pro",
"flux-pro": "flux-pro/v1.1",
"dalle-3": "dalle-e-3",
}
persona_ids = {
'gpt-4o': "gpt",
'gpt-4o-mini': "amigo",
'o1-preview': "openai-o-one",
'o1-mini': "openai-o-one-mini",
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo': "llama-three-point-one",
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo': "llama-3-2",
'claude-3-sonnet-20240229': "claude",
'gemini-1.5-pro': "gemini-1-5-pro",
'flux-pro/v1.1': "flux-1-1-pro",
'flux-realism': "flux-realism",
'flux-pro': "flux-pro",
'dalle-e-3': "dalle-three",
}
@classmethod
def get_personaId(cls, model: str) -> str:
return cls.persona_ids[model]
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
stream: bool = False,
timeout: int = 300,
frequency_penalty: float = 0,
max_tokens: int = 4000,
presence_penalty: float = 0,
temperature: float = 0.5,
top_p: float = 0.95,
**kwargs
) -> AsyncResult:
model = cls.get_model(model)
device_uuid = str(uuid.uuid4())
max_retries = 3
retry_count = 0
while retry_count < max_retries:
try:
headers = {
"Accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": "Bearer",
"Cache-Control": "no-cache",
"Content-Type": "application/json",
"Origin": cls.url,
"Pragma": "no-cache",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
"x-device-language": "en-US",
"x-device-platform": "web",
"x-device-uuid": device_uuid,
"x-device-version": "1.0.41",
}
async with aiohttp.ClientSession(headers=headers) as session:
if model not in cls.image_models:
data = {
"messages": messages,
"model": model,
"personaId": cls.get_personaId(model),
"frequency_penalty": frequency_penalty,
"max_tokens": max_tokens,
"presence_penalty": presence_penalty,
"stream": stream,
"temperature": temperature,
"top_p": top_p
}
async with session.post(cls.chat_api_endpoint, json=data, proxy=proxy, timeout=timeout) as response:
if response.status != 200:
error_text = await response.text()
yield f"Error: {response.status}, {error_text}"
return
buffer = ""
async for line in response.content:
line = line.decode('utf-8').strip()
if line.startswith('data: '):
if line == 'data: [DONE]':
break
try:
chunk = json.loads(line[6:]) # Remove 'data: ' prefix
if 'choices' in chunk and len(chunk['choices']) > 0:
choice = chunk['choices'][0]
if 'delta' in choice:
content = choice['delta'].get('content')
elif 'text' in choice:
content = choice['text']
else:
content = None
if content:
yield content
except json.JSONDecodeError:
continue
else:
prompt = messages[-1]['content']
data = {
"prompt": prompt,
"model": model,
"personaId": cls.get_personaId(model)
}
async with session.post(cls.image_api_endpoint, json=data, proxy=proxy) as response:
if response.status != 200:
error_text = await response.text()
yield f"Error: {response.status}, {error_text}"
return
response_data = await response.json()
if "data" in response_data:
image_urls = []
for item in response_data["data"]:
if "url" in item:
image_url = item["url"]
image_urls.append(image_url)
if image_urls:
yield ImageResponse(image_urls, prompt)
else:
yield None
break
except (aiohttp.ClientError, Exception) as e:
retry_count += 1
if retry_count >= max_retries:
raise e
device_uuid = str(uuid.uuid4())
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = AmigoChat()
def pipes(self):
return [
{
"id": f"amigochat/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
async def pipe(self, body: dict) -> AsyncResult:
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
timeout = body.get("timeout", 300)
logger.debug(f"Pipe called with: model={model}, raw={raw}")
try:
async for chunk in self.provider.create_async_generator(
model, messages, raw=raw, proxy=proxy, timeout=timeout
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
import asyncio
async def test():
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "amigochat/gpt_4o_mini",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
"raw": False,
}
print("\nProbando con el modelo 'gpt-4o-mini':")
async for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
asyncio.run(test())
```
Este código sigue la estructura de OpenWebUI, implementando la clase `AmigoChat` para interactuar con los servicios de chat e imágenes de AmigoChat. El manejo de errores está incluido, junto con los reintentos en caso de errores de solicitud.
Para usar este código en OpenWebUI:
1. Asegúrate de tener instalado `aiohttp` y `pydantic`.
2. Guarda este código en un archivo, por ejemplo `amigochat_provider.py`.
3. Coloca el archivo en la ubicación correcta donde OpenWebUI pueda acceder a él.
4. En la configuración de OpenWebUI, referencia este archivo y la clase `Pipe` como tu proveedor personalizado.
Si tienes problemas adicionales o necesitas más configuraciones, por favor, házmelo saber.
user: usa import requests
assistant: Claro, podemos adaptar el código para utilizar la biblioteca `requests`, que es otra opción para realizar solicitudes HTTP. Aunque `requests` es generalmente sincrónica, podemos adaptarla para funcionar dentro de la estructura asíncrona usando `asyncio`. Aquí está la versión actualizada del código usando `requests`:
```python
from __future__ import annotations
from typing import List, Dict, Any, Optional
from pydantic import BaseModel
import requests
import json
import logging
import uuid
# Configurar logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# Definiciones de tipos
Messages = List[Dict[str, Any]]
class AsyncGeneratorProvider:
pass
class ProviderModelMixin:
@classmethod
def get_model(cls, model: str) -> str:
return cls.model_aliases.get(model, model)
class AmigoChat(AsyncGeneratorProvider, ProviderModelMixin):
url = "https://amigochat.io/chat/"
chat_api_endpoint = "https://api.amigochat.io/v1/chat/completions"
image_api_endpoint = "https://api.amigochat.io/v1/images/generations"
working = True
supports_stream = True
supports_system_message = True
supports_message_history = True
default_model = 'gpt-4o-mini'
chat_models = [
'gpt-4o',
default_model,
'o1-preview',
'o1-mini',
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo',
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo',
'claude-3-sonnet-20240229',
'gemini-1.5-pro',
]
image_models = [
'flux-pro/v1.1',
'flux-realism',
'flux-pro',
'dalle-e-3',
]
models = [*chat_models, *image_models]
model_aliases = {
"o1": "o1-preview",
"llama-3.1-405b": "meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo",
"llama-3.2-90b": "meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo",
"claude-3.5-sonnet": "claude-3-sonnet-20240229",
"gemini-pro": "gemini-1.5-pro",
"flux-pro": "flux-pro/v1.1",
"dalle-3": "dalle-e-3",
}
persona_ids = {
'gpt-4o': "gpt",
'gpt-4o-mini': "amigo",
'o1-preview': "openai-o-one",
'o1-mini': "openai-o-one-mini",
'meta-llama/Meta-Llama-3.1-405B-Instruct-Turbo': "llama-three-point-one",
'meta-llama/Llama-3.2-90B-Vision-Instruct-Turbo': "llama-3-2",
'claude-3-sonnet-20240229': "claude",
'gemini-1.5-pro': "gemini-1-5-pro",
'flux-pro/v1.1': "flux-1-1-pro",
'flux-realism': "flux-realism",
'flux-pro': "flux-pro",
'dalle-e-3': "dalle-three",
}
@classmethod
def get_personaId(cls, model: str) -> str:
return cls.persona_ids[model]
@classmethod
def create_generator(
cls,
model: str,
messages: Messages,
proxy: Optional[str] = None,
stream: bool = False,
timeout: int = 300,
frequency_penalty: float = 0,
max_tokens: int = 4000,
presence_penalty: float = 0,
temperature: float = 0.5,
top_p: float = 0.95,
**kwargs
):
model = cls.get_model(model)
device_uuid = str(uuid.uuid4())
max_retries = 3
retry_count = 0
while retry_count < max_retries:
try:
headers = {
"Accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"authorization": "Bearer",
"Cache-Control": "no-cache",
"Content-Type": "application/json",
"Origin": cls.url,
"Pragma": "no-cache",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36",
"x-device-language": "en-US",
"x-device-platform": "web",
"x-device-uuid": device_uuid,
"x-device-version": "1.0.41",
}
proxies = {"http": proxy, "https": proxy} if proxy else None
if model not in cls.image_models:
# Chat completion
data = {
"messages": messages,
"model": model,
"personaId": cls.get_personaId(model),
"frequency_penalty": frequency_penalty,
"max_tokens": max_tokens,
"presence_penalty": presence_penalty,
"stream": stream,
"temperature": temperature,
"top_p": top_p
}
response = requests.post(cls.chat_api_endpoint, json=data, headers=headers, proxies=proxies, timeout=timeout)
response.raise_for_status()
buffer = ""
for line in response.iter_lines():
line = line.decode('utf-8').strip()
if line.startswith('data: '):
if line == 'data: [DONE]':
break
try:
chunk = json.loads(line[6:]) # Remove 'data: ' prefix
if 'choices' in chunk and len(chunk['choices']) > 0:
choice = chunk['choices'][0]
if 'delta' in choice:
content = choice['delta'].get('content')
elif 'text' in choice:
content = choice['text']
else:
content = None
if content:
yield content
except json.JSONDecodeError:
continue
else:
# Image generation
prompt = messages[-1]['content']
data = {
"prompt": prompt,
"model": model,
"personaId": cls.get_personaId(model)
}
response = requests.post(cls.image_api_endpoint, json=data, headers=headers, proxies=proxies, timeout=timeout)
response.raise_for_status()
response_data = response.json()
if "data" in response_data:
image_urls = [item["url"] for item in response_data["data"] if "url" in item]
if image_urls:
yield ImageResponse(image_urls, prompt)
else:
yield None
break
except (requests.RequestException, Exception) as e:
retry_count += 1
if retry_count >= max_retries:
logger.error(f"Failed after {max_retries} attempts: {str(e)}")
raise e
device_uuid = str(uuid.uuid4())
class Pipe:
class Valves(BaseModel):
NAME_PREFIX: str = ""
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.provider = AmigoChat()
def pipes(self):
return [
{
"id": f"amigochat/{model.lower().replace('-', '_').replace('/', '_')}",
"name": f"{self.valves.NAME_PREFIX}{model.upper()}",
}
for model in self.provider.models
]
def pipe(self, body: dict):
model = body["model"].split("/")[-1].replace("_", "-")
messages = body["messages"]
raw = body.get("raw", False)
proxy = body.get("proxy")
timeout = body.get("timeout", 300)
logger.debug(f"Pipe called with: model={model}, raw={raw}")
try:
for chunk in self.provider.create_generator(
model, messages, raw=raw, proxy=proxy, timeout=timeout
):
yield chunk
except Exception as e:
logger.error(f"Error in pipe: {str(e)}", exc_info=True)
yield f"Error: {str(e)}"
# Instancia de Pipe para ser utilizada por OpenWebUI
pipe = Pipe()
# Función de prueba
if __name__ == "__main__":
# Probar el método pipes
models = pipe.pipes()
print("Modelos disponibles:", models)
# Probar el método pipe con un mensaje simple
test_body = {
"model": "amigochat/gpt_4o_mini",
"messages": [
{"role": "system", "content": "Eres un asistente útil."},
{"role": "user", "content": "Hola, ¿cómo estás?"},
],
"raw": False,
}
print("\nProbando con el modelo 'gpt-4o-mini':")
for chunk in pipe.pipe(test_body):
print(chunk, end="", flush=True)
print("\nPrueba completada.")
```
### Principales Cambios:
1. **Uso de `requests`**: La implementación fue cambiada para usar la librería `requests` en lugar de `aiohttp`. Esto simplifica el manejo de sesiones en una sintaxis sincrónica.
2. **Manejo de Errores**: Se utiliza un bucle `while` con un contador de reintentos para manejar los fallos en las solicitudes, asegurando varios intentos antes de fallar de forma definitiva.
3. **Manejo de Próxies**: Se ha agregado soporte para un proxy opcional utilizando el diccionario `proxies`.
Este enfoque sincrónico puede ser más fácil de entender y es relevante si se quiere integrar en un flujo de trabajo que no usa `asyncio`. Sin embargo, es importante recordar que `requests` no se aprovecha de `asyncio` y, por lo tanto, es menos eficiente en operaciones que podrían beneficiarse de la asincronía.
user: ahora a llegado a contestar el modelo un maximo de tokens: 688, y se a parado! no termina su contestacion hasta acabar todo lo deja a medias cuando supera estas cantidades