wassname/stampy-chat
1
0
Fork 0
mirror of https://github.com/wassname/stampy-chat.git synced 2026-09-11 12:50:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

delete old api folder

Browse Source
This commit is contained in:
Fraser
2023-03-30 09:36:03 -04:00
parent e0c7eb8cc7
commit 55d7bc8d68
4 changed files with 0 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/api/chat.py
-173
View File
@@ -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
web/api/get_blocks.py
-189
View File
@@ -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
web/api/requirements.txt
-4
View File
@@ -1,4 +0,0 @@
openai==0.27.2
numpy==1.24.2
tenacity==8.2.2
tiktoken
web/api/semantic_search.py
-34
View File
@@ -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'))