# Main application entry point for NovaChat from flask import Flask, render_template, session, jsonify, request from flask_session import Session import logging import asyncio from datetime import datetime from ollama import AsyncClient, ChatResponse from typing import Any from werkzeug.exceptions import HTTPException from modules.llm.ollama_client import list_models, pull_model from modules.chat.chat_handler import process_message from modules.authentication import configure_auth from modules.search.news import fetch_news from modules.search.arxiv import fetch_arxiv from modules.search.web import fetch_ddg, fetch_torrents # Configure logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) # Create file handler which logs even debug messages fh = logging.FileHandler('app.log') fh.setLevel(logging.DEBUG) # Create formatter and add it to the handlers formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') fh.setFormatter(formatter) # Add the handlers to the logger logger.addHandler(fh) app = Flask(__name__, template_folder='frontend/templates') app.config['SECRET_KEY'] = 'your-secret-key' app.config['SESSION_TYPE'] = 'filesystem' Session(app) # Register error handlers @app.errorhandler(Exception) def handle_exception(e): # Log the error logger.error(f"Unhandled Exception: {str(e)}", exc_info=True) # Handle HTTP exceptions if isinstance(e, HTTPException): response = { "error": e.name, "message": e.description, "status": e.code, "timestamp": datetime.now().isoformat() } return jsonify(response), e.code # Handle other exceptions response = { "error": "Internal Server Error", "message": "An unexpected error occurred", "status": 500, "timestamp": datetime.now().isoformat() } return jsonify(response), 500 @app.route('/') def index(): if 'chat_history' not in session: session['chat_history'] = [] return render_template('index.html', chat_history=session['chat_history']) @app.route('/api/models') def get_models(): # Get configured required models required_models = config['ollama'].get('required_models', []) # Get available models from Ollama available_models = list_models() # Check and pull missing models for model in required_models: if model not in available_models: try: pull_model(model) available_models.append(model) app.logger.info(f'Auto-pulled required model: {model}') except Exception as e: app.logger.error(f'Failed to auto-pull {model}: {str(e)}') # Set default model in session if not set if 'selected_model' not in session and required_models: session['selected_model'] = required_models[0] return jsonify({ 'available': available_models, 'required': required_models, 'selected': session.get('selected_model') }) @app.route('/api/pull-model', methods=['POST']) def add_model(): model_name = request.json.get('model') if not model_name: return jsonify({'error': 'Model name required'}), 400 try: pull_model(model_name) return jsonify({'message': f'Model {model_name} pulled successfully'}), 200 except Exception as e: app.logger.error(f'Model pull failed: {str(e)}') return jsonify({ 'error': 'Model deployment failed', 'details': str(e), 'resolution': 'Check model name validity and network connection' }), 500 import yaml with open('config.yaml') as f: config = yaml.safe_load(f) # Initialize authentication configure_auth(app) # ——— System prompts ——— # ——— Compatibility patch for asyncio.to_thread ——— if not hasattr(asyncio, 'to_thread'): def _to_thread(func, *args, **kwargs): loop = asyncio.get_event_loop() return loop.run_in_executor(None, lambda: func(*args, **kwargs)) asyncio.to_thread = _to_thread # ——— System prompts and session management ——— from modules.system_prompts import system_prompts MODE = "COGITO" # ——— Tool functions and descriptors ——— # ——— Chat endpoints ——— @app.route('/chat', methods=['POST']) async def chat() -> Any: if not request.is_json: return jsonify(error="Content-Type must be application/json"), 415 try: request_data = request.get_json() except Exception as e: app.logger.error(f"Invalid JSON: {str(e)}") return jsonify(error="Invalid JSON format"), 400 global MODE user_message = request_data.get('message', '').strip() app.logger.info("User ▶ %s", user_message) if not user_message: return jsonify(error="No message provided"), 400 # Mode toggles if user_message == "Enable flirty mode": MODE = "FLIRTY" system_prompts.init_chat_history("FLIRTY") elif user_message == "Enable companion mode": MODE = "COMPANION" system_prompts.init_chat_history("COMPANION") elif user_message == "Enable investigative mode": MODE = "INVESTIGATE" system_prompts.init_chat_history("INVESTIGATE") elif user_message == "Back to normal.": MODE = "COGITO" system_prompts.init_chat_history("COGITO") else: if 'chat_history' not in session: init_chat_history(MODE) session['chat_history'].append({"role": "user", "content": user_message}) chat_history = session['chat_history'] model_name = "hf.co/openerotica/..." if MODE == "FLIRTY" else "huihui_ai/qwen2.5-1m-abliterated" client = AsyncClient() response: ChatResponse = await client.chat( model=model_name, messages=chat_history, tools=[fetch_news_tool, fetch_arxiv_tool, fetch_ddg_tool, fetch_torrents_tool], options={"temperature": 0.4} ) assistant_text = response.message.content # Process tool calls if response.message.tool_calls: for call in response.message.tool_calls: fname = call.function.name args = call.function.arguments or {} if fname in available_functions: app.logger.info("Invoking %s(%s)", fname, args) result = await asyncio.to_thread(available_functions[fname], **args) session['chat_history'].append({"role": "tool", "name": fname, "content": "\n\n".join(result)}) # Rerun with tool outputs follow_up: ChatResponse = await client.chat( model=model_name, messages=session['chat_history'], options={"temperature": 0.4} ) assistant_text = follow_up.message.content session['chat_history'].append({"role": "assistant", "content": assistant_text}) session.modified = True app.logger.info("Nova ▶ %s", assistant_text) return jsonify(chat_history=session['chat_history']) from flask import session SYSTEM_PROMPT = "You are a helpful AI assistant." def init_chat_history(mode="COGITO"): """Initialize chat history with a system message (hidden from UI).""" session['chat_history'] = [ {"role": "system", "content": SYSTEM_PROMPT} ] session['chat_history'].append({"role": "system", "content": f"Mode set to {mode}."}) @app.route('/clear', methods=['POST']) def clear_history() -> Any: init_chat_history(MODE) return jsonify(chat_history=session['chat_history']) from modules.tools import fetch_news_tool, fetch_arxiv_tool, fetch_ddg_tool, fetch_torrents_tool, available_functions if __name__ == '__main__': logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # print("Starting Nova server on http://localhost:9125") app.run(host=config['server']['host'], port=config['server']['port'], debug=False, use_reloader=False)