""" Ponto de entrada principal do sistema de IA Chatbot """ import os import asyncio import argparse from typing import Optional from core.brain import Brain from core.personality import PersonalityFactory from adapters.api_adapter import APIAdapter from adapters.socket_adapter import SocketAdapter from tools.calculator import CalculatorTool from tools.system_tools import SystemTools from tools.search_tool import SearchTool from utils.logger import logger class ChatbotSystem: """ Sistema principal que orquestra todos os componentes """ def __init__(self): self.brain = Brain() self.api_adapter: Optional[APIAdapter] = None self.socket_adapter: Optional[SocketAdapter] = None self._setup_tools() self._setup_personality() def _setup_tools(self): """Registra ferramentas disponíveis""" # Calculadora calc = CalculatorTool() self.brain.register_tool( "calculator", lambda expression: calc.format_result(calc.calculate(expression)), "Realiza cálculos matemáticos. Parâmetro: expression (string)" ) # Ferramentas do sistema sys_tools = SystemTools() self.brain.register_tool( "get_time", lambda: sys_tools.get_time(), "Retorna hora atual" ) self.brain.register_tool( "get_date", lambda: sys_tools.get_date(), "Retorna data atual" ) self.brain.register_tool( "format_currency", lambda value, currency="BRL": sys_tools.format_currency(float(value), currency), "Formata valor como moeda. Parâmetros: value (float), currency (string, default BRL)" ) # Busca search = SearchTool() self.brain.register_tool( "web_search", lambda query: asyncio.run(search.search(query)), "Busca na web. Parâmetro: query (string)" ) logger.info("Ferramentas registradas", extra={"tools": list(self.brain.tools.keys())}) def _setup_personality(self): """Configura personalidade padrão""" personality = PersonalityFactory.support_agent() # A personalidade pode ser acessada e modificada via brain logger.info("Personalidade configurada", extra={"tone": personality.tone.value}) def setup_api(self): """Configura adapter de API REST""" self.api_adapter = APIAdapter(self.brain) logger.info("API adapter configurado") def setup_websocket(self): """Configura adapter WebSocket""" self.socket_adapter = SocketAdapter(self.brain) logger.info("WebSocket adapter configurado") async def chat_cli(self): """ Interface de chat via linha de comando Para testes e demonstração """ print("=" * 50) print("🤖 Assistente IA - Modo Interativo") print("=" * 50) print("Comandos: /sair, /limpar, /historico, /ajuda") print("-" * 50) user_id = "cli_user" session_id = f"cli_session_{os.urandom(4).hex()}" while True: try: message = input("\n👤 Você: ").strip() if not message: continue if message == "/sair": print("👋 Até logo!") break if message == "/limpar": await self.brain.memory.clear_history(session_id) print("🧹 Histórico limpo!") continue if message == "/historico": history = await self.brain.memory.get_history(session_id) print("\n📜 Histórico:") for msg in history[-10:]: role = "👤" if msg["role"] == "user" else "🤖" print(f"{role} {msg['content'][:100]}...") continue if message == "/ajuda": print(""" Comandos disponíveis: /sair - Encerrar conversa /limpar - Limpar histórico /historico - Ver últimas mensagens /ajuda - Mostrar esta ajuda """) continue # Processar mensagem print("\n🤖 Assistente: ", end="", flush=True) result = await self.brain.process( message=message, user_id=user_id, session_id=session_id ) print(result.response) print(f"\n⚡ Intenção: {result.intent} | Tokens: {result.tokens_used} | Tempo: {result.processing_time:.2f}s") except KeyboardInterrupt: print("\n👋 Até logo!") break except Exception as e: logger.error(f"Erro no CLI: {e}") print(f"\n❌ Erro: {e}") async def run_api_server(self, host: str = "0.0.0.0", port: int = 8000): """ Inicia servidor API com FastAPI """ try: from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn app = FastAPI(title="Chatbot IA API", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) self.setup_api() class ChatRequest(BaseModel): message: str user_id: str = "anonymous" session_id: Optional[str] = None context: Optional[dict] = None @app.post("/chat") async def chat(request: ChatRequest): """Endpoint principal de chat""" result = await self.api_adapter.handle_message(request.dict()) if not result["success"]: raise HTTPException(status_code=400, detail=result["error"]) return result @app.post("/chat/stream") async def chat_stream(request: ChatRequest): """Endpoint de streaming""" from fastapi.responses import StreamingResponse async def event_generator(): async for chunk in self.brain.stream_process( message=request.message, user_id=request.user_id, session_id=request.session_id ): yield f"data: {chunk}\n\n" yield "data: [DONE]\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream" ) @app.get("/health") async def health(): """Health check""" return {"status": "ok", "version": "1.0.0"} @app.get("/stats") async def stats(): """Estatísticas do sistema""" return { "tools": list(self.brain.tools.keys()), "memory_sessions": len(await self.brain.memory.get_all_sessions()) } logger.info(f"Iniciando servidor API em {host}:{port}") uvicorn.run(app, host=host, port=port) except ImportError: logger.error("FastAPI não instalado. Instale com: pip install fastapi uvicorn") raise async def run_websocket_server(self, host: str = "0.0.0.0", port: int = 8001): """ Inicia servidor WebSocket """ try: import websockets import json self.setup_websocket() async def handler(websocket, path): connection_id = f"ws_{os.urandom(4).hex()}" user_id = "ws_user" await self.socket_adapter.handle_connect( connection_id=connection_id, user_id=user_id, send_func=websocket.send ) try: async for message in websocket: data = json.loads(message) await self.socket_adapter.handle_message(connection_id, data) except websockets.exceptions.ConnectionClosed: pass finally: await self.socket_adapter.handle_disconnect(connection_id) logger.info(f"Iniciando servidor WebSocket em {host}:{port}") async with websockets.serve(handler, host, port): await asyncio.Future() # Run forever except ImportError: logger.error("websockets não instalado. Instale com: pip install websockets") raise def main(): """Função principal""" parser = argparse.ArgumentParser(description="Chatbot IA System") parser.add_argument( "mode", choices=["cli", "api", "websocket"], help="Modo de execução" ) parser.add_argument("--host", default="0.0.0.0", help="Host do servidor") parser.add_argument("--port", type=int, help="Porta do servidor") args = parser.parse_args() # Verificar API key if not os.getenv("OPENAI_API_KEY"): print("⚠️ AVISO: OPENAI_API_KEY não configurada!") print("Defina a variável de ambiente ou algumas funcionalidades não funcionarão.") system = ChatbotSystem() if args.mode == "cli": asyncio.run(system.chat_cli()) elif args.mode == "api": port = args.port or 8000 asyncio.run(system.run_api_server(args.host, port)) elif args.mode == "websocket": port = args.port or 8001 asyncio.run(system.run_websocket_server(args.host, port)) if __name__ == "__main__": main()