Spaces:
Runtime error
Runtime error
File size: 7,366 Bytes
75df934 |
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 |
# my_app/model_manager.py
import google.generativeai as genai
import chat.arxiv_bot.arxiv_bot_utils as utils
import json
model = None
def create_model():
with open("apikey.txt","r") as apikey:
key = apikey.readline()
genai.configure(api_key=key)
for m in genai.list_models():
if 'generateContent' in m.supported_generation_methods:
print(m.name)
print("He was there")
config = genai.GenerationConfig(max_output_tokens=2048,
temperature=0.7)
safety_settings = [
{
"category": "HARM_CATEGORY_DANGEROUS",
"threshold": "BLOCK_NONE",
},
{
"category": "HARM_CATEGORY_HARASSMENT",
"threshold": "BLOCK_NONE",
},
{
"category": "HARM_CATEGORY_HATE_SPEECH",
"threshold": "BLOCK_NONE",
},
{
"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
"threshold": "BLOCK_NONE",
},
{
"category": "HARM_CATEGORY_DANGEROUS_CONTENT",
"threshold": "BLOCK_NONE",
},
]
global model
model = genai.GenerativeModel("gemini-pro",
generation_config=config,
safety_settings=safety_settings)
return model
def get_model():
global model
if model is None:
# Khởi tạo model ở đây
model = create_model() # Giả sử create_model là hàm tạo model của bạn
return model
def extract_keyword_prompt(query):
"""A prompt that return a JSON block as arguments for querying database"""
prompt = (
"""[INST] SYSTEM: You are an assistant that choose only one action below based on guest question.
1. If the guest question is asking for a single specific document or article with explicit title, you need to respond the information in JSON format with 2 keys "title", "author" if found any above. The authors are separated with the word 'and'.
2. If the guest question is asking for relevant informations about a topic, you need to respond the information in JSON format with 2 keys "keywords", "description", include a list of keywords represent the main academic topic, \
and a description about the main topic. You may paraphrase the keywords to add more. \
3. If the guest is not asking for any informations or documents, you need to respond with a polite answer in JSON format with 1 key "answer".
QUESTION: '{query}'
[/INST]
ANSWER:
"""
).format(query=query)
return prompt
def make_answer_prompt(input, contexts):
"""A prompt that return the final answer, based on the queried context"""
prompt = (
"""[INST] You are a library assistant that help to search articles and documents based on user's question.
From guest's question, you have found some records and documents that may help. Now you need to answer the guest with the information found.
If no information found in the database, you may generate some other recommendation related to user's question using your own knowledge. Each article or paper must have a link to the pdf download page.
You should answer in a conversational form politely.
QUESTION: '{input}'
INFORMATION: '{contexts}'
[/INST]
ANSWER:
"""
).format(input=input, contexts=contexts)
return prompt
def response(args):
"""Create response context, based on input arguments"""
keys = list(dict.keys(args))
if "answer" in keys:
return args['answer'], None # trả lời trực tiếp
if "keywords" in keys:
# perform query
query_texts = args["description"]
keywords = args["keywords"]
results = utils.db.query_relevant(keywords=keywords, query_texts=query_texts)
# print(results)
ids = results['metadatas'][0]
if len(ids) == 0:
# go crawl some
new_records = utils.crawl_arxiv(keyword_list=keywords, max_results=10)
print("Got new records: ",len(new_records))
if type(new_records) == str:
return "Error occured, information not found", new_records
utils.db.add(new_records)
utils.sqldb.add(new_records)
results = utils.db.query_relevant(keywords=keywords, query_texts=query_texts)
ids = results['metadatas'][0]
print("Re-queried on chromadb, results: ",ids)
paper_id = [id['paper_id'] for id in ids]
paper_info = utils.sqldb.query_id(paper_id)
print(paper_info)
records = [] # get title (2), author (3), link (6)
result_string = ""
if paper_info:
for i in range(len(paper_info)):
result_string += "Title: {}, Author: {}, Link: {}".format(paper_info[i][2],paper_info[i][3],paper_info[i][6])
records.append([paper_info[i][2],paper_info[i][3],paper_info[i][6]])
return result_string, records
else:
return "Information not found", "Information not found"
# invoke llm and return result
if "title" in keys:
title = args['title']
authors = utils.authors_str_to_list(args['author'])
paper_info = utils.sqldb.query(title = title,author = authors)
# if query not found then go crawl brh
# print(paper_info)
if len(paper_info) == 0:
new_records = utils.crawl_exact_paper(title=title,author=authors)
print("Got new records: ",len(new_records))
if type(new_records) == str:
# print(new_records)
return "Error occured, information not found", "Information not found"
utils.db.add(new_records)
utils.sqldb.add(new_records)
paper_info = utils.sqldb.query(title = title,author = authors)
print("Re-queried on chromadb, results: ",paper_info)
# -------------------------------------
records = [] # get title (2), author (3), link (6)
result_string = ""
for i in range(len(paper_info)):
result_string += "Title: {}, Author: {}, Link: {}".format(paper_info[i][2],paper_info[i][3],paper_info[i][6])
records.append([paper_info[i][2],paper_info[i][3],paper_info[i][6]])
# process results:
if len(result_string) == 0:
return "Information not found", "Information not found"
return result_string, records
# invoke llm and return result
def full_chain_single_question(input_prompt):
try:
first_prompt = extract_keyword_prompt(input_prompt)
temp_answer = model.generate_content(first_prompt).text
args = json.loads(utils.trimming(temp_answer))
contexts, results = response(args)
if not results:
# print(contexts)
return "Random question, direct return", contexts
else:
output_prompt = make_answer_prompt(input_prompt,contexts)
answer = model.generate_content(output_prompt).text
return temp_answer, answer
except Exception as e:
# print(e)
return temp_answer, "Error occured: " + str(e)
|