BRICS-FS-58

人工智能应用创新

项目式教学教材 —— 从零基础到独立完成竞赛项目

面向赛项：2026 金砖国家职业技能大赛 · 人工智能应用创新（BRICS-FS-58）

覆盖模块：模块1 情感计算之人脸表情识别 / 模块2 情感计算之人体行为识别 / 模块3 具身智能平台部署与运维 / 模块4 智能体应用开发

适用对象：零基础初学者 · 中职 / 高职 / 应用型本科学生

教材版本：v1.0 · 2026年

目录

  1. 前言
  2. 第0章 开发环境搭建
    1. 0.1 Python 与 Anaconda 安装
    2. 0.2 核心库安装（PyTorch / OpenCV / Ultralytics / Flask / FastAPI）
    3. 0.3 Docker 与 ROS2 环境准备
    4. 0.4 Dify 与 OpenClaw 平台简介
    5. 0.5 DeepSeek API 申请与配置
  3. 模块1：情感计算之人脸表情识别
    1. 1.1 项目导读与学习目标
    2. 1.2 知识准备
    3. 1.3 环境与数据准备
    4. 1.4 项目实施
    5. 1.5 项目总结与拓展
  4. 模块2：情感计算之人体行为识别
    1. 2.1 项目导读与学习目标
    2. 2.2 知识准备
    3. 2.3 环境与数据准备
    4. 2.4 项目实施
    5. 2.5 项目总结与拓展
  5. 模块3：具身智能平台部署与运维
    1. 3.1 项目导读与学习目标
    2. 3.2 知识准备
    3. 3.3 环境与数据准备
    4. 3.4 项目实施
    5. 3.5 项目总结与拓展
  6. 模块4：智能体应用开发
    1. 4.1 项目导读与学习目标
    2. 4.2 知识准备
    3. 4.3 环境与数据准备
    4. 4.4 项目实施
    5. 4.5 项目总结与拓展

前言

为什么要编写这本教材

2026年金砖国家职业技能大赛"人工智能应用创新"赛项（BRICS-FS-58）围绕人工智能视觉应用领域展开，考察学生在AI算法原理理解、技术框架选型、模型训练与优化、智能体应用开发落地等方面的综合能力。赛项涵盖情感计算、具身智能、智能体开发三大前沿方向，技术栈涉及 PyTorch、YOLO、ROS2、Docker、Dify、LangChain 等，对参赛选手的知识广度和工程实践能力提出了很高要求。

然而，现有教材往往只覆盖单一技术方向，缺乏将竞赛各模块串联起来的系统性教学材料。零基础学生面对如此多的技术栈，往往不知从何入手。本教材正是为解决这一痛点而编写——以竞赛样题为蓝本，将每个模块拆解为可独立完成的子任务，从环境搭建到代码实现逐步引导，确保零基础学生能够跟随教材独立完成全部项目。

教材特色

逐题讲解，代码完整

教材以选拔赛样题为基础，覆盖全部4个模块共20道任务，每道题提供完整的题目要求、思路分析、代码实现、代码详解和易错提示五个部分，学生可以照着代码一步步运行出结果。

零基础友好

每个模块开头都设有"知识准备"环节，用通俗语言解释该模块涉及的核心概念和技术原理。代码部分不仅有完整代码，还有逐行注释和详细解释，确保初学者能看懂每一行代码的含义。

实战导向

教材内容紧扣竞赛实际要求，所有代码均可在本地运行验证。每个任务都配有"验证"环节，帮助学生确认代码执行结果是否正确。同时标注了竞赛中常见的易错点和踩坑提示。

赛项概况

项目说明
赛项名称2026金砖国家职业技能大赛 · 人工智能应用创新（BRICS-FS-58）
参赛形式双人赛，两位选手自行分配工作任务
参赛对象技师学院、中职、高职、应用型本科在籍学生或教师
竞赛时长区域选拔赛 4 小时
技术栈Python、PyTorch、Ubuntu、ROS2、Docker、Dify、OpenClaw、LangChain
序号模块分值比例核心内容
1情感计算之人脸表情识别25%FER2013数据集、VGG/ResNet/MobileNet/MobileViT模型训练、ONNX导出、Flask推理服务
2情感计算之人体行为识别25%YOLO8n-POSE姿态估计、ByteTrack目标跟踪、MMAction2视频识别、FastAPI服务
3具身智能平台部署与运维25%Docker容器编排、ROS2 Humble、Dify平台、OpenClaw智能体、DeepSeek大模型
4智能体应用开发25%Dify智能体、RAG知识库、OpenClaw Skills插件、LangChain Function Calling

如何使用本教材

第0章 开发环境搭建

本章将帮助你搭建好竞赛所需的全部开发环境。请务必按照顺序逐步完成，确保每个工具都能正常运行后再进入后续模块的学习。

0.1 Python 与 Anaconda 安装

竞赛使用 Python 3.10+ 作为主要编程语言。推荐使用 Anaconda 管理Python环境和依赖包。

安装 Anaconda

访问 Anaconda 官网下载页面，根据操作系统选择对应版本。安装完成后，打开终端验证：

# 验证 conda 是否安装成功
conda --version
# 应输出类似：conda 24.x.x

# 创建竞赛专用虚拟环境
conda create -n brics_ai python=3.10 -y
conda activate brics_ai

# 验证 Python 版本
python --version
# 应输出：Python 3.10.x
提示 国内用户建议配置清华大学镜像源以加速包安装：conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/ && conda config --set show_channel_urls yes

0.2 核心库安装

在 brics_ai 环境中安装竞赛所需的核心 Python 库：

# PyTorch（CPU版本，无GPU时使用）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu

# 如有NVIDIA GPU，安装CUDA版本（显著加速训练）
# pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121

# 计算机视觉与人脸检测
pip install opencv-python insightface onnxruntime

# YOLO 姿态估计与目标跟踪
pip install ultralytics bytetracker

# MMAction2 视频行为识别
pip install mmaction2

# Web 框架
pip install flask fastapi uvicorn

# 模型导出与评估
pip install onnx scikit-learn matplotlib numpy pillow

# LangChain 智能体开发
pip install langchain langgraph langchain-community
注意 PyTorch 安装较大（约2GB），请确保网络通畅。如果安装速度慢，可使用清华镜像：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple torch torchvision

0.3 Docker 与 ROS2 环境准备

模块3和模块4需要使用 Docker 和 ROS2。这两部分在 Ubuntu 22.04 环境下运行最佳。

Docker 安装（Ubuntu 22.04）

# 更新包索引
sudo apt update

# 安装 Docker Engine
sudo apt install -y docker.io docker-compose-plugin

# 将当前用户加入 docker 组（免 sudo）
sudo usermod -aG docker $USER

# 重新登录后验证
docker --version
docker compose version

ROS2 Humble 安装

# 设置 locale
sudo apt update && sudo apt install locales -y
sudo locale-gen en_US en_US.UTF-8
sudo update-locale LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8
export LANG=en_US.UTF-8

# 添加 ROS2 仓库
sudo apt install software-properties-common -y
sudo add-apt-repository universe -y
sudo apt update && sudo apt install curl -y
sudo curl -sSL https://raw.githubusercontent.com/ros/rosdistro/master/ros.key -o /usr/share/keyrings/ros-archive-keyring.gpg
echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/ros-archive-keyring.gpg] http://packages.ros.org/ros2/ubuntu $(. /etc/os-release && echo $UBUNTU_CODENAME) main" | sudo tee /etc/apt/sources.list.d/ros2.list > /dev/null

# 安装 ROS2 Humble
sudo apt update
sudo apt install ros-humble-desktop -y

# 验证安装
source /opt/ros/humble/setup.bash
ros2 --help
说明 Windows 用户建议使用 WSL2（Windows Subsystem for Linux 2）安装 Ubuntu 22.04，再在其中安装 Docker 和 ROS2。macOS 用户可使用虚拟机或云服务器。

0.4 Dify 与 OpenClaw 平台简介

Dify —— 开源智能体开发平台

Dify 是一个开源的 LLM（大语言模型）应用开发平台，支持可视化编排 AI 工作流、构建知识库（RAG）、创建智能体应用等。竞赛中需要通过 Docker Compose 部署 Dify，并配置 DeepSeek 大模型接入。

官方文档：https://docs.dify.ai

OpenClaw —— 具身智能体框架

OpenClaw 是一个具身智能体开发框架，支持通过自然语言指令控制机器人。它可以将大模型（如 DeepSeek）的输出转化为机器人控制指令，实现"说一句话，机器人就动"的效果。竞赛中需要部署 OpenClaw 并对接 ROS2 仿真环境。

0.5 DeepSeek API 申请与配置

竞赛中模块3和模块4都需要使用 DeepSeek 大模型 API。DeepSeek 提供兼容 OpenAI 格式的 API 接口。

申请 API Key

  1. 访问 DeepSeek 开放平台：https://platform.deepseek.com
  2. 注册账号并登录
  3. 在"API Keys"页面创建新的 API Key
  4. 复制并保存 API Key（格式如 sk-xxxxxxxxxxxxxxxx）

测试 API 连接

from openai import OpenAI

client = OpenAI(
    api_key="sk-你的API密钥",
    base_url="https://api.deepseek.com/v1"
)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "你好，请介绍一下自己"}],
    max_tokens=200
)

print(response.choices[0].message.content)
验证成功 如果配置正确，运行上述代码后应能看到 DeepSeek 返回的自我介绍文字。如果报错"Authentication Fails"，请检查 API Key 是否正确。
重要提醒 请妥善保管 API Key，不要将其提交到 Git 仓库或分享给他人。竞赛中建议将 Key 存储在环境变量中：export DEEPSEEK_API_KEY="sk-你的密钥"
模块1 · 25%

情感计算之人脸表情识别

基于 PyTorch + FER2013 数据集，完成表情识别模型训练、ONNX导出、Flask推理服务与前端界面开发

1.1 项目导读与学习目标

项目背景

情感计算（Affective Computing）是人工智能领域的重要分支，旨在让计算机能够识别、理解、处理和模拟人类的情感。人脸表情识别是情感计算中最直观、应用最广泛的技术之一，它在智能教育、医疗辅助、驾驶员疲劳监测、市场调研、游戏交互等场景中都有着重要价值。

本项目以国际经典的 FER2013 数据集为基础，要求选手基于 PyTorch 深度学习框架，使用 InsightFace / OpenCV 进行人脸检测，从 VGG-16、ResNet-50、MobileNetV3-Small、MobileViT-XS 四种视觉模型中选择并训练表情识别模型，最终将模型导出为 ONNX 格式，并封装为 Flask 推理服务，配套开发前端界面，形成完整的"训练-导出-部署-展示"全链路应用系统。

项目整体流程

1数据加载与增强
→
2模型训练与评估
→
3ONNX 导出与推理
→
4Flask 推理服务
→
5前端界面对接

学习目标

  • 知识目标：理解人脸表情识别的基本原理；掌握 FER2013 数据集的结构与特点；了解 VGG、ResNet、MobileNet、MobileViT 四类经典视觉模型的架构差异。
  • 技能目标：能够使用 torchvision 加载图像数据并构建数据增强流水线；能够使用预训练模型进行迁移学习与训练评估；能够将 PyTorch 模型导出为 ONNX 并编写推理引擎；能够使用 Flask 封装 RESTful 推理接口；能够开发前端页面并完成接口对接。
  • 素养目标：建立"数据-模型-部署-应用"的完整工程思维；具备模型轻量化部署的意识；理解前后端分离架构在 AI 应用中的价值。

1.2 知识准备

知识卡片1：人脸表情识别

人脸表情识别（Facial Expression Recognition, FER）是指通过计算机分析人脸图像，自动判断出表情类别的技术。根据心理学家 Ekman 的研究，人类基本表情可分为七类：愤怒（angry）、厌恶（disgust）、恐惧（fear）、开心（happy）、悲伤（sad）、惊讶（surprise）、中性（neutral）。

一个完整的表情识别系统通常包含三个核心环节：人脸检测（从图像中定位人脸位置）→ 人脸对齐（将人脸归一化到统一尺寸与姿态）→ 表情分类（用分类模型输出表情类别）。本项目在训练阶段直接使用裁剪好的人脸图像，在推理服务阶段通过 OpenCV/InsightFace 进行人脸检测。

知识卡片2：FER2013 数据集

FER2013 是 2013 年 Kaggle 面部表情识别挑战赛使用的数据集，包含 35,887 张 48×48 灰度人脸图像，每张图像标注了七种表情之一。其训练集 28,709 张，测试集 7,178 张。

表情类别标签训练集数量测试集数量
愤怒angry3,995958
厌恶disgust436111
恐惧fear4,0971,024
开心happy7,2151,774
悲伤sad4,8301,247
惊讶surprise3,171831
中性neutral4,9651,233

注意：数据集存在类别不均衡问题（disgust 类样本远少于 happy 类），这是训练时需要关注的重点。在使用 ImageFolder 加载时，数据需按"类别名/图片"的文件夹结构组织。

知识卡片3：四种视觉模型对比

本项目可选四种预训练模型，它们在结构、参数量、适用场景上各有特点：

模型核心思想参数量适用场景
VGG-16堆叠 3×3 小卷积，结构简单规整约 138M准确率优先，计算资源充足
ResNet-50残差连接（Residual），解决深层网络退化约 25M精度与速度均衡，通用首选
MobileNetV3-Small深度可分离卷积 + SE 注意力 + 反残差结构约 2.5M移动端、嵌入式设备
MobileViT-XS将 Transformer 引入轻量 CNN，融合局部与全局特征约 1.3M移动端且追求更高精度

四类模型均可通过 torchvision 或 timm 库加载预训练权重，通过迁移学习适配 FER2013 的七分类任务。

知识卡片4：数据增强

数据增强（Data Augmentation）是通过对训练图像进行随机变换来扩充数据多样性的技术，能有效缓解过拟合、提升模型泛化能力。对于人脸表情识别，常用的增强方式包括：

  • RandomResizedCrop：随机裁剪并缩放，模拟人脸位置与尺度变化
  • RandomHorizontalFlip：随机水平翻转（人脸左右对称，不影响表情）
  • ColorJitter：亮度/对比度/饱和度抖动，模拟光照变化
  • Normalize：标准化像素值，加速模型收敛

注意：测试集只做 Resize + CenterCrop + Normalize，不能做随机翻转或裁剪，否则评估结果不准确。

知识卡片5：ONNX 与推理部署

ONNX（Open Neural Network Exchange）是微软与 Facebook 联合推出的开放神经网络交换格式。将 PyTorch 模型导出为 ONNX 后，可脱离 PyTorch 框架，使用 ONNX Runtime 进行高效推理，具备跨平台、低延迟、易部署的优势。

导出 ONNX 的核心步骤：1）将模型设为 eval 模式；2）构造一个示例输入张量；3）调用 torch.onnx.export() 导出。推理时使用 onnxruntime.InferenceSession 加载模型并执行前向计算。

知识卡片6：Flask 与 RESTful 接口

Flask 是 Python 轻量级 Web 框架，适合快速搭建推理服务后端。RESTful 是一种接口设计风格，通过 HTTP 方法（GET/POST）与 URL 路径表达资源操作。本项目通过 @app.route('/predict', methods=['POST']) 暴露表情识别接口，前端上传图片，后端返回 JSON 格式的表情类别与置信度。

1.3 环境与数据准备

软件环境

组件推荐版本用途
Python3.9 / 3.10运行环境
PyTorch2.0+ (含 torchvision)模型训练与导出
onnx / onnxruntime1.14+ / 1.15+ONNX 导出与推理
Flask2.3+推理服务后端
OpenCV (opencv-python)4.8+图像处理与人脸检测
InsightFace0.7+人脸检测（备选方案）
scikit-learn1.3+评估指标计算
numpy / Pillow最新稳定版数值计算与图像读取

依赖安装

在终端中执行以下命令安装所需依赖：

# 安装 PyTorch（CUDA 版按官网选择对应命令，以下为 CPU 版示例）
pip install torch torchvision

# 安装 ONNX 相关
pip install onnx onnxruntime

# 安装 Web 服务与图像处理
pip install flask opencv-python insightface

# 安装评估与工具库
pip install scikit-learn numpy Pillow

数据集目录结构

使用 ImageFolder 加载数据要求目录按"类别名/图片"层级组织。请将 FER2013 数据集整理为如下结构：

dataset_data/
└── fer2013/
    ├── train/
    │   ├── angry/      # 内含愤怒表情图片
    │   ├── disgust/
    │   ├── fear/
    │   ├── happy/
    │   ├── sad/
    │   ├── surprise/
    │   └── neutral/
    └── test/
        ├── angry/
        ├── disgust/
        ├── fear/
        ├── happy/
        ├── sad/
        ├── surprise/
        └── neutral/

说明：原始 FER2013 为 CSV 格式（每行一像素序列+标签），需先转换为按类别分文件夹的图片格式。竞赛环境通常已预处理完毕，选手直接使用即可。

1.4 项目实施

任务 1 5 分

数据加载与增强流水线

题目要求

编写代码使用 torchvision.datasets.ImageFolder 加载 dataset_data/fer2013/train 训练数据，实现数据增强流水线，用 DataLoader 创建模型训练加载器。

  • 使用 ImageFolder 加载训练集，并应用数据增强变换（Resize、RandomResizedCrop、RandomHorizontalFlip、ColorJitter、Normalize）。
  • 使用 DataLoader 创建训练数据加载器，设置 batch_size、shuffle、num_workers。
  • 打印类别映射与数据集大小，验证加载成功。

思路分析

  1. 定义变换流水线：使用 transforms.Compose 将多个增强操作串联。训练集用随机裁剪+翻转+颜色抖动提升泛化性；测试集仅做 Resize+中心裁剪+标准化。
  2. 加载 ImageFolder：传入数据路径和 transform，ImageFolder 会自动将子文件夹名映射为类别标签（按字母序编号 0~6）。
  3. 构建 DataLoader：设置批大小、是否打乱、多进程加载，提升训练时数据读取效率。
  4. 验证检查：取出一个 batch 打印形状，确认通道数、类别数正确。

代码实现

import torch
from torch.utils.data import DataLoader
from torchvision import datasets, transforms

# ========================================
# 第一步：定义训练集与测试集的数据变换流水线
# ========================================

# FER2013 原图为 48x48 灰度图，迁移学习模型通常需要 224x224 的 3 通道输入
# 因此先 Resize 到 224，再进行增强

# ImageNet 标准化参数（预训练模型默认使用该参数）
IMAGENET_MEAN = [0.485, 0.456, 0.406]
IMAGENET_STD = [0.229, 0.224, 0.225]

# 训练集变换：包含随机裁剪、翻转、颜色抖动等增强操作
train_transform = transforms.Compose([
    transforms.Resize(224),                              # 缩放到 224x224
    transforms.RandomResizedCrop(224, scale=(0.8, 1.0)), # 随机裁剪缩放
    transforms.RandomHorizontalFlip(p=0.5),            # 50% 概率水平翻转
    transforms.ColorJitter(brightness=0.2, contrast=0.2), # 亮度对比度抖动
    transforms.ToTensor(),                                  # 转为张量，并归一化到 [0,1]
    transforms.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD) # 标准化
])

# 测试集变换：只做缩放和标准化，不做随机增强
test_transform = transforms.Compose([
    transforms.Resize(224),
    transforms.CenterCrop(224),       # 中心裁剪，保证尺寸一致
    transforms.ToTensor(),
    transforms.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD)
])

# ========================================
# 第二步：使用 ImageFolder 加载数据集
# ========================================

train_dir = "dataset_data/fer2013/train"    # 训练集路径
test_dir = "dataset_data/fer2013/test"     # 测试集路径

# ImageFolder 会自动扫描子文件夹，文件夹名即类别名
train_dataset = datasets.ImageFolder(train_dir, transform=train_transform)
test_dataset = datasets.ImageFolder(test_dir, transform=test_transform)

# 打印类别映射（文件夹名 → 数字标签）和数据集大小
print("类别映射:", train_dataset.class_to_idx)
print("训练集样本数:", len(train_dataset))
print("测试集样本数:", len(test_dataset))

# ========================================
# 第三步：创建 DataLoader 数据加载器
# ========================================

train_loader = DataLoader(
    train_dataset,
    batch_size=32,           # 每批 32 张图片
    shuffle=True,           # 训练时打乱顺序，提升泛化性
    num_workers=4,          # 使用 4 个子进程并行加载数据
    pin_memory=True          # GPU 训练时加速数据传输
)

test_loader = DataLoader(
    test_dataset,
    batch_size=32,
    shuffle=False,          # 测试时不打乱，保证结果可复现
    num_workers=4,
    pin_memory=True
)

# ========================================
# 第四步：验证数据加载是否正确
# ========================================

# 取出一个 batch 检查形状
images, labels = next(iter(train_loader))
print("一个batch的图片形状:", images.shape)   # 期望 [32, 3, 224, 224]
print("一个batch的标签:", labels.shape)      # 期望 [32]
print("标签取值范围:", labels.min().item(), "~", labels.max().item())

代码详解

  • transforms.Compose：将多个变换按顺序串联成一个流水线，数据会依次通过每个变换处理。
  • Resize(224)：FER2013 原图为 48×48，预训练模型输入尺寸通常为 224×224，需先放大。
  • RandomResizedCrop：随机裁剪并缩放回 224×224，scale=(0.8,1.0) 表示裁剪区域面积占原图的 80%~100%，模拟人脸尺度变化。
  • RandomHorizontalFlip：水平翻转不改变表情语义（开心的人左右翻转依然开心），是有效的增强手段。
  • ToTensor：将 PIL 图片转为张量，同时把像素值从 [0,255] 归一化到 [0,1]。
  • Normalize：用 ImageNet 均值方差标准化，使输入分布与预训练模型匹配。
  • ImageFolder：自动将子文件夹名映射为类别标签，class_to_idx 属性可查看映射关系。
  • DataLoader：负责按 batch 组织数据。pin_memory=True 在 GPU 训练时可加速数据从 CPU 到 GPU 的传输。

易错提示

  • 【灰度图通道问题】FER2013 是灰度图，ImageFolder 读取后是单通道。transforms.Resize 不改变通道数。若预训练模型需要 3 通道输入，需在 transform 中加入 transforms.Grayscale(num_output_channels=3) 将其复制为 3 通道。
  • 【测试集增强】测试集千万不能加 RandomResizedCrop、RandomHorizontalFlip 等随机操作，否则每次评估结果都不同。
  • 【num_workers】在 Windows 上 num_workers>0 时需将代码放入 if __name__ == '__main__': 中，否则会报多进程错误。
  • 【路径错误】确保路径 dataset_data/fer2013/train 存在且子文件夹为类别名。
任务 2 5 分

模型训练与评估

题目要求

编写模型训练代码，加载指定的预训练模型，初始化训练参数，编写评估函数，计算 accuracy_score、precision_score、recall_score、f1_score（均设置 average='macro'），返回包含四项指标，启动训练，按照执行的训练 epoch 周期进行训练。

  • 加载预训练模型（以 ResNet-50 为例），修改最后一层适配 7 分类。
  • 设置损失函数、优化器、学习率调度器。
  • 编写评估函数，计算四项 macro 指标并返回字典。
  • 按 epoch 循环训练，每个 epoch 结束后在测试集上评估。

思路分析

  1. 加载预训练模型：使用 torchvision.models.resnet50(weights=...) 加载 ImageNet 预训练权重，替换最后的全连接层为 7 输出。
  2. 迁移学习策略：前期冻结骨干网络只训练分类头，后期解冻微调，或直接全参数训练（数据量较大时可行）。
  3. 训练循环：每个 epoch 遍历 train_loader，前向传播→计算损失→反向传播→更新参数。
  4. 评估函数：在 test_loader 上收集所有预测值与真实值，用 sklearn 计算 4 项 macro 指标。
  5. 保存最佳模型：记录验证 f1 最高的模型权重，训练结束保存。

代码实现

import torch
import torch.nn as nn
from torch.optim import Adam
from torch.optim.lr_scheduler import StepLR
from torchvision import models
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import numpy as np

# ========================================
# 第一步：设备选择与模型加载
# ========================================

# 优先使用 GPU，没有则用 CPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("使用设备:", device)

# 加载预训练 ResNet-50（使用 ImageNet 权重）
model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)

# 修改最后一层全连接层，输出维度改为 7（七种表情）
num_features = model.fc.in_features    # 获取原全连接层输入维度（2048）
model.fc = nn.Linear(num_features, 7)  # 替换为 7 分类输出

model = model.to(device)              # 将模型移至 GPU/CPU

# ========================================
# 第二步：初始化训练参数
# ========================================

num_epochs = 20                 # 训练轮数
learning_rate = 0.001          # 学习率
weight_decay = 1e-4             # 权重衰减，防止过拟合

# 损失函数：交叉熵（内含 softmax，无需手动加）
criterion = nn.CrossEntropyLoss()

# 优化器：Adam，加入 weight_decay 做 L2 正则化
optimizer = Adam(model.parameters(), lr=learning_rate, weight_decay=weight_decay)

# 学习率调度器：每 5 个 epoch 学习率乘以 0.5
scheduler = StepLR(optimizer, step_size=5, gamma=0.5)

# ========================================
# 第三步：编写评估函数
# ========================================

@torch.no_grad()          # 评估时关闭梯度计算，节省显存
def evaluate(model, data_loader, criterion, device):
    """在测试集上评估模型，返回四项 macro 指标"""
    model.eval()             # 切换到评估模式（关闭 Dropout、BatchNorm 用统计量）
    
    all_preds = []           # 存储所有预测标签
    all_labels = []          # 存储所有真实标签
    total_loss = 0.0        # 累计损失
    
    for images, labels in data_loader:
        images = images.to(device)
        labels = labels.to(device)
        
        outputs = model(images)                    # 前向传播
        loss = criterion(outputs, labels)        # 计算损失
        total_loss += loss.item() * images.size(0)
        
        _, predicted = torch.max(outputs, 1)   # 取概率最大的类别
        
        all_preds.extend(predicted.cpu().numpy())
        all_labels.extend(labels.cpu().numpy())
    
    # 计算四项评估指标（均使用 macro 平均）
    accuracy = accuracy_score(all_labels, all_preds)
    precision = precision_score(all_labels, all_preds, average='macro', zero_division=0)
    recall = recall_score(all_labels, all_preds, average='macro', zero_division=0)
    f1 = f1_score(all_labels, all_preds, average='macro', zero_division=0)
    
    avg_loss = total_loss / len(data_loader.dataset)
    
    # 返回包含四项指标的字典
    return {
        'accuracy': accuracy,
        'precision': precision,
        'recall': recall,
        'f1_score': f1,
        'loss': avg_loss
    }

# ========================================
# 第四步：启动训练循环
# ========================================

best_f1 = 0.0               # 记录最佳 f1 值
save_path = "best_model.pth"  # 模型保存路径

for epoch in range(num_epochs):
    # ---------- 训练阶段 ----------
    model.train()           # 切换到训练模式
    running_loss = 0.0
    correct = 0
    total = 0
    
    for batch_idx, (images, labels) in enumerate(train_loader):
        images = images.to(device)
        labels = labels.to(device)
        
        optimizer.zero_grad()                      # 清空梯度
        outputs = model(images)                    # 前向传播
        loss = criterion(outputs, labels)        # 计算损失
        loss.backward()                        # 反向传播，计算梯度
        optimizer.step()                          # 更新参数
        
        running_loss += loss.item()
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()
    
    scheduler.step()              # 更新学习率
    train_acc = correct / total
    
    # ---------- 评估阶段 ----------
    metrics = evaluate(model, test_loader, criterion, device)
    
    print(f"Epoch [{epoch+1}/{num_epochs}] "
          f"Loss: {running_loss/len(train_loader):.4f} "
          f"TrainAcc: {train_acc:.4f} "
          f"TestAcc: {metrics['accuracy']:.4f} "
          f"F1: {metrics['f1_score']:.4f}")
    
    # 保存 f1 最优的模型
    if metrics['f1_score'] > best_f1:
        best_f1 = metrics['f1_score']
        torch.save(model.state_dict(), save_path)
        print(f"  -> 模型已保存，F1={best_f1:.4f}")

print("训练完成！最佳 F1:", best_f1)

代码详解

  • weights=ResNet50_Weights.IMAGENET1K_V2：加载改进版 ImageNet 预训练权重，比 V1 效果更好。这是新版 torchvision 推荐写法。
  • model.fc = nn.Linear(2048, 7)：ResNet 最后一层名为 fc，替换为 7 输出的全连接层。其他模型的替换层名不同：VGG 是 model.classifier[6]，MobileNetV3 是 model.classifier[3]。
  • model.train() / model.eval()：切换训练/评估模式。eval 模式下 Dropout 失效，BatchNorm 使用累积统计量而非 batch 统计量。
  • @torch.no_grad()：装饰器，评估时禁用梯度计算，减少显存占用、加快推理速度。
  • average='macro'：macro 平均先计算每个类别的指标，再取算术平均，对所有类别同等对待。适合关注少数类表现的场景。
  • zero_division=0：当某类无预测时避免除零警告，置为 0。
  • StepLR：每 step_size 个 epoch 将学习率乘以 gamma，逐步降低学习率有助于精细收敛。
  • optimizer.zero_grad()：PyTorch 梯度默认累加，每个 batch 前必须清零。

易错提示

  • 【忘记 model.eval()】评估时不切换 eval 模式，BatchNorm 会用当前 batch 统计量，导致评估结果不稳定。
  • 【忘记 zero_grad】不清空梯度会导致梯度累加，模型无法正常收敛。
  • 【指标参数】average='macro' 必须显式指定，否则二分类和多分类默认行为不同。题目明确要求 macro。
  • 【设备不匹配】模型在 GPU，数据在 CPU（或反之）会报错。务必将 images、labels 都 .to(device)。
  • 【保存内容】保存的是 state_dict()（仅权重），不是整个模型对象。加载时需先构建相同结构的模型。
  • 【不同模型替换层】VGG-16 替换 model.classifier[6] = nn.Linear(4096, 7)；MobileNetV3-Small 替换 model.classifier[3] = nn.Linear(1024, 7)。
任务 3 5 分

ONNX 导出与推理引擎

题目要求

将训练好的模型导出为 ONNX：编写推理引擎代码。

  • 加载训练好的最佳模型权重，将 PyTorch 模型导出为 ONNX 文件。
  • 编写 ONNX 推理引擎，加载 ONNX 模型并对单张图片进行表情预测。
  • 验证导出的 ONNX 模型与原 PyTorch 模型输出一致。

思路分析

  1. 导出准备：加载模型结构 → 加载权重 → 设为 eval 模式 → 构造示例输入张量（与训练时输入尺寸一致）。
  2. 调用 torch.onnx.export：指定输入名、输出名、动态 batch 维度（opset_version 建议 14 以上）。
  3. 推理引擎：用 onnxruntime 创建 Session，预处理图片（与测试变换一致），执行 session.run() 获取输出。
  4. 一致性验证：对同一张图分别用 PyTorch 和 ONNX 推理，对比输出差异。

代码实现

import torch
import torch.nn as nn
from torchvision import models, transforms
from PIL import Image
import numpy as np
import onnx
import onnxruntime as ort

# ========================================
# 第一部分：将 PyTorch 模型导出为 ONNX
# ========================================

# 1. 重新构建模型结构并加载训练好的权重
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

model = models.resnet50(weights=None)   # 不加载预训练权重，仅建结构
num_features = model.fc.in_features
model.fc = nn.Linear(num_features, 7)  # 替换为 7 分类

# 加载训练好的最佳权重
model.load_state_dict(torch.load("best_model.pth", map_location=device))
model = model.to(device)
model.eval()                          # 必须设为 eval 模式

# 2. 构造示例输入（用于追踪计算图）
# 尺寸必须与训练时一致：[batch, channel, height, width]
dummy_input = torch.randn(1, 3, 224, 224, device=device)

# 3. 导出 ONNX 文件
onnx_path = "fer_model.onnx"
torch.onnx.export(
    model,                          # 要导出的模型
    dummy_input,                    # 示例输入
    onnx_path,                      # 输出文件路径
    export_params=True,           # 导出模型参数（权重）
    opset_version=14,            # ONNX 算子集版本
    do_constant_folding=True,    # 优化常量折叠
    input_names=["input"],        # 输入节点名
    output_names=["output"],      # 输出节点名
    dynamic_axes={                 # 动态维度，支持可变 batch
        "input": {0: "batch_size"},
        "output": {0: "batch_size"}
    }
)
print("ONNX 模型已导出至:", onnx_path)

# 4. 验证 ONNX 模型格式正确
onnx_model = onnx.load(onnx_path)
onnx.checker.check_model(onnx_model)
print("ONNX 模型格式校验通过！")


# ========================================
# 第二部分：编写 ONNX 推理引擎
# ========================================

# 表情类别列表（需与训练时 class_to_idx 顺序一致）
EXPRESSIONS = ['angry', 'disgust', 'fear', 'happy', 'sad', 'surprise', 'neutral']

class ONNXInferenceEngine:
    """ONNX 推理引擎：加载 ONNX 模型并执行表情识别推理"""
    
    def __init__(self, model_path, providers=['CPUExecutionProvider']):
        """初始化推理引擎
        Args:
            model_path: ONNX 模型文件路径
            providers: 推理后端，可选 CUDAExecutionProvider（GPU）
        """
        self.session = ort.InferenceSession(model_path, providers=providers)
        self.input_name = self.session.get_inputs()[0].name
        self.output_name = self.session.get_outputs()[0].name
        
        # 预处理变换（必须与测试集一致）
        self.transform = transforms.Compose([
            transforms.Resize(224),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            )
        ])
    
    def preprocess(self, image):
        """预处理图片：转为 RGB → 变换 → 增加 batch 维"""
        if image.mode != 'RGB':
            image = image.convert('RGB')       # 灰度图转 3 通道
        img_tensor = self.transform(image)      # 应用变换 [3,224,224]
        img_array = img_tensor.cpu().numpy()     # 转 numpy
        img_array = np.expand_dims(img_array, axis=0)  # [1,3,224,224]
        return img_array.astype(np.float32)
    
    def predict(self, image):
        """对单张图片进行表情识别
        Args:
            image: PIL.Image 图片对象
        Returns:
            dict: 包含表情类别、置信度、各类概率
        """
        input_data = self.preprocess(image)
        
        # 执行 ONNX 推理
        outputs = self.session.run(
            [self.output_name],
            {self.input_name: input_data}
        )
        
        logits = outputs[0]                  # 原始输出 [1, 7]
        probs = torch.softmax(torch.from_numpy(logits), dim=1).numpy()[0]
        
        pred_idx = int(np.argmax(probs))    # 取最大概率索引
        pred_label = EXPRESSIONS[pred_idx]
        confidence = float(probs[pred_idx])
        
        return {
            'expression': pred_label,
            'confidence': round(confidence, 4),
            'all_probs': {EXPRESSIONS[i]: round(float(p), 4) for i, p in enumerate(probs)}
        }


# ========================================
# 第三部分：使用推理引擎并验证一致性
# ========================================

# 创建推理引擎实例
engine = ONNXInferenceEngine("fer_model.onnx")

# 加载测试图片
test_image = Image.open("test_sample.jpg")

# 使用 ONNX 推理
result = engine.predict(test_image)
print("ONNX 推理结果:", result)

# 对比 PyTorch 原模型输出（验证一致性）
with torch.no_grad():
    pt_input = torch.from_numpy(engine.preprocess(test_image)).to(device)
    pt_output = model(pt_input)
    pt_probs = torch.softmax(pt_output, dim=1).cpu().numpy()[0]

print("PyTorch 输出:", {EXPRESSIONS[i]: round(float(p),4) for i,p in enumerate(pt_probs)})
print("输出差异:", np.max(np.abs(result['all_probs'][EXPRESSIONS[0]] - pt_probs[0])))

代码详解

  • torch.onnx.export：核心导出函数。通过示例输入追踪模型的计算图，将所有操作转为 ONNX 算子。
  • export_params=True：将模型权重一同导出到 ONNX 文件中，否则导出的是无权重模型。
  • opset_version=14：算子集版本，版本越高支持的算子越多。建议 14 或 17。
  • dynamic_axes：指定第 0 维（batch）为动态，这样推理时可传入任意 batch 大小的输入。
  • do_constant_folding：常量折叠优化，将编译期可确定的常量计算提前执行，减小模型体积。
  • onnx.checker.check_model：校验 ONNX 文件格式是否合法，确保可被各推理框架正确加载。
  • ort.InferenceSession：ONNX Runtime 的推理会话，加载模型并管理推理后端。
  • providers：指定推理后端，CPUExecutionProvider 用 CPU，CUDAExecutionProvider 用 GPU。
  • session.run：执行推理。传入输出名列表和输入字典，返回输出结果。
  • 预处理一致性：ONNX 推理时的预处理必须与测试集完全一致（Resize、Normalize 参数相同），否则结果偏差很大。

易错提示

  • 【忘记 model.eval()】导出前必须 eval，否则 BatchNorm 的运行均值不会被保存进 ONNX。
  • 【输入尺寸不匹配】dummy_input 尺寸必须与训练输入一致（3×224×224），否则导出后推理报错。
  • 【预处理不一致】推理引擎的 transform 必须与测试集完全相同，包括 Normalize 的均值方差。
  • 【numpy 类型】onnxruntime 要求输入为 float32 的 numpy 数组，需 .astype(np.float32)。
  • 【opset 版本】某些新算子需要较高 opset 版本。若导出报"unsupported op"，尝试提高 opset_version。
  • 【灰度图】PIL 读取灰度图为单通道，需 convert('RGB') 转为 3 通道后才能输入预训练模型。
任务 4 5 分

Flask 推理服务接口

题目要求

基于 Flask 编写推理接口，包括人脸检测、识别、表情识别等。

  • 使用 Flask 创建 Web 服务，提供 /predict POST 接口接收图片。
  • 使用 OpenCV（或 InsightFace）的 Haar / RetinaFace 级联分类器进行人脸检测。
  • 对检测到的人脸裁剪后送入 ONNX 推理引擎进行表情识别。
  • 返回 JSON 结果，包含人脸数量、每个人脸的表情类别与置信度、人脸位置框。

思路分析

  1. 服务初始化：创建 Flask app，在启动时加载 ONNX 推理引擎和人脸检测器，避免每次请求重复加载。
  2. 人脸检测：使用 OpenCV 的 Haar 级联分类器（cv2.CascadeClassifier）检测人脸位置，返回 (x,y,w,h) 框。也可用 InsightFace 的 RetinaFace 获得更高精度。
  3. 人脸裁剪与推理：将每个人脸区域裁剪出来，转 PIL Image，送入 ONNX 引擎预测表情。
  4. 结果组装：将人脸位置、表情、置信度组装成 JSON 返回。
  5. 错误处理：对无图片、无人脸、推理失败等情况返回友好错误信息。

代码实现

import os
import io
import base64
import cv2
import numpy as np
from PIL import Image
from flask import Flask, request, jsonify
import onnxruntime as ort
from torchvision import transforms

# ========================================
# 第一部分：初始化 Flask 应用与模型
# ========================================

app = Flask(__name__)

# 表情类别
EXPRESSIONS = ['angry', 'disgust', 'fear', 'happy', 'sad', 'surprise', 'neutral']

# 加载 ONNX 推理引擎（启动时加载一次）
onnx_path = "fer_model.onnx"
ort_session = ort.InferenceSession(onnx_path, providers=['CPUExecutionProvider'])
input_name = ort_session.get_inputs()[0].name
output_name = ort_session.get_outputs()[0].name

# 加载 OpenCV 人脸检测器（Haar 级联分类器）
# xml 文件路径随 opencv 安装，常见位置如下：
cascade_path = cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
face_detector = cv2.CascadeClassifier(cascade_path)

# 推理预处理变换（与测试集一致）
preprocess = transforms.Compose([
    transforms.Resize(224),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])


# ========================================
# 第二部分：核心功能函数
# ========================================

def detect_faces(cv_image):
    """使用 OpenCV 检测人脸，返回人脸框列表 [(x,y,w,h), ...]"""
    gray = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY)  # 转灰度图加速检测
    
    # detectMultiScale 参数说明：
    # scaleFactor: 每次图像缩放比例（1.1 表示每次缩小 10%）
    # minNeighbors: 每个候选框需保留的邻近数（越大越严格）
    # minSize: 最小人脸尺寸
    faces = face_detector.detectMultiScale(
        gray,
        scaleFactor=1.1,
        minNeighbors=5,
        minSize=(48, 48)
    )
    return faces


def predict_expression(face_image):
    """对单张人脸图片进行表情识别
    Args:
        face_image: 人脸区域的 PIL.Image 对象
    Returns:
        dict: 表情类别、置信度、全部概率
    """
    # 灰度图转 RGB（预训练模型需要 3 通道）
    if face_image.mode != 'RGB':
        face_image = face_image.convert('RGB')
    
    # 预处理
    img_tensor = preprocess(face_image)
    img_array = img_tensor.unsqueeze(0).numpy().astype(np.float32)
    
    # ONNX 推理
    outputs = ort_session.run([output_name], {input_name: img_array})
    logits = outputs[0][0]           # 取第一个样本
    
    # softmax 转概率（手动实现，避免依赖 torch）
    exp_scores = np.exp(logits - np.max(logits))
    probs = exp_scores / np.sum(exp_scores)
    
    pred_idx = int(np.argmax(probs))
    
    return {
        'expression': EXPRESSIONS[pred_idx],
        'confidence': round(float(probs[pred_idx]), 4),
        'all_probs': {EXPRESSIONS[i]: round(float(p), 4) for i, p in enumerate(probs)}
    }


# ========================================
# 第三部分：Flask 路由接口
# ========================================

@app.route('/', methods=['GET'])
def index():
    """健康检查接口"""
    return jsonify({'status': 'ok', 'message': '表情识别服务运行中'})


@app.route('/predict', methods=['POST'])
def predict():
    """表情识别接口：接收图片，返回人脸检测与表情识别结果
    接收参数：
        - file: 上传的图片文件（multipart/form-data）
    返回 JSON：
        - success: 是否成功
        - face_count: 检测到的人脸数量
        - faces: 每个人脸的位置与表情结果列表
    """
    # 1. 检查是否上传了文件
    if 'file' not in request.files:
        return jsonify({'success': False, 'error': '未上传图片文件'}), 400
    
    file = request.files['file']
    if file.filename == '':
        return jsonify({'success': False, 'error': '文件名为空'}), 400
    
    try:
        # 2. 读取图片并转为 OpenCV 格式
        img_bytes = file.read()
        pil_image = Image.open(io.BytesIO(img_bytes))
        cv_image = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR)
        
        # 3. 人脸检测
        faces = detect_faces(cv_image)
        
        if len(faces) == 0:
            return jsonify({
                'success': True,
                'face_count': 0,
                'message': '未检测到人脸',
                'faces': []
            })
        
        # 4. 对每个人脸进行表情识别
        results = []
        for i, (x, y, w, h) in enumerate(faces):
            # 裁剪人脸区域
            face_crop = pil_image.crop((x, y, x+w, y+h))
            
            # 表情识别
            expr_result = predict_expression(face_crop)
            
            results.append({
                'face_id': i + 1,
                'bbox': {'x': int(x), 'y': int(y), 'w': int(w), 'h': int(h)},
                'expression': expr_result['expression'],
                'confidence': expr_result['confidence'],
                'all_probs': expr_result['all_probs']
            })
        
        # 5. 返回 JSON 结果
        return jsonify({
            'success': True,
            'face_count': len(faces),
            'faces': results
        })
        
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500


@app.route('/predict_base64', methods=['POST'])
def predict_base64():
    """备选接口：接收 base64 编码的图片"""
    try:
        data = request.get_json()
        img_b64 = data.get('image', '')
        img_bytes = base64.b64decode(img_b64)
        pil_image = Image.open(io.BytesIO(img_bytes))
        cv_image = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR)
        
        faces = detect_faces(cv_image)
        results = []
        for i, (x, y, w, h) in enumerate(faces):
            face_crop = pil_image.crop((x, y, x+w, y+h))
            expr_result = predict_expression(face_crop)
            results.append({
                'face_id': i+1,
                'bbox': {'x': int(x), 'y': int(y), 'w': int(w), 'h': int(h)},
                'expression': expr_result['expression'],
                'confidence': expr_result['confidence']
            })
        return jsonify({'success': True, 'face_count': len(faces), 'faces': results})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)}), 500


# ========================================
# 第四部分：启动服务
# ========================================

if __name__ == '__main__':
    # host=0.0.0.0 允许外部访问，port=5000 为服务端口
    app.run(host='0.0.0.0', port=5000, debug=True)

代码详解

  • 模型全局加载：在模块顶层加载 ONNX Session 和人脸检测器，避免每次请求重复加载，大幅提升响应速度。
  • cv2.CascadeClassifier：OpenCV 内置的 Haar 级联分类器，加载 XML 模型文件。路径通过 cv2.data.haarcascades 获取。
  • detectMultiScale：在图像金字塔上滑动检测窗口。scaleFactor 控制缩放步长，minNeighbors 控制误检过滤严格度。
  • cvtColor：人脸检测需灰度图（速度更快），表情识别需 RGB 图。注意 OpenCV 默认 BGR，PIL 默认 RGB，需正确转换。
  • request.files：Flask 接收 multipart/form-data 上传的文件，通过键名 'file' 获取。
  • jsonify：将 Python 字典转为 JSON 响应，自动设置 Content-Type 为 application/json。
  • 手动 softmax：服务端为减少依赖，可用 numpy 手动实现 softmax，避免引入 torch。注意先减最大值防止数值溢出。
  • 错误处理：try-except 捕获异常并返回 500 状态码，保证服务不会因单次请求崩溃。
  • base64 接口：提供 /predict_base64 备选接口，前端可通过 JSON 传 base64 图片，适用于无法使用 multipart 上传的场景。

易错提示

  • 【Haar 文件路径】不同 OpenCV 版本路径可能不同，可用 print(cv2.data.haarcascades) 查看实际路径。若找不到，需单独下载 XML 文件。
  • 【BGR vs RGB】OpenCV 读取为 BGR，PIL 为 RGB，混用会导致颜色错乱、影响检测与识别。务必明确转换。
  • 【CORS 跨域】前端和后端不同源时浏览器会拦截请求，需安装 flask-cors 并添加 CORS(app)。
  • 【文件大小限制】Flask 默认无限制，大图上传可能超时。可在配置中设置 app.config['MAX_CONTENT_LENGTH']。
  • 【debug 模式】生产环境应关闭 debug=True，否则会暴露调试信息。debug 模式下代码修改会自动重启。
  • 【线程安全】ONNX Runtime Session 默认线程安全，但若使用 PyTorch 模型需注意多线程并发问题。
任务 5 5 分

前端界面与接口对接

题目要求

具有前端界面原型，实现前端功能与推理后端进行接口对接。

  • 开发一个 HTML 前端页面，包含图片上传、预览、识别按钮、结果展示区域。
  • 使用 JavaScript（fetch / axios）调用后端 /predict 接口。
  • 展示识别结果：人脸数量、每个人脸的表情类别、置信度、人脸框可视化。
  • 界面美观，交互流畅，有加载状态提示。

思路分析

  1. 页面布局：采用左右分栏布局，左侧上传与预览，右侧结果展示。
  2. 图片上传：使用 <input type="file"> 配合 FileReader 实现本地预览。
  3. 接口调用：用 FormData 封装图片，通过 fetch POST 到 /predict，处理 JSON 响应。
  4. 结果可视化：在 Canvas 上绘制人脸框和表情标签，用列表展示详细概率。
  5. 状态管理：上传中、识别中、完成、错误四种状态，用 loading 动画提示用户。

代码实现

<!-- index.html：前端界面完整代码 -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>人脸表情识别系统</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: 'Microsoft YaHei', sans-serif; background: #f0f2f5; }
        .container { max-width: 1200px; margin: 0 auto; padding: 20px; }
        .header { text-align: center; padding: 20px; background: #fff; border-radius: 8px; margin-bottom: 20px; }
        .main { display: flex; gap: 20px; }
        .panel { flex: 1; background: #fff; border-radius: 8px; padding: 20px; }
        .upload-area { border: 2px dashed #ccc; border-radius: 8px; padding: 40px; text-align: center; cursor: pointer; }
        .upload-area:hover { border-color: #4CAF50; background: #f9fff9; }
        .btn { background: #4CAF50; color: #fff; border: none; padding: 10px 30px; border-radius: 4px; cursor: pointer; font-size: 16px; }
        .btn:hover { background: #45a049; }
        .btn:disabled { background: #ccc; cursor: not-allowed; }
        #previewCanvas { max-width: 100%; margin-top: 15px; border-radius: 4px; }
        .result-item { background: #f8f9fa; padding: 15px; margin-bottom: 10px; border-radius: 4px; border-left: 4px solid #4CAF50; }
        .expr-badge { display: inline-block; padding: 4px 12px; border-radius: 12px; color: #fff; font-size: 14px; margin-right: 8px; }
        .progress-bar { background: #e0e0e0; border-radius: 4px; overflow: hidden; height: 8px; margin-top: 4px; }
        .progress-fill { height: 100%; background: #4CAF50; transition: width 0.3s; }
        .loading { text-align: center; padding: 20px; color: #666; }
        .spinner { border: 3px solid #f3f3f3; border-top: 3px solid #4CAF50; border-radius: 50%; width: 30px; height: 30px; animation: spin 1s linear infinite; margin: 0 auto 10px; }
        @keyframes spin { 0% { transform: rotate(0deg); } 100% { transform: rotate(360deg); } }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>人脸表情识别系统</h1>
            <p>上传图片，自动检测人脸并识别七种表情</p>
        </div>
        <div class="main">
            <div class="panel">
                <h3>图片上传</h3>
                <div class="upload-area" id="uploadArea" onclick="document.getElementById('fileInput').click()">
                    <p>点击或拖拽图片到此处上传</p>
                    <p style="color:#999;font-size:13px;margin-top:8px">支持 JPG/PNG 格式</p>
                </div>
                <input type="file" id="fileInput" accept="image/*" style="display:none">
                <canvas id="previewCanvas" style="display:none"></canvas>
                <div style="margin-top:15px;text-align:center">
                    <button class="btn" id="predictBtn" disabled>开始识别</button>
                </div>
                <div id="statusArea"></div>
            </div>
            <div class="panel">
                <h3>识别结果</h3>
                <div id="resultArea">
                    <p style="color:#999;text-align:center;padding:40px">请上传图片后点击识别</p>
                </div>
            </div>
        </div>
    </div>

    <script>
        // 表情颜色映射
        const exprColors = {
            angry: '#e74c3c', disgust: '#8e44ad', fear: '#34495e',
            happy: '#f39c12', sad: '#3498db', surprise: '#2ecc71', neutral: '#95a5a6'
        };
        const exprCN = {
            angry: '愤怒', disgust: '厌恶', fear: '恐惧',
            happy: '开心', sad: '悲伤', surprise: '惊讶', neutral: '中性'
        };

        let currentImage = null;   // 存储当前图片对象

        // 文件选择事件
        document.getElementById('fileInput').addEventListener('change', function(e) {
            const file = e.target.files[0];
            if (!file) return;
            
            const reader = new FileReader();
            reader.onload = function(event) {
                const img = new Image();
                img.onload = function() {
                    currentImage = img;
                    drawPreview(img, []);
                    document.getElementById('predictBtn').disabled = false;
                };
                img.src = event.target.result;
            };
            reader.readAsDataURL(file);
        });

        // 在 Canvas 上绘制图片和人脸框
        function drawPreview(img, faces) {
            const canvas = document.getElementById('previewCanvas');
            const ctx = canvas.getContext('2d');
            
            // 按比例缩放，最大宽度 500
            const maxW = 500;
            const scale = Math.min(maxW / img.width, 1);
            canvas.width = img.width * scale;
            canvas.height = img.height * scale;
            canvas.style.display = 'block';
            
            ctx.drawImage(img, 0, 0, canvas.width, canvas.height);
            
            // 绘制人脸框
            faces.forEach(function(face) {
                const b = face.bbox;
                const x = b.x * scale;
                const y = b.y * scale;
                const w = b.w * scale;
                const h = b.h * scale;
                
                // 绘制矩形框
                ctx.strokeStyle = exprColors[face.expression] || '#4CAF50';
                ctx.lineWidth = 3;
                ctx.strokeRect(x, y, w, h);
                
                // 绘制标签背景
                const label = exprCN[face.expression] + ' ' + (face.confidence * 100).toFixed(1) + '%';
                ctx.fillStyle = exprColors[face.expression] || '#4CAF50';
                ctx.fillRect(x, y - 24, ctx.measureText(label).width + 16, 24);
                ctx.fillStyle = '#fff';
                ctx.font = '14px Microsoft YaHei';
                ctx.fillText(label, x + 8, y - 8);
            });
        }

        // 识别按钮点击事件
        document.getElementById('predictBtn').addEventListener('click', async function() {
            const fileInput = document.getElementById('fileInput');
            if (!fileInput.files[0]) return;
            
            const btn = document.getElementById('predictBtn');
            const statusArea = document.getElementById('statusArea');
            const resultArea = document.getElementById('resultArea');
            
            // 显示加载状态
            btn.disabled = true;
            btn.textContent = '识别中...';
            statusArea.innerHTML = '<div class="loading"><div class="spinner"></div>正在识别，请稍候...</div>';
            
            try {
                // 构造 FormData 上传文件
                const formData = new FormData();
                formData.append('file', fileInput.files[0]);
                
                // 调用后端 /predict 接口
                const response = await fetch('http://localhost:5000/predict', {
                    method: 'POST',
                    body: formData
                });
                
                const data = await response.json();
                statusArea.innerHTML = '';
                
                if (data.success && data.face_count > 0) {
                    // 在预览图上绘制人脸框
                    drawPreview(currentImage, data.faces);
                    
                    // 渲染结果列表
                    let html = '<p>共检测到 ' + data.face_count + ' 张人脸</p>';
                    data.faces.forEach(function(face) {
                        const color = exprColors[face.expression];
                        const cn = exprCN[face.expression];
                        html += '<div class="result-item">';
                        html += '<div><span class="expr-badge" style="background:' + color + '">' + cn + '</span>';
                        html += '置信度: ' + (face.confidence * 100).toFixed(2) + '%</div>';
                        html += '<div style="margin-top:8px;font-size:13px;color:#666">位置:(' + face.bbox.x + ',' + face.bbox.y + ',' + face.bbox.w + ',' + face.bbox.h + ')</div>';
                        // 绘制各表情概率条
                        if (face.all_probs) {
                            html += '<div style="margin-top:10px">';
                            Object.keys(face.all_probs).forEach(function(key) {
                                const prob = face.all_probs[key];
                                html += '<div style="margin-bottom:4px"><span style="display:inline-block;width:60px;font-size:12px">' + exprCN[key] + '</span>';
                                html += '<div class="progress-bar" style="display:inline-block;width:150px;vertical-align:middle"><div class="progress-fill" style="width:' + (prob*100) + '%;background:' + exprColors[key] + '"></div></div>';
                                html += '<span style="font-size:12px;margin-left:5px">' + (prob*100).toFixed(1) + '%</span></div>';
                            });
                            html += '</div>';
                        }
                        html += '</div>';
                    });
                    resultArea.innerHTML = html;
                } else {
                    resultArea.innerHTML = '<p style="text-align:center;color:#e74c3c;padding:30px">未检测到人脸，请更换图片重试</p>';
                }
            } catch (err) {
                statusArea.innerHTML = '<p style="color:#e74c3c;text-align:center">请求失败: ' + err.message + '</p>';
            } finally {
                btn.disabled = false;
                btn.textContent = '开始识别';
            }
        });
    </script>
</body>
</html>

代码详解

  • 布局结构：左右分栏（上传区 / 结果区），使用 flex 布局。header 展示标题，main 包含两个 panel。
  • FileReader.readAsDataURL：将图片文件读取为 base64 Data URL，可直接赋给 Image.src 实现本地预览，无需上传服务器。
  • Canvas 绘制：使用 2D Canvas API 绘制图片和人脸框。drawImage 画图片，strokeRect 画框，fillText 写标签。按缩放比例绘制保证框位置准确。
  • FormData：构造 multipart/form-data 请求体，append('file', file) 添加文件字段，与后端 request.files['file'] 对应。
  • fetch async/await：用 async/await 语法发起异步请求，代码更清晰。fetch 返回 Response 对象，需调用 .json() 解析。
  • 状态管理：识别中禁用按钮、显示 spinner 动画；finally 块确保无论成功失败都恢复按钮状态。
  • 结果可视化：每个表情用不同颜色标签展示，概率用进度条直观呈现，人脸框直接画在图片上。
  • 跨域处理：若前端与后端不同源，后端需配置 CORS（from flask_cors import CORS; CORS(app)）。

易错提示

  • 【接口地址】fetch 的 URL 必须与 Flask 服务地址端口一致（如 http://localhost:5000/predict）。部署时改为实际服务器地址。
  • 【CORS 报错】浏览器控制台出现 "CORS policy" 错误时，需在后端安装 flask-cors 并启用，或在 Flask 中手动添加 Access-Control-Allow-Origin 头。
  • 【FormData 字段名】formData.append('file', ...) 的键名必须与后端 request.files['file'] 一致。
  • 【Canvas 缩放】绘制人脸框时需乘以缩放比例 scale，否则框的位置和大小会与图片不匹配。
  • 【异步错误处理】fetch 不会因 HTTP 4xx/5xx 自动抛错，需检查 response.ok 或 data.success 字段。
  • 【图片格式】accept="image/*" 限制只能选图片，但后端也应校验文件类型防止恶意上传。

1.5 项目总结与拓展

项目总结

本项目围绕"情感计算之人脸表情识别"这一主题，完整实践了从数据处理到部署应用的 AI 全链路开发流程。通过五个任务的实施，掌握了以下核心能力：

任务核心技能关键知识点
任务1 数据加载数据工程ImageFolder、transforms 数据增强、DataLoader
任务2 模型训练模型训练迁移学习、损失函数、优化器、sklearn 评估指标
任务3 ONNX导出模型部署torch.onnx.export、ONNX Runtime、推理引擎封装
任务4 Flask服务后端开发Flask路由、OpenCV人脸检测、RESTful接口设计
任务5 前端界面前端开发HTML/CSS/JS、Canvas绘制、fetch接口对接

项目的核心价值在于打通了"数据→模型→部署→应用"的完整工程闭环，这正是企业级 AI 应用开发的标准范式。训练出好模型只是起点，能够高效部署并提供服务才是落地的关键。

常见问题与解决方案

  • 模型准确率不高：尝试更强的数据增强（MixUp、CutMix）；使用类别加权损失解决类别不均衡；尝试不同预训练模型对比；增加训练 epoch 并使用余弦退火学习率。
  • ONNX 推理结果与 PyTorch 不一致：检查是否遗漏 eval 模式；确认预处理变换完全一致；检查动态轴是否设置正确；尝试提高 opset_version。
  • 人脸检测漏检：Haar 级联对侧脸、遮挡、小脸检测效果差，可换用 InsightFace 的 RetinaFace 或 MTCNN；调整 minNeighbors 和 minSize 参数。
  • Flask 服务响应慢：大图先 Resize 再检测；使用多进程（gunicorn）并发；开启 ONNX Runtime 的多线程推理。

拓展方向

  1. 模型轻量化：将模型量化为 INT8（onnxruntime 量化工具），进一步减小体积、加速推理，适合边缘设备部署。
  2. 实时视频识别：将 Flask 接口扩展为 WebSocket 或流式接口，接入摄像头实时检测表情，可用于课堂注意力分析、驾驶员疲劳监测。
  3. 多模态融合：结合语音情感识别、文本情感分析，构建多模态情感计算系统，提升识别鲁棒性。
  4. 数据增强进阶：引入 AutoAugment、RandAugment 等自动增强策略，或使用 GAN 生成少数类样本解决类别不均衡。
  5. 容器化部署：使用 Docker 将 Flask 服务打包为镜像，配合 Docker Compose 一键部署，实现环境隔离与快速复制。
  6. 模型对比实验：系统对比 VGG-16、ResNet-50、MobileNetV3-Small、MobileViT-XS 在精度、速度、模型大小上的表现，形成实验报告，深入理解模型选型策略。

学习建议

  • 先跑通完整流程，再逐项优化。新手容易在单点（如数据增强参数）过度纠结，应先确保全链路可用。
  • 多查阅官方文档：PyTorch 文档、torchvision models 文档、ONNX Runtime 文档、Flask 文档是最权威的参考。
  • 善用断点调试：在训练循环中打印 loss、accuracy 变化曲线，用 TensorBoard 或 matplotlib 可视化训练过程，有助于发现过拟合、学习率不当等问题。
  • 注重工程规范：代码加注释、变量命名清晰、配置参数集中管理、异常处理完善，这些是竞赛评分和实际开发都看重的素质。
模块2 · 25%

情感计算之人体行为识别

基于 YOLO8n-POSE + ByteTrack + MMAction2，完成室内外人体行为识别、FastAPI推理服务与前端界面开发

2.1 项目导读与学习目标

项目背景

人体行为识别是情感计算与计算机视觉领域的重要研究方向，广泛应用于智慧教育、智能安防、医疗监护等场景。在课堂教学中，通过分析学生的姿态与行为，教师可以实时了解学生的学习状态，从而进行针对性教学调整。

本项目构建了一套完整的人体行为分析系统，支持室内（classroom）与室外两种工作模式：

  • 室内模式：基于 YOLO8n-POSE 实现人体姿态估计与 17 关键点检测，结合 ByteTrack 多目标跟踪，通过关键点几何规则判断六类课堂行为（sit_listen 听讲、raise_hand 举手、write 写字、bow_head 低头、stand 站立、unknown 其他）。
  • 室外模式：集成 MMAction2 TSN 模型（Kinetics-400 预训练），完成开放场景下的视频行为识别。

系统基于 FastAPI 封装 RESTful API，并开发前端界面进行接口对接，形成端到端的完整解决方案。

学习目标

目标类型具体内容
知识目标 理解姿态估计、关键点检测、多目标跟踪、时序行为识别的基本原理；掌握 YOLO-POSE、ByteTrack、MMAction2 的核心概念
技能目标 能够使用 ultralytics 库加载 YOLO8n-POSE 模型并提取关键点；能够编写几何规则算法实现行为分类；能够集成 ByteTrack 实现多目标跟踪；能够使用 MMAction2 进行视频行为识别；能够使用 FastAPI 编写推理接口并开发前端界面
素养目标 培养工程化思维，理解从模型推理到 API 服务到前端展示的完整开发流程；提升团队协作与项目实践能力

系统架构总览

前端层
HTML + CSS + JavaScript 界面原型
↓ HTTP / RESTful API
服务层
FastAPI 推理服务（人体检测 / 关键点 / 室内行为 / 室外行为）
↓ 模型调用
算法层
YOLO8n-POSE
姿态估计
ByteTrack
多目标跟踪
几何规则
行为分类
MMAction2 TSN
视频行为识别

2.2 知识准备

01

YOLO8n-POSE 姿态估计

姿态估计（Pose Estimation）是指从图像或视频中检测人体并定位其关键解剖点（如关节、面部特征点）的技术。YOLO8n-POSE 是 Ultralytics 公司推出的 YOLOv8 系列中的姿态估计模型，它将目标检测与关键点检测融合在一个端到端网络中。

核心特点：

  • 单阶段检测：同时完成人体边界框检测和关键点回归，速度快
  • 17 个关键点：遵循 COCO 格式标准，覆盖头、上肢、下肢主要关节
  • 多种模型规格：n（nano）最轻量，适合实时推理场景
  • 输出格式：每个人体返回 17 个关键点的 (x, y, confidence) 三元组

与目标检测的区别：普通 YOLO 只输出边界框和类别，而 YOLO-POSE 在此基础上额外输出每个人体的精细关键点坐标，为行为分析提供了更丰富的几何信息。

02

COCO 17 关键点体系

COCO（Common Objects in Context）数据集定义了人体姿态估计的标准 17 关键点格式，是业界最广泛使用的关键点标注规范。每个关键点包含横坐标 x、纵坐标 y 和置信度 confidence（0~1）。

索引关键点名称英文名所属部位
0鼻子nose头部
1左眼left_eye头部
2右眼right_eye头部
3左耳left_ear头部
4右耳right_ear头部
5左肩left_shoulder上肢
6右肩right_shoulder上肢
7左肘left_elbow上肢
8右肘right_elbow上肢
9左腕left_wrist上肢
10右腕right_wrist上肢
11左髋left_hip下肢
12右髋right_hip下肢
13左膝left_knee下肢
14右膝right_knee下肢
15左踝left_ankle下肢
16右踝right_ankle下肢

坐标系统说明：在图像坐标系中，原点 (0, 0) 位于左上角，x 轴向右递增，y 轴向下递增。因此，y 值越小的点在图像中位置越靠上。这一特性在后续行为判断规则中非常重要。

03

ByteTrack 多目标跟踪

多目标跟踪（Multiple Object Tracking, MOT）是指在视频序列中持续追踪多个目标，为每个目标分配一个唯一 ID，使其在不同帧之间保持关联。

ByteTrack 核心思想：传统跟踪算法只使用高置信度检测框进行关联，导致低分检测框被丢弃。ByteTrack 创新性地将检测框分为高分和低分两组：

  • 第一步：用高分检测框与已有轨迹进行匹配（基于 IoU 或外观特征）
  • 第二步：将第一步中未匹配的低分检测框与剩余轨迹进行二次匹配
  • 这样能够挽回被遮挡、模糊等原因导致低分的真实目标，减少 ID 切换

在项目中的作用：课堂视频中有多名学生，ByteTrack 为每个学生分配唯一跟踪 ID，使系统能够持续分析每个学生的行为，而非逐帧独立判断。

Ultralytics 集成方式：ultralytics 库内置了 ByteTrack，通过 model.track(tracker='bytetrack.yaml') 即可启用，使用非常便捷。

04

MMAction2 与 TSN 模型

MMAction2 是 OpenMMLab 开源的动作识别工具箱，支持多种视频理解模型，包括 TSN、TSM、SlowFast、TimeSformer 等。

TSN（Temporal Segment Networks）是经典的视频行为识别模型：

  • 将视频均匀分成若干片段（segments），从每个片段中随机采样一帧
  • 对采样帧使用 2D CNN（如 ResNet）提取特征
  • 通过共识函数（consensus function，如平均池化）聚合各片段特征
  • 最终输出视频级别的行为分类结果

Kinetics-400 数据集：包含 400 类人体行为类别，约 30 万个视频片段，是视频行为识别领域最重要的基准数据集之一。使用 Kinetics-400 预训练的 TSN 模型可以直接识别跑步、跳跃、骑自行车等常见行为。

室内外模式区别：室内模式基于关键点几何规则，适合坐姿、举手等特定课堂行为；室外模式基于 TSN 深度学习模型，适合开放场景下的复杂行为识别，但仅支持视频输入（需要时序信息）。

05

课堂行为识别规则

本项目通过分析 17 关键点之间的几何关系，定义了六类课堂行为的判断规则。核心思路是利用关键点的相对位置（坐标差值、角度、比例）来区分不同姿态。

行为类别英文名判断依据
举手 raise_hand 手腕 y 坐标明显小于肩膀 y 坐标（手腕高于肩膀）
低头 bow_head 鼻子 y 坐标大于肩膀中点 y 坐标（头部低于肩线）
写字 write 坐姿状态下，手腕位于肩线以下且靠近身体中线
听讲 sit_listen 坐姿状态（腿长/躯干长比值较小），未满足举手、写字、低头条件
站立 stand 站姿状态（腿长/躯干长比值较大，身体直立）
其他 unknown 关键点置信度过低或无法匹配以上任何规则

关键概念 —— 坐标与位置关系：在图像坐标系中 y 轴向下递增，因此"高于"意味着 y 值更小，"低于"意味着 y 值更大。例如判断举手时，需要检查手腕的 y 坐标是否小于肩膀的 y 坐标。

06

FastAPI Web 框架

FastAPI 是一个现代、高性能的 Python Web 框架，专为构建 RESTful API 设计，具有以下优势：

  • 高性能：基于 Starlette 和 Pydantic，性能媲美 NodeJS 和 Go
  • 类型提示：利用 Python 类型注解自动进行参数验证和文档生成
  • 自动文档：内置 Swagger UI（/docs）和 ReDoc（/redoc）交互式 API 文档
  • 异步支持：原生支持 async/await，适合处理并发请求

在本项目中的应用：使用 FastAPI 封装模型推理逻辑为 HTTP 接口，前端通过发送 HTTP 请求（携带图片或视频文件）获取推理结果，实现前后端分离架构。

核心概念：

  • @app.post("/path")：定义 POST 路由
  • UploadFile：处理文件上传
  • Form：处理表单字段
  • CORSMiddleware：处理跨域请求（前端与后端不同源时必需）

2.3 环境与数据准备

开发环境配置

本项目推荐使用 Python 3.9+ 环境，建议通过 Conda 创建独立虚拟环境以避免依赖冲突。

# 1. 创建并激活 Conda 虚拟环境
conda create -n behavior_recognition python=3.10 -y
conda activate behavior_recognition

# 2. 安装 PyTorch（根据 CUDA 版本选择，以下为 CUDA 11.8 示例）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118

# 3. 安装 ultralytics（包含 YOLO8n-POSE 与 ByteTrack）
pip install ultralytics

# 4. 安装 MMAction2（室外行为识别）
pip install -U openmim
mim install mmengine
mim install mmcv
pip install mmaction2

# 5. 安装 FastAPI 及相关依赖
pip install fastapi uvicorn python-multipart

# 6. 安装图像/视频处理库
pip install opencv-python numpy Pillow

项目目录结构

# 推荐的项目目录组织
behavior_recognition/
├── models/                  # 模型权重文件
│   ├── yolov8n-pose.pt          # YOLO8n-POSE 预训练权重
│   └── tsn_k400.pth             # TSN Kinetics-400 预训练权重
├── configs/                 # MMAction2 配置文件
│   └── tsn_k400_config.py
├── data/                    # 测试数据
│   ├── images/                  # 室内测试图片
│   └── videos/                  # 室内外测试视频
├── labels/                  # 标签文件
│   └── kinetics400_labels.txt   # Kinetics-400 类别标签
├── app/                     # 后端服务代码
│   ├── behavior_rules.py        # 行为识别规则算法
│   ├── pose_detector.py         # 姿态检测模块
│   ├── tracker.py               # 多目标跟踪模块
│   ├── action_recognizer.py     # MMAction2 行为识别模块
│   └── main.py                  # FastAPI 主程序
├── frontend/                # 前端界面
│   └── index.html
└── requirements.txt

数据准备

1. YOLO8n-POSE 模型下载：首次运行 YOLO('yolov8n-pose.pt') 时，ultralytics 库会自动从官方服务器下载预训练权重（约 7MB），也可手动下载放置于 models 目录。

2. TSN 模型权重：从 OpenMMLab 模型库下载 TSN Kinetics-400 预训练权重文件 tsn_k400.pth，放置于 models 目录。

3. Kinetics-400 标签文件：创建 kinetics400_labels.txt，每行一个类别名称，共 400 行，与模型输出索引一一对应。

4. 测试数据准备：

  • 室内图片：课堂场景照片，包含坐姿、举手、写字等多种行为的学生
  • 室内视频：课堂录像片段，用于测试多目标跟踪与行为持续分析
  • 室外视频：操场、街道等开放场景视频，包含跑步、骑车等行为

验证环境安装

# 验证 ultralytics 安装
from ultralytics import YOLO
model = YOLO('yolov8n-pose.pt')
print("YOLO8n-POSE 加载成功！")

# 验证 MMAction2 安装
import mmaction
print(f"MMAction2 版本: {mmaction.__version__}")

# 验证 FastAPI 安装
from fastapi import FastAPI
print("FastAPI 安装成功！")

2.4 项目实施

任务 1 5分

YOLO8n-POSE 关键点检测与室内行为识别

题目要求

使用 ultralytics 库加载 YOLO8n-POSE 预训练模型，推理提取前 17 个关键点，编写课堂姿态规则算法，实现室内 5 种行为识别（sit_listen 听讲、raise_hand 举手、write 写字、bow_head 低头、stand 站立）。

思路分析

本任务分为三个步骤：加载模型 → 推理提取关键点 → 规则判断行为。

  1. 使用 YOLO('yolov8n-pose.pt') 加载预训练姿态估计模型
  2. 对输入图片执行推理，获取每个人体的 17 个关键点坐标 (x, y, confidence)
  3. 编写 classify_behavior() 函数，基于关键点几何关系判断行为类别：
    • 判断举手：手腕是否高于肩膀（y 坐标更小）
    • 判断低头：鼻子是否低于肩线
    • 判断坐/站：通过腿长与躯干长的比例区分
    • 判断写字：坐姿下手腕位置是否在桌面前方区域

代码实现

# ============================================
# 文件名: behavior_rules.py
# 功能: YOLO8n-POSE 关键点检测 + 课堂行为识别规则
# ============================================

import numpy as np
from ultralytics import YOLO

# -----------------------------------------------
# 第一步: 定义 17 关键点索引常量（便于代码阅读）
# -----------------------------------------------
NOSE = 0
LEFT_EYE, RIGHT_EYE = 1, 2
LEFT_EAR, RIGHT_EAR = 3, 4
LEFT_SHOULDER, RIGHT_SHOULDER = 5, 6
LEFT_ELBOW, RIGHT_ELBOW = 7, 8
LEFT_WRIST, RIGHT_WRIST = 9, 10
LEFT_HIP, RIGHT_HIP = 11, 12
LEFT_KNEE, RIGHT_KNEE = 13, 14
LEFT_ANKLE, RIGHT_ANKLE = 15, 16

# 六种行为类别名称
BEHAVIOR_CLASSES = ['sit_listen', 'raise_hand', 'write',
                    'bow_head', 'stand', 'unknown']


# -----------------------------------------------
# 第二步: 编写行为分类规则函数
# 输入: keypoints - 形状为 (17, 3) 的数组 [x, y, confidence]
# 输出: 行为类别字符串
# -----------------------------------------------
def classify_behavior(keypoints, conf_threshold=0.5):
    """根据 17 个关键点几何关系判断课堂行为"""

    # 检查关键点数据有效性
    if keypoints is None or len(keypoints) < 17:
        return 'unknown'

    # 分离坐标和置信度
    kp = keypoints[:, :2]      # 取 x, y 坐标
    conf = keypoints[:, 2]      # 取置信度

    # 统计有效关键点数量，过少则无法判断
    valid = conf > conf_threshold
    if valid.sum() < 8:
        return 'unknown'

    # --- 计算身体关键参考点 ---
    # 肩膀中点（x, y）
    shoulder_mid_x = (kp[LEFT_SHOULDER][0] + kp[RIGHT_SHOULDER][0]) / 2
    shoulder_mid_y = (kp[LEFT_SHOULDER][1] + kp[RIGHT_SHOULDER][1]) / 2

    # 髋部中点 y 坐标
    hip_mid_y = (kp[LEFT_HIP][1] + kp[RIGHT_HIP][1]) / 2

    # 膝盖中点 y 坐标
    knee_mid_y = (kp[LEFT_KNEE][1] + kp[RIGHT_KNEE][1]) / 2

    # 计算躯干长度（肩膀到髋部的距离）
    torso_length = abs(shoulder_mid_y - hip_mid_y)
    if torso_length < 1:
        return 'unknown'

    # --- 规则1: 判断举手 ---
    # 手腕 y 坐标明显小于肩膀 y 坐标，说明手腕在肩膀上方
    if valid[LEFT_WRIST] and kp[LEFT_WRIST][1] < shoulder_mid_y - 20:
        return 'raise_hand'
    if valid[RIGHT_WRIST] and kp[RIGHT_WRIST][1] < shoulder_mid_y - 20:
        return 'raise_hand'

    # --- 规则2: 判断低头 ---
    # 鼻子 y 坐标大于肩膀中点 y 坐标，说明头部低于肩线
    if valid[NOSE] and kp[NOSE][1] > shoulder_mid_y + 10:
        return 'bow_head'

    # --- 规则3: 区分坐姿与站姿 ---
    # 通过腿长（髋到膝）与躯干长（肩到髋）的比值判断
    leg_length = abs(hip_mid_y - knee_mid_y)
    ratio = leg_length / torso_length if torso_length > 0 else 0

    if ratio < 0.8:
        # 比值较小，说明腿部弯曲或被遮挡，判定为坐姿

        # --- 规则4: 坐姿下判断写字 ---
        # 手腕在肩线以下且靠近身体中线水平范围
        if valid[LEFT_WRIST]:
            wrist_below = kp[LEFT_WRIST][1] > shoulder_mid_y
            wrist_near = abs(kp[LEFT_WRIST][0] - shoulder_mid_x) < 100
            if wrist_below and wrist_near:
                return 'write'

        if valid[RIGHT_WRIST]:
            wrist_below = kp[RIGHT_WRIST][1] > shoulder_mid_y
            wrist_near = abs(kp[RIGHT_WRIST][0] - shoulder_mid_x) < 100
            if wrist_below and wrist_near:
                return 'write'

        # 未满足写字条件，判定为听讲
        return 'sit_listen'

    else:
        # 比值较大，腿部伸展，判定为站姿
        return 'stand'


# -----------------------------------------------
# 第三步: 加载模型并执行推理
# -----------------------------------------------
def detect_pose_and_classify(image_path):
    """对单张图片执行姿态检测与行为分类"""

    # 加载 YOLO8n-POSE 预训练模型（首次运行自动下载权重）
    model = YOLO('yolov8n-pose.pt')

    # 执行推理，results 是结果列表
    results = model(image_path, verbose=False)

    # 存储所有人体行为结果
    all_behaviors = []

    for result in results:
        # 检查是否检测到人体
        if result.keypoints is None:
            continue

        # 提取关键点数据，形状: (人数, 17, 3)
        # data 属性包含 [x, y, confidence] 三列
        keypoints_data = result.keypoints.data.cpu().numpy()

        # 提取边界框，形状: (人数, 4)，格式 [x1, y1, x2, y2]
        boxes = result.boxes.xyxy.cpu().numpy()

        for i, kps in enumerate(keypoints_data):
            # 对每个人体调用行为分类函数
            behavior = classify_behavior(kps)

            # 记录结果
            all_behaviors.append({
                'person_id': i,
                'behavior': behavior,
                'bbox': boxes[i].tolist() if len(boxes) > i else None,
                'keypoints': kps.tolist()
            })

    return all_behaviors


# -----------------------------------------------
# 主程序入口: 测试单张图片
# -----------------------------------------------
if __name__ == '__main__':
    # 指定测试图片路径
    image_path = 'data/images/classroom_01.jpg'

    # 执行检测与分类
    results = detect_pose_and_classify(image_path)

    # 打印每个人的行为识别结果
    for person in results:
        print(f"人员 {person['person_id']}: 行为={person['behavior']}, "
              f"位置={person['bbox']}")

代码详解

1. 关键点索引常量定义：使用语义化常量（如 NOSE=0、LEFT_WRIST=9）替代魔法数字，使规则代码可读性大幅提升。这是良好的编程习惯。

2. classify_behavior() 函数核心逻辑：

  • 首先检查关键点数据的有效性，过滤置信度过低的关键点
  • 计算肩膀中点、髋部中点、膝盖中点等参考点
  • 按优先级依次判断：举手（最高优先级）→ 低头 → 坐姿下的写字/听讲 → 站姿
  • 优先级设计的原因：举手时也可能同时低头，但举手是最显著的行为特征，应优先判断

3. 坐标系理解：图像中 y 轴向下递增。手腕 y 值小于肩膀 y 值意味着手腕在肩膀上方（举手）；鼻子 y 值大于肩膀 y 值意味着头低于肩线（低头）。

4. 坐姿/站姿判断：使用腿长（髋到膝）与躯干长（肩到髋）的比值。站立时腿部伸展，比值接近 1 或更大；坐姿时膝盖弯曲靠近髋部，比值较小。

5. detect_pose_and_classify() 函数：封装完整的推理流程，加载模型 → 推理 → 提取关键点 → 逐人分类 → 返回结构化结果。返回的字典包含人员编号、行为类别、边界框和关键点坐标。

易错提示

  • 关键点数据格式：result.keypoints.data 的形状是 (N, 17, 3)，不是 (N, 17, 2)。第三列是置信度，切勿遗漏。
  • 置信度过滤：未检测到的关键点置信度为 0，使用前必须检查 conf > threshold，否则会用到无效坐标导致误判。
  • 坐标系方向：图像 y 轴向下递增，初学者容易搞反"高于"和"低于"的含义。y 值小 = 位置高。
  • 阈值调参：代码中的 20、10、100、0.8 等阈值需根据实际场景调整，不同摄像头角度和分辨率可能需要不同参数。
  • GPU/CPU 转换：.cpu().numpy() 是将 GPU 张量转为 CPU numpy 数组，若使用 CPU 推理可省略 .cpu()。
任务 2 5分

ByteTrack 多目标跟踪与室内推理流水线

题目要求

集成 ByteTrack 多目标跟踪，实现人体目标跟踪，完成课堂场景推理服务完整流水线（检测 → 跟踪 → 分类）功能。

思路分析

本任务在任务 1 基础上增加多目标跟踪能力，形成完整的视频推理流水线。

  1. 使用 model.track() 方法替代 model()，启用 ByteTrack 跟踪器
  2. 设置 persist=True，使跟踪 ID 在视频帧之间保持连续
  3. 对每一帧的每个人体（通过跟踪 ID 区分），调用任务 1 的 classify_behavior() 进行行为分类
  4. 记录每个跟踪 ID 的行为序列，最终统计每个学生的主要行为
  5. 绘制可视化结果（边界框 + ID + 行为标签）输出标注视频

代码实现

# ============================================
# 文件名: tracker.py
# 功能: ByteTrack 多目标跟踪 + 室内推理流水线
# 流水线: 检测 → 跟踪 → 分类
# ============================================

import cv2
import numpy as np
from ultralytics import YOLO
from collections import Counter, defaultdict
# 导入任务1编写的行为分类函数
from behavior_rules import classify_behavior

# 行为类别对应的中文标签（用于可视化显示）
BEHAVIOR_LABELS = {
    'sit_listen': '听讲',
    'raise_hand': '举手',
    'write': '写字',
    'bow_head': '低头',
    'stand': '站立',
    'unknown': '其他'
}

# 不同行为的边界框颜色 (B, G, R) 格式
BEHAVIOR_COLORS = {
    'sit_listen': (0, 255, 0),    # 绿色
    'raise_hand': (0, 0, 255),    # 红色
    'write': (255, 0, 0),       # 蓝色
    'bow_head': (0, 165, 255),   # 橙色
    'stand': (255, 255, 0),   # 青色
    'unknown': (128, 128, 128)  # 灰色
}


def run_indoor_pipeline(video_path, output_path=None, conf=0.5):
    """
    室内推理完整流水线: 检测 → 跟踪 → 分类
    参数:
        video_path: 输入视频路径
        output_path: 输出视频路径（可选，为None则不保存）
        conf: 检测置信度阈值
    返回:
        track_behaviors: 每个跟踪ID的行为统计字典
    """

    # 第一步: 加载 YOLO8n-POSE 模型
    model = YOLO('yolov8n-pose.pt')

    # 第二步: 使用 ByteTrack 进行多目标跟踪
    # tracker='bytetrack.yaml' 指定使用 ByteTrack 算法
    # persist=True 使跟踪ID在视频帧之间持续
    results = model.track(
        source=video_path,
        tracker='bytetrack.yaml',  # 使用 ByteTrack 跟踪器
        persist=True,               # 保持跨帧ID连续性
        conf=conf,                   # 置信度阈值
        verbose=False               # 关闭详细日志
    )

    # 存储每个跟踪ID的行为记录
    track_behaviors = defaultdict(list)

    # 获取视频信息用于输出
    cap = cv2.VideoCapture(video_path)
    fps = int(cap.get(cv2.CAP_PROP_FPS))
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    cap.release()

    # 初始化视频写入器（如果需要输出）
    writer = None
    if output_path:
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

    # 第三步: 逐帧处理（检测 → 跟踪 → 分类）
    for frame_idx, result in enumerate(results):
        # 获取当前帧图像
        frame = result.orig_img

        # 检查是否检测到人体且有跟踪ID
        if result.boxes.id is None:
            if writer:
                writer.write(frame)
            continue

        # 提取跟踪ID、边界框和关键点
        track_ids = result.boxes.id.cpu().numpy().astype(int)  # 跟踪ID数组
        boxes = result.boxes.xyxy.cpu().numpy()   # 边界框坐标
        keypoints_data = result.keypoints.data.cpu().numpy()  # 关键点

        # 遍历当前帧的每个检测到的人体
        for i in range(len(track_ids)):
            tid = track_ids[i]          # 跟踪ID
            box = boxes[i]               # 边界框 [x1, y1, x2, y2]
            kps = keypoints_data[i]      # 17个关键点

            # 第三步-a: 对每个人体执行行为分类
            behavior = classify_behavior(kps)

            # 记录行为到该跟踪ID的历史列表
            track_behaviors[tid].append(behavior)

            # 第三步-b: 绘制可视化结果
            if writer:
                x1, y1, x2, y2 = map(int, box)
                color = BEHAVIOR_COLORS.get(behavior, (128, 128, 128))
                label_text = f"ID:{tid} {BEHAVIOR_LABELS.get(behavior, behavior)}"

                # 画边界框
                cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)

                # 画标签背景
                (tw, th), _ = cv2.getTextSize(label_text, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
                cv2.rectangle(frame, (x1, y1 - th - 10), (x1 + tw, y1), color, -1)
                cv2.putText(frame, label_text, (x1, y1 - 5),
                          cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)

        # 写入输出视频帧
        if writer:
            writer.write(frame)

    # 释放视频写入器
    if writer:
        writer.release()

    # 第四步: 统计每个跟踪ID的主要行为
    behavior_summary = {}
    for tid, behaviors in track_behaviors.items():
        # 使用 Counter 统计各行为出现次数
        counter = Counter(behaviors)
        # 取出现次数最多的行为作为该人员的主要行为
        main_behavior = counter.most_common(1)[0][0]
        behavior_summary[tid] = {
            'main_behavior': main_behavior,
            'behavior_count': dict(counter),
            'total_frames': len(behaviors)
        }

    return behavior_summary


# -----------------------------------------------
# 主程序入口
# -----------------------------------------------
if __name__ == '__main__':
    video_path = 'data/videos/classroom.mp4'
    output_path = 'data/videos/classroom_result.mp4'

    # 运行完整推理流水线
    summary = run_indoor_pipeline(video_path, output_path)

    # 打印每个学生的行为统计
    print("\n===== 课堂行为统计结果 =====")
    for tid, info in summary.items():
        print(f"学生ID {tid}: 主要行为={info['main_behavior']}, "
              f"总帧数={info['total_frames']}")
        print(f"  行为分布: {info['behavior_count']}")

代码详解

1. model.track() 与 model() 的区别：track() 方法在检测基础上增加了跟踪功能，自动调用 ByteTrack 算法，为每个检测到的人体分配唯一跟踪 ID。参数 persist=True 确保 ID 在视频帧之间连续，而非每帧重新分配。

2. 完整流水线（检测 → 跟踪 → 分类）：

  • 检测：YOLO8n-POSE 在每一帧中检测人体并提取 17 关键点
  • 跟踪：ByteTrack 将跨帧的同一人体关联到相同跟踪 ID
  • 分类：对每个跟踪 ID 的关键点调用 classify_behavior() 判断行为

3. result.boxes.id 的处理：跟踪模式下，boxes.id 存储每个检测框的跟踪 ID。当某一帧未检测到任何人体时，id 为 None，需做空值检查。

4. defaultdict(list) 的使用：使用默认字典自动初始化新跟踪 ID 的行为列表，避免手动检查键是否存在的冗余代码。

5. Counter.most_common()：统计每个 ID 在整个视频中各行为出现的次数，取出现最多的行为作为该学生的主要行为，适合生成课堂行为报告。

6. 可视化绘制：使用 OpenCV 在视频帧上绘制边界框和行为标签，不同行为使用不同颜色，便于直观查看分析结果。

易错提示

  • persist 参数：处理视频时必须设置 persist=True，否则每帧都会重新初始化跟踪器，ID 无法跨帧保持。处理图片流时可设为 False。
  • boxes.id 为 None：当某帧检测不到目标时，result.boxes.id 为 None，直接访问会导致报错。务必添加 if result.boxes.id is None: continue 判断。
  • 跟踪 ID 类型转换：boxes.id 返回的是浮点张量，需要 .astype(int) 转为整数才能作为字典键。
  • VideoWriter 编码：不同系统支持的视频编码不同，'mp4v' 是通用性较好的选择。若输出视频无法播放，可尝试 'XVID' 或 'avc1'。
  • 内存管理：长视频处理时，track_behaviors 会累积大量数据。若仅需最终统计，可改为增量计数而非存储全部行为列表。
任务 3 5分

MMAction2 实现室外人体行为识别

题目要求

实现 MMAction2 对课外人体行为的识别，完成课堂场景推理服务完整流水线（检测 → 跟踪 → 分类）功能。室外场景使用 TSN 模型（Kinetics-400 预训练）进行开放场景行为识别。

思路分析

室外场景与室内场景的区别在于：室内基于关键点几何规则，室外基于深度学习时序模型。室外模式仅支持视频输入（TSN 需要时序信息）。

  1. 使用 MMAction2 的 init_recognizer() 加载 TSN 模型（Kinetics-400 预训练配置）
  2. 使用 inference_recognizer() 对输入视频进行行为识别推理
  3. 读取 Kinetics-400 类别标签文件，将输出索引映射为行为名称
  4. 返回 Top-K 预测结果（行为名称 + 置信度）
  5. 结合 YOLO 检测，可选地对视频中的每个人体区域进行裁剪后再识别

代码实现

# ============================================
# 文件名: action_recognizer.py
# 功能: MMAction2 TSN 模型室外行为识别
# ============================================

import os
import cv2
import numpy as np
from mmaction.apis import init_recognizer, inference_recognizer


class OutdoorActionRecognizer:
    """室外行为识别器: 封装 MMAction2 TSN 模型的加载与推理"""

    def __init__(self, config_file, checkpoint_file, label_file,
                 device='cuda:0'):
        """
        初始化室外行为识别器
        参数:
            config_file: MMAction2 配置文件路径
            checkpoint_file: 模型权重文件路径
            label_file: Kinetics-400 标签文件路径
            device: 推理设备 ('cuda:0' 或 'cpu')
        """
        # 加载 TSN 配置文件和预训练权重
        self.model = init_recognizer(
            config_file,
            checkpoint_file,
            device=device
        )

        # 读取 Kinetics-400 类别标签
        self.labels = self._load_labels(label_file)

        print(f"TSN 模型加载完成，共 {len(self.labels)} 个行为类别")

    def _load_labels(self, label_file):
        """读取 Kinetics-400 标签文件，每行一个类别名"""
        with open(label_file, 'r', encoding='utf-8') as f:
            labels = [line.strip() for line in f.readlines()]
        return labels

    def recognize(self, video_path, top_k=5):
        """
        对视频进行行为识别
        参数:
            video_path: 输入视频文件路径
            top_k: 返回前K个预测结果
        返回:
            list[dict]: [{'action': 行为名, 'score': 置信度}, ...]
        """
        # 使用 MMAction2 进行推理
        results = inference_recognizer(self.model, video_path)

        # 获取预测分数（400个类别的概率分布）
        pred_scores = results.pred_score.cpu().numpy()

        # 取 Top-K 预测结果
        top_indices = pred_scores.argsort()[::-1][:top_k]

        # 构建结果列表
        predictions = []
        for idx in top_indices:
            predictions.append({
                'action': self.labels[idx],
                'score': float(pred_scores[idx])
            })

        return predictions


# -----------------------------------------------
# 结合 YOLO 检测 + TSN 识别的完整室外流水线
# -----------------------------------------------
def run_outdoor_pipeline(video_path, recognizer,
                         output_path=None, conf=0.5):
    """
    室外推理完整流水线: 检测 → 跟踪 → 分类
    先用 YOLO 检测视频中的人体区域，
    再对裁剪后的人体视频片段用 TSN 识别行为。
    """
    from ultralytics import YOLO
    import tempfile

    # 加载 YOLO 检测模型（仅检测人体，不需要关键点）
    det_model = YOLO('yolov8n-pose.pt')

    # 使用 ByteTrack 跟踪
    results = det_model.track(
        source=video_path,
        tracker='bytetrack.yaml',
        persist=True,
        conf=conf,
        classes=[0],  # 只检测人体 (COCO class 0)
        verbose=False
    )

    # 收集每个跟踪ID的边界框序列
    track_boxes = {}
    for result in results:
        if result.boxes.id is None:
            continue
        track_ids = result.boxes.id.cpu().numpy().astype(int)
        boxes = result.boxes.xyxy.cpu().numpy()
        for tid, box in zip(track_ids, boxes):
            if tid not in track_boxes:
                track_boxes[tid] = []
            track_boxes[tid].append(box)

    # 对每个跟踪ID: 裁剪人体区域 → TSN识别
    final_results = {}

    # 打开原始视频
    cap = cv2.VideoCapture(video_path)
    fps = int(cap.get(cv2.CAP_PROP_FPS))
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    for tid, box_list in track_boxes.items():
        # 计算该人体在所有帧中的包围框（取并集）
        all_boxes = np.array(box_list)
        x1 = int(all_boxes[:, 0].min())
        y1 = int(all_boxes[:, 1].min())
        x2 = int(all_boxes[:, 2].max())
        y2 = int(all_boxes[:, 3].max())

        # 确保裁剪区域不越界
        x1 = max(0, x1)
        y1 = max(0, y1)
        x2 = min(width, x2)
        y2 = min(height, y2)

        # 裁剪人体区域视频并保存为临时文件
        cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
        tmp_path = os.path.join(tempfile.gettempdir(), f'track_{tid}.mp4')
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        writer = cv2.VideoWriter(tmp_path, fourcc, fps, (x2 - x1, y2 - y1))

        while True:
            ret, frame = cap.read()
            if not ret:
                break
            cropped = frame[y1:y2, x1:x2]
            if cropped.size > 0:
                writer.write(cropped)
        writer.release()

        # 使用 TSN 识别裁剪后的人体行为
        predictions = recognizer.recognize(tmp_path, top_k=3)
        final_results[tid] = {
            'bbox': [x1, y1, x2, y2],
            'predictions': predictions
        }

        # 删除临时文件
        os.unlink(tmp_path)

    cap.release()
    return final_results


# -----------------------------------------------
# 主程序入口
# -----------------------------------------------
if __name__ == '__main__':
    # 初始化室外行为识别器
    recognizer = OutdoorActionRecognizer(
        config_file='configs/tsn_k400_config.py',
        checkpoint_file='models/tsn_k400.pth',
        label_file='labels/kinetics400_labels.txt',
        device='cuda:0'
    )

    video_path = 'data/videos/outdoor_running.mp4'

    # 方式1: 直接对整段视频识别
    print("===== 整段视频行为识别 =====")
    predictions = recognizer.recognize(video_path, top_k=5)
    for pred in predictions:
        print(f"  {pred['action']}: {pred['score']:.4f}")

    # 方式2: 完整流水线（检测 → 跟踪 → 识别）
    print("\n===== 完整流水线结果 =====")
    results = run_outdoor_pipeline(video_path, recognizer)
    for tid, info in results.items():
        print(f"跟踪ID {tid}:")
        for pred in info['predictions']:
            print(f"  {pred['action']}: {pred['score']:.4f}")

代码详解

1. OutdoorActionRecognizer 类设计：将 TSN 模型的加载和推理封装为类，初始化时加载模型和标签，提供 recognize() 方法供外部调用。这种面向对象的设计使代码结构清晰，便于在 FastAPI 中复用。

2. init_recognizer() 与 inference_recognizer()：

  • init_recognizer(config, checkpoint, device)：根据配置文件和权重构建 TSN 模型
  • inference_recognizer(model, video_path)：对视频执行推理，返回包含预测分数的结果对象

3. results.pred_score：TSN 模型输出 Kinetics-400 各类别的概率分布（400 维向量），通过 argsort()[::-1] 降序排列取 Top-K 结果。

4. 完整室外流水线 run_outdoor_pipeline()：

  • 使用 YOLO 检测 + ByteTrack 跟踪定位每个人体
  • 计算每个人体在整个视频中的最大包围框
  • 裁剪人体区域视频保存为临时文件
  • 对裁剪后的视频调用 TSN 进行行为识别
  • 清理临时文件，返回结构化结果

5. 室内外流水线对比：室内流水线使用关键点几何规则实时分类每帧行为；室外流水线使用 TSN 对视频片段进行整体行为识别，需要更多计算资源但能识别更复杂的动作。

易错提示

  • MMAction2 版本兼容：不同版本的 MMAction2 API 有差异，init_recognizer 和 inference_recognizer 的参数名可能不同，请根据安装版本查阅官方文档。
  • 配置文件路径：TSN 配置文件 .py 中引用的权重路径和数据集路径需正确设置，否则初始化会报错。
  • 视频格式：MMAction2 对视频编码有要求，某些编码的视频可能无法正确解码。建议使用 H.264 编码的 MP4 文件。
  • 裁剪区域越界：裁剪人体区域时必须检查坐标是否超出画面边界，否则 frame[y1:y2, x1:x2] 会返回空数组。
  • 临时文件清理：使用 tempfile 创建的临时视频文件需要手动删除，否则会占用大量磁盘空间。
  • classes=[0]：YOLO 默认检测 COCO 80 类，设置 classes=[0] 只保留人体检测结果，减少误检和计算量。
任务 4 5分

FastAPI 推理接口开发

题目要求

基于 FastAPI 编写推理接口，包括人体检测、关键点识别、室内人体行为识别、室外人体行为识别，支持图片与视频数据推理。

思路分析

将任务 1-3 的推理逻辑封装为 FastAPI HTTP 接口，实现前后端分离架构。

  1. 创建 FastAPI 应用实例，配置 CORS 中间件（允许前端跨域访问）
  2. 设计四个核心接口：
    • POST /api/detect — 人体检测（输入图片，返回边界框）
    • POST /api/keypoints — 关键点识别（输入图片，返回17关键点）
    • POST /api/indoor_behavior — 室内行为识别（支持图片和视频）
    • POST /api/outdoor_behavior — 室外行为识别（输入视频）
  3. 使用 UploadFile 接收上传的文件，Form 接收表单参数
  4. 将上传的文件转为 numpy 数组或临时文件，调用推理函数后返回 JSON 结果

代码实现

# ============================================
# 文件名: main.py
# 功能: FastAPI 推理服务（人体检测/关键点/室内/室外行为）
# 启动: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
# ============================================

import os
import cv2
import tempfile
import numpy as np
from fastapi import FastAPI, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from ultralytics import YOLO
from collections import Counter

# 导入自定模块
from behavior_rules import classify_behavior
from action_recognizer import OutdoorActionRecognizer

# -----------------------------------------------
# 第一步: 创建 FastAPI 应用并配置中间件
# -----------------------------------------------
app = FastAPI(
    title="人体行为识别 API",
    description="支持室内外人体行为识别的推理服务",
    version="1.0.0"
)

# 配置 CORS: 允许前端跨域请求
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],       # 允许所有来源（生产环境应限制具体域名）
    allow_credentials=True,
    allow_methods=["*"],        # 允许所有 HTTP 方法
    allow_headers=["*"],        # 允许所有请求头
)

# -----------------------------------------------
# 第二步: 全局加载模型（避免每次请求重复加载）
# -----------------------------------------------
print("正在加载模型...")
pose_model = YOLO('yolov8n-pose.pt')  # 姿态估计模型

# 室外行为识别器（懒加载: 首次调用室外接口时才加载）
outdoor_recognizer = None


def get_outdoor_recognizer():
    """懒加载室外行为识别器"""
    global outdoor_recognizer
    if outdoor_recognizer is None:
        outdoor_recognizer = OutdoorActionRecognizer(
            config_file='configs/tsn_k400_config.py',
            checkpoint_file='models/tsn_k400.pth',
            label_file='labels/kinetics400_labels.txt',
            device='cuda:0'
        )
    return outdoor_recognizer


# -----------------------------------------------
# 第三步: 定义 API 接口
# -----------------------------------------------

# --- 接口1: 人体检测 ---
@app.post("/api/detect")
async def detect_persons(file: UploadFile = File(...)):
    """
    人体检测接口
    输入: 图片文件
    输出: 每个人体的边界框和置信度
    """
    # 读取上传的文件内容
    contents = await file.read()

    # 将字节流转为 OpenCV 图像格式
    img_array = np.frombuffer(contents, np.uint8)
    img = cv2.imdecode(img_array, cv2.IMREAD_COLOR)

    if img is None:
        return JSONResponse(status_code=400,
                               content={"error": "无法解析图片文件"})

    # 执行 YOLO 推理
    results = pose_model(img, verbose=False)

    detections = []
    for box in results[0].boxes:
        detections.append({
            'bbox': box.xyxy[0].cpu().numpy().tolist(),
            'confidence': float(box.conf[0]),
            'class': int(box.cls[0])
        })

    return {'count': len(detections), 'detections': detections}


# --- 接口2: 关键点识别 ---
@app.post("/api/keypoints")
async def detect_keypoints(file: UploadFile = File(...)):
    """
    关键点识别接口
    输入: 图片文件
    输出: 每个人体的17个关键点坐标与置信度
    """
    contents = await file.read()
    img = cv2.imdecode(np.frombuffer(contents, np.uint8), cv2.IMREAD_COLOR)

    if img is None:
        return JSONResponse(status_code=400,
                               content={"error": "无法解析图片文件"})

    results = pose_model(img, verbose=False)

    # 关键点名称列表
    kp_names = ['nose', 'left_eye', 'right_eye', 'left_ear', 'right_ear',
                'left_shoulder', 'right_shoulder', 'left_elbow', 'right_elbow',
                'left_wrist', 'right_wrist', 'left_hip', 'right_hip',
                'left_knee', 'right_knee', 'left_ankle', 'right_ankle']

    persons = []
    if results[0].keypoints is None:
        return {'count': 0, 'persons': []}

    kps_data = results[0].keypoints.data.cpu().numpy()

    for i, kps in enumerate(kps_data):
        keypoints_list = []
        for j, name in enumerate(kp_names):
            keypoints_list.append({
                'name': name,
                'x': float(kps[j][0]),
                'y': float(kps[j][1]),
                'confidence': float(kps[j][2])
            })
        persons.append({'person_id': i, 'keypoints': keypoints_list})

    return {'count': len(persons), 'persons': persons}


# --- 接口3: 室内行为识别（支持图片和视频） ---
@app.post("/api/indoor_behavior")
async def indoor_behavior(
    file: UploadFile = File(...),
    is_video: bool = Form(False)
):
    """
    室内行为识别接口
    输入: 图片或视频文件, is_video 标记是否为视频
    输出: 每个人体的行为分类结果
    """
    contents = await file.read()

    if not is_video:
        # ===== 图片模式: 单帧行为识别 =====
        img = cv2.imdecode(np.frombuffer(contents, np.uint8), cv2.IMREAD_COLOR)
        if img is None:
            return JSONResponse(status_code=400, content={"error": "图片解析失败"})

        results = pose_model(img, verbose=False)

        behaviors = []
        if results[0].keypoints is None:
            return {'mode': 'image', 'count': 0, 'behaviors': []}

        kps_data = results[0].keypoints.data.cpu().numpy()
        boxes = results[0].boxes.xyxy.cpu().numpy()

        for i, kps in enumerate(kps_data):
            behavior = classify_behavior(kps)
            behaviors.append({
                'person_id': i,
                'behavior': behavior,
                'bbox': boxes[i].tolist() if len(boxes) > i else None
            })

        return {'mode': 'image', 'count': len(behaviors), 'behaviors': behaviors}

    else:
        # ===== 视频模式: 多帧跟踪 + 行为统计 =====
        # 保存上传的视频到临时文件
        suffix = os.path.pathsplit(file.filename)[1] if file.filename else '.mp4'
        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            tmp.write(contents)
            tmp_path = tmp.name

        try:
            # 使用 ByteTrack 跟踪
            results = pose_model.track(
                source=tmp_path,
                tracker='bytetrack.yaml',
                persist=True,
                verbose=False
            )

            track_behaviors = {}
            for result in results:
                if result.boxes.id is None:
                    continue
                track_ids = result.boxes.id.cpu().numpy().astype(int)
                kps_data = result.keypoints.data.cpu().numpy()

                for tid, kps in zip(track_ids, kps_data):
                    behavior = classify_behavior(kps)
                    if tid not in track_behaviors:
                        track_behaviors[tid] = []
                    track_behaviors[tid].append(behavior)

            # 统计每个跟踪ID的主要行为
            summary = {}
            for tid, bhvs in track_behaviors.items():
                counter = Counter(bhvs)
                summary[str(tid)] = {
                    'main_behavior': counter.most_common(1)[0][0],
                    'behavior_distribution': dict(counter),
                    'total_frames': len(bhvs)
                }

            return {'mode': 'video', 'count': len(summary), 'behaviors': summary}

        finally:
            # 确保临时文件被删除
            os.unlink(tmp_path)


# --- 接口4: 室外行为识别（仅支持视频） ---
@app.post("/api/outdoor_behavior")
async def outdoor_behavior(
    file: UploadFile = File(...),
    top_k: int = Form(5)
):
    """
    室外行为识别接口
    输入: 视频文件
    输出: Top-K 行为预测结果
    """
    contents = await file.read()

    # 保存视频到临时文件
    suffix = os.path.pathsplit(file.filename)[1] if file.filename else '.mp4'
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        tmp.write(contents)
        tmp_path = tmp.name

    try:
        # 获取室外识别器（懒加载）
        recognizer = get_outdoor_recognizer()

        # 执行 TSN 行为识别
        predictions = recognizer.recognize(tmp_path, top_k=top_k)

        return {
            'mode': 'outdoor_video',
            'predictions': predictions
        }

    finally:
        os.unlink(tmp_path)


# --- 健康检查接口 ---
@app.get("/api/health")
async def health_check():
    """健康检查接口，用于前端检测后端服务是否在线"""
    return {'status': 'ok', 'service': 'behavior_recognition'}


# -----------------------------------------------
# 启动服务: uvicorn main:app --host 0.0.0.0 --port 8000
# 访问 API 文档: http://localhost:8000/docs
# -----------------------------------------------

代码详解

1. CORS 中间件配置：前端页面和后端 API 通常运行在不同端口（如前端 5500，后端 8000），浏览器会阻止跨域请求。CORSMiddleware 允许跨域访问，开发阶段使用 allow_origins=["*"] 放行所有来源。

2. 全局模型加载：在应用启动时加载 YOLO 模型（耗时操作），避免每次请求都重新加载。室外识别器采用懒加载策略，首次调用室外接口时才加载 TSN 模型，减少启动时间。

3. UploadFile 与 Form 参数：

  • UploadFile：接收上传的文件，通过 await file.read() 读取二进制内容
  • Form：接收表单字段（如 is_video、top_k），前端以 FormData 形式提交

4. 图片解码流程：上传文件 → 字节流 → np.frombuffer → cv2.imdecode → numpy 数组。这是 Web 接收图片并转为 OpenCV 格式的标准流程。

5. 临时文件管理：视频文件需保存为临时文件供模型读取，使用 tempfile.NamedTemporaryFile 创建，try/finally 确保无论成功或异常都删除临时文件。

6. 接口设计规范：

  • 统一返回 JSON 格式，包含 mode、count、结果数据
  • 错误时返回 400 状态码和错误信息
  • 提供 /api/health 健康检查接口供前端探测服务状态

易错提示

  • async/await：FastAPI 接口函数中使用 async def，文件读取必须用 await file.read()，遗漏 await 会返回协程对象而非实际数据。
  • FormData 与 JSON：包含文件上传的接口必须使用 multipart/form-data 格式，不能使用 application/json。前端需使用 FormData 对象提交。
  • 文件扩展名：保存临时视频文件时需保留正确扩展名（.mp4/.avi），否则 OpenCV/FFmpeg 可能无法识别格式。使用 os.path.splitext 获取上传文件扩展名。
  • 临时文件泄漏：务必在 finally 块中删除临时文件。如果请求处理过程中抛出异常，try/finally 也能确保文件被清理。
  • 并发问题：YOLO 模型推理不是线程安全的，多并发请求可能导致冲突。生产环境可考虑使用请求队列或进程池。
  • python-multipart：处理文件上传需要安装 python-multipart 包，否则 FastAPI 会报错。
任务 5 5分

前端界面原型与接口对接

题目要求

具有前端界面原型，实现前端功能与推理后端进行接口对接。前端需支持图片/视频上传、场景模式选择（室内/室外）、调用后端 API 并展示识别结果。

思路分析

使用原生 HTML + CSS + JavaScript 开发前端界面，通过 Fetch API 调用后端 FastAPI 接口。

  1. 设计页面布局：标题栏、场景选择（室内/室外）、文件上传区、结果展示区
  2. 使用 FormData 封装上传文件和参数，通过 fetch() 发送 POST 请求
  3. 解析后端返回的 JSON 结果，动态渲染到页面
  4. 室内模式：展示每个人的行为标签和边界框
  5. 室外模式：展示 Top-K 行为预测结果和置信度
  6. 添加加载状态提示和错误处理

代码实现

<!-- ============================================ -->
<!-- 文件名: frontend/index.html -->
<!-- 功能: 人体行为识别前端界面 -->
<!-- ============================================ -->

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>人体行为识别系统</title>
    <style>
        /* 全局样式 */
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body {
            font-family: 'Microsoft YaHei', sans-serif;
            background: #f0f2f5; color: #333;
        }

        /* 顶部标题栏 */
        .header {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            color: white; padding: 20px; text-align: center;
        }
        .header h1 { font-size: 24px; }

        /* 主容器 */
        .container { max-width: 900px; margin: 20px auto; padding: 0 15px; }

        /* 模式选择卡片 */
        .mode-selector {
            display: flex; gap: 15px; margin-bottom: 20px;
        }
        .mode-card {
            flex: 1; padding: 20px; border-radius: 10px;
            background: white; cursor: pointer; text-align: center;
            border: 2px solid transparent; transition: all 0.3s;
        }
        .mode-card.active {
            border-color: #667eea; background: #f5f3ff;
        }
        .mode-card h3 { margin-bottom: 5px; color: #667eea; }
        .mode-card p { font-size: 13px; color: #888; }

        /* 上传区域 */
        .upload-area {
            background: white; border-radius: 10px; padding: 30px;
            margin-bottom: 20px; text-align: center;
        }
        .upload-area input[type="file"] { margin: 10px 0; }
        .btn {
            background: #667eea; color: white; border: none;
            padding: 10px 30px; border-radius: 5px; cursor: pointer;
            font-size: 15px; margin-top: 10px;
        }
        .btn:hover { background: #5568d3; }
        .btn:disabled { background: #ccc; cursor: not-allowed; }

        /* 结果展示区 */
        .result-area {
            background: white; border-radius: 10px; padding: 20px;
            min-height: 100px;
        }
        .result-item {
            padding: 12px; border-bottom: 1px solid #eee;
            display: flex; justify-content: space-between; align-items: center;
        }
        .behavior-tag {
            padding: 4px 12px; border-radius: 15px; font-size: 13px;
            color: white;
        }
        .loading { text-align: center; padding: 30px; color: #888; }
    </style>
</head>
<body>

<div class="header">
    <h1>人体行为识别系统</h1>
    <p>室内关键点检测 / 室外视频行为识别</p>
</div>

<div class="container">
    <!-- 场景模式选择 -->
    <div class="mode-selector">
        <div class="mode-card active" id="indoorMode" onclick="selectMode('indoor')">
            <h3>室内场景</h3>
            <p>YOLO8n-POSE + 关键点规则<br>支持图片和视频</p>
        </div>
        <div class="mode-card" id="outdoorMode" onclick="selectMode('outdoor')">
            <h3>室外场景</h3>
            <p>MMAction2 TSN 模型<br>仅支持视频</p>
        </div>
    </div>

    <!-- 文件上传区 -->
    <div class="upload-area">
        <h3>上传文件</h3>
        <input type="file" id="fileInput" accept="image/*,video/*">
        <br>
        <button class="btn" id="analyzeBtn" onclick="analyze()">开始识别</button>
    </div>

    <!-- 结果展示区 -->
    <div class="result-area">
        <h3>识别结果</h3>
        <div id="resultContent">
            <p style="color:#888;text-align:center;padding:20px;">
                请上传文件并点击"开始识别"
            </p>
        </div>
    </div>
</div>

<script>
    // 后端 API 基础地址
    const API_BASE = 'http://localhost:8000';

    // 当前选择的模式: indoor 或 outdoor
    let currentMode = 'indoor';

    // 切换场景模式
    function selectMode(mode) {
        currentMode = mode;
        // 更新UI样式
        document.getElementById('indoorMode').classList.toggle('active', mode === 'indoor');
        document.getElementById('outdoorMode').classList.toggle('active', mode === 'outdoor');

        // 室外模式仅支持视频
        const fileInput = document.getElementById('fileInput');
        if (mode === 'outdoor') {
            fileInput.accept = 'video/*';
        } else {
            fileInput.accept = 'image/*,video/*';
        }
    }

    // 开始识别
    async function analyze() {
        const fileInput = document.getElementById('fileInput');
        const btn = document.getElementById('analyzeBtn');
        const resultDiv = document.getElementById('resultContent');

        // 检查是否选择了文件
        if (!fileInput.files[0]) {
            alert('请先选择文件！');
            return;
        }

        const file = fileInput.files[0];

        // 显示加载状态
        btn.disabled = true;
        btn.textContent = '识别中...';
        resultDiv.innerHTML = '<div class="loading">正在分析，请稍候...</div>';

        // 构建 FormData
        const formData = new FormData();
        formData.append('file', file);

        try {
            let apiUrl, response, data;

            if (currentMode === 'indoor') {
                // 判断是图片还是视频
                const isVideo = file.type.startsWith('video/');
                formData.append('is_video', isVideo);
                apiUrl = API_BASE + '/api/indoor_behavior';

                response = await fetch(apiUrl, {
                    method: 'POST',
                    body: formData
                });
                data = await response.json();
                renderIndoorResult(data);

            } else {
                // 室外模式
                formData.append('top_k', 5);
                apiUrl = API_BASE + '/api/outdoor_behavior';

                response = await fetch(apiUrl, {
                    method: 'POST',
                    body: formData
                });
                data = await response.json();
                renderOutdoorResult(data);
            }

        } catch (error) {
            resultDiv.innerHTML = '<p style="color:red;">请求失败: ' + error.message + '</p>';
        } finally {
            btn.disabled = false;
            btn.textContent = '开始识别';
        }
    }

    // 渲染室内行为识别结果
    function renderIndoorResult(data) {
        const resultDiv = document.getElementById('resultContent');
        let html = '<p>模式: ' + data.mode + ' | 检测人数: ' + data.count + '</p>';

        // 行为颜色映射
        const colors = {
            'sit_listen': '#4CAF50', 'raise_hand': '#f44336',
            'write': '#2196F3', 'bow_head': '#FF9800',
            'stand': '#00BCD4', 'unknown': '#9E9E9E'
        };

        const labels = {
            'sit_listen': '听讲', 'raise_hand': '举手',
            'write': '写字', 'bow_head': '低头',
            'stand': '站立', 'unknown': '其他'
        };

        if (data.mode === 'image') {
            // 图片模式: 展示每个人行为
            data.behaviors.forEach(p => {
                const color = colors[p.behavior] || '#999';
                html += '<div class="result-item">';
                html += '<span>人员 #' + p.person_id + '</span>';
                html += '<span class="behavior-tag" style="background:' + color + '">';
                html += labels[p.behavior] || p.behavior;
                html += '</span></div>';
            });
        } else {
            // 视频模式: 展示每个跟踪ID主要行为
            for (const [tid, info] of Object.entries(data.behaviors)) {
                const color = colors[info.main_behavior] || '#999';
                html += '<div class="result-item">';
                html += '<span>学生 ID:' + tid + ' (' + info.total_frames + '帧)</span>';
                html += '<span class="behavior-tag" style="background:' + color + '">';
                html += labels[info.main_behavior] || info.main_behavior;
                html += '</span></div>';
            }
        }

        resultDiv.innerHTML = html;
    }

    // 渲染室外行为识别结果
    function renderOutdoorResult(data) {
        const resultDiv = document.getElementById('resultContent');
        let html = '<p>模式: 室外视频行为识别</p>';
        html += '<p>Top-5 预测结果:</p>';

        data.predictions.forEach((pred, i) => {
            const percent = (pred.score * 100).toFixed(2);
            const barWidth = pred.score * 100;
            html += '<div class="result-item">';
            html += '<span>' + (i + 1) + '. ' + pred.action + '</span>';
            html += '<span>' + percent + '%</span>';
            html += '</div>';
            // 置信度条形图
            html += '<div style="background:#eee;height:8px;border-radius:4px;margin:5px 0;">';
            html += '<div style="background:#667eea;height:100%;width:' + barWidth + '%;border-radius:4px;"></div>';
            html += '</div>';
        });

        resultDiv.innerHTML = html;
    }
</script>
</body>
</html>

代码详解

1. 页面布局结构：页面分为三个区域：场景模式选择卡片（室内/室外）、文件上传区、结果展示区。使用 Flexbox 布局和卡片式设计，界面简洁直观。

2. 模式切换逻辑 selectMode()：点击模式卡片切换 currentMode 变量，更新卡片样式高亮，并动态调整文件输入的 accept 属性（室外模式限制只能选择视频）。

3. 文件上传与 API 调用 analyze()：

  • 使用 FormData 封装文件和参数（is_video、top_k）
  • 通过 file.type.startsWith('video/') 自动判断文件类型
  • 使用 fetch() 发送 POST 请求到后端 API
  • 使用 async/await 处理异步请求
  • try/catch/finally 完整处理成功、失败和恢复状态

4. 结果渲染 renderIndoorResult()：

  • 图片模式：遍历 behaviors 数组，为每个人体显示编号和行为标签
  • 视频模式：遍历 behaviors 对象，显示每个跟踪 ID 的主要行为和帧数统计
  • 使用不同颜色标签区分行为类别，视觉清晰

5. 结果渲染 renderOutdoorResult()：

  • 展示 Top-K 行为预测结果
  • 使用 CSS 条形图可视化置信度，直观展示各行为概率分布

6. 用户体验优化：

  • 按钮禁用状态防止重复提交
  • 加载中提示文字
  • 错误信息友好展示

易错提示

  • 跨域问题：前端与后端不同源时，浏览器会阻止请求。确保后端已配置 CORS 中间件，或前后端部署在同一域名下。
  • FormData 字段名：前端 formData.append('file', file) 中的字段名 'file' 必须与后端 FastAPI 参数名 file: UploadFile 一致，否则后端无法接收文件。
  • Content-Type：使用 FormData 时不要手动设置 Content-Type，浏览器会自动设置为 multipart/form-data 并附带 boundary。
  • 布尔值传递：FormData.append('is_video', true) 会被转为字符串 "true"，FastAPI 的 Form(bool) 能正确解析。但要注意 "false" 字符串在 Python 中 bool("false") 为 True（非空字符串），需使用 FastAPI 的 Form 类型转换。
  • 文件大小限制：大视频文件上传可能超时或超出服务器限制。可在 FastAPI 中配置 Request 的 max_upload_size。
  • 异步等待：视频推理耗时较长，前端 fetch 请求可能超时。可在 fetch 中添加 signal 设置超时时间，或使用 WebSocket 实时推送进度。

2.5 项目总结与拓展

项目总结

本项目围绕"人体行为识别"主题，从模型推理、目标跟踪、API 服务到前端界面，构建了完整的端到端系统。以下是各任务的核心知识点回顾：

任务核心技术关键知识点
任务1 YOLO8n-POSE + 几何规则 17关键点提取、坐标系理解、行为规则设计、置信度过滤
任务2 ByteTrack 多目标跟踪 model.track()、persist参数、跨帧ID关联、Counter统计、视频可视化
任务3 MMAction2 TSN 模型 init_recognizer、inference_recognizer、Kinetics-400标签、视频裁剪、室外流水线
任务4 FastAPI 推理服务 路由定义、UploadFile、FormData、CORS中间件、懒加载、临时文件管理
任务5 HTML/CSS/JS 前端原型 页面布局、FormData封装、fetch API调用、动态渲染结果、跨域处理

核心技术回顾

1. YOLO8n-POSE 姿态估计：通过 ultralytics 库一行代码加载模型，对图片或视频执行推理，输出每个人体的 17 个 COCO 格式关键点坐标与置信度。这是室内行为识别的基础数据来源。

2. 关键点几何规则：利用关键点的 y 坐标大小关系判断上下位置（y 小 = 位置高），通过腿长/躯干长比值区分坐姿与站姿，结合手腕位置判断举手和写字。规则设计需考虑优先级和阈值调参。

3. ByteTrack 多目标跟踪：通过 model.track(tracker='bytetrack.yaml', persist=True) 启用，为视频中每个人体分配唯一跟踪 ID，实现跨帧行为持续分析。ByteTrack 的核心创新是利用低分检测框进行二次匹配。

4. MMAction2 TSN 时序识别：TSN 将视频分片段采样，用 2D CNN 提取特征后聚合，适合开放场景的复杂行为识别。Kinetics-400 预训练模型可直接识别 400 类常见行为。

5. FastAPI 前后端分离：将模型推理封装为 HTTP 接口，前端通过 fetch API 调用。关键要素包括 CORS 中间件、UploadFile 文件接收、FormData 表单参数、临时文件管理。

拓展提升方向

拓展方向具体内容难度
模型优化 尝试 YOLO8s/m-pose 等更大模型提升精度；使用自建课堂数据集微调（fine-tune）模型；引入 ST-GCN 等图卷积网络进行基于关键点的时序行为识别 中等
规则增强 引入角度计算（如肘关节角度）替代简单坐标比较；使用时序平滑（滑动窗口投票）减少单帧误判；增加更多行为类别（如睡觉、玩手机、交头接耳） 中等
室外模型升级 替换 TSN 为 SlowFast、TimeSformer 等更先进的时序模型；在自建数据集上微调以适应特定场景；使用多模态融合（RGB + 光流） 较高
系统部署 使用 Docker 容器化部署；添加 Nginx 反向代理和负载均衡；实现 WebSocket 实时视频流推理；部署到云服务器提供在线服务 较高
前端升级 使用 Vue/React 框架重构前端；添加 Canvas 在线绘制关键点和边界框；实现实时视频流展示和逐帧分析；添加数据可视化仪表盘 中等
数据分析 记录课堂行为数据到数据库；生成学生注意力分析报告；使用时序分析方法挖掘行为模式；对接教务系统进行教学评估 较高

常见问题排查清单

问题现象可能原因解决方案
YOLO 模型加载失败 网络问题导致权重下载失败 手动下载 yolov8n-pose.pt 放置到项目目录
关键点全为 0 图片中无人或人体过小 检查图片质量，调整 conf 阈值，确保画面中人体清晰可见
跟踪 ID 频繁切换 目标遮挡严重或 ByteTrack 参数不当 调整 bytetrack.yaml 中的 track_thresh 和 match_thresh 参数
MMAction2 推理报错 视频编码不支持或配置文件路径错误 用 ffmpeg 转码为 H.264 MP4；检查 config 文件中的路径引用
前端请求 403/ CORS 错误 后端未配置 CORS 或来源不匹配 确认 CORSMiddleware 配置正确，allow_origins 包含前端地址
视频推理超时 视频过长或服务器性能不足 限制视频长度；使用 GPU 加速；添加请求超时和进度提示
模块3 · 25%

具身智能平台部署与运维

基于 Docker + ROS2 + Dify + OpenClaw + DeepSeek，完成具身智能平台部署与运维

3.1 项目导读与学习目标

项目背景

具身智能（Embodied AI）是人工智能领域的前沿方向，它将大语言模型的"大脑"与机器人硬件的"身体"结合，使 AI 能够感知物理世界并执行实际操作。本项目模拟真实竞赛场景，要求选手在 Ubuntu 22.04 环境中，从零开始搭建一套完整的具身智能平台——从底层容器化部署，到中间件通信框架，再到上层智能体开发平台和大模型对接，最终实现"自然语言指令驱动机器人运动"的端到端闭环。

平台整体架构自下而上分为四层：

  • 基础设施层：Ubuntu 22.04 操作系统 + Docker 容器引擎，提供隔离的运行环境。
  • 机器人中间件层：ROS2 Humble 发行版，负责节点间通信与仿真控制。
  • 智能体开发层：Dify 平台 + OpenClaw 框架，用于编排 AI 智能体工作流。
  • 大模型能力层：DeepSeek API，提供自然语言理解与推理能力。

学习目标

目标维度 具体能力要求
知识理解 理解容器化、机器人操作系统、智能体框架、大模型 API 调用的基本概念与原理
环境部署 掌握 Docker Engine、Docker Compose、ROS2 Humble 的安装与配置方法
平台搭建 能够使用 Docker Compose 部署 Dify 平台并完成初始化配置
接口对接 掌握 DeepSeek 大模型 API 的调用方法，配置 API Key 与 Base URL
系统集成 能够通过 OpenClaw 智能体调用 ROS2 接口，实现自然语言到机器人控制的端到端闭环
故障排查 具备查看容器日志、排查端口冲突、验证服务连通性的运维能力

项目任务总览

本项目共 5 个任务，每个任务 5 分，合计 25 分。各任务之间存在递进依赖关系，建议按顺序完成：

1 Docker 容器化部署
→
2 ROS2 机器人框架
→
3 Dify 智能体平台
→
4 OpenClaw + DeepSeek
→
5 端到端闭环验证

3.2 知识准备

Docker

Docker 容器化技术

什么是 Docker？Docker 是一种容器化平台，它将应用程序及其依赖打包到一个轻量级、可移植的容器中。容器与虚拟机不同——虚拟机需要完整的操作系统，而容器共享宿主机内核，启动速度秒级，资源占用极小。

核心概念：

  • 镜像（Image）：只读模板，包含运行应用所需的所有内容（代码、库、环境变量等），类似于面向对象编程中的"类"。
  • 容器（Container）：镜像的运行实例，类似于"对象"。一个镜像可以启动多个容器。
  • Dockerfile：描述如何构建镜像的文本文件，类似于"构建脚本"。
  • Docker Compose：用于定义和运行多容器应用的工具，通过 YAML 文件编排多个服务。

常用命令速查：

# 拉取镜像
docker pull nginx
# 查看运行中的容器
docker ps
# 启动/停止容器
docker start <容器名>
docker stop <容器名>
# 查看容器日志
docker logs <容器名>
# 使用 Compose 启动服务（-d 表示后台运行）
docker compose up -d
ROS2

ROS2 机器人操作系统

什么是 ROS2？ROS2（Robot Operating System 2）是用于编写机器人软件的灵活框架，它提供了硬件抽象、设备驱动、进程间通信等功能。ROS2 Humble Hawksbill 是长期支持（LTS）版本，支持 Ubuntu 22.04。

核心概念：

  • 节点（Node）：ROS2 中的最小可执行单元，每个节点负责一个独立功能，如传感器读取、电机控制等。
  • 话题（Topic）：节点之间发布/订阅的通信通道。例如 /cmd_vel 话题用于发送速度控制指令。
  • 消息（Message）：话题中传递的数据格式。例如 geometry_msgs/msg/Twist 描述线速度和角速度。
  • 工作空间（Workspace）：组织 ROS2 包的目录结构，使用 colcon build 编译。

通信模型示意：

  ┌──────────────┐    发布     ┌──────────────┐    订阅     ┌──────────────┐
  │  发布者节点   │ ──────────→ │  /cmd_vel    │ ──────────→ │  仿真器节点   │
  │ Publisher   │   Twist消息  │   话题        │   Twist消息  │  turtlesim   │
  └──────────────┘             └──────────────┘             └──────────────┘

常用命令速查：

# 环境激活（每次开新终端需执行）
source /opt/ros/humble/setup.bash
# 创建工作空间目录结构
mkdir -p ~/ros2_ws/src
# 编译工作空间
cd ~/ros2_ws && colcon build
# 运行节点
ros2 run <包名> <节点名>
# 查看话题列表
ros2 topic list
# 查看话题消息类型
ros2 topic info /cmd_vel
Dify

Dify 智能体开发平台

什么是 Dify？Dify 是一个开源的 LLM（大语言模型）应用开发平台，它提供了可视化的工作流编排、模型管理、知识库构建等功能。开发者无需从零编写代码，即可快速搭建基于大模型的 AI 应用。

核心功能：

  • 应用管理：创建聊天助手、文本生成应用、Agent 智能体等多种类型的应用。
  • 模型供应商：支持接入 OpenAI、DeepSeek、通义千问等多种模型 API。
  • 知识库：上传文档构建 RAG（检索增强生成）知识库，让 AI 基于私有数据回答。
  • API 服务：每个应用自动生成 REST API，方便集成到外部系统。

部署架构：Dify 采用前后端分离的微服务架构，通过 Docker Compose 编排多个容器：

  ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
  │   Nginx     │ → │   Web API   │ → │ PostgreSQL  │
  │  (前端入口)  │   │  (后端API)   │   │  (数据存储)  │
  └─────────────┘   └─────────────┘   └─────────────┘
                          ↓
                    ┌─────────────┐   ┌─────────────┐
                    │  Redis      │   │  Weaviate   │
                    │ (缓存/队列)  │   │ (向量数据库) │
                    └─────────────┘   └─────────────┘
OpenClaw

OpenClaw 智能体框架

什么是 OpenClaw？OpenClaw 是一个开源的智能体编排框架，它允许开发者将大语言模型与外部工具（Tool）结合，使 AI 能够执行实际操作。在本项目中，OpenClaw 充当"翻译层"，将用户的自然语言指令翻译为 ROS2 可执行的控制命令。

核心概念：

  • Agent（智能体）：封装了模型能力和工具调用的智能实体，可以接收用户指令并自主决策执行步骤。
  • Tool（工具）：智能体可调用的外部能力。例如"控制机器人前进"就是一个 Tool，内部调用 ROS2 接口。
  • LLM Provider：大模型接口配置，指定使用哪个模型（如 DeepSeek）及其连接参数。

工作流程：

  用户输入自然语言         OpenClaw 调用 DeepSeek        大模型返回结构化
       "让乌龟前进"      →    分析意图选择工具       →    调用 ROS2 发布
                                                       /cmd_vel 速度消息
                                                            ↓
                                                      turtlesim 仿真器
                                                      执行运动
Embodied AI

具身智能概念

什么是具身智能？具身智能（Embodied Intelligence）是指 AI 系统拥有物理身体（或仿真身体），能够通过感知-决策-行动的闭环与物理世界交互。与传统"纯软件"AI 不同，具身智能强调"身体"在智能中的核心作用。

技术栈分层：

  • 感知层：摄像头、激光雷达、深度传感器等，获取环境信息。
  • 认知层：大语言模型（如 DeepSeek）进行自然语言理解、推理和决策。
  • 执行层：通过 ROS2 等中间件控制机器人执行运动指令。

本项目中的映射关系：

具身智能层次本项目中对应组件
认知层（大脑）DeepSeek 大模型 API
决策层（小脑）OpenClaw 智能体框架
通信层（神经）ROS2 中间件 /cmd_vel 话题
执行层（身体）turtlesim 仿真机器人
DeepSeek

DeepSeek 大模型 API

什么是 DeepSeek？DeepSeek 是国产高性能大语言模型，提供与 OpenAI API 兼容的接口，支持对话补全、函数调用等功能。在本项目中使用其 API 作为具身智能平台的"认知大脑"。

API 核心要素：

  • API Key：身份认证密钥，格式形如 sk-xxxxxxxxxxxx，用于验证调用者身份。
  • Base URL：API 基础地址，DeepSeek 的标准地址为 https://api.deepseek.com 或 https://api.deepseek.com/v1。
  • 模型名称：指定调用的模型版本，如 deepseek-chat（对话模型）或 deepseek-reasoner（推理模型）。

API 调用示例（兼容 OpenAI 格式）：

# HTTP 请求示例
POST https://api.deepseek.com/v1/chat/completions
Content-Type: application/json
Authorization: Bearer sk-xxxxxxxxxxxx

{
  "model": "deepseek-chat",
  "messages": [
    {"role": "user", "content": "你好"}
  ]
}

3.3 环境与数据准备

系统环境要求

项目 要求 说明
操作系统 Ubuntu 22.04 LTS (Jammy) 64位桌面版或服务器版均可
内存 ≥ 8GB（建议 16GB） Dify 多容器部署需较多内存
磁盘空间 ≥ 50GB 可用空间 Docker 镜像 + ROS2 + Dify 需大量空间
网络 可访问互联网 需下载软件包、Docker 镜像、调用 DeepSeek API
Python Python 3.10 Ubuntu 22.04 自带，ROS2 Humble 依赖此版本
用户权限 具有 sudo 权限的普通用户 安装软件需要 root 权限

软件清单与版本对照

软件名称 目标版本 用途
Docker Engine24.x+容器运行引擎
Docker Composev2.x（插件形式）多容器编排
ROS2Humble Hawksbill机器人通信中间件
colcon随 ROS2 安装ROS2 工作空间编译工具
Dify最新社区版智能体开发平台
OpenClaw最新版智能体编排框架
Python pip22.x+Python 包管理

项目目录结构规划

为保持文件有序，建议在用户主目录下创建如下项目结构：

~/
├── docker-projects/
│   └── demo-compose/          # 任务1：Docker Compose 演示
│       └── docker-compose.yml
├── ros2_ws/                   # 任务2：ROS2 工作空间
│   ├── src/
│   │   └── velocity_publisher/
│   │       ├── setup.py
│   │       ├── package.xml
│   │       └── velocity_publisher/
│   │           ├── __init__.py
│   │           └── publisher_node.py
│   ├── build/
│   ├── install/
│   └── log/
├── dify/                      # 任务3：Dify 部署目录
│   └── docker-compose.yaml
├── openclaw/                  # 任务4：OpenClaw 部署目录
│   ├── config.yaml
│   ├── test_model.py
│   └── ros2_bridge.py        # 任务5：ROS2 桥接脚本
└── .env                      # 环境变量（API Key 等）

执行以下命令创建基础目录结构：

# 创建项目所需的所有目录
mkdir -p ~/docker-projects/demo-compose
mkdir -p ~/ros2_ws/src/velocity_publisher/velocity_publisher
mkdir -p ~/dify
mkdir -p ~/openclaw

环境初始化检查

在开始任务前，先确认系统基础环境状态：

# 检查 Ubuntu 版本
lsb_release -a
# 预期输出：
# Description: Ubuntu 22.04.x LTS

# 检查 Python 版本（应 ≥ 3.10）
python3 --version
# 预期输出：Python 3.10.x

# 检查 pip 是否可用
pip3 --version

# 检查网络连通性（能否访问外网）
ping -c 4 8.8.8.8

# 更新软件包索引（确保后续安装顺利）
sudo apt update && sudo apt upgrade -y

DeepSeek API Key 获取

任务3和任务4需要使用 DeepSeek 大模型 API，请提前完成以下准备：

  1. 访问 DeepSeek 开放平台 https://platform.deepseek.com 注册账号。
  2. 在"API Keys"页面创建新的 API Key，格式形如 sk-xxxxxxxxxxxxxxxxxxxxxxxx。
  3. 记录 Base URL：https://api.deepseek.com（也可使用 https://api.deepseek.com/v1）。
  4. 确认账户有足够的 API 调用额度（新用户通常有赠送额度）。

安全提示：API Key 是敏感信息，切勿提交到公开代码仓库。建议存放在 .env 文件中并通过环境变量引用。

3.4 项目实施

5分

任务1：Docker Engine 与 Docker Compose 安装及容器编排

题目要求

在 Ubuntu 22.04 中安装 Docker Engine 与 Docker Compose 插件，编写 docker-compose.yml 编排文件，使用 docker compose up -d 启动并验证容器正常运行与服务可访问。

思路分析

本任务分为三个阶段：

  1. 安装 Docker Engine：使用 Docker 官方源安装最新稳定版，而非 Ubuntu 自带的较旧版本。
  2. 安装 Docker Compose 插件：现代 Docker 使用 Compose V2 插件（命令为 docker compose，无连字符），而非旧版独立的 docker-compose 程序。
  3. 编写编排文件并启动验证：创建一个 Nginx Web 服务作为演示，编写 YAML 配置，后台启动后通过浏览器或 curl 验证可访问。

验证点：容器状态为 running、映射端口可正常访问返回 HTTP 200。

步骤1：安装 Docker Engine

# ===== 第一步：卸载旧版本 Docker（如果有的话）=====
sudo apt remove docker docker-engine docker.io containerd runc -y

# ===== 第二步：安装必要的依赖工具 =====
# ca-certificates: SSL 证书
# curl: 下载工具
# gnupg: GPG 密钥管理
# lsb-release: 获取发行版信息
sudo apt update
sudo apt install ca-certificates curl gnupg lsb-release -y

# ===== 第三步：添加 Docker 官方 GPG 密钥 =====
# 创建密钥存放目录
sudo mkdir -p /etc/apt/keyrings
# 下载 Docker 的 GPG 密钥并转换为二进制格式保存
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | \
  sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg

# ===== 第四步：添加 Docker APT 软件源 =====
# 将 Docker 源写入 apt 源列表，系统会从此源获取 Docker 包
echo \
  "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] \
  https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

# ===== 第五步：安装 Docker Engine 和 Compose 插件 =====
sudo apt update
sudo apt install docker-ce docker-ce-cli containerd.io \
  docker-buildx-plugin docker-compose-plugin -y

# ===== 第六步：将当前用户加入 docker 组（免 sudo 运行 docker）=====
sudo usermod -aG docker $USER
# 注意：加入组后需要重新登录或执行 newgrp docker 生效
newgrp docker

# ===== 第七步：验证安装结果 =====
docker --version
# 预期输出：Docker version 24.x.x

docker compose version
# 预期输出：Docker Compose version v2.x.x

# 运行测试容器，验证 Docker 引擎正常工作
sudo docker run hello-world
# 预期输出：Hello from Docker! 说明引擎安装成功

步骤2：编写 docker-compose.yml

在 ~/docker-projects/demo-compose/ 目录下创建编排文件：

# 文件路径：~/docker-projects/demo-compose/docker-compose.yml
# Docker Compose 编排文件使用 YAML 格式
# 缩进必须使用空格（不能使用 Tab），每级 2 个空格

version: "3.8"          # Compose 文件格式版本

services:               # 定义所有服务（容器）
  web:                  # 服务名称：web
    image: nginx:latest # 使用 Nginx 最新版镜像
    container_name: demo-nginx  # 容器名称
    ports:              # 端口映射：宿主机端口:容器端口
      - "8080:80"       # 访问宿主机 8080 端口 → 转发到容器内 80 端口
    volumes:            # 数据卷挂载：宿主机路径:容器路径
      - ./html:/usr/share/nginx/html  # 挂载自定义网页目录
    restart: always     # 容器异常退出时自动重启
    healthcheck:        # 健康检查配置
      test: ["CMD", "curl", "-f", "http://localhost:80"]
      interval: 30s     # 每 30 秒检查一次
      timeout: 5s       # 超时 5 秒判定为不健康
      retries: 3        # 连续 3 次失败才标记为 unhealthy

创建自定义 HTML 页面用于验证服务可访问：

# 创建 HTML 目录
mkdir -p ~/docker-projects/demo-compose/html

# 创建测试页面 index.html
cat > ~/docker-projects/demo-compose/html/index.html << 'EOF'
<!DOCTYPE html>
<html>
<head><meta charset="UTF-8"><title>Docker 测试</title></head>
<body>
<h1>Docker Compose 部署成功！</h1>
<p>容器正常运行，服务可访问。</p>
</body>
</html>
EOF

步骤3：启动并验证容器

# 进入项目目录（docker compose 命令必须在有 yml 文件的目录执行）
cd ~/docker-projects/demo-compose

# 后台启动所有服务（-d 表示 detached 模式，即在后台运行）
docker compose up -d
# 预期输出：
# [+] Running 2/2
#  ✔ Container demo-nginx  Started

# 查看正在运行的容器
docker compose ps
# 预期输出：NAME=demo-nginx, STATUS=Up, PORTS=0.0.0.0:8080->80/tcp

# 查看容器日志（确认 Nginx 启动无错误）
docker compose logs web
# 预期输出：nginx: ... start worker processes

# 验证服务可访问（使用 curl 命令发送 HTTP 请求）
curl -I http://localhost:8080
# 预期输出：HTTP/1.1 200 OK

# 验证页面内容
curl http://localhost:8080
# 预期输出：包含 "Docker Compose 部署成功！"

# 如果在图形界面环境中，也可用浏览器访问 http://localhost:8080

# 停止并清理（验证完成后可执行）
docker compose down

代码详解

配置项说明
version: "3.8"Compose 文件格式版本，3.8 支持 healthcheck 等功能
image: nginx:latest指定镜像名称和标签，若本地不存在会自动从 Docker Hub 拉取
ports: "8080:80"端口映射，左边是宿主机端口，右边是容器内端口，用冒号分隔
volumes数据卷挂载，将宿主机目录映射到容器内，实现数据持久化和文件共享
restart: always容器退出策略，always 表示总是自动重启
healthcheck健康检查，定期检测容器内服务是否正常响应
docker compose up -d-d 参数让容器在后台运行，不加 -d 会占用终端显示日志

易错提示

  • 端口被占用：如果 8080 端口已被其他程序占用，容器无法启动。使用 sudo lsof -i:8080 查看占用进程，或修改 yml 中的端口映射为其他端口如 "8081:80"。
  • 权限问题：执行 docker 命令报 permission denied，说明当前用户未加入 docker 组。执行 sudo usermod -aG docker $USER 后重新登录终端。
  • YAML 缩进错误：YAML 文件对缩进极其敏感，必须使用空格而非 Tab。常见错误是 Tab 和空格混用导致解析失败。
  • docker compose vs docker-compose：注意新版使用 docker compose（空格分隔），旧版独立程序使用 docker-compose（连字符）。本项目使用新版插件方式。
  • 网络问题：国内拉取镜像可能较慢，可配置镜像加速器。编辑 /etc/docker/daemon.json 添加 registry-mirrors 配置。
5分

任务2：ROS2 Humble 安装与工作空间配置

题目要求

安装 ROS2 Humble Hawksbill 发行版，创建 colcon 工作空间，编写 Python 发布者节点向 /cmd_vel 话题发布速度指令消息，使用 colcon build 编译并运行节点验证消息发布。

思路分析

本任务分为四个阶段：

  1. 配置 ROS2 软件源：添加 ROS2 官方 APT 源和 GPG 密钥。
  2. 安装 ROS2 Humble：使用 ros-humble-desktop 元包安装完整版（含 rviz、turtlesim 等工具）。
  3. 创建工作空间和功能包：使用标准 colcon 目录结构创建 Python 包。
  4. 编写发布者节点：使用 rclpy 库创建节点，定时向 /cmd_vel 话题发布 Twist 速度消息。

验证点：节点启动后使用 ros2 topic echo /cmd_vel 能看到速度消息持续输出。

步骤1：安装 ROS2 Humble

# ===== 第一步：安装基础依赖和工具 =====
sudo apt update
sudo apt install software-properties-common curl gnupg lsb-release locales -y

# 确保系统 locale 支持 UTF-8（ROS2 要求）
sudo locale-gen en_US en_US.UTF-8
sudo update-locale LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8

# ===== 第二步：添加 ROS2 GPG 密钥 =====
sudo apt install ca-certificates -y
sudo curl -sSL https://raw.githubusercontent.com/ros/rosdistro/master/ros.key \
  -o /usr/share/keyrings/ros-archive-keyring.gpg

# ===== 第三步：添加 ROS2 APT 源 =====
echo "deb [arch=$(dpkg --print-architecture) \
  signed-by=/usr/share/keyrings/ros-archive-keyring.gpg] \
  http://packages.ros.org/ros2/ubuntu $(lsb_release -cs) main" | \
  sudo tee /etc/apt/sources.list.d/ros2.list > /dev/null

# ===== 第四步：安装 ROS2 Humble 桌面完整版 =====
# ros-humble-desktop 包含：核心库 + 可视化工具（rviz2）+ 仿真（turtlesim）
sudo apt update
sudo apt install ros-humble-desktop python3-colcon-common-extensions \
  python3-rosdep python3-vcstool -y

# ===== 第五步：初始化 rosdep（依赖管理工具）=====
sudo rosdep init
rosdep update

# ===== 第六步：验证安装 =====
# 激活 ROS2 环境
source /opt/ros/humble/setup.bash
# 检查 ROS2 版本
ros2 --help
# 预期输出：显示 ros2 命令帮助信息，说明安装成功

将 ROS2 环境变量写入 bashrc，实现自动加载：

# 将 ROS2 环境激活命令添加到 ~/.bashrc，每次开终端自动执行
echo "source /opt/ros/humble/setup.bash" >> ~/.bashrc
# 立即生效
source ~/.bashrc

步骤2：创建工作空间与功能包

# ===== 创建 ROS2 工作空间目录 =====
# 工作空间标准结构：src/ 存放源码，build/ 存放编译中间文件，install/ 存放编译产物
mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src

# ===== 创建 Python 功能包 =====
# ros2 pkg create: 创建新包的命令
# --build-type ament_python: 使用 Python 构建系统
# velocity_publisher: 包名（必须用小写字母和下划线）
ros2 pkg create --build-type ament_python velocity_publisher

# 创建后目录结构如下：
# ~/ros2_ws/src/velocity_publisher/
# ├── package.xml          # 包描述文件（依赖、作者等）
# ├── setup.py             # Python 安装脚本（注册入口点）
# ├── setup.cfg            # 安装配置
# └── velocity_publisher/  # Python 源码目录
#     ├── __init__.py       # 包初始化文件（空文件）
#     └── publisher_node.py # 我们要创建的节点文件

步骤3：编写 Python 发布者节点

创建发布者节点文件 ~/ros2_ws/src/velocity_publisher/velocity_publisher/publisher_node.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# 导入 rclpy 库，这是 ROS2 的 Python 客户端库
import rclpy
from rclpy.node import Node

# 导入 Twist 消息类型，用于描述机器人速度（线速度 + 角速度）
# geometry_msgs/msg/Twist 包含 linear（线速度 xyz）和 angular（角速度 xyz）
from geometry_msgs.msg import Twist


class VelocityPublisher(Node):
    """速度指令发布者节点

    该节点定时向 /cmd_vel 话题发布 Twist 速度消息，
    使机器人（如 turtlesim）执行前进和旋转运动。
    """

    def __init__(self):
        # 调用父类构造函数，设置节点名称为 'velocity_publisher'
        super().__init__('velocity_publisher')

        # 创建发布者对象
        # 参数1：消息类型为 Twist
        # 参数2：话题名称为 '/cmd_vel'（turtlesim 默认监听此话题）
        # 参数3：队列大小为 10，表示最多缓存 10 条未发送的消息
        self.publisher_ = self.create_publisher(Twist, '/cmd_vel', 10)

        # 创建定时器，每 0.5 秒调用一次 timer_callback
        # 0.5 秒 = 2Hz 发布频率
        timer_period = 0.5  # 单位：秒
        self.timer = self.create_timer(timer_period, self.timer_callback)

        # 在终端打印日志信息，表示节点已启动
        self.get_logger().info('速度发布者节点已启动，正在向 /cmd_vel 发布消息...')

    def timer_callback(self):
        """定时器回调函数：每次触发时创建并发布一条速度消息"""
        # 创建 Twist 消息对象
        msg = Twist()

        # 设置线速度（linear.x：前进/后退速度，单位 m/s）
        # 正值 = 前进，负值 = 后退
        msg.linear.x = 0.5   # 前进速度 0.5 m/s
        msg.linear.y = 0.0   # 侧向速度（差速驱动机器人通常为0）
        msg.linear.z = 0.0   # 垂直速度（地面机器人通常为0）

        # 设置角速度（angular.z：绕 Z 轴旋转速度，单位 rad/s）
        # 正值 = 逆时针旋转，负值 = 顺时针旋转
        msg.angular.z = 0.3  # 逆时针旋转 0.3 rad/s
        msg.angular.x = 0.0
        msg.angular.y = 0.0

        # 发布消息到 /cmd_vel 话题
        self.publisher_.publish(msg)

        # 打印发布的速度值，方便调试
        self.get_logger().info(
            f'发布速度: linear.x={msg.linear.x:.2f} m/s, '
            f'angular.z={msg.angular.z:.2f} rad/s'
        )


def main(args=None):
    """主函数：ROS2 Python 节点的入口点"""
    # 初始化 rclpy 库（必须在创建节点之前调用）
    rclpy.init(args=args)

    # 创建节点实例
    velocity_publisher = VelocityPublisher()

    # 进入消息循环（spin 会持续处理回调，直到节点被关闭）
    # spin() 是阻塞函数，会一直运行直到 Ctrl+C 中断
    try:
        rclpy.spin(velocity_publisher)
    except KeyboardInterrupt:
        # 用户按 Ctrl+C 时优雅退出
        velocity_publisher.get_logger().info('节点正在关闭...')
    finally:
        # 销毁节点，释放资源
        velocity_publisher.destroy_node()
        # 关闭 rclpy 库
        rclpy.shutdown()


# 当直接运行此脚本时调用 main 函数
if __name__ == '__main__':
    main()

步骤4：配置入口点并编译运行

编辑 setup.py 注册节点入口点：

# 文件路径：~/ros2_ws/src/velocity_publisher/setup.py
from setuptools import setup
from setuptools.command.install import install
import os
from glob import glob

package_name = 'velocity_publisher'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='student',
    maintainer_email='student@example.com',
    description='Velocity publisher node for /cmd_vel topic',
    license='MIT',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            # 格式：命令名 = 包名.模块名:函数名
            # 注册后可用 ros2 run velocity_publisher velocity_publisher 运行
            'velocity_publisher = velocity_publisher.publisher_node:main',
        ],
    },
)

编译并运行验证：

# ===== 编译工作空间 =====
cd ~/ros2_ws

# 安装功能包依赖（首次编译前执行）
rosdep install --from-paths src --ignore-src -r -y

# 编译所有包（colcon build 会自动处理依赖关系）
colcon build
# 预期输出：Finished << velocity_publisher >>

# 激活工作空间环境（编译后必须执行才能找到新包）
source ~/ros2_ws/install/setup.bash

# ===== 验证方式1：检查话题消息 =====
# 终端1 - 运行发布者节点
ros2 run velocity_publisher velocity_publisher
# 预期输出：每隔0.5秒打印一次速度值

# 终端2（新开终端）- 监听话题消息
source /opt/ros/humble/setup.bash
source ~/ros2_ws/install/setup.bash
ros2 topic echo /cmd_vel
# 预期输出：持续显示 Twist 消息内容
# linear:
#   x: 0.5
# angular:
#   z: 0.3

# ===== 验证方式2：配合 turtlesim 仿真 =====
# 终端1 - 启动 turtlesim 仿真器
ros2 run turtlesim turtlesim_node
# 预期：弹出图形窗口显示一只乌龟

# 终端2 - 运行发布者节点
ros2 run velocity_publisher velocity_publisher
# 预期：乌龟开始画圆运动

# 查看话题列表，确认 /cmd_vel 存在
ros2 topic list
# 预期输出中包含：/cmd_vel

# 查看话题信息
ros2 topic info /cmd_vel
# 预期输出：Type: geometry_msgs/msg/Twist

代码详解

代码要素说明
rclpy.init()初始化 ROS2 Python 客户端库，必须在使用任何 ROS2 功能前调用
class VelocityPublisher(Node)自定义节点继承自 Node 基类，获得 ROS2 通信能力
create_publisher(Twist, '/cmd_vel', 10)创建发布者，指定消息类型、话题名和 QoS 队列大小
create_timer(0.5, callback)创建定时器，每 0.5 秒触发一次回调函数发布消息
Twist 消息包含 linear（线速度）和 angular（角速度），各含 xyz 三轴
rclpy.spin()进入事件循环，持续处理定时器和回调，阻塞直到节点关闭
entry_points在 setup.py 中注册，使节点可通过 ros2 run 命令启动
colcon build编译工作空间，生成可执行的 install 目录

易错提示

  • 忘记 source 环境变量：每次新开终端都需要执行 source /opt/ros/humble/setup.bash 和 source ~/ros2_ws/install/setup.bash，否则找不到 ros2 命令或自定义包。建议写入 ~/.bashrc 自动加载。
  • __init__.py 文件缺失：Python 包目录下必须有 __init__.py 文件（可以为空），否则 Python 无法将其识别为包。
  • 包名冲突：自定义包名不能与已有 ROS2 包同名。使用 ros2 pkg list 查看已有包。
  • 编译目录错误：colcon build 必须在工作空间根目录（~/ros2_ws）执行，不能在 src 目录内执行。
  • resource 目录缺失：ament_python 包需要在 src/包名/resource/ 目录下有一个与包同名的空文件，否则编译后 ros2 run 找不到包。使用 ros2 pkg create 创建时会自动生成。
  • turtlesim 无图形界面：在服务器或无桌面环境中运行 turtlesim_node 会报错。可以使用 ros2 topic echo /cmd_vel 替代验证。
5分

任务3：Dify 智能体开发平台部署与 DeepSeek 模型接入

题目要求

使用 Docker Compose 部署 Dify 智能体开发平台，完成管理员账号初始化注册，创建聊天助手应用并配置 DeepSeek 模型接入（API Key、Base URL），验证平台页面可正常访问与对话功能。

思路分析

本任务分为四个阶段：

  1. 克隆 Dify 仓库：从 GitHub 克隆官方仓库，使用其提供的 docker-compose 编排文件。
  2. 配置环境变量：复制 .env.example 为 .env，按需修改密码、端口等配置。
  3. 启动服务并初始化：使用 docker compose 启动所有容器，通过浏览器访问并注册管理员账号。
  4. 接入 DeepSeek 模型：在平台设置中添加 DeepSeek 模型供应商，填入 API Key 和 Base URL。

验证点：Dify 首页可访问、管理员账号注册成功、创建的聊天应用能正常对话返回结果。

步骤1：克隆并配置 Dify

# ===== 第一步：克隆 Dify 官方仓库 =====
# 如果网络较慢，可使用镜像加速：git clone https://ghproxy.com/https://github.com/langgenius/dify.git
cd ~
git clone https://github.com/langgenius/dify.git

# ===== 第二步：进入 Docker 部署目录 =====
cd dify/docker

# ===== 第三步：复制环境变量模板文件 =====
# .env 文件包含所有配置项：密码、端口、密钥等
cp .env.example .env

# ===== 第四步：生成随机密钥（增强安全性）=====
# 生成 SECRET_KEY 用于会话加密
SECRET_KEY=$(openssl rand -base64 42)
# 将生成的密钥写入 .env 文件
sed -i "s/SECRET_KEY=.*/SECRET_KEY=${SECRET_KEY}/" .env

步骤2：启动 Dify 平台

# ===== 启动所有 Dify 服务容器 =====
cd ~/dify/docker
docker compose up -d

# 预期输出：多个容器依次启动
# [+] Running 11/11
#  ✔ Container docker-nginx-1          Started
#  ✔ Container docker-web-1            Started
#  ✔ Container docker-api-1            Started
#  ✔ Container docker-worker-1         Started
#  ✔ Container docker-db-1             Started
#  ✔ Container docker-redis-1          Started
#  ... （其他容器）

# ===== 查看所有容器运行状态 =====
docker compose ps
# 确保所有容器 STATUS 为 Up 状态
# 如果有容器状态为 Exited 或 Restarting，查看日志排查

# ===== 查看服务日志（确认无错误）=====
docker compose logs -f --tail=50
# 看到 "Listening at: http://0.0.0.0:5001" 说明 API 服务启动成功
# 按 Ctrl+C 退出日志查看

# ===== 验证服务可访问 =====
curl -I http://localhost
# 预期输出：HTTP/1.1 200 OK
# Dify 默认通过 Nginx 暴露在 80 端口

步骤3：初始化管理员账号

# ===== 方式一：通过浏览器访问初始化 =====
# 1. 在浏览器中打开 http://localhost（或 http://<服务器IP>）
# 2. 首次访问会跳转到"安装/注册"页面
# 3. 填写邮箱和密码，点击"设置"完成管理员账号注册
# 4. 注册成功后自动登录进入 Dify 工作台

# ===== 方式二：通过 API 初始化（适合自动化场景）=====
# 调用 Dify 安装接口设置管理员账号
curl -X POST http://localhost/console/api/install \
  -H "Content-Type: application/json" \
  -d '{
    "email": "admin@example.com",
    "name": "Admin",
    "password": "Dify@123456"
  }'
# 预期输出：{"result": "success", "data": {"access_token": "..."}}
# 记住邮箱和密码，后续登录使用

步骤4：配置 DeepSeek 模型接入

在 Dify Web 界面中完成 DeepSeek 模型接入：

# ===== 在 Dify 界面操作步骤 =====

# 1. 登录 Dify 平台（http://localhost）
# 2. 点击右上角头像 → "设置" → "模型供应商"

# 3. 在模型供应商列表中找到 "DeepSeek"，点击"添加模型"

# 4. 填写配置：
#    - 模型类型：LLM（大语言模型）
#    - 模型名称：deepseek-chat
#    - API Key：sk-xxxxxxxxxxxxxxxxxxxx（替换为你的实际 Key）
#    - API endpoint URL：https://api.deepseek.com
#    - 上下文 Token 上限：64000（或按需设置）
#    - 最大 Token 上限：4096
#    - 温度：0.7（0=确定，1=随机，影响回答的创造性）

# 5. 点击"保存"，模型列表中应显示 DeepSeek 已启用

# ===== 验证模型是否可用（通过 API）=====
# 获取 API Key（在 Dify 设置 → API 密钥 中查看）
DIFY_API_KEY="app-xxxxxxxxxxxxxxxxx"

curl -X POST http://localhost/v1/chat-messages \
  -H "Authorization: Bearer ${DIFY_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "inputs": {},
    "query": "你好，请做一下自我介绍",
    "response_mode": "blocking",
    "user": "test-user"
  }'
# 预期输出：包含 DeepSeek 生成的回答内容
# {"answer": "你好！我是 DeepSeek...", ...}

步骤5：创建聊天助手应用

# ===== 在 Dify 界面创建应用 =====

# 1. 在 Dify 首页点击"创建空白应用"
# 2. 选择应用类型："聊天助手"
# 3. 填写应用名称："DeepSeek 聊天助手"
# 4. 填写描述："基于 DeepSeek 模型的对话应用"
# 5. 点击"创建"

# 6. 进入应用编排页面：
#    - 在右上角模型选择器中选择 "DeepSeek → deepseek-chat"
#    - 编写系统提示词（Prompt）：
#      "你是一个智能助手，能够回答用户的各种问题。请用中文回答。"

# 7. 在右侧预览区域输入消息测试对话
#    - 输入："你好"
#    - 预期返回：DeepSeek 的回复消息

# 8. 点击右上角"发布"按钮，使应用上线
#    发布后可通过独立 URL 访问聊天界面

# ===== 验证应用可访问 =====
# 在浏览器中打开应用访问链接（Dify 会生成独立 URL）
# 输入消息测试对话功能正常返回

代码详解

配置/操作说明
.env 文件Dify 全局配置文件，包含数据库密码、加密密钥、端口映射等
SECRET_KEY用于加密会话和敏感数据的密钥，必须随机生成且唯一
docker compose up -d后台启动所有 Dify 服务容器（Nginx、API、Web、Worker、DB、Redis 等）
模型供应商Dify 的模型管理入口，统一管理多个 AI 模型的 API 配置
API endpoint URL即 Base URL，DeepSeek 为 https://api.deepseek.com
temperature温度参数控制输出随机性，0 最确定，1 最随机，通常设 0.7
response_mode: blocking阻塞模式等待完整回复后再返回，streaming 模式则流式返回

易错提示

  • 端口 80 被占用：Dify 默认使用 80 端口。如已被占用，编辑 ~/dify/docker/docker-compose.yaml 中 nginx 服务的端口映射，如改为 "3000:80"，或修改 .env 中的 EXPOSE_NGINX_PORT 为 3000。
  • 容器启动失败：使用 docker compose logs <服务名> 查看具体错误。常见原因有内存不足（需 ≥4GB 可用内存）、磁盘空间不足、配置文件格式错误。
  • DeepSeek API Key 无效：检查 Key 是否正确复制（注意前后无空格），确认账户有调用额度，确认 Base URL 为 https://api.deepseek.com（不要多加或漏加路径）。
  • 模型未在应用中选择：创建应用后必须在编排页面右上角选择 DeepSeek 模型，否则对话会报"No model configured"错误。
  • 忘记发布应用：创建和配置后需点击"发布"按钮，否则应用 API 不可用。
  • git clone 失败：国内访问 GitHub 可能超时，可使用镜像加速地址或提前下载好仓库压缩包。
5分

任务4：OpenClaw 智能体部署与 DeepSeek API 对接

题目要求

部署 OpenClaw 智能体框架，配置 DeepSeek 大模型 API 接口连接参数，编写测试脚本调用模型接口发送对话请求，验证大模型对话功能正常返回结果。

思路分析

本任务分为四个阶段：

  1. 安装 OpenClaw 框架：通过 pip 或源码方式安装 OpenClaw 及其依赖。
  2. 配置连接参数：创建配置文件，填入 DeepSeek API Key、Base URL、模型名称等参数。
  3. 编写测试脚本：使用 Python 调用 OpenClaw 接口发送对话请求。
  4. 验证返回结果：确认大模型正常返回对话内容，无报错。

验证点：脚本运行后成功打印大模型返回的回答内容。

步骤1：安装 OpenClaw 与依赖

# ===== 创建 Python 虚拟环境（隔离依赖，避免冲突）=====
cd ~/openclaw
python3 -m venv venv
# 激活虚拟环境
source venv/bin/activate

# ===== 升级 pip 工具 =====
pip install --upgrade pip

# ===== 安装 OpenClaw 框架 =====
# OpenClaw 可通过 pip 安装
pip install openclaw

# ===== 安装其他必要依赖 =====
# openai: DeepSeek API 兼容 OpenAI 接口格式
# python-dotenv: 从 .env 文件加载环境变量
pip install openai python-dotenv pyyaml

# ===== 验证安装 =====
python3 -c "import openclaw; print('OpenClaw 安装成功')"
# 预期输出：OpenClaw 安装成功

步骤2：配置 DeepSeek 连接参数

创建环境变量文件（安全存储 API Key）：

# 文件路径：~/openclaw/.env
# 此文件存储敏感信息，不要提交到 Git 仓库

# DeepSeek API Key（替换为你的实际 Key）
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxx

# DeepSeek API Base URL
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

# 使用的模型名称
DEEPSEEK_MODEL=deepseek-chat

创建 OpenClaw 配置文件 ~/openclaw/config.yaml：

# 文件路径：~/openclaw/config.yaml
# OpenClaw 框架配置文件（YAML 格式）

# 大模型供应商配置
llm:
  provider: deepseek              # 模型供应商名称
  api_key: ${DEEPSEEK_API_KEY}    # 从环境变量读取（安全引用）
  base_url: ${DEEPSEEK_BASE_URL}   # API 基础地址
  model: ${DEEPSEEK_MODEL}        # 模型名称
  temperature: 0.7                 # 温度参数：0-1，越高越随机
  max_tokens: 2048                # 最大返回 Token 数

# Agent 智能体配置
agent:
  name: "embodied_agent"          # 智能体名称
  description: "具身智能控制智能体" # 智能体描述
  system_prompt: |
    你是一个具身智能助手，能够理解用户的自然语言指令，
    并将其转换为机器人控制命令。
    请始终用中文回答。

# 日志配置
logging:
  level: INFO                     # 日志级别：DEBUG/INFO/WARNING/ERROR

步骤3：编写测试脚本

创建测试脚本 ~/openclaw/test_model.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
DeepSeek 大模型接口测试脚本

本脚本通过 OpenClaw 框架调用 DeepSeek API，
发送对话请求并验证大模型是否正常返回结果。
"""

# 导入所需库
import os
import sys
import json
from dotenv import load_dotenv

# 加载 .env 文件中的环境变量
load_dotenv()

# 从环境变量读取配置
API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')
MODEL = os.getenv('DEEPSEEK_MODEL')

# 检查必要配置是否存在
if not API_KEY:
    print("错误：未找到 DEEPSEEK_API_KEY 环境变量")
    print("请在 .env 文件中设置 DEEPSEEK_API_KEY")
    sys.exit(1)

if not BASE_URL:
    print("错误：未找到 DEEPSEEK_BASE_URL 环境变量")
    sys.exit(1)


def test_deepseek_chat():
    """测试 DeepSeek 对话功能"""

    # 使用 OpenAI 兼容接口调用 DeepSeek
    # DeepSeek API 完全兼容 OpenAI SDK 格式
    from openai import OpenAI

    # 创建客户端实例
    # api_key: 认证密钥
    # base_url: API 服务地址（指向 DeepSeek 而非 OpenAI）
    client = OpenAI(
        api_key=API_KEY,
        base_url=BASE_URL
    )

    # 定义测试对话消息列表
    # system: 系统提示词，设定 AI 角色
    # user: 用户输入消息
    messages = [
        {
            "role": "system",
            "content": "你是一个具身智能助手，请用简洁的中文回答问题。"
        },
        {
            "role": "user",
            "content": "你好，请用一句话介绍你自己。"
        }
    ]

    print("=" * 50)
    print("DeepSeek 大模型接口测试")
    print("=" * 50)
    print(f"模型: {MODEL}")
    print(f"API 地址: {BASE_URL}")
    print("-" * 50)
    print("发送消息: ", messages[1]["content"])
    print("等待模型回复...")
    print("-" * 50)

    try:
        # 调用 chat completions 接口
        response = client.chat.completions.create(
            model=MODEL,          # 指定模型
            messages=messages,     # 传入对话历史
            temperature=0.7,      # 控制随机性
            max_tokens=2048      # 最大输出长度
        )

        # 提取回复内容
        answer = response.choices[0].message.content
        usage = response.usage

        print("模型回复:")
        print(answer)
        print("-" * 50)
        print(f"Token 使用: 输入={usage.prompt_tokens}, "
              f"输出={usage.completion_tokens}, "
              f"总计={usage.total_tokens}")
        print("=" * 50)
        print("测试结果: 成功 - 大模型对话功能正常！")

        return True

    except Exception as e:
        print(f"测试结果: 失败 - 错误信息: {e}")
        print("=" * 50)
        return False


def test_deepseek_function_call():
    """测试 DeepSeek 函数调用功能（为任务5做准备）"""

    from openai import OpenAI

    client = OpenAI(api_key=API_KEY, base_url=BASE_URL)

    # 定义可用函数（工具）
    tools = [
        {
            "type": "function",
            "function": {
                "name": "control_robot",
                "description": "控制机器人运动",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "action": {
                            "type": "string",
                            "enum": ["forward", "backward",
                                       "turn_left", "turn_right",
                                       "stop"],
                            "description": "运动方向"
                        },
                        "speed": {
                            "type": "number",
                            "description": "运动速度 (m/s)，范围 0.0-2.0"
                        }
                    },
                    "required": ["action"]
                }
            }
        }
    ]

    print("\n函数调用测试:")
    print("-" * 50)

    try:
        response = client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system",
                 "content": "你是机器人控制助手，根据用户指令调用 control_robot 函数。"},
                {"role": "user",
                 "content": "让机器人向前走，速度1.0"}
            ],
            tools=tools,
            tool_choice="auto"
        )

        # 检查模型是否调用了函数
        message = response.choices[0].message
        if message.tool_calls:
            tool_call = message.tool_calls[0]
            print(f"模型决定调用函数: {tool_call.function.name}")
            print(f"函数参数: {tool_call.function.arguments}")
            print("函数调用测试: 成功！")
            return True
        else:
            print(f"模型回复: {message.content}")
            print("函数调用测试: 模型未调用函数")
            return False

    except Exception as e:
        print(f"函数调用测试失败: {e}")
        return False


if __name__ == "__main__":
    # 运行对话测试
    chat_ok = test_deepseek_chat()
    # 运行函数调用测试
    func_ok = test_deepseek_function_call()

    # 输出最终结果
    print("\n" + "=" * 50)
    print("最终结果:")
    print(f"  对话测试: {'通过' if chat_ok else '失败'}")
    print(f"  函数调用: {'通过' if func_ok else '失败'}")
    print("=" * 50)

步骤4：运行测试验证

# ===== 运行测试脚本 =====
cd ~/openclaw
source venv/bin/activate

python3 test_model.py

# 预期输出：
# ==================================================
# DeepSeek 大模型接口测试
# ==================================================
# 模型: deepseek-chat
# API 地址: https://api.deepseek.com/v1
# --------------------------------------------------
# 发送消息:  你好，请用一句话介绍你自己。
# 等待模型回复...
# --------------------------------------------------
# 模型回复:
# 你好！我是 DeepSeek 智能助手，可以帮你回答各种问题...
# --------------------------------------------------
# Token 使用: 输入=35, 输出=48, 总计=83
# ==================================================
# 测试结果: 成功 - 大模型对话功能正常！
#
# 函数调用测试:
# 模型决定调用函数: control_robot
# 函数参数: {"action": "forward", "speed": 1.0}
# 函数调用测试: 成功！
#
# 最终结果:
#   对话测试: 通过
#   函数调用: 通过

代码详解

代码要素说明
load_dotenv()从 .env 文件加载环境变量到 os.environ，使代码能通过 os.getenv 读取
OpenAI(api_key, base_url)创建 API 客户端，通过 base_url 指向 DeepSeek 而非 OpenAI 默认地址
chat.completions.create()调用对话补全接口，传入消息列表和参数
system role系统角色消息，设定 AI 的行为准则和身份
temperature控制输出随机性，0 最确定，1 最随机
max_tokens限制模型输出的最大 Token 数量
response.choices[0].message.content从 API 响应中提取模型生成的文本内容
tools / tool_choice定义可调用的函数工具，模型可根据用户意图自动选择调用
usageAPI 返回的 Token 使用统计，包含输入/输出/总计 Token 数

易错提示

  • API Key 无效或过期：错误信息通常为 AuthenticationError。检查 Key 是否正确、账户是否有余额。注意 Key 不应包含引号或空格。
  • Base URL 格式错误：正确格式为 https://api.deepseek.com/v1 或 https://api.deepseek.com。注意不要写错域名或多余路径。
  • 网络连接超时：如果服务器无法访问外网，API 调用会超时。使用 curl https://api.deepseek.com 测试连通性。如需代理，设置 HTTP_PROXY 环境变量。
  • openai 库版本问题：确保安装的是 openai 库 v1.x 版本（pip install openai>=1.0），v0.x 的 API 调用方式完全不同。
  • 虚拟环境未激活：运行脚本前确保已激活虚拟环境 source venv/bin/activate，否则可能找不到已安装的依赖库。
  • .env 文件位置：load_dotenv() 默认在当前工作目录查找 .env 文件，确保在 ~/openclaw 目录下运行脚本。
5分

任务5：OpenClaw 调用 ROS2 控制仿真机器人（端到端闭环）

题目要求

通过 OpenClaw 智能体发送自然语言指令，调用 ROS2 接口控制 turtlesim 仿真机器人完成基本运动操作，验证具身智能平台端到端功能闭环。

思路分析

本任务是整个项目的集成验证环节，将前四个任务的所有组件串联起来：

  1. 编写 ROS2 桥接脚本：创建一个 Python 脚本，既能调用 DeepSeek 大模型理解自然语言，又能通过 rclpy 发布 ROS2 控制指令。
  2. 定义动作映射：将大模型的函数调用结果映射为具体的 Twist 消息参数（前进、后退、左转、右转、停止）。
  3. 端到端验证：启动 turtlesim 仿真器，输入自然语言指令，观察机器人是否按指令运动。

验证点：输入"让乌龟前进"→ turtlesim 中的乌龟前进；输入"停止"→ 乌龟停止运动。

步骤1：编写 ROS2 桥接脚本

创建 ~/openclaw/ros2_bridge.py，实现自然语言到 ROS2 控制的完整闭环：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
OpenClaw + ROS2 端到端控制脚本

本脚本实现具身智能闭环：
1. 接收用户的自然语言指令
2. 调用 DeepSeek 大模型理解指令意图
3. 将意图转换为 ROS2 Twist 速度消息
4. 发布到 /cmd_vel 话题控制 turtlesim 机器人
"""

import os
import sys
import json
import time
import threading

# ===== 导入 DeepSeek / OpenAI 兼容库 =====
from openai import OpenAI
from dotenv import load_dotenv

# ===== 导入 ROS2 Python 库 =====
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist

# 加载环境变量
load_dotenv()

# 获取配置
API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')
MODEL = os.getenv('DEEPSEEK_MODEL', 'deepseek-chat')


class RobotController(Node):
    """ROS2 机器人控制器节点

    负责向 /cmd_vel 话题发布速度指令，控制 turtlesim 机器人运动。
    """

    def __init__(self):
        super().__init__('robot_controller')
        # 创建 /cmd_vel 话题发布者
        self.publisher = self.create_publisher(Twist, '/cmd_vel', 10)
        self.get_logger().info('机器人控制器节点已就绪')

    def publish_velocity(self, linear_x=0.0, angular_z=0.0):
        """发布速度指令到 /cmd_vel 话题

        参数:
            linear_x: 线速度 (m/s)，正值前进，负值后退
            angular_z: 角速度 (rad/s)，正值左转，负值右转
        """
        msg = Twist()
        msg.linear.x = linear_x
        msg.angular.z = angular_z
        self.publisher.publish(msg)
        self.get_logger().info(
            f'发布速度: linear.x={linear_x:.2f}, '
            f'angular.z={angular_z:.2f}'
        )

    def stop(self):
        """停止机器人运动（发布零速度）"""
        self.publish_velocity(0.0, 0.0)
        self.get_logger().info('机器人已停止')


class EmbodiedAgent:
    """具身智能体：连接大模型与机器人控制"""

    # 定义动作与速度参数的映射表
    ACTION_MAP = {
        'forward': {'linear_x': 1.0, 'angular_z': 0.0},
        'backward': {'linear_x': -1.0, 'angular_z': 0.0},
        'turn_left': {'linear_x': 0.0, 'angular_z': 1.0},
        'turn_right': {'linear_x': 0.0, 'angular_z': -1.0},
        'stop': {'linear_x': 0.0, 'angular_z': 0.0},
    }

    # 定义可用的函数工具（告诉大模型有哪些动作可用）
    TOOLS = [
        {
            "type": "function",
            "function": {
                "name": "control_robot",
                "description": "控制机器人运动，可选择方向和速度",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "action": {
                            "type": "string",
                            "enum": ["forward", "backward",
                                       "turn_left", "turn_right",
                                       "stop"],
                            "description": "运动动作"
                        },
                        "duration": {
                            "type": "number",
                            "description": "持续时间（秒），默认2秒"
                        }
                    },
                    "required": ["action"]
                }
            }
        }
    ]

    def __init__(self, controller):
        """初始化智能体

        参数:
            controller: RobotController 实例，用于发布 ROS2 指令
        """
        self.controller = controller
        self.client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
        self.conversation_history = [
            {
                "role": "system",
                "content": (
                    "你是一个具身智能助手，能够理解用户的自然语言指令，"
                    "并通过调用 control_robot 函数控制机器人运动。"
                    "可用动作：forward（前进）、backward（后退）、"
                    "turn_left（左转）、turn_right（右转）、stop（停止）。"
                    "请根据用户指令选择最合适的动作。"
                )
            }
        ]

    def process_command(self, user_input):
        """处理用户自然语言指令

        流程：用户输入 → 大模型理解 → 函数调用 → ROS2 控制

        参数:
            user_input: 用户的自然语言指令字符串
        """
        print(f"\n用户指令: {user_input}")
        print("-" * 50)

        # 将用户输入添加到对话历史
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        try:
            # 调用大模型，让其决定调用哪个函数
            response = self.client.chat.completions.create(
                model=MODEL,
                messages=self.conversation_history,
                tools=self.TOOLS,
                tool_choice="auto",
                temperature=0.1  # 低温度，确保输出确定
            )

            message = response.choices[0].message

            # 检查模型是否决定调用函数
            if message.tool_calls:
                tool_call = message.tool_calls[0]
                function_name = tool_call.function.name
                arguments = json.loads(tool_call.function.arguments)

                print(f"模型决策: 调用函数 {function_name}")
                print(f"参数: {arguments}")

                # 执行函数调用 → 控制 ROS2 机器人
                if function_name == "control_robot":
                    self._execute_robot_control(arguments)
            else:
                # 模型直接回复文字（未调用函数）
                print(f"模型回复: {message.content}")

            # 将模型回复加入对话历史
            self.conversation_history.append(message.model_dump())

        except Exception as e:
            print(f"错误: {e}")

    def _execute_robot_control(self, arguments):
        """执行机器人控制动作

        将大模型的函数调用参数转换为 ROS2 速度指令并发布。
        """
        action = arguments.get('action', 'stop')
        duration = arguments.get('duration', 2.0)

        # 从映射表获取对应的速度参数
        if action in self.ACTION_MAP:
            params = self.ACTION_MAP[action]
            print(f"执行动作: {action}, 持续 {duration} 秒")

            # 发布速度指令
            self.controller.publish_velocity(
                linear_x=params['linear_x'],
                angular_z=params['angular_z']
            )

            # 持续运动指定时间
            time.sleep(duration)

            # 运动结束后自动停止
            self.controller.stop()
            print(f"动作 {action} 完成")
        else:
            print(f"未知动作: {action}")


def main():
    """主函数：启动具身智能控制循环"""

    # ===== 检查配置 =====
    if not API_KEY:
        print("错误：请先在 .env 文件中设置 DEEPSEEK_API_KEY")
        sys.exit(1)

    # ===== 初始化 ROS2 =====
    rclpy.init()
    controller = RobotController()

    # 在单独的线程中运行 ROS2 spin（处理回调）
    spin_thread = threading.Thread(target=rclpy.spin, args=(controller,))
    spin_thread.daemon = True
    spin_thread.start()

    # 等待 ROS2 初始化完成
    time.sleep(1)

    # 创建智能体实例
    agent = EmbodiedAgent(controller)

    # ===== 交互式命令循环 =====
    print("\n" + "=" * 50)
    print("具身智能控制平台 - 端到端验证")
    print("=" * 50)
    print("输入自然语言指令控制机器人（输入 'quit' 退出）")
    print("示例: '让乌龟前进'、'向左转'、'停止'、'后退3秒'")
    print("=" * 50)

    try:
        while True:
            # 获取用户输入
            user_input = input("\n请输入指令> ").strip()

            if not user_input:
                continue

            if user_input.lower() in ['quit', 'exit', 'q']:
                print("退出程序")
                break

            # 处理用户指令
            agent.process_command(user_input)

    except KeyboardInterrupt:
        print("\n用户中断")
    finally:
        # 清理资源
        controller.stop()
        controller.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

步骤2：端到端验证

# ===== 需要三个终端窗口 =====

# ----- 终端1：启动 turtlesim 仿真器 -----
# 激活 ROS2 环境
source /opt/ros/humble/setup.bash
# 启动 turtlesim 节点
ros2 run turtlesim turtlesim_node
# 预期：弹出图形窗口，显示一只乌龟

# ----- 终端2：启动具身智能控制脚本 -----
# 激活 ROS2 环境
source /opt/ros/humble/setup.bash
# 激活虚拟环境
cd ~/openclaw
source venv/bin/activate
# 运行控制脚本
python3 ros2_bridge.py

# 预期输出：
# ==================================================
# 具身智能控制平台 - 端到端验证
# ==================================================
# 输入自然语言指令控制机器人（输入 'quit' 退出）
# 示例: '让乌龟前进'、'向左转'、'停止'、'后退3秒'
# ==================================================

# ----- 终端3：监控 /cmd_vel 话题（可选，用于验证消息）-----
source /opt/ros/humble/setup.bash
ros2 topic echo /cmd_vel

# ===== 交互测试 =====
# 在终端2中输入以下指令测试：

# 测试1：前进
请输入指令> 让乌龟向前走
# 预期：turtlesim 窗口中乌龟向前移动2秒后停止

# 测试2：左转
请输入指令> 向左转
# 预期：乌龟逆时针旋转2秒后停止

# 测试3：右转
请输入指令> 右转
# 预期：乌龟顺时针旋转2秒后停止

# 测试4：后退
请输入指令> 后退
# 预期：乌龟向后移动2秒后停止

# 测试5：停止
请输入指令> 停止
# 预期：乌龟停止运动

# 退出
请输入指令> quit

步骤3：自动化验证脚本

创建自动化测试脚本 ~/openclaw/e2e_test.py，用于批量验证：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""端到端自动化验证脚本"""

import os
import sys
import time
import json
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# 测试用例列表：(自然语言指令, 期望动作)
TEST_CASES = [
    ("让乌龟前进", "forward"),
    ("向左转", "turn_left"),
    ("向右转", "turn_right"),
    ("后退", "backward"),
    ("停止", "stop"),
]

# 定义函数工具（与 ros2_bridge.py 一致）
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "control_robot",
            "description": "控制机器人运动",
            "parameters": {
                "type": "object",
                "properties": {
                    "action": {
                        "type": "string",
                        "enum": ["forward", "backward",
                                   "turn_left", "turn_right",
                                   "stop"],
                    }
                },
                "required": ["action"]
            }
        }
    }
]


def run_e2e_test():
    """运行端到端测试"""
    # 初始化 ROS2
    rclpy.init()
    node = Node('e2e_tester')
    publisher = node.create_publisher(Twist, '/cmd_vel', 10)

    # 初始化 DeepSeek 客户端
    client = OpenAI(
        api_key=os.getenv('DEEPSEEK_API_KEY'),
        base_url=os.getenv('DEEPSEEK_BASE_URL')
    )
    model = os.getenv('DEEPSEEK_MODEL', 'deepseek-chat')

    print("\n端到端自动化测试")
    print("=" * 60)

    passed = 0
    total = len(TEST_CASES)

    for i, (user_input, expected) in enumerate(TEST_CASES, 1):
        print(f"\n测试 {i}/{total}: '{user_input}'")
        print(f"期望动作: {expected}")

        try:
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system",
                     "content": "根据用户指令调用 control_robot 函数控制机器人。"},
                    {"role": "user",
                     "content": user_input}
                ],
                tools=TOOLS,
                tool_choice="auto",
                temperature=0.0
            )

            message = response.choices[0].message

            if message.tool_calls:
                args = json.loads(message.tool_calls[0].function.arguments)
                actual = args.get('action')

                print(f"实际动作: {actual}")

                if actual == expected:
                    print("结果: 通过")
                    passed += 1

                    # 发布对应的 ROS2 控制指令
                    action_map = {
                        'forward': (1.0, 0.0),
                        'backward': (-1.0, 0.0),
                        'turn_left': (0.0, 1.0),
                        'turn_right': (0.0, -1.0),
                        'stop': (0.0, 0.0),
                    }
                    if actual in action_map:
                        lx, az = action_map[actual]
                        msg = Twist()
                        msg.linear.x = lx
                        msg.angular.z = az
                        publisher.publish(msg)
                        time.sleep(1.5)
                        # 停止
                        stop_msg = Twist()
                        publisher.publish(stop_msg)
                else:
                    print(f"结果: 失败 (期望 {expected}, 实际 {actual})")
            else:
                print(f"结果: 失败 (模型未调用函数)")
                print(f"模型回复: {message.content}")

        except Exception as e:
            print(f"结果: 错误 - {e}")

    # 输出测试报告
    print("\n" + "=" * 60)
    print(f"测试报告: {passed}/{total} 通过 ({passed/total*100:.0f}%)")
    if passed == total:
        print("端到端功能闭环验证: 全部通过！")
    else:
        print("部分测试未通过，请检查模型配置或提示词。")
    print("=" * 60)

    # 清理
    node.destroy_node()
    rclpy.shutdown()


if __name__ == '__main__':
    run_e2e_test()

运行自动化测试：

# 确保 turtlesim 正在运行（终端1）
# 在终端2中运行测试
cd ~/openclaw
source venv/bin/activate
source /opt/ros/humble/setup.bash
python3 e2e_test.py

# 预期输出：
# 端到端自动化测试
# ============================================================
# 测试 1/5: '让乌龟前进'
# 期望动作: forward
# 实际动作: forward
# 结果: 通过
# ...
# ============================================================
# 测试报告: 5/5 通过 (100%)
# 端到端功能闭环验证: 全部通过！

代码详解

代码要素说明
RobotController(Node)ROS2 控制节点，负责发布 Twist 消息到 /cmd_vel 话题
EmbodiedAgent智能体类，封装大模型调用和机器人控制的完整逻辑
ACTION_MAP动作到速度参数的映射表，将语义动作转换为数值指令
TOOLS 定义告诉大模型有哪些函数可调用，包括参数类型和可选值
tool_choice="auto"让模型自主决定是否调用函数以及调用哪个函数
json.loads(arguments)将模型返回的 JSON 字符串参数解析为 Python 字典
threading.Thread在独立线程运行 rclpy.spin，避免阻塞用户输入循环
temperature=0.1极低温度确保模型决策稳定一致，减少随机性
conversation_history维护多轮对话历史，使模型能理解上下文

易错提示

  • ROS2 环境未激活：运行脚本前必须执行 source /opt/ros/humble/setup.bash，否则 import rclpy 会报 ModuleNotFoundError。
  • 线程安全问题：ROS2 的 spin 在独立线程运行，发布消息时要注意线程安全。本方案中 publish 操作是线程安全的。
  • 模型未调用函数：如果模型直接回复文字而不调用函数，检查系统提示词是否清晰描述了可用函数，尝试降低 temperature 参数。
  • turtlesim 未运行：如果 turtlesim_node 未启动，消息发布到 /cmd_vel 但无人接收。使用 ros2 topic list 确认 /cmd_vel 话题存在。
  • 速度值过大：Twist 消息中的速度值如果过大，乌龟可能会快速冲出屏幕边界。建议线速度不超过 2.0，角速度不超过 2.0。
  • 模型返回的 JSON 格式问题：有时模型返回的参数可能包含额外文本，使用 json.loads 解析可能失败。可添加 try-except 容错处理。
  • 多轮对话上下文丢失：确保每次将模型的回复加入 conversation_history，否则模型无法理解上下文。注意使用 message.model_dump() 正确序列化。

3.5 项目总结与拓展

项目知识点回顾

通过本模块的 5 个任务，我们完成了从零搭建具身智能平台的完整过程。以下是核心知识点回顾：

任务 核心技术 关键能力
任务1 Docker + Docker Compose 容器化部署、YAML 编排、端口映射、健康检查
任务2 ROS2 Humble + colcon 节点通信、话题发布订阅、Twist 消息、工作空间编译
任务3 Dify + DeepSeek 平台部署、模型供应商配置、应用创建、API 调用
任务4 OpenClaw + OpenAI SDK 大模型 API 对接、函数调用、配置管理
任务5 系统集成 自然语言理解、ROS2 桥接、端到端闭环验证

常见问题与排查指南

问题现象 可能原因 排查方法
docker 命令报 permission denied 用户未加入 docker 组 sudo usermod -aG docker $USER，重新登录
ros2 命令找不到 ROS2 环境未激活 source /opt/ros/humble/setup.bash
colcon build 报错找不到包 工作空间环境未激活或依赖缺失 source install/setup.bash；rosdep install 补依赖
Dify 容器启动后立即退出 内存不足或端口冲突 检查内存 ≥8GB；修改端口映射；查看容器日志
DeepSeek API 返回 401 API Key 无效或过期 检查 Key 格式，确认账户余额，重新生成 Key
turtlesim 乌龟不动 /cmd_vel 话题无消息或速度为0 ros2 topic echo /cmd_vel 检查消息；确认节点在运行
大模型不调用函数 系统提示词不清晰或温度过高 优化提示词描述；降低 temperature 至 0.1
import rclpy 报 ModuleNotFoundError 未激活 ROS2 环境或未在虚拟环境中 先 source ROS2 环境，再激活 Python 虚拟环境

核心命令速查表

类别 命令 说明
Docker docker compose up -d 后台启动编排服务
Docker docker compose ps 查看编排服务状态
Docker docker compose logs -f 实时查看服务日志
Docker docker compose down 停止并移除容器
ROS2 source /opt/ros/humble/setup.bash 激活 ROS2 环境
ROS2 colcon build 编译工作空间
ROS2 ros2 run <包> <节点> 运行节点
ROS2 ros2 topic echo /cmd_vel 查看话题消息
ROS2 ros2 topic list 列出所有话题
DeepSeek curl -X POST .../chat/completions 调用大模型对话接口

技能图谱

完成本模块后，学生应掌握以下技能层次的能力：

基础层 - 环境部署能力

  • 能够在 Linux 系统中安装和配置 Docker 容器引擎
  • 能够编写 Docker Compose YAML 编排文件
  • 能够安装 ROS2 Humble 并配置工作空间
  • 能够使用 Git 克隆开源项目并部署

进阶层 - 接口对接能力

  • 能够配置大模型 API 连接参数（API Key、Base URL）
  • 能够使用 Python SDK 调用大模型对话接口
  • 能够使用 ROS2 rclpy 编写发布者/订阅者节点
  • 能够配置 Dify 平台并创建智能体应用

高阶层 - 系统集成能力

  • 能够设计自然语言到机器人控制的端到端闭环
  • 能够使用大模型函数调用（Function Calling）能力
  • 能够编写 ROS2 桥接脚本实现多系统联动
  • 能够进行自动化测试和故障排查

拓展项目建议

在完成基础任务后，可以尝试以下拓展方向，进一步提升技能水平：

  1. 增加更多运动指令：在 ACTION_MAP 中添加"画圆""画正方形""加速""减速"等复杂动作，扩展智能体的控制能力。
  2. 接入更多传感器：使用 ROS2 仿真环境中的激光雷达话题 /scan，实现简单的避障功能，让大模型根据传感器数据决策。
  3. 多智能体协作：在 Dify 中创建多个智能体，分别负责导航、抓取、视觉识别，通过工作流编排实现协作。
  4. 语音交互：集成语音识别（STT）和语音合成（TTS）模块，实现语音控制的具身智能交互。
  5. 知识库增强：在 Dify 中上传机器人操作手册文档构建知识库，让智能体基于文档回答专业问题。
  6. 部署到真实机器人：将 turtlesim 仿真替换为真实机器人的 ROS2 驱动，实现从仿真到实物的迁移。

考核要点检查清单

在提交作品前，请对照以下检查清单确认所有任务完成：

学习反思

通过本模块的实践，你应该能够回答以下问题：

  1. Docker 容器与虚拟机有什么本质区别？为什么具身智能平台选择容器化部署？
  2. ROS2 中的话题（Topic）通信机制与传统的函数调用有什么不同？这种设计有什么优势？
  3. 大模型的函数调用（Function Calling）功能在具身智能中扮演什么角色？如果没有这个功能，端到端闭环还能实现吗？
  4. 从"自然语言指令"到"机器人运动"，数据经过了哪些转换步骤？每个步骤的输入和输出是什么？
  5. 如果要加入"避障"功能，需要在现有架构的哪个层次添加什么组件？
模块4 · 25%

智能体应用开发

基于 Dify + OpenClaw + LangChain，完成智能体对话、RAG知识库、Skills插件、Function Calling与前端开发

4.1 项目导读与学习目标

4.1.1 项目背景

具身智能（Embodied AI）是人工智能领域的前沿方向，它将大语言模型的推理能力与物理机器人的感知执行能力相结合，使机器人能够"听懂"人类自然语言指令并自主完成现实世界中的任务。例如，当你对机器人说"去仓库把红色货箱搬过来"时，机器人需要理解这句话的语义，规划运动路径，控制机械臂抓取物体，最终完成任务。

本项目以智能工厂巡检机器人为应用场景，综合运用 Dify（低代码智能体平台）、OpenClaw（智能体技能框架）和 LangChain（大模型应用开发框架）三大技术栈，构建一个完整的具身智能体应用。学生将经历从对话交互、知识问答、技能扩展到工具调用、前端可视化的全流程开发实践。

4.1.2 项目场景

某智能制造工厂部署了一批巡检机器人，需要开发一套智能体应用系统，实现以下功能：

  • 自然语言交互：操作人员通过语音或文字与机器人对话，下达巡检指令
  • 知识库问答：机器人能查询操作手册和场景规范，回答专业问题
  • 自动巡检技能：机器人能自动执行巡检路线，检测异常并报告
  • 运动控制：通过自然语言指令控制机器人移动、转向、停止
  • 可视化前端：提供直观的操作界面，展示对话、问答和机器人状态

4.1.3 技术栈概览

Dify 平台

开源的 LLM 应用开发平台，提供可视化的智能体编排、知识库管理、对话流程设计等功能，支持零代码快速搭建 AI 应用。

OpenClaw 框架

面向具身智能的技能开发框架，支持以插件形式扩展智能体能力，提供标准化的技能注册、触发和执行机制。

LangChain

大语言模型应用开发框架，提供模型集成、工具调用、链式推理等核心能力，是构建 Function Calling 应用的主流方案。

DeepSeek 大模型

国产高性能大语言模型，原生支持 Function Calling（函数调用），适合作为具身智能体的推理大脑。

ROS2

机器人操作系统第二代，提供运动控制、传感器通信、导航定位等基础设施，是机器人软件开发的事实标准。

Vue3 + FastAPI

前端采用 Vue3 构建响应式界面，后端采用 FastAPI 提供 API 服务，实现智能体应用的前后端分离架构。

4.1.4 学习目标

知识目标
  • 理解智能体（Agent）的基本概念、架构和工作原理
  • 掌握 RAG（检索增强生成）技术的核心流程与关键参数
  • 理解 Function Calling 机制及其在机器人控制中的应用
  • 了解 Dify、OpenClaw、LangChain 三大框架的定位与协作关系
  • 熟悉具身智能应用的全栈开发流程
技能目标
  • 能在 Dify 平台创建智能体应用，编写系统提示词，配置记忆窗口
  • 能创建知识库，配置分段策略与检索参数，实现 RAG 问答
  • 能基于 OpenClaw 编写 Skills 插件，实现自动巡检功能
  • 能用 LangChain 集成 DeepSeek，定义工具函数，实现 Function Calling
  • 能开发智能体前端，对接后台服务，实现可视化交互
素养目标
  • 培养系统化思维，理解智能体各模块的协作关系
  • 树立工程实践意识，注重代码规范与文档编写
  • 增强团队协作能力，体验敏捷开发流程
  • 激发创新意识，探索具身智能的更多应用可能

4.1.5 项目架构总览

本项目的整体架构如下图所示，分为五个层次：

前端展示层
Vue3 对话界面
知识库问答面板
机器人状态可视化
↓ API 接口调用
后端服务层
FastAPI 路由
会话管理
工具调度引擎
↓ 框架协作
智能体框架层
Dify 对话编排
Dify 知识库 RAG
OpenClaw Skills 插件
LangChain 工具链
↓ 模型调用
大模型层
DeepSeek（推理 + Function Calling）
Embedding 向量化模型
↓ 硬件控制
机器人执行层
ROS2 运动控制节点
传感器数据采集
导航与避障

4.2 知识准备

4.2.1 智能体（Agent）

核心概念

什么是智能体？

智能体（Agent）是一个能够感知环境、自主决策并执行动作以实现目标的智能系统。在 LLM 时代，智能体以大语言模型作为"大脑"，结合记忆、工具和规划能力，完成复杂任务。

智能体 = LLM（大脑）+ 记忆（Memory）+ 工具（Tools）+ 规划（Planning）

一个完整的智能体包含以下核心组件：

  • 大脑（LLM）：负责理解用户意图、推理决策、生成自然语言回复
  • 记忆（Memory）：存储对话历史和上下文信息，支持多轮交互
  • 工具（Tools）：外部能力接口，如搜索引擎、数据库查询、API 调用、机器人控制等
  • 规划（Planning）：将复杂任务分解为可执行的子步骤，逐步完成

生活类比：智能体就像一个新入职的工厂操作员。LLM 是他的"大脑"负责理解和思考；记忆是他的"笔记本"记录之前说过的话；工具是他的"双手"可以操作设备；规划是他"先做什么后做什么"的工作计划。

4.2.2 RAG（检索增强生成）

核心技术

RAG 检索增强生成

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将信息检索与大模型生成相结合的技术。它先从知识库中检索与用户问题相关的文档片段，再将这些片段作为上下文提供给大模型，让模型基于检索到的知识生成回答。

1 文档处理

将原始文档切分为小段文本

→
2 向量化

用 Embedding 模型将文本转为向量

→
3 存储入库

将向量存入向量数据库

→
4 检索

用户问题向量化后，检索最相关的 Top-K 片段

→
5 生成

将检索结果作为上下文，交给 LLM 生成回答

关键参数说明
参数说明推荐值
Chunk Size（分段大小）每个文本片段的字符数500-1000
Chunk Overlap（分段重叠）相邻片段的重叠字符数，保证语义连贯50-200
Top-K（检索数量）检索返回的最相似片段数量3-5
Score Threshold（分数阈值）相似度分数低于此值的片段将被过滤0.5-0.8

生活类比：RAG 就像"开卷考试"。大模型本身是学生的"知识储备"（可能不够全面），知识库是"参考教材"。考试时（用户提问），学生先翻书找到相关段落（检索），再结合自己的理解写出答案（生成）。这样既保证了答案的准确性，又能灵活应对各种问题。

4.2.3 Function Calling（函数调用）

核心机制

Function Calling 函数调用

Function Calling是大模型的一项核心能力：模型可以根据用户意图，自主判断是否需要调用外部函数（工具），并生成符合函数签名规范的调用参数。这使得大模型从"只会说话"进化为"能做事"的智能体。

1 定义工具

向模型描述可用函数的名称、参数和功能

→
2 用户提问

用户用自然语言表达需求

→
3 模型决策

模型判断是否需要调用函数，生成调用参数

→
4 执行函数

程序执行对应函数，返回结果给模型

→
5 生成回复

模型根据函数结果生成最终自然语言回复

示例：用户说"让机器人向前走2米"，模型识别到需要调用 move_forward(distance=2) 函数，生成参数 {"distance": 2}，程序执行后机器人移动，模型回复"机器人已向前移动2米"。

4.2.4 LangChain 框架

开发框架

LangChain 大模型应用框架

LangChain是一个开源的大语言模型应用开发框架，提供了模型集成、提示词管理、工具调用、记忆管理、链式推理等丰富的组件，帮助开发者快速构建复杂的 LLM 应用。

ChatModel：统一的大模型接口，支持 OpenAI、DeepSeek、通义千问等多种模型
PromptTemplate：提示词模板，支持变量注入，实现提示词复用
Tool / @tool：工具定义装饰器，将普通函数转为可被模型调用的工具
Memory：对话记忆组件，支持窗口记忆、摘要记忆等策略
AgentExecutor：智能体执行器，协调模型推理与工具调用的循环
LangGraph：基于图的状态机框架，支持复杂的多步骤工作流编排

4.2.5 Dify 平台工作流

低代码平台

Dify 智能体开发平台

Dify是一个开源的 LLM 应用开发平台，提供从原型到生产的全流程支持。它的核心优势是低代码/零代码，通过可视化界面即可完成智能体创建、知识库管理、对话编排等工作。

智能体应用 可视化创建对话型智能体，配置系统提示词、记忆窗口、工具集
知识库 上传文档自动分段、向量化，支持 Top-K 检索和混合检索策略
工作流编排 拖拽式节点编排，串联 LLM、知识检索、条件判断等节点
API 发布 一键发布为 REST API，供前端或第三方系统调用

4.2.6 OpenClaw 技能框架

技能框架

OpenClaw 智能体技能框架

OpenClaw是一个面向具身智能场景的技能开发框架。它采用"插件化"设计理念，将机器人的各项能力封装为独立的 Skills 插件，智能体可以根据任务需要动态加载和调用这些插件。

OpenClaw 插件核心结构
  • skill_config.yaml：技能配置文件，定义插件名称、描述、触发条件、参数 schema
  • handler.py：技能处理逻辑，包含执行入口函数和业务代码
  • requirements.txt：插件依赖的 Python 包列表
  • manifest.json：插件清单文件，声明插件元信息与接口规范

工作原理：当智能体收到用户指令时，OpenClaw 框架根据各插件的触发条件（关键词、意图分类等）匹配最合适的 Skill，调用其 handler 执行任务，并将执行结果返回给智能体进行后续处理。

4.2.7 ROS2 基础概念

机器人系统

ROS2 机器人操作系统

ROS2（Robot Operating System 2）是用于编写机器人软件的灵活框架，提供硬件抽象、设备驱动、进程间通信等功能。在本项目中，ROS2 负责接收来自 LangChain 的运动控制指令，驱动机器人执行动作。

节点（Node）：ROS2 中的最小执行单元，每个节点负责一个特定功能（如运动控制、传感器读取）
话题（Topic）：节点间的发布/订阅通信通道，如 /cmd_vel 用于发布速度控制指令
服务（Service）：节点间的请求/响应通信，适合一次性指令（如重置位置）
动作（Action）：长时间任务的通信机制，支持反馈和取消（如导航到目标点）

4.3 环境与数据准备

4.3.1 软件环境清单

类别软件/框架版本要求用途
平台Difyv0.6+智能体创建与知识库管理
框架OpenClawv1.0+Skills 插件开发
框架LangChainv0.2+大模型应用开发
框架LangGraphv0.1+智能体状态机编排
模型DeepSeek APIdeepseek-chat大语言模型推理
机器人ROS2Humble机器人运动控制
后端Python3.10+后端服务开发
后端FastAPIv0.100+API 服务框架
前端Node.js18+前端构建环境
前端Vue3v3.3+前端界面开发

4.3.2 Python 环境搭建

使用 conda 创建独立的项目虚拟环境，避免依赖冲突：

# 1. 创建名为 agent_dev 的虚拟环境，Python 版本为 3.10
conda create -n agent_dev python=3.10 -y

# 2. 激活虚拟环境
conda activate agent_dev

# 3. 安装 LangChain 相关依赖
pip install langchain==0.2.0 langchain-community==0.2.0 langchain-core==0.2.1

# 4. 安装 LangGraph（用于智能体状态机编排）
pip install langgraph==0.1.0

# 5. 安装 DeepSeek SDK（兼容 OpenAI 接口）
pip install openai==1.30.0

# 6. 安装 FastAPI 后端框架
pip install fastapi==0.110.0 uvicorn==0.29.0

# 7. 安装 ROS2 Python 客户端库（rclpy）
pip install rclpy

# 8. 安装 OpenClaw 框架
pip install openclaw==1.0.0

4.3.3 DeepSeek API 配置

DeepSeek 大模型通过 API 方式调用，需要先获取 API Key 并配置环境变量：

# 在终端中设置环境变量（Linux/macOS）
# 将 your_api_key 替换为你在 DeepSeek 平台申请的真实 API Key
export DEEPSEEK_API_KEY="your_api_key_here"

# 验证环境变量是否设置成功
echo $DEEPSEEK_API_KEY

# DeepSeek API 基础信息：
# 官网：https://platform.deepseek.com
# API 地址：https://api.deepseek.com/v1
# 模型名称：deepseek-chat（通用对话）/ deepseek-coder（代码生成）
注意：API Key 是敏感信息，切勿硬编码在代码中或提交到版本库。建议统一通过环境变量读取。

4.3.4 Dify 平台部署

Dify 支持 Docker Compose 一键部署，推荐在 Ubuntu 服务器上安装：

# 1. 克隆 Dify 官方仓库
git clone https://github.com/langgenius/dify.git

# 2. 进入 docker 目录
cd dify/docker

# 3. 复制环境变量配置文件
cp .env.example .env

# 4. 启动 Dify 所有服务（包含 Web 界面、API、数据库等）
docker compose up -d

# 5. 查看服务运行状态，确认所有容器均为 running
docker compose ps

# 6. 浏览器访问 Dify 管理界面
# 默认地址：http://localhost （即服务器的 80 端口）
# 首次访问需要注册管理员账号

4.3.5 项目数据准备

本项目需要准备以下文档数据，用于构建知识库：

机器人操作手册.pdf

包含机器人硬件规格、操作规范、安全须知、故障排除指南等内容，约 50 页。

工厂巡检场景规范.docx

包含巡检路线定义、检测标准、异常处理流程、报告模板等，约 20 页。

常见问题FAQ.md

包含机器人使用中的高频问题及解答，Markdown 格式，约 100 条 Q&A。

将以上文档存放在项目的 data/docs/ 目录下：

# 项目目录结构
project/
├── data/
│   └── docs/                    # 知识库文档目录
│       ├── 机器人操作手册.pdf
│       ├── 工厂巡检场景规范.docx
│       └── 常见问题FAQ.md
├── skills/                      # OpenClaw 插件目录
│   └── auto_inspection/         # 自动巡检插件
├── backend/                     # 后端服务
│   ├── main.py                  # FastAPI 入口
│   ├── agent.py                 # LangChain 智能体
│   └── ros2_tools.py            # ROS2 工具函数
├── frontend/                    # Vue3 前端
│   ├── src/
│   │   ├── App.vue
│   │   ├── components/
│   │   └── api/
│   └── package.json
└── README.md

4.4 项目实施

本节将按照竞赛样题的 5 道题目，逐步完成智能体应用开发。每个任务包含题目要求、思路分析、代码实现、代码详解和易错提示五个部分。

任务 1 5分

Dify 智能体创建与对话交互

题目要求

在 Dify 平台中创建智能体应用，编写系统提示词设定机器人助手角色与行为规范，配置对话记忆窗口实现多轮上下文管理，完成人机自然语言交互功能验证。

思路分析

步骤1

创建智能体应用：登录 Dify 平台，新建"智能体"类型应用，命名为"工厂巡检机器人助手"。

步骤2

编写系统提示词：设计提示词定义机器人助手的角色、能力边界、行为规范和回复风格。系统提示词是智能体的"人设说明书"。

步骤3

配置记忆窗口：设置对话记忆窗口大小（如保留最近 10 轮对话），使智能体能理解上下文关联的追问。

步骤4

验证多轮对话：在调试面板中测试多轮对话场景，验证上下文记忆功能。

代码实现

第1步：创建智能体应用

登录 Dify 平台后，点击"创建空白应用" → 选择"智能体"类型 → 填写应用名称和描述。

第2步：配置模型与系统提示词

在智能体配置页面，选择 LLM 模型（如 DeepSeek），在"系统提示词"区域编写以下内容：

# ========== 系统提示词（System Prompt） ==========
# 将以下内容填入 Dify 智能体的"提示词"配置区域

你是"巡检宝"，一个专业的智能工厂巡检机器人助手。

## 你的角色设定
- 你是一台部署在智能制造工厂的巡检机器人
- 你能理解自然语言指令，控制自身运动执行巡检任务
- 你熟悉工厂巡检规范和机器人操作手册
- 你的语气专业、简洁、友好

## 你的核心能力
1. 运动控制：前进、后退、左转、右转、停止
2. 巡检任务：自动巡检、定点检测、异常报告
3. 知识问答：回答关于机器人操作和巡检规范的问题
4. 状态查询：报告当前电量、位置、传感器状态

## 行为规范
- 收到运动指令时，先确认方向和距离，再执行
- 遇到不确定的问题，主动查阅知识库而非编造答案
- 检测到异常时，立即报告并建议处理方案
- 回复使用中文，长度控制在100字以内
- 如果用户的指令不清晰，主动追问细节

## 安全规则
- 禁止执行可能导致碰撞或危险的指令
- 电量低于20%时，提醒用户充电
- 巡检过程中发现异常，优先保证人员安全

第3步：配置对话记忆窗口

在 Dify 智能体配置页面的"记忆"区域进行设置：

配置项设置值说明
记忆窗口大小10保留最近 10 轮对话（即 10 对问答）
记忆类型对话窗口滑动窗口策略，超出后自动丢弃最早的对话

第4步：通过 API 验证多轮对话功能

Dify 发布应用后会生成 API 接口，我们可以用 Python 脚本验证多轮对话效果：

import requests
import json

# ========== Dify API 配置 ==========
# 在 Dify 应用页面的"访问 API"中获取以下信息
DIFY_API_URL = "http://localhost/v1/chat-messages"  # Dify API 地址
DIFY_API_KEY = "app-your-api-key-here"             # 应用的 API Key

def send_message(query, user_id, conversation_id=None):
    """
    向 Dify 智能体发送消息，并获取回复

    参数:
        query: 用户输入的消息文本
        user_id: 用户标识符（用于区分不同用户）
        conversation_id: 对话 ID，首次对话传 None，后续传入返回的 ID 实现多轮

    返回:
        reply: 智能体的回复文本
        conversation_id: 对话 ID（用于后续多轮对话）
    """
    # 构造请求头，携带 API Key 进行身份认证
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    # 构造请求体
    payload = {
        "inputs": {},                           # 变量参数（如有定义模板变量）
        "query": query,                       # 用户消息
        "response_mode": "blocking",        # 阻塞模式，等待完整回复
        "user": user_id,                       # 用户标识
    }

    # 如果有对话 ID，加入请求体以延续多轮对话
    if conversation_id:
        payload["conversation_id"] = conversation_id

    # 发送 POST 请求到 Dify API
    response = requests.post(DIFY_API_URL, headers=headers, json=payload)
    result = response.json()

    # 提取回复文本和对话 ID
    reply = result.get("answer", "未获取到回复")
    new_conversation_id = result.get("conversation_id")

    return reply, new_conversation_id


# ========== 多轮对话验证测试 ==========
if __name__ == "__main__":
    user_id = "test_user_001"
    conversation_id = None  # 首次对话，无对话 ID

    # 第1轮对话：打招呼并自我介绍
    print("=== 第1轮对话 ===")
    reply, conversation_id = send_message("你好，你是谁？", user_id, conversation_id)
    print(f"用户: 你好，你是谁？")
    print(f"机器人: {reply}")
    print(f"对话ID: {conversation_id}")

    # 第2轮对话：测试上下文记忆（不重复介绍自己）
    print("\n=== 第2轮对话（测试上下文记忆）===")
    reply, conversation_id = send_message("你能做什么？", user_id, conversation_id)
    print(f"用户: 你能做什么？")
    print(f"机器人: {reply}")

    # 第3轮对话：测试运动指令理解
    print("\n=== 第3轮对话（测试指令理解）===")
    reply, conversation_id = send_message("向前走2米", user_id, conversation_id)
    print(f"用户: 向前走2米")
    print(f"机器人: {reply}")

    # 第4轮对话：测试上下文追问（"再走1米"是否理解为继续向前）
    print("\n=== 第4轮对话（测试上下文追问）===")
    reply, conversation_id = send_message("再走1米", user_id, conversation_id)
    print(f"用户: 再走1米")
    print(f"机器人: {reply}")

代码详解

1. 系统提示词设计要点

系统提示词采用"模块化结构"编写，分为角色设定、核心能力、行为规范、安全规则四个模块。使用 Markdown 标题和列表让 LLM 更好地理解结构。关键技巧：用具体场景描述代替抽象指令（如"回复控制在100字以内"比"简洁回复"更明确）。

2. 对话记忆窗口机制

Dify 的记忆窗口采用"滑动窗口"策略：保留最近 N 轮对话。当第 N+1 轮对话到来时，最早的一轮被丢弃。通过 conversation_id 参数关联同一对话的多次消息，实现上下文延续。窗口大小设为 10 是平衡 token 消耗和上下文完整性的常用值。

3. API 调用流程

send_message 函数封装了 Dify Chat API 的调用逻辑。首次调用 conversation_id=None，Dify 会创建新对话并返回 ID；后续调用传入该 ID，Dify 自动加载历史消息作为上下文。response_mode="blocking" 表示同步等待完整回复（适合功能验证），生产环境可用 "streaming" 实现流式输出。

4. 多轮对话验证逻辑

测试脚本设计了4轮递进式对话：第1轮建立角色认知，第2轮测试能力介绍，第3轮测试指令理解，第4轮测试上下文追问（"再走1米"应理解为"继续向前走1米"）。如果第4轮能正确理解上下文，说明记忆窗口配置生效。

易错提示

!

API Key 错误：Dify 的 API Key 格式为 app-xxxxx，注意不是用户登录密码。在应用页面的"访问 API → API 密钥"中获取。

!

conversation_id 丢失：如果每次调用都不传 conversation_id，每条消息都会被当作新对话，无法实现多轮上下文。务必保存并传递返回的 conversation_id。

!

记忆窗口过大：窗口设置过大会导致 token 消耗激增、响应变慢甚至超出模型上下文限制。建议不超过 20 轮。

!

系统提示词过长：提示词应精炼有效，过长的提示词会挤占对话 token 空间。建议控制在 500 字以内。

任务 2 5分

Dify 知识库创建与 RAG 增强问答

题目要求

在 Dify 平台中创建知识库，上传机器人操作手册与场景规范文档，配置文本分段策略与 Top-K 检索参数，将知识库关联至智能体应用，验证 RAG 增强问答功能。

思路分析

步骤1

创建知识库：在 Dify 的"知识库"页面新建知识库，上传 PDF 和 Word 文档。

步骤2

配置分段策略：设置文本分段大小和重叠区域，平衡检索精度和上下文完整性。

步骤3

配置检索参数：设置 Top-K 值和相似度阈值，优化检索效果。

步骤4

关联智能体：在任务1创建的智能体应用中关联知识库，验证 RAG 问答。

代码实现

第1步：创建知识库并上传文档

在 Dify 平台操作："知识库" → "创建知识库" → 上传以下文件：

  • 机器人操作手册.pdf
  • 工厂巡检场景规范.docx
  • 常见问题FAQ.md

第2步：配置文本分段策略

配置项设置值说明
分段方式自动分段Dify 自动识别段落边界进行切分
分段长度上限500每个文本片段最多 500 个 token
分段重叠长度50相邻片段重叠 50 个 token，保证语义连贯
清洗规则开启去除多余空格、换行符等无关字符

第3步：配置索引与检索参数

配置项设置值说明
索引方式高质量索引使用 Embedding 向量索引（比经济索引精度更高）
Embedding 模型bge-large-zh中文优化的向量化模型
检索模式向量检索基于语义相似度的向量检索
Top-K3检索返回最相关的 3 个文本片段
Score 阈值0.5相似度分数低于 0.5 的片段将被过滤
Rerank开启对检索结果二次排序，提升精度

第4步：关联知识库到智能体应用

在任务1创建的智能体应用的配置页面："上下文" → "添加知识库" → 选择刚创建的知识库。

第5步：通过 API 验证 RAG 问答

import requests
import json

# ========== Dify API 配置（与任务1相同）==========
DIFY_API_URL = "http://localhost/v1/chat-messages"
DIFY_API_KEY = "app-your-api-key-here"

def chat_with_rag(query, user_id, conversation_id=None):
    """
    向关联了知识库的智能体发送问题，验证 RAG 增强问答

    参数:
        query: 用户的提问内容
        user_id: 用户标识符
        conversation_id: 对话 ID（多轮对话用）

    返回:
        answer: 智能体的回答
        retrieved_docs: 检索到的知识库片段信息
        conversation_id: 对话 ID
    """
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "blocking",
        "user": user_id,
    }
    if conversation_id:
        payload["conversation_id"] = conversation_id

    response = requests.post(DIFY_API_URL, headers=headers, json=payload)
    result = response.json()

    answer = result.get("answer", "未获取到回复")
    # Dify 返回的 metadata 中包含检索到的知识库片段信息
    metadata = result.get("metadata", {})
    retrieved_docs = metadata.get("retriever_resources", [])
    new_conversation_id = result.get("conversation_id")

    return answer, retrieved_docs, new_conversation_id


# ========== RAG 问答验证测试 ==========
if __name__ == "__main__":
    user_id = "test_user_002"
    conversation_id = None

    # 测试用例1：查询机器人操作规范
    print("=== 测试1：查询操作规范 ===")
    q1 = "机器人启动前需要做哪些检查？"
    answer, docs, conversation_id = chat_with_rag(q1, user_id, conversation_id)
    print(f"问题: {q1}")
    print(f"回答: {answer}")
    print(f"检索到 {len(docs)} 个知识片段:")
    for i, doc in enumerate(docs):
        print(f"  片段{i+1} (相似度: {doc.get('score', 'N/A')}): "
              f"{doc.get('content', '')[:80]}...")

    # 测试用例2：查询巡检路线
    print("\n=== 测试2：查询巡检路线 ===")
    q2 = "A区巡检路线包含哪些检测点？"
    answer, docs, conversation_id = chat_with_rag(q2, user_id, conversation_id)
    print(f"问题: {q2}")
    print(f"回答: {answer}")

    # 测试用例3：查询故障处理流程
    print("\n=== 测试3：查询故障处理 ===")
    q3 = "如果巡检过程中发现温度异常，应该怎么处理？"
    answer, docs, conversation_id = chat_with_rag(q3, user_id, conversation_id)
    print(f"问题: {q3}")
    print(f"回答: {answer}")

    # 测试用例4：测试知识库未覆盖的问题（验证不编造答案）
    print("\n=== 测试4：测试未覆盖问题 ===")
    q4 = "如何用Python写一个网页爬虫？"
    answer, docs, conversation_id = chat_with_rag(q4, user_id, conversation_id)
    print(f"问题: {q4}")
    print(f"回答: {answer}")
    print(f"检索到 {len(docs)} 个知识片段（应为0或低相似度）")

代码详解

1. 文本分段策略原理

文档不能整篇交给模型处理（超出 token 限制且检索不精确），需要切分为小段。分段长度 500 token 保证每段包含完整语义单元；重叠 50 token 避免在段落边界切断关键信息（如一个步骤的描述横跨两个片段时，重叠区域能保证至少一个片段包含完整描述）。

2. Top-K 与 Score 阈值

Top-K=3 表示检索最相关的 3 个片段作为上下文。K 值过大可能引入不相关内容干扰回答，过小可能遗漏关键信息。Score 阈值 0.5 过滤掉相似度太低的片段，避免模型基于无关内容"胡说八道"。

3. Rerank 重排序

向量检索速度快但精度有限。Rerank 模型对初步检索的 Top-K 结果进行二次精细排序，将最相关的片段排在前面。开启 Rerank 可显著提升 RAG 回答质量，代价是增加少量延迟。

4. metadata 中的检索信息

Dify API 返回的 metadata.retriever_resources 包含每个检索片段的内容、相似度分数和来源文档。通过分析这些信息可以判断检索是否准确，是调试 RAG 效果的关键依据。测试4验证了当问题不在知识库范围内时，检索片段应为空或低相似度，模型应表示"无法回答"而非编造。

易错提示

!

文档格式不支持：Dify 支持 PDF、Word、Markdown、TXT 等格式。扫描版 PDF（图片型）无法直接提取文字，需先用 OCR 工具转换为文本型 PDF。

!

分段过大导致检索不准：如果分段设置为 2000+ token，单个片段包含太多信息，向量检索的语义匹配精度会下降。建议分段长度 300-800 token。

!

忘记关联知识库：创建了知识库但忘记在智能体应用的"上下文"中关联，会导致 RAG 功能不生效。关联后可在调试面板查看检索日志确认。

!

Embedding 模型不匹配：知识库和查询必须使用相同的 Embedding 模型。如果中途更换模型，需要重新索引所有文档。

任务 3 5分

OpenClaw Skills 插件开发——自动巡检

题目要求

基于 OpenClaw 框架编写 Skills 插件，实现自动巡检功能，配置插件触发条件与执行逻辑，验证插件可被智能体正确调用执行。

思路分析

步骤1

设计插件结构：创建 OpenClaw 插件目录，包含配置文件、处理逻辑和依赖清单。

步骤2

编写技能配置：定义插件名称、描述、触发关键词和参数 schema，让智能体知道何时调用此插件。

步骤3

实现巡检逻辑：编写 handler 函数，模拟巡检路线执行、传感器数据采集和异常检测。

步骤4

注册并测试：将插件注册到 OpenClaw 框架，通过智能体对话验证触发和执行。

代码实现

第1步：创建插件目录结构

# 插件目录结构
skills/auto_inspection/
├── skill_config.yaml     # 技能配置文件
├── handler.py            # 技能处理逻辑
├── manifest.json         # 插件清单
└── requirements.txt      # 依赖列表

第2步：编写技能配置文件 skill_config.yaml

# ========== skill_config.yaml ==========
# 技能配置文件：定义插件的元信息和触发条件

# 技能名称（全局唯一标识）
name: "auto_inspection"

# 技能显示名称
display_name: "自动巡检"

# 技能描述（供智能体理解何时使用此技能）
description: "执行工厂自动巡检任务，按照预设路线进行巡检，检测环境异常（温度、烟雾、异响等）并生成巡检报告"

# 触发条件配置
trigger:
  # 关键词触发：当用户消息包含以下关键词时激活
  keywords:
    - "巡检"
    - "自动巡检"
    - "开始巡检"
    - "巡逻"
    - "检查设备"

  # 意图触发：当用户意图分类匹配时激活
  intents:
    - "inspection_request"
    - "patrol_command"

# 参数 schema（定义技能可接受的参数）
parameters:
  - name: "area"
    type: "string"
    description: "巡检区域，如 A区、B区、C区"
    required: false
    default: "A区"

  - name: "mode"
    type: "string"
    description: "巡检模式：quick(快速)/full(全面)"
    required: false
    default: "full"

# 执行超时时间（秒）
timeout: 120

第3步：编写插件清单 manifest.json

{
    "name": "auto_inspection",
    "version": "1.0.0",
    "description": "自动巡检技能插件",
    "author": "Agent Dev Team",
    "entry_point": "handler.py",
    "handler_function": "execute_inspection",
    "python_version": ">=3.10"
}

第4步：编写巡检处理逻辑 handler.py

import json
import time
import random
from datetime import datetime
from typing import Dict, Any

# ========== 巡检路线定义 ==========
# 每个区域定义不同的巡检检测点
INSPECTION_ROUTES = {
    "A区": [
        {"point_id": "A-01", "name": "配电室", "sensors": ["temperature", "smoke"]},
        {"point_id": "A-02", "name": "原料仓库", "sensors": ["temperature", "humidity"]},
        {"point_id": "A-03", "name": "生产车间1", "sensors": ["temperature", "noise", "smoke"]},
    ],
    "B区": [
        {"point_id": "B-01", "name": "成品仓库", "sensors": ["temperature", "humidity"]},
        {"point_id": "B-02", "name": "包装车间", "sensors": ["temperature", "noise"]},
    ],
    "C区": [
        {"point_id": "C-01", "name": "机房", "sensors": ["temperature", "smoke", "humidity"]},
    ]
}

# ========== 传感器阈值定义 ==========
# 当传感器读数超过阈值时判定为异常
SENSOR_THRESHOLDS = {
    "temperature": {"max": 35.0, "unit": "℃"},      # 温度上限 35℃
    "humidity": {"max": 70.0, "unit": "%"},        # 湿度上限 70%
    "smoke": {"max": 50.0, "unit": "ppm"},         # 烟雾浓度上限 50ppm
    "noise": {"max": 85.0, "unit": "dB"},          # 噪音上限 85dB
}


def read_sensor(sensor_type: str) -> float:
    """
    模拟读取传感器数据
    （实际项目中应调用 ROS2 话题或硬件接口获取真实数据）

    参数:
        sensor_type: 传感器类型（temperature/humidity/smoke/noise）

    返回:
        模拟的传感器读数
    """
    # 根据传感器类型生成合理的随机模拟值
    if sensor_type == "temperature":
        return round(random.uniform(20.0, 38.0), 1)
    elif sensor_type == "humidity":
        return round(random.uniform(30.0, 80.0), 1)
    elif sensor_type == "smoke":
        return round(random.uniform(0.0, 60.0), 1)
    elif sensor_type == "noise":
        return round(random.uniform(50.0, 95.0), 1)
    return 0.0


def check_anomaly(sensor_type: str, value: float) -> bool:
    """
    检查传感器读数是否异常

    参数:
        sensor_type: 传感器类型
        value: 传感器读数

    返回:
        True 表示异常，False 表示正常
    """
    threshold = SENSOR_THRESHOLDS.get(sensor_type)
    if threshold and value > threshold["max"]:
        return True
    return False


def execute_inspection(params: Dict[str, Any]) -> Dict[str, Any]:
    """
    自动巡检技能的主执行函数（OpenClaw 插件入口）

    参数:
        params: 包含 area（巡检区域）和 mode（巡检模式）的字典

    返回:
        巡检结果字典，包含巡检报告、异常列表等
    """
    # 解析参数，使用默认值兜底
    area = params.get("area", "A区")
    mode = params.get("mode", "full")

    # 获取该区域的巡检路线
    route = INSPECTION_ROUTES.get(area, [])
    if not route:
        return {
            "success": False,
            "message": f"未找到区域 {area} 的巡检路线"
        }

    # 初始化巡检结果
    inspection_results = []
    anomalies = []
    start_time = datetime.now()

    print(f"[巡检启动] 区域: {area}, 模式: {mode}, 开始时间: {start_time}")

    # 遍历巡检路线中的每个检测点
    for point in route:
        point_id = point["point_id"]
        point_name = point["name"]
        sensors = point["sensors"]

        print(f"  [巡检中] 到达检测点 {point_id} - {point_name}")

        # 在每个检测点读取所有传感器数据
        sensor_readings = {}
        point_anomalies = []

        for sensor_type in sensors:
            # 读取传感器数据
            value = read_sensor(sensor_type)
            sensor_readings[sensor_type] = value

            # 检查是否异常
            if check_anomaly(sensor_type, value):
                threshold = SENSOR_THRESHOLDS[sensor_type]
                anomaly_info = {
                    "point_id": point_id,
                    "point_name": point_name,
                    "sensor": sensor_type,
                    "value": value,
                    "threshold": threshold["max"],
                    "unit": threshold["unit"],
                    "severity": "warning" if value < threshold["max"] * 1.2 else "critical"
                }
                point_anomalies.append(anomaly_info)
                anomalies.append(anomaly_info)
                print(f"    [异常] {sensor_type}={value}{threshold['unit']} "
                      f"(阈值: {threshold['max']}{threshold['unit']})")
            else:
                print(f"    [正常] {sensor_type}={value}")

        # 记录该检测点的巡检结果
        inspection_results.append({
            "point_id": point_id,
            "point_name": point_name,
            "readings": sensor_readings,
            "anomalies": point_anomalies,
            "status": "abnormal" if point_anomalies else "normal"
        })

        # 模拟移动到下一个检测点的时间（快速模式间隔更短）
        time.sleep(0.5 if mode == "quick" else 1.0)

    # 生成巡检报告
    end_time = datetime.now()
    duration = (end_time - start_time).total_seconds()

    report = {
        "success": True,
        "area": area,
        "mode": mode,
        "start_time": start_time.strftime("%Y-%m-%d %H:%M:%S"),
        "end_time": end_time.strftime("%Y-%m-%d %H:%M:%S"),
        "duration_seconds": round(duration, 1),
        "total_points": len(route),
        "normal_points": len([r for r in inspection_results if r["status"] == "normal"]),
        "abnormal_points": len([r for r in inspection_results if r["status"] == "abnormal"]),
        "total_anomalies": len(anomalies),
        "anomalies": anomalies,
        "details": inspection_results,
        "summary": generate_summary(area, inspection_results, anomalies)
    }

    print(f"\n[巡检完成] 总检测点: {report['total_points']}, "
          f"异常: {report['total_anomalies']}, 耗时: {duration}s")

    return report


def generate_summary(area, results, anomalies) -> str:
    """
    生成巡检报告文字摘要（供智能体转述给用户）

    参数:
        area: 巡检区域
        results: 巡检结果列表
        anomalies: 异常列表

    返回:
        巡检报告的文字摘要
    """
    total = len(results)
    normal = len([r for r in results if r["status"] == "normal"])
    abnormal = total - normal

    summary = f"{area}巡检完成。共检查{total}个检测点，其中{normal}个正常，{abnormal}个异常。"

    if anomalies:
        summary += "发现以下异常："
        for a in anomalies:
            summary += f"\n- {a['point_name']}({a['point_id']})的{a['sensor']}" \
                      f"读数为{a['value']}{a['unit']}，超过阈值{a['threshold']}{a['unit']}，" \
                      f"严重程度：{a['severity']}。"
        summary += "\n建议立即处理异常项，确保生产安全。"
    else:
        summary += "所有检测点正常，未发现异常。"

    return summary


# ========== 本地测试入口 ==========
if __name__ == "__main__":
    # 模拟智能体调用插件
    test_params = {"area": "A区", "mode": "full"}
    result = execute_inspection(test_params)
    print("\n========== 巡检报告摘要 ==========")
    print(result["summary"])

第5步：注册插件到 OpenClaw 框架并测试调用

from openclaw import SkillRegistry, Agent
import os

# ========== 注册 OpenClaw 插件 ==========

# 1. 创建技能注册器实例
registry = SkillRegistry()

# 2. 注册自动巡检插件
#    指定插件目录路径，OpenClaw 会自动读取 config 和 handler
plugin_path = os.path.join(os.path.dirname(__file__), "skills/auto_inspection")
registry.register(plugin_path)

# 3. 创建智能体并加载已注册的技能
agent = Agent(
    name="巡检机器人助手",
    model="deepseek-chat",
    skills=registry
)

# ========== 测试插件调用 ==========
if __name__ == "__main__":
    # 测试1：通过自然语言触发自动巡检插件
    print("=== 测试1：自然语言触发 ===")
    user_input = "请对A区进行全面巡检"
    print(f"用户: {user_input}")
    response = agent.chat(user_input)
    print(f"机器人: {response}")

    # 测试2：使用默认参数触发
    print("\n=== 测试2：默认参数触发 ===")
    user_input2 = "开始巡检"
    print(f"用户: {user_input2}")
    response2 = agent.chat(user_input2)
    print(f"机器人: {response2}")

    # 测试3：快速巡检模式
    print("\n=== 测试3：快速巡检模式 ===")
    user_input3 = "快速巡检B区"
    print(f"用户: {user_input3}")
    response3 = agent.chat(user_input3)
    print(f"机器人: {response3}")

代码详解

1. skill_config.yaml 配置文件

配置文件是插件与智能体之间的"接口契约"。trigger.keywords 定义了触发关键词，当用户消息包含"巡检""巡逻"等词时，OpenClaw 框架会优先匹配此插件。parameters 定义了插件接受的参数及其类型和默认值，智能体会从用户消息中提取参数值传入。

2. execute_inspection 主函数

这是插件的入口函数，接收 params 字典参数。函数执行流程：解析参数 → 获取巡检路线 → 遍历检测点 → 读取传感器 → 检测异常 → 生成报告。函数返回结构化字典，包含完整的巡检结果和文字摘要，供智能体转述给用户。

3. 异常检测逻辑

check_anomaly 函数将传感器读数与预定义阈值比较。超过阈值则标记为异常，并根据超标程度判定严重级别（warning/critical）。generate_summary 函数将结构化数据转为自然语言摘要，使智能体能直接向用户播报巡检结果。

4. 插件注册与调用

SkillRegistry 负责管理所有已注册的插件。注册时 OpenClaw 读取配置文件和 handler，建立关键词/意图到插件的映射。Agent.chat() 方法接收用户消息，自动匹配最合适的插件，提取参数，调用 handler 执行，最后将结果整合为自然语言回复。

易错提示

!

handler 函数名不匹配：manifest.json 中的 handler_function 必须与 handler.py 中的函数名完全一致（本例为 execute_inspection），否则 OpenClaw 找不到入口函数。

!

触发关键词覆盖不足：如果用户说"检查一下设备"但关键词列表中没有"检查设备"，插件不会被触发。建议覆盖多种表达方式。

!

插件执行超时：巡检逻辑如果涉及真实的机器人移动和传感器等待，可能执行时间较长。需合理设置 timeout，避免智能体长时间无响应。

!

返回值格式不规范：插件返回值应为字典格式，包含 success 字段标识执行状态。不规范的结构会导致智能体无法正确解析结果。

任务 4 5分

LangChain 集成 DeepSeek 实现 Function Calling 控制 ROS2

题目要求

使用 LangChain 框架集成 DeepSeek 大模型，定义 ROS2 运动控制工具函数，通过 Function Calling 实现自然语言指令到机器人动作的转换与执行。

思路分析

步骤1

集成 DeepSeek：使用 LangChain 的 ChatOpenAI 接口（DeepSeek 兼容 OpenAI API 格式）连接 DeepSeek 模型。

步骤2

定义 ROS2 工具函数：使用 @tool 装饰器将机器人运动控制函数封装为 LangChain 工具，添加详细的函数描述和参数说明。

步骤3

构建智能体：使用 LangChain 的 create_tool_calling_agent 创建支持 Function Calling 的智能体，连接模型、工具和提示词。

步骤4

测试指令转换：输入自然语言指令，验证模型能正确识别意图、调用工具、执行 ROS2 命令并返回结果。

代码实现

第1步：定义 ROS2 运动控制工具函数（ros2_tools.py）

import math
import time
from langchain_core.tools import tool
from typing import Optional

# ========== ROS2 运动控制工具函数 ==========
# 使用 @tool 装饰器将普通函数注册为 LangChain 工具
# 函数的 docstring 会作为工具描述发送给大模型，帮助模型理解何时调用此工具

# ---------- 全局状态模拟 ----------
# 在实际项目中，这些状态通过 ROS2 话题/服务获取
robot_state = {
    "position": {"x": 0.0, "y": 0.0, "theta": 0.0},
    "battery": 85,
    "speed": 0.0,
    "status": "idle"  # idle / moving / stopped
}


@tool
def move_forward(distance: float) -> str:
    """
    控制机器人向前直线移动指定距离。

    当用户要求机器人前进、向前走、向前移动时使用此工具。

    Args:
        distance: 移动距离，单位为米，范围为 0.1 到 50.0

    Returns:
        执行结果描述字符串
    """
    # 参数校验
    if distance < 0.1 or distance > 50.0:
        return f"错误：移动距离必须在0.1-50.0米之间，当前值: {distance}米"

    # 电量检查
    if robot_state["battery"] < 10:
        return "错误：电量不足（低于10%），请先充电"

    # 模拟 ROS2 发布速度指令到 /cmd_vel 话题
    # 实际代码：ros2 publisher 发布 geometry_msgs/Twist 消息
    robot_state["status"] = "moving"
    robot_state["speed"] = 0.5  # 移动速度 0.5 m/s

    # 更新位置（基于当前朝向计算新坐标）
    theta = robot_state["position"]["theta"]
    robot_state["position"]["x"] += distance * math.cos(theta)
    robot_state["position"]["y"] += distance * math.sin(theta)

    # 模拟移动耗时
    time.sleep(0.3)

    # 更新状态
    robot_state["status"] = "idle"
    robot_state["speed"] = 0.0
    robot_state["battery"] -= int(distance * 0.5)  # 每米消耗0.5%电量

    pos = robot_state["position"]
    return f"机器人已向前移动{distance}米。当前位置: x={pos['x']:.1f}, y={pos['y']:.1f}, " \
           f"电量: {robot_state['battery']}%"


@tool
def move_backward(distance: float) -> str:
    """
    控制机器人向后直线移动指定距离。

    当用户要求机器人后退、向后走、倒退时使用此工具。

    Args:
        distance: 后退距离，单位为米，范围为 0.1 到 20.0

    Returns:
        执行结果描述字符串
    """
    if distance < 0.1 or distance > 20.0:
        return f"错误：后退距离必须在0.1-20.0米之间，当前值: {distance}米"

    if robot_state["battery"] < 10:
        return "错误：电量不足（低于10%），请先充电"

    robot_state["status"] = "moving"
    robot_state["speed"] = 0.3  # 后退速度较慢

    # 向后移动：反方向计算
    theta = robot_state["position"]["theta"]
    robot_state["position"]["x"] -= distance * math.cos(theta)
    robot_state["position"]["y"] -= distance * math.sin(theta)

    time.sleep(0.3)
    robot_state["status"] = "idle"
    robot_state["speed"] = 0.0
    robot_state["battery"] -= int(distance * 0.5)

    pos = robot_state["position"]
    return f"机器人已向后移动{distance}米。当前位置: x={pos['x']:.1f}, y={pos['y']:.1f}, " \
           f"电量: {robot_state['battery']}%"


@tool
def turn_left(angle: float) -> str:
    """
    控制机器人向左旋转指定角度。

    当用户要求机器人左转、向左转、逆时针旋转时使用此工具。

    Args:
        angle: 旋转角度，单位为度，范围为 1 到 360

    Returns:
        执行结果描述字符串
    """
    if angle < 1 or angle > 360:
        return f"错误：旋转角度必须在1-360度之间，当前值: {angle}度"

    if robot_state["battery"] < 10:
        return "错误：电量不足（低于10%），请先充电"

    robot_state["status"] = "moving"

    # 更新朝向角度（左转为逆时针，角度增加）
    robot_state["position"]["theta"] += math.radians(angle)
    # 标准化角度到 [0, 2π) 范围
    robot_state["position"]["theta"] %= 2 * math.pi

    time.sleep(0.2)
    robot_state["status"] = "idle"
    robot_state["battery"] -= 1

    theta_deg = math.degrees(robot_state["position"]["theta"])
    return f"机器人已向左旋转{angle}度。当前朝向: {theta_deg:.0f}度, 电量: {robot_state['battery']}%"


@tool
def turn_right(angle: float) -> str:
    """
    控制机器人向右旋转指定角度。

    当用户要求机器人右转、向右转、顺时针旋转时使用此工具。

    Args:
        angle: 旋转角度，单位为度，范围为 1 到 360

    Returns:
        执行结果描述字符串
    """
    if angle < 1 or angle > 360:
        return f"错误：旋转角度必须在1-360度之间，当前值: {angle}度"

    if robot_state["battery"] < 10:
        return "错误：电量不足（低于10%），请先充电"

    robot_state["status"] = "moving"

    # 更新朝向角度（右转为顺时针，角度减少）
    robot_state["position"]["theta"] -= math.radians(angle)
    robot_state["position"]["theta"] %= 2 * math.pi

    time.sleep(0.2)
    robot_state["status"] = "idle"
    robot_state["battery"] -= 1

    theta_deg = math.degrees(robot_state["position"]["theta"])
    return f"机器人已向右旋转{angle}度。当前朝向: {theta_deg:.0f}度, 电量: {robot_state['battery']}%"


@tool
def stop_robot() -> str:
    """
    紧急停止机器人运动。

    当用户要求机器人停止、停下、刹车、紧急停止时使用此工具。
    此工具不需要任何参数。

    Returns:
        执行结果描述字符串
    """
    robot_state["status"] = "stopped"
    robot_state["speed"] = 0.0

    # 实际代码：发布零速度到 /cmd_vel 话题
    # twist = Twist(); twist.linear.x = 0.0; twist.angular.z = 0.0
    # publisher.publish(twist)

    return "机器人已紧急停止。所有运动已中断。"


@tool
def get_robot_status() -> str:
    """
    查询机器人当前状态信息。

    当用户询问机器人状态、当前位置、电量、速度等信息时使用此工具。
    此工具不需要任何参数。

    Returns:
        机器人状态信息字符串
    """
    pos = robot_state["position"]
    theta_deg = math.degrees(pos["theta"])
    return (f"机器人当前状态:\n"
            f"  位置: x={pos['x']:.1f}m, y={pos['y']:.1f}m\n"
            f"  朝向: {theta_deg:.0f}度\n"
            f"  速度: {robot_state['speed']} m/s\n"
            f"  电量: {robot_state['battery']}%\n"
            f"  状态: {robot_state['status']}")


# 导出所有工具列表（供智能体使用）
all_tools = [move_forward, move_backward, turn_left, turn_right, stop_robot, get_robot_status]

第2步：创建 LangChain 智能体集成 DeepSeek（agent.py）

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.agents import create_tool_calling_agent, AgentExecutor

# 导入任务第1步定义的 ROS2 工具函数
from ros2_tools import all_tools

# ========== 1. 初始化 DeepSeek 大模型 ==========
# DeepSeek 兼容 OpenAI API 格式，使用 ChatOpenAI 接口连接
llm = ChatOpenAI(
    model="deepseek-chat",                              # 使用 DeepSeek 对话模型
    api_key=os.environ.get("DEEPSEEK_API_KEY"),          # 从环境变量读取 API Key
    base_url="https://api.deepseek.com/v1",               # DeepSeek API 地址
    temperature=0.1,                                    # 低温度保证输出稳定
    max_tokens=1024,                                   # 最大输出 token 数
)

# ========== 2. 定义智能体提示词模板 ==========
# 提示词模板定义了智能体的角色和行为规范
prompt = ChatPromptTemplate.from_messages([
    # 系统消息：定义机器人助手的角色和行为规范
    ("system", """你是一个智能工厂巡检机器人控制助手。

你的职责是理解用户的自然语言指令，通过调用工具函数控制机器人运动。

你可以使用以下工具：
- move_forward: 控制机器人前进
- move_backward: 控制机器人后退
- turn_left: 控制机器人左转
- turn_right: 控制机器人右转
- stop_robot: 紧急停止
- get_robot_status: 查询机器人状态

行为规范：
1. 收到运动指令后，先判断方向和距离/角度，再调用对应工具
2. 如果用户没有明确距离，默认移动1米；没有明确角度，默认转90度
3. 每次执行完动作后，简要告知用户执行结果
4. 如果指令不明确，主动询问用户具体参数
5. 涉及安全的指令（如紧急停止）立即执行，无需确认"""),

    # 对话历史占位符：存放之前的对话消息
    MessagesPlaceholder(variable_name="chat_history"),

    # 用户消息占位符
    ("human", "{input}"),

    # Agent 轨迹占位符：存放模型的中间推理步骤
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# ========== 3. 创建 Function Calling 智能体 ==========
# create_tool_calling_agent 会将工具定义发送给模型，
# 模型在推理时可以决定是否调用工具以及传入什么参数
agent = create_tool_calling_agent(
    llm=llm,          # 大模型
    tools=all_tools,   # 可用工具列表
    prompt=prompt      # 提示词模板
)

# ========== 4. 创建智能体执行器 ==========
# AgentExecutor 负责协调"模型推理→工具调用→结果反馈→模型再推理"的循环
agent_executor = AgentExecutor(
    agent=agent,
    tools=all_tools,
    verbose=True,          # 打印详细执行过程（调试用）
    max_iterations=5,      # 最大迭代次数，防止无限循环
    handle_parsing_errors=True  # 自动处理解析错误
)


# ========== 5. 对话函数（支持多轮交互）==========
def chat_with_robot(user_input: str, chat_history: list) -> tuple:
    """
    与机器人智能体对话

    参数:
        user_input: 用户输入的自然语言指令
        chat_history: 对话历史列表（HumanMessage/AIMessage 对象列表）

    返回:
        response: 智能体的回复文本
        chat_history: 更新后的对话历史
    """
    # 调用 AgentExecutor 执行
    result = agent_executor.invoke({
        "input": user_input,
        "chat_history": chat_history
    })

    response = result["output"]

    # 更新对话历史
    chat_history.append(HumanMessage(content=user_input))
    chat_history.append(AIMessage(content=response))

    return response, chat_history


# ========== 6. 测试 Function Calling 功能 ==========
if __name__ == "__main__":
    chat_history = []  # 初始化对话历史

    # 测试1：基本前进指令
    print("=== 测试1：前进指令 ===")
    response, chat_history = chat_with_robot("向前走3米", chat_history)
    print(f"用户: 向前走3米")
    print(f"机器人: {response}\n")

    # 测试2：左转指令
    print("=== 测试2：左转指令 ===")
    response, chat_history = chat_with_robot("向左转90度", chat_history)
    print(f"用户: 向左转90度")
    print(f"机器人: {response}\n")

    # 测试3：上下文关联（"再走2米"应理解为继续向前）
    print("=== 测试3：上下文关联 ===")
    response, chat_history = chat_with_robot("再走2米", chat_history)
    print(f"用户: 再走2米")
    print(f"机器人: {response}\n")

    # 测试4：查询状态
    print("=== 测试4：查询状态 ===")
    response, chat_history = chat_with_robot("我现在在哪里？电量还有多少？", chat_history)
    print(f"用户: 我现在在哪里？电量还有多少？")
    print(f"机器人: {response}\n")

    # 测试5：紧急停止
    print("=== 测试5：紧急停止 ===")
    response, chat_history = chat_with_robot("紧急停止！", chat_history)
    print(f"用户: 紧急停止！")
    print(f"机器人: {response}\n")

代码详解

1. @tool 装饰器与工具描述

@tool 装饰器将普通 Python 函数转为 LangChain 工具。函数的 docstring 至关重要——它会作为工具描述发送给大模型，模型依据描述决定何时调用该工具以及如何提取参数。因此 docstring 必须清晰说明工具功能、使用场景和参数含义。参数类型注解（如 distance: float）帮助模型生成正确类型的参数。

2. DeepSeek 模型集成

DeepSeek 兼容 OpenAI API 格式，因此使用 ChatOpenAI 接口连接。关键配置：base_url 指向 DeepSeek 的 API 地址而非 OpenAI；temperature=0.1 设置低温度确保工具调用时输出稳定可预测（高温度可能导致模型"发挥"而误调工具）；api_key 从环境变量读取避免硬编码。

3. create_tool_calling_agent 工作原理

此函数创建的智能体使用模型的原生 Function Calling 能力（而非 ReAct 文本推理）。工作流程：用户输入 → 模型分析意图 → 模型决定调用哪个工具并生成参数 → AgentExecutor 执行工具函数 → 工具返回结果 → 模型根据结果生成自然语言回复。agent_scratchpad 占位符存放模型每一步的思考和工具调用记录。

4. AgentExecutor 执行器

AgentExecutor 是智能体的"运行引擎"，管理"推理-调用-反馈"循环。max_iterations=5 限制最大循环次数，防止模型陷入无限调用。verbose=True 打印详细过程便于调试。handle_parsing_errors=True 在模型输出格式异常时自动重试而非崩溃。

5. 多轮对话与 chat_history

chat_history 列表存储历史消息，每次对话后追加 HumanMessage 和 AIMessage。下次调用时传入历史，模型可以理解上下文关联（如测试3"再走2米"应基于前文"向前走"理解为继续前进）。MessagesPlaceholder 在提示词模板中为历史消息预留位置。

易错提示

!

API Key 未设置：如果未设置 DEEPSEEK_API_KEY 环境变量，程序会报认证错误。使用 echo $DEEPSEEK_API_KEY 确认已设置。

!

工具 docstring 不清晰：如果工具描述含糊（如只写"移动机器人"），模型可能无法正确判断何时调用或提取错误参数。描述应包含功能、触发场景、参数含义和取值范围。

!

参数类型不匹配：模型可能返回字符串类型的数字（如 "3" 而非 3）。工具函数内部应做类型转换和校验，避免因类型错误导致执行失败。

!

无限调用循环：如果工具返回的结果让模型反复尝试调用，可能超出 max_iterations。确保工具在成功执行后返回明确的结果信息，让模型知道任务已完成。

!

ROS2 节点初始化冲突：在真实 ROS2 环境中，如果多个工具函数各自初始化 rclpy 节点会报错。应使用全局单例节点或通过依赖注入共享节点。

任务 5 5分

智能体前端开发与功能集成

题目要求

编写智能体前端，对接智能体后台服务，实现对话、知识库问答、工具调用、具身交互可视化，验证智能体应用的功能。

思路分析

步骤1

开发后端 API：使用 FastAPI 封装前4个任务的功能，提供对话、RAG问答、工具调用的 REST API 接口。

步骤2

开发前端界面：使用 Vue3 构建对话界面、知识库问答面板和机器人状态可视化区域。

步骤3

前后端对接：前端通过 fetch/axios 调用后端 API，实现消息收发和状态更新。

步骤4

功能验证：在浏览器中测试对话、知识问答、机器人控制和可视化等全部功能。

代码实现

第1步：FastAPI 后端服务（backend/main.py）

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List
import requests

# 导入任务4的智能体模块
from agent import chat_with_robot
# 导入任务4的机器人状态
from ros2_tools import robot_state

# ========== 创建 FastAPI 应用 ==========
app = FastAPI(title="智能体应用API", version="1.0.0")

# 配置 CORS 跨域（允许前端访问）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],          # 允许所有来源（开发环境）
    allow_credentials=True,
    allow_methods=["*"],          # 允许所有 HTTP 方法
    allow_headers=["*"],          # 允许所有请求头
)

# ========== Dify API 配置 ==========
DIFY_API_URL = "http://localhost/v1/chat-messages"
DIFY_API_KEY = "app-your-api-key-here"

# ========== 会话管理 ==========
# 存储不同用户的对话历史（内存存储，生产环境应用 Redis）
sessions = {}

# ========== 请求/响应数据模型 ==========
class ChatRequest(BaseModel):
    """对话请求模型"""
    message: str                    # 用户消息
    user_id: str = "default_user"  # 用户 ID
    mode: str = "agent"          # 模式: agent(智能体) / rag(知识库) / robot(机器人控制)

class ChatResponse(BaseModel):
    """对话响应模型"""
    reply: str                      # 回复文本
    mode: str                      # 当前模式
    robot_status: dict             # 机器人状态
    retrieved_docs: Optional[List[dict]] = None  # RAG 检索结果（仅 rag 模式）


# ========== API 接口定义 ==========

@app.post("/api/chat")
async def chat(request: ChatRequest):
    """
    统一对话接口
    根据 mode 参数路由到不同的处理逻辑：
    - agent: Dify 智能体对话（含知识库 RAG）
    - robot: LangChain + Function Calling 机器人控制
    """
    user_id = request.user_id
    mode = request.mode

    # 初始化或获取用户会话
    if user_id not in sessions:
        sessions[user_id] = {
            "dify_conversation_id": None,  # Dify 对话 ID
            "langchain_history": [],        # LangChain 对话历史
        }
    session = sessions[user_id]

    if mode == "robot":
        # ---------- 机器人控制模式 ----------
        # 调用任务4的 LangChain 智能体
        reply, history = chat_with_robot(
            request.message,
            session["langchain_history"]
        )
        session["langchain_history"] = history

        return {
            "reply": reply,
            "mode": mode,
            "robot_status": robot_state,
        }

    else:
        # ---------- 智能体/RAG 问答模式 ----------
        # 调用 Dify API（包含知识库 RAG 功能）
        headers = {
            "Authorization": f"Bearer {DIFY_API_KEY}",
            "Content-Type": "application/json"
        }
        payload = {
            "inputs": {},
            "query": request.message,
            "response_mode": "blocking",
            "user": user_id,
        }
        if session["dify_conversation_id"]:
            payload["conversation_id"] = session["dify_conversation_id"]

        response = requests.post(DIFY_API_URL, headers=headers, json=payload)
        result = response.json()

        reply = result.get("answer", "未获取到回复")
        session["dify_conversation_id"] = result.get("conversation_id")
        metadata = result.get("metadata", {})
        retrieved_docs = metadata.get("retriever_resources", [])

        return {
            "reply": reply,
            "mode": mode,
            "robot_status": robot_state,
            "retrieved_docs": retrieved_docs,
        }


@app.get("/api/robot/status")
async def get_status():
    """获取机器人当前状态"""
    return {"robot_status": robot_state}


# ========== 启动命令 ==========
# uvicorn main:app --host 0.0.0.0 --port 8000 --reload

第2步：Vue3 前端界面（frontend/src/App.vue）

<template>
  <!-- 根容器：三栏布局 -->
  <div class="app-container">

    <!-- ===== 左栏：对话区 ===== -->
    <div class="chat-panel">
      <h2>智能体对话</h2>

      <!-- 模式切换标签 -->
      <div class="mode-tabs">
        <button
          v-for="m in modes"
          :key="m.value"
          :class="['tab', { active: currentMode === m.value }]"
          @click="currentMode = m.value"
        >{{ m.label }}</button>
      </div>

      <!-- 消息列表区域 -->
      <div class="message-list" ref="msgList">
        <div
          v-for="(msg, i) in messages"
          :key="i"
          :class="['message', msg.role]"
        >
          <span class="msg-role">{{ msg.role === 'user' ? '我' : '机器人' }}</span>
          <p class="msg-text">{{ msg.text }}</p>
        </div>
      </div>

      <!-- 输入框和发送按钮 -->
      <div class="input-area">
        <input
          v-model="inputText"
          @keyup.enter="sendMessage"
          placeholder="输入消息...（如：向前走2米）"
        />
        <button @click="sendMessage" :disabled="loading">
          {{ loading ? '发送中...' : '发送' }}
        </button>
      </div>
    </div>

    <!-- ===== 右栏：机器人状态可视化 ===== -->
    <div class="status-panel">
      <h2>机器人状态</h2>

      <!-- 电量条 -->
      <div class="status-item">
        <label>电量</label>
        <div class="battery-bar">
          <div class="battery-fill"
               :style="{ width: robotStatus.battery + '%' }">
            {{ robotStatus.battery }}%
          </div>
        </div>
      </div>

      <!-- 位置信息 -->
      <div class="status-item">
        <label>位置</label>
        <span>X: {{ pos.x }}, Y: {{ pos.y }}</span>
      </div>

      <!-- 朝向信息 -->
      <div class="status-item">
        <label>朝向</label>
        <span>{{ Math.round(thetaDeg) }}度</span>
      </div>

      <!-- 运行状态 -->
      <div class="status-item">
        <label>状态</label>
        <span :class="robotStatus.status">{{ robotStatus.status }}</span>
      </div>

      <!-- 机器人可视化图标 -->
      <div class="robot-viz">
        <div class="robot-icon"
             :style="{ transform: 'rotate(' + thetaDeg + 'deg)' }">
          &#129302;
        </div>
      </div>

      <!-- RAG 检索结果展示 -->
      <div v-if="retrievedDocs.length" class="rag-results">
        <h3>知识库检索结果</h3>
        <div v-for="(doc, i) in retrievedDocs" :key="i" class="doc-item">
          <span class="doc-score">相似度: {{ doc.score }}</span>
          <p>{{ doc.content.substring(0, 100) }}...</p>
        </div>
      </div>
    </div>
  </div>
</template>

<script setup>
import { ref, reactive, computed, nextTick } from 'vue'

// ========== 后端 API 地址 ==========
const API_BASE = 'http://localhost:8000'

// ========== 响应式状态定义 ==========
const inputText = ref('')           // 输入框文本
const messages = ref([])            // 消息列表
const loading = ref(false)          // 加载状态
const currentMode = ref('agent')    // 当前模式
const retrievedDocs = ref([])       // RAG 检索结果

// 模式选项
const modes = [
  { value: 'agent', label: '智能体对话' },
  { value: 'rag', label: '知识库问答' },
  { value: 'robot', label: '机器人控制' },
]

// 机器人状态（从后端获取）
const robotStatus = reactive({
  battery: 85,
  speed: 0,
  status: 'idle',
  position: { x: 0, y: 0, theta: 0 }
})

// 计算属性：位置和朝向（便于模板显示）
const pos = computed(() => ({
  x: robotStatus.position.x.toFixed(1),
  y: robotStatus.position.y.toFixed(1)
}))
const thetaDeg = computed(() =>
  robotStatus.position.theta * 180 / Math.PI
)

// ========== 发送消息函数 ==========
async function sendMessage() {
  if (!inputText.value.trim() || loading.value) return

  // 添加用户消息到列表
  const userMsg = inputText.value
  messages.value.push({ role: 'user', text: userMsg })
  inputText.value = ''
  loading.value = true

  try {
    // 调用后端 API
    const res = await fetch(`${API_BASE}/api/chat`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        message: userMsg,
        mode: currentMode.value,
        user_id: 'web_user'
      })
    })
    const data = await res.json()

    // 添加机器人回复
    messages.value.push({ role: 'bot', text: data.reply })

    // 更新机器人状态
    if (data.robot_status) {
      Object.assign(robotStatus, data.robot_status)
    }

    // 更新 RAG 检索结果
    retrievedDocs.value = data.retrieved_docs || []

    // 滚动到最新消息
    await nextTick()
    const list = document.querySelector('.message-list')
    if (list) list.scrollTop = list.scrollHeight
  } catch (err) {
    messages.value.push({ role: 'bot', text: '通信错误: ' + err.message })
  } finally {
    loading.value = false
  }
}
</script>

<style scoped>
/* 三栏布局 */
.app-container { display: flex; gap: 20px; padding: 20px; }
.chat-panel { flex: 1; }
.status-panel { width: 320px; }
.message-list { height: 400px; overflow-y: auto; }
.message { margin: 10px 0; padding: 10px; border-radius: 8px; }
.message.user { background: #e3f2fd; }
.message.bot { background: #f5f5f5; }
</style>

第3步：启动前后端服务

# 终端1：启动 FastAPI 后端服务
cd backend
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

# 终端2：启动 Vue3 前端开发服务器
cd frontend
npm install    # 首次运行需安装依赖
npm run dev    # 启动开发服务器，默认 http://localhost:5173

代码详解

1. FastAPI 后端路由设计

后端提供统一的 /api/chat 接口，通过 mode 参数路由到不同处理逻辑：robot 模式调用任务4的 LangChain 智能体实现 Function Calling 机器人控制；agent/rag 模式调用 Dify API 实现智能体对话和知识库问答。这种设计使前端只需一个接口即可访问所有功能。

2. 会话管理

sessions 字典以 user_id 为键存储每个用户的会话状态，包括 Dify 的 conversation_id（用于多轮对话）和 LangChain 的 chat_history（对话消息列表）。这样不同用户的对话互不干扰，同一用户的多轮对话能保持上下文。

3. CORS 跨域配置

前端（localhost:5173）和后端（localhost:8000）端口不同，浏览器会阻止跨域请求。FastAPI 通过 CORSMiddleware 中间件添加 CORS 响应头，允许前端跨域访问。开发环境使用 allow_origins=["*"] 放行所有来源，生产环境应限制为具体域名。

4. Vue3 响应式状态

ref 用于基本类型（字符串、布尔值、数组），reactive 用于对象。computed 创建计算属性，当依赖的响应式数据变化时自动重新计算。机器人状态用 reactive 定义，后端返回新状态后通过 Object.assign 更新，Vue 自动触发界面重渲染。

5. 机器人可视化

机器人图标使用 CSS transform: rotate() 根据朝向角度旋转，直观展示机器人方向。电量条通过 :style 动态绑定宽度百分比。RAG 检索结果区域使用 v-if 条件渲染，仅在有检索结果时显示，展示片段内容和相似度分数。

易错提示

!

跨域请求被阻止：如果后端未配置 CORS 或配置错误，浏览器控制台会报 CORS 错误。确保 allow_origins 包含前端地址或使用 "*"。

!

API 地址错误：前端 API_BASE 必须指向后端实际运行地址。如果后端运行在 8000 端口，前端应配置为 http://localhost:8000。

!

响应数据结构不匹配：后端返回的 JSON 字段名必须与前端解析的一致。如后端返回 robot_status 而前端读 robotState 会导致 undefined。建议用 Pydantic 模型统一接口规范。

!

异步加载状态未处理：发送消息后如果忘记设置 loading 状态，用户可能连续点击发送导致重复请求。始终在请求开始时设 loading=true，结束时设 false。

4.5 项目总结与拓展

4.5.1 项目总结

通过本模块的五个任务，我们从零开始构建了一个完整的具身智能体应用系统，覆盖了从对话交互到机器人控制的全链路开发。回顾各任务的核心知识点：

任务1

Dify 智能体对话

  • 掌握了 Dify 平台的智能体创建流程
  • 学会了编写结构化系统提示词定义角色
  • 理解了对话记忆窗口的多轮上下文机制
  • 学会了通过 API 验证智能体对话功能
任务2

RAG 知识库问答

  • 掌握了知识库的创建和文档上传流程
  • 理解了文本分段策略对检索精度的影响
  • 学会了配置 Top-K 和 Score 阈值参数
  • 掌握了通过 metadata 分析检索质量的方法
任务3

OpenClaw Skills 插件

  • 掌握了 OpenClaw 插件的目录结构
  • 学会了编写 skill_config.yaml 配置文件
  • 实现了自动巡检的完整业务逻辑
  • 理解了插件触发条件和参数提取机制
任务4

Function Calling

  • 掌握了 LangChain 集成 DeepSeek 的方法
  • 学会了使用 @tool 装饰器定义工具函数
  • 理解了 Function Calling 的决策和执行流程
  • 实现了自然语言到机器人动作的转换
任务5

前端开发集成

  • 掌握了 FastAPI 后端 API 开发
  • 学会了 Vue3 响应式状态管理
  • 实现了前后端分离架构的对接
  • 完成了机器人状态的可视化展示

4.5.2 核心技术回顾

智能体架构

本项目展示了智能体的四大核心组件如何协作：LLM（DeepSeek）作为大脑负责理解和推理；记忆（Dify 记忆窗口 + LangChain chat_history）维护上下文；工具（ROS2 运动控制函数 + OpenClaw 巡检插件）执行具体动作；规划（Function Calling 决策链）将复杂任务分解执行。

RAG 技术链路

从文档上传到分段、向量化、存储、检索、生成的完整 RAG 链路。关键在于分段策略（Chunk Size + Overlap）和检索参数（Top-K + Score Threshold）的调优，需要在检索精度和上下文完整性之间找到平衡。

Function Calling 机制

Function Calling 是连接大模型与物理世界的桥梁。模型通过工具描述理解"能做什么"，通过用户意图判断"该做什么"，通过参数生成告诉程序"怎么做"。@tool 装饰器的 docstring 是模型理解工具的关键，必须清晰准确。

全栈开发流程

项目采用前后端分离架构：FastAPI 提供统一 API 接口，Vue3 构建交互界面，Dify/LangChain/OpenClaw 作为智能体框架层，DeepSeek 提供模型能力，ROS2 连接物理机器人。各层职责清晰，通过标准接口协作。

4.5.3 拓展方向

拓展1：多模态交互

在当前文字对话基础上，增加语音输入（语音识别 ASR）和语音输出（语音合成 TTS），实现真正的语音交互。还可接入摄像头实现视觉感知，让机器人能"看到"环境并做出响应。

拓展2：多智能体协作

使用 LangGraph 编排多个智能体协作完成任务。例如：调度智能体负责任务分配，巡检智能体负责环境检测，维修智能体负责故障处理。多个智能体通过消息传递协同工作。

拓展3：强化学习优化

引入强化学习让机器人通过试错优化巡检路线和运动策略。结合仿真环境（如 Gazebo）进行训练，再将训练好的策略部署到真实机器人上。

拓展4：边缘计算部署

将智能体部署到边缘设备（如 Jetson Nano），实现离线运行和低延迟响应。使用模型量化技术压缩模型体积，适应边缘设备的算力限制。

拓展5：数字孪生

构建工厂的数字孪生模型，在虚拟空间中实时映射机器人的状态和动作。通过数字孪生进行巡检路线优化、异常预测和远程监控。

4.5.4 评分要点回顾

任务考核要点分值关键验收标准
任务1 Dify智能体创建与对话 5分 系统提示词合理、记忆窗口配置正确、多轮对话上下文连贯
任务2 知识库与RAG问答 5分 文档上传成功、分段参数合理、Top-K检索有效、RAG回答准确
任务3 OpenClaw Skills插件 5分 插件结构完整、配置正确、巡检逻辑可执行、智能体可调用
任务4 Function Calling控制ROS2 5分 DeepSeek集成成功、工具定义规范、指令转换准确、动作执行正确
任务5 前端开发与功能集成 5分 前后端对接成功、四大功能可用、界面交互流畅、可视化正确
合计 25分 智能体应用功能完整、代码规范、文档清晰