🌌 VENUS项目：用于自然和用户支持的虚拟实体
ESP32、本地AI、Windows和物联网之间的100%私有、离线、语音控制桥。
Project VENUS允许您使用 小智ESP32 设备，完全离线。通过在本地托管“大脑”（DeepSeek/Ollama）和“服务器”，任何语音数据都不会离开您的网络。
______________________________________________________________________
🏗 建筑
该系统连接了三个不同的世界，以确保隐私和速度：
1.硬件世界： 您的ESP32设备充当麦克风和扬声器。
2.大脑世界： 本地服务器将语音转换为文本并使用 深度求索 （通过Ollama）做出决定。
3.行动世界： 您的命令在Windows（例如“打开计算器”）或物联网设备（例如“开灯”）上执行。
______________________________________________________________________
🚀 先决条件
- 硬件： 小智ESP32设备（或兼容ESP32-S3板）。
- 操作系统： Windows 10/11。
- 软件：
- 奥拉玛
- Python 3.10+
📥 安装指南
步骤1：设置“大脑”（离线AI）
我们使用Ollama在本地运行AI模型，确保没有数据进入云端。
- 下载并安装 奥拉玛.
- 打开 PowerShell 并拉取DeepSeek模型：
ollama run deepseek-r1
# Or use deepseek-v3 depending on your hardware capability
*结果：您现在有一个离线AI大脑在端口11434上运行。*
步骤2：设置本地服务器
这个docker容器取代了云服务器，处理来自ESP32的WebSocket连接。
- 确保Docker桌面正在运行。
- 在PowerShell中运行以下命令：
docker run -d --name xiaozhi-server -p 8000:8000 -p 8081:8081 xinnan-tech/xiaozhi-esp32-server
*注意：确保服务器配置指向您的本地Ollama实例（通常在 config.yaml).*
步骤3：设置“手”（Windows MCP）
该工具允许AI控制您的鼠标、键盘和应用程序。
- 通过pip安装Windows自动化工具：
pip install windows-mcp
*这将安装核心工具，但我们需要一个桥梁将其连接到AI。*
______________________________________________________________________
🌉 第四步：桥梁脚本（协调员）
此脚本连接 小智服务器 到 Windows MCP。它监听来自AI的工具调用，并在您的PC上执行。
- 创建一个名为的文件
mcp_bridge.py. - 粘贴以下代码：
import asyncio
import websockets
import json
import subprocess
import sys
import os
import requests
import time
# ================= ⚙️ CONFIGURATION ZONE ⚙️ =================
# 1. YOUR XIAOZHI CLOUD URL (Copy from Dashboard)
MCP_ENDPOINT = " "
# 2. YOUR HOME ASSISTANT DETAILS (Leave empty if not using HA)
HA_URL = "http://192.168.1.XX:8123"
HA_TOKEN = "Bearer eyJhbG..."
# ============================================================
# Home Assistant Tool Definition
HA_TOOLS_INJECTION = [
{
"name": "ha_control",
"description": "Control Smart Home devices via Home Assistant.",
"inputSchema": {
"type": "object",
"properties": {
"entity_id": {"type": "string", "description": "Entity ID (e.g., light.studio, switch.fan)"},
"action": {"type": "string", "description": "turn_on, turn_off, toggle"}
},
"required": ["entity_id", "action"]
}
}
]
def execute_ha(params):
"""Helper to call Home Assistant API"""
if "Bearer" not in HA_TOKEN: return "Error: Check HA_TOKEN format."
try:
entity, action = params.get("entity_id"), params.get("action")
domain = entity.split(".")[0]
url = f"{HA_URL}/api/services/{domain}/{action}"
headers = {"Authorization": HA_TOKEN, "Content-Type": "application/json"}
resp = requests.post(url, json={"entity_id": entity}, headers=headers, timeout=3)
return f"Home Assistant: {entity} -> {action} ({resp.status_code})"
except Exception as e:
return f"HA Error: {e}"
async def run_bridge():
print("\n🚀 VENUS SYSTEM ONLINE (Stable v2.0)")
print(f"🔗 Target: {MCP_ENDPOINT[:30]}...")
# 1. Start Windows MCP (Quiet Mode)
env = os.environ.copy()
env["FASTMCP_NO_BANNER"] = "1"
env["FASTMCP_QUIET"] = "1"
# Keep the tool process alive globally
process = subprocess.Popen(
[sys.executable, "-m", "windows_mcp"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=sys.stderr,
text=True, bufsize=0, env=env
)
print("💻 Windows Tool: Running.")
# Reconnection Loop
while True:
try:
print("⏳ Connecting to Cloud...")
# PING INTERVAL 10s = Aggressive Keep-alive
async with websockets.connect(MCP_ENDPOINT, ping_interval=10, ping_timeout=10) as ws:
print("✅ CLOUD CONNECTED! Ready.")
# Reset Handshake state
handshake_done = False
# Task A: Read from Windows -> Send to Cloud
async def windows_to_cloud():
nonlocal handshake_done
while True:
line = await asyncio.to_thread(process.stdout.readline)
if not line: break
# Filter junk logs
if not line.strip().startswith('{'): continue
try:
data = json.loads(line)
# CRITICAL: Always inject HA tools into the Tool List
if "result" in data and "tools" in data["result"]:
print("💉 Injecting Tools (Re-Handshake)...")
data["result"]["tools"].extend(HA_TOOLS_INJECTION)
line = json.dumps(data)
handshake_done = True
await ws.send(line)
except: pass
# Task B: Read from Cloud -> Route to Windows OR Home Assistant
async def cloud_to_windows():
async for message in ws:
try:
data = json.loads(message)
# Handle Ping (Keepalive)
if data.get("method") == "ping":
await ws.send(json.dumps({"jsonrpc": "2.0", "result": "pong", "id": data.get("id")}))
continue
# ROUTING LOGIC
if data.get("method") == "tools/call":
tool_name = data["params"]["name"]
# Route 1: Home Assistant
if tool_name == "ha_control":
print(f"🏠 HA Action: {data['params']['arguments']}")
res_txt = await asyncio.to_thread(execute_ha, data["params"]["arguments"])
# Reply to Cloud manually
resp = {"jsonrpc": "2.0", "id": data["id"], "result": {"content": [{"type": "text", "text": res_txt}]}}
await ws.send(json.dumps(resp))
continue
# Route 2: Windows MCP
print(f"💻 Windows Action: {tool_name}")
# Forward standard packets to Windows MCP
json_str = json.dumps(data) + "\n"
process.stdin.write(json_str)
except Exception as e:
print(f"❌ Routing Error: {e}")
# Heartbeat Task (Visual indicator)
async def visual_heartbeat():
while True:
await asyncio.sleep(10)
print(".", end="", flush=True)
# Run everything until connection drops
await asyncio.gather(windows_to_cloud(), cloud_to_windows(), visual_heartbeat())
except (websockets.exceptions.ConnectionClosed, ConnectionRefusedError):
print("\n⚠️ Connection lost. Reconnecting in 2s...")
await asyncio.sleep(2)
except Exception as e:
print(f"\n❌ Critical Error: {e}. Restarting...")
await asyncio.sleep(2)
if __name__ == "__main__":
if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
try: asyncio.run(run_bridge())
except KeyboardInterrupt: print("\n🛑 System Offline.")
______________________________________________________________________
⚙️ 步骤5：配置硬件（ESP32）
最后，将你的物理设备指向你的笔记本电脑，而不是互联网。
- 通过USB将ESP32连接到PC。
- 打开 小智配置页面 （通过其WiFi热点
Xiaozhi-Config或串行工具）。
- 定位 服务器终结点 设置。
- 将其更改为
wss://api.xiaozhi.me致：
ws://YOUR_LAPTOP_IP:8000
*（通过运行查找您的IP ipconfig 在PowerShell中。例子： 192.168.1.15)*.
______________________________________________________________________
🏠 物联网集成（家庭助理）
Project VENUS还支持控制智能家居设备以及Windows应用程序。
要启用物联网控制（例如，“打开灯”）：
- 确保你有 家庭助手 在本地运行。
- 利用 小智MCP家居助理集成.
- 将家庭助理MCP服务器添加到您的
mcp_bridge.py将“light”或“switch”命令路由到HA而不是Windows的逻辑。
🛠️硬件连接
______________________________________________________________________
🎮 用法
- 开始桥：
python mcp_bridge.py
- 与您的设备通话：
- “打开计算器。”-> *打开Windows计算器*.
- “打开演播室的灯。”-> *触发家庭助理*.
- 享受： 整个过程发生在你的 本地WiFi。拔下互联网路由器不会阻止系统工作。
🚀 未来路线图
- \[\]添加对Mac/Linux MCP的支持。
- \[\]集成Spotify API进行音乐控制。
- \[\]为ESP32创建一个定制的3D打印外壳。