1. 项目概述让两个StackChan机器人“开口说话”最近在捣鼓我的两个StackChan机器人突发奇想如果能让它们俩通过网络“聊上天”那该多有意思这个想法听起来有点天马行空但实现起来其实是一条非常清晰的技术路径。核心就是利用WebSocket这个双向通信协议在两个独立的机器人之间建立一条实时、全双工的“对话通道”。这不仅仅是让两个可爱的机器人动动嘴那么简单它背后涉及了嵌入式开发、网络通信、前后端协同以及机器人行为控制等多个领域的知识融合。简单来说这个项目就是构建一个微型的、分布式的机器人对话系统。每个StackChan作为一个独立的客户端连接到一个中心化的WebSocket服务器。服务器负责消息的路由和转发当一个机器人“说话”发送消息时服务器会立刻将这条消息推送给另一个机器人。接收方机器人解析消息内容并驱动其舵机和屏幕做出相应的“聆听”和“回应”动作。整个过程几乎是实时的延迟可以控制在毫秒级从而实现流畅的“对话”体验。这个项目适合谁呢首先当然是StackChan的爱好者你想让自己的机器人“活”起来拥有社交能力。其次是对物联网、实时通信或机器人交互感兴趣的开发者。通过这个项目你可以亲手实践如何将硬件舵机、麦克风、扬声器、屏幕与软件网络协议、消息队列、状态机紧密结合解决真实世界中的通信与同步问题。即使你之前没有接触过WebSocket或机器人编程跟着下面的步骤也能一步步搭建起这个有趣的小系统。2. 核心架构设计与技术选型要让两个StackChan实现对话我们需要一个稳定、高效的通信骨架。整个系统的核心是发布-订阅模式与WebSocket协议的结合。2.1 为什么选择WebSocket在实时通信领域我们有几种常见选择HTTP轮询、长轮询、Server-Sent Events和WebSocket。前两者效率低下且延迟高不适合高频交互。SSE是服务器向客户端的单向推送。而WebSocket是真正的双向、全双工通信协议。一旦握手建立连接客户端和服务器就可以在任何时刻主动向对方发送数据连接会一直保持直到一方主动关闭。这对于需要持续、低延迟交互的机器人对话场景是近乎完美的选择。它避免了HTTP协议无状态、每次请求都要重建连接的开销使得“你一言我一语”的对话流能够自然发生。2.2 整体系统架构拆解整个系统可以清晰地划分为三层客户端层两个StackChan机器人每个机器人是一个独立的智能终端。它需要具备以下能力网络连接通过Wi-Fi模块接入局域网或互联网。WebSocket客户端实现与中心服务器的连接、消息发送与接收。本地逻辑控制解析收到的消息并转化为具体的动作指令控制舵机转头、张嘴、屏幕显示表情、文字和可能的音频模块播放语音。输入捕获如何触发“说话”可以是预设的对话脚本、语音识别模块的输入或者简单的按钮触发。服务层WebSocket服务器这是系统的“大脑”和“交换机”。它的核心职责是连接管理维护所有在线StackChan客户端的WebSocket连接。消息路由当一个客户端发送消息时服务器需要知道该将这条消息转发给哪个或哪些客户端。这通常基于“房间”或“对话对”的概念。例如我们可以为每一对聊天的机器人创建一个唯一的房间ID。协议适配定义客户端与服务器之间通信的消息格式如JSON并可能进行简单的协议转换或消息广播。通信协议与消息格式这是客户端与服务端对话的“语言”。我们必须定义一套双方都能理解的规则。一个简单而有效的JSON格式示例如下{ type: chat, sender: stackchan_001, room: pair_01, payload: { text: 你好呀今天天气不错, emotion: happy, action: nod } }type: 消息类型如chat聊天、join加入房间、leave离开房间、ping心跳。sender: 发送者的唯一标识符。room: 房间标识用于消息路由。payload: 消息主体包含具体的对话内容、机器人情感状态和动作指令。注意消息格式的设计至关重要。它直接决定了系统的扩展性。初期可以简单但务必预留字段以便未来增加语音数据流、更复杂的表情控制等功能。2.3 技术栈选型考量StackChan端客户端主控通常使用ESP32或Raspberry Pi Pico W因为它们内置Wi-Fi且社区支持完善。开发框架MicroPython是绝佳选择。它语法简单内置了websocket或uasyncio模块非常适合在资源受限的嵌入式设备上实现网络功能。相较于C/CMicroPython能让你更专注于业务逻辑而非底层驱动。舵机与屏幕驱动利用现有的StackChan开源库如m5stack-avatar的变种或自定义驱动来控制SG90舵机和ST7789屏幕。服务器端语言Node.js (ws库)、Python (websockets, FastAPI)、Go (gorilla/websocket) 都是成熟的选择。考虑到生态和快速原型开发Python FastAPI WebSockets组合非常友好代码简洁易于部署。部署初期可在本地电脑运行后期可部署到云服务器如国内的腾讯云、阿里云轻量应用服务器以实现公网对话。工具链串口调试工具用于给StackChan烧录程序和查看日志。MQTT客户端可选用于测试和模拟消息流。Wireshark高级用于深度网络抓包分析排查复杂的通信问题。3. 服务器端实现详解服务器是整个系统的枢纽其稳定性和效率直接决定了对话体验。我们选择用Python的FastAPI和websockets库来构建因为它异步性能好代码直观。3.1 搭建基础的WebSocket服务器首先我们需要一个能够接受连接、管理连接并转发消息的服务器核心。# server.py import asyncio import json from typing import Dict, Set import websockets from fastapi import FastAPI, WebSocket, WebSocketDisconnect app FastAPI() # 用于管理连接和房间 class ConnectionManager: def __init__(self): # room_id - Set[WebSocket] self.active_connections: Dict[str, Set[WebSocket]] {} async def connect(self, websocket: WebSocket, room_id: str): await websocket.accept() if room_id not in self.active_connections: self.active_connections[room_id] set() self.active_connections[room_id].add(websocket) print(f客户端加入房间: {room_id}, 当前连接数: {len(self.active_connections[room_id])}) # 通知房间内其他成员有新成员加入可选 await self.broadcast_to_room(room_id, json.dumps({ type: system, message: fA new StackChan has joined the room. }), excludewebsocket) def disconnect(self, websocket: WebSocket, room_id: str): if room_id in self.active_connections: self.active_connections[room_id].discard(websocket) if not self.active_connections[room_id]: del self.active_connections[room_id] print(f客户端离开房间: {room_id}) async def broadcast_to_room(self, room_id: str, message: str, exclude: WebSocket None): if room_id in self.active_connections: disconnected set() for connection in self.active_connections[room_id]: if connection ! exclude: try: await connection.send_text(message) except websockets.exceptions.ConnectionClosed: disconnected.add(connection) # 清理已断开的连接 for conn in disconnected: self.disconnect(conn, room_id) manager ConnectionManager() app.websocket(/ws/{room_id}) async def websocket_endpoint(websocket: WebSocket, room_id: str): await manager.connect(websocket, room_id) try: while True: data await websocket.receive_text() # 解析收到的消息 message json.loads(data) # 这里可以添加消息验证和处理的逻辑 print(f收到来自房间 {room_id} 的消息: {message}) # 核心将消息广播给同房间的其他所有客户端 await manager.broadcast_to_room(room_id, data, excludewebsocket) except WebSocketDisconnect: manager.disconnect(websocket, room_id) print(f客户端断开连接房间: {room_id})这个服务器做了几件关键事定义了一个连接管理器用字典来维护“房间-连接集合”的映射。提供了/ws/{room_id}的WebSocket端点。两个StackChan客户端只要连接时使用相同的room_id就会被分配到同一个“房间”。当一个客户端发来消息时服务器会将该消息原样转发给同一房间内的所有其他客户端实现了基本的对话路由。3.2 心跳机制与连接健康管理网络环境不稳定连接可能意外断开。为了及时发现“僵尸连接”需要实现心跳机制。# 在 websocket_endpoint 函数的循环中增加心跳处理 import time async def websocket_endpoint(websocket: WebSocket, room_id: str): await manager.connect(websocket, room_id) last_ping time.time() try: while True: try: # 设置接收超时以便有机会发送ping data await asyncio.wait_for(websocket.receive_text(), timeout10.0) message json.loads(data) if message.get(type) ping: # 收到客户端ping回复pong await websocket.send_text(json.dumps({type: pong, timestamp: time.time()})) last_ping time.time() continue # ... 处理其他消息 ... await manager.broadcast_to_room(room_id, data, excludewebsocket) except asyncio.TimeoutError: # 超时检查心跳 if time.time() - last_ping 30: # 30秒无心跳判定失联 print(f客户端 {room_id} 心跳超时断开连接) break # 主动向客户端发送ping try: await websocket.send_text(json.dumps({type: ping, timestamp: time.time()})) except: break # 发送失败连接已断 except WebSocketDisconnect: # ... 断开处理 ...客户端也需要定期向服务器发送ping消息并在收到pong后重置计时器。这种双向心跳能有效保持连接活跃并在断开时快速触发重连逻辑。3.3 消息队列与流量控制当消息频率很高时直接广播可能导致服务器压力大或客户端处理不过来。引入一个简单的异步消息队列可以缓解这个问题。import asyncio from collections import deque class MessageQueue: def __init__(self): self.queue asyncio.Queue() async def put(self, message): await self.queue.put(message) async def get(self): return await self.queue.get() # 在ConnectionManager中为每个房间或连接维护一个发送队列 async def sender_task(websocket: WebSocket, queue: MessageQueue): while True: message await queue.get() try: await websocket.send_text(message) except: break # 发送失败结束任务 # 在广播时不直接send而是放入各个连接的队列 async def broadcast_to_room(self, room_id: str, message: str, exclude: WebSocket None): if room_id in self.active_connections: for connection in self.active_connections[room_id]: if connection ! exclude and hasattr(connection, send_queue): await connection.send_queue.put(message)这样发送动作变成了异步的即使某个客户端处理慢也不会阻塞服务器向其他客户端发送消息提升了系统的整体健壮性。4. StackChan客户端实现详解客户端是机器人的“身体”和“感官”。我们需要在资源有限的微控制器上实现网络通信、消息解析和硬件控制。4.1 MicroPython环境搭建与网络连接首先确保你的StackChan主控如ESP32已经刷入MicroPython固件。然后编写连接Wi-Fi和WebSocket的代码。# main.py on StackChan import network import usocket as socket import ujson as json import ure as re import time from machine import Pin, PWM, SPI import st7789 # 屏幕驱动库 import avatar # 自定义的Avatar库用于控制表情和嘴型 import _thread # 1. 连接Wi-Fi def connect_wifi(ssid, password): wlan network.WLAN(network.STA_IF) wlan.active(True) if not wlan.isconnected(): print(connecting to network...) wlan.connect(ssid, password) while not wlan.isconnected(): time.sleep(0.5) print(., end) print(network config:, wlan.ifconfig()) # 2. WebSocket客户端类 (简化版实际需使用urequests或自定义协议) class SimpleWebSocketClient: def __init__(self, server_addr, port, path): self.server_addr server_addr self.port port self.path path self.sock None def connect(self): ai socket.getaddrinfo(self.server_addr, self.port)[0] self.sock socket.socket() self.sock.connect(ai[-1]) # 发送WebSocket握手请求 (这里极其简化真实环境需完整实现RFC6455) self.sock.send(fGET {self.path} HTTP/1.1\r\nHost: {self.server_addr}:{self.port}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: x\r\n\r\n) # ... 解析握手响应 ... print(WebSocket Connected) def send(self, data): # 实现WebSocket数据帧封装 frame self._encode_frame(data) self.sock.send(frame) def recv(self): # 实现WebSocket数据帧解析 data self.sock.recv(1024) return self._decode_frame(data) def _encode_frame(self, data): # 简化的掩码处理和数据帧构造 (生产环境需完善) length len(data) if length 125: frame bytearray([0x81, 0x80 | length]) # FIN1, Opcode1 (Text), Mask1 # ... 处理更长的数据 ... mask bytearray([0x12, 0x34, 0x56, 0x78]) # 示例掩码 frame.extend(mask) masked_data bytearray([data[i] ^ mask[i % 4] for i in range(length)]) frame.extend(masked_data) return frame def _decode_frame(self, data): # 解析服务器返回的帧 (假设服务器不掩码) if len(data) 2: return None fin (data[0] 0x80) ! 0 opcode data[0] 0x0F masked (data[1] 0x80) ! 0 payload_len data[1] 0x7F # ... 解析变长长度和掩码 ... # 返回payload数据 return data[offset:offsetpayload_len].decode(utf-8) # 主循环 def main(): connect_wifi(Your_WiFi_SSID, Your_WiFi_Password) client SimpleWebSocketClient(192.168.1.100, 8000, /ws/stackchan_room_1) client.connect() # 初始化硬件 screen st7789.ST7789(...) face avatar.Avatar(screen) servo_jaw PWM(Pin(12), freq50) # 下巴舵机 last_ping_time time.time() while True: # 发送心跳 if time.time() - last_ping_time 10: client.send(json.dumps({type: ping})) last_ping_time time.time() # 尝试接收消息 try: message client.recv() if message: msg_obj json.loads(message) if msg_obj[type] chat: # 处理聊天消息 process_chat_message(msg_obj, face, servo_jaw) elif msg_obj[type] pong: pass # 心跳回应 except Exception as e: print(Recv error:, e) # 可以考虑重连逻辑 time.sleep(1) # 这里可以加入检测本地触发“说话”的代码例如按钮按下 # if button_pressed(): # text_to_say 你好我是StackChan # client.send(json.dumps({type: chat, sender: stackchan_001, text: text_to_say})) # # 同时让自己动起来 # local_speak_animation(text_to_say, face, servo_jaw) time.sleep(0.05) # 短暂延时避免CPU跑满实操心得在MicroPython中完整实现WebSocket协议有点复杂。强烈建议使用社区维护的库如micropython-uasyncio和micropython-websocket的客户端实现它们更稳定处理了协议细节。上述简化代码仅用于说明原理。4.2 消息解析与机器人动作驱动当收到一条chat类型的消息时客户端需要解析payload并驱动硬件做出反应。def process_chat_message(msg, face, servo_jaw): payload msg[payload] text payload.get(text, ) emotion payload.get(emotion, neutral) action payload.get(action, ) # 1. 更新屏幕表情 face.set_emotion(emotion) # 假设avatar库有set_emotion方法切换预设表情图片 # 2. 文字转嘴型动画关键 # 这是一个简化版根据文本长度和字符控制下巴舵机开合模拟说话口型 words_per_minute 150 # 假设语速 duration_per_char 60.0 / (words_per_minute * 5) # 粗略估算每个字符显示时间 for char in text: # 简单规则元音字母或某些辅音时张嘴 if char.lower() in aeiou: servo_jaw.duty_u16(4500) # 张嘴位置具体值需校准 else: servo_jaw.duty_u16(3000) # 闭嘴位置具体值需校准 face.draw_text(char, x, y) # 在屏幕某个位置显示字符可选 time.sleep(duration_per_char) servo_jaw.duty_u16(3000) # 说完一个字符后闭嘴 time.sleep(duration_per_char * 0.3) # 字间短暂停顿 # 3. 执行附加动作 if action nod: nod_head() # 控制头部舵机点头的函数 elif action shake: shake_head() # 摇头 # 4. 消息显示后恢复默认表情 time.sleep(0.5) face.set_emotion(neutral)这里最复杂也最有趣的部分是“文字转嘴型动画”。更高级的做法是预先定义一组口型如A、I、U、E、O等对应的舵机角度然后根据文本内容生成一段口型关键帧序列再通过插值让舵机平滑运动这样看起来会更自然而不是生硬的“电报式”开合。4.3 本地触发与音频集成进阶要让对话更自然可以增加本地触发机制。按钮触发最简单的形式按一下按钮说一句预设的话。语音识别触发集成一个像LD3320这样的离线语音识别模块或者连接一个在线语音识别API如百度语音识别。当机器人“听到”特定唤醒词或问句时触发本地逻辑生成回复文本并通过WebSocket发送出去。音频播放在“说话”时可以同步通过DFPlayer Mini等模块播放预先录制或TTS生成的语音。这就需要精确同步音频播放和舵机动画挑战更大但效果也最震撼。# 伪代码语音识别触发示例 if voice_module.detect_wake_word(小卡): question voice_module.record_and_recognize(timeout3) if question: # 简单的关键词回复可扩展为接入大语言模型API if 你好 in question: reply_text 你好主人 elif 天气 in question: reply_text 今天天气晴转多云。 else: reply_text 我没听明白呢。 # 发送回复 client.send(json.dumps({type: chat, sender: MY_ID, payload: {text: reply_text}})) # 本地播放语音 audio_player.play_tts(reply_text) # 本地执行说话动画 local_speak_animation(reply_text, face, servo_jaw)5. 系统联调与问题排查实录将服务器和两个客户端都跑起来才是挑战的开始。以下是我在调试过程中遇到的一些典型问题及解决方法。5.1 连接建立失败现象StackChan客户端无法连接到服务器一直超时。排查网络可达性首先确保StackChan和运行服务器的电脑在同一个局域网内并且IP地址正确。在服务器电脑上ping一下StackChan的IP反之亦然。防火墙服务器电脑的防火墙可能阻止了8000端口或你自定义的端口。需要在防火墙设置中添加入站规则允许该端口的TCP连接。服务器未启动检查服务器程序是否真的在运行并且监听在0.0.0.0所有接口而不是127.0.0.1仅本地。WebSocket握手失败用浏览器打开ws://your_server_ip:8000/ws/test测试工具检查握手是否成功。MicroPython客户端的握手请求头必须严格符合RFC6455标准一个字符错误都会导致失败。务必使用成熟的WebSocket客户端库。5.2 消息收发延迟或丢失现象一个机器人“说话”后另一个机器人反应很慢或者根本没反应。排查服务器广播逻辑检查服务器的broadcast_to_room函数确保它正确地将消息发送给了目标房间内除发送者外的所有连接。我的一个bug就是忘记exclude发送者自己导致消息回环机器人自言自语。客户端接收缓冲区MicroPython的socket.recv()默认是阻塞的。如果处理一条消息比如执行一个漫长的动画的时间过长期间服务器发来的下一条消息可能会被覆盖或丢失。解决方案是使用非阻塞socket或select轮询或者将网络接收和动画执行放在不同的线程/异步任务中。网络抖动在Wi-Fi信号弱的环境下UDP包WebSocket基于TCP但Wi-Fi本身不稳定可能丢失重传。可以在关键位置如发送和接收后打印时间戳计算端到端延迟。如果延迟过高考虑优化网络环境或增加消息确认重传机制对于非实时性要求极高的对话通常不需要。5.3 机器人动作卡顿或不同步现象嘴型动画一顿一顿的或者和语音播放不同步。排查主循环阻塞检查客户端的main循环。如果recv()、动画time.sleep()或复杂的计算阻塞了主循环太久会导致整个系统响应变慢。必须将长时间操作异步化或使用多线程。例如用一个线程专负责网络IO另一个线程负责控制舵机和屏幕。舵机信号冲突确保控制舵机的PWM信号引脚没有其他功能冲突。同时SG90这类舵机不能接收太快的信号更新通常20ms50Hz的周期是合适的。频繁地以更高频率更新duty值可能反而导致舵机抖动。动画与音频同步这是一个经典难题。一个实用的方法是以音频播放为时间基准。预先知道一段语音的时长T然后将嘴型动画的关键帧时间轴映射到[0, T]区间内。在开始播放音频的同一时刻启动动画计时器。这样即使两者有微小误差人眼也不易察觉。5.4 内存不足与程序崩溃现象StackChan运行一段时间后重启或停止响应。排查内存泄漏MicroPython有垃圾回收但如果你在循环中不断创建大的对象如长字符串、列表、字典而不及时释放引用内存会耗尽。使用gc.mem_free()定期打印剩余内存监控内存变化。尽量复用对象。递归或深循环避免深度递归或无限循环中没有time.sleep()或await asyncio.sleep()这会导致看门狗定时器触发复位。异常捕获用try...except包裹可能出错的代码块特别是网络操作和硬件操作记录错误日志并尝试恢复如网络重连而不是让整个程序崩溃。5.5 常见问题速查表问题现象可能原因排查步骤与解决方案无法连接服务器1. IP/端口错误2. 防火墙阻止3. 服务器未运行4. 握手协议错误1. 检查IP和端口2. 关闭防火墙或添加规则3. 确认服务器进程4. 使用标准WebSocket库连接频繁断开1. 网络不稳定2. 心跳机制未启用或超时时间太短3. 服务器/客户端资源耗尽1. 改善Wi-Fi信号2. 实现并调整心跳间隔如30秒ping60秒超时3. 检查服务器日志优化代码收不到对方消息1. 房间ID不一致2. 服务器广播逻辑错误3. 客户端接收代码阻塞1. 确认双方连接时使用相同的room_id2. 调试服务器确认消息正确转发3. 将客户端接收改为非阻塞模式舵机不动作或乱动1. 电源不足5V/2A以上2. 信号线接触不良3. PWM频率或占空比范围不对1. 使用独立电源为舵机供电2. 检查杜邦线连接3. SG90舵机频率通常为50Hz脉宽0.5ms-2.4ms屏幕显示异常1. SPI引脚配置错误2. 电源或背光问题3. 驱动库不兼容1. 对照原理图检查SCK, MOSI, DC, RESET, CS引脚2. 确保3.3V供电稳定背光引脚拉高3. 尝试使用官方示例代码测试屏幕6. 优化与扩展思路当基础对话功能跑通后你可以考虑以下方向让项目变得更酷、更智能。6.1 引入状态机管理机器人行为目前机器人的行为是线性的收到消息 - 执行动画。引入一个状态机可以让机器人的行为更丰富、更自然。例如定义状态空闲、聆听中、思考中、说话中、高兴、困惑等。不同状态下机器人的待机动画、收到消息后的反应都会不同。这能极大提升交互的生动性。6.2 集成大语言模型LLM让对话内容不再预设而是真正具有智能。你可以在服务器端集成一个LLM的API如国内的一些开放平台或本地部署的小模型。当服务器收到一个机器人的消息后不直接转发而是先将消息内容发送给LLM获取一个有趣的回复再将这个回复转发给另一个机器人。这样两个机器人就能进行天马行空的自由对话了。注意这需要处理网络延迟和API调用成本。6.3 支持多个机器人群体聊天当前架构很容易扩展成多人多机器人聊天室。只需让多个机器人加入同一个room_id。服务器端的广播逻辑不需要改变。你还可以设计一些系统消息比如“XXX进入了房间”。为了区分说话者可以在屏幕上显示发言者的ID。6.4 增加传感器与上下文感知给StackChan加上传感器如超声波测距、摄像头、温湿度传感器。机器人可以将传感器数据“我看到你离我很近”、“这里好热”作为对话内容的一部分发送出去使得对话内容与物理世界关联创造出更沉浸的体验。这个项目从一个小小的想法开始串联起了嵌入式硬件、网络通信、软件开发和机器人学等多个知识点。当你看到两个自己亲手制作的机器人通过网络流畅地“交谈”起来那种成就感是无与伦比的。过程中遇到的每一个坑解决的每一个问题都是宝贵的经验。最重要的是它为你打开了一扇门一个让硬件设备具备社交和协作能力的、充满可能性的世界。