基于云端OpenClaw的情绪互动机器人系统-Milk-V Duo S端可靠通信、执行反馈与情绪速度自适应控制设计（9）
·
前几周的工作中，我已经完成了 Milk-V Duo S 端动作执行状态机、动作安全校验、风险分级机制以及复杂动作组设计。现在 Duo S 端已经不再只是简单地执行云端传来的动作编号，而是能够对动作序列进行校验、规范化、调度执行和异常恢复。
但是在继续联调过程中，我发现端侧还有两个问题需要进一步优化。
第一个问题是可靠通信与执行反馈。Duo S 端通过 HTTP 轮询从云端 Webhook 服务获取动作指令，但是实际比赛或演示环境中，WiFi 可能不稳定，云端接口可能短暂无响应，ACK 也可能回传失败。如果端侧只是简单地 requests.get() 和 requests.post()，一旦网络抖动，机器人就会出现状态不明确的问题。
第二个问题是动作表现缺少情绪强度差异。比如同样是挥手动作，如果用户只是普通问候，机器人可以正常挥手；如果用户表达非常开心，挥手动作应该更快、更有活力；如果用户情绪低落或害怕，动作应该更慢、更收缩。也就是说，云端 OpenClaw 不仅应该决定“机器人做什么动作”，还应该影响“机器人以什么速度做动作”。
因此，本周我在 Milk-V Duo S 端继续设计了两个优化模块：
- 可靠通信与执行反馈机制
- 基于情绪强度的动作速度自适应控制机制
本次优化的目标是让 Duo S 端从“能执行动作”进一步升级为“能可靠通信、能回传执行结果、能根据情绪强度调整动作表现”。
一、为什么要设计可靠通信机制
之前的 Duo S 端轮询代码大致如下：
def poll_action():
try:
response = requests.get(url, timeout=5)
return response.json()
except Exception as e:
print("轮询失败:", e)
return None这段代码可以完成基本通信，但是存在几个问题：
- 网络失败后没有失败次数统计；
- 无法区分短暂失败和长时间离线；
- ACK 回传失败后没有重试；
- 云端无法知道机器人是否真正执行成功；
- Duo S 端无法根据通信状态进入安全保护；
- 终端日志中只能看到“失败”，看不到完整执行报告。
在机器人项目中，通信失败不是简单的软件错误。因为机器人端控制的是实体舵机，如果云端和端侧状态不同步，就可能出现“云端认为执行完成，但机器人实际失败”的情况。因此我决定把通信模块从简单的 HTTP 请求升级为可靠通信客户端。
二、可靠通信模块整体设计
我将 Duo S 端通信机制设计成以下流程：
轮询云端任务
↓
网络成功：解析动作任务
↓
交给动作调度器执行
↓
生成执行报告
↓
回传 ACK
↓
ACK 成功：进入下一轮轮询
↓
ACK 失败：重试并记录日志
如果网络连续失败：
失败 1-2 次：正常重试
失败 3 次：降低轮询频率
失败 5 次：执行站立保护
失败 10 次：进入离线保护模式
网络恢复：退出离线模式技术理解：
Duo S 端的 HTTP 轮询不是简单的“请求一次成功或失败”，而是一个长期运行的通信过程。它需要考虑网络抖动、云端无任务、重复任务、ACK 丢失、异常 JSON 等情况。可靠通信机制的本质，是让机器人端在网络不稳定时仍然保持可控状态。
三、使用 AI 辅助生成可靠通信客户端
提示词 1：生成 CloudClient 通信模块
请帮我生成一个 Python 模块 cloud_client.py，用于 Milk-V Duo S 端和云端 OpenClaw Webhook 服务通信。
项目背景：
Milk-V Duo S 端需要通过 HTTP 轮询云端服务获取动作任务，接口为：
GET /poll/{client_id}
POST /ack/{client_id}
要求：
1. 使用 requests 标准 HTTP 库。
2. 定义 CloudClient 类。
3. 初始化参数包括 server、client_id、token=None、timeout=5。
4. 提供 poll_action() 方法：
- 向 /poll/{client_id} 发送 GET 请求
- 支持 Bearer Token
- 处理网络异常、非 200 状态码、JSON 解析异常
- 成功时重置网络失败计数
- 失败时增加 fail_count
5. 提供 ack_action(report) 方法：
- 向 /ack/{client_id} 发送 POST 请求
- 发送执行报告 JSON
- ACK 失败时最多重试 3 次
6. 提供 is_offline() 方法：
- 当连续失败次数达到 offline_threshold 时返回 True
7. 提供 get_poll_interval() 方法：
- 网络正常时返回基础轮询间隔
- 网络失败次数较多时增加轮询间隔
8. 代码要适合嵌入式 Linux Python 环境运行，不依赖复杂第三方库。AI 生成代码：cloud_client.py
import time
import requests
class CloudClient:
"""
Milk-V Duo S 端云端通信客户端。
负责从 OpenClaw Webhook 服务轮询动作任务，并回传执行结果 ACK。
"""
def __init__(
self,
server,
client_id,
token=None,
timeout=5,
base_poll_interval=2,
offline_threshold=10
):
self.server = server.rstrip("/")
self.client_id = client_id
self.token = token
self.timeout = timeout
self.base_poll_interval = base_poll_interval
self.offline_threshold = offline_threshold
self.fail_count = 0
self.last_error = None
self.offline_mode = False
def _headers(self):
"""
构造 HTTP 请求头。
如果配置了 token，则使用 Bearer Token 认证。
"""
headers = {
"Content-Type": "application/json"
}
if self.token:
headers["Authorization"] = f"Bearer {self.token}"
return headers
def poll_action(self):
"""
从云端轮询动作任务。
成功返回云端 JSON。
失败返回 None，并增加 fail_count。
"""
url = f"{self.server}/poll/{self.client_id}"
try:
response = requests.get(
url,
headers=self._headers(),
timeout=self.timeout
)
if response.status_code != 200:
self.fail_count += 1
self.last_error = f"HTTP 状态码异常: {response.status_code}"
print(f"[CloudClient] 轮询失败: {self.last_error}")
return None
try:
data = response.json()
except Exception as e:
self.fail_count += 1
self.last_error = f"JSON 解析失败: {e}"
print(f"[CloudClient] 轮询失败: {self.last_error}")
return None
# 一旦成功收到云端响应，就认为网络恢复
if self.fail_count > 0:
print("[CloudClient] 网络通信恢复，fail_count 清零")
self.fail_count = 0
self.last_error = None
self.offline_mode = False
return data
except Exception as e:
self.fail_count += 1
self.last_error = str(e)
print(f"[CloudClient] 轮询异常: {e}")
return None
def ack_action(self, report, max_retry=3):
"""
向云端回传动作执行结果。
如果 ACK 失败，进行有限次数重试。
"""
url = f"{self.server}/ack/{self.client_id}"
for attempt in range(1, max_retry + 1):
try:
response = requests.post(
url,
headers=self._headers(),
json=report,
timeout=self.timeout
)
if response.status_code == 200:
print(f"[CloudClient] ACK 回传成功，第 {attempt} 次尝试")
return True
print(
f"[CloudClient] ACK 回传失败，第 {attempt} 次尝试，"
f"状态码: {response.status_code}"
)
except Exception as e:
print(f"[CloudClient] ACK 回传异常，第 {attempt} 次尝试: {e}")
time.sleep(1)
print("[CloudClient] ACK 多次重试后仍然失败")
return False
def is_offline(self):
"""
判断是否进入离线模式。
"""
if self.fail_count >= self.offline_threshold:
self.offline_mode = True
return self.offline_mode
def get_poll_interval(self):
"""
根据网络失败次数动态调整轮询间隔。
正常情况下 2 秒轮询一次；
连续失败后逐渐降低请求频率，避免无意义地频繁请求。
"""
if self.fail_count <= 2:
return self.base_poll_interval
if self.fail_count <= 5:
return self.base_poll_interval * 2
if self.fail_count <= 10:
return self.base_poll_interval * 4
return self.base_poll_interval * 6
def get_network_status(self):
"""
返回当前网络状态，用于执行报告和日志记录。
"""
return {
"fail_count": self.fail_count,
"offline_mode": self.offline_mode,
"last_error": self.last_error
}技术理解
这个模块相比原来的简单 requests.get()，主要增加了以下能力：
| 功能 | 作用 |
|---|---|
| fail_count | 统计连续网络失败次数 |
| offline_mode | 判断机器人是否进入离线保护 |
| get_poll_interval | 根据失败次数动态调整轮询频率 |
| ack_action 重试 | 避免 ACK 因短暂网络波动丢失 |
| get_network_status | 让执行报告中包含网络状态 |
| token 支持 | 保留后续接口认证扩展能力 |
我认为这里最重要的是 fail_count。它让 Duo S 端能够区分“一次网络抖动”和“长时间离线”。如果只是偶尔失败，程序继续重试即可；如果连续失败很多次，就应该让机器人回到安全姿态，而不是一直保持上一个动作状态。
四、离线保护机制设计
可靠通信不仅要处理网络请求，还要考虑断网时机器人应该做什么。
我的设计是：
连续失败 < 3 次：
正常重试
连续失败 >= 3 次：
降低轮询频率
连续失败 >= 5 次：
执行站立回正
连续失败 >= 10 次：
进入离线保护模式技术理解：
机器人不能因为断网就处于未知状态。如果网络断开前机器人刚执行完下蹲保护、前进招手或鞠躬动作，而端侧没有及时回正，那么机器人可能长时间保持不稳定姿态。因此 Duo S 端需要在网络长时间异常时主动执行安全站立动作。
五、使用 AI 辅助生成离线保护模块
提示词 2：生成通信异常处理和离线保护代码
请帮我生成一个 Python 模块 network_safety.py，用于 Milk-V Duo S 端机器人网络异常时的安全保护。
项目背景：
机器人通过 CloudClient 轮询云端。如果连续网络失败，机器人需要逐步进入保护策略：
1. 失败次数小于 3：只打印日志
2. 失败次数达到 3：提示降低轮询频率
3. 失败次数达到 5：执行站立动作
4. 失败次数达到 10：进入离线保护模式
5. 避免每一轮都重复执行站立动作
要求：
1. 定义 NetworkSafetyManager 类。
2. 初始化时传入 robot_controller。
3. 提供 handle_network_status(network_status) 方法。
4. 根据 fail_count 判断保护策略。
5. 当 fail_count >= 5 时只执行一次 stand()，避免重复发送站立指令。
6. 当 fail_count 归零时，退出离线保护并重置标志位。
7. 代码要有清晰日志。AI 生成代码：network_safety.py
import time
class NetworkSafetyManager:
"""
网络异常安全保护模块。
当 Duo S 端连续无法连接云端时，让机器人进入安全姿态。
"""
def __init__(self, robot):
self.robot = robot
self.has_stand_for_network_error = False
self.in_offline_protection = False
def handle_network_status(self, network_status):
"""
根据网络状态执行保护策略。
network_status 示例：
{
"fail_count": 5,
"offline_mode": False,
"last_error": "timeout"
}
"""
fail_count = network_status.get("fail_count", 0)
offline_mode = network_status.get("offline_mode", False)
last_error = network_status.get("last_error")
if fail_count == 0:
if self.in_offline_protection or self.has_stand_for_network_error:
print("[NetworkSafety] 网络恢复，退出离线保护状态")
self.has_stand_for_network_error = False
self.in_offline_protection = False
return
if fail_count < 3:
print(f"[NetworkSafety] 网络短暂异常，fail_count={fail_count}, error={last_error}")
return
if fail_count >= 3 and fail_count < 5:
print("[NetworkSafety] 网络连续异常，开始降低轮询频率")
return
if fail_count >= 5 and not self.has_stand_for_network_error:
print("[NetworkSafety] 网络异常达到阈值，执行站立保护")
try:
self.robot.stand()
time.sleep(5)
self.has_stand_for_network_error = True
except Exception as e:
print(f"[NetworkSafety] 站立保护执行失败: {e}")
if offline_mode:
if not self.in_offline_protection:
print("[NetworkSafety] 进入离线保护模式")
self.in_offline_protection = True技术理解
这里有一个细节：当 fail_count >= 5 时，不应该每一轮循环都执行站立动作。否则 Duo S 端会不断向串口发送站立命令，影响舵机控制板正常状态。因此我加入了：
self.has_stand_for_network_error = False这个标志位保证网络异常期间只执行一次站立保护。等网络恢复后，再重置这个标志位。
六、执行反馈报告机制设计
原来的 ACK 回传只包含：
{
"command_id": command_id
}这样云端只能知道“Duo S 收到了这个任务并尝试确认”，但不知道机器人到底执行了什么、是否成功、耗时多久、失败原因是什么。
因此我把 ACK 升级为执行报告，内容包括：
| 字段 | 含义 |
| command_id | 云端任务编号 |
| client_id | Duo S 客户端 ID |
| status | done 或 failed |
| original_sequence | 云端原始动作序列 |
| normalized_sequence | 端侧规范化后的动作序列 |
| executed_actions | 实际执行过的动作 |
| emotion | 云端识别情绪 |
| intensity | 情绪强度 |
| speed | 端侧计算出的动作速度 |
| start_time | 执行开始时间 |
| end_time | 执行结束时间 |
| duration | 实际耗时 |
| error | 错误信息 |
| network_status | 当前网络状态 |
技术理解：
执行反馈报告让云端和前端能够真正知道机器人端发生了什么。比如前端显示“执行完成”，就不应该只是云端成功发送了动作，而应该是 Duo S 端真的执行完成并回传了结果。这是端云闭环的重要一步。
七、使用 AI 辅助生成执行报告模块
提示词 3：生成执行报告模块
请帮我生成一个 Python 模块 execution_report.py，用于 Milk-V Duo S 端生成机器人动作执行报告。
项目背景：
机器人从云端 OpenClaw 获取 command_id、action_sequence、emotion、intensity。Duo S 端执行动作后，需要向云端 ACK 接口回传详细执行报告。
要求：
1. 不使用 dataclass，保证兼容嵌入式 Python 环境。
2. 定义 ExecutionReport 类。
3. 初始化参数包括 command_id、client_id、original_sequence、emotion=None、intensity=None。
4. 记录 start_time、end_time、duration。
5. 记录 normalized_sequence、executed_actions、speed_map、status、error。
6. 提供 set_normalized_sequence(sequence)。
7. 提供 add_executed_action(action_id, speed)。
8. 提供 mark_done()。
9. 提供 mark_failed(error)。
10. 提供 to_dict(network_status=None) 方法，返回可 JSON 序列化的字典。
11. 时间使用 time.strftime 和 time.time。AI 生成代码：execution_report.py
import time
class ExecutionReport:
"""
Duo S 端动作执行报告。
用于向云端 ACK 接口回传机器人真实执行情况。
"""
def __init__(
self,
command_id,
client_id,
original_sequence,
emotion=None,
intensity=None
):
self.command_id = command_id
self.client_id = client_id
self.original_sequence = original_sequence
self.normalized_sequence = []
self.executed_actions = []
self.speed_map = {}
self.emotion = emotion
self.intensity = intensity
self.status = "running"
self.error = None
self.start_timestamp = time.time()
self.end_timestamp = None
self.start_time = time.strftime("%Y-%m-%d %H:%M:%S")
self.end_time = None
self.duration = None
def set_normalized_sequence(self, sequence):
"""
记录端侧安全校验后生成的规范化动作序列。
"""
self.normalized_sequence = list(sequence)
def add_executed_action(self, action_id, speed):
"""
记录已经执行过的动作和对应速度。
"""
self.executed_actions.append(action_id)
self.speed_map[str(action_id)] = speed
def mark_done(self):
"""
标记执行成功。
"""
self.status = "done"
self.end_timestamp = time.time()
self.end_time = time.strftime("%Y-%m-%d %H:%M:%S")
self.duration = round(self.end_timestamp - self.start_timestamp, 3)
def mark_failed(self, error):
"""
标记执行失败。
"""
self.status = "failed"
self.error = str(error)
self.end_timestamp = time.time()
self.end_time = time.strftime("%Y-%m-%d %H:%M:%S")
self.duration = round(self.end_timestamp - self.start_timestamp, 3)
def to_dict(self, network_status=None):
"""
转换为可以直接 JSON 序列化的字典。
"""
return {
"command_id": self.command_id,
"client_id": self.client_id,
"status": self.status,
"emotion": self.emotion,
"intensity": self.intensity,
"original_sequence": self.original_sequence,
"normalized_sequence": self.normalized_sequence,
"executed_actions": self.executed_actions,
"speed_map": self.speed_map,
"start_time": self.start_time,
"end_time": self.end_time,
"duration": self.duration,
"error": self.error,
"network_status": network_status or {}
}技术理解
这个执行报告模块解决了一个重要问题：以前 ACK 只是“确认收到”，现在 ACK 是“执行结果报告”。
例如 Duo S 执行 [11, 12] 时，端侧可能会自动规范化成：
[0, 11, 12, 0]如果只回传原始序列，云端并不知道 Duo S 实际执行了站立回正动作。而执行报告可以同时记录：
"original_sequence": [11, 12],
"normalized_sequence": [0, 11, 12, 0],
"executed_actions": [0, 11, 12, 0]这样云端、前端和端侧日志就能保持一致。
八、为什么要设计情绪强度速度自适应
在之前的版本中，机器人执行动作时速度基本固定：
robot.run_action(action_id, speed=100)但是情绪互动机器人不应该所有动作都用同样速度执行。因为动作速度本身也是情绪表达的一部分。
例如：
| 情绪 | 动作速度设计 |
| 开心 | 动作更快、更有活力 |
| 兴奋 | 动作速度最高，但要限制上限 |
| 感谢 | 动作稍慢，更柔和 |
| 难过 | 动作更慢，表现低落 |
| 害怕 | 动作更慢，避免突兀 |
| 平静 | 默认速度 |
| 惊讶 | 起始动作可稍快 |
技术理解：
云端 OpenClaw 生成的动作序列解决的是“动作选择”问题，而 Duo S 端速度控制解决的是“动作表现”问题。两个输入同样生成挥手动作，但如果情绪强度不同，机器人身体语言也应该不同。
九、情绪速度控制设计
我设计了一个 speed_controller.py 模块，它根据两个因素计算速度：
- 情绪类型
emotion - 情绪强度
intensity
其中 intensity 取值范围为 0 到 1。
基本规则：
happy / excited:
强度越高，速度越快
sad / fear:
强度越高，速度越慢
thanks / calm:
速度略低，动作更柔和
angry:
不直接高速执行，避免动作过猛
默认：
使用 100 速度此外，还需要考虑动作风险等级。比如高风险动作即使情绪很强，也不能无限加速。因此我加入了速度上下限限制：
普通动作：60 - 130
中风险动作：60 - 115
高风险动作：60 - 100这样可以避免复杂动作因速度过快导致不稳定。
十、使用 AI 辅助生成情绪速度控制模块
提示词 4：生成速度自适应控制代码
请帮我生成一个 Python 模块 speed_controller.py，用于 Milk-V Duo S 端机器人根据情绪类型和情绪强度自适应调整动作速度。
项目背景：
云端 OpenClaw Agent 返回 emotion 和 intensity，Milk-V Duo S 端执行动作组时通过串口命令中的 SM{speed} 控制动作速度，例如：
PL0 SQ11 SM100 ONCE
要求：
1. 定义 SpeedController 类。
2. 初始化默认速度 default_speed=100。
3. intensity 范围为 0 到 1，如果为空或非法则使用 0.5。
4. 根据 emotion 计算速度：
- happy、excited：强度越高速度越快
- sad、fear：强度越高速度越慢
- thanks、calm：速度略慢且柔和
- surprise：中高速度
- angry：不超过 105，避免动作过猛
- 其他情绪使用默认速度
5. 根据动作风险等级限制速度：
- low 风险最高 130
- medium 风险最高 115
- high 风险最高 100
6. 最低速度不能低于 60。
7. 提供 calc_speed(emotion, intensity, risk_level=None) 方法。
8. 提供 calc_wait_time(base_duration, speed) 方法，根据速度调整等待时间，并加入安全缓冲。
9. 不依赖第三方库。AI 生成代码：speed_controller.py
class SpeedController:
"""
基于情绪强度的动作速度自适应控制器。
将云端 emotion/intensity 转换为舵机控制指令中的 SM 速度参数。
"""
def __init__(self, default_speed=100):
self.default_speed = default_speed
self.min_speed = 60
self.low_risk_max_speed = 130
self.medium_risk_max_speed = 115
self.high_risk_max_speed = 100
def _normalize_intensity(self, intensity):
"""
将情绪强度规范化到 0.0 - 1.0。
如果 intensity 缺失或格式错误，则使用 0.5。
"""
try:
value = float(intensity)
except Exception:
return 0.5
if value < 0:
return 0.0
if value > 1:
return 1.0
return value
def _risk_max_speed(self, risk_level):
"""
根据动作风险等级限制最高速度。
risk_level 可以是字符串，也可以是 Enum.value。
"""
if risk_level is None:
return self.medium_risk_max_speed
risk_text = str(risk_level).upper()
if "HIGH" in risk_text:
return self.high_risk_max_speed
if "MEDIUM" in risk_text:
return self.medium_risk_max_speed
if "LOW" in risk_text:
return self.low_risk_max_speed
return self.medium_risk_max_speed
def calc_speed(self, emotion, intensity, risk_level=None):
"""
根据情绪类型、情绪强度和动作风险等级计算最终速度。
"""
intensity = self._normalize_intensity(intensity)
if emotion is None:
emotion = "neutral"
emotion = str(emotion).lower()
speed = self.default_speed
if emotion in ["happy", "joy", "excited", "开心", "高兴", "兴奋"]:
# 开心和兴奋：强度越高，速度越快
speed = 95 + int(35 * intensity)
elif emotion in ["sad", "悲伤", "难过", "失落"]:
# 难过：强度越高，动作越慢
speed = 100 - int(30 * intensity)
elif emotion in ["fear", "scared", "害怕", "恐惧", "紧张"]:
# 害怕：动作收缩、变慢
speed = 95 - int(25 * intensity)
elif emotion in ["thanks", "thankful", "gratitude", "感谢", "谢谢"]:
# 感谢：速度略慢，表现柔和
speed = 85
elif emotion in ["calm", "neutral", "平静", "中性"]:
speed = self.default_speed
elif emotion in ["surprise", "惊讶"]:
speed = 100 + int(15 * intensity)
elif emotion in ["angry", "生气", "愤怒"]:
# 生气不直接高速执行，避免动作过猛
speed = 95 + int(10 * intensity)
else:
speed = self.default_speed
# 根据动作风险等级限制速度上限
max_speed = self._risk_max_speed(risk_level)
if speed > max_speed:
speed = max_speed
if speed < self.min_speed:
speed = self.min_speed
return int(speed)
def calc_wait_time(self, base_duration, speed):
"""
根据动作速度调整等待时间。
base_duration 是动作在 SM100 附近测试得到的基础执行时间。
speed 越大，理论执行时间越短；speed 越小，等待时间越长。
为了安全，加入 0.5 秒缓冲，避免动作尚未结束就执行下一个动作。
"""
try:
base_duration = float(base_duration)
speed = float(speed)
except Exception:
return base_duration
if speed <= 0:
speed = self.default_speed
wait_time = base_duration * (self.default_speed / speed)
wait_time += 0.5
if wait_time < 1:
wait_time = 1
return round(wait_time, 2)技术理解
这个模块的重点不是简单地把 intensity 乘一个系数，而是同时考虑情绪和动作风险。
例如：
emotion = "excited"
intensity = 0.9
action_id = 11如果只看情绪，速度可能会算到 126。但是 11-前进招手 是高风险复杂动作，因此速度上限会被限制为 100。这样既能体现情绪强度，又不会让高风险动作过快执行。
这就是端侧控制和云端语义的结合：云端决定情绪，端侧结合机械风险进行二次控制。
十一、动作元数据中加入风险等级
速度控制需要知道动作风险等级，因此要和前一篇博客中的 action_meta.py 配合使用。
提示词 5：补充动作风险信息读取函数
请帮我补充 action_meta.py 中的辅助函数，用于给 speed_controller.py 提供动作风险等级。
要求：
1. 已经存在 ACTION_META 字典，每个动作包含 risk 字段。
2. 添加 get_action_risk(action_id) 函数。
3. 如果动作不存在，默认返回 MEDIUM。
4. 添加 is_complex_action(action_id) 函数。
5. 不依赖第三方库。AI 生成代码：补充到 action_meta.py
def get_action_risk(action_id):
"""
获取动作风险等级。
如果动作不存在，默认按中风险处理。
"""
meta = get_action_meta(action_id)
if meta is None:
return RiskLevel.MEDIUM
return meta.get("risk", RiskLevel.MEDIUM)
def is_complex_action(action_id):
"""
判断动作是否为复杂动作组。
"""
meta = get_action_meta(action_id)
if meta is None:
return False
return bool(meta.get("complex", False))技术理解
这里默认未知动作是中风险，而不是低风险。这样设计更加保守。机器人安全控制中，未知情况不应该被当成安全情况处理，而应该采用更谨慎的策略。
十二、将速度控制集成到动作调度器
原来的动作调度器执行动作时是：
robot.run_action(action_id)
time.sleep(duration)现在需要改成：
speed = speed_controller.calc_speed(emotion, intensity, risk)
robot.run_action(action_id, speed=speed)
wait_time = speed_controller.calc_wait_time(duration, speed)
time.sleep(wait_time)也就是说，同样的动作编号，在不同情绪强度下会生成不同速度。
例如：
[1] + happy + 0.9可能执行：
PL0 SQ1 SM126 ONCE而：
[1] + sad + 0.8可能执行：
PL0 SQ1 SM76 ONCE动作编号相同，但速度不同，情绪表达也不同。
十三、使用 AI 辅助修改动作调度器
提示词 6：把速度控制接入 ActionScheduler
请帮我修改 action_scheduler.py，让 Milk-V Duo S 端动作调度器支持情绪速度自适应控制。
已有模块：
1. RobotController，提供 run_action(action_id, speed=None)
2. SpeedController，提供 calc_speed(emotion, intensity, risk_level) 和 calc_wait_time(base_duration, speed)
3. action_meta.py，提供 get_action_duration、get_action_name、get_action_risk
4. ExecutionReport，提供 add_executed_action(action_id, speed)
要求：
1. ActionScheduler 初始化时创建 SpeedController。
2. run_sequence 增加参数 emotion=None、intensity=None、report=None。
3. 每个动作执行前读取动作风险等级。
4. 根据 emotion、intensity、risk_level 计算 speed。
5. 执行动作时传入 speed。
6. 根据 speed 计算 wait_time，而不是固定 sleep duration。
7. 如果传入 report，则记录 executed_action 和 speed。
8. 保留异常恢复 recover()。AI 生成代码：新版 action_scheduler.py 核心部分
import time
from robot_state import RobotState
from action_validator import validate_sequence
from action_meta import get_action_duration
from action_meta import get_action_name
from action_meta import get_action_risk
from action_meta import is_high_risk
from speed_controller import SpeedController
class ActionScheduler:
"""
Milk-V Duo S 端动作调度状态机。
在原有安全校验基础上，加入情绪强度速度自适应控制。
"""
def __init__(self, robot):
self.robot = robot
self.state = RobotState.IDLE
self.running = False
self.speed_controller = SpeedController(default_speed=100)
def set_state(self, new_state):
old_state = self.state
self.state = new_state
print(f"[Scheduler] 状态切换: {old_state.value} -> {new_state.value}")
def run_sequence(self, sequence, emotion=None, intensity=None, report=None):
"""
执行动作序列，并根据情绪强度自适应调整动作速度。
"""
if self.running:
print("[Scheduler] 当前已有动作正在执行，拒绝新序列")
return {
"ok": False,
"reason": "机器人正在执行上一条动作序列"
}
self.running = True
try:
self.set_state(RobotState.VALIDATING)
result = validate_sequence(sequence)
if not result["ok"]:
print(f"[Scheduler] 动作序列校验失败: {result['reason']}")
self.set_state(RobotState.ERROR)
self.recover()
return {
"ok": False,
"reason": result["reason"]
}
safe_sequence = result["sequence"]
if report is not None:
report.set_normalized_sequence(safe_sequence)
print(f"[Scheduler] 校验通过，规范化动作序列: {safe_sequence}")
self.set_state(RobotState.RUNNING)
for action_id in safe_sequence:
action_name = get_action_name(action_id)
base_duration = get_action_duration(action_id)
risk_level = get_action_risk(action_id)
speed = self.speed_controller.calc_speed(
emotion=emotion,
intensity=intensity,
risk_level=risk_level
)
wait_time = self.speed_controller.calc_wait_time(
base_duration=base_duration,
speed=speed
)
if is_high_risk(action_id):
print(
f"[Scheduler] 高风险动作: {action_id} - {action_name}, "
f"risk={risk_level.value}, speed={speed}"
)
print(
f"[Scheduler] 执行动作: {action_id} - {action_name}, "
f"emotion={emotion}, intensity={intensity}, "
f"speed={speed}, wait={wait_time}s"
)
ok = self.robot.run_action(action_id, speed=speed)
if not ok:
raise RuntimeError(f"动作 {action_id} 串口发送失败")
if report is not None:
report.add_executed_action(action_id, speed)
time.sleep(wait_time)
print(f"[Scheduler] 动作完成: {action_id} - {action_name}")
self.set_state(RobotState.IDLE)
return {
"ok": True,
"reason": "动作序列执行完成"
}
except Exception as e:
print(f"[Scheduler] 执行异常: {e}")
self.set_state(RobotState.ERROR)
self.recover()
return {
"ok": False,
"reason": str(e)
}
finally:
self.running = False
def recover(self):
"""
异常恢复流程。
"""
self.set_state(RobotState.RECOVERING)
try:
self.robot.stop()
time.sleep(1)
self.robot.stand()
time.sleep(5)
print("[Scheduler] 安全回正完成")
except Exception as e:
print(f"[Scheduler] 安全恢复失败: {e}")
finally:
self.set_state(RobotState.IDLE)技术理解
这里的关键变化是：动作调度器不再只关心“执行哪个动作”，还关心“用什么速度执行”。
原来的执行逻辑是动作编号驱动：
action_id → run_action现在变成：
action_id + emotion + intensity + risk_level
↓
speed
↓
run_action(action_id, speed)这样 Duo S 端就成为了一个真正参与情绪表达的端侧控制器，而不是简单的编号执行器。
十四、主程序集成：通信、执行报告与速度控制闭环
最后需要把 CloudClient、NetworkSafetyManager、ExecutionReport、ActionScheduler 集成到 main.py 中。
新的主循环流程为：
初始化 RobotController
初始化 CloudClient
初始化 NetworkSafetyManager
初始化 ActionScheduler
↓
轮询云端任务
↓
如果网络失败：进入网络保护判断
↓
如果收到任务：解析 command_id/action_sequence/emotion/intensity
↓
创建 ExecutionReport
↓
调用 ActionScheduler 执行动作
↓
执行成功或失败后生成报告
↓
ACK 回传执行报告
↓
进入下一轮十五、使用 AI 辅助生成完整主程序
提示词 7：生成集成版 main.py
请帮我生成一个 Milk-V Duo S 端主程序 main.py，集成可靠通信、网络异常保护、执行反馈报告和情绪速度自适应控制。
已有模块：
1. RobotController
2. ActionScheduler
3. CloudClient
4. NetworkSafetyManager
5. ExecutionReport
6. RobotState
云端 poll 接口返回示例：
{
"command_id": "cmd_001",
"action_sequence": [11, 12, 0],
"emotion": "happy",
"intensity": 0.8
}
要求：
1. 初始化 robot、scheduler、cloud、network_safety。
2. 使用 last_command_id 防止重复执行。
3. 每轮先 poll_action。
4. 如果 poll 失败，调用 network_safety.handle_network_status。
5. 如果返回 no_action，继续轮询。
6. 如果收到 action_sequence：
- 创建 ExecutionReport
- 调用 scheduler.run_sequence(sequence, emotion, intensity, report)
- 根据执行结果 mark_done 或 mark_failed
- 调用 cloud.ack_action(report.to_dict(network_status))
7. ACK 失败也要打印日志。
8. 根据 cloud.get_poll_interval() 动态 sleep。
9. KeyboardInterrupt 时停止机器人、执行站立并关闭串口。AI 生成代码：main.py
import time
from robot_state import RobotState
from robot_controller import RobotController
from action_scheduler import ActionScheduler
from cloud_client import CloudClient
from network_safety import NetworkSafetyManager
from execution_report import ExecutionReport
SERVER = "http://47.93.27.196:8765"
CLIENT_ID = "milk_duos_001"
TOKEN = None
def main():
print("[Main] Milk-V Duo S 机器人端程序启动")
robot = RobotController(
port="/dev/ttyS0",
baudrate=115200,
default_speed=100
)
scheduler = ActionScheduler(robot)
cloud = CloudClient(
server=SERVER,
client_id=CLIENT_ID,
token=TOKEN,
timeout=5,
base_poll_interval=2,
offline_threshold=10
)
network_safety = NetworkSafetyManager(robot)
last_command_id = None
try:
while True:
scheduler.set_state(RobotState.POLLING)
data = cloud.poll_action()
network_status = cloud.get_network_status()
if data is None:
network_safety.handle_network_status(network_status)
scheduler.set_state(RobotState.IDLE)
sleep_time = cloud.get_poll_interval()
print(f"[Main] 网络异常，{sleep_time} 秒后重试")
time.sleep(sleep_time)
continue
network_safety.handle_network_status(network_status)
if data.get("status") == "no_action":
scheduler.set_state(RobotState.IDLE)
time.sleep(cloud.get_poll_interval())
continue
command_id = data.get("command_id")
sequence = data.get("action_sequence")
emotion = data.get("emotion")
intensity = data.get("intensity")
if not command_id or not sequence:
print(f"[Main] 云端响应缺少 command_id 或 action_sequence: {data}")
scheduler.set_state(RobotState.IDLE)
time.sleep(cloud.get_poll_interval())
continue
if command_id == last_command_id:
print(f"[Main] 检测到重复 command_id，跳过执行: {command_id}")
scheduler.set_state(RobotState.IDLE)
time.sleep(cloud.get_poll_interval())
continue
scheduler.set_state(RobotState.RECEIVED)
print(
f"[Main] 收到动作任务: command_id={command_id}, "
f"sequence={sequence}, emotion={emotion}, intensity={intensity}"
)
report = ExecutionReport(
command_id=command_id,
client_id=CLIENT_ID,
original_sequence=sequence,
emotion=emotion,
intensity=intensity
)
execute_result = scheduler.run_sequence(
sequence=sequence,
emotion=emotion,
intensity=intensity,
report=report
)
if execute_result.get("ok"):
report.mark_done()
else:
report.mark_failed(execute_result.get("reason"))
scheduler.set_state(RobotState.ACKING)
ack_ok = cloud.ack_action(
report.to_dict(
network_status=cloud.get_network_status()
)
)
if ack_ok:
print("[Main] 执行报告已成功回传云端")
else:
print("[Main] 执行报告回传失败，等待下一轮继续运行")
last_command_id = command_id
scheduler.set_state(RobotState.IDLE)
time.sleep(cloud.get_poll_interval())
except KeyboardInterrupt:
print("[Main] 用户手动终止程序")
try:
robot.stop()
time.sleep(1)
robot.stand()
time.sleep(5)
except Exception as e:
print(f"[Main] 退出时安全回正失败: {e}")
robot.close()
print("[Main] 程序已退出")
if __name__ == "__main__":
main()技术理解
这份主程序把本周设计的几个模块串成了完整闭环：
云端任务
↓
可靠轮询
↓
网络异常保护
↓
执行报告创建
↓
状态机动作调度
↓
情绪速度自适应
↓
执行结果记录
↓
ACK 报告回传这样 Duo S 端不只是“收到编号就动”，而是每次动作执行都有完整上下文：
- 是哪个 command_id；
- 原始动作序列是什么；
- 端侧规范化后执行了什么；
- 情绪类型是什么；
- 情绪强度是多少；
- 每个动作速度是多少；
- 是否执行成功；
- 失败原因是什么；
- 当前网络状态如何。
这对于后续调试、演示和项目答辩都非常有帮助。
十六、测试样例设计
为了验证速度控制是否有效，我设计了几组测试输入。
测试 1：开心强情绪
云端返回：
{
"command_id": "cmd_happy_001",
"action_sequence": [11, 12],
"emotion": "happy",
"intensity": 0.9
}Duo S 端可能规范化为：
[0, 11, 12, 0]其中普通站立动作速度可以接近默认值，11-前进招手 因为是高风险动作，速度被限制在 100 左右；12-鞠躬谢谢 是中风险动作，速度可以略高，但不会超过 115。
测试 2：害怕强情绪
云端返回：
{
"command_id": "cmd_fear_001",
"action_sequence": [13],
"emotion": "fear",
"intensity": 0.8
}Duo S 端会执行类似：
PL0 SQ0 SM75 ONCE
PL0 SQ13 SM75 ONCE
PL0 SQ0 SM75 ONCE这里动作速度会变慢，让下蹲保护动作更符合害怕、收缩的情绪。
测试 3：感谢语义
云端返回：
{
"command_id": "cmd_thanks_001",
"action_sequence": [12],
"emotion": "thanks",
"intensity": 0.6
}Duo S 端会让鞠躬谢谢动作以较柔和的速度执行，不会太快，避免感谢动作显得突兀。
十七、本周优化后的文件结构
本次优化后，Milk-V Duo S 端代码结构可以设计为：
milkv-Duo-S/
├── main.py # 主程序入口，集成通信、调度、反馈
├── cloud_client.py # 可靠通信客户端
├── network_safety.py # 网络异常与离线保护
├── execution_report.py # 执行反馈报告
├── speed_controller.py # 情绪强度速度自适应控制
├── action_scheduler.py # 动作调度状态机
├── action_validator.py # 动作安全校验
├── action_meta.py # 动作元数据和风险分级
├── robot_state.py # 状态机枚举
├── robot_controller.py # 串口控制封装
└── test_servo.py # 本地测试脚本相比最初版本，现在 Duo S 端已经形成了一个比较完整的端侧执行框架。
十八、本周总结
本周我将原本的 Duo S 端轮询与动作执行逻辑进一步升级，主要完成了两个方向的设计：
第一，设计了可靠通信与执行反馈机制。通过 CloudClient、NetworkSafetyManager 和 ExecutionReport，Duo S 端能够处理网络异常、动态调整轮询间隔、进入离线保护，并把真实执行结果回传云端。这样云端和前端不再只是知道“动作已发送”，而是能够知道“机器人是否真正执行完成”。
第二，设计了基于情绪强度的动作速度自适应控制机制。通过 SpeedController，Duo S 端可以根据云端返回的 emotion 和 intensity 动态调整串口指令中的 SM 速度参数。这样同一个动作组在不同情绪下会有不同表现：开心时更活泼，难过时更缓慢，感谢时更柔和，害怕时更收缩。
通过这次优化，我对 Milk-V Duo S 端在整个项目中的作用有了更深入的认识：
云端 OpenClaw 负责理解情绪和生成动作序列，但是机器人最终能不能稳定、自然、安全地表达情绪，关键还取决于 Duo S 端。Duo S 端不仅要负责“执行动作”，还要负责通信可靠性、异常保护、执行反馈、速度控制和机械安全约束。
也就是说，Duo S 端不是一个简单的动作播放器，而是一个运行在嵌入式设备上的机器人端执行控制系统。它把云端生成的抽象情绪和动作编号，进一步转换成带有速度、节奏、反馈和安全策略的真实机器人身体表达。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
6
0- 0
已为社区贡献4条内容
所有评论(0)