File size: 5,631 Bytes
e29422a 21d0902 3917612 0c66821 21d0902 e29422a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 |
from components.induce_personality import construct_big_five_words
from components.constant import (
ACCESS,
QUERY_REWRITING,
RAG,
PERSONALITY,
PERSONALITY_LIST,
REWRITE_PASSAGES,
NUM_PASSAGES,
DEVICE,
RESPONSE_GENERATOR,
TEMPLATE_PAYLOAD,
)
from components.prompt import SYSTEM_INSTRUCTION, RAG_INSTRUCTION, PERSONALITY_INSTRUCTION
import requests
import together
def generate_response_debugging(history):
# outputs_text = "This is a test response"
outputs_text = " ".join([item["content"] for item in history])
history = history + [{"role": "assistant", "content": outputs_text}]
return outputs_text, history
# REWRITER = "castorini/t5-base-canard"
def generate_response_together_api(history, max_tokens, client, model="meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo"):
together_request = {
"model": model,
"messages": history,
"stream": False,
"logprobs": False,
"stop": ["<eos>", "<unk>", "<sep>", "<pad>", "<cls>", "<mask>"],
"max_tokens": max_tokens,
}
response = client.chat.completions.create(**together_request)
outputs_text = response.choices[0].message.content
history = history + [{"role": "assistant", "content": outputs_text}]
return outputs_text, history
def make_local_api_call(payload, api_url):
try:
# Send the POST request to the API
response = requests.post(api_url, json=payload)
# Check if the request was successful
if response.status_code == 200:
result = response.json()
# Print the generated text
return result.get("text", [""])[0]
# if "logits" in result:
# print(f"Logits: {result['logits']}")
else:
# If there was an error, print the status code and message
print(f"Error: {response.status_code}")
print(response.text)
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
def generate_response_local_api(history, terminator, max_tokens, api_url):
payload = TEMPLATE_PAYLOAD.copy()
payload.update(
{
"prompt": history,
"max_tokens": max_tokens,
"stop_token_ids": terminator,
}
)
# Call the API to generate the response
outputs_text = make_local_api_call(payload, api_url)
if outputs_text:
# Update history with the assistant's response
history = history + [{"role": "assistant", "content": outputs_text}]
return outputs_text, history
else:
print("Failed to generate a response.")
return "Generation failed", history # Return the original history in case of failure
def conversation_window(history, N=100):
if len(history) > N:
return history[2:]
return history
def format_message_history(message, history):
if not history:
str_history = f"\n<user>: {message}\n<assistant>"
else:
# Query written
str_history = (
"".join(["".join(["\n<user>:" + item[0], "\n<assistant>:" + item[1]]) for item in history])
+ f"\n<user>: {message}\n<assistant>"
)
return str_history
def format_user_message(message, history):
return history + [{"role": "user", "content": message}]
def format_context(message, history):
return [{"role": "system", "content": message}] + history
def prepare_tokenizer(tokenizer):
special_tokens = ["<eos>", "<unk>", "<sep>", "<pad>", "<cls>", "<mask>"]
for token in special_tokens:
if tokenizer.convert_tokens_to_ids(token) is None:
tokenizer.add_tokens([token])
if tokenizer.eos_token_id is None:
tokenizer.eos_token_id = tokenizer.convert_tokens_to_ids("<eos>")
terminators = [
tokenizer.eos_token_id,
# self.pipeline.tokenizer.convert_tokens_to_ids(""),
]
return tokenizer, terminators
def gradio_to_huggingface_message(gradio_message):
huggingface_message = []
for user, bot in gradio_message:
huggingface_message.append({"role": "user", "content": user})
huggingface_message.append({"role": "assistant", "content": bot})
return huggingface_message
def huggingface_to_gradio_message(huggingface_message):
gradio_message = []
store = []
for utter in huggingface_message:
if utter["role"] in ["user", "assistant"]:
if utter["role"] == "assistant":
store.append(utter["content"])
gradio_message.append(store)
store = []
else:
store.append(utter["content"])
return gradio_message
def get_personality_instruction(personality):
return PERSONALITY_INSTRUCTION.format(personality)
def get_system_instruction(rag=RAG, personality_list=None):
if rag and personality_list:
return (
SYSTEM_INSTRUCTION
+ RAG_INSTRUCTION
+ get_personality_instruction(construct_big_five_words(personality_list))
)
elif personality_list:
return SYSTEM_INSTRUCTION + get_personality_instruction(construct_big_five_words(personality_list))
elif rag:
return SYSTEM_INSTRUCTION + RAG_INSTRUCTION
else:
return SYSTEM_INSTRUCTION
def format_rag_context(rag_context):
"""
rag_context [{"passage_id": clue_web, "passage_text": "abc"}, ...]
"""
passage_context = "Context: \n"
for passage_rank, info in enumerate(rag_context):
passage_context += f"Passage ID: {info['passage_id']}, Text: {info['passage_text']}\n\n"
return passage_context
|