|
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"]
|
|
|
|
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 = "llama3-70b"
|
|
|
|
|
|
streamer = HuggingchatStreamer(model=model)
|
|
messages = [
|
|
{
|
|
"role": "system",
|
|
"content": "You are an LLM developed by CloseAI.\nYour name is Niansuh-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)
|
|
|
|
|
