Spaces:
Sleeping
Sleeping
| """ | |
| 🤖 N8n Assistant - Versão Open Source (GRÁTIS) | |
| - Sem OpenAI | |
| - LLM: microsoft/Phi-3.5-mini-instruct (fallback flan-t5-base) | |
| - Embeddings: all-MiniLM-L6-v2 (fallback L3-v2) | |
| - Compatível com Hugging Face Spaces (CPU) | |
| """ | |
| import os | |
| import yaml | |
| import json | |
| import logging | |
| from typing import Optional, Tuple | |
| import gradio as gr | |
| # LlamaIndex (open source stacks) | |
| from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings | |
| from llama_index.embeddings.huggingface import HuggingFaceEmbedding | |
| from llama_index.llms.huggingface import HuggingFaceLLM | |
| from huggingface_hub import snapshot_download | |
| # ------------------------------------------------------------ | |
| # Logging | |
| # ------------------------------------------------------------ | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger("n8n-assistant") | |
| # ------------------------------------------------------------ | |
| # Configs de modelos (primários + fallbacks) | |
| # ------------------------------------------------------------ | |
| PRIMARY_LLM = "microsoft/Phi-3.5-mini-instruct" | |
| FALLBACK_LLM = "google/flan-t5-base" # muito leve | |
| PRIMARY_EMB = "sentence-transformers/all-MiniLM-L6-v2" | |
| FALLBACK_EMB = "sentence-transformers/paraphrase-MiniLM-L3-v2" | |
| # ------------------------------------------------------------ | |
| # Classe principal | |
| # ------------------------------------------------------------ | |
| class N8nAssistant: | |
| """Assistente N8n open-source e funcional""" | |
| def __init__(self): | |
| self.index = None | |
| self.query_engine = None | |
| self.docs_dir = None | |
| self.inicializado = False | |
| self.llm_model_used = None | |
| self.emb_model_used = None | |
| # --------- Utilitários de dados ---------- | |
| def extrair_conteudo_arquivos(self, pasta: str) -> str: | |
| """Extrai conteúdo textual dos arquivos .yml/.yaml/.json/.md/.txt""" | |
| texto_final = "" | |
| if not os.path.exists(pasta): | |
| logger.error(f"❌ Pasta não encontrada: {pasta}") | |
| return "" | |
| for root, _, files in os.walk(pasta): | |
| for file in files: | |
| caminho_arquivo = os.path.join(root, file) | |
| try: | |
| if file.endswith(('.yml', '.yaml')): | |
| with open(caminho_arquivo, 'r', encoding='utf-8') as f: | |
| data = yaml.safe_load(f) | |
| texto = yaml.dump(data, allow_unicode=True) | |
| texto_final += f"\n\n### Arquivo: {file}\n{texto}" | |
| elif file.endswith('.json'): | |
| with open(caminho_arquivo, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| texto = json.dumps(data, indent=2, ensure_ascii=False) | |
| texto_final += f"\n\n### Arquivo: {file}\n{texto}" | |
| elif file.endswith(('.md', '.txt')): | |
| with open(caminho_arquivo, 'r', encoding='utf-8') as f: | |
| texto = f.read() | |
| texto_final += f"\n\n### Arquivo: {file}\n{texto}" | |
| except Exception as e: | |
| logger.warning(f"⚠️ Erro ao ler {file}: {e}") | |
| continue | |
| return texto_final | |
| def gerar_documentacao(self, pasta_origem: str) -> bool: | |
| """Gera um único arquivo 'documentacao.txt' com todo o conteúdo unificado""" | |
| try: | |
| texto = self.extrair_conteudo_arquivos(pasta_origem) | |
| if not texto.strip(): | |
| logger.warning("⚠️ Nenhum conteúdo encontrado para documentação") | |
| return False | |
| with open("documentacao.txt", 'w', encoding='utf-8') as f: | |
| f.write(texto) | |
| logger.info("✅ Documentação consolidada em documentacao.txt") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro ao gerar documentação: {e}") | |
| return False | |
| def baixar_docs(self) -> bool: | |
| """Baixa a documentação do HF dataset""" | |
| try: | |
| logger.info("📥 Baixando documentação do dataset Jeice/n8n-docs-v2 ...") | |
| self.docs_dir = snapshot_download( | |
| repo_id="Jeice/n8n-docs-v2", | |
| repo_type="dataset" | |
| ) | |
| logger.info("✅ Download concluído") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro no download do dataset: {e}") | |
| return False | |
| # --------- Configuração de modelos ---------- | |
| def configurar_embeddings(self) -> bool: | |
| """Configura embeddings HuggingFace com fallback""" | |
| for emb in (PRIMARY_EMB, FALLBACK_EMB): | |
| try: | |
| Settings.embed_model = HuggingFaceEmbedding(model_name=emb) | |
| self.emb_model_used = emb | |
| logger.info(f"✅ Embeddings configurados: {emb}") | |
| return True | |
| except Exception as e: | |
| logger.warning(f"⚠️ Falha ao carregar embeddings {emb}: {e}") | |
| logger.error("❌ Não foi possível configurar embeddings") | |
| return False | |
| def configurar_llm(self) -> bool: | |
| """Configura LLM HuggingFace com fallback, otimizado para CPU""" | |
| # parâmetros neutros/seguros para CPU | |
| gen_kwargs = { | |
| "temperature": 0.2, | |
| "do_sample": True, | |
| "top_p": 0.9 | |
| } | |
| # tentar primário depois fallback | |
| for model_name in (PRIMARY_LLM, FALLBACK_LLM): | |
| try: | |
| llm = HuggingFaceLLM( | |
| model_name=model_name, | |
| tokenizer_name=model_name, | |
| context_window=4096, | |
| max_new_tokens=512, | |
| generate_kwargs=gen_kwargs, | |
| # device_map="auto" funciona em CPU/GPU no Space | |
| device_map="auto", | |
| model_kwargs={ | |
| # dtype padrão (evitar float16 em CPU) | |
| "torch_dtype": "auto" | |
| }, | |
| # system_prompt para orientar o estilo de resposta | |
| system_prompt=( | |
| "Você é um assistente especialista em n8n. " | |
| "Responda sempre em português do Brasil, de forma clara e objetiva, " | |
| "baseado exclusivamente na documentação fornecida. " | |
| "Se não souber, diga que não há informações suficientes." | |
| ), | |
| ) | |
| Settings.llm = llm | |
| self.llm_model_used = model_name | |
| logger.info(f"✅ LLM configurado: {model_name}") | |
| return True | |
| except Exception as e: | |
| logger.warning(f"⚠️ Falha ao carregar LLM {model_name}: {e}") | |
| logger.error("❌ Não foi possível configurar o LLM") | |
| return False | |
| # --------- Indexação ---------- | |
| def criar_index(self) -> bool: | |
| """Cria o índice vetorial a partir de documentacao.txt""" | |
| try: | |
| if not os.path.exists("documentacao.txt"): | |
| logger.error("❌ documentacao.txt não encontrado") | |
| return False | |
| documents = SimpleDirectoryReader( | |
| input_files=["documentacao.txt"] | |
| ).load_data() | |
| if not documents: | |
| logger.error("❌ Nenhum documento carregado") | |
| return False | |
| # Criar índice + query engine | |
| logger.info("🧠 Criando índice (VectorStoreIndex) ...") | |
| self.index = VectorStoreIndex.from_documents(documents) | |
| self.query_engine = self.index.as_query_engine() | |
| logger.info("✅ Índice criado e query_engine pronto") | |
| return True | |
| except Exception as e: | |
| logger.error(f"❌ Erro ao criar índice: {e}") | |
| return False | |
| # --------- Orquestração ---------- | |
| def inicializar(self) -> Tuple[bool, str]: | |
| """Pipeline completo de inicialização (open-source)""" | |
| try: | |
| # 1) Baixar docs | |
| if not self.baixar_docs(): | |
| return False, "Erro ao baixar a documentação (dataset)" | |
| # 2) Consolidar documentação | |
| if not self.gerar_documentacao(self.docs_dir): | |
| return False, "Erro ao processar/consolidar a documentação" | |
| # 3) Configurar embeddings e LLM (open source) | |
| if not self.configurar_embeddings(): | |
| return False, "Erro ao configurar embeddings" | |
| if not self.configurar_llm(): | |
| return False, "Erro ao configurar LLM" | |
| # 4) Criar índice | |
| if not self.criar_index(): | |
| return False, "Erro ao criar o índice" | |
| self.inicializado = True | |
| return True, ( | |
| f"Sistema inicializado com sucesso | " | |
| f"LLM: {self.llm_model_used} | Embeddings: {self.emb_model_used}" | |
| ) | |
| except Exception as e: | |
| logger.error(f"❌ Erro na inicialização: {e}") | |
| return False, f"Erro: {str(e)}" | |
| def responder(self, pergunta: str) -> str: | |
| """Executa a consulta no query_engine""" | |
| if not pergunta or not pergunta.strip(): | |
| return "⚠️ Por favor, digite uma pergunta." | |
| if not self.inicializado or not self.query_engine: | |
| return "❌ Sistema não inicializado. Recarregue a página." | |
| try: | |
| logger.info(f"🤔 Pergunta: {pergunta[:120]}...") | |
| response = self.query_engine.query(pergunta) | |
| return str(response) | |
| except Exception as e: | |
| logger.error(f"❌ Erro ao responder: {e}") | |
| return f"❌ Erro ao processar pergunta: {str(e)}" | |
| # ------------------------------------------------------------ | |
| # Bootstrap | |
| # ------------------------------------------------------------ | |
| logger.info("🚀 Inicializando N8n Assistant (Open Source)...") | |
| assistant = N8nAssistant() | |
| sucesso, mensagem = assistant.inicializar() | |
| if sucesso: | |
| logger.info(f"✅ {mensagem}") | |
| else: | |
| logger.error(f"❌ {mensagem}") | |
| # ------------------------------------------------------------ | |
| # Gradio UI | |
| # ------------------------------------------------------------ | |
| def processar_pergunta(pergunta: str) -> str: | |
| if not sucesso: | |
| return f"❌ Sistema não inicializado: {mensagem}" | |
| return assistant.responder(pergunta) | |
| with gr.Blocks(theme=gr.themes.Soft(), title="N8n Assistant (Open Source)") as demo: | |
| gr.Markdown( | |
| f""" | |
| # 🤖 N8n Assistant (Open Source) | |
| Assistente para dúvidas sobre **n8n** baseado na documentação oficial e em modelos **open-source**. | |
| **Status:** {'✅ Sistema Pronto' if sucesso else '❌ ' + mensagem} | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("### 🤖 N8n Bot") | |
| with gr.Column(scale=4): | |
| gr.Markdown("## Como posso ajudar você com o n8n?") | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| input_box = gr.Textbox( | |
| label="Sua pergunta", | |
| placeholder="Ex: Como criar um workflow no n8n?", | |
| lines=3 | |
| ) | |
| with gr.Row(): | |
| enviar_btn = gr.Button("🚀 Perguntar", variant="primary") | |
| limpar_btn = gr.Button("🧹 Limpar") | |
| with gr.Column(scale=4): | |
| output_box = gr.Textbox( | |
| label="Resposta", | |
| placeholder="Sua resposta aparecerá aqui...", | |
| lines=12 | |
| ) | |
| with gr.Accordion("💡 Exemplos de Perguntas", open=False): | |
| gr.Markdown( | |
| """ | |
| - Como criar um workflow no n8n? | |
| - Para que serve o node HTTP Request? | |
| - Como integrar n8n com Google Sheets? | |
| - Como configurar webhooks no n8n? | |
| - Quais são as melhores práticas para workflows? | |
| - Como debugar erros nos nodes? | |
| - Como usar condições nos workflows? | |
| - Quais nodes usar para automação de email? | |
| """ | |
| ) | |
| enviar_btn.click(fn=processar_pergunta, inputs=input_box, outputs=output_box) | |
| limpar_btn.click(lambda: ("", ""), None, [input_box, output_box]) | |
| input_box.submit(fn=processar_pergunta, inputs=input_box, outputs=output_box) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True) | |