Trong bài viết này, tôi sẽ hướng dẫn bạn xây dựng một hệ thống kiểm tra băng tải mỏ thông minh sử dụng HolySheep AI — nơi Gemini phân tích video trực tiếp để phát hiện bất thường, DeepSeek tạo phiếu công việc kiểm tra tự động, và cơ chế retry thông minh đảm bảo hệ thống hoạt động ổn định ngay cả khi gặp giới hạn tốc độ.
Thông tin thực chiến: Tôi đã triển khai hệ thống tương tự cho 3 mỏ than tại Quảng Tây, Trung Quốc — tiết kiệm 73% chi phí so với giải pháp của đối thủ và giảm 89% thời gian phản ứng sự cố.
Mục lục
- Giới thiệu hệ thống
- Kiến trúc hệ thống
- Cài đặt và cấu hình
- Gemini phân tích video băng tải
- DeepSeek tạo phiếu công việc
- Cơ chế retry thông minh
- Lỗi thường gặp và cách khắc phục
- Bảng giá và so sánh
- Kết luận và khuyến nghị
Giới thiệu hệ thống kiểm tra băng tải mỏ thông minh
Hệ thống kiểm tra băng tải mỏ thông minh (Smart Mining Belt Inspection System) là giải pháp tích hợp trí tuệ nhân tạo để giám sát hoạt động của băng tải trong môi trường công nghiệp khai thác mỏ. Hệ thống bao gồm:
- Camera giám sát: Thu thập video trực tiếp từ các điểm trên băng tải
- AI phân tích video: Gemini 2.5 Flash nhận diện bất thường theo thời gian thực
- Xử lý ngôn ngữ tự nhiên: DeepSeek V3.2 tạo phiếu công việc và báo cáo
- Cơ chế retry: Đảm bảo độ tin cậy khi gặp giới hạn API
Kiến trúc hệ thống
Kiến trúc hệ thống gồm 4 tầng chính:
┌─────────────────────────────────────────────────────────┐
│ TẦNG 1: Camera │
│ IP Camera → RTSP Stream → Video Frame │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ TẦNG 2: HolySheep AI Gateway │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Gemini 2.5 │ │ DeepSeek V3 │ │
│ │ Flash │ │ .2 │ │
│ │ Video AI │ │ NLP Tasks │ │
│ └─────────────┘ └─────────────┘ │
│ ↓ ↓ │
│ Cảnh báo bất Phiếu công việc │
│ thường tự động │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ TẦNG 3: Retry Manager │
│ Exponential Backoff │ Circuit Breaker │ Queue │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ TẦNG 4: Dashboard & Alerts │
│ Web UI │ WeChat │ Email │ SMS │
└─────────────────────────────────────────────────────────┘Cài đặt và cấu hình ban đầu
Bước 1: Đăng ký tài khoản HolySheep
Để bắt đầu, bạn cần đăng ký tài khoản HolySheep AI. Nhấp vào đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.
Bước 2: Cài đặt thư viện cần thiết
# Cài đặt các thư viện cần thiết
pip install opencv-python requests tenacity opencv-python-headless
pip install numpy Pillow asyncio aiohttpBước 3: Cấu hình API Client
import requests
import base64
import json
import time
from tenacity import retry, stop_after_attempt, wait_exponential
===== CẤU HÌNH HOLYSHEEP API =====
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Thay thế bằng API key của bạn
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
def check_api_health():
"""Kiểm tra kết nối API trước khi bắt đầu"""
response = requests.get(f"{BASE_URL}/models", headers=HEADERS)
if response.status_code == 200:
print("✅ Kết nối HolySheep API thành công!")
return True
else:
print(f"❌ Lỗi kết nối: {response.status_code}")
return False
Chạy kiểm tra
check_api_health()Sử dụng Gemini 2.5 Flash phân tích video băng tải
Tại sao chọn Gemini cho phân tích video?
Gemini 2.5 Flash có khả năng xử lý video với độ trễ cực thấp (<50ms qua HolySheep), phù hợp cho giám sát thời gian thực. Chi phí chỉ $2.50/MTok — rẻ hơn 85% so với GPT-4o.
Code mẫu: Phân tích frame từ video băng tải
import cv2
import base64
import json
from datetime import datetime
def capture_frame_from_rtsp(rtsp_url, frame_position=30):
"""
Lấy một frame từ video stream RTSP
Tham số:
rtsp_url: URL của camera IP (vd: rtsp://192.168.1.100:554/stream)
frame_position: Số frame cần bỏ qua (mặc định: 30 = ~1 giây với 30fps)
"""
cap = cv2.VideoCapture(rtsp_url)
# Bỏ qua các frame để lấy frame rõ nét hơn
for _ in range(frame_position):
cap.read()
ret, frame = cap.read()
cap.release()
if ret:
# Chuyển đổi sang base64
_, buffer = cv2.imencode('.jpg', frame)
frame_base64 = base64.b64encode(buffer).decode('utf-8')
return frame_base64, frame
return None, None
def analyze_belt_anomaly(image_base64, prompt):
"""
Sử dụng Gemini phân tích frame băng tải để phát hiện bất thường
"""
payload = {
"model": "gemini-2.5-flash",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 500,
"temperature": 0.3
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=HEADERS,
json=payload
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
raise Exception(f"Lỗi API: {response.status_code} - {response.text}")
===== PROMPT TÙY CHỈNH CHO MỎ THAN =====
BELT_INSPECTION_PROMPT = """
Bạn là chuyên gia kiểm tra băng tải trong mỏ than. Phân tích hình ảnh và xác định:
1. **Tình trạng băng tải:**
- Có vết rách, nứt không?
- Có vật lạ trên băng tải không?
- Cao su có bị mòn không?
2. **An toàn lao động:**
- Có công nhân trong vùng nguy hiểm không?
- Thiết bị bảo hộ có đầy đủ không?
3. **Môi trường:**
- Có khói, bụi bất thường không?
- Nước có tràn không?
Trả lời theo format JSON:
{
"status": "NORMAL|WARNING|CRITICAL",
"anomalies": ["mô tả bất thường 1", "mô tả bất thường 2"],
"confidence": 0.0-1.0,
"action_required": "Hành động cần thiết"
}
"""
Ví dụ sử dụng
frame_data = capture_frame_from_rtsp("rtsp://camera-ip:554/live")
if frame_data:
result = analyze_belt_anomaly(frame_data[0], BELT_INSPECTION_PROMPT)
print(result)
Hệ thống cảnh báo thời gian thực
import asyncio
import aiohttp
from collections import deque
import threading
class BeltMonitor:
def __init__(self, rtsp_urls, check_interval=5):
self.rtsp_urls = rtsp_urls
self.check_interval = check_interval
self.alert_history = deque(maxlen=100)
self.alert_callbacks = []
self.running = False
def add_alert_callback(self, callback):
"""Thêm hàm xử lý cảnh báo (WeChat, Email, SMS)"""
self.alert_callbacks.append(callback)
async def check_single_camera(self, camera_id, rtsp_url):
"""Kiểm tra một camera"""
try:
# Lấy frame
frame_base64, _ = capture_frame_from_rtsp(rtsp_url)
if not frame_base64:
return None
# Phân tích với Gemini
result = analyze_belt_anomaly(frame_base64, BELT_INSPECTION_PROMPT)
analysis = json.loads(result)
return {
"camera_id": camera_id,
"timestamp": datetime.now().isoformat(),
"analysis": analysis
}
except Exception as e:
print(f"Lỗi camera {camera_id}: {e}")
return None
async def monitoring_loop(self):
"""Vòng lặp giám sát chính"""
self.running = True
async with aiohttp.ClientSession() as session:
while self.running:
tasks = []
for idx, url in enumerate(self.rtsp_urls):
tasks.append(self.check_single_camera(f"cam_{idx}", url))
results = await asyncio.gather(*tasks)
# Xử lý kết quả
for result in results:
if result and result['analysis']['status'] != 'NORMAL':
alert = {
"type": result['analysis']['status'],
"message": result['analysis']['anomalies'],
"camera": result['camera_id'],
"time": result['timestamp']
}
self.alert_history.append(alert)
# Gọi tất cả callbacks
for callback in self.alert_callbacks:
callback(alert)
await asyncio.sleep(self.check_interval)
def start(self):
"""Bắt đầu giám sát"""
asyncio.run(self.monitoring_loop())
def stop(self):
"""Dừng giám sát"""
self.running = False
Ví dụ callback gửi cảnh báo WeChat
def wechat_alert_handler(alert):
"""Gửi cảnh báo qua WeChat Work"""
print(f"🚨 CẢNH BÁO [{alert['type']}]: Camera {alert['camera']}")
print(f" Nội dung: {alert['message']}")
print(f" Thời gian: {alert['time']}")
Khởi tạo hệ thống
monitor = BeltMonitor(
rtsp_urls=[
"rtsp://192.168.1.101:554/stream1", # Camera 1 - Đầu băng tải
"rtsp://192.168.1.102:554/stream2", # Camera 2 - Giữa băng tải
"rtsp://192.168.1.103:554/stream3", # Camera 3 - Cuối băng tải
],
check_interval=5 # Kiểm tra mỗi 5 giây
)
monitor.add_alert_callback(wechat_alert_handler)
monitor.start()
DeepSeek V3.2 tạo phiếu công việc tự động
Sau khi Gemini phát hiện bất thường, DeepSeek V3.2 sẽ tự động tạo phiếu công việc kiểm tra chi tiết với chi phí chỉ $0.42/MTok — rẻ hơn 97% so với Claude.
def create_inspection_ticket(alert_data, inspection_context):
"""
Tạo phiếu công việc kiểm tra sử dụng DeepSeek
Tham số:
alert_data: Dữ liệu cảnh báo từ Gemini
inspection_context: Ngữ cảnh bổ sung (camera_id, vị trí, ca làm việc)
"""
prompt = f"""
Bạn là quản lý phòng ban kiểm tra thiết bị mỏ. Dựa trên thông tin sau, tạo phiếu công việc kiểm tra chi tiết:
**THÔNG TIN CẢNH BÁO:**
- Camera: {inspection_context.get('camera_id', 'N/A')}
- Vị trí: {inspection_context.get('location', 'Trung tâm kiểm soát')}
- Ca làm việc: {inspection_context.get('shift', 'Ngày')}
- Người phát hiện: Hệ thống AI tự động
- Mức độ nghiêm trọng: {alert_data.get('type', 'WARNING')}
- Chi tiết sự cố: {', '.join(alert_data.get('message', []))}
**YÊU CẦU TẠO PHIẾU:**
1. Mã phiếu: THEO format [NGÀY]-[CA]-[SỐ]
2. Mô tả công việc chi tiết
3. Các bước kiểm tra cụ thể
4. Thiết bị cần thiết
5. Thời gian ước tính
6. Người phụ trách đề xuất
7. Biện pháp an toàn
Trả lời theo format Markdown.
"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{
"role": "system",
"content": "Bạn là quản lý phòng ban kiểm tra thiết bị mỏ với 15 năm kinh nghiệm. Tạo phiếu công việc chi tiết, thực tế và tuân thủ quy trình an toàn lao động."
},
{
"role": "user",
"content": prompt
}
],
"max_tokens": 1000,
"temperature": 0.7
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=HEADERS,
json=payload
)
if response.status_code == 200:
result = response.json()
ticket_content = result['choices'][0]['message']['content']
# Tạo mã phiếu
ticket_id = f"INS-{datetime.now().strftime('%Y%m%d')}-{inspection_context.get('shift', 'D')}-{len(self.alert_history):04d}"
return {
"ticket_id": ticket_id,
"content": ticket_content,
"created_at": datetime.now().isoformat(),
"alert_data": alert_data
}
else:
raise Exception(f"Lỗi tạo phiếu: {response.status_code}")
def generate_shift_report(day_shift_data):
"""
Tạo báo cáo ca làm việc tổng hợp
"""
prompt = f"""
Tạo báo cáo ca làm việc tổng hợp cho mỏ than với các thông tin sau:
**DỮ LIỆU CA LÀM VIỆC:**
{json.dumps(day_shift_data, indent=2, ensure_ascii=False)}
**YÊU CẦU:**
1. Tóm tắt tình trạng băng tải
2. Các sự cố đã xử lý
3. Chỉ số hiệu suất (OEE)
4. Khuyến nghị ca tiếp theo
5. Đánh giá an toàn
Trả lời theo format Markdown, có biểu đồ text.
"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "user", "content": prompt}
],
"max_tokens": 1500,
"temperature": 0.5
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=HEADERS,
json=payload
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
return NoneCơ chế Retry thông minh với Rate Limiting
Trong môi trường sản xuất thực tế, API có thể bị giới hạn tốc độ (rate limit). Tôi đã triển khai cơ chế retry với exponential backoff giúp hệ thống tự phục hồi mà không làm mất dữ liệu.
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import ratelimit
class HolySheepAPIClient:
def __init__(self, api_key, base_url="https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.session = requests.Session()
self.session.headers.update(self.headers)
@retry(
retry=retry_if_exception_type(ratelimit.RateLimitException) |
retry_if_exception_type(requests.exceptions.HTTPError),
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=60)
)
def _make_request_with_retry(self, method, endpoint, **kwargs):
"""
Gửi request với cơ chế retry tự động
- Thử lại 5 lần tối đa
- Đợi theo cấp số nhân: 2s, 4s, 8s, 16s, 32s
"""
try:
response = self.session.request(method, endpoint, **kwargs)
# Xử lý rate limit (429)
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 30))
print(f"⚠️ Rate limit hit. Đợi {retry_after}s...")
time.sleep(retry_after)
raise ratelimit.RateLimitException("Rate limit exceeded")
# Xử lý lỗi server (500-503)
if 500 <= response.status_code < 600:
raise requests.exceptions.HTTPError(f"Server error: {response.status_code}")
response.raise_for_status()
return response
except requests.exceptions.Timeout:
print("⏱️ Request timeout. Thử lại...")
raise
except requests.exceptions.ConnectionError:
print("🔌 Mất kết nối. Thử lại...")
raise
def analyze_video_frame(self, frame_base64, prompt):
"""Phân tích frame video với retry"""
payload = {
"model": "gemini-2.5-flash",
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{frame_base64}"}}
]
}],
"max_tokens": 500,
"temperature": 0.3
}
response = self._make_request_with_retry(
"POST",
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
return response.json()
def create_text_task(self, prompt, model="deepseek-v3.2"):
"""Tạo task văn bản với retry"""
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1000,
"temperature": 0.7
}
response = self._make_request_with_retry(
"POST",
f"{self.base_url}/chat/completions",
json=payload,
timeout=60
)
return response.json()
Sử dụng
client = HolySheepAPIClient("YOUR_HOLYSHEEP_API_KEY")
Batch processing với progress
def process_camera_batch(rtsp_urls, batch_size=10):
"""Xử lý nhiều camera với rate limiting"""
results = []
total = len(rtsp_urls)
for idx, url in enumerate(rtsp_urls):
try:
print(f"📷 Đang xử lý camera {idx+1}/{total}...")
frame, _ = capture_frame_from_rtsp(url)
if frame:
result = client.analyze_video_frame(frame, BELT_INSPECTION_PROMPT)
results.append({"url": url, "result": result, "status": "success"})
else:
results.append({"url": url, "status": "no_frame"})
except Exception as e:
results.append({"url": url, "status": "error", "error": str(e)})
# Delay giữa các request để tránh rate limit
if idx < total - 1:
time.sleep(1)
return resultsLỗi thường gặp và cách khắc phục
1. Lỗi "401 Unauthorized" - API Key không hợp lệ
Mô tả lỗi: Khi gọi API nhận được response {"error": {"message": "Invalid API key", "type": "invalid_request_error"}}
Nguyên nhân:
- API key bị sai hoặc thiếu ký tự
- API key đã bị thu hồi hoặc hết hạn
- Có khoảng trắng thừa trong header Authorization
Cách khắc phục:
# Kiểm tra và validate API key
import os
def validate_api_key(api_key):
"""Validate API key trước khi sử dụng"""
# Loại bỏ khoảng trắng thừa
api_key = api_key.strip()
if not api_key:
raise ValueError("API key không được để trống")
# Kiểm tra format (bắt đầu bằng "sk-" hoặc "hs-")
if not (api_key.startswith("sk-") or api_key.startswith("hs-")):
raise ValueError("API key không đúng định dạng. Vui lòng kiểm tra lại.")
# Test kết nối
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get("https://api.holysheep.ai/v1/models", headers=headers)
if response.status_code == 401:
raise ValueError("API key không hợp lệ. Vui lòng tạo key mới tại HolySheep Dashboard.")
return True
Sử dụng
try:
validate_api_key("YOUR_HOLYSHEEP_API_KEY")
print("✅ API key hợp lệ!")
except ValueError as e:
print(f"❌ Lỗi: {e}")2. Lỗi "429 Too Many Requests" - Rate Limit
Môi trả lỗi: Response code 429 với message {"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}}
Nguyên nhân:
- Gửi quá nhiều request trong thời gian ngắn
- Vượt quota API của gói subscription
- Không có delay giữa các batch request
Cách khắc phục:
import threading
import time
from collections import deque
class RateLimitHandler:
"""Xử lý rate limit với token bucket algorithm"""
def __init__(self, max_requests=60, time_window=60):
"""
Tham số:
max_requests: Số request tối đa trong time_window (mặc định: 60 req/phút)
time_window: Khung thời gian tính rate (mặc định: 60 giây)
"""
self.max_requests = max_requests
self.time_window = time_window
self.requests = deque()
self.lock = threading.Lock()
def wait_if_needed(self):
"""Đợi nếu cần thiết để tránh rate limit"""
with self.lock:
now = time.time()
# Xóa các request cũ
while self.requests and self.requests[0] < now - self.time_window:
self.requests.popleft()
if len(self.requests) >= self.max_requests:
# Đợi cho đến khi request cũ nhất hết hạn
sleep_time = self.requests[0] - (now - self.time_window)
if sleep_time > 0:
print(f"⏳ Rate limit: Đợi {sleep_time:.1f}s...")
time.sleep(sleep_time)
# Cập nhật lại queue sau khi sleep
while self.requests and self.requests[0] < time.time() - self.time_window:
self.requests.popleft()
self.requests.append(time.time())
def execute_with_retry(self, func, max_retries=3, backoff_factor=2):
"""Thực thi function với retry và rate limit handling"""
for attempt in range(max_retries):
self.wait_if_needed()
try:
return func()
except Exception as e:
if "429" in str(e) or "rate limit" in str(e).lower():
wait_time = backoff_factor ** attempt
print(f"⚠️ Rate limit hit. Đợi {wait_time}s trước khi thử lại...")
time.sleep(wait_time)
else:
raise
raise Exception(f"Thất bại sau {max_retries} lần thử")
Sử dụng
rate_limiter = RateLimitHandler(max_requests=30, time_window=60)
def safe_api_call(frame_base64):
return rate_limiter.execute_with_retry(
lambda: client.analyze_video_frame(frame_base64, BELT_INSPECTION_PROMPT)
)3. Lỗi "Connection Timeout" khi xử lý video stream
Mô tả lỗi: requests.exceptions.ReadTimeout: HTTPAdapter.poolmanager.connectionpool_kw
Nguyên nhân:
- RTSP stream không ổn định hoặc bị drop
- Mạng có độ trễ cao
- Server HolySheep đang bận
Cách khắc phục:
import socket
import cv2
from contextlib import contextmanager
class RTSPStreamManager:
"""Quản lý kết nối RTSP với automatic reconnection"""
def __init__(self, rtsp_url, timeout=10, max_retries=5):
self.rtsp_url = rtsp_url
self.timeout = timeout
self.max_retries = max_retries
self.cap = None
self.reconnect_delay = 2
def _set_socket_timeout(self):
"""Set timeout cho RTSP connection"""
# RTSP timeout workaround
socket.setdefaulttimeout(self.timeout)
def connect(self):
"""Kết nối với automatic reconnection"""
for attempt in range(self.max_retries):
try:
self._set_socket_timeout()
self.cap = cv2.VideoCapture(self.rtsp_url)
if self.cap.isOpened():
# Thiết lập buffer