import os import json import sqlite3 import requests from flask import Flask, render_template, request, jsonify, send_from_directory from werkzeug.utils import secure_filename app = Flask(__name__) app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB max upload app.config['UPLOAD_FOLDER'] = os.path.join(app.instance_path, 'uploads') os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True) app.secret_key = os.urandom(24) # Error Handlers @app.errorhandler(413) def request_entity_too_large(error): return jsonify({"error": "File too large (Max 16MB)"}), 413 @app.errorhandler(404) def page_not_found(error): return render_template('index.html'), 200 # SPA fallback @app.errorhandler(500) def internal_error(error): return jsonify({"error": "Internal Server Error"}), 500 # Database Setup DB_PATH = os.path.join(app.instance_path, 'echo_mimic.db') os.makedirs(app.instance_path, exist_ok=True) def init_db(): conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS personas (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, role TEXT, bio TEXT, traits JSON, avatar_style TEXT, avatar_path TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''') # Check for schema migration (add avatar_path if missing) c.execute("PRAGMA table_info(personas)") columns = [info[1] for info in c.fetchall()] if 'avatar_path' not in columns: print("Migrating database: adding avatar_path column") c.execute("ALTER TABLE personas ADD COLUMN avatar_path TEXT") conn.commit() # Check if empty and seed c.execute("SELECT count(*) FROM personas") if c.fetchone()[0] == 0: seed_data = [ ("Einstein (Demo)", "Physicist", "The father of relativity.", json.dumps({"Openness": 95, "Conscientiousness": 80, "Extraversion": 40, "Agreeableness": 70, "Neuroticism": 30}), "sketch", ""), ("Sherlock (Demo)", "Detective", "High-functioning sociopath.", json.dumps({"Openness": 90, "Conscientiousness": 95, "Extraversion": 20, "Agreeableness": 10, "Neuroticism": 60}), "realistic", "") ] c.executemany("INSERT INTO personas (name, role, bio, traits, avatar_style, avatar_path) VALUES (?, ?, ?, ?, ?, ?)", seed_data) conn.commit() print("Database seeded with default personas.") conn.commit() conn.close() init_db() # SiliconFlow API Configuration SF_API_KEY = "sk-vimuseiptfbomzegyuvmebjzooncsqbyjtlddrfodzcdskgi" SF_API_URL = "https://api.siliconflow.cn/v1/chat/completions" MODEL_NAME = "Qwen/Qwen2.5-7B-Instruct" def call_silicon_flow(messages, temperature=0.7): headers = { "Authorization": f"Bearer {SF_API_KEY}", "Content-Type": "application/json" } payload = { "model": MODEL_NAME, "messages": messages, "temperature": temperature, "max_tokens": 1024 } try: response = requests.post(SF_API_URL, json=payload, headers=headers, timeout=10) response.raise_for_status() data = response.json() return data['choices'][0]['message']['content'] except Exception as e: print(f"SiliconFlow Error: {e}") return None @app.route('/') def index(): return render_template('index.html') @app.route('/api/generate_persona', methods=['POST']) def generate_persona(): data = request.json desc = data.get('description', 'A helpful assistant') system_prompt = """ You are an expert Character Designer. Based on the user's description, generate a detailed persona profile in JSON format. Return ONLY the JSON object, no markdown, no other text. JSON Structure: { "name": "Name", "role": "Job Title / Role", "bio": "A short 2-sentence biography.", "traits": { "Openness": 1-100, "Conscientiousness": 1-100, "Extraversion": 1-100, "Agreeableness": 1-100, "Neuroticism": 1-100 }, "speaking_style": "Keywords describing how they talk (e.g. formal, slang, poetic)", "catchphrases": ["phrase 1", "phrase 2"] } """ messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": f"Create a persona for: {desc}"} ] content = call_silicon_flow(messages, temperature=0.8) if content: try: # Clean up potential markdown code blocks if "```json" in content: content = content.split("```json")[1].split("```")[0].strip() elif "```" in content: content = content.split("```")[1].strip() persona = json.loads(content) return jsonify({"status": "success", "data": persona}) except json.JSONDecodeError: return jsonify({"status": "error", "message": "Failed to parse AI response"}), 500 else: # Fallback Mock Data return jsonify({ "status": "success", "data": { "name": "Nova (Mock)", "role": "AI Assistant", "bio": "A fallback persona generated because the API call failed.", "traits": {"Openness": 80, "Conscientiousness": 90, "Extraversion": 50, "Agreeableness": 85, "Neuroticism": 20}, "speaking_style": "Polite and direct", "catchphrases": ["How can I help?", "Processing request."] } }) @app.route('/api/chat', methods=['POST']) def chat(): data = request.json message = data.get('message') history = data.get('history', []) persona = data.get('persona', {}) if not message: return jsonify({"error": "No message provided"}), 400 # Construct System Prompt from Persona sys_prompt = f""" You are roleplaying as {persona.get('name', 'AI')}. Role: {persona.get('role', 'Assistant')} Bio: {persona.get('bio', '')} Speaking Style: {persona.get('speaking_style', 'Normal')} Your personality traits (1-100): - Openness: {persona.get('traits', {}).get('Openness', 50)} - Conscientiousness: {persona.get('traits', {}).get('Conscientiousness', 50)} - Extraversion: {persona.get('traits', {}).get('Extraversion', 50)} - Agreeableness: {persona.get('traits', {}).get('Agreeableness', 50)} - Neuroticism: {persona.get('traits', {}).get('Neuroticism', 50)} Stay in character at all times. Keep responses concise and engaging. """ messages = [{"role": "system", "content": sys_prompt}] # Add last 10 messages for context for msg in history[-10:]: messages.append({"role": msg['role'], "content": msg['content']}) messages.append({"role": "user", "content": message}) response_text = call_silicon_flow(messages) if not response_text: response_text = f"[{persona.get('name')} is offline - Mock Mode] I heard you say: {message}" return jsonify({"response": response_text}) @app.route('/api/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({"error": "No file part"}), 400 file = request.files['file'] if file.filename == '': return jsonify({"error": "No selected file"}), 400 if file: filename = secure_filename(file.filename) # Handle non-ascii filenames if not filename: filename = "uploaded_file_" + os.urandom(4).hex() file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) file.save(file_path) return jsonify({"status": "success", "path": f"/uploads/{filename}", "filename": filename}) @app.route('/uploads/') def uploaded_file(filename): return send_from_directory(app.config['UPLOAD_FOLDER'], filename) @app.route('/api/save_persona', methods=['POST']) def save_persona(): data = request.json try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute("INSERT INTO personas (name, role, bio, traits, avatar_style, avatar_path) VALUES (?, ?, ?, ?, ?, ?)", (data['name'], data['role'], data['bio'], json.dumps(data['traits']), data.get('avatar_style', 'default'), data.get('avatar_path', ''))) conn.commit() conn.close() return jsonify({"status": "success"}) except Exception as e: return jsonify({"status": "error", "message": str(e)}), 500 @app.route('/api/personas', methods=['GET']) def get_personas(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row c = conn.cursor() c.execute("SELECT * FROM personas ORDER BY created_at DESC") rows = c.fetchall() conn.close() personas = [] for row in rows: p = dict(row) p['traits'] = json.loads(p['traits']) personas.append(p) return jsonify({"status": "success", "data": personas}) if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)