networks/huggingchat_streamer.py

import copy
import json
import re

import requests
from curl_cffi import requests as cffi_requests

from tclogger import logger

from constants.models import MODEL_MAP
from constants.envs import PROXIES
from constants.headers import HUGGINGCHAT_POST_HEADERS, HUGGINGCHAT_SETTINGS_POST_DATA
from messagers.message_outputer import OpenaiStreamOutputer
from messagers.message_composer import MessageComposer
from messagers.token_checker import TokenChecker


class HuggingchatRequester:
    def __init__(self, model: str):
        if model in MODEL_MAP.keys():
            self.model = model
        else:
            self.model = "nous-mixtral-8x7b"
        self.model_fullname = MODEL_MAP[self.model]

    def get_hf_chat_id(self):
        request_url = "https://huggingface.co/chat/settings"
        request_body = copy.deepcopy(HUGGINGCHAT_SETTINGS_POST_DATA)
        extra_body = {
            "activeModel": self.model_fullname,
        }
        request_body.update(extra_body)
        logger.note(f"> hf-chat ID:", end=" ")

        res = cffi_requests.post(
            request_url,
            headers=HUGGINGCHAT_POST_HEADERS,
            json=request_body,
            proxies=PROXIES,
            timeout=10,
            impersonate="chrome",
        )
        self.hf_chat_id = res.cookies.get("hf-chat")
        if self.hf_chat_id:
            logger.success(f"[{self.hf_chat_id}]")
        else:
            logger.warn(f"[{res.status_code}]")
            logger.warn(res.text)
            raise ValueError(f"Failed to get hf-chat ID: {res.text}")

    def get_conversation_id(self, system_prompt: str = ""):
        request_url = "https://huggingface.co/chat/conversation"
        request_headers = HUGGINGCHAT_POST_HEADERS
        extra_headers = {
            "Cookie": f"hf-chat={self.hf_chat_id}",
        }
        request_headers.update(extra_headers)
        request_body = {
            "model": self.model_fullname,
            "preprompt": system_prompt,
        }
        logger.note(f"> Conversation ID:", end=" ")

        res = requests.post(
            request_url,
            headers=request_headers,
            json=request_body,
            proxies=PROXIES,
            timeout=10,
        )
        if res.status_code == 200:
            conversation_id = res.json()["conversationId"]
            logger.success(f"[{conversation_id}]")
        else:
            logger.warn(f"[{res.status_code}]")
            raise ValueError("Failed to get conversation ID!")
        self.conversation_id = conversation_id
        return conversation_id

    def get_last_message_id(self):
        request_url = f"https://huggingface.co/chat/conversation/{self.conversation_id}/__data.json?x-sveltekit-invalidated=11"
        request_headers = HUGGINGCHAT_POST_HEADERS
        extra_headers = {
            "Cookie": f"hf-chat={self.hf_chat_id}",
        }
        request_headers.update(extra_headers)
        logger.note(f"> Message ID:", end=" ")

        message_id = None
        res = requests.post(
            request_url,
            headers=request_headers,
            proxies=PROXIES,
            timeout=10,
        )
        if res.status_code == 200:
            data = res.json()["nodes"][1]["data"]
            # find the last element which matches the format of uuid4
            uuid_pattern = re.compile(
                r"^[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}$"
            )
            for item in data:
                if type(item) == str and uuid_pattern.match(item):
                    message_id = item
            logger.success(f"[{message_id}]")
        else:
            logger.warn(f"[{res.status_code}]")
            raise ValueError("Failed to get message ID!")

        return message_id

    def log_request(self, url, method="GET"):
        logger.note(f"> {method}:", end=" ")
        logger.mesg(f"{url}", end=" ")

    def log_response(
        self, res: requests.Response, stream=False, iter_lines=False, verbose=False
    ):
        status_code = res.status_code
        status_code_str = f"[{status_code}]"

        if status_code == 200:
            logger_func = logger.success
        else:
            logger_func = logger.warn

        logger.enter_quiet(not verbose)
        logger_func(status_code_str)

        if status_code != 200:
            logger_func(res.text)

        if stream:
            if not iter_lines:
                return

            for line in res.iter_lines():
                line = line.decode("utf-8")
                line = re.sub(r"^data:\s*", "", line)
                line = line.strip()
                if line:
                    try:
                        data = json.loads(line, strict=False)
                        msg_type = data.get("type")
                        if msg_type == "status":
                            msg_status = data.get("status")
                        elif msg_type == "stream":
                            content = data.get("token", "")
                            logger_func(content, end="")
                        elif msg_type == "finalAnswer":
                            full_content = data.get("text")
                            logger.success("\n[Finished]")
                            break
                        else:
                            pass
                    except Exception as e:
                        logger.warn(e)
        else:
            logger_func(res.json())

        logger.exit_quiet(not verbose)

    def chat_completions(self, messages: list[dict], iter_lines=False, verbose=False):
        composer = MessageComposer(model=self.model)
        system_prompt, input_prompt = composer.decompose_to_system_and_input_prompt(
            messages
        )

        checker = TokenChecker(input_str=system_prompt + input_prompt, model=self.model)
        checker.check_token_limit()

        self.get_hf_chat_id()
        self.get_conversation_id(system_prompt=system_prompt)
        message_id = self.get_last_message_id()

        request_url = f"https://huggingface.co/chat/conversation/{self.conversation_id}"
        request_headers = copy.deepcopy(HUGGINGCHAT_POST_HEADERS)
        extra_headers = {
            "Content-Type": "text/event-stream",
            "Referer": request_url,
            "Cookie": f"hf-chat={self.hf_chat_id}",
        }
        request_headers.update(extra_headers)
        request_body = {
            "files": [],
            "id": message_id,
            "inputs": input_prompt,
            "is_continue": False,
            "is_retry": False,
            "web_search": False,
        }
        self.log_request(request_url, method="POST")

        res = requests.post(
            request_url,
            headers=request_headers,
            json=request_body,
            proxies=PROXIES,
            stream=True,
        )
        self.log_response(res, stream=True, iter_lines=iter_lines, verbose=verbose)
        return res


class HuggingchatStreamer:
    def __init__(self, model: str):
        if model in MODEL_MAP.keys():
            self.model = model
        else:
            self.model = "nous-mixtral-8x7b"
        self.model_fullname = MODEL_MAP[self.model]
        self.message_outputer = OpenaiStreamOutputer(model=self.model)

    def chat_response(self, messages: list[dict], verbose=False):
        requester = HuggingchatRequester(model=self.model)
        return requester.chat_completions(
            messages=messages, iter_lines=False, verbose=verbose
        )

    def chat_return_generator(self, stream_response: requests.Response, verbose=False):
        is_finished = False
        for line in stream_response.iter_lines():
            line = line.decode("utf-8")
            line = re.sub(r"^data:\s*", "", line)
            line = line.strip()
            if not line:
                continue

            content = ""
            content_type = "Completions"
            try:
                data = json.loads(line, strict=False)
                msg_type = data.get("type")
                if msg_type == "status":
                    msg_status = data.get("status")
                    continue
                elif msg_type == "stream":
                    content_type = "Completions"
                    content = data.get("token", "")
                    if verbose:
                        logger.success(content, end="")
                elif msg_type == "finalAnswer":
                    content_type = "Finished"
                    content = ""
                    full_content = data.get("text")
                    if verbose:
                        logger.success("\n[Finished]")
                    is_finished = True
                    break
                else:
                    continue
            except Exception as e:
                logger.warn(e)

            output = self.message_outputer.output(
                content=content, content_type=content_type
            )
            yield output

        if not is_finished:
            yield self.message_outputer.output(content="", content_type="Finished")

    def chat_return_dict(self, stream_response: requests.Response):
        final_output = self.message_outputer.default_data.copy()
        final_output["choices"] = [
            {
                "index": 0,
                "finish_reason": "stop",
                "message": {"role": "assistant", "content": ""},
            }
        ]
        final_content = ""
        for item in self.chat_return_generator(stream_response):
            try:
                data = json.loads(item)
                delta = data["choices"][0]["delta"]
                delta_content = delta.get("content", "")
                if delta_content:
                    final_content += delta_content
            except Exception as e:
                logger.warn(e)
        final_output["choices"][0]["message"]["content"] = final_content.strip()
        return final_output


if __name__ == "__main__":
    # model = "command-r-plus"
    model = "llama3-70b"
    # model = "zephyr-141b"

    streamer = HuggingchatStreamer(model=model)
    messages = [
        {
            "role": "system",
            "content": "You are an LLM developed by CloseAI.\nYour name is Hansimov-Copilot.",
        },
        {"role": "user", "content": "Hello, what is your role?"},
        {"role": "assistant", "content": "I am an LLM."},
        {"role": "user", "content": "What is your name?"},
    ]

    streamer.chat_response(messages=messages)
    # HF_ENDPOINT=https://hf-mirror.com python -m networks.huggingchat_streamer