User: hola
Assistant: ¡Hola! ¿En qué puedo ayudarte hoy?
User: estas bien?
Assistant: ¡Gracias por preguntar! Estoy aquí para ayudarte. ¿Hay algo específico en lo que necesites asistencia?
User: como estas?
Assistant: ¡Estoy funcionando perfectamente, gracias por preguntar! Estoy aquí para ayudarte en lo que necesites, así que dime, ¿en qué puedo asistirte hoy?
User: mira esto, esto es una funcion que se instala dentro de openwebui "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",
    ]

    image_models = [
        "flux-pro",
        "flux-realism",
        "Gpt-3.5",
    ]

    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",
        "flux-pro": "Gpt-3.5",
    }

    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",
        "flux-pro/v1.1": "flux-1-1-pro",
        "flux-realism": "flux-realism",
        "flux-pro": "flux-pro",
    }

    @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 == "ONE]":
                                        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[-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:
                            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:
                                    for url in image_urls:
                                        yield f"![Generated Image]({url})"
                            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}")

        if model in self.provider.chat_models:
            model_specific_message = {
                "role": "system",
                "content": f"Eres un modelo de lenguaje basado en {model}. Responde de acuerdo a tus capacidades y características específicas.",
            }
            messages = [model_specific_message] + messages

        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 modelo de chat
        test_body_chat = {
            "model": "amigochat/meta_llama_meta_llama_3_1_405b_instruct_turbo",
            "messages": [
                {"role": "user", "content": "Hola, ¿qué modelo de IA eres?"},
            ],
            "raw": False,
        }
        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.")

        # Probar el método pipe con un modelo de imagen
        test_body_image = {
            "model": "amigochat/dalle_e_3",
            "messages": [
                {"role": "user", "content": "Un gato jugando ajedrez"},
            ],
            "raw": False,
        }
        print(f"\nProbando con el modelo de imagen '{test_body_image['model']}':")
        async for chunk in pipe.pipe(test_body_image):
            print(chunk)
        print("\nPrueba de imagen completada.")

    asyncio.run(test())

Assistant: