Make chatbot queries by API

1
import json
2
import time
3
4
import requests  # third-party module for making HTTP requests
5
6
# define URLs
7
api_base_url = <API-BASE-URL>
8
queries_endpoint_url = f'{api_base_url}/v2/queries'
9
10
# include authentication credentials as request headers
11
api_token = <API-TOKEN>
12
ib_context = <IB-CONTEXT>
13
14
api_headers = {
15
    'Authorization': f'Bearer {api_token}',
16
    'IB-Context': ib_context
17
}
18
19
# provide the chatbot ID
20
source_app = {'type': 'CHATBOT',
21
              'id':   <CHATBOT-ID>}
22
23
# create payload for POST to 'queries' endpoint
24
data = {'query': <QUERY>,
25
        'model_name': <MODEL-NAME>,
26
        'include_source_info': <INCLUDE-SOURCE-INFO>,
27
        'source_app': source_app}
28
29
# send the query to the chatbot and capture the 'query_id'
30
response = requests.post(url=queries_endpoint_url,
31
                         headers=api_headers,
32
                         data=json.dumps(data))
33
resp_data = response.json()
34
query_id = resp_data.get('query_id')
35
36
# get the status of the query
37
while True:
38
    response = requests.get(url=f'{queries_endpoint_url}/{query_id}',
39
                            headers=api_headers)
40
    response_json = response.json()
41
    status = response_json.get('status')
42
    if status == 'COMPLETE':
43
        break
44
    time.sleep(5)
45
46
# print the query's reply and its sources
47
results = response_json['results']
48
for result in results:
49
    reply = result['response']
50
    print(f"REPLY\n{reply}\n")
51
52
    sources = result['source_documents']
53
    print("SOURCE DOCUMENTS")
54
    for source in sources:
55
        print(f"{source['name']}")
56
        pages = source['pages']
57
        for page in pages:
58
            page = int(page['page_number'])  # convert from float
59
            print(f"  page {page}")

1
import json
2
import sys
3
import time
4
from typing import Any, Final, Literal, Optional
5
6
import requests  # third-party module for making HTTP requests
7
from requests import Response
8
9
# define URLs
10
API_BASE_URL: Final[str] = <API-BASE-URL>
11
QUERIES_ENDPOINT_URL: Final[str] = f'{API_BASE_URL}/v2/queries'
12
13
# how long to wait for the Chatbot to process the query
14
# and generate a reply
15
PROCESSING_TIMEOUT_SECS: Final[int] = <PROCESSING-TIMEOUT-SECS>
16
17
# how long to wait for an HTTP response to each HTTP request
18
RESPONSE_TIMEOUT_SECS: Final[int] = <RESPONSE-TIMEOUT-SECS>
19
20
# how long to wait between query processing status checks
21
STATUS_POLLING_INTERVAL_SECS: Final[int] = <STATUS-POLLING-INTERVAL-SECS>
22
23
# put authentication credentials in a header for API calls
24
API_TOKEN: Final[str] = <API-TOKEN>
25
IB_CONTEXT: Final[str] = <IB-CONTEXT>
26
API_HEADERS: Final[dict[str, str]] = {
27
    'Authorization': f'Bearer {API_TOKEN}',
28
    'IB-Context': IB_CONTEXT
29
}
30
31
# provide the chatbot ID
32
SOURCE_APP: Final[dict[str, str]] = {'type': 'CHATBOT',
33
                                     'id':   <CHATBOT-ID>}
34
35
# create payload for POST to 'queries' endpoint
36
data: dict[str, Any] = {'query': <QUERY>,
37
'model_name': <MODEL-NAME>,
38
'include_source_info': <INCLUDE-SOURCE-INFO>,
39
'source_app': SOURCE_APP}
40
41
# send the query to the chatbot and capture the 'query_id'
42
try:
43
    response: Response = requests.post(url=QUERIES_ENDPOINT_URL,
44
                                       headers=API_HEADERS,
45
                                       data=json.dumps(data),
46
                                       timeout=RESPONSE_TIMEOUT_SECS)
47
except requests.exceptions.ConnectionError as conn_err:
48
    sys.exit(f"Error: Couldn't submit request to API: {conn_err}")
49
except requests.exceptions.Timeout as timeout_err:
50
    sys.exit(f"Error: Request to API timed out: {timeout_err}")
51
52
if not response.ok:
53
    sys.exit("Error: Bad response from API. "
54
             f"Response status code: {response.status_code}. "
55
             f"Response content: {response.text}")
56
57
response_json: dict[str, Any] = response.json()
58
query_id: Optional[str] = response_json.get('query_id')
59
60
if not query_id:
61
    sys.exit("Error: Chatbot didn't return a query_id")
62
63
# get the query status and check for chatbot processing timeout
64
elapsed_time_secs: int = 0
65
while elapsed_time_secs < PROCESSING_TIMEOUT_SECS:
66
    try:
67
        response = requests.get(url=f'{QUERIES_ENDPOINT_URL}/{query_id}',
68
                                headers=API_HEADERS)
69
    except requests.exceptions.ConnectionError as conn_err:
70
        sys.exit(f"Error: Couldn't submit request to API: {conn_err}")
71
    except requests.exceptions.Timeout as timeout_err:
72
        sys.exit(f"Error: Request to API timed out: {timeout_err}")
73
    except Exception as err:
74
        sys.exit(f"Error: Request to API failed: {err}")
75
76
    # check if we got back an expected status code
77
    if response.ok:
78
        response_json = response.json()
79
    else:
80
        sys.exit("Error: Bad response from API. "
81
                 f"Response status code: {response.status_code}. "
82
                 f"Response content: {response.text}")
83
84
    # stop polling once the processing status is 'COMPLETE'
85
    status: Optional[Literal['COMPLETE', 'FAILED', 'RUNNING']] = response_json.get('status')
86
    if status == 'COMPLETE':
87
        break
88
    time.sleep(STATUS_POLLING_INTERVAL_SECS)
89
    elapsed_time_secs += STATUS_POLLING_INTERVAL_SECS
90
91
# handle processing timeout
92
if elapsed_time_secs >= PROCESSING_TIMEOUT_SECS:
93
    sys.exit(f"Error: Chatbot didn't finish processing the query "
94
             f"within {PROCESSING_TIMEOUT_SECS} seconds")
95
96
# Print the reply and its sources. Assume the API returns a well-
97
# formed reply, so no need for error handling when parsing the reply
98
results: list[dict] = response_json['results']
99
result: dict[str, Any]
100
for result in results:
101
    reply: str = result['response']
102
    print(f"REPLY\n{reply}\n")
103
    sources: list[dict[str, Any]] = result['source_documents']
104
    if sources:
105
        print("SOURCE DOCUMENTS")
106
        source: dict[str, Any]
107
        for source in sources:
108
            print(f"{source['name']}")
109
            pages: list[dict] = source['pages']
110
            page: dict[str, str | float]
111
            for page in pages:
112
                page_num: int = int(page['page_number'])
113
                print(f"  page {page_num}")