Là một kỹ sư đã triển khai hơn 50 dự án IoT trong 5 năm qua, tôi nhận ra rằng việc đưa AI đến gần nơi dữ liệu được tạo ra — thay vì phụ thuộc hoàn toàn vào đám mây — là xu hướng tất yếu. Bài viết này sẽ hướng dẫn bạn từng bước cách thiết kế kiến trúc Edge AI với AWS Greengrass, ngay cả khi bạn chưa từng động đến API hay cloud trước đó.

Mục Lục

Edge AI Là Gì Và Tại Sao Bạn Cần Nó

Edge AI nghĩa là chạy AI ngay tại thiết bị gần bạn — như camera an ninh, cảm biến nhà máy, hay robot — thay vì gửi dữ liệu lên server đám mây rồi chờ phản hồi. Phương pháp này mang lại 3 lợi ích quan trọng:

Gợi ý ảnh chụp màn hình: Sơ đồ so sánh Cloud AI vs Edge AI về độ trễ và chi phí

AWS Greengrass Giải Quyết Vấn Đề Gì

AWS Greengrass là phần mềm của Amazon giúp bạn chạy Lambda functions, container, và machine learning models ngay trên thiết bị cục bộ. Nó đóng vai trò như một "bộ não trung gian" giữa thiết bị IoT và đám mây AWS.

Trong kinh nghiệm triển khai thực tế của tôi, AWS Greengrass đặc biệt hữu ích khi:

Kiến Trúc Cơ Bản Của Hệ Thống Edge AI

Kiến trúc tôi hay sử dụng cho các dự án nhỏ và vừa gồm 4 tầng:

Gợi ý ảnh chụp màn hình: Sơ đồ kiến trúc 4 tầng với các luồng dữ liệu

Cài Đặt Môi Trường Từ Con Số 0

Bước 1: Chuẩn bị thiết bị

Tôi khuyên bắt đầu với Raspberry Pi 4 (4GB RAM) hoặc Ubuntu VM nếu bạn muốn test nhanh trên laptop. Chi phí thiết bị test khoảng $50-100.

# Cập nhật hệ thống Ubuntu/Debian
sudo apt update && sudo apt upgrade -y

Cài đặt Python 3.8+ và pip

sudo apt install -y python3 python3-pip python3-venv

Kiểm tra phiên bản

python3 --version

Output mong đợi: Python 3.10.x hoặc cao hơn

Bước 2: Tạo tài khoản AWS và cấu hình IAM

Đăng nhập AWS Console, tạo IAM user với quyền:

# Cài đặt AWS CLI
curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install

Cấu hình credentials

aws configure

AWS Access Key ID: [Nhập của bạn]

AWS Secret Access Key: [Nhập của bạn]

Default region name: ap-southeast-1 (Singapore) hoặc us-east-1

Default output format: json

Bước 3: Cài đặt AWS Greengrass Core

# Tải Greengrass installer
curl -s https://d2s8p88vq9s1eb.cloudfront.net/greengrass/v2/latest/linux-arm64-installer.tar.gz -o greengrass-installer.tar.gz

Giải nén

sudo tar -xzf greengrass-installer.tar.gz -C /tmp

Cài đặt (thay YOUR_THING_NAME bằng tên device của bạn)

sudo -E env "PATH=$PATH" /tmp/installer/installer.sh \ --root-path /greengrass/v2 \ --thing-name my-edge-ai-device \ --component-default-user root \ --provision true \ --setup-system-service true \ --deploy-devtools true

Kiểm tra Greengrass đã chạy chưa

sudo systemctl status greengrass

Gợi ý ảnh chụp màn hình: AWS Console > IoT Core > Things > my-edge-ai-device đang ở trạng thái "Connected"

Triển Khai Mô Hình AI Thực Chiến

Tạo Custom Component cho Edge Inference

Đây là phần quan trọng nhất — tôi sẽ hướng dẫn cách deploy một mô hình image classification chạy cục bộ.

# Tạo cấu trúc thư mục component
mkdir -p ~/greengrass/components/com.edgeai.image-classifier/artifacts
mkdir -p ~/greengrass/components/com.edgeai.image-classifier/recipes

Tạo recipe file (recipes/recipe.yaml)

cat > ~/greengrass/components/com.edgeai.image-classifier/recipes/recipe.yaml << 'EOF' RecipeFormatVersion: "2020-01-25" ComponentName: "com.edgeai.image-classifier" ComponentVersion: "1.0.0" ComponentConfiguration: DefaultConfiguration: inference: model_path: "/model/mobilenet_v2.tflite" confidence_threshold: 0.75 ComponentType: "Lambda" LambdaExecutionParameters: Timeout: 30 MemorySize: 512 Artifacts: - URI: "s3://BUCKET_NAME/model/mobilenet_v2.tflite" Permission: Read: "READ" Lifecycle: Run: "python3 {artifacts:path}/inference.py" EOF echo "Recipe created successfully!"
# Tạo Python inference script (artifacts/inference.py)
import json
import time
import os

def lambda_handler(event, context):
    """Xử lý inference cục bộ trên Edge device"""
    
    # Đọc cấu hình từ event
    image_data = event.get("image_data", "")
    threshold = float(os.environ.get('CONFIDENCE_THRESHOLD', '0.75'))
    
    # === THAY THẾ: Tích hợp HolyShehe AI cho inference thông minh ===
    # Khi cần xử lý phức tạp hoặc mô hình lớn, gọi API
    if event.get("use_cloud_inference", False):
        result = call_holysheep_api(image_data)
        return {"status": "cloud", "result": result}
    
    # Inference cục bộ với mô hình nhẹ
    start_time = time.time()
    local_result = local_inference(image_data, threshold)
    latency = (time.time() - start_time) * 1000
    
    return {
        "status": "success",
        "predictions": local_result,
        "latency_ms": round(latency, 2),
        "inference_location": "edge"
    }

def local_inference(image_data, threshold):
    """Mock inference - thay bằng TensorFlow Lite inference thực tế"""
    # Trong production, đây sẽ là:
    # import tflite_runtime.interpreter as tflite
    # interpreter = tflite.Interpreter(model_path)
    return [
        {"class": "cat", "confidence": 0.92},
        {"class": "dog", "confidence": 0.05},
        {"class": "bird", "confidence": 0.02}
    ]

def call_holysheep_api(image_data):
    """Gọi HolyShehe AI API cho inference phức tạp"""
    import urllib.request
    import urllib.error
    
    api_url = "https://api.holysheep.ai/v1/chat/completions"
    
    payload = json.dumps({
        "model": "gpt-4.1",
        "messages": [
            {"role": "user", "content": f"Analyze this image: {image_data[:100]}..."}
        ],
        "temperature": 0.3
    }).encode('utf-8')
    
    req = urllib.request.Request(
        api_url,
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY', 'YOUR_HOLYSHEEP_API_KEY')}"
        },
        method="POST"
    )
    
    try:
        with urllib.request.urlopen(req, timeout=10) as response:
            return json.loads(response.read().decode('utf-8'))
    except urllib.error.URLError as e:
        return {"error": str(e), "fallback": "using_local_model"}

print("Inference component loaded successfully!")
# Deploy component lên device
aws greengrassv2 create-deployment \
  --target-arn "arn:aws:iot:ap-southeast-1:123456789012:thing/my-edge-ai-device" \
  --components '{"com.edgeai.image-classifier": {"componentVersion": "1.0.0"}}' \
  --deployment-name "edge-ai-deployment-$(date +%Y%m%d)"

Theo dõi deployment

echo "Checking deployment status..." sleep 10 aws greengrassv2 get-deployment \ --target-arn "arn:aws:iot:ap-southeast-1:123456789012:thing/my-edge-ai-device" \ --query "deploymentStatus"

Trạng thái "IN_PROGRESS" nghĩa là đang deploy

Gợi ý ảnh chụp màn hình: AWS IoT Greengrass > Components > com.edgeai.image-classifier hiển thị version 1.0.0

Tích Hợp HolyShehe AI Cho Inference Thông Minh

Trong các dự án thực tế, tôi thường kết hợp Edge AI cục bộ với HolyShehe AI API vì những lý do sau:

Đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.

# Cấu hình HolyShehe AI client trên Edge device

File: /greengrass/components/com.edgeai.smart-analyzer/artifacts/client.py

import json import urllib.request import urllib.error import base64 import time class HolySheheAIClient: """Client tối ưu cho Edge-Greengrass integration""" BASE_URL = "https://api.holysheep.ai/v1" def __init__(self, api_key: str): self.api_key = api_key self.models = { "fast": "gemini-2.5-flash", # $2.50/Mtok - nhanh, rẻ "balanced": "deepseek-v3.2", # $0.42/Mtok - tiết kiệm nhất "powerful": "gpt-4.1" # $8/Mtok - chất lượng cao } def analyze_image_edge(self, image_base64: str, context: str) -> dict: """Phân tích ảnh với AI, quyết định xem có cần cloud support không""" # Bước 1: Inference nhanh cục bộ local_result = self.local_quick_check(image_base64) if local_result["confidence"] >= 0.9: # Model cục bộ tự tin → trả ngay return {"source": "edge", "result": local_result} # Bước 2: Cần AI mạnh hơn → gọi HolyShehe return self.call_holysheep(context, image_base64) def call_holysheep(self, context: str, image_data: str) -> dict: """Gọi HolyShehe AI API với error handling đầy đủ""" payload = { "model": self.models["balanced"], # Dùng DeepSeek tiết kiệm chi phí "messages": [ {"role": "system", "content": "Bạn là chuyên gia phân tích hình ảnh công nghiệp."}, {"role": "user", "content": f"Context: {context}\nImage: {image_data[:500]}...\nPhân tích và đưa ra quyết định."} ], "temperature": 0.3, "max_tokens": 500 } start_time = time.time() try: req = urllib.request.Request( f"{self.BASE_URL}/chat/completions", data=json.dumps(payload).encode('utf-8'), headers={ "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" }, method="POST" ) with urllib.request.urlopen(req, timeout=30) as response: result = json.loads(response.read().decode('utf-8')) latency_ms = (time.time() - start_time) * 1000 return { "source": "holysheep", "result": result["choices"][0]["message"]["content"], "latency_ms": round(latency_ms, 2), "model_used": self.models["balanced"] }