""" Motor principal de processamento da IA """ import json import asyncio from typing import Dict, List, Optional, Any, AsyncGenerator from dataclasses import dataclass from config.settings import settings from config.prompts import SYSTEM_PROMPT, CLASSIFICATION_PROMPT, TOOL_USE_PROMPT from core.memory import ConversationMemory from core.context import ContextManager from services.llm_service import LLMService from services.knowledge_base import KnowledgeBase @dataclass class ProcessingResult: """Resultado do processamento""" response: str intent: str used_tools: List[str] confidence: float tokens_used: int processing_time: float class Brain: """ Cérebro central da IA - orquestra todo o processamento """ def __init__(self): self.llm = LLMService(settings.llm) self.memory = ConversationMemory(settings.memory) self.context = ContextManager() self.knowledge = KnowledgeBase() self.tools: Dict[str, Any] = {} def register_tool(self, name: str, tool_func: callable, description: str): """Registra uma ferramenta disponível""" self.tools[name] = { "function": tool_func, "description": description } async def process( self, message: str, user_id: str, session_id: Optional[str] = None, system_context: Optional[str] = None ) -> ProcessingResult: """ Processa uma mensagem do usuário Fluxo: 1. Classifica intenção 2. Recupera contexto/memória 3. Busca conhecimento relevante 4. Decide uso de ferramentas 5. Gera resposta 6. Armazena na memória """ import time start_time = time.time() # 1. Classificar intenção intent = await self._classify_intent(message) # 2. Recuperar memória e contexto history = await self.memory.get_history(session_id or user_id) context = self.context.get_context(user_id) # 3. Buscar conhecimento relevante relevant_info = await self.knowledge.search(message, top_k=3) # 4. Verificar necessidade de ferramentas used_tools = [] tool_result = None if intent in ["acao_sistema", "calculo", "busca_informacao"]: tool_result = await self._execute_tools(message) used_tools = [t["name"] for t in tool_result] if tool_result else [] # 5. Construir prompt e gerar resposta prompt = self._build_prompt( message=message, history=history, context=context, relevant_info=relevant_info, tool_results=tool_result, system_context=system_context ) response, tokens = await self.llm.generate(prompt) # 6. Armazenar na memória await self.memory.add_message( session_id or user_id, role="user", content=message ) await self.memory.add_message( session_id or user_id, role="assistant", content=response ) processing_time = time.time() - start_time return ProcessingResult( response=response, intent=intent, used_tools=used_tools, confidence=0.95, tokens_used=tokens, processing_time=processing_time ) async def stream_process( self, message: str, user_id: str, session_id: Optional[str] = None ) -> AsyncGenerator[str, None]: """ Processa mensagem com streaming de resposta """ history = await self.memory.get_history(session_id or user_id) relevant_info = await self.knowledge.search(message, top_k=3) prompt = self._build_prompt( message=message, history=history, context={}, relevant_info=relevant_info, tool_results=None, system_context=None ) full_response = [] async for chunk in self.llm.generate_stream(prompt): full_response.append(chunk) yield chunk # Salvar após completar complete_response = "".join(full_response) await self.memory.add_message(session_id or user_id, "user", message) await self.memory.add_message(session_id or user_id, "assistant", complete_response) async def _classify_intent(self, message: str) -> str: """Classifica a intenção da mensagem""" prompt = CLASSIFICATION_PROMPT.format(message=message[:500]) result, _ = await self.llm.generate(prompt, max_tokens=50) return result.strip().lower() async def _execute_tools(self, message: str) -> Optional[List[Dict]]: """Executa ferramentas necessárias""" if not self.tools: return None tools_desc = "\n".join([ f"- {name}: {info['description']}" for name, info in self.tools.items() ]) prompt = TOOL_USE_PROMPT.format( tools=tools_desc, message=message ) result, _ = await self.llm.generate(prompt, max_tokens=200) # Parse resposta da ferramenta if "TOOL:" in result: lines = result.strip().split("\n") tool_name = None params = {} for line in lines: if line.startswith("TOOL:"): tool_name = line.replace("TOOL:", "").strip() elif line.startswith("PARAMS:"): try: params = json.loads(line.replace("PARAMS:", "").strip()) except: pass if tool_name and tool_name in self.tools: try: output = await self.tools[tool_name]["function"](**params) return [{"name": tool_name, "output": output}] except Exception as e: return [{"name": tool_name, "error": str(e)}] return None def _build_prompt( self, message: str, history: List[Dict], context: Dict, relevant_info: List[str], tool_results: Optional[List[Dict]], system_context: Optional[str] ) -> str: """Constrói o prompt completo""" # Formatar histórico history_str = "" for msg in history[-settings.memory.max_history:]: role = "Usuário" if msg["role"] == "user" else "Assistente" history_str += f"{role}: {msg['content']}\n" # Formatar informações relevantes info_str = "\n".join(relevant_info) if relevant_info else "Nenhuma informação específica." # Adicionar resultados de ferramentas if tool_results: info_str += "\n\nResultados de ferramentas:\n" for tool in tool_results: if "error" in tool: info_str += f"- {tool['name']}: Erro - {tool['error']}\n" else: info_str += f"- {tool['name']}: {tool['output']}\n" system = SYSTEM_PROMPT.format( name=settings.system.name, system_context=system_context or "Sistema operacional normal.", conversation_history=history_str, relevant_info=info_str ) return f"{system}\n\nUsuário: {message}\n\nAssistente:"