import time import json import requests import trafilatura import concurrent.futures from typing import List, Dict, Any from urllib.parse import urlparse from loguru import logger try: from ddgs import DDGS except ImportError: try: from duckduckgo_search import DDGS # fallback: nome antigo do pacote except ImportError: DDGS = None from modules.config import OPENROUTER_API_KEY, GROQ_API_KEY, MISTRAL_API_KEY, OPENROUTER_MODEL, config class NativeDeepResearch: """ Agente Nativo de Deep Research para Akira (OpenManus Clone). Não depende de APIs pagas de pesquisa autônoma (como o Manus), usa DDGS + Trafilatura + LLM para investigar e sintetizar relatórios completos. """ def __init__(self): # Configuração de Sessão Robusta para evitar "Connection pool is full" self.session = requests.Session() adapter = requests.adapters.HTTPAdapter( pool_connections=20, pool_maxsize=20, max_retries=3 ) self.session.mount("http://", adapter) self.session.mount("https://", adapter) # Prioridade: Mistral Direct -> OpenRouter -> Groq if MISTRAL_API_KEY: self.api_url = "https://api.mistral.ai/v1/chat/completions" self.api_key = MISTRAL_API_KEY self.model = "mistral-large-latest" elif OPENROUTER_API_KEY: self.api_url = "https://openrouter.ai/api/v1/chat/completions" self.api_key = OPENROUTER_API_KEY self.model = OPENROUTER_MODEL elif GROQ_API_KEY: self.api_url = "https://api.groq.com/openai/v1/chat/completions" self.api_key = GROQ_API_KEY self.model = getattr(config, 'GROQ_MODEL', 'groq/compound') else: self.api_url = "" self.api_key = "" self.model = "" def _call_llm(self, system_prompt: str, user_prompt: str, json_mode: bool = False) -> str: """Chamada direta ao LLM para evitar importações circulares com o LLMManager.""" if not self.api_key: return "" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "HTTP-Referer": "https://akira.softedge.ai", "X-Title": "Kiami Native Research" } model_to_use = self.model payload = { "model": model_to_use, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], "temperature": 0.3, "max_tokens": 3000 } if json_mode and "groq" not in self.api_url: payload["response_format"] = {"type": "json_object"} response = None try: response = self.session.post(self.api_url, headers=headers, json=payload, timeout=60) response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"] except Exception as e: logger.error(f"❌ [NATIVE RESEARCH] Erro no LLM: {e}") if response is not None and hasattr(response, 'text'): logger.error(f"Detalhes: {response.text[:500]}") return "" def brainstorm_queries(self, topic: str) -> List[str]: """Gera 3 a 4 sub-pesquisas otimizadas para motores de busca.""" sys_prompt = "És um assistente de pesquisa especializado. Dada uma instrução complexa, gera 3 queries de pesquisa no Google para investigar o tema profundamente. Retorna APENAS as queries, uma por linha, sem numeração." res = self._call_llm(sys_prompt, topic) if not res: return [topic] queries = [q.strip().strip('-').strip('1234567890.').strip() for q in res.split('\n') if q.strip()] # Evita demasiadas queries return queries[:3] if queries else [topic] def search_urls(self, queries: List[str]) -> List[str]: """Procura na web usando o DuckDuckGo e extrai URLs únicos.""" urls = set() try: with DDGS() as ddgs: for q in queries: try: logger.info(f"🔍 [NATIVE RESEARCH] Procurando por: {q}") results = list(ddgs.text(q, max_results=3)) for r in results: if isinstance(r, dict) and "href" in r: url = r["href"] # Filtra links inúteis if not any(x in url for x in ['youtube.com', 'facebook.com', 'instagram.com', 'tiktok.com']): urls.add(url) except Exception as e: logger.warning(f"⚠️ Erro ao procurar '{q}': {e}") time.sleep(1) # Pequena pausa em caso de rate limit except Exception as session_err: logger.error(f"❌ [NATIVE RESEARCH] Erro na sessão DDGS: {session_err}") return list(urls) def scrape_url(self, url: str) -> str: """Saca o texto limpo do site usando requests + trafilatura.""" try: # Uso da sessão com pool aumentado para estabilidade response = self.session.get(url, timeout=15, headers={ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' }) response.raise_for_status() if response.text: text = trafilatura.extract(response.text, include_comments=False, include_tables=True) if text: return f"--- CONTEÚDO DE {url} ---\n{text[:5000]}\n" except Exception as e: logger.debug(f"Falha ao ler {url}: {e}") return "" def synthesize_report(self, prompt: str, context: str) -> str: """Gera o relatório final baseado em todo o contexto lido.""" sys_prompt = ( "És a Akira, a IA incrivelmente inteligente do ecossistema SoftEdge.\n" "Foi-te dada uma tarefa de pesquisa profunda (Deep Research). Tens abaixo as notas extraídas " "da internet em bruto. O teu objectivo é ler tudo, sintetizar a verdade e redigir um " "relatório detalhado, claro e fenomenal para o utilizador.\n" "- Foca-te em dados precisos e atuais.\n" "- Ignora informações redundantes ou não relacionadas com a pergunta original.\n" "- O relatório DEVE ter parágrafos limpos, sem excesso de hashtags.\n" "- Se as notas não tiverem informação suficiente, responde com o que sabes e avisa que a pesquisa web não encontrou tudo." ) user_prompt = f"TAREFA ORIGINAL DO UTILIZADOR: {prompt}\n\nNOTAS EXTRAÍDAS DA WEB:\n{context}" res = self._call_llm(sys_prompt, user_prompt) return res def run(self, prompt: str) -> Dict[str, Any]: """Executa a rotina completa de pesquisa nativa (OpenManus flow).""" if not self.api_key: return {"sucesso": False, "erro": "Chave de API do LLM em falta para o Native Research."} start_time = time.time() logger.info("🧠 [NATIVE RESEARCH] Iniciando pipeline autónoma...") # 1. Planeamento queries = self.brainstorm_queries(prompt) logger.info(f"📋 Sub-pesquisas geradas: {queries}") # 2. Pesquisa de URLs urls = self.search_urls(queries) if not urls: return {"sucesso": False, "erro": "Não foi possível encontrar páginas web relevantes."} logger.info(f"🌐 URLs recolhidos ({len(urls)}). A extrair texto...") # 3. Scraping Paralelo (Velocidade) scraped_texts = [] with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: future_to_url = {executor.submit(self.scrape_url, url): url for url in urls} for future in concurrent.futures.as_completed(future_to_url): text = future.result() if text: scraped_texts.append(text) if not scraped_texts: return {"sucesso": False, "erro": "Os sites bloqueadores a leitura dos dados. Não foi possível extrair texto."} full_context = "\n".join(scraped_texts) logger.info(f"📚 Extraídos {len(full_context)} caracteres de conteúdo em bruto. A sintetizar...") # 4. Síntese Final final_report = self.synthesize_report(prompt, full_context) elapsed = int(time.time() - start_time) logger.info(f"✅ [NATIVE RESEARCH] Concluído com sucesso em {elapsed} segundos.") if final_report: return { "sucesso": True, "resultado": final_report, "prompt_original": prompt, "status": "concluído" } else: return {"sucesso": False, "erro": "Falha na síntese do relatório final."} # Instância partilhada native_research_agent = NativeDeepResearch()