Spaces:
Sleeping
Sleeping
File size: 8,330 Bytes
a190548 8853856 43f57f4 a190548 8853856 a190548 8853856 a190548 8853856 a190548 8853856 a190548 8853856 a190548 8853856 a190548 8853856 43f57f4 8853856 a190548 8853856 a190548 43f57f4 a190548 43f57f4 8853856 |
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
import requests
from frontend.app import common_functions
import streamlit as st
API_BASE_URL = "http://localhost:8000/knowledge-base"
def upsert_data():
"""
Displays a form to upsert data into the Pinecone database.
Features:
- Users can provide 'Input', 'Output', and 'Instruction'.
- Displays appropriate success or error messages.
- Improved error handling with detailed feedback.
"""
st.subheader("Enter the data to upsert")
with st.form("upsert_form"):
# Input Fields
input_text = st.text_area("Input", placeholder="Enter input text here...", height=150)
output_text = st.text_area("Output", placeholder="Enter output text here...", height=150)
instruction_text = st.text_input("Instruction", placeholder="Provide guidance for the data...")
upsert_submit = st.form_submit_button("🚀 Upsert Data")
if upsert_submit:
# ✅ Validation Check
if not input_text.strip() or not output_text.strip() or not instruction_text.strip():
st.error("❗ All fields are required. Please fill out each section before submitting.")
return
# ✅ Payload Creation
payload = {
"data": [
{
"input": input_text.strip(),
"output": output_text.strip(),
"instruction": instruction_text.strip()
}
]
}
# API Call
with st.spinner("⏳ Processing your data..."):
try:
response = requests.post(f"{API_BASE_URL}/upsert-data", json=payload)
response_data = response.json()
if response.status_code == 200:
st.success(f"✅ Data successfully upserted: {response_data.get('message', 'Success')}")
st.toast("🎉 Upsert successful!")
elif response.status_code == 400:
st.warning(f"⚠️ Bad Request: {response_data.get('detail', 'Check your input data.')}")
elif response.status_code == 500:
st.error("❌ Internal Server Error. Please try again later.")
else:
st.error(f"❗ Unexpected error: {response_data.get('detail', 'Unknown issue occurred.')}")
except requests.exceptions.RequestException as e:
st.error(f"❌ Network error: {e}")
def delete_records():
"""
Displays a form to delete records from the Pinecone database.
Features:
- Users can input comma-separated IDs for deletion.
- Includes validation checks for empty or malformed IDs.
- Enhanced error handling for better user feedback.
"""
st.subheader("Enter id to delete")
with st.form("delete_form"):
ids_input = st.text_area("IDs to Delete", placeholder="Enter IDs separated by commas (e.g., id_123, id_456)")
ids_to_delete = [id.strip() for id in ids_input.split(",") if id.strip()]
delete_submit = st.form_submit_button("🗑️ Delete Records")
if delete_submit:
# ✅ Validation Check
if not ids_to_delete:
st.error("❗ Please provide at least one valid ID.")
return
# 🔒 Confirmation Prompt for Safety
if not st.confirm("Are you sure you want to delete the selected records? This action is irreversible."):
st.info("❗ Deletion canceled.")
return
# ✅ Payload Creation
payload = {"ids_to_delete": ids_to_delete}
# ✅ API Call with Improved Error Handling
with st.spinner("⏳ Deleting records..."):
try:
response = requests.post(f"{API_BASE_URL}/delete-records", json=payload)
response_data = response.json()
if response.status_code == 200:
st.success(f"✅ {response_data.get('message', 'Records successfully deleted.')}")
st.toast("🎯 Deletion successful!")
elif response.status_code == 400:
st.warning(f"⚠️ Bad Request: {response_data.get('detail', 'Check the provided IDs.')}")
elif response.status_code == 404:
st.warning("⚠️ No matching records found. Please verify the provided IDs.")
elif response.status_code == 500:
st.error("❌ Internal Server Error. Please try again later.")
else:
st.error(f"❗ Unexpected error: {response_data.get('detail', 'Unknown issue occurred.')}")
except requests.exceptions.RequestException as e:
st.error(f"❌ Network error: {e}")
def render_metadata_fetch_form():
"""
Renders a form to fetch metadata based on user concerns.
Features:
- Input validation to ensure meaningful data is provided.
- Displays metadata in a visually appealing format.
- Improved error handling with detailed messages.
"""
st.header("📋 Fetch Metadata")
with st.form("fetch_metadata_form"):
# 📝 Input Fields
prompt_text = st.text_area(
"Describe Your Concern",
"e.g., I've been feeling anxious lately.",
help="Provide a brief description of your concern to get relevant metadata."
)
n_result = st.number_input(
"Number of Results",
min_value=1,
value=3,
help="Specify the number of search results you'd like to retrieve."
)
score_threshold = st.number_input(
"Score Threshold",
min_value=0.0,
max_value=1.0,
value=0.47,
help="Set the minimum relevance score for the results. Higher values ensure more accurate matches."
)
metadata_submit = st.form_submit_button("🔍 Fetch Metadata")
if metadata_submit:
# ✅ Input Validation
if not prompt_text.strip():
st.warning("❗ Please provide a valid concern description.")
return
payload = {
"prompt": prompt_text.strip(),
"n_result": n_result,
"score_threshold": score_threshold
}
# 🔄 Enhanced API Request with Better Error Handling
with st.spinner("⏳ Fetching metadata..."):
try:
response = requests.post(f"{API_BASE_URL}/fetch-metadata", json=payload)
response.raise_for_status()
metadata = response.json().get('metadata', [])
# ✅ Display Results
if metadata:
st.success(f"✅ Found {len(metadata)} relevant result(s).")
st.subheader("Search Results")
for index, entry in enumerate(metadata, start=1):
common_functions.typewriter_effect(f"**🧠 Question:** {entry.get('question', 'N/A')}",speed=0)
common_functions.typewriter_effect(f"**💬 Answer:** {entry.get('answer', 'N/A')}",speed=0)
st.markdown(f"**📈 Score:** `{entry.get('score', 'N/A')}`")
st.markdown(f"**🆔 ID:** `{entry.get('id', 'N/A')}`")
else:
st.info("🤔 No relevant data found. Try refining your concern or adjusting the score threshold.")
# Exception Handling for Specific Errors
except requests.exceptions.HTTPError as http_err:
st.error(f"❌ HTTP Error: {http_err}")
except requests.exceptions.ConnectionError:
st.error("❌ Network error. Please check your internet connection.")
except requests.exceptions.Timeout:
st.error("❌ Request timed out. Please try again later.")
except requests.exceptions.RequestException as e:
st.error(f"❌ Unexpected error: {e}")
|