"""
虫群Swarm Web界面 — 纯FastAPI+HTML，零额外依赖
功能：
- 聊天对话（本地记忆+API推理+MOA聚合）
- 系统状态面板
- 记忆管理
"""
import os, sys, time, json, asyncio
from typing import Optional
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
app = FastAPI(title="虫群Swarm", version="8.0")
# ============================================================
# 全局状态
# ============================================================
class SwarmAppState:
def __init__(self):
self.node = None
self.initialized = False
self.chat_history = []
def initialize(self):
if self.initialized:
return True
try:
from core.swarm_node import SwarmNode
from core.aggregation_protocol.types import PermissionLevel
self.node = SwarmNode(
node_id="web_user",
name="Web用户",
permission_level=PermissionLevel.QUEEN,
model_config="tiny",
initial_balance=100.0,
)
self.node.start()
self.initialized = True
return True
except Exception as e:
print(f"[WebApp] 初始化失败: {e}")
import traceback; traceback.print_exc()
return False
state = SwarmAppState()
# ============================================================
# HTML页面
# ============================================================
HTML_PAGE = """
虫群 Swarm
"""
# ============================================================
# API路由
# ============================================================
@app.get("/", response_class=HTMLResponse)
async def index():
return HTML_PAGE
@app.get("/api/status")
async def get_status():
if not state.initialized:
state.initialize()
if not state.initialized or not state.node:
return {"ready": False}
node = state.node
mem_status = node.memory.get_status()
api_count = len(node.api_manager.models) if node._api_models_available else 0
balance = node.get_balance()
return {
"ready": True,
"model": mem_status.get("model_config", "tiny"),
"api_models": api_count,
"memories": mem_status.get("total_memories", 0),
"balance": balance,
"queries": node.stats.get("queries_processed", 0),
}
@app.post("/api/chat")
async def chat(req: dict):
if not state.initialized:
state.initialize()
if not state.initialized or not state.node:
return {"response": "系统初始化失败", "source": "error", "confidence": 0}
message = req.get("message", "")
use_api = req.get("use_api", True)
use_moa = req.get("use_moa", True)
# 存储用户消息
state.chat_history.append({"role": "user", "content": message})
# 智能查询
result = state.node.smart_query(
message, use_api=use_api, use_moa=use_moa
)
# 存储AI回答到记忆
if result.get("response"):
state.node.store_memory(message, result["response"], importance=0.5)
state.chat_history.append({"role": "assistant", "content": result["response"]})
return result
@app.post("/api/memory")
async def store_memory(req: dict):
"""手动存储记忆"""
if not state.initialized or not state.node:
return {"ok": False}
mid = state.node.store_memory(
req.get("question", ""),
req.get("answer", ""),
importance=req.get("importance", 0.5)
)
return {"ok": True, "memory_id": mid}
@app.get("/api/memory/stats")
async def memory_stats():
if not state.initialized or not state.node:
return {"total": 0}
return state.node.memory.get_status()
# 启动
if __name__ == "__main__":
import uvicorn
print("=" * 50)
print("🦠 虫群 Swarm Web界面")
print("=" * 50)
uvicorn.run(app, host="0.0.0.0", port=7860)