diff --git a/web/api/chat.py b/web/api/chat.py deleted file mode 100644 index 8a926b1..0000000 --- a/web/api/chat.py +++ /dev/null @@ -1,173 +0,0 @@ -# ---------------------------------- web code ---------------------------------- - -import json -from http.server import BaseHTTPRequestHandler - -class handler(BaseHTTPRequestHandler): - - def do_POST(self): - - self.send_response(200) - self.send_header('Content-type', 'application/json') - self.end_headers() - - content_length = int(self.headers['Content-Length']) - post_data = self.rfile.read(content_length) - data = json.loads(post_data) - - self.wfile.write(chat(data['query'], data['history']).encode('utf-8')) - -# ------------------------------- env, constants ------------------------------- - -from api.get_blocks import get_top_k_blocks, Block - -from typing import List, Dict -import openai -import os -import requests -import tiktoken - -# OpenAI models -EMBEDDING_MODEL = "text-embedding-ada-002" -COMPLETIONS_MODEL = "gpt-3.5-turbo" -MODERATION_ENDPOINT = "https://api.openai.com/v1/moderations" - -# OpenAI parameters -LEN_EMBEDDINGS = 1536 -MAX_TOKEN_LEN_PROMPT = 4095 # This may be 8191, unsure. -TRUNCATE_CONTEXT = 2000 - -# OpenAI API key -OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY') -openai.api_key = OPENAI_API_KEY - -# --------------------------------- prompt code -------------------------------- - - - - -def limit_tokens(text: str, max_tokens: int, encoding_name: str = "cl100k_base") -> str: - encoding = tiktoken.get_encoding(encoding_name) - tokens = encoding.encode(text)[:max_tokens] - return encoding.decode(tokens) - - - - - - -def construct_prompt(query: str, history: List[Dict[str, str]], context: List[Block]) -> List[Dict[str, str]]: - """ - Args: - query (str): The user query. - history (List[Dict[str, str]]): The previous dialogue. Defaults to []. - blocks (List[Block]): The top-k most relevant blocks from the Alignment Research Dataset. Defaults to []. - - History takes the format: history=[ - {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": "Who won the world series in 2020?"}, - {"role": "assistant", "content": "The Los Angeles Dodgers won the World Series in 2020."}, - {"role": "user", "content": "Where was it played?"} - {"role": "assistant", "content": "Los Angeles, California."} - ] - - Returns: List[Dict[str, str]]: The prompt in messages format. - """ - - # Initialize prompt with system description - prompt = [{"role": "system", "content": "You are a helpful assistant knowledgeable about AI Alignment and Safety."}] - # Add previous dialogue - prompt.extend(history) - - instruction_prompt = \ - "Please give a clear and coherent answer to my question (written after \"Q:\") " \ - "using the following sources. Each source is labeled with a letter. Feel free to " \ - "use the sources in any order, and try to use multiple sources in your answer." - - prompt.append({"role": "user", "content": instruction_prompt}) - - # Add context from top-k blocks - context_prompt = "" - for i, block in enumerate(context): - context_prompt += f"[{chr(ord('a') + i)}] {block.title} - {block.author} - {block.date}\n\n{block.text}\n\n\n" - - context_prompt = context_prompt[:-2] # trim last two newlines - - context_prompt = limit_tokens(context_prompt, TRUNCATE_CONTEXT) # truncate to about 2k tokens - - prompt.append({"role": "user", "content": f"{context_prompt}"}) - - # Add user query - question_prompt = "In your answer, please cite any claims you make back to each source " \ - "using the format: [a], [b], etc. If you use multiple sources to make a claim " \ - "cite all of them. For example: \"AGI is concerning [c, d, e].\"" - - question_prompt += "\n\n\nQ: " + query - - prompt.append({"role": "user", "content": question_prompt}) - - return prompt - - - - -# ------------------------------------------------------------------------------ - - -def normal_completion(prompt: List[Dict[str, str]]) -> str: - """ - This function uses the OpenAI ChatCompletions API to answer a user query. - - Args: - messages (Dict[str, str]): A dictionary containing the system prompt and user prompt, in addition to any previous dialogue. - - Returns: - str: The answer generated by the API. - - Raises: - Exception: If the API call fails. - """ - try: - return openai.ChatCompletion.create( - model=COMPLETIONS_MODEL, - messages=prompt - )["choices"][0]["message"]["content"] - except Exception as e: - print(e) - return "I'm sorry, I failed to process your query. Please try again. If the problem persists, please contact the administrator." - -def chat(query: str, history: List[Dict[str, str]] = [], k: str = 10, HyDE: bool = False) -> str: - """ - This function uses the OpenAI ChatCompletions API to answer a user query. - It first checks if the query is offensive, and if so, raises an exception. - Then, it finds the top-k most relevant blocks from the Alignment Research Dataset and uses them as context for the ChatCompletions API. - It uses the blocks to generate a prompt for the ChatCompletions API. - Finally, it uses the ChatCompletions API to generate an answer to the user query. - - Args: - query (str): The user query. - previous_dialogue (List[Dict[str, str]]): The previous dialogue. Defaults to []. - k (str): The number of blocks to use as context. - mode (str): The mode to use for the ChatCompletions API. Defaults to "standard". - HyDE (bool): Whether to use the HyDE technique for semantic search. This makes search slower, but better. Defaults to False. - stream (bool): Whether to stream the results from the ChatCompletions API. Defaults to True. - stream_delay (float): The delay between each word in the streamed response when streaming a hard-coded response. Defaults to 0.1. - - Returns: - str: The answer to the user query. - - Raises: - Exception: If the query is offensive. - """ - - - - # 2. Find the top-k most relevant blocks from the Alignment Research Dataset - top_k_blocks: List[Block] = get_top_k_blocks(query, k, HyDE) - - # 3. Generate a prompt for the ChatCompletions API - prompt: List[Dict[str, str]] = construct_prompt(query, history, top_k_blocks) - - # 4. Use the top-k most relevant blocks as context for the ChatCompletions API, and generate an answer to the user query - completion: str = normal_completion(prompt) - return completion diff --git a/web/api/get_blocks.py b/web/api/get_blocks.py deleted file mode 100644 index 6d3a510..0000000 --- a/web/api/get_blocks.py +++ /dev/null @@ -1,189 +0,0 @@ -from typing import List, Tuple -import dataclasses -import itertools -import json -import pickle -import numpy as np -import openai -import regex as re -import time - -EMBEDDING_MODEL = "text-embedding-ada-002" -COMPLETIONS_MODEL = "gpt-3.5-turbo" - -import pathlib -project_path = pathlib.Path(__file__).parent -PATH_TO_DATASET_DICT = project_path / "dataset_dict.pkl" - -class Dataset: - pass - # def __init__(self, path_to_dataset: str = PATH_TO_DATASET_PKL): - # self.path_to_dataset = path_to_dataset # .json - # with open(self.path_to_dataset, 'rb') as f: - # self. - # self.load_dataset() - - # def load_dataset(self): # Load the dataset from the saved .json file - # with open(self.path_to_dataset, 'rb') as f: - # dataset_dict = json.load(f) - # self.metadata = dataset_dict['metadata'] - # self.embedding_strings = dataset_dict['embedding_strings'] - # self.embeddings_metadata_index = dataset_dict['embeddings_metadata_index'] - # self.articles_count = dataset_dict['articles_count'] - # self.total_articles_count = dataset_dict['total_articles_count'] - # self.total_char_count = dataset_dict['total_char_count'] - # self.total_word_count = dataset_dict['total_word_count'] - # self.total_sentence_count = dataset_dict['total_sentence_count'] - # self.total_block_count = dataset_dict['total_block_count'] - # self.sources_so_far = dataset_dict['sources_so_far'] - # self.info_types = dataset_dict['info_types'] - # self.embeddings = np.array(dataset_dict['embeddings']) - -@dataclasses.dataclass -class Block: - title: str - author: str - date: str - url: str - tags: str - text: str - -def get_embedding(text: str) -> np.ndarray: - """Get the embedding for a given text. The function will retry with exponential backoff if the API rate limit is reached, up to 4 times. - - Args: - text (str): The text to get the embedding for. - - Returns: - np.ndarray: The embedding for the given text. - """ - max_retries = 4 - max_wait_time = 10 - - for attempt in range(max_retries): - try: - result = openai.Embedding.create( - model=EMBEDDING_MODEL, - input=text - ) - return result["data"][0]["embedding"] - except openai.error.RateLimitError as e: - if attempt + 1 == max_retries: - raise e - wait_time = min(max_wait_time, (2 ** attempt)) # Exponential backoff - time.sleep(wait_time) - -def get_top_k_blocks(user_query: str, k: int = 10, HyDE: bool = False) -> List[Block]: - """Get the top k blocks that are most semantically similar to the query, using the provided dataset. - - Args: - query (str): The query to be searched for. - k (int, optional): The number of blocks to return. - HyDE (bool, optional): Whether to use HyDE or not. Defaults to False. - - Returns: - List[Block]: A list of the top k blocks that are most semantically similar to the query. - """ - # Get the dataset (in data/dataset.json) - # metadataset = Dataset() - # Get the dataset (in data/dataset_5percent.pkl) - with open(PATH_TO_DATASET_DICT, 'rb') as f: - metadataset = pickle.load(f) - - # Get the embedding for the query. - query_embedding = get_embedding(user_query) - - # If HyDE is enabled, produce a no-context ChatCompletion to the query. - if HyDE: - messages = [ - {"role": "system", "content": "You are a knowledgeable AI Alignment assistant."}, - {"role": "user", "content": f"Do your best to answer the question/instruction, even if you don't know the correct answer or action for sure.\nQ: {user_query}"}, - ] - HyDE_completion = openai.ChatCompletion.create( - model=COMPLETIONS_MODEL, - messages=messages - )["choices"][0]["message"]["content"] - HyDe_completion_embedding = get_embedding(f"Question: {user_query}\n\nAnswer: {HyDE_completion}") - - similarity_scores = np.dot(metadataset["embeddings"], HyDe_completion_embedding) - else: - similarity_scores = np.dot(metadataset["embeddings"], query_embedding) - - ordered_blocks = np.argsort(similarity_scores)[::-1] # Sort the blocks by similarity score - - top_k_block_indices = ordered_blocks[:k] # Get the top k indices of the blocks - top_k_metadata_indexes = [metadataset["embeddings_metadata_index"][i] for i in top_k_block_indices] - - # -------------------------------------------------------------------------- - - # we've got some sort of truncation issue with the dataset. Ideally, delete these lines - - tkbi = [] - tkmi = [] - - bl = len(metadataset["embedding_strings"]) - ml = len(metadataset["metadata"]) - - for i in range(len(top_k_block_indices)): - if top_k_block_indices[i] >= bl or top_k_metadata_indexes[i] >= ml: - print("!!!TRUNCATION ERROR!!!") - print(f"- top_k_block_indices: {top_k_block_indices}, sampling array of length {bl}") - print(f"- top_k_metadata_indexes: {top_k_metadata_indexes}, sampling array of length {ml}") - else: - tkbi.append(top_k_block_indices[i]) - tkmi.append(top_k_metadata_indexes[i]) - - - - # -------------------------------------------------------------------------- - - top_k_texts = [metadataset["embedding_strings"][i] for i in tkbi] - top_k_metadata = [metadataset["metadata"][i] for i in tkmi] - - - # Combine the top k texts and metadata into a list of Block objects - top_k_metadata_and_text = [list(top_k_metadata[i]) + [strip_block(top_k_texts[i])] for i in range(len(top_k_metadata))] - blocks = [Block(*block) for block in top_k_metadata_and_text] - - return unify(blocks) - - - -# for all blocks that are "the same" (same title, author, date, url, tags), -# combine their text with "\n\n.....\n\n" in between. Return them in order such -# that the combined block has the minimum index of the blocks combined. - -def unify(blocks: List[Block]) -> List[Block]: - - key = lambda bi: (bi[0].title or "", bi[0].author or "", bi[0].date or "", bi[0].url or "", bi[0].tags or "") - - blocks_plus_old_index = [(block, i) for i, block in enumerate(blocks)] - blocks_plus_old_index.sort(key=key) - - unified_blocks: List[Tuple[Block, int]] = [] - - for key, group in itertools.groupby(blocks_plus_old_index, key=key): - group = list(group) - if len(group) == 0: continue - - text = "\n\n\n.....\n\n\n".join([block[0].text for block in group]) - - min_index = min([block[1] for block in group]) - - unified_blocks.append((Block(key[0], key[1], key[2], key[3], key[4], text), min_index)) - - unified_blocks.sort(key=lambda bi: bi[1]) - blocks = [block for block, _ in unified_blocks] - return blocks - - -# we the title and authors inside the contents of the block, so that searches for -# the title or author will pull it up. This strips it back out. -def strip_block(text: str) -> str: - r = re.match(r"^\"(.*)\"\s*-\s*Title:.*$", text, re.DOTALL) - if not r: - print("Warning: couldn't strip block") - print(text) - return r.group(1) if r else text - - diff --git a/web/api/requirements.txt b/web/api/requirements.txt deleted file mode 100644 index ce36a42..0000000 --- a/web/api/requirements.txt +++ /dev/null @@ -1,4 +0,0 @@ -openai==0.27.2 -numpy==1.24.2 -tenacity==8.2.2 -tiktoken diff --git a/web/api/semantic_search.py b/web/api/semantic_search.py deleted file mode 100644 index 32079b9..0000000 --- a/web/api/semantic_search.py +++ /dev/null @@ -1,34 +0,0 @@ -# ---------------------------------- web code ---------------------------------- - -import json -import dataclasses -from api.get_blocks import get_top_k_blocks -from http.server import BaseHTTPRequestHandler - -@dataclasses.dataclass -class Block: - title: str - author: str - date: str - url: str - tags: str - text: str - -class Encoder(json.JSONEncoder): - def default(self, o): - return dataclasses.asdict(o) if dataclasses.is_dataclass(o) else super().default(o) - -class handler(BaseHTTPRequestHandler): - - def do_POST(self): - - self.send_response(200) - self.send_header('Content-type', 'application/json') - self.end_headers() - - content_length = int(self.headers['Content-Length']) - post_data = self.rfile.read(content_length) - data = json.loads(post_data) - - self.wfile.write(json.dumps(get_top_k_blocks(data['query']), cls = Encoder).encode('utf-8')) -