这是课程的终极项目——将前24课学到的所有技术集成为一个完整的智能服务机器人系统：
1. 集成导航、语音、视觉、操控四大核心能力
2. 实现完整的感知→理解→决策→执行→反馈闭环
3. 支持酒店/医院/商场/迎宾等多种场景
4. 具备异常处理、远程监控、OTA更新能力
5. 完成从仿真到实机的部署验证import json, math, random, time
from collections import defaultdict
class IntelligentServiceRobot:
"""智能服务机器人 - 完整系统集成"""
def __init__(self, name="SmartBot-01"):
self.name = name
# 子系统
self.navigation = NavigationSystem()
self.interaction = InteractionSystem()
self.task_manager = TaskManager()
self.health = HealthMonitor()
self.cloud = CloudConnector()
self.state = "idle"
self.position = {"x": 0, "y": 0, "floor": 1}
self.battery = 100
self.session_log = []
def process_request(self, user_input, user_id="guest"):
"""处理用户请求 - 完整流水线"""
log = [f"[输入] {user_input}"]
# 1. 语音识别
asr_result = self.interaction.asr(user_input)
log.append(f"[ASR] {asr_result}")
# 2. 意图理解
nlu_result = self.interaction.nlu(asr_result)
log.append(f"[NLU] 意图={nlu_result['intent']} 槽位={nlu_result['slots']}")
# 3. 情感检测
emotion = self.interaction.detect_emotion(user_input)
log.append(f"[情感] {emotion['emotion']} ({emotion['confidence']:.0%})")
# 4. 任务规划
task = self.task_manager.create_task(nlu_result, emotion)
log.append(f"[任务] {task['id']}: {task['type']}")
# 5. 执行
result = self.execute_task(task)
log.append(f"[执行] {'成功' if result['success'] else '失败'}")
# 6. 语音反馈
response = self.interaction.generate_response(result, emotion)
log.append(f"[回复] {response}")
# 7. 健康检查
self.health.update(self.battery, self.state)
self.session_log.extend(log)
return {"response": response, "task_id": task["id"], "logs": log}
def execute_task(self, task):
self.battery -= random.uniform(1, 3)
self.state = "busy"
result = {"success": random.random() > 0.05, "task_id": task["id"]}
self.state = "idle"
return result
class NavigationSystem:
def navigate(self, target): return {"distance": random.uniform(5, 50), "time": random.uniform(30, 180)}
class InteractionSystem:
def asr(self, text): return text
def nlu(self, text):
intents = {"去": "navigate", "送": "deliver", "你好": "greet", "查询": "query"}
for kw, intent in intents.items():
if kw in text: return {"intent": intent, "slots": {"target": text.replace(kw,"")}}
return {"intent": "unknown", "slots": {}}
def detect_emotion(self, text):
emotions = {"开心": "happy", "生气": "angry", "着急": "anxious"}
for kw, emo in emotions.items():
if kw in text: return {"emotion": emo, "confidence": 0.8}
return {"emotion": "neutral", "confidence": 0.6}
def generate_response(self, result, emotion):
if result["success"]:
return "好的，马上为您处理！"
return "抱歉，出了点问题，我再来一次。"
class TaskManager:
def __init__(self): self.counter = 0
def create_task(self, nlu, emotion):
self.counter += 1
return {"id": f"T{self.counter:04d}", "type": nlu["intent"], "priority": 3 if emotion["emotion"] == "anxious" else 2}
class HealthMonitor:
def update(self, battery, state):
if battery < 20: return {"alert": "low_battery"}
return {"status": "ok"}
class CloudConnector:
def sync(self): return {"synced": True}
robot = IntelligentServiceRobot()
print("智能服务机器人 - 完整系统集成")
print("=" * 55)
requests = [
"你好，请带我去3楼会议室",
"帮我送咖啡给张总，比较着急",
"查询洗手间在哪里",
]
for req in requests:
print(f"\n🎤 用户: {req}")
result = robot.process_request(req)
print(f"🤖 机器人: {result['response']}")
for log in result['logs']:
print(f" {log}")
print(f"\n📊 电量: {robot.battery:.1f}%")
print("✅ 完整系统集成验证通过")class SystemArchitecture:
"""系统架构文档"""
ARCHITECTURE = {
"应用层": {
"任务规划": "BehaviorTree + FSM",
"对话管理": "状态机 + LLM兜底",
"场景理解": "多模态融合",
},
"能力层": {
"导航": "Nav2 (A* + DWA + SLAM)",
"语音": "Whisper(ASR) + VITS(TTS)",
"视觉": "YOLOv8 + ArcFace + MediaPipe",
"操控": "逆运动学 + 抓取规划",
},
"通信层": {
"中间件": "ROS2 (DDS)",
"云端": "MQTT + REST API",
"前端": "WebSocket",
},
"硬件层": {
"传感器": "激光雷达 + RGB-D + 麦克风阵列",
"执行器": "差速驱动 + 机械臂",
"计算": "NVIDIA Jetson + 边缘推理",
},
}
def print_architecture(self):
for layer, components in self.ARCHITECTURE.items():
print(f"\n🏗️ {layer}:")
for comp, tech in components.items():
print(f" ├── {comp}: {tech}")
arch = SystemArchitecture()
print("系统架构文档")
print("=" * 55)
arch.print_architecture()
print("\n✅ 架构文档生成完成")class DeploymentChecklist:
"""部署检查清单"""
def __init__(self):
self.checks = {
"硬件": [
{"item": "激光雷达标定", "status": True},
{"item": "相机内外参标定", "status": True},
{"item": "IMU校准", "status": True},
{"item": "轮式里程计校准", "status": True},
{"item": "机械臂零点标定", "status": False},
],
"软件": [
{"item": "ROS2节点通信测试", "status": True},
{"item": "导航建图验证", "status": True},
{"item": "语音交互测试", "status": True},
{"item": "电梯协议对接", "status": False},
{"item": "云端服务连通性", "status": True},
],
"安全": [
{"item": "急停按钮测试", "status": True},
{"item": "碰撞检测验证", "status": True},
{"item": "速度限制设置", "status": True},
{"item": "隐私数据加密", "status": False},
{"item": "网络安全配置", "status": True},
],
"运维": [
{"item": "充电桩对接", "status": True},
{"item": "远程监控连通", "status": True},
{"item": "告警通知配置", "status": False},
{"item": "OTA更新测试", "status": True},
{"item": "日志收集确认", "status": True},
],
}
def run_check(self):
total = 0; passed = 0
print("部署检查清单")
print("=" * 55)
for category, items in self.checks.items():
print(f"\n📋 {category}:")
for item in items:
status = "✅" if item["status"] else "❌"
print(f" {status} {item['item']}")
total += 1
if item["status"]: passed += 1
rate = passed / total * 100
print(f"\n📊 通过率: {passed}/{total} ({rate:.0f}%)")
if rate < 100:
print("⚠️ 存在未通过项目，请完成后再部署！")
else:
print("🎉 全部通过，可以部署！")
return rate
checklist = DeploymentChecklist()
checklist.run_check()| 阶段 | 课程 | 核心收获 |
|---|---|---|
| 🧭 导航 | 1-5 | SLAM、A*、避障、电梯、多楼层 |
| 🗣️ 交互 | 6-10 | ASR/TTS、NLU、对话管理、情感、多模态 |
| 🎯 任务 | 11-15 | 行为树、物体识别、抓取、引导、人脸 |
| 🔧 集成 | 16-20 | ROS2、云端、监控、调度、异常处理 |
| 🚀 实战 | 21-25 | 酒店、医院、商场、迎宾、毕业项目 |
完成端到端场景演示：选择一个场景（酒店/医院/商场），从用户开口说话到任务完成的完整流程，所有子系统协同工作。
撰写技术论文：总结你的系统架构设计、关键技术选择、性能优化策略，3000字以上。
实现实机迁移：将仿真代码迁移到真实机器人上，记录仿真与实机的差异和适配方案。