app.py

"""gradio app for Chat Interface for DataStax Langflow calls"""

import json
from typing import Optional, Sequence, Tuple
from uuid import uuid4
import requests
import gradio as gr
import os

BASE_API_URL = "https://api.langflow.astra.datastax.com"
LANGFLOW_ID = "e99b525a-84dd-4cf2-bac6-8e5ebc6a7d17"
FLOW_ID = "bfbf7237-50dd-40dd-baba-cc680288d788"
APPLICATION_TOKEN = os.getenv('DATASTAX_TOKEN')

ENDPOINT = 'ai_traveller'
# SESSION_ID = str(uuid4())

TWEAKS = {
    "Memory-fCuCs": {
    "n_messages": 100,
    "order": "Ascending",
    "sender": "Machine and User",
    "sender_name": "",
    "template": "{sender_name}: {text}"
  },
    "ChatInput-PN5E4": {},
    "Prompt-vGvVG": {},
    "GoogleGenerativeAIModel-6n0Ft": {},
    "ChatOutput-HjfF1": {}
}


def call_travel_ai(
    message: str,
    history: Sequence[Tuple[str, str]],
    session_id: str,
    endpoint: str = ENDPOINT,
    output_type: str = "chat",
    input_type: str = "chat",
    tweaks: Optional[dict] = None,
    application_token: Optional[str] = APPLICATION_TOKEN
) -> dict:
    """
    Run a flow with a given message and optional tweaks.

    :param message: The message to send to the flow
    :param endpoint: The ID or the endpoint name of the flow
    :param tweaks: Optional tweaks to customize the flow
    :return: The JSON response from the flow
    """
    api_url = f"{BASE_API_URL}/lf/{LANGFLOW_ID}/api/v1/run/{endpoint}"

    payload = {
        'session_id': session_id,
        "input_value": message,
        "output_type": output_type,
        "input_type": input_type,
    }
    headers = None
    if tweaks:
        payload["tweaks"] = tweaks
    if application_token:
        headers = {
            "Authorization": "Bearer " + application_token,
            "Content-Type": "application/json"
        }
    response = requests.post(
        api_url,
        json=payload,
        headers=headers,
        timeout=360
    )
    response = response.json()
    resp_message = ''
    for resp in response['outputs']:
        for _resp in resp['outputs']:
            for message in _resp['messages']:
                resp_message = message['message']

    return resp_message

def generate_session_id():
    return str(uuid4())

def render_value(value):
    return value

with gr.Blocks() as demo:
    session_id = gr.Textbox(
        value=generate_session_id,
        visible=False,
        interactive=False,
        label="Session ID",
    )

    chat_interface = gr.ChatInterface(
        call_travel_ai,
        type='messages',
        title=f"AI Travel Partner",
        additional_inputs=[session_id],
        autofocus=True,
        fill_height=True,
    )

demo.launch(share=False, debug=True,)