Spaces:
Sleeping
Sleeping
File size: 4,519 Bytes
d44b06c 8b55673 78f71d4 8b55673 f37954c 20d91c3 8b55673 d44b06c 8b55673 d44b06c 8b55673 78f71d4 8b55673 d44b06c 8b55673 d44b06c 8b55673 d44b06c 8b55673 d44b06c 8b55673 d44b06c 8b55673 d44b06c 8b55673 | 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 | import streamlit as st
import requests
import os
import json
from dotenv import load_dotenv
import logging
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
# Configurar logging
logging.basicConfig(level=logging.INFO)
# Cargar variables de entorno
load_dotenv()
# Configuraci贸n de la API de Flowise
#API_URL = os.getenv("API_URL", "https://livanarzuaga-test.hf.space/api/v1/prediction/2a2e535d-1807-4701-8468-c0ede26bcdbf")
API_URL = os.getenv("API_URL", "https://wilkie-hf-space-tuto.hf.space/api/v1/prediction/037dd3d4-8492-45c4-9c5f-6ff84b8e95ee")
session = requests.Session()
# Configurar reintentos
retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
def query(payload, api_url=API_URL):
try:
response = session.post(api_url, json=payload)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as http_err:
logging.error(f"HTTP error occurred: {http_err}")
return {"error": "HTTP error occurred. Please try again later."}
except requests.exceptions.ConnectionError as conn_err:
logging.error(f"Connection error occurred: {conn_err}")
return {"error": "Connection error occurred. Please try again later."}
except requests.exceptions.Timeout as timeout_err:
logging.error(f"Timeout error occurred: {timeout_err}")
return {"error": "Timeout error occurred. Please try again later."}
except requests.exceptions.RequestException as req_err:
logging.error(f"An error occurred: {req_err}")
return {"error": "An error occurred. Please try again later."}
def validate_input(user_input):
if not user_input or len(user_input.strip()) == 0:
return False, "Input cannot be empty."
return True, ""
def get_response(user_input, history, api_url=API_URL):
is_valid, error_message = validate_input(user_input)
if not is_valid:
return error_message
# Incluir el historial completo en el payload
payload = {"question": user_input, "history": history, "overrideConfig": {"sessionId": "user1"}}
response = query(payload, api_url)
return response.get('text', "Error: No API response.")
def export_conversation(past, generated, file_path="conversation.json"):
try:
conversation = {
"past": past,
"generated": generated
}
with open(file_path, "w") as f:
json.dump(conversation, f)
return True
except Exception as e:
logging.error(f"Error exporting conversation: {e}")
return str(e)
# Inicializar claves en st.session_state
if 'history' not in st.session_state:
st.session_state['history'] = []
if 'generated' not in st.session_state:
st.session_state['generated'] = ["Hi I can answer your weather related questions"]
if 'past' not in st.session_state:
st.session_state['past'] = ["Hello"]
# Sidebar con opciones
st.sidebar.title("Opciones")
if st.sidebar.button("Nueva conversaci贸n"):
st.session_state['past'] = ["Hello"]
st.session_state['generated'] = ["Hi I can answer your weather related questions"]
st.session_state['history'] = []
st.sidebar.success("Nueva conversaci贸n iniciada")
if st.sidebar.button("Exportar conversaci贸n"):
result = export_conversation(st.session_state['past'], st.session_state['generated'])
if result is True:
st.sidebar.success("Conversaci贸n exportada como conversation.json")
else:
st.sidebar.error(f"Error al exportar la conversaci贸n: {result}")
# Contenedor para el historial del chat
chat_container = st.container()
# Contenedor para la entrada de texto del usuario
with st.form(key='my_form', clear_on_submit=True):
user_input = st.text_input("Chat:", placeholder="Type here to chat", key='input')
submit_button = st.form_submit_button(label='Send')
if submit_button and user_input:
output = get_response(user_input, st.session_state['history'])
st.session_state['past'].append(user_input)
st.session_state['generated'].append(output)
st.session_state['history'].append({"user": user_input, "bot": output})
# Mostrar el historial del chat
with chat_container.container():
for i in range(len(st.session_state['generated'])):
st.write(f"**User:** {st.session_state['past'][i]}")
st.write(f"**Bot:** {st.session_state['generated'][i]}")
|