""" 🤖 N8n Assistant - Open Source (GRÁTIS, CPU-friendly) - Sem OpenAI - LLM: google/flan-t5-base (fallback flan-t5-small) - Embeddings: all-MiniLM-L6-v2 (fallback paraphrase-MiniLM-L3-v2) - Baixa dataset Jeice/n8n-docs-v2 e gera documentacao.txt - Logs detalhados p/ depuração """ import os import json import yaml import logging from typing import Tuple import gradio as gr from huggingface_hub import snapshot_download # LlamaIndex from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings, ServiceContext from llama_index.core.settings import Settings as LISettings from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.llms.huggingface import HuggingFaceLLM # ------------------------- # Logging # ------------------------- logging.basicConfig(level=logging.INFO) logger = logging.getLogger("n8n-assistant") # ------------------------- # Modelos (CPU-friendly) # ------------------------- PRIMARY_LLM = "google/flan-t5-base" FALLBACK_LLM = "google/flan-t5-small" PRIMARY_EMB = "sentence-transformers/all-MiniLM-L6-v2" FALLBACK_EMB = "sentence-transformers/paraphrase-MiniLM-L3-v2" # ------------------------- # App # ------------------------- class N8nAssistant: def __init__(self): self.docs_dir = None self.index = None self.query_engine = None self.inicializado = False self.llm_model_used = None self.emb_model_used = None # ---------- Dataset ---------- def baixar_docs(self) -> bool: """Baixa o dataset com a documentação.""" try: logger.info("📥 Baixando dataset Jeice/n8n-docs-v2 ...") self.docs_dir = snapshot_download( repo_id="Jeice/n8n-docs-v2", repo_type="dataset" ) logger.info(f"✅ Dataset baixado em: {self.docs_dir}") try: logger.info(f"📂 Itens no diretório raiz do dataset: {os.listdir(self.docs_dir)}") data_path = os.path.join(self.docs_dir, "data") if os.path.isdir(data_path): logger.info(f"📂 Pasta /data encontrada. Itens: {os.listdir(data_path)}") except Exception as e: logger.warning(f"⚠️ Não consegui listar arquivos do dataset: {e}") return True except Exception as e: logger.error(f"❌ Erro ao baixar dataset: {e}") return False # ---------- Consolidação ---------- def extrair_conteudo_arquivos(self, pasta: str) -> str: """Varre todas as subpastas e agrega .yml/.yaml/.json/.md/.txt em um único texto.""" extensoes = ('.yml', '.yaml', '.json', '.md', '.txt') texto_final = [] if not os.path.exists(pasta): logger.error(f"❌ Pasta não existe: {pasta}") return "" total_arquivos = 0 for root, _, files in os.walk(pasta): logger.info(f"🔎 Explorando: {root} | {len(files)} arquivos") for file in files: caminho = os.path.join(root, file) if not file.lower().endswith(extensoes): continue total_arquivos += 1 try: if file.lower().endswith(('.yml', '.yaml')): with open(caminho, 'r', encoding='utf-8') as f: data = yaml.safe_load(f) texto = yaml.dump(data, allow_unicode=True, sort_keys=False) elif file.lower().endswith('.json'): with open(caminho, 'r', encoding='utf-8') as f: data = json.load(f) texto = json.dumps(data, ensure_ascii=False, indent=2) else: # .md / .txt with open(caminho, 'r', encoding='utf-8', errors='ignore') as f: texto = f.read() texto_final.append(f"\n\n### Arquivo: {os.path.relpath(caminho, pasta)}\n{texto}") except Exception as e: logger.warning(f"⚠️ Erro lendo {caminho}: {e}") logger.info(f"🧾 Total de arquivos agregados: {total_arquivos}") return "".join(texto_final) def gerar_documentacao(self) -> bool: """Gera documentacao.txt a partir do dataset (raiz + /data se existir).""" try: if not self.docs_dir: logger.error("❌ docs_dir não definido") return False partes = [] # raiz do dataset partes.append(self.extrair_conteudo_arquivos(self.docs_dir)) # subpasta /data (comum em datasets do HF) data_path = os.path.join(self.docs_dir, "data") if os.path.isdir(data_path): partes.append(self.extrair_conteudo_arquivos(data_path)) texto = "\n".join([p for p in partes if p and p.strip()]) if not texto.strip(): logger.error("❌ Nenhum conteúdo válido encontrado no dataset") return False with open("documentacao.txt", "w", encoding="utf-8") as f: f.write(texto) # Loga um preview preview = texto[:1500] logger.info(f"📝 documentacao.txt gerado (preview 1500 chars):\n{preview}") return True except Exception as e: logger.error(f"❌ Erro ao gerar documentacao.txt: {e}") return False # ---------- Modelos ---------- def configurar_embeddings(self) -> bool: for emb in (PRIMARY_EMB, FALLBACK_EMB): try: LISettings.embed_model = HuggingFaceEmbedding(model_name=emb) self.emb_model_used = emb logger.info(f"✅ Embeddings carregados: {emb}") return True except Exception as e: logger.warning(f"⚠️ Falhou carregar embeddings {emb}: {e}") return False def configurar_llm(self) -> bool: gen_kwargs = { "temperature": 0.2, "do_sample": True, "top_p": 0.9 } for name in (PRIMARY_LLM, FALLBACK_LLM): try: llm = HuggingFaceLLM( model_name=name, tokenizer_name=name, context_window=2048, max_new_tokens=384, # menor = mais leve em CPU generate_kwargs=gen_kwargs, device_map="auto", model_kwargs={"torch_dtype": "auto"}, system_prompt=( "Você é um assistente especialista em n8n. " "Responda 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." ), ) LISettings.llm = llm self.llm_model_used = name logger.info(f"✅ LLM carregado: {name}") return True except Exception as e: logger.warning(f"⚠️ Falhou carregar LLM {name}: {e}") return False # ---------- Index ---------- def criar_index(self) -> bool: try: if not os.path.exists("documentacao.txt"): logger.error("❌ documentacao.txt não existe") return False # Carrega o único arquivo consolidado docs = SimpleDirectoryReader(input_files=["documentacao.txt"]).load_data() if not docs: logger.error("❌ Nenhum documento carregado de documentacao.txt") with open("documentacao.txt", "r", encoding="utf-8") as f: logger.error("📄 documentacao.txt (trecho): " + f.read()[:1200]) return False logger.info(f"📚 {len(docs)} documento(s) prontos para indexação") self.index = VectorStoreIndex.from_documents(docs) self.query_engine = self.index.as_query_engine() logger.info("✅ Índice e QueryEngine criados") return True except Exception as e: logger.error(f"❌ Erro ao criar índice: {e}") return False # ---------- Orquestração ---------- def inicializar(self) -> Tuple[bool, str]: try: if not self.baixar_docs(): return False, "Erro ao baixar dataset" if not self.gerar_documentacao(): return False, "Erro ao gerar documentacao.txt" if not self.configurar_embeddings(): return False, "Erro ao configurar embeddings" if not self.configurar_llm(): return False, "Erro ao configurar LLM" if not self.criar_index(): return False, "Erro ao criar índice" self.inicializado = True return True, f"Pronto | LLM: {self.llm_model_used} | Emb: {self.emb_model_used}" except Exception as e: logger.error(f"❌ Erro na inicialização: {e}") return False, f"Erro na inicialização: {e}" def responder(self, pergunta: str) -> str: if 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]}") resp = self.query_engine.query(pergunta) return str(resp) except Exception as e: logger.error(f"❌ Erro na resposta: {e}") return f"❌ Erro ao processar a pergunta: {e}" # ------------------------- # Bootstrap # ------------------------- logger.info("🚀 Subindo N8n Assistant (Open Source, CPU)...") assistant = N8nAssistant() ok, status_msg = assistant.inicializar() if ok: logger.info(f"✅ {status_msg}") else: logger.error(f"❌ {status_msg}") # ------------------------- # Gradio UI # ------------------------- def processar_pergunta(pergunta: str) -> str: if not ok: return f"❌ Sistema não inicializado: {status_msg}" return assistant.responder(pergunta) with gr.Blocks(theme=gr.themes.Soft(), title="N8n Assistant") as demo: gr.Markdown( f""" # 🤖 N8n Assistant (Open Source) Assistente baseado na documentação oficial do **n8n** (dataset do HF). **Status:** {'✅ ' + status_msg if ok else '❌ ' + status_msg} """ ) 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): pergunta = gr.Textbox( label="Sua pergunta", placeholder="Ex: Como configurar um Webhook Trigger no n8n?", lines=3 ) enviar = gr.Button("🚀 Perguntar", variant="primary") limpar = gr.Button("🧹 Limpar") with gr.Column(scale=4): resposta = gr.Textbox( label="Resposta", placeholder="A resposta aparecerá aqui...", lines=14 ) with gr.Accordion("💡 Exemplos", open=False): gr.Markdown( """ - Como configurar webhooks no n8n? - Para que serve o node HTTP Request? - Como integrar com Google Sheets? - Como debugar erros nos nodes? - Quais são boas práticas de workflows? """ ) enviar.click(fn=processar_pergunta, inputs=pergunta, outputs=resposta) limpar.click(lambda: ("", ""), None, [pergunta, resposta]) pergunta.submit(fn=processar_pergunta, inputs=pergunta, outputs=resposta) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)