Companion AI

#!/usr/bin/env python3 """ AI VRM Companion — Minimal & Production-Ready """ import os, io, json, asyncio, time, logging from flask import Flask, request, send_from_directory, Response, jsonify import edge_tts from llama_cpp import Llama app = Flask(__name__) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger(__name__) # ── Paths ── BASE = os.path.dirname(os.path.abspath(__file__)) MODEL_DIR = os.path.join(BASE, "model") ANIM_DIR = os.path.join(BASE, "animation") # Log what we have log.info(f"BASE: {BASE}") log.info(f"MODEL_DIR exists: {os.path.exists(MODEL_DIR)}") log.info(f"ANIM_DIR exists: {os.path.exists(ANIM_DIR)}") if os.path.exists(MODEL_DIR): log.info(f"MODEL files: {os.listdir(MODEL_DIR)}") if os.path.exists(ANIM_DIR): log.info(f"ANIM files: {os.listdir(ANIM_DIR)}") # ── LLM ── log.info("Loading MiniCPM-1B Q4_K_M ...") t0 = time.time() llm = Llama( model_path="/app/minicpm5-1b-Q4_K_M.gguf", n_ctx=2048, n_threads=min(os.cpu_count() or 2, 4), n_batch=2048, verbose=False, use_mmap=True, use_mlock=False, f16_kv=True, ) log.info(f"Model loaded in {time.time()-t0:.1f}s") # ── TTS ── async def tts_to_buffer(text: str, voice="zh-CN-XiaoyiNeural") -> io.BytesIO: buf = io.BytesIO() async for chunk in edge_tts.Communicate(text, voice).stream(): if chunk["type"] == "audio": buf.write(chunk["data"]) buf.seek(0) return buf # ── Routes ── @app.route("/model/") def model(fn): return send_from_directory(MODEL_DIR, fn, max_age=2592000) @app.route("/animation/") def animation(fn): return send_from_directory(ANIM_DIR, fn, max_age=2592000) @app.route("/tts") def tts_route(): text = request.args.get("text", "") if not text: return "Missing text", 400 try: buf = asyncio.run(tts_to_buffer(text)) return Response(buf.read(), mimetype="audio/mpeg") except Exception as e: return str(e), 500 @app.route("/chat", methods=["POST"]) def chat(): data = request.get_json(force=True) or {} prompt = data.get("prompt", "").strip() exprs = data.get("expressions", []) if not prompt: return jsonify({"error": "prompt required"}), 400 system = ( "You are a cute anime companion. Reply in JSON ONLY. " "Format: {\"text\":\"your spoken reply\",\"expressions\":{...}}. " f"Available expressions (0-1 values): {', '.join(exprs)}. " "Pick 1-3 expressions that match your mood. Keep text short (1-2 sentences)." ) full_prompt = f"{system}\n\nUser: {prompt}\nCompanion:" t0 = time.time() out = llm(full_prompt, max_tokens=150, temperature=0.8, top_p=0.9, stop=["User:", "\n\n"]) raw = out["choices"][0]["text"].strip() tok = out.get("usage", {}).get("completion_tokens", 0) if tok: log.info(f"Gen: {tok} tokens in {time.time()-t0:.2f}s ({tok/(time.time()-t0):.1f} tok/s)") try: if "```" in raw: raw = raw.split("```")[-2] if raw.count("```") >= 2 else raw.replace("```json", "").replace("```", "") result = json.loads(raw.strip()) if "text" not in result: raise ValueError("no text key") except Exception: result = {"text": raw or "I'm here for you!", "expressions": {"neutral": 1.0}} return jsonify(result) @app.route("/health") def health(): return jsonify({"ok": True}) @app.route("/") def index(): return Response(HTML, mimetype="text/html") # ── Frontend ── HTML = r""" Companion AI

""" if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) app.run(host="0.0.0.0", port=port, debug=False, threaded=True)