Making chatbot queries by API

1
# Query Chatbot
2
def query_chatbot(data: Dict[Text, Any]) -> Response:
3
  print('Querying chatbot')
4
  response = requests.post(url=url, headers=API_HEADERS, data=json.dumps(data))
5
  return response
6
7
data: Dict[Text, Any] = {}
8
data['query'] = '<CHATBOT-QUERY>' # Define your chatbot query
9
data['model_name'] = '<MODEL-NAME>' # Define model to use, 'multistep-lite' (multistep model) or 'multistep' (research mode)
10
# Set include_source_info to 'true' to get source documents from which the response is fetched
11
data['include_source_info'] = False
12
data['source_app'] = [{'type':'CHATBOT','uuid':'<CHATBOT-ID>'}]
13
14
# Query the chatbot
15
response = query_chatbot(data)
16
17
if response.ok:
18
  resp_data = response.json()
19
  query_id = resp_data.get('query_id')
20
  if not query_id:
21
    print('Unable to query the chatbot.')
22
23
  print(f'Response from query chatbot : {resp_data}')
24
else:
25
  print(f'Unable to connect to the chatbot query API - {response}.')
26
  print(f'Response status code: {response.status_code}')
27
  print(f'Response content: {response.text}')

1
{
2
    "query_id": "f2ef9702-b018-4a45-9b2e-d1fb575b42ed"
3
}

1
def get_query_status(query_id: Text) -> Tuple[Dict, Text]:
2
  print(f'Getting query request status for {query_id}')
3
  while True:
4
    response = requests.get(f'{url}/{query_id}', headers=API_HEADERS)
5
    if not response.ok:
6
        return None, f'Error while getting the status for query request :  {response.status_code} - {response.text}'
7
8
    response_json = response.json()
9
    status = response_json.get('status')
10
    if not status:
11
        return None, f'Getting the status for query request : {response_json}'
12
13
    # If status of query request is COMPLETE stop polling and return
14
    if status in ['COMPLETE']:
15
      return response_json, None
16
17
    print('Query still processing, please wait!')
18
19
    # If it is RUNNING, the script waits for 5 seconds before the next poll
20
    time.sleep(5)
21
22
# Get the status of query request
23
query_id = '<QUERY-ID>' # Define query ID
24
response_json, err = get_query_status(query_id=query_id)
25
if err:
26
  print(f'Error occurred: {err}')
27
else:
28
  print(f'Query request status : {response_json}')

1
{
2
   "query_id":"f2ef9702-b018-4a45-9b2e-d1fb575b42ed",
3
   "results":[
4
      {
5
         "response":"<QUERY-RESPONSE>",
6
         "source_documents":[
7
            {
8
               "name":"file1.pdf",
9
               "pages":[
10
               ]
11
            },
12
            {
13
               "name":"file2.pdf",
14
               "pages":[
15
               ]
16
            }
17
         ]
18
      }
19
   ],
20
   "status":"COMPLETE"
21
}

1
import time, json
2
from typing import Dict, Text, Tuple, Any
3
import requests
4
from requests import Response
5
6
API_TOKEN = '<API-TOKEN>' # Define your API token
7
IB_CONTEXT = '<ORGANIZATION-ID or USER-ID>' # Define organization ID to use organization account, define user ID to use community account
8
9
API_BASE_URL = 'https://aihub.instabase.com/api' # Edit root API URL if your organization has a custom domain
10
url = API_BASE_URL + 'v2/queries'
11
API_HEADERS = {
12
    'Authorization': f'Bearer {API_TOKEN}',
13
    'IB-Context': f'{IB_CONTEXT}'
14
}
15
16
def query_chatbot(data: Dict[Text, Any]) -> Response:
17
  print('Querying chatbot')
18
  response = requests.post(url=url, headers=API_HEADERS, data=json.dumps(data))
19
  return response
20
21
def get_query_status(query_id: Text) -> Tuple[Dict, Text]:
22
  print(f'Getting query request status for {query_id}')
23
  while True:
24
    response = requests.get(f'{url}/{query_id}', headers=API_HEADERS)
25
    if not response.ok:
26
        return None, f'Error while getting the status for query request :  {response.status_code} - {response.text}'
27
28
    response_json = response.json()
29
    status = response_json.get('status')
30
    if not status:
31
        return None, f'Getting the status for query request : {response_json}'
32
33
    # If status of query request is COMPLETE stop polling and return
34
    if status in ['COMPLETE']:
35
      return response_json, None
36
37
    print('Query still processing, please wait!')
38
39
    # If it is RUNNING, the script waits for 5 seconds before the next poll
40
    time.sleep(5)
41
42
data: Dict[Text, Any] = {}
43
data['query'] = '<CHATBOT-QUERY>' # Define your chatbot query
44
data['model_name'] = '<MODEL-NAME>' # Define model to use, 'multistep-lite' (multistep model) or 'multistep' (research mode)
45
data['include_source_info'] = False # Set to 'true' to get information about source documents referenced when generating response
46
data['source_app'] = [{'type':'CHATBOT','uuid':'<CHATBOT-ID>'}] # Define chatbot ID
47
48
# Query the chatbot
49
response = query_chatbot(data)
50
query_id = None
51
if response.ok:
52
  resp_data = response.json()
53
  query_id = resp_data.get('query_id')
54
  if not query_id:
55
    print('Unable to query the chatbot.')
56
57
  print(f'Response from query chatbot : {resp_data}')
58
else:
59
  print(f'Unable to connect to the chatbot query API - {response}.')
60
  print(f'Response status code: {response.status_code}')
61
  print(f'Response content: {response.text}')
62
63
64
# Get the status of query request
65
if query_id:
66
  response_json, err = get_query_status(query_id=query_id)
67
  if err:
68
    print(f'Error occurred: {err}')
69
  else:
70
    print(f'Query request status : {response_json}')