Trong ngành logistics hiện đại, an toàn kho bãi là ưu tiên hàng đầu. Việc kiểm tra thủ công tốn kém, dễ bỏ sót và không thể giám sát 24/7. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống巡检 thông minh với chi phí giảm 85% so với API chính thức, sử dụng HolySheep AI làm nền tảng trung tâm.
Bảng so sánh: HolySheep vs API chính thức vs Relay services
| Tiêu chí | HolySheep AI | API chính thức | Relay services khác |
|---|---|---|---|
| GPT-4o (hình ảnh) | $8/1M tokens | $15/1M tokens | $10-12/1M tokens |
| DeepSeek V3.2 | $0.42/1M tokens | $0.42/1M tokens | $0.50-0.60/1M tokens |
| Độ trễ trung bình | <50ms | 100-300ms | 80-200ms |
| Thanh toán | WeChat/Alipay, USD | Chỉ thẻ quốc tế | Hạn chế |
| Tín dụng miễn phí | ✅ Có | ❌ Không | ❌ Không |
| Hỗ trợ video frame | ✅ Đầy đủ | ✅ Đầy đủ | ⚠️ Giới hạn |
Tổng quan giải pháp
Hệ thống bao gồm 3 module chính:
- Video Frame Extractor: Trích xuất frame từ camera IP/Số từ video CCTV
- GPT-4o Vision Analyzer: Phân tích từng frame để phát hiện bất thường
- DeepSeek Risk Classifier: Phân loại mức độ nguy hiểm và đề xuất xử lý
Kiến trúc hệ thống
Trước khi đi vào code chi tiết, hãy xem kiến trúc tổng thể:
+------------------+ +------------------+ +------------------+
| Camera IP | | Video Store | | Alert System |
| (RTSP/HLS) |---->| (Local/NAS) |---->| (Telegram/ |
+------------------+ +------------------+ | Email/SMS) |
+------------------+
^
|
+------------------+----------+
| |
+------v------+ +-------v-------+
| Frame Extract | | GPT-4o |
| (FFmpeg) |--->| Vision API |
+--------------+ +--------------+
|
+------v------+
| DeepSeek |
| Risk Class |
+-------------+
Triển khai chi tiết
1. Cài đặt và cấu hình ban đầu
# Cài đặt các thư viện cần thiết
pip install opencv-python requests Pillow python-dotenv schedule
Tạo file .env với API key của HolySheep
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Cấu hình camera
CAMERA_RTSP=rtsp://admin:[email protected]:554/stream1
CAMERA_INTERVAL=30 # Giây giữa mỗi lần kiểm tra
Cấu hình cảnh báo
ALERT_TELEGRAM_BOT_TOKEN=your_bot_token
ALERT_TELEGRAM_CHAT_ID=your_chat_id
EOF
echo "Đã tạo file cấu hình. Chỉnh sửa .env với thông tin của bạn."2. Module trích xuất frame từ video
import cv2
import base64
import os
from datetime import datetime
from pathlib import Path
class VideoFrameExtractor:
"""Trích xuất frame từ video stream hoặc file"""
def __init__(self, rtsp_url=None, output_dir="frames"):
self.rtsp_url = rtsp_url
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
def extract_frame_from_rtsp(self, timestamp=None):
"""Trích xuất 1 frame từ RTSP stream"""
if not self.rtsp_url:
raise ValueError("RTSP URL not configured")
cap = cv2.VideoCapture(self.rtsp_url)
if not cap.isOpened():
raise ConnectionError(f"Không thể kết nối camera: {self.rtsp_url}")
# Đọc frame
ret, frame = cap.read()
cap.release()
if not ret:
raise RuntimeError("Không thể đọc frame từ camera")
return self._process_frame(frame, timestamp)
def extract_frame_from_file(self, video_path, frame_position=0):
"""Trích xuất frame từ file video"""
cap = cv2.VideoCapture(str(video_path))
if frame_position > 0:
cap.set(cv2.CAP_PROP_POS_FRAMES, frame_position)
ret, frame = cap.read()
cap.release()
if not ret:
raise RuntimeError(f"Không thể đọc frame tại vị trí {frame_position}")
return self._process_frame(frame)
def _process_frame(self, frame, timestamp=None):
"""Xử lý và mã hóa frame"""
if timestamp is None:
timestamp = datetime.now()
# Chuyển đổi màu (OpenCV dùng BGR, cần RGB cho GPT-4o)
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
# Mã hóa thành JPEG
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 90]
_, buffer = cv2.imencode('.jpg', frame_rgb, encode_param)
# Mã hóa base64
base64_image = base64.b64encode(buffer).decode('utf-8')
# Lưu file local
filename = f"frame_{timestamp.strftime('%Y%m%d_%H%M%S')}.jpg"
filepath = self.output_dir / filename
cv2.imwrite(str(filepath), frame)
return {
'base64': base64_image,
'filepath': str(filepath),
'timestamp': timestamp.isoformat(),
'shape': frame.shape
}
Sử dụng
extractor = VideoFrameExtractor(rtsp_url="rtsp://admin:[email protected]:554/stream1")
frame_data = extractor.extract_frame_from_rtsp()
print(f"Đã trích xuất frame: {frame_data['filepath']}")3. Module phân tích frame với GPT-4o Vision
import requests
import json
import os
from dotenv import load_dotenv
load_dotenv()
class SafetyAnalyzer:
"""Phân tích frame an toàn sử dụng GPT-4o Vision qua HolySheep"""
# Prompt chuyên biệt cho kiểm tra an toàn kho bãi
SAFETY_PROMPT = """Bạn là chuyên gia kiểm tra an toàn logistics. Phân tích hình ảnh kho bãi và trả về JSON:
{
"has_issues": true/false,
"issues": [
{
"type": "的类型",
"severity": "critical/high/medium/low",
"location": "Vị trí trong ảnh",
"description": "Mô tả chi tiết vấn đề"
}
],
"summary": "Tóm tắt ngắn tình trạng an toàn"
}
Kiểm tra các vấn đề:
- Pallet đổ, hàng hóa chất đống không an toàn
- Khu vực cấm có người xâm phạm
- Lối thoát hiểm bị chặn
- Rò rỉ nước/chất lỏng
- Hỏa hoạn, khói
- Thiếu thiết bị PCCC
- Nhân viên không đeo thiết bị bảo hộ
"""
def __init__(self):
self.api_key = os.getenv('HOLYSHEEP_API_KEY')
self.base_url = os.getenv('HOLYSHEEP_BASE_URL', 'https://api.holysheep.ai/v1')
def analyze_frame(self, frame_data):
"""Gửi frame lên GPT-4o Vision để phân tích"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": self.SAFETY_PROMPT
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{frame_data['base64']}"
}
}
]
}
],
"max_tokens": 1024,
"temperature": 0.1
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise RuntimeError(f"API Error: {response.status_code} - {response.text}")
result = response.json()
content = result['choices'][0]['message']['content']
# Parse JSON từ response
# GPT có thể trả về trong code block
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
return json.loads(content.strip())
Sử dụng
analyzer = SafetyAnalyzer()
analysis_result = analyzer.analyze_frame(frame_data)
print(f"Kết quả phân tích:")
print(f"- Có vấn đề: {analysis_result['has_issues']}")
print(f"- Số lượng vấn đề: {len(analysis_result.get('issues', []))}")
print(f"- Tóm tắt: {analysis_result['summary']}")4. Module phân loại rủi ro với DeepSeek
import requests
import json
from datetime import datetime
class RiskClassifier:
"""Phân loại và xếp hạng rủi ro sử dụng DeepSeek"""
RISK_PROMPT = """Bạn là chuyên gia đánh giá rủi ro an toàn kho bãi. Dựa trên các vấn đề được phát hiện, hãy:
1. Xác định mức độ ưu tiên xử lý (1-5)
2. Đề xuất thời gian xử lý cho từng vấn đề
3. Ước tính chi phí nếu không xử lý
4. Đề xuất biện pháp khắc phục
Trả về JSON:
{
"risk_level": "critical/high/medium/low/minimal",
"risk_score": 0-100,
"priority_issues": [
{
"issue": "Mô tả vấn đề",
"priority": 1-5,
"deadline": "Khuyến nghị thời gian xử lý",
"estimated_cost_if_ignored": "Chi phí ước tính",
"action": "Biện pháp khắc phục"
}
],
"overall_recommendation": "Khuyến nghị tổng thể"
}
"""
def __init__(self):
self.api_key = os.getenv('HOLYSHEEP_API_KEY')
self.base_url = os.getenv('HOLYSHEEP_BASE_URL', 'https://api.holysheep.ai/v1')
def classify_risk(self, analysis_result):
"""Phân loại rủi ro dựa trên kết quả phân tích"""
if not analysis_result.get('has_issues'):
return {
"risk_level": "minimal",
"risk_score": 5,
"priority_issues": [],
"overall_recommendation": "Không phát hiện vấn đề an toàn nghiêm trọng. Tiếp tục giám sát bình thường."
}
# Tạo context từ các vấn đề
issues_text = json.dumps(analysis_result.get('issues', []), indent=2, ensure_ascii=False)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": "Bạn là chuyên gia đánh giá rủi ro logistics với 10 năm kinh nghiệm."
},
{
"role": "user",
"content": f"Các vấn đề được phát hiện:\n{issues_text}\n\n{self.RISK_PROMPT}"
}
],
"max_tokens": 1500,
"temperature": 0.3
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise RuntimeError(f"DeepSeek API Error: {response.status_code}")
result = response.json()
content = result['choices'][0]['message']['content']
# Parse JSON
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
return json.loads(content.strip())
Sử dụng
classifier = RiskClassifier()
risk_result = classifier.classify_risk(analysis_result)
print(f"Mức độ rủi ro: {risk_result['risk_level'].upper()}")
print(f"Điểm rủi ro: {risk_result['risk_score']}/100")
print(f"Khuyến nghị: {risk_result['overall_recommendation']}")5. Module tạo báo cáo và cảnh báo
import json
from datetime import datetime
from pathlib import Path
import requests
class ReportGenerator:
"""Tạo báo cáo và gửi cảnh báo"""
def __init__(self):
self.report_dir = Path("reports")
self.report_dir.mkdir(parents=True, exist_ok=True)
def generate_html_report(self, frame_data, analysis, risk):
"""Tạo báo cáo HTML chi tiết"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
html = f"""
<!DOCTYPE html>
<html lang="vi">
<head>
<meta charset="UTF-8">
<title>Báo cáo Kiểm tra An toàn - {timestamp}</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.header {{ background: #2c3e50; color: white; padding: 20px; }}
.critical {{ background: #e74c3c; color: white; }}
.high {{ background: #f39c12; color: white; }}
.medium {{ background: #f1c40f; }}
.low {{ background: #27ae60; color: white; }}
table {{ border-collapse: collapse; width: 100%; margin: 20px 0; }}
th, td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
th {{ background-color: #34495e; color: white; }}
img {{ max-width: 400px; border: 2px solid #ddd; }}
</style>
</head>
<body>
<div class="header">
<h1>🏭 Báo cáo Kiểm tra An toàn Kho bãi</h1>
<p>Thời gian: {timestamp}</p>
<p>Camera: {frame_data.get('camera_id', 'N/A')}</p>
</div>
<h2>📊 Tổng quan</h2>
<table>
<tr>
<th>Chỉ số</th>
<th>Giá trị</th>
</tr>
<tr>
<td>Mức độ rủi ro</td>
<td class="{risk.get('risk_level', 'low')}">
{risk.get('risk_level', 'unknown').upper()}
(Điểm: {risk.get('risk_score', 0)}/100)
</td>
</tr>
<tr>
<td>Số vấn đề phát hiện</td>
<td>{len(analysis.get('issues', []))}</td>
</tr>
<tr>
<td>Tình trạng</td>
<td>{"⚠️ CẢNH BÁO" if analysis.get('has_issues') else "✅ AN TOÀN"}</td>
</tr>
</table>
<h2>📷 Hình ảnh ghi nhận</h2>
<img src="{frame_data.get('image_path', '')}" alt="Frame capture">
<h2>🚨 Chi tiết vấn đề</h2>
<table>
<tr>
<th>Loại</th>
<th>Mức độ</th>
<th>Vị trí</th>
<th>Mô tả</th>
</tr>
"""
for issue in analysis.get('issues', []):
html += f"""
<tr>
<td>{issue.get('type', 'N/A')}</td>
<td class="{issue.get('severity', 'low')}">{issue.get('severity', 'N/A').upper()}</td>
<td>{issue.get('location', 'N/A')}</td>
<td>{issue.get('description', 'N/A')}</td>
</tr>
"""
html += """
💡 Khuyến nghị
""" + risk.get('overall_recommendation', 'Tiếp tục giám sát') + """
📋 Vấn đề ưu tiên xử lý
Ưu tiên
Vấn đề
Thời hạn
Biện pháp
"""
for idx, priority in enumerate(risk.get('priority_issues', []), 1):
html += f"""
{idx}
{priority.get('issue', 'N/A')}
{priority.get('deadline', 'N/A')}
{priority.get('action', 'N/A')}
"""
html += """