Create app.py

app.py

@@ -0,0 +1,104 @@
+"""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'
+
+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,)