Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, Request | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.responses import FileResponse | |
| from fastapi import FastAPI, BackgroundTasks, HTTPException, Query | |
| from fastapi.responses import StreamingResponse | |
| from starlette.concurrency import run_in_threadpool | |
| from datasets import load_dataset | |
| from fastapi import FastAPI, Depends, HTTPException, status | |
| from fastapi.security import OAuth2PasswordBearer | |
| import random | |
| import json | |
| from genson import SchemaBuilder | |
| from pathvalidate import sanitize_filename | |
| from openai import OpenAI | |
| import hashlib | |
| import base64 | |
| from pprint import pprint | |
| import asyncio | |
| import importlib.util | |
| import traceback | |
| import sys | |
| import json | |
| import jsonschema | |
| from utils import extract_code | |
| import numpy as np | |
| import os | |
| import requests | |
| import secrets | |
| import urllib.parse | |
| app = FastAPI() | |
| client_id = os.getenv("OAUTH_CLIENT_ID") | |
| client_secret = os.getenv("OAUTH_CLIENT_SECRET") | |
| space_host = os.getenv("SPACE_HOST") | |
| client = OpenAI( | |
| base_url="https://openrouter.ai/api/v1", | |
| api_key=os.environ.get('OPENROUTER_KEY') | |
| ) | |
| oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") | |
| state_queue_map = {} | |
| def is_sharegpt(sample): | |
| schema = { | |
| '$schema': 'http://json-schema.org/schema#', | |
| 'type': 'object', | |
| 'properties': { | |
| 'conversations': { | |
| 'type': 'array', | |
| 'items': { | |
| 'type': 'object', | |
| 'properties': { | |
| 'from': { | |
| 'type': 'string', | |
| 'enum': [ | |
| 'human', | |
| 'gpt', | |
| 'system']}, | |
| 'value': { | |
| 'type': 'string'}}, | |
| 'required': [ | |
| 'from', | |
| 'value']}}}, | |
| 'required': ['conversations']} | |
| try: | |
| jsonschema.validate(instance=sample, schema=schema) | |
| return True | |
| except jsonschema.exceptions.ValidationError as e: | |
| return False | |
| def sha256(string): | |
| # Create a hashlib object for SHA-256 | |
| sha256_hash = hashlib.sha256() | |
| # Update the hash object with your string encoded as bytes | |
| sha256_hash.update(string.encode('utf-8')) | |
| return sha256_hash.hexdigest() | |
| def get_adapter_name(sample): | |
| builder = SchemaBuilder() | |
| builder.add_object(sample) | |
| schema = builder.to_schema() | |
| return sha256(json.dumps(schema)) | |
| def has_adapter(sample): | |
| adapter_name = get_adapter_name(sample) | |
| module_name = f"dataset_adapters.{adapter_name}" | |
| module_spec = importlib.util.find_spec(module_name) | |
| if module_spec is None: | |
| return False | |
| return True | |
| def auto_tranform(sample): | |
| adapter_name = get_adapter_name(sample) | |
| if not has_adapter(sample): | |
| create_adapter(sample, adapter_name) | |
| module_name = f"dataset_adapters.{adapter_name}" | |
| spec = importlib.util.spec_from_file_location( | |
| module_name, f"dataset_adapters/{adapter_name}.py") | |
| dynamic_module = importlib.util.module_from_spec(spec) | |
| sys.modules[module_name] = dynamic_module | |
| spec.loader.exec_module(dynamic_module) | |
| transformed_data = dynamic_module.transform_data(sample) | |
| if isinstance(transformed_data, list): | |
| return {'conversations': transformed_data} | |
| return transformed_data | |
| with open(f"dataset_adapters/{adapter_name}.py", 'w') as file: | |
| file.write(code_string) | |
| def create_adapter(sample, adapter_name): | |
| builder = SchemaBuilder() | |
| builder.add_object(sample) | |
| schema = builder.to_schema() | |
| prompt = f"""Make me minimal and efficient python code to convert data in the shape of | |
| initial data shape | |
| ==========➡️📑📐========== | |
| ```jsonschema | |
| {schema} | |
| ``` | |
| ==========➡️📑📐========== | |
| to equivalent data in the form | |
| final data shape | |
| ==========⬇️📑📐========== | |
| ```jsonschema | |
| {{'$schema': 'http://json-schema.org/schema#', 'type': 'object', 'properties': {{'conversations': {{'type': 'array', 'items': {{'type': 'object', 'properties': {{'from': {{ 'type': 'string', 'enum': ['human', 'gpt', 'system'] }}, 'value': {{'type': 'string'}}}}, 'required': ['from', 'value']}}}}}}, 'required': ['conversations']}} | |
| ``` | |
| ==========⬇️📑📐========== | |
| the data to transform is | |
| ```json | |
| {sample} | |
| ``` | |
| Inside the data to transform, `input` and `instruction` is usually associated with `"from" : "human"` while `output` is usually associated with `"from" : "gpt"` | |
| For transforming the data you shall use python. Make robust and elegant python code that will do the transformation | |
| your code will contain a function `def transform_data(data):` that does the transformation and outputs the newly shaped data. Only the data, no schema. Your code snippet will include only the function signature and body. I know how to call it. You won't need to import anything, I will take care of parsing and dumping json. You work with dicts. Remember to be careful if you iterate over the data because I want the output conversation to always start with the prompt. In other words, always process "input" before "output" and "instruction" before "output". Such heuristics are very important. If there is "instruction" and "input" and the "input" is not empty, concat the input at the end of the first message. If the data contains no "system" message, human always speaks first. If it contains a "system" message, the "system" message is first, then human, then gpt, then alternating if needed | |
| "human" ALWAYS SPEAKS BEFORE "gpt", if you suspect your code makes "gpt speak first, fix it | |
| MOST IMPORTANT IS THAT YOU look at the initial data shape (➡️📑📐) to ground your transformation into final data shape (⬇️📑📐) | |
| Your output should contain only the code for `def transform_data(data):`, signature and body. Put the code inside markdown code block""" | |
| response = client.chat.completions.create( | |
| model="openai/gpt-4-1106-preview", | |
| messages=[ | |
| {"role": "system", "content": """You are ChatGPT, a large language model trained by OpenAI, based on the GPT-4 architecture. | |
| Knowledge cutoff: 2023-04 | |
| Current date: 2023-11-05 | |
| Image input capabilities: Enabled"""}, | |
| {"role": "user", "content": prompt} | |
| ] | |
| ) | |
| val = response.choices[0].message.content | |
| code_string = extract_code(val) | |
| if code_string is None: | |
| raise Exception("hey la") | |
| with open(f"dataset_adapters/{adapter_name}.py", 'w') as file: | |
| file.write(code_string) | |
| async def get_sample(hash: str = Query(..., alias="hash")): | |
| res = await get_sample_by_hash(hash) | |
| if res is None: | |
| raise HTTPException(status_code=404, detail="Item not found") | |
| data, dataset = res | |
| sample = auto_tranform(json.loads(data)) | |
| return {'sample': sample, 'dataset': dataset} | |
| def generate_random_string(length=16): | |
| return secrets.token_hex(length) | |
| async def get_oauth_token(): | |
| queue = asyncio.Queue() | |
| async def event_stream(queue, state): | |
| state_queue_map[state] = queue | |
| redirect_uri = f'https://{space_host}/login/callback' | |
| auth_url = ( | |
| f"https://huggingface.co/oauth/authorize?" | |
| f"redirect_uri={urllib.parse.quote(redirect_uri)}&" | |
| f"client_id={client_id}&" | |
| f"scope=openid%20profile&" | |
| f"response_type=code&" | |
| f"state={state}" | |
| ) | |
| yield f"data: {json.dumps({ 'url' : auth_url })}\n\n" | |
| try: | |
| while True: | |
| message = await queue.get() | |
| if 'end_stream' in message and message['end_stream']: | |
| break | |
| yield f"data: {json.dumps(message)}\n\n" | |
| finally: | |
| del state_queue_map[state] | |
| state = generate_random_string() | |
| return StreamingResponse( | |
| event_stream(queue, state), | |
| media_type="text/event-stream") | |
| async def get_random_sample(background_tasks: BackgroundTasks, dataset_name: str = Query(..., alias="dataset-name"), index: str = Query(None, alias="index")): | |
| queue = asyncio.Queue() | |
| def event_stream(queue): | |
| yield f"data: {json.dumps({'status': 'grab_sample'})}\n\n" | |
| try: | |
| headers = {"Authorization": f"Bearer {os.environ.get('HF_TOKEN')}"} | |
| API_URL = f"https://datasets-server.huggingface.co/info?dataset={dataset_name}" | |
| def query(): | |
| response = requests.get(API_URL, headers=headers) | |
| return response.json() | |
| data = query() | |
| splits = data['dataset_info']['default']['splits'] | |
| split = next(iter(splits.values())) | |
| num_samples = split['num_examples'] | |
| split_name = split['name'] | |
| idx = random.randint( | |
| 0, num_samples) if index is None else int(index) | |
| API_URL = f"https://datasets-server.huggingface.co/rows?dataset={dataset_name}&config=default&split=train&offset={idx}&length=1" | |
| def query(): | |
| headers = { | |
| "Authorization": f"Bearer {os.environ.get('HF_TOKEN')}"} | |
| response = requests.get(API_URL, headers=headers) | |
| if response.status_code != 200: | |
| raise Exception("hugging face api error") | |
| return response.json() | |
| data = query() | |
| random_sample = data['rows'][0]['row'] | |
| hashed = sha256(json.dumps(random_sample)) | |
| except Exception as e: | |
| message = "" | |
| if hasattr(e, 'message'): | |
| message = e.message | |
| else: | |
| message = str(e) | |
| print("error : ", message) | |
| yield f"data: {json.dumps({'status': 'error', 'message' : message })}\n\n" | |
| transformed_data = random_sample | |
| success = True | |
| if not is_sharegpt(random_sample): | |
| try: | |
| if not has_adapter(random_sample): | |
| yield f"data: {json.dumps({'status': 'creating_adapter'})}\n\n" | |
| transformed_data = auto_tranform(random_sample) | |
| except Exception as e: | |
| success = False | |
| if hasattr(e, 'message'): | |
| print("error : ", e.message) | |
| else: | |
| print("error : ", e) | |
| yield f"data: {json.dumps({'status': 'error'})}\n\n" | |
| if success: | |
| yield f"data: {json.dumps({'status': 'done', 'data' : transformed_data, 'index' : str(idx)})}\n\n" | |
| return StreamingResponse( | |
| event_stream(queue), | |
| media_type="text/event-stream") | |
| async def get_random_sample(dataset_name: str = Query(..., alias="dataset-name")): | |
| try: | |
| dataset = load_dataset(dataset_name, streaming=True) | |
| split = [key for key in dataset.keys() if "train" in key] | |
| dataset = load_dataset(dataset_name, split=split[0], streaming=True) | |
| buffer_size = 100 # Define a reasonable buffer size | |
| samples_buffer = [ | |
| sample for _, sample in zip( | |
| range(buffer_size), dataset)] | |
| random_sample = random.choice(samples_buffer) | |
| hashed = sha256(json.dumps(random_sample)) | |
| sanitized = sanitize_filename(dataset_name) | |
| module_name = f"dataset_adapters.{sanitized}" | |
| module_spec = importlib.util.find_spec(module_name) | |
| if module_spec is None: | |
| create_adapter(random_sample, sanitized) | |
| spec = importlib.util.spec_from_file_location( | |
| module_name, f"dataset_adapters/{sanitized}.py") | |
| dynamic_module = importlib.util.module_from_spec(spec) | |
| sys.modules[module_name] = dynamic_module | |
| spec.loader.exec_module(dynamic_module) | |
| transformed_data = dynamic_module.transform_data(random_sample) | |
| return transformed_data | |
| except FileNotFoundError: | |
| raise HTTPException(status_code=404, detail="Dataset not found") | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def oauth_callback(code: str, state: str): | |
| credentials = f"{client_id}:{client_secret}" | |
| credentials_bytes = credentials.encode("ascii") | |
| base64_credentials = base64.b64encode(credentials_bytes) | |
| auth_header = f"Basic {base64_credentials.decode('ascii')}" | |
| username = "" | |
| try: | |
| token_response = requests.post( | |
| 'https://huggingface.co/oauth/token', | |
| headers={'Authorization': auth_header}, | |
| data={ | |
| 'grant_type': 'authorization_code', | |
| 'code': code, | |
| 'redirect_uri': f'https://{space_host}/login/callback', | |
| 'client_id': client_id | |
| } | |
| ) | |
| print(token_response.status_code, token_response.text) | |
| if token_response.status_code == 200: | |
| tokens = token_response.json() | |
| access_token = tokens.get('access_token') | |
| if access_token: | |
| url = "https://huggingface.co/oauth/userinfo" | |
| payload = "" | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Authorization": f"Bearer {access_token}" | |
| } | |
| response = requests.request( | |
| "GET", url, data=payload, headers=headers) | |
| data = response.json() | |
| username = data["preferred_username"] | |
| picture = data["picture"] | |
| if state in state_queue_map: | |
| queue = state_queue_map[state] | |
| await queue.put({"access_token": access_token, "username": username, "picture" : f"https://huggingface.co{picture}" }) | |
| await queue.put({"end_stream": True}) | |
| else: | |
| username = "" | |
| else: | |
| access_token = "" | |
| except Exception: | |
| traceback.print_exc() | |
| access_token = "" | |
| return {"access_token": access_token, "username": username} | |
| async def get_oauth_config(request: Request): | |
| return { | |
| "client_id": client_id, | |
| "redirect_uri": f'https://{space_host}/login/callback' | |
| } | |
| async def get_current_user(token: str = Depends(oauth2_scheme)): | |
| if not token: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Missing token", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| url = "https://huggingface.co/oauth/userinfo" | |
| headers = {"Authorization": f"Bearer {token}"} | |
| response = requests.get(url, headers=headers) | |
| if response.status_code != 200: | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid token", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| user_info = response.json() | |
| return user_info | |
| async def gated_route(current_user: str = Depends(get_current_user)): | |
| # Your logic here. The endpoint will only be accessible if the token is valid | |
| return {"message": "You are authorized to access this route"} | |
| def index() -> FileResponse: | |
| return FileResponse(path="static/index.html", media_type="text/html") | |
| app.mount("/", StaticFiles(directory="static"), name="static") | |