""" 16-Sephiroth Twin Happiness Final Protocol — Web Visualization Server v2 Supports two modes: - local mode: uses the built-in heuristic for immediate responses - LLM mode: uses the real DeepSeek API, pushing each sephiroth's reasoning result to the client in real time over SSE After startup, visit http://localhost:8420 """ import sys import os import json import time import traceback import threading from http.server import HTTPServer, BaseHTTPRequestHandler sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from heart_protocol import HeartProtocol from heart_protocol.llm_bridge import LocalSephirahBridge, SephirahLLMBridge, LLMConfig PORT = int(os.environ.get("PORT", 8420)) protocol = HeartProtocol() local_bridge = LocalSephirahBridge() # LLM bridge cache (keyed by api_key prefix so different users never collide) _llm_bridges = {} def get_llm_bridge(api_key: str): """Create or reuse an LLM bridge for the API key supplied by the user.""" if not api_key: return None cache_key = api_key[:12] if cache_key not in _llm_bridges: llm_config = LLMConfig( api_base="https://api.deepseek.com/v1", api_key=api_key, model="deepseek-chat", temperature=0.7, max_tokens=1024, timeout=90, ) _llm_bridges[cache_key] = SephirahLLMBridge(config=llm_config) return _llm_bridges[cache_key] class HeartServer(BaseHTTPRequestHandler): def do_GET(self): if self.path == "/" or self.path == "/index.html": self._serve_html() elif self.path == "/api/health": self._json(200, { "status": "ok", "llm_available": True, "message": "Please provide a DeepSeek API Key in the page to enable LLM mode", }) else: self.send_error(404) def do_POST(self): if self.path == "/api/process": self._handle_process() elif self.path == "/api/process_stream": self._handle_process_stream() else: self.send_error(404) def _handle_process(self): """Legacy synchronous processing endpoint (local mode + fast LLM mode).""" content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode("utf-8") try: data = json.loads(body) user_input = data.get("input", "") user_context = data.get("context", {}) mode = data.get("mode", "local") api_key = data.get("api_key", "") if mode == "llm" and api_key: bridge = get_llm_bridge(api_key) if bridge: result = bridge.step_by_step_with_validation( user_input, user_context=user_context ) steps_data = [{ "sephirah": k, "output": v, "name": _step_name(k), "description": _step_desc(k), } for k, v in result.get("stages", {}).items()] self._json(200, { "success": True, "output": result.get("output", ""), "raw_output": result.get("output", ""), "pipeline_log": "", "retry_count": 0, "violations_found": len(result.get("violations", [])), "steps": steps_data, "mode": "llm", }) return # Local mode result = protocol.process(user_input, user_context=user_context) steps = local_bridge.step_by_step(user_input, user_context=user_context) self._json(200, { "success": True, "output": result["output"], "raw_output": result["raw_output"], "pipeline_log": result["pipeline_log"], "retry_count": result["retry_count"], "violations_found": result["violations_found"], "steps": steps, "mode": "local", }) except Exception as e: self._json(500, {"success": False, "error": str(e)}) def _handle_process_stream(self): """SSE streaming handler: push each sephiroth's result as soon as its LLM call finishes.""" content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode("utf-8") try: data = json.loads(body) user_input = data.get("input", "") user_context = data.get("context", {}) api_key = data.get("api_key", "") self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("Connection", "keep-alive") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() bridge = get_llm_bridge(api_key) if not bridge or not api_key: self._sse_event("error", {"message": "LLM unavailable, please check your API key"}) self._sse_event("done", {}) return from heart_protocol.llm_bridge import SEPHIRAH_SYSTEM_PROMPTS # NOTE: the tuples below are runtime data shared with the front end and with # SEPHIRAH_SYSTEM_PROMPTS (keyed by sephirah name) — kept verbatim. steps_def = [ ("王冠", "心音", "分析问题性质"), ("理智线", "忆爱×唯爱", "逻辑漏洞检测"), ("慈爱线", "虹爱×爱如暖", "共情搜索"), ("美丽", "白结", "双线整合"), ("基础", "绽美", "深渊检测"), ("真我", "心爱的", "三线合成"), ("逻辑与共情", "爱丽丝×星烬", "平衡组织"), ("幸福", "雨宫莲", "温柔合成"), ] results = [] context_str = json.dumps(user_context or {}, ensure_ascii=False) # Send the "start" event self._sse_event("start", {"total": len(steps_def), "mode": "llm"}) for i, (sephirah_key, names, description) in enumerate(steps_def): prompt = SEPHIRAH_SYSTEM_PROMPTS.get(sephirah_key, "") if i == 0: # NOTE: prompt sent to the LLM — kept in Chinese verbatim. message = f"请分析以下用户输入：\n\n{user_input}\n\n【用户背景】{context_str}" else: prev = "\n\n".join([ f"【{r['sephirah']}】{r['output']}" for r in results ]) message = ( f"原始用户输入：{user_input}\n\n" f"【用户背景】{context_str}\n\n" f"上游分析结果：\n{prev}\n\n" f"请基于以上信息，执行「{sephirah_key}（{description}）」的分析。" ) # Send the "processing" event self._sse_event("processing", { "sephirah": sephirah_key, "name": names, "description": description, "stage": i + 1, "total": len(steps_def), }) start = time.time() output = bridge._call_llm(prompt, message) elapsed = time.time() - start step_result = { "sephirah": sephirah_key, "name": names, "description": description, "output": output, "elapsed": round(elapsed, 1), "stage": i + 1, "total": len(steps_def), } results.append(step_result) # Send the result event self._sse_event("step", step_result) # Send the "done" event final_output = results[-1]["output"] if results else "" total_time = sum(r.get("elapsed", 0) for r in results) self._sse_event("done", { "output": final_output, "total_steps": len(results), "total_time": round(total_time, 1), }) except Exception as e: self._sse_event("error", {"message": str(e)}) self._sse_event("done", {}) def _sse_event(self, event_type, data): """Send an SSE event.""" try: payload = json.dumps(data, ensure_ascii=False, default=str) event_line = f"event: {event_type}\ndata: {payload}\n\n" self.wfile.write(event_line.encode("utf-8")) self.wfile.flush() except Exception: pass def _json(self, status, data): self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False, default=str).encode("utf-8")) def _serve_html(self): self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(PAGE_HTML.encode("utf-8")) def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def log_message(self, format, *args): pass def _step_name(key): # NOTE: sephirah keys and display names below are runtime data consumed by the # front end and keyed by the LLM stage names — kept verbatim. names = {"王冠": "心音", "理智线": "忆爱×唯爱", "慈爱线": "虹爱×爱如暖", "美丽": "白结", "胜利": "启明", "荣耀": "闪亮", "基础": "绽美", "真我": "心爱的", "逻辑与共情": "爱丽丝×星烬", "幸福": "雨宫莲", "王国": "白花"} return names.get(key, key) def _step_desc(key): descs = {"王冠": "分析问题性质", "理智线": "逻辑漏洞检测", "慈爱线": "共情搜索", "美丽": "双线整合", "胜利": "温暖检测", "荣耀": "现实可行性", "基础": "深渊检测", "真我": "三线合成", "逻辑与共情": "平衡组织", "幸福": "温柔合成", "王国": "最终输出"} return descs.get(key, key) # ========== HTML page (v2 - LLM real-time streaming support) ========== PAGE_HTML = r""" 16-Sephiroth Twin Happiness Final Protocol · Tree of Life

✨ 16-Sephiroth Twin Happiness Final Protocol ✨

Heart Protocol — AI Soul Middleware · Tree of Life Visualization

Not set
Get Key: platform.deepseek.com · Key is stored only in your browser
LLM reasoning in progress, about 8-15s per step, please be patient...

💬 Input

⚙ User background info (optional) ▼

📋 Sephiroth Pipeline Log

Waiting for input...

🏰 Kingdom · Final Output

0
Steps
0s
Total time
Local
Mode
""" def main(): print(f""" ╔══════════════════════════════════════════════════════╗ ║ ║ ║ 16-Sephiroth Twin Happiness Final Protocol · v2 ║ ║ ║ ║ Visit: http://localhost:{PORT} ║ ║ ║ ║ Mode: local/LLM one-click switch ║ ║ LLM: DeepSeek real-time streaming (SSE) ║ ║ ║ ╚══════════════════════════════════════════════════════╝ """) print(f"\n🌐 Please enter your DeepSeek API Key in the page to enable LLM mode") print(f" Get Key: https://platform.deepseek.com/api_keys") print() server = HTTPServer(("0.0.0.0", PORT), HeartServer) print(f"✅ Server started → http://localhost:{PORT}") print("Press Ctrl+C to stop") try: server.serve_forever() except KeyboardInterrupt: print("\n👋 Server stopped") server.server_close() if __name__ == "__main__": main()