from __future__ import annotations
import json, requests, re
from curl_cffi import requests as cf_reqs
from ..typing import CreateResult, Messages
from .base_provider import ProviderModelMixin, AbstractProvider
from .helper import format_prompt, get_connector, get_cookies
class HuggingChat(AbstractProvider, ProviderModelMixin):
url = "https://huggingface.co/chat"
working = True
supports_stream = True
default_model = "mistralai/Mixtral-8x7B-Instruct-v0.1"
models = [
"HuggingFaceH4/zephyr-orpo-141b-A35b-v0.1",
'CohereForAI/c4ai-command-r-plus',
'mistralai/Mixtral-8x7B-Instruct-v0.1',
'google/gemma-1.1-7b-it',
'NousResearch/Nous-Hermes-2-Mixtral-8x7B-DPO',
'mistralai/Mistral-7B-Instruct-v0.2',
'meta-llama/Meta-Llama-3-70B-Instruct',
'microsoft/Phi-3-mini-4k-instruct',
'01-ai/Yi-1.5-34B-Chat'
]
model_aliases = {
"mistralai/Mistral-7B-Instruct-v0.1": "mistralai/Mistral-7B-Instruct-v0.2"
}
@classmethod
def create_completion(
cls,
model: str,
messages: Messages,
stream: bool,
**kwargs
) -> CreateResult:
if (model in cls.models) :
session = requests.Session()
headers = {
'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',
'cache-control': 'no-cache',
'origin': 'https://huggingface.co',
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': 'https://huggingface.co/chat/',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
'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 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
}
json_data = {
'searchEnabled': True,
'ethicsModalAccepted': True,
'ethicsModalAcceptedAt': None,
'activeModel': 'CohereForAI/c4ai-command-r-plus', # doesn't matter
'hideEmojiOnSidebar': False,
'shareConversationsWithModelAuthors': False,
'customPrompts': {},
'assistants': [],
'tools': {},
'disableStream': False,
'recentlySaved': False,
}
response = cf_reqs.post('https://huggingface.co/chat/settings', headers=headers, json=json_data)
session.cookies.update(response.cookies)
response = session.post('https://huggingface.co/chat/conversation',
headers=headers, json={'model': model})
conversationId = response.json()['conversationId']
response = session.get(f'https://huggingface.co/chat/conversation/{conversationId}/__data.json?x-sveltekit-invalidated=11',
headers=headers,
)
messageId = extract_id(response.json())
settings = {
"inputs": format_prompt(messages),
"id": messageId,
"is_retry": False,
"is_continue": False,
"web_search": False,
# enable tools
"tools": {
"websearch": True,
"document_parser": False,
"query_calculator": False,
"image_generation": False,
"image_editing": False,
"fetch_url": False,
}
}
payload = {
"data": json.dumps(settings),
}
response = session.post(f"https://huggingface.co/chat/conversation/{conversationId}",
headers=headers, data=payload, stream=True,
)
first_token = True
for line in response.iter_lines():
line = json.loads(line)
if "type" not in line:
raise RuntimeError(f"Response: {line}")
elif line["type"] == "stream":
token = line["token"]
if first_token:
token = token.lstrip().replace('\u0000', '')
first_token = False
#yield token
yield (token)
elif line["type"] == "finalAnswer":
break
def extract_id(response: dict) -> str:
data = response["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):
return item
return None
