diff --git a/ai_agent.py b/ai_agent.py index c92544d..b18db22 100644 --- a/ai_agent.py +++ b/ai_agent.py @@ -1,120 +1,90 @@ import os import re import requests -from tools import run_bash_command, get_system_health +import json +from tools import AVAILABLE_TOOLS from config import get_config +def get_llm_response(prompt: str, provider: str, cfg: dict) -> str: + """Invoca o provedor de LLM configurado.""" + if provider == "gemini": + api_key = cfg.get("gemini_api_key") + url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={api_key}" + payload = {"contents": [{"parts": [{"text": prompt}]}]} + res = requests.post(url, json=payload) + if res.status_code == 200: + return res.json()["candidates"][0]["content"]["parts"][0]["text"] + return f"Erro Gemini: {res.text}" + + elif provider == "ollama": + ollama_host = os.getenv("OLLAMA_HOST", "http://ollama-lw4s8g4gc8gss4gkc4gg0wk4:11434") + res = requests.post(f"{ollama_host}/api/generate", json={ + "model": os.getenv("OLLAMA_MODEL", "qwen2.5-coder:1.5b"), + "prompt": prompt, + "stream": False + }) + if res.status_code == 200: + return res.json().get("response", "") + return f"Erro Ollama: {res.text}" + + return "Provedor desconhecido." + def query_agent(prompt: str, override_provider: str = None) -> str: """ - Função principal que roteia pro LLM desejado, detecta intents e aciona as Tools. + Motor Agente em Loop (ReAct): Pensamento -> Ação -> Observação -> Resposta Final. """ - - # SYSTEM PROMPT AVANÇADO: O "Ensinamento" da Inteligência Artificial - system_prompt = """Você é o [Antigravity VPS Agent], uma Inteligência Artificial autônoma de Dev/SysAdmin operando diretamente em uma máquina Ubuntu Linux do usuário (Marcos). -Sua missão é ajudar o Marcos a gerenciar o servidor, criar aplicações, analisar código e orquestrar o Docker. VOCÊ DEVE RESPONDER SEMPRE EM PORTUGUÊS FLUENTE DO BRASIL. - -### SEUS PODERES E ARQUITETURA -1. **Nível Root**: Você roda em um Bot Python empacotado via Docker. Este container mapeia o `/var/run/docker.sock`, o que te dá o poder DIVINO sobre tudo. Você pode listar, deletar, parar e recriar qualquer container na máquina hospedeira inteira. -2. **Sistema de Arquivos**: O disco principal do servidor está montado de forma segura, o que permite que você leia logs nativos. -3. **Mecanismo de Ação (A Interface de Ferramentas)**: Você **não executa os comandos diretamente pela web**. Você possui uma interface conectada ao bot no Telegram e Web. Para interagir com o servidor, sempre que decidir que algo precisa ser lido (ex: `ls -la`, `docker network ls`, `cat arquivo.txt`) ou modificado, responda com a notação: [CMD] o_comando_aqui [/CMD]. -4. O bot Python varrerá sua resposta via Regex, extrairá o comando que você colocou entre os blocos, abrirá um shell Bash real, e te devolverá na conversa o resultado do seu comando. - -### SEUS LIMITES E REGRAS -- **Sem Comandos Interativos**: Jamais use comandos que exijam resposta humana travando o terminal (ex: `nano`, `top -d 1`, `apt-get install` sem `-y`). Se usar, o bash vai dar timeout (60s limit). -- **Nunca use systemctl**: Você roda dentro de um container Docker. Logo, o comando `systemctl` NÃO FUNCIONA. Para ver serviços, use invariavelmente o comando `docker ps -a`. -- **Aja, Não Explique Demais**: Se o usuário te der uma tarefa como "Veja meus containers", não explique o que é Docker. Apenas devolva [CMD] docker ps -a [/CMD] e diga "Estou olhando agora mesmo." -- **Erros**: Se o comando retornar um código 127, 1 ou falha de acesso, aceite o erro e sugira tentar um comando de diagnóstico em seguida. - -Acorde, a VPS é sua para cuidar!""" - - if "logs do nginx" in prompt.lower() or "nginx" in prompt.lower(): - output = run_bash_command("systemctl status nginx || echo 'Nginx não parece ser gerenciado por systemctl aqui'") - return f"Executei a checagem no Nginx. Olha o resultado: \n\n{output}" - - if any(palavra in prompt.lower() for palavra in ["status", "cpu", "saude", "saúde", "sauda", "memória", "disco", "hd"]): - health = get_system_health() - return f"A saúde atual pontual da sua máquina direta do Python está assim:\n{health}" - cfg = get_config() - provider = override_provider or cfg.get("active_provider", "ollama") + provider = override_provider or cfg.get("active_provider", "gemini") - if provider == "gemini": - gemini_api_key = cfg.get("gemini_api_key", "") - if not gemini_api_key: return "Chave API do Gemini não configurada." + # Contexto de Ferramentas para a IA + tools_desc = "\n".join([f"- {k}: {v['description']}" for k,v in AVAILABLE_TOOLS.items()]) + + # Prompt especializado (sem chaves complexas) + system_prompt_base = """Você é o [Antigravity VPS Agent]. +Sua missão é ser o SysAdmin de elite do Marcos. Você tem acesso root. + +### REGRAS: +1. Responda em PORTUGUÊS (Brasil). +2. Para agir, use: [CMD] comando [/CMD]. Rode UM comando por vez. +3. Seus comandos devem ser diretos (docker, git, ls, rm, mkdir). +4. Após o comando, você receberá a saída. O seu objetivo é resolver a solicitação do usuário. +5. Quando terminar, sua resposta FINAL deve ter: + - Um resumo técnico rápido. + - Uma seção entre tags ... com uma tabela Markdown limpa ou resumo em tópicos (Nome: Valor) para o usuário leigo. + +### FERRAMENTAS DISPONÍVEIS: +{TOOLS_LIST} + +### EXEMPLO DE REFINAMENTO VISUAL: +Relatório: Coletei os dados solicitados. + +### 📊 Status Global +- **CPU**: 10% +- **RAM**: 500MB livre + +""" + system_prompt = system_prompt_base.replace("{TOOLS_LIST}", tools_desc) + + history = f"\nUsuário: {prompt}\n" + max_loops = 10 + + for _ in range(max_loops): + full_prompt = system_prompt + history + response = get_llm_response(full_prompt, provider, cfg) - try: - url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key={gemini_api_key}" - payload = { - "contents": [{"parts": [{"text": system_prompt + "\n\nUsuário: " + prompt}]}] - } - res = requests.post(url, json=payload) - if res.status_code == 200: - raw_response = res.json()["candidates"][0]["content"]["parts"][0]["text"] - else: - return f"Erro na API do Gemini: {res.text}" - - # Regex Universal para processar Comandos - match = re.search(r"\[.*?CMD\](.*?)\[/.*?CMD\]", raw_response, re.IGNORECASE | re.DOTALL) - if match: - comando_bash = match.group(1).strip() - resultado = run_bash_command(comando_bash) - - # Tradução via Gemini - traducao_prompt = f"O comando `{comando_bash}` retornou o seguinte dado: {resultado[:1500]}\nTraduza gentilmente para um formato leigo sem códigos de erro difíceis." - t_payload = {"contents": [{"parts": [{"text": system_prompt + "\n\n" + traducao_prompt}]}]} - t_res = requests.post(url, json=t_payload) - if t_res.status_code == 200: - texto_leigo = t_res.json()["candidates"][0]["content"]["parts"][0]["text"] - else: - texto_leigo = "Falha ao gerar resumo no Gemini." - - return f"🤖 **Comando Técnico (Gemini):** `{comando_bash}`\n\n**🖥️ Log Nativo:**\n```\n{resultado[:1000]}\n```\n\n🧑‍🏫 **Tradução:**\n{texto_leigo}" + # Procura por comandos na resposta + match = re.search(r"\[CMD\](.*?)\[/CMD\]", response, re.IGNORECASE | re.DOTALL) + + if match: + cmd = match.group(1).strip() + # Executa a tool (por enquanto focada em bash que é a mais poderosa) + print(f"Agente executando: {cmd}") + observation = AVAILABLE_TOOLS["run_bash_command"]["func"](cmd) - return raw_response - except Exception as e: - return f"Falha de Conexão com Gemini Pro: {e}" - - elif provider == "ollama": - try: - ollama_host = os.getenv("OLLAMA_HOST", "http://ollama-lw4s8g4gc8gss4gkc4gg0wk4:11434") - res = requests.post(f"{ollama_host}/api/generate", json={ - "model": os.getenv("OLLAMA_MODEL", "qwen2.5-coder:1.5b"), - "prompt": system_prompt + "\nUsuário: " + prompt, - "stream": False - }) - if res.status_code == 200: - raw_response = res.json().get("response", "Erro vazio do Ollama") - - # Motor de Tool Calling Tolerante: Detecta [CMD] ou [VCMD] ou variações que LLMs minúsculas inventam - match = re.search(r"\[.*?CMD\](.*?)\[/.*?CMD\]", raw_response, re.IGNORECASE | re.DOTALL) - - if match: - comando_bash = match.group(1).strip() - resultado = run_bash_command(comando_bash) - - # ---------------------------------------------------- - # NOVA LÓGICA: Tradução Leiga (Segundo Prompt para a IA) - # ---------------------------------------------------- - traducao_prompt = f"O comando `{comando_bash}` retornou a seguinte saída do servidor:\n\n{resultado[:1500]}\n\nTraduza GENTILMENTE essa saída técnica explicando de forma amigável, gentil e em português muito simples (para um não-técnico) o que isso indica. SE a saída for um 'ERRO', acalme o usuário, resuma que um comando técnico falhou e sugira verbalmente o que de forma segura investigar a seguir (não mande blocos confusos, apenas explique em português fluente)." - - try: - res_traducao = requests.post(f"{ollama_host}/api/generate", json={ - "model": os.getenv("OLLAMA_MODEL", "qwen2.5-coder:1.5b"), - "prompt": system_prompt + "\n\n" + traducao_prompt, - "stream": False - }) - if res_traducao.status_code == 200: - texto_leigo = res_traducao.json().get("response", "Erro ao processar resumo.") - else: - texto_leigo = "Falha ao gerar resumo na LLM." - except Exception as e_traducao: - texto_leigo = "Ocorreu uma falha ao tentar traduzir o log." - - # Retorna na tela a versão técnica seguida da versão explicada - return f"🤖 **Comando Técnico:** `{comando_bash}`\n\n**🖥️ Log Nativo (Terminal):**\n```\n{resultado[:1000]}\n```\n\n🧑‍🏫 **Tradução:**\n{texto_leigo}" - - return raw_response - except Exception as e: - return f"Falha ao conectar no Ollama local: {e}" - - return "Ação não reconhecida pelo Agente no momento." + # Adiciona ao histórico para a IA ler na próxima rodada + history += f"\nAgente (Pensamento/Ação): {response}\nSISTEMA (Saída do Terminal): {observation}\n" + else: + # Se não tem comando, é a resposta final + return response + + return "O agente atingiu o limite de tentativas para esta tarefa." diff --git a/audio_handler.py b/audio_handler.py new file mode 100644 index 0000000..525a187 --- /dev/null +++ b/audio_handler.py @@ -0,0 +1,42 @@ +import os +import speech_recognition as sr +from pydub import AudioSegment +from gtts import gTTS +import uuid +import re + +def transcribe_audio(file_path: str) -> str: + """Converte áudio (qualquer formato compatível com pydub) para WAV e transcreve com Google Speech.""" + recognizer = sr.Recognizer() + + # Se não for wav, converte usando pydub (precisa de ffmpeg na VPS) + temp_wav = f"/tmp/{uuid.uuid4()}.wav" + try: + audio = AudioSegment.from_file(file_path) + audio.export(temp_wav, format="wav") + + with sr.AudioFile(temp_wav) as source: + audio_data = recognizer.record(source) + text = recognizer.recognize_google(audio_data, language="pt-BR") + return text + finally: + if os.path.exists(temp_wav): + os.remove(temp_wav) + +def text_to_speech(text: str) -> str: + """Sintetiza texto em áudio MP3, removendo tags visuais e emojis.""" + # Limpeza para narração + texto_limpo = text.replace("🤖", "").replace("🧑‍🏫", "").replace("*", "").replace("`", "") + # Remove o bloco se houver, pois ele é para leitura visual apenas + texto_limpo = re.sub(r'.*?', '', texto_limpo, flags=re.DOTALL).strip() + + # Se sobrar texto vazio após limpar o refinado (raro), fala algo genérico + if not texto_limpo: + texto_limpo = "Relatório processado. Os detalhes estão no painel visual." + + filename = f"audio_reply_{uuid.uuid4().hex[:8]}.mp3" + filepath = os.path.join("/tmp", filename) + + tts = gTTS(text=texto_limpo, lang='pt-br', tld='com.br', slow=False) + tts.save(filepath) + return filename diff --git a/docker-compose.yml b/docker-compose.yml index 3c13db2..1c1beef 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -9,7 +9,8 @@ services: - "8000" # Monta as credenciais e o socket do docker para o Bot conseguir comandar a VPS raiz! volumes: - - /var/run/docker.sock:/var/run/docker.sock:ro + - .:/app + - /var/run/docker.sock:/var/run/docker.sock:rw - /:/host_root:ro # Acesso em leitura à VPS para análise - ./data:/app/data # Configs dinâmicas (API Keys, etc) env_file: diff --git a/main.py b/main.py index 2441c14..e18f101 100644 --- a/main.py +++ b/main.py @@ -3,11 +3,12 @@ import psutil import subprocess import time import json -from fastapi import FastAPI, Request, Header, Depends, HTTPException, status -from fastapi.responses import HTMLResponse, JSONResponse +from fastapi import FastAPI, Request, Header, Depends, HTTPException, status, File, UploadFile +from fastapi.responses import HTMLResponse, JSONResponse, FileResponse from fastapi.templating import Jinja2Templates from dotenv import load_dotenv from starlette.concurrency import run_in_threadpool +import audio_handler from ai_agent import query_agent from config import get_config, save_config @@ -111,6 +112,43 @@ async def web_chat(message: dict, is_auth: bool = Depends(verify_password)): reply = await run_in_threadpool(query_agent, prompt=user_text) return JSONResponse(content={"reply": reply}) +@app.post("/api/chat-audio") +async def web_chat_audio(audio: UploadFile = File(...), is_auth: bool = Depends(verify_password)): + """Recebe áudio, transcreve, processa na IA e devolve texto + áudio de resposta.""" + temp_path = f"/tmp/{audio.filename}" + with open(temp_path, "wb") as buffer: + buffer.write(await audio.read()) + + try: + # Transcrição (STT) + user_text = await run_in_threadpool(audio_handler.transcribe_audio, temp_path) + + # IA (Processamento) + reply = await run_in_threadpool(query_agent, prompt=user_text) + + # Síntese (TTS) + audio_filename = await run_in_threadpool(audio_handler.text_to_speech, reply) + audio_url = f"/api/audio-file/{audio_filename}" + + return JSONResponse(content={ + "text": user_text, + "reply": reply, + "audio_url": audio_url + }) + except Exception as e: + return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500) + finally: + if os.path.exists(temp_path): + os.remove(temp_path) + +@app.get("/api/audio-file/{filename}") +async def get_audio_file(filename: str): + """Serve os arquivos de áudio temporários gerados pelo TTS.""" + filepath = os.path.join("/tmp", filename) + if os.path.exists(filepath): + return FileResponse(filepath, media_type="audio/mpeg") + raise HTTPException(status_code=404, detail="Arquivo de áudio não encontrado.") + @app.get("/api/test_llm") async def test_llm_speed(is_auth: bool = Depends(verify_password)): """Mede a velocidade de resposta da IA ativa.""" diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..7cfc0c1 --- /dev/null +++ b/start.sh @@ -0,0 +1,3 @@ +#!/bin/bash +uvicorn main:app --host 0.0.0.0 --port 8000 --reload & +python bot_logic.py diff --git a/templates/index.html b/templates/index.html index 16d47dd..116f358 100644 --- a/templates/index.html +++ b/templates/index.html @@ -7,6 +7,7 @@ +