Spaces:
Sleeping
Sleeping
| 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]}") | |