Spaces:
Sleeping
Sleeping
File size: 3,309 Bytes
72dab5c 9f952c3 72dab5c fc24275 72dab5c b2d74f7 72dab5c b2d74f7 9f952c3 72dab5c ab70608 8bdfec4 72dab5c 9f952c3 66570ef 13cc9d4 fbae619 13cc9d4 66570ef 72dab5c 66570ef 18cec4a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | import gradio as gr
import requests
import os
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from ping import add_ping_route
# Configuration
LANGFLOW_API_URL = os.environ.get("LANGFLOW_API_URL", "")
LANGFLOW_API_KEY = os.environ.get("LANGFLOW_API_KEY", "")
HF_API_KEY = os.environ.get("HF_API_KEY", "")
if not LANGFLOW_API_URL:
print("FATAL: LANGFLOW_API_URL secret not found or is empty.")
else:
print("SUCCESS: LANGFLOW_API_URL loaded securely.")
if not LANGFLOW_API_KEY:
print("FATAL: LANGFLOW_API_KEY secret not found or is empty.")
else:
print("SUCCESS: LANGFLOW_API_KEY loaded securely.")
if not HF_API_KEY:
print("FATAL: HF_API_KEY secret not found or is empty.")
else:
print("SUCCESS: HF_API_KEY loaded securely.")
# Function that calls the LangFlow chat
def call_langflow(message, history):
"""
Call Langflow API and return the response
"""
headers = {
"Content-Type": "application/json",
}
# Add API keys
if HF_API_KEY:
headers["Authorization"] = f"Bearer {HF_API_KEY}"
if LANGFLOW_API_KEY:
headers["x-api-key"] = f"{LANGFLOW_API_KEY}"
# Adjust this payload based on your Langflow API structure
payload = {
"input_value": message,
"output_type": "chat",
"input_type": "chat",
"tweaks": {}
}
try:
response = requests.post(
LANGFLOW_API_URL,
json=payload,
headers=headers,
timeout=30
)
response.raise_for_status()
# Parse response - adjust based on your API response structure
data = response.json()
# Common Langflow response structures:
# Option 1: data["outputs"][0]["outputs"][0]["results"]["message"]["text"]
# Option 2: data["result"]["message"]
# Adjust the following line based on your actual response:
bot_message = data["outputs"][0]["outputs"][0]["results"]["message"]["text"]
return bot_message
except requests.exceptions.RequestException as e:
return f"Error connecting to Langflow: {str(e)}"
except (KeyError, IndexError) as e:
return f"Error parsing response: {str(e)}\nResponse: {data}"
#Create FastAPI App for health check.
app = FastAPI(title="Chatbot with Ping API")
# Optional CORS setup
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Add /ping route from ping.py
add_ping_route(app, call_langflow)
# Create Gradio Chat Interface with custom CSS
with gr.Blocks(
css="""
footer {display: none !important;}
.footer {display: none !important;}
#footer {display: none !important;}
.svelte-1p1dq6v {display: none !important;}
"""
) as demo:
gr.ChatInterface(
fn=call_langflow,
title="CPA Chatbot POC",
description="You can use this chatbot to answer questions related to Crown Pointe Academy policies.",
examples=["Summarize the school's uniform policy", "Can a student wear earrings?"],
retry_btn=None,
undo_btn=None,
)
# Mount Gradio app to FastAPI at root path
app = gr.mount_gradio_app(app, demo, path="/") |