手语智能识别方案设计：基于 RAG + MCP + YOLO + Agent 的小而美工程实践

# 导入YOLO模型
from ultralytics import YOLO
import cv2

# 加载预训练模型
model = YOLO('yolov8n-pose.pt')

# 视频处理函数
def process_video(video_path):
    cap = cv2.VideoCapture(video_path)
    
    while cap.isOpened():
        success, frame = cap.read()
        if success:
            # 运行YOLO推理
            results = model(frame, conf=0.5, iou=0.7)
            
            # 可视化结果
            annotated_frame = results[0].plot()
            cv2.imshow('YOLO Hand Detection', annotated_frame)
            
            # 提取关键点
            keypoints = results[0].keypoints.numpy()
            if keypoints is not None:
                # 处理关键点数据
                process_keypoints(keypoints)
            
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
        else:
            break
    
    cap.release()
    cv2.destroyAllWindows()

# 关键点处理函数
def process_keypoints(keypoints):
    # 提取左右手关键点
    left_hand = keypoints[:, 9:17, :]  # 左手关键点索引
    right_hand = keypoints[:, 17:25, :]  # 右手关键点索引
    
    # 计算手型特征
    left_hand_shape = calculate_hand_shape(left_hand)
    right_hand_shape = calculate_hand_shape(right_hand)
    
    # 输出手型信息
    print(f"Left hand shape: {left_hand_shape}")
    print(f"Right hand shape: {right_hand_shape}")

# 手型计算函数
def calculate_hand_shape(hand_keypoints):
    # 计算手指间距离，判断手型
    # 简化实现，实际可使用更复杂的算法
    if hand_keypoints is None or hand_keypoints.shape[0] == 0:
        return "unknown"
    
    # 计算拇指与食指距离
    thumb_tip = hand_keypoints[0, 4, :2]  # 拇指指尖
    index_tip = hand_keypoints[0, 8, :2]  # 食指指尖
    distance = ((thumb_tip[0] - index_tip[0]) ** 2 + (thumb_tip[1] - index_tip[1]) ** 2) ** 0.5
    
    if distance < 50:
        return "OK"
    else:
        return "Open"

# 运行示例
if __name__ == "__main__":
    process_video("hand_sign_video.mp4")

{
  "timestamp": 1234567890,
  "hand": "both",
  "shape": "index_point",
  "motion": "circle_clockwise",
  "location": "chest_center",
  "duration_ms": 450,
  "confidence": 0.92
}

import numpy as np
import json
from datetime import datetime

# 原子识别类
class SignAtomRecognizer:
    def __init__(self):
        # 初始化规则引擎和CNN模型
        self.rule_engine = RuleEngine()
        self.cnn_model = self.load_cnn_model()
        
    def load_cnn_model(self):
        # 加载预训练的CNN模型
        # 简化实现，实际可使用PyTorch或TensorFlow
        return None
    
    def recognize_atom(self, keypoints_sequence, timestamp):
        # 从关键点序列识别原子
        
        # 1. 提取手型特征
        hand_shape = self.rule_engine.detect_hand_shape(keypoints_sequence)
        
        # 2. 提取动作特征
        motion = self.rule_engine.detect_motion(keypoints_sequence)
        
        # 3. 确定空间位置
        location = self.rule_engine.detect_location(keypoints_sequence)
        
        # 4. 计算持续时间
        duration_ms = len(keypoints_sequence) * 33.3  # 假设30fps
        
        # 5. 确定手的数量（左手/右手/双手）
        hand = self.rule_engine.detect_hand_count(keypoints_sequence)
        
        # 6. 计算置信度
        confidence = self.calculate_confidence(hand_shape, motion, location)
        
        # 生成标准化JSON输出
        atom = {
            "timestamp": timestamp,
            "hand": hand,
            "shape": hand_shape,
            "motion": motion,
            "location": location,
            "duration_ms": duration_ms,
            "confidence": confidence
        }
        
        return atom
    
    def calculate_confidence(self, hand_shape, motion, location):
        # 简化的置信度计算
        confidence = 0.8
        if hand_shape != "unknown":
            confidence += 0.05
        if motion != "unknown":
            confidence += 0.05
        if location != "unknown":
            confidence += 0.05
        return min(confidence, 0.95)

# 规则引擎类
class RuleEngine:
    def detect_hand_shape(self, keypoints_sequence):
        # 基于关键点距离检测手型
        return "index_point"  # 简化返回
    
    def detect_motion(self, keypoints_sequence):
        # 基于关键点变化检测动作
        return "circle_clockwise"  # 简化返回
    
    def detect_location(self, keypoints_sequence):
        # 基于关键点坐标检测位置
        return "chest_center"  # 简化返回
    
    def detect_hand_count(self, keypoints_sequence):
        # 检测手的数量
        return "both"  # 简化返回

# 示例使用
if __name__ == "__main__":
    recognizer = SignAtomRecognizer()
    
    # 模拟关键点序列
    keypoints_sequence = [np.random.rand(21, 3) for _ in range(15)]  # 15帧关键点
    timestamp = datetime.now().timestamp()
    
    # 识别原子
    atom = recognizer.recognize_atom(keypoints_sequence, timestamp)
    
    # 输出结果
    print(json.dumps(atom, indent=2))

from flask import Flask, request, jsonify
from ultralytics import YOLO
from sign_atom_recognizer import SignAtomRecognizer

app = Flask(__name__)

# 初始化模型
yolo_model = YOLO('yolov8n-pose.pt')
atom_recognizer = SignAtomRecognizer()

# API端点：从视频提取原子序列
@app.route('/detect_sign_atoms', methods=['POST'])
def detect_sign_atoms():
    try:
        video_path = request.json['video_path']
        
        # 1. 使用YOLO提取关键点
        results = yolo_model.predict(video_path, stream=True)
        
        # 2. 处理关键点，生成原子序列
        atom_sequence = []
        for result in results:
            keypoints = result.keypoints.numpy()
            if keypoints is not None:
                timestamp = result.timestamp
                atom = atom_recognizer.recognize_atom(keypoints, timestamp)
                atom_sequence.append(atom)
        
        return jsonify({
            'success': True,
            'data': atom_sequence
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        })

# API端点：添加自定义手语
@app.route('/add_custom_sign', methods=['POST'])
def add_custom_sign():
    try:
        label = request.json['label']
        samples = request.json['samples']
        
        # 简化实现，实际应处理视频样本
        # 1. 提取关键点
        # 2. 生成初步原子标注
        # 3. 保存到数据库
        
        return jsonify({
            'success': True,
            'message': f'Custom sign "{label}" added successfully'
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        })

# API端点：本地微调模型
@app.route('/retrain_local_model', methods=['POST'])
def retrain_local_model():
    try:
        params = request.json['params']
        
        # 简化实现，实际应触发增量训练
        # 1. 加载训练数据
        # 2. 执行增量训练
        # 3. 保存更新后的模型
        
        return jsonify({
            'success': True,
            'message': 'Model retrained successfully'
        })
    except Exception as e:
        return jsonify({
            'success': False,
            'error': str(e)
        })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

# Agent决策规则
class AgentDecisionEngine:
    def __init__(self, rag_client):
        self.rag_client = rag_client
        
    def decide_next_action(self, atom_sequence, context_history, user_preferences):
        # 1. 计算平均置信度
        avg_confidence = sum(atom['confidence'] for atom in atom_sequence) / len(atom_sequence)
        
        # 2. 检查是否存在歧义原子
        has_ambiguous_atom = any(atom['confidence'] < 0.7 for atom in atom_sequence)
        
        # 3. 决策逻辑
        if avg_confidence < 0.8 and context_history:
            # 调用RAG检索相似语义
            return self._call_rag(atom_sequence, context_history)
        elif has_ambiguous_atom:
            # 请求用户确认
            return self._request_user_confirmation(atom_sequence)
        elif user_preferences.get('direct_output', False):
            # 直接输出
            return self._direct_output(atom_sequence)
        else:
            # 回退本地规则匹配
            return self._fallback_to_rules(atom_sequence)
    
    def _call_rag(self, atom_sequence, context_history):
        # 调用RAG服务
        query = self._format_atom_query(atom_sequence)
        rag_result = self.rag_client.retrieve(query, context_history)
        return {
            'action': 'rag_enhanced',
            'result': rag_result
        }
    
    def _request_user_confirmation(self, atom_sequence):
        # 生成确认请求
        return {
            'action': 'user_confirmation',
            'atoms': atom_sequence
        }
    
    def _direct_output(self, atom_sequence):
        # 直接生成输出
        output = self._generate_direct_output(atom_sequence)
        return {
            'action': 'direct_output',
            'result': output
        }
    
    def _fallback_to_rules(self, atom_sequence):
        # 使用规则匹配生成输出
        output = self._rule_based_matching(atom_sequence)
        return {
            'action': 'rule_based',
            'result': output
        }
    
    def _format_atom_query(self, atom_sequence):
        # 将原子序列格式化为RAG查询
        query_parts = []
        for atom in atom_sequence:
            query_parts.append(f"{atom['shape']} {atom['motion']} at {atom['location']}")
        return ' '.join(query_parts)
    
    def _generate_direct_output(self, atom_sequence):
        # 直接生成输出
        return "生成的文本结果"
    
    def _rule_based_matching(self, atom_sequence):
        # 基于规则匹配生成输出
        return "规则匹配的文本结果"

from sentence_transformers import SentenceTransformer
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.llms import HuggingFacePipeline
from langchain.chains import RetrievalQA

class RAGService:
    def __init__(self):
        # 初始化嵌入模型
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        
        # 初始化向量数据库
        self.vector_db = self._init_vector_db()
        
        # 初始化LLM
        self.llm = self._init_llm()
        
        # 初始化RetrievalQA链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vector_db.as_retriever(k=5)
        )
    
    def _init_vector_db(self):
        # 初始化向量数据库
        # 简化实现，实际应加载预构建的向量数据库
        embeddings = HuggingFaceEmbeddings(model_name='all-MiniLM-L6-v2')
        return FAISS.from_texts(["示例文本"], embeddings)
    
    def _init_llm(self):
        # 初始化小型LLM
        # 简化实现，实际可使用Llama.cpp或其他轻量级LLM
        return HuggingFacePipeline.from_model_id(
            model_id="google/flan-t5-small",
            task="text2text-generation"
        )
    
    def retrieve(self, query, context_history):
        # 检索并生成增强文本
        
        # 1. 格式化查询，添加上下文历史
        formatted_query = self._format_query(query, context_history)
        
        # 2. 调用RetrievalQA链
        result = self.qa_chain.invoke(formatted_query)
        
        return result
    
    def _format_query(self, query, context_history):
        # 格式化查询，添加上下文历史
        context_str = "\n".join([f"历史输入: {h['input']}, 历史输出: {h['output']}" for h in context_history[-5:]])
        return f"上下文历史:\n{context_str}\n\n当前查询: {query}\n\n请生成准确的文本结果："

# 安装Python依赖
pip install -r requirements.txt

# requirements.txt内容
albumentations==1.3.1
cv2==4.8.1
docker==6.1.3
faiss-cpu==1.7.4
flask==2.3.3
langchain==0.0.309
numpy==1.25.2
opencv-python==4.8.1.78
sentence-transformers==2.2.2
torch==2.0.1
ultralytics==8.0.224

# 拉取镜像
docker pull sign-language-recognition:latest

# 运行容器
docker run -d -p 5000:5000 -v ./data:/app/data sign-language-recognition:latest

# 运行YOLO手部检测脚本
python yolo_hand_detection.py --video hand_sign_video.mp4 --output annotated_video.mp4

# 输出结果
正在处理视频: hand_sign_video.mp4
检测到215帧，平均FPS: 32
生成标注视频: annotated_video.mp4

# 运行原子识别脚本
python sign_atom_recognizer.py --keypoints keypoints.json --output atoms.json

# 输出结果
加载关键点数据: keypoints.json
识别到12个原子
生成原子序列: atoms.json

# 启动MCP服务
python mcp_server.py

# 输出结果
 * Serving Flask app 'mcp_server' (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:5000
 * Running on http://192.168.1.100:5000
Press CTRL+C to quit