Azure IoT Edge 集成大模型 API 完整教程

Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi triển khai Azure IoT Edge kết hợp với HolySheep AI cho một dự án IoT thông minh tại doanh nghiệp thương mại điện tử. Dự án ban đầu gặp vấn đề về độ trễ phản hồi khi xử lý 50,000+ yêu cầu từ chatbot chăm sóc khách hàng mỗi ngày. Sau khi tích hợp LLM trực tiếp trên Edge gateway, độ trễ giảm từ 800ms xuống còn 45ms — một cải thiện đáng kinh ngạc.

Tại sao cần tích hợp LLM trên Azure IoT Edge?

Azure IoT Edge cho phép chạy workload AI ngay trên thiết bị Edge thay vì phải gửi dữ liệu lên cloud. Khi kết hợp với HolySheep AI — nền tảng API LLM với độ trễ dưới 50ms và chi phí chỉ từ $0.42/MTok — bạn có một giải pháp vừa nhanh vừa tiết kiệm.

Lợi ích chính:

• Giảm độ trễ phản hồi từ 800ms xuống dưới 50ms
• Tiết kiệm chi phí bandwidth — dữ liệu không cần upload lên cloud
• Xử lý offline khi mất kết nối internet
• Tỷ giá ¥1=$1, tiết kiệm 85%+ so với các provider khác

Kiến trúc hệ thống

Kiến trúc tổng thể gồm 3 tầng:

Tầng 1: IoT Devices (Sensors, Cameras, PLC)
         ↓
Tầng 2: Azure IoT Edge Gateway (Local LLM Cache)
         ↓
Tầng 3: HolySheep AI Cloud API (Fallback/Heavy Processing)
         ↓
Tầng 4: Dashboard & Analytics

Chuẩn bị môi trường

1. Cài đặt Azure IoT Edge trên Ubuntu 22.04

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

Cài đặt Docker Engine
curl -fssl https://get.docker.com | sh
sudo usermod -aG docker $USER

Cài đặt Azure IoT Edge runtime
curl https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor > microsoft.gpg
sudo mv microsoft.gpg /etc/apt/trusted.gpg.d/microsoft.gpg
sudo apt-get update
sudo apt-get install aziot-edge -y

Kiểm tra version
aziot-edge --version

2. Tạo IoT Hub và đăng ký Edge Device

# Kết nối Azure CLI
az login
az account set --subscription "your-subscription-id"

Tạo Resource Group
az group create --name iot-edge-rg --location eastus

Tạo IoT Hub
az iot hub create --resource-group iot-edge-rg --name holysheep-iot-hub --sku S1

Tạo Edge Device
az iot hub device-identity create \
  --hub-name holysheep-iot-hub \
  --device-id edge-gateway-01 \
  --edge-enabled

Lấy connection string
az iot hub device-identity connection-string show \
  --hub-name holysheep-iot-hub \
  --device-id edge-gateway-01

3. Cấu hình config.yaml cho IoT Edge

# /etc/aziot/config.yaml
provisioning:
  source: "manual"
  device_connection_string: "HostName=holysheep-iot-hub.azure-devices.net;DeviceId=edge-gateway-01;SharedAccessKey=xxxxxxx"
  dynamic_reprovisioning: false

agent:
  name: "edgeAgent"
  type: "docker"
  env:
    EGHOME: "."
  config:
    image: "mcr.microsoft.com/azureiotedge-agent:1.4"
    auth: {}

hostname: "edge-gateway-01"

watchdog:
  max_retries: "infinite"

connect:
  management_uri: "unix:///var/run/iotedge/mgmt.sock"
  workload_uri: "unix:///var/run/iotedge/workload.sock"

listen:
  management_uri: "unix:///var/run/iotedge/mgmt.sock"
  workload_uri: "unix:///var/run/iotedge/workload.sock"

moby_runtime:
  uri: "unix:///var/run/docker.sock"
  network: "azure-iot-edge"

# Áp dụng cấu hình và khởi động service
sudo iotedge config apply
sudo systemctl restart iotedge

Kiểm tra trạng thái
sudo iotedge system status
sudo iotedge list

Tạo Module LLM Edge Service

4. Viết Python Module kết nối HolySheep AI

Tạo thư mục và file Dockerfile cho custom module:

# Dockerfile.edge.llm
FROM python:3.11-slim

WORKDIR /app

Cài đặt dependencies
RUN pip install --no-cache-dir \
    fastapi==0.109.0 \
    uvicorn==0.27.0 \
    httpx==0.26.0 \
    redis==5.0.1 \
    pydantic==2.5.3

Copy source code
COPY app.py /app/
COPY requirements.txt /app/

ENV PYTHONUNBUFFERED=1

EXPOSE 5000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "5000"]

# app.py - LLM Edge Gateway Module
import os
import json
import hashlib
from datetime import datetime, timedelta
from typing import Optional, List
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import httpx

app = FastAPI(title="HolySheep LLM Edge Gateway")

Cấu hình - SỬ DỤNG HOLYSHEEP AI
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
MODEL_NAME = os.getenv("LLM_MODEL", "gpt-4.1")
CACHE_TTL = int(os.getenv("CACHE_TTL", "3600"))

In-memory cache cho responses phổ biến
response_cache = {}

class ChatMessage(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    temperature: float = 0.7
    max_tokens: int = 1000
    stream: bool = False

class ChatResponse(BaseModel):
    model: str
    content: str
    usage: dict
    cached: bool = False
    latency_ms: float

def get_cache_key(messages: List[ChatMessage], temperature: float) -> str:
    """Tạo cache key từ request content"""
    content_hash = hashlib.md5(
        json.dumps({
            "messages": [{"role": m.role, "content": m.content} for m in messages],
            "temperature": temperature
        }, sort_keys=True).encode()
    ).hexdigest()
    return content_hash

async def call_holysheep_api(messages: List[ChatMessage], temperature: float, max_tokens: int):
    """Gọi HolySheep AI API với retry logic"""
    headers = {
        "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": MODEL_NAME,
        "messages": [{"role": m.role, "content": m.content} for m in messages],
        "temperature": temperature,
        "max_tokens": max_tokens
    }
    
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            f"{HOLYSHEEP_BASE_URL}/chat/completions",
            headers=headers,
            json=payload
        )
        
        if response.status_code != 200:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"HolySheep API Error: {response.text}"
            )
        
        return response.json()

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    """Endpoint chính - xử lý chat completion với caching"""
    start_time = datetime.now()
    
    # Kiểm tra cache
    cache_key = get_cache_key(request.messages, request.temperature)
    if cache_key in response_cache:
        cached_response = response_cache[cache_key]
        if datetime.now() < cached_response["expires"]:
            latency = (datetime.now() - start_time).total_seconds() * 1000
            return JSONResponse({
                **cached_response["data"],
                "cached": True,
                "latency_ms": round(latency, 2)
            })
    
    # Gọi HolySheep API
    try:
        result = await call_holysheep_api(
            request.messages,
            request.temperature,
            request.max_tokens
        )
        
        # Lưu vào cache
        response_cache[cache_key] = {
            "data": result,
            "expires": datetime.now() + timedelta(seconds=CACHE_TTL)
        }
        
        latency = (datetime.now() - start_time).total_seconds() * 1000
        return JSONResponse({
            **result,
            "cached": False,
            "latency_ms": round(latency, 2)
        })
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint cho IoT Edge"""
    return {
        "status": "healthy",
        "model": MODEL_NAME,
        "cache_size": len(response_cache),
        "timestamp": datetime.now().isoformat()
    }

@app.get("/v1/models")
async def list_models():
    """Trả về thông tin model đang sử dụng"""
    return {
        "object": "list",
        "data": [{
            "id": MODEL_NAME,
            "object": "model",
            "created": 1677610602,
            "owned_by": "holysheep"
        }]
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

5. Triển khai Module lên IoT Edge

# Tạo deployment manifest
cat > deployment.edge.json << 'EOF'
{
  "schemaVersion": "1.4",
  "runtime": {
    "settings": {
      "loggingDriver": "json-file"
    }
  },
  "systemModules": {
    "edgeAgent": {
      "settings": {
        "image": "mcr.microsoft.com/azureiotedge-agent:1.4",
        "createOptions": "{}"
      }
    },
    "edgeHub": {
      "settings": {
        "image": "mcr.microsoft.com/azureiotedge-hub:1.4",
        "createOptions": "{\"HostConfig\":{\"PortBindings\":{\"443/tcp\":[{\"HostPort\":\"443\"}],\"5671/tcp\":[{\"HostPort\":\"5671\"}],\"8883/tcp\":[{\"HostPort\":\"8883\"}]}}}"
      },
      "type": "docker",
      "status": "running",
      "restartPolicy": "always"
    }
  },
  "moduleTypes": {
    "type": "docker",
    "settings": {
      "image": "${CONTAINER_REGISTRY}/llm-edge-gateway:1.0.0",
      "createOptions": "{\"HostConfig\":{\"PortBindings\":{\"5000/tcp\":[{\"HostPort\":\"5000\"}]}}}"
    },
    "env": {
      "HOLYSHEEP_API_KEY": {
        "value": "YOUR_HOLYSHEEP_API_KEY"
      },
      "LLM_MODEL": {
        "value": "gpt-4.1"
      }
    }
  },
  "modules": {
    "llm-edge-gateway": {
      "version": "1.0.0",
      "type": "docker",
      "status": "running",
      "restartPolicy": "always",
      "settings": {
        "image": "${CONTAINER_REGISTRY}/llm-edge-gateway:1.0.0",
        "createOptions": "{\"HostConfig\":{\"PortBindings\":{\"5000/tcp\":[{\"HostPort\":\"5000\"}]}}}"
      },
      "env": {
        "HOLYSHEEP_API_KEY": {
          "value": "YOUR_HOLYSHEEP_API_KEY"
        },
        "LLM_MODEL": {
          "value": "gpt-4.1"
        }
      }
    }
  }
}
EOF

Deploy lên Edge device
az iot edge set-modules \
  --hub-name holysheep-iot-hub \
  --device-id edge-gateway-01 \
  --content deployment.edge.json

Theo dõi trạng thái deployment
az iot edge show-module-twin \
  --hub-name holysheep-iot-hub \
  --device-id edge-gateway-01 \
  --module-name llm-edge-gateway

Tích hợp với RAG System trên Edge

6. Xây dựng Edge RAG Module

# edge_rag_module.py - RAG Inference trên Edge
from typing import List, Tuple, Optional
import numpy as np
from sentence_transformers import SentenceTransformer
import chromadb
from chromadb.config import Settings

class EdgeRAG:
    """RAG System chạy trên Azure IoT Edge"""
    
    def __init__(self, 
                 embedding_model: str = "all-MiniLM-L6-v2",
                 collection_name: str = "product_knowledge",
                 holysheep_api_key: str = "YOUR_HOLYSHEEP_API_KEY"):
        self.embedding_model = SentenceTransformer(embedding_model)
        self.client = chromadb.Client(Settings(
            persist_directory="/data/chroma_db",
            anonymized_telemetry=False
        ))
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            metadata={"description": "Product knowledge base"}
        )
        self.api_key = holysheep_api_key
        
    def add_documents(self, documents: List[str], metadatas: List[dict] = None):
        """Thêm documents vào vector database"""
        embeddings = self.embedding_model.encode(documents)
        ids = [f"doc_{i}" for i in range(len(documents))]
        self.collection.add(
            embeddings=embeddings.tolist(),
            documents=documents,
            ids=ids,
            metadatas=metadatas or [{"source": "manual"} for _ in documents]
        )
        
    def retrieve_context(self, query: str, top_k: int = 3) -> List[Tuple[str, dict, float]]:
        """Tìm kiếm documents liên quan đến query"""
        query_embedding = self.embedding_model.encode([query])
        results = self.collection.query(
            query_embeddings=query_embedding.tolist(),
            n_results=top_k
        )
        
        contexts = []
        for i in range(len(results['documents'][0])):
            contexts.append((
                results['documents'][0][i],
                results['metadatas'][0][i],
                results['distances'][0][i]
            ))
        return contexts
    
    def build_prompt(self, query: str, system_prompt: str = None) -> str:
        """Build prompt với retrieved context"""
        contexts = self.retrieve_context(query)
        
        context_text = "\n\n".join([
            f"[Document {i+1}] {doc}" 
            for i, (doc, _, _) in enumerate(contexts)
        ])
        
        if system_prompt is None:
            system_prompt = """Bạn là trợ lý AI chăm sóc khách hàng. 
Dựa trên thông tin được cung cấp trong ngữ cảnh, hãy trả lời câu hỏi một cách chính xác.
Nếu không có thông tin trong ngữ cảnh, hãy nói rằng bạn không biết."""
        
        full_prompt = f"""{system_prompt}

Ngữ cảnh:
{context_text}

Câu hỏi: {query}

Câu trả lời:"""
        return full_prompt

Sử dụng Edge RAG với HolySheep API
async def rag_chat(EdgeRAG rag_system, query: str):
    """Chat với RAG system qua HolySheep API"""
    import httpx
    
    prompt = rag_system.build_prompt(query)
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://localhost:5000/v1/chat/completions",
            json={
                "model": "gpt-4.1",
                "messages": [
                    {"role": "user", "content": prompt}
                ],
                "temperature": 0.3,
                "max_tokens": 500
            },
            headers={"X-API-Key": rag_system.api_key}
        )
        return response.json()

Khởi tạo và test
if __name__ == "__main__":
    import asyncio
    
    rag = EdgeRAG(
        embedding_model="all-MiniLM-L6-v2",
        collection_name="ecommerce_products"
    )
    
    # Thêm sample knowledge base
    rag.add_documents([
        "Sản phẩm laptop ASUS ZenBook có bảo hành 24 tháng",
        "Chính sách đổi trả trong 30 ngày với sản phẩm chưa qua sử dụng",
        "Miễn phí vận chuyển cho đơn hàng từ 500.000 VNĐ"
    ])
    
    # Test query
    async def test():
        result = await rag_chat(rag, "Laptop ASUS có bảo hành bao lâu?")
        print(result)
    
    asyncio.run(test())

Bảng so sánh chi phí

Provider	Giá/MTok	50K requests/ngày	Tiết kiệm
OpenAI GPT-4	$30	$1,500	-
Anthropic Claude	$15	$750	-
HolySheep AI	$0.42	$21	85%+

Với HolySheep AI, dự án của tôi tiết kiệm được $1,479/ngày (~¥10,600) khi xử lý khối lượng lớn yêu cầu. Đặc biệt, tỷ giá