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