| import ast
|
| import gradio as gr
|
| from functions import example_question_generator, chatbot_func
|
| from data_sources import connect_graphql
|
| from utils import message_dict
|
|
|
| import os
|
| from dotenv import load_dotenv
|
|
|
| load_dotenv()
|
|
|
| graphql_sample_endpoint = os.getenv("GRAPHQL_SAMPLE_ENDPOINT")
|
| graphql_sample_api_token = os.getenv("GRAPHQL_SAMPLE_API_TOKEN")
|
| graphql_sample_header_name = os.getenv("GRAPHQL_SAMPLE_HEADER_NAME")
|
|
|
| def hide_info():
|
| return gr.update(visible=False)
|
|
|
| with gr.Blocks() as demo:
|
| description = gr.HTML("""
|
| <!-- Header -->
|
| <div class="max-w-4xl mx-auto mb-12 text-center">
|
| <div class="bg-blue-50 border border-blue-200 rounded-lg max-w-2xl mx-auto">
|
| <p>This tool allows users to communicate with and query real time data from a GraphQL API endpoint using natural
|
| language and the above features.</p>
|
| <p style="font-weight:bold;">Notice: API querying is the most difficult and experimental feature so far.
|
| This tool may have variable performance and quality, although it should get better over time as I evaluate use.
|
| No login information is retained and credentials are passed as session variables until the user leaves or
|
| refreshes the page in which they disappear. They are never saved to any files.</p>
|
| <p style="font-weight:bold;"> I don't include a function that allows the system to run mutations and I instruct the agent to not alter any data, but it could in theory be possible,
|
| although my testing wasn't able to get the system to alter or write to the api. I would be careful to make sure permissions are restricted for the
|
| api token being used.
|
| And of course, it's probably best to use caution when connecting to a strange AI tool with an unfamiliar author.
|
| This should be for demonstration purposes.</p>
|
| <p>Contact me if this is something you would like built in your organization, on your infrastructure, and with the requisite privacy and control a production
|
| database analytics tool requires.</p>
|
| </div>
|
| </div>
|
| """, elem_classes="description_component")
|
|
|
| graphql_url = gr.Textbox(label="GraphQL Endpoint URL", value=graphql_sample_endpoint)
|
| with gr.Row():
|
| api_token_header_name = gr.Textbox(label="API Token Header Name", value=graphql_sample_header_name)
|
| api_token = gr.Textbox(label="API Token", value=graphql_sample_api_token, type="password")
|
|
|
| submit = gr.Button(value="Submit")
|
| submit.click(fn=hide_info, outputs=description)
|
|
|
| @gr.render(inputs=[graphql_url,api_token,api_token_header_name], triggers=[submit.click])
|
| def api_chat(request: gr.Request, graphql_url=graphql_url.value, api_token=api_token.value, api_token_header_name=api_token_header_name.value):
|
| if request.session_hash not in message_dict:
|
| message_dict[request.session_hash] = {}
|
| message_dict[request.session_hash]['graphql'] = None
|
| if graphql_url:
|
| print("GraphQL API")
|
| process_message = process_graphql(graphql_url, api_token, api_token_header_name, request.session_hash)
|
| gr.HTML(value=process_message[1], padding=False)
|
| if process_message[0] == "success":
|
| if "qdl-app-testing" in graphql_url:
|
| example_questions = [
|
| ["Describe the dataset"],
|
| ["What is the total revenue for this shopify store?"],
|
| ["What is the average duration from the fulfillment of an order to its delivery?"],
|
| ["What is the total value of orders processed in the current month?"],
|
| ["Which product has the highest number of variants in the inventory?"],
|
| ["How many gift cards have been issued this year, and what is their total value?"],
|
| ["How many active apps are currently installed on the store?"],
|
| ["What is the total count of abandoned checkouts over the last month?"]
|
| ]
|
| else:
|
| try:
|
| generated_examples = ast.literal_eval(example_question_generator(request.session_hash, 'graphql', graphql_url, process_message[2], ''))
|
| example_questions = [
|
| ["Describe the dataset"]
|
| ]
|
| for example in generated_examples:
|
| example_questions.append([example])
|
| except Exception as e:
|
| print("GRAPHQL QUESTION GENERATION ERROR")
|
| print(e)
|
| example_questions = [
|
| ["Describe the dataset"],
|
| ["List the columns in the dataset"],
|
| ["What could this data be used for?"],
|
| ]
|
| session_hash = gr.Textbox(visible=False, value=request.session_hash)
|
| graphql_api_string = gr.Textbox(visible=False, value=graphql_url)
|
| graphql_api_token = gr.Textbox(visible=False, value=api_token)
|
| graphql_token_header = gr.Textbox(visible=False, value=api_token_header_name)
|
| titles = gr.Textbox(value=process_message[2], interactive=False, label="GraphQL Types")
|
| data_source = gr.Textbox(visible=False, value='graphql')
|
| schema = gr.Textbox(visible=False, value='')
|
| bot = gr.Chatbot(type='messages', label="GraphQL Chat Window", render_markdown=True, sanitize_html=False, show_label=True, render=False, visible=True, elem_classes="chatbot")
|
| chat = gr.ChatInterface(
|
| fn=chatbot_func,
|
| type='messages',
|
| chatbot=bot,
|
| title="Chat with your Graphql API",
|
| examples=example_questions,
|
| concurrency_limit=None,
|
| additional_inputs=[session_hash, data_source, titles, schema, graphql_api_string, graphql_api_token, graphql_token_header]
|
| )
|
|
|
| def process_graphql(graphql_url, api_token, api_token_header_name, session_hash):
|
| if graphql_url:
|
| process_message = connect_graphql(graphql_url, api_token, api_token_header_name, session_hash)
|
| return process_message
|
|
|
| if __name__ == "__main__":
|
| demo.launch() |
