app.py

import gradio as gr
import requests
import json
from decouple import Config

config = Config('.env')


def query_vectara(question):
    user_message = question

    # Read authentication parameters from the .env file
    CUSTOMER_ID = config('CUSTOMER_ID')
    CORPUS_ID = config('CORPUS_ID')
    API_KEY = config('API_KEY')

    # Define the headers
    api_key_header = {
        "customer-id": CUSTOMER_ID,
        "x-api-key": API_KEY
    }

    # Define the request body in the structure provided in the example
    request_body = {
        "query": [
            {
                "query": user_message,
                "queryContext": "",
                "start": 1,
                "numResults": 10,
                "contextConfig": {
                    "charsBefore": 0,
                    "charsAfter": 0,
                    "sentencesBefore": 2,
                    "sentencesAfter": 2,
                    "startTag": "%START_SNIPPET%",
                    "endTag": "%END_SNIPPET%",
                },
                "rerankingConfig": {
                    "rerankerId": 272725718,
                    "mmrConfig": {
                        "diversityBias": 0.27
                    }
                },
                "corpusKey": [
                    {
                        "customerId": CUSTOMER_ID,
                        "corpusId": CORPUS_ID,
                        "semantics": 0,
                        "metadataFilter": "",
                        "lexicalInterpolationConfig": {
                            "lambda": 0
                        },
                        "dim": []
                    }
                ],
                "summary": [
                    {
                        "maxSummarizedResults": 5,
                        "responseLang": "eng",
                        "summarizerPromptName": "vectara-summary-ext-v1.2.0"
                    }
                ]
            }
        ]
    }

    # Make the API request using Gradio
    response = requests.post(
        "https://api.vectara.io/v1/query",
        json=request_body,  # Use json to automatically serialize the request body
        verify=True,
        headers=api_key_header
    )
  
    if response.status_code == 200:  
        query_data = response.json()  
        print(query_data)  
        if query_data:  
            sources_info = []  
              
            # Iterate over all response sets  
            for response_set in query_data.get('responseSet', []):  
                # Extract sources  
                for source in response_set.get('response', []):  
                    source_metadata = source.get('metadata', [])  
                    source_info = {}  
  
                    for metadata in source_metadata:  
                        metadata_name = metadata.get('name', '')  
                        metadata_value = metadata.get('value', '')  
  
                        if metadata_name == 'title':  
                            source_info['title'] = metadata_value  
                        elif metadata_name == 'author':  
                            source_info['author'] = metadata_value  
                        elif metadata_name == 'pageNumber':  
                            source_info['page number'] = metadata_value  
  
                    if source_info:  
                        sources_info.append(source_info)  
              
            return f"Sources:\n{json.dumps(sources_info, indent=2)}"  
        else:  
            return "No data found in the response."  
    else:  
        return f"Error: {response.status_code}"  


def convert_to_markdown(vectara_response):
    if vectara_response:
        response_set = vectara_response.get('responseSet', [{}])[0]
        summary = response_set.get('summary', [{}])[0]
        summary_text = summary.get('text', 'No summary available')
        sources = response_set.get('response', [])[:5]
        sources_text = [source.get('text', '') for source in sources]

        # Format the summary as Markdown
        markdown_summary = f'**Summary:** {summary_text}\n\n'

        # Format the sources as a numbered list
        markdown_sources = "\n".join([f"{i+1}. {source}" for i, source in enumerate(sources_text)])

        return f"{markdown_summary}**Sources:**\n{markdown_sources}"
    else:
        return "No data found in the response."


iface = gr.Interface(
    fn=query_vectara,
    inputs=[gr.Textbox(label="Input Text")],
    outputs=[gr.Markdown(label="Output Text")],
    title="Vectara Chatbot",
    description="Ask me anything using the Vectara API!"
)

iface.launch()