Spaces:
Running
Running
File size: 3,758 Bytes
7f46a81 |
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 |
import requests
import json
import re
from urllib.parse import quote
def extract_between_tags(text, start_tag, end_tag):
start_index = text.find(start_tag)
end_index = text.find(end_tag, start_index)
return text[start_index+len(start_tag):end_index-len(end_tag)]
class VectaraQuery():
def __init__(self, api_key: str, customer_id: int, corpus_ids: list):
self.customer_id = customer_id
self.corpus_ids = corpus_ids
self.api_key = api_key
def submit_query(self, query_str: str):
corpora_key_list = [{
'customer_id': str(self.customer_id), 'corpus_id': str(corpus_id), 'lexical_interpolation_config': {'lambda': 0.025}
} for corpus_id in self.corpus_ids
]
endpoint = f"https://api.vectara.io/v1/query"
start_tag = "%START_SNIPPET%"
end_tag = "%END_SNIPPET%"
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"customer-id": str(self.customer_id),
"x-api-key": self.api_key,
"grpc-timeout": "60S"
}
body = {
'query': [
{
'query': query_str,
'start': 0,
'numResults': 7,
'corpusKey': corpora_key_list,
'context_config': {
'sentences_before': 3,
'sentences_after': 3,
'start_tag': start_tag,
'end_tag': end_tag,
},
'summary': [
{
'responseLang': 'eng',
'maxSummarizedResults': 7,
}
]
}
]
}
response = requests.post(endpoint, data=json.dumps(body), verify=True, headers=headers)
if response.status_code != 200:
print(f"Query failed with code {response.status_code}, reason {response.reason}, text {response.text}")
return "Sorry, something went wrong in my brain. Please try again later."
res = response.json()
summary = res['responseSet'][0]['summary'][0]['text']
responses = res['responseSet'][0]['response']
docs = res['responseSet'][0]['document']
pattern = r'\[\d{1,2}\]'
matches = [match.span() for match in re.finditer(pattern, summary)]
# figure out unique list of references
refs = []
for match in matches:
start, end = match
response_num = int(summary[start+1:end-1])
doc_num = responses[response_num-1]['documentIndex']
metadata = {item['name']: item['value'] for item in docs[doc_num]['metadata']}
text = extract_between_tags(responses[response_num-1]['text'], start_tag, end_tag)
url = f"{metadata['url']}#:~:text={quote(text)}"
if url not in refs:
refs.append(url)
# replace references with markdown links
refs_dict = {url:(inx+1) for inx,url in enumerate(refs)}
for match in reversed(matches):
start, end = match
response_num = int(summary[start+1:end-1])
doc_num = responses[response_num-1]['documentIndex']
metadata = {item['name']: item['value'] for item in docs[doc_num]['metadata']}
text = extract_between_tags(responses[response_num-1]['text'], start_tag, end_tag)
url = f"{metadata['url']}#:~:text={quote(text)}"
citation_inx = refs_dict[url]
summary = summary[:start] + f'[\[{citation_inx}\]]({url})' + summary[end:]
return summary
|