Spaces:

unb-lamfo-sgd
/

prototipo-2-agente

Sleeping

App Files Files Community

Pecximenes commited on Oct 22, 2024

Commit

0746d2c

1 Parent(s): 70a5baf

Adding feedback's and login's features

Browse files

Files changed (6) hide show

interface/app.py +24 -14
interface/auxiliary_functions.py +60 -7
interface/chatbot.py +97 -43
interface/login.py +110 -0
interface/main.py +0 -122
requirements.txt +4 -2

interface/app.py CHANGED Viewed

@@ -9,6 +9,7 @@ sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 # Importando módulos necessários
 from interface.chatbot import Chatbot
 def display_message(message):
@@ -33,21 +34,30 @@ def display_page_message(message):
 def main():
     app = Chatbot()
-    with st.sidebar:
-        app.create_sidebar()
-    app.mount_chatbot()
-    if app.response:
-        app.add_to_history(app.response, role="user")
-        app.generate_awnser(app.response)
-        app.response = ""  # Zerando a variável após uso
-    for message in st.session_state.chat_history:
-        display_message(message)
-        if message["type"] == "page" and message == st.session_state.chat_history[-1]:
-            app.display_suggestions(app.links)
 if __name__ == "__main__":

 # Importando módulos necessários
 from interface.chatbot import Chatbot
+from interface.login import Login
 def display_message(message):
 def main():
     app = Chatbot()
+    login = Login()
+    if not st.session_state.user_authorized:
+        login.mount_login_page()
+    else:
+        with st.sidebar:
+            app.create_sidebar()  # Corrigido para chamar o método que cria o sidebar
+        app.mount_chatbot() # Recebe o input de texto feito pelo usuário
+        if app.response:
+            app.add_to_history(app.response, role="user")
+            app.generate_answer(app.response)
+            app.response = ""  # Zerando a variável após uso
+            st.session_state.is_feedback_active = True # Ativando o feedback
+        for message in st.session_state.chat_history:
+            display_message(message)
+            if message["type"] == "page" and message == st.session_state.chat_history[-1]:
+                app.display_suggestions(app.links)
+        if st.session_state.is_feedback_active:
+            app.display_feedback()
 if __name__ == "__main__":

interface/auxiliary_functions.py CHANGED Viewed

@@ -1,9 +1,13 @@
-# Funções utilitárias
 import re
 import requests
 from bs4 import BeautifulSoup
 from urllib.parse import urlparse, unquote
 def extract_content(text, tag):
     """Extrai conteúdo de um texto com base na tag fornecida."""
@@ -12,29 +16,78 @@ def extract_content(text, tag):
     return match.group(1).strip() if match else None
 def fetch_webpage_content(url):
-    """Obtém o conteúdo HTML de uma URL e retorna o conteúdo principal."""
     try:
         response = requests.get(url)
         response.raise_for_status()
         soup = BeautifulSoup(response.text, 'html.parser')
         main_content = soup.find('div', id='main')
         if main_content:
             body_tag = soup.find('body')
             if body_tag:
                 body_tag.clear()
                 body_tag.append(main_content)
-            return str(soup)
-        return "<html><body><p>Could not find main content on the page.</p></body></html>"
     except requests.RequestException as e:
-        return f"<html><body><p>Error fetching the webpage: {str(e)}</p></body></html>"
 def extract_links(html_content):
     """Extrai todos os links (URLs) de um conteúdo HTML."""
     soup = BeautifulSoup(html_content, 'html.parser')
     return [a_tag['href'] for a_tag in soup.find_all('a', href=True)]
 def url_to_suggestion(url):
-    """Converte uma URL longa em uma sugestão amigável."""
     path = unquote(urlparse(url).path).strip('/').split('/')
-    return path[-1].replace('-', ' ').replace('_', ' ').capitalize() if path else url

 import re
 import requests
 from bs4 import BeautifulSoup
 from urllib.parse import urlparse, unquote
+import streamlit as st
+from openai import OpenAI
+from qdrant_client import QdrantClient
+import edgedb
+import os
 def extract_content(text, tag):
     """Extrai conteúdo de um texto com base na tag fornecida."""
     return match.group(1).strip() if match else None
 def fetch_webpage_content(url):
+    """Obtém o conteúdo HTML de uma URL e retorna o conteúdo principal, desativando links específicos, mas mantendo os links originais para sugestões."""
     try:
         response = requests.get(url)
         response.raise_for_status()
+        # Usando BeautifulSoup para processar o HTML
         soup = BeautifulSoup(response.text, 'html.parser')
+        # Localiza o conteúdo principal pelo ID 'main'
         main_content = soup.find('div', id='main')
         if main_content:
+            # Armazenar os links originais para sugestões
+            original_links = [link['href'] for link in main_content.find_all('a', class_='govbr-card-content')]
+            # Desativar links com a classe "govbr-card-content" (apenas visualmente)
+            for link in main_content.find_all('a', class_='govbr-card-content'):
+                link['href'] = 'javascript:void(0);'  # Impede a navegação
+                link['style'] = 'pointer-events: none; cursor: not-allowed;'  # Estilo para desativar visualmente
+            # Remover elementos específicos (exemplo: div com a classe 'outstanding-header')
+            for unwanted_element in main_content.find_all('div', class_='outstanding-header'):
+                unwanted_element.decompose()  # Remove o elemento do DOM
+            # Limpa o conteúdo do <body> e insere o conteúdo principal
             body_tag = soup.find('body')
             if body_tag:
                 body_tag.clear()
                 body_tag.append(main_content)
+            # Retorna o conteúdo modificado e os links originais
+            return str(soup), original_links
+        # Mensagem de erro caso não encontre o conteúdo principal
+        return "<html><body><p>Could not find main content on the page.</p></body></html>", []
     except requests.RequestException as e:
+        # Retorna uma mensagem de erro em caso de falha na requisição
+        return f"<html><body><p>Error fetching the webpage: {str(e)}</p></body></html>", []
 def extract_links(html_content):
     """Extrai todos os links (URLs) de um conteúdo HTML."""
     soup = BeautifulSoup(html_content, 'html.parser')
     return [a_tag['href'] for a_tag in soup.find_all('a', href=True)]
 def url_to_suggestion(url):
+    """Converte uma URL longa em uma sugestão amigável e natural."""
     path = unquote(urlparse(url).path).strip('/').split('/')
+    # Garantir que há pelo menos duas partes significativas no caminho
+    if len(path) > 2:
+        # Remover hífens e capitalizar cada palavra
+        section = path[-2].replace('-', ' ').title()
+        subsection = path[-1].replace('-', ' ').title()
+        # Formatar a sugestão de forma mais natural
+        return f"Acesso à seção '{section}' sobre '{subsection}'"
+    return None
+@st.cache_resource
+def connect_to_services():
+    oa_client = OpenAI(
+        api_key=os.environ.get("OPENAI_API_KEY")
+    )
+    qdrant_client = QdrantClient(
+        url=os.environ.get("QDRANT_URL"),
+        api_key=os.environ.get("QDRANT_KEY")
+    )
+    edgedb_client = edgedb.create_client()
+    return oa_client, qdrant_client, edgedb_client

interface/chatbot.py CHANGED Viewed

@@ -14,8 +14,7 @@ from interface.auxiliary_functions import (
     url_to_suggestion
 )
-class Chatbot():
     def __init__(self):
         # Configuração inicial do histórico de chat
         if "chat_history" not in st.session_state:
@@ -29,55 +28,33 @@ class Chatbot():
         # Conteúdo dos botões do sidebar
         if "topics" not in st.session_state:
             st.session_state["topics"] = [
-                "Niveis da conta govbr.",
                 "Dúvidas no reconhecimento facial.",
                 "Como recuperar minha conta gov.br",
-                "Dúvidas para aumentar o nível com a cin."
             ]
         # Pergunta do usuário no chatbot
         self.response = ""
         self.links = []
-    def display_suggestions(self, links):
-        suggestions = [(link, url_to_suggestion(link)) for link in links if url_to_suggestion(link)]
-        st.subheader("Sugestões de Passos:")
-        if suggestions:
-            st.write("Você também pode se interessar em:")
-            for link, suggestion in suggestions:
-                st.write(f"- {suggestion}")
-        else:
-            st.write("Não há sugestões disponíveis no momento.")
-    def mount_chatbot(self):
-        # Exibição do título e subtítulo
-        st.title("Bem-vindo à ajuda do gov.br")
-        st.caption("💬 Qual a sua dificuldade hoje? Estou aqui para ajudar!")
-        # Exibição do espaço para mandar mensagem
-        if user_query := st.chat_input(placeholder="Digite sua mensagem"):
-            st.session_state.chat_history.append({"type": "text","role": "user", "content": user_query})
-            #TODO Aplicar o llm_rag nessa parte
-            full_link = self.generate_awnser(user_query)
     def create_sidebar(self):
         st.image('https://www.gov.br/++theme++padrao_govbr/img/govbr-logo-large.png', width=200)
         st.header("Tópicos frequentes")
-        for topic in st.session_state.topics:
             if st.button(topic, key=topic):
                 self.response = topic
         # Espaços em branco para organização
         for _ in range(5):
             st.write("")
-        # Botão centralizado
         col1, col2, col3 = st.columns([1, 2, 1])
         with col2:
             if st.button("LIMPAR HISTÓRICO"):
@@ -87,21 +64,52 @@ class Chatbot():
                     "content": "Como eu posso ajudar?",
                     "links": []
                 }]
     def add_to_history(self, message, role="user", type="text"):
         st.session_state.chat_history.append({
             "role": role,
             "type": type,
             "content": message
-            })
-    def generate_awnser(self, input_message):
-        # Retorna o caminho do site gov que responde a pergunta
         response_text = send_message(input_message, pipeline)
-        # Armazena a resposta do agent com a tag <path>
         self.add_to_history(response_text, role="assistant")
         # Extração do caminho da resposta do bot
@@ -109,11 +117,57 @@ class Chatbot():
         if path:
             base_url = "https://www.gov.br/governodigital/pt-br/"
             full_url = base_url + path
             with st.spinner("Obtendo conteúdo da página..."):
-                content = fetch_webpage_content(full_url)
             self.add_to_history(content, role="assistant", type="page")
             extracted_links = extract_links(content)
-            self.links = extracted_links

     url_to_suggestion
 )
+class Chatbot:
     def __init__(self):
         # Configuração inicial do histórico de chat
         if "chat_history" not in st.session_state:
         # Conteúdo dos botões do sidebar
         if "topics" not in st.session_state:
             st.session_state["topics"] = [
+                "Níveis da conta govbr.",
                 "Dúvidas no reconhecimento facial.",
                 "Como recuperar minha conta gov.br",
+                "Dúvidas para aumentar o nível com a CIN."
             ]
+        if "is_feedback_active" not in st.session_state:
+            st.session_state.is_feedback_active = False
         # Pergunta do usuário no chatbot
         self.response = ""
         self.links = []
     def create_sidebar(self):
+        """Cria o sidebar com tópicos e opções."""
         st.image('https://www.gov.br/++theme++padrao_govbr/img/govbr-logo-large.png', width=200)
         st.header("Tópicos frequentes")
+        for topic in st.session_state["topics"]:
             if st.button(topic, key=topic):
                 self.response = topic
         # Espaços em branco para organização
         for _ in range(5):
             st.write("")
+        # Botão centralizado para limpar histórico
         col1, col2, col3 = st.columns([1, 2, 1])
         with col2:
             if st.button("LIMPAR HISTÓRICO"):
                     "content": "Como eu posso ajudar?",
                     "links": []
                 }]
+                st.session_state.is_feedback_active = False
+    def display_suggestions(self, links):
+        """Exibe as sugestões de links no chat."""
+        if links:
+            st.subheader("Você pode acessar a página na web em:")
+            st.write(links[0])  # O primeiro link será o da página principal
+            st.subheader("Próximos Passos:")
+            suggestions = [(link, url_to_suggestion(link)) for link in links[1:] if url_to_suggestion(link)]
+            if suggestions:
+                st.write("Você também pode se interessar em:")
+                for link, suggestion in suggestions:
+                    st.write(f"- [{suggestion}]({link})")  # Exibe os links clicáveis no formato Markdown
+            else:
+                st.write("Não há sugestões de navegação adicionais.")
+        else:
+            st.write("Não há links disponíveis para exibição.")
+    def mount_chatbot(self):
+        """Configura o chatbot com título, subtítulo e espaço de entrada."""
+        st.title("Bem-vindo à ajuda do gov.br")
+        st.caption("💬 Qual a sua dificuldade hoje? Estou aqui para ajudar!")
+        # Exibição do espaço para mandar mensagem
+        if user_query := st.chat_input(placeholder="Digite sua mensagem"):
+            st.session_state.chat_history.append({"type": "text", "role": "user", "content": user_query})
+            self.generate_answer(user_query)
+            st.session_state.is_feedback_active = True   # Ativando o feedback
     def add_to_history(self, message, role="user", type="text"):
+        """Adiciona uma mensagem ao histórico de chat."""
         st.session_state.chat_history.append({
             "role": role,
             "type": type,
             "content": message
+        })
+    def generate_answer(self, input_message):
+        """Gera uma resposta para a consulta do usuário."""
+        # Chama o pipeline para obter a resposta
         response_text = send_message(input_message, pipeline)
+        # Armazena a resposta do agente com a tag <path>
         self.add_to_history(response_text, role="assistant")
         # Extração do caminho da resposta do bot
         if path:
             base_url = "https://www.gov.br/governodigital/pt-br/"
             full_url = base_url + path
+            # Armazena o link principal
+            self.links = [full_url]
             with st.spinner("Obtendo conteúdo da página..."):
+                content, original_links = fetch_webpage_content(full_url)  # Agora retornando também os links desativados
+            # Adiciona o conteúdo da página ao histórico do chatbot
             self.add_to_history(content, role="assistant", type="page")
+            # Extrai links ativos do conteúdo da página e armazena para sugestões
             extracted_links = extract_links(content)
+            # Combinar links ativos e links inibidos
+            all_links = extracted_links + original_links
+            self.links.extend(all_links)
+    def display_feedback(self):
+        user_input = st.session_state.chat_history[-2]["content"]
+        bot_output = st.session_state.chat_history[-1]["content"]
+        with st.expander("Avaliação do atendimento"):
+            # st.write(f'O que achou da resposta para a pergunta "{user_input}"?')
+            st.write(f'O que achou da resposta para a pergunta?')
+            rate = st.feedback("stars")
+            # rate = st.feedback("faces")
+            text_feedback = st.text_input("Comentários extras:")
+            # Botão para confirmar a avaliação
+            if rate is not None:
+                if st.button("Enviar Avaliação"):
+                    try:
+                        feedback_rate = rate + 1
+                        # TODO Colocar nessa parte a estrutura para adicionar o feedback_data ao banco de dados
+                        # st.session_state.services["edgedb_client"].query('''
+                        #     INSERT Feedback {
+                        #         rating := <int16>$rating,
+                        #         content := <str>$content,
+                        #         message := <str>$content
+                        #     }
+                        # ''',
+                        # message_id=bot_output,
+                        # rating=feedback_rate,
+                        # content=text_feedback,
+                        # )
+                        # st.session_state.chat_history
+                        st.success(f"Avaliação enviada!")
+                    except Exception as e:
+                        print(e)
+                        st.error("Erro ao enviar avaliação!")
+                    st.session_state.is_feedback_active = False # Desativando o feedback

interface/login.py ADDED Viewed

	@@ -0,0 +1,110 @@

+import streamlit as st
+import sys
+import os
+import bcrypt
+# Add the parent directory to the Python path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from auxiliary_functions import connect_to_services
+class Login:
+    def __init__(self):
+        if "username" not in st.session_state:
+            st.session_state.username = ""
+        if "user_authorized" not in st.session_state:
+            st.session_state.user_authorized = False
+        if "services" not in st.session_state:
+            oa_client, qdrant_client, edgedb_client =  connect_to_services()
+            st.session_state.services = {
+                "oa_client": oa_client,
+                "qdrant_client": qdrant_client,
+                "edgedb_client": edgedb_client
+            }
+        self.username = ""
+        self.password = ""
+    def mount_login_page(self):
+        # Exibição do título e subtítulo
+        st.title("Bem-vindo à ajuda do gov.br")
+        with st.form("login_form"):
+            st.title("Login")
+            username = st.text_input('E-mail:')
+            password = st.text_input('Senha:', type='password')
+            submitted = st.form_submit_button("Entrar")
+            st.text("Não possui uma conta?")
+            create_account = st.form_submit_button("Criar uma conta")
+            if create_account:
+                if not username:
+                    st.warning("Digite um email para cadastrar a conta.")
+                if not password:
+                    st.warning("Digite uma senha para cadastrar a conta.")
+                if username and password:
+                    if "@" not in username:
+                        st.warning("Digite um email válido para cadastrar a conta.")
+                    else:
+                        user = self.validate_credentials(username, password, return_just_user=True)
+                        if user:
+                            st.warning("Já existe uma conta usando este email")
+                        else:
+                            # Gerando senha encriptada
+                            hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
+                            hashed_password_str = hashed_password.decode('utf-8')
+                            st.session_state.services["edgedb_client"].query('''
+                                INSERT User {
+                                    username := <str>$username,
+                                    password := <str>$password
+                                }
+                            ''',
+                                username=username, \
+                                password=hashed_password_str, \
+                            )
+                            st.success("Conta cadastrada com sucesso.")
+            if submitted:
+                if username and password:
+                    checked = self.validate_credentials(username, password)
+                    if checked:
+                        st.success("Usuário logado!")
+                        st.session_state.user_authorized = True
+                        st.session_state.username = username
+                        st.rerun()
+                    else:
+                        st.error("Nome de usuário ou senha inválidos.")
+                else:
+                    st.warning("Digite o nome do usuário e a senha.")
+    def validate_credentials(self, username, password, return_just_user=False):
+        user = st.session_state.services["edgedb_client"].query('''
+            SELECT User {
+                username,
+                password
+            } FILTER .username = <str>$username
+        ''', username=username)
+        if return_just_user:
+            if not user:
+                return False
+            return True
+        if not user or not bcrypt.checkpw(password.encode('utf-8'), user[0].password.encode('utf-8')):
+            return False
+        return True

interface/main.py DELETED Viewed

@@ -1,122 +0,0 @@
-import streamlit as st
-import streamlit.components.v1 as components
-import re
-import sys
-import os
-import requests
-from bs4 import BeautifulSoup
-import difflib
-from urllib.parse import urlparse, unquote
-from pipelines.message import send_message
-from agent import pipeline, knowledge, graph, graph_data
-# Configuração do caminho
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-# Substitui o caminho do gráfico pelo conteúdo do graph_data
-graph = graph.replace('"../memory/graph_data.json"', f'{graph_data}')
-# Variáveis globais
-extracted_links = []
-# Configuração inicial do histórico de chat
-if "chat_history" not in st.session_state:
-    st.session_state.chat_history = []
-# Funções utilitárias
-def extract_content(text, tag):
-    """Extrai conteúdo de um texto com base na tag fornecida."""
-    pattern = rf'<{tag}>(.*?)</{tag}>'
-    match = re.search(pattern, text, re.DOTALL)
-    return match.group(1).strip() if match else None
-def fetch_webpage_content(url):
-    """Obtém o conteúdo HTML de uma URL e retorna o conteúdo principal."""
-    try:
-        response = requests.get(url)
-        response.raise_for_status()
-        soup = BeautifulSoup(response.text, 'html.parser')
-        main_content = soup.find('div', id='main')
-        if main_content:
-            body_tag = soup.find('body')
-            if body_tag:
-                body_tag.clear()
-                body_tag.append(main_content)
-            return str(soup)
-        return "<html><body><p>Could not find main content on the page.</p></body></html>"
-    except requests.RequestException as e:
-        return f"<html><body><p>Error fetching the webpage: {str(e)}</p></body></html>"
-def extract_links(html_content):
-    """Extrai todos os links (URLs) de um conteúdo HTML."""
-    soup = BeautifulSoup(html_content, 'html.parser')
-    return [a_tag['href'] for a_tag in soup.find_all('a', href=True)]
-def url_to_suggestion(url):
-    """Converte uma URL longa em uma sugestão amigável."""
-    path = unquote(urlparse(url).path).strip('/').split('/')
-    return path[-1].replace('-', ' ').replace('_', ' ').capitalize() if path else url
-def display_suggestions(links):
-    """Exibe sugestões para o usuário com base nos links extraídos."""
-    suggestions = [(link, url_to_suggestion(link)) for link in links if url_to_suggestion(link)]
-    st.subheader("Sugestões de Passos:")
-    if suggestions:
-        st.write("Você também pode se interessar em:")
-        for link, suggestion in suggestions:
-            st.write(f"- {suggestion}")
-    else:
-        st.write("Não há sugestões disponíveis no momento.")
-def display_knowledge_tree(structure):
-    """Exibe a estrutura de conhecimento na barra lateral do Streamlit."""
-    with st.sidebar.expander("Knowledge Structure", expanded=True):
-        components.html(graph, height=600, scrolling=True)
-        st.markdown(structure)
-def navigate_page(full_url):
-    """Função para navegar para uma nova página e atualizar o conteúdo."""
-    global extracted_links
-    with st.spinner("Obtendo conteúdo da página..."):
-        content = fetch_webpage_content(full_url)
-    extracted_links = extract_links(content)
-    st.session_state.chat_history.append({"type": "page", "content": content, "links": extracted_links})
-    st.subheader("Conteúdo da Página Web:")
-    components.html(content, height=600, scrolling=True)
-    display_suggestions(extracted_links)
-    st.subheader("URL Completa:")
-    st.write(full_url)
-def main():
-    st.title("Alfred - Assistente de IA para Serviços gov.br")
-    st.write("Qual a sua dificuldade hoje? Estou aqui para ajudar!")
-    display_knowledge_tree(knowledge)
-    # Exibe o histórico do chat
-    for message in st.session_state.chat_history:
-        if message["type"] == "text":
-            st.markdown(message["content"])
-        elif message["type"] == "page":
-            st.subheader("Conteúdo da Página Web:")
-            components.html(message["content"], height=400, scrolling=True)
-    with st.form(key='user_input_form', clear_on_submit=True):
-        user_query = st.text_input("Digite sua mensagem aqui:")
-        submit_button = st.form_submit_button(label='Enviar')
-        if submit_button and user_query:
-            response_text = send_message(user_query, pipeline)
-            st.write("Bot:", response_text)
-            st.session_state.chat_history.append({"type": "text", "content": f"**Você:** {user_query}"})
-            # Extração do caminho da resposta do bot
-            path = extract_content(response_text, "path")
-            if path:
-                base_url = "https://www.gov.br/governodigital/pt-br/"
-                full_url = base_url + path
-                navigate_page(full_url)
-            st.session_state.chat_history.append({"type": "text", "content": f"**Bot:** {response_text}"})
-if __name__ == "__main__":
-    main()

requirements.txt CHANGED Viewed

@@ -13,6 +13,8 @@ docker==7.1.0
 paramiko==3.4.0
 duckduckgo_search==6.1.12
 inputimeout==1.0.4
-streamlit==1.25.0
 beautifulsoup4==4.12.3
-bs4==0.0.2

 paramiko==3.4.0
 duckduckgo_search==6.1.12
 inputimeout==1.0.4
+streamlit==1.39.0
 beautifulsoup4==4.12.3
+bs4==0.0.2
+qdrant-client==1.12.0
+edgedb==2.1.0