app.py

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

def query_vectara(question, chat_history, uploaded_file=None):
    # Handle file upload to Vectara if a file is provided
    if uploaded_file is not None:
        customer_id = config('CUSTOMER_ID')  # Read from .env file
        corpus_id = config('CORPUS_ID')  # Read from .env file
        api_key = config('API_KEY')  # Read from .env file
        url = f"https://api.vectara.io/v1/upload?c={customer_id}&o={corpus_id}"

        post_headers = {
            "x-api-key": api_key,
            "customer-id": customer_id
        }

        files = {
            "file": (uploaded_file.name, uploaded_file),
            "doc_metadata": (None, json.dumps({"metadata_key": "metadata_value"})),  # Replace with your metadata
        }
        response = requests.post(url, files=files, headers=post_headers)

        if response.status_code == 200:
            upload_status = "File uploaded successfully"
        else:
            upload_status = "Failed to upload the file"
    else:
        upload_status = "No file uploaded"

    # Get the user's message from the chat history
    user_message = chat_history[-1][0]

    # Query Vectara API
    query_url = "https://api.vectara.io/v1/query/v1/query"

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}",
        "customer-id": customer_id,
    }

    query_body = {
        "query": [
            {
                "query": user_message,
                "queryContext": "",
                "start": 0,
                "numResults": 10,
                "contextConfig": {
                    "charsBefore": 0,
                    "charsAfter": 0,
                    "sentencesBefore": 2,
                    "sentencesAfter": 2,
                    "startTag": "%START_SNIPPET%",
                    "endTag": "%END_SNIPPET%",
                },
                "rerankingConfig": {
                    "rerankerId": 272725718,
                    "mmrConfig": {
                        "diversityBias": 0.3
                    }
                },
                "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"
                    }
                ]
            }
        ]
    }

    query_response = requests.post(query_url, json=query_body, headers=headers)

    if query_response.status_code == 200:
        query_data = query_response.json()
        response_message = f"{upload_status}\n\nResponse from Vectara API: {json.dumps(query_data, indent=2)}"
    else:
        response_message = f"{upload_status}\n\nError: {query_response.status_code}"

    return response_message

# Create a Gradio ChatInterface with a text input and an optional file upload input
iface = gr.Interface(
    fn=query_vectara,
    inputs=[gr.Textbox(label="Input Text"), gr.File(label="Upload a file")],
    outputs=gr.Textbox(label="Output Text"),
    title="Vectara Chatbot",
    description="Ask me anything using the Vectara API!"
)

iface.launch()