軽量エージェントのエッジデプロイ
Raspberry Pi やエッジデバイス上で小規模モデルを実行し、低遅延で応答させます。
「軽量エージェントのエッジデプロイ」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エッジAIエージェント
エッジエージェントは、クラウドではなく、Raspberry Pi、Jetson Nano、産業用PCなどのIoTデバイスやローカルゲートウェイ上で直接動作します。利点は、ラウンドトリップが不要なため低遅延であること、オフラインでも動作すること、帯域幅コストが低いことです。一方で、計算能力とRAMに制約があるため、より小型で効率的なモデルが必要になります。
エッジデプロイ用モデルの選択
エッジデバイスではGPT-4oやClaude Opusを実行できません。Raspberry Pi 5（RAM 8 GB）に収まる小型モデルには、Phi-3-mini（38億パラメーター）、Gemma-2B、TinyLlama-1.1Bがあります。これらのモデルを4ビット量子化すると、必要なRAMは1～3 GBになり、CPU上で毎秒5～15トークンの速度で動作します。
# Model size reference for edge selection:
EDGE_MODELS = {
'tinyllama-1.1b-q4': {
'params': '1.1B', 'quantization': 'Q4_K_M',
'ram_gb': 0.8, 'tokens_per_sec_cpu': 15,
'use_case': 'simple classification, keyword detection'
},
'phi-3-mini-q4': {
'params': '3.8B', 'quantization': 'Q4_K_M',
'ram_gb': 2.5, 'tokens_per_sec_cpu': 8,
'use_case': 'reasoning, multi-step decisions'
},
'gemma-2b-q4': {
'params': '2B', 'quantization': 'Q4_K_M',
'ram_gb': 1.5, 'tokens_per_sec_cpu': 10,
'use_case': 'general assistant tasks'
}
}
for name, info in EDGE_MODELS.items():
print(f'{name}: {info["ram_gb"]}GB RAM, '
f'{info["tokens_per_sec_cpu"]} tok/s — {info["use_case"]}')Raspberry PiへのOllamaのインストール
Ollamaは、ローカルハードウェア上で小型LLMを実行する最も簡単な方法です。モデルのダウンロード、量子化、OpenAI SDK互換のローカルREST APIを管理します。1つのコマンドでインストールでき、別のコマンドでモデルを取得して起動できます。
# Install Ollama (run on the Raspberry Pi terminal):
# curl -fsSL https://ollama.ai/install.sh | sh
# Pull and run a model:
# ollama pull phi3:mini
# ollama serve (starts API on localhost:11434)
# Python client — uses the OpenAI-compatible endpoint:
from openai import OpenAI
local_client = OpenAI(
base_url='http://localhost:11434/v1',
api_key='ollama' # Ollama ignores this but it is required by the SDK
)
def local_inference(prompt: str, model: str = 'phi3:mini') -> str:
response = local_client.chat.completions.create(
model=model,
messages=[{'role': 'user', 'content': prompt}],
max_tokens=256,
temperature=0.1
)
return response.choices[0].message.content
result = local_inference('Is temperature 45C dangerous for a server room?')
print(result)4ビットGGUF量子化
GGUF（GPT-Generated Unified Format）は、llama.cppとOllamaが量子化モデルに使用するファイル形式です。Q4_K_Mは4ビットの混合量子化を意味し、fp32と比べて約75%小さい一方、エッジでの推論タスクにおける品質低下は最小限に抑えられます。
# Understanding quantisation quality levels:
QUANTISATION_GUIDE = {
'Q2_K': {'size_multiplier': 0.25, 'quality': 'poor', 'ram': 'minimal'},
'Q4_K_M': {'size_multiplier': 0.45, 'quality': 'good', 'ram': 'low'},
'Q5_K_M': {'size_multiplier': 0.55, 'quality': 'better', 'ram': 'medium'},
'Q8_0': {'size_multiplier': 0.75, 'quality': 'near-original', 'ram': 'high'},
'F16': {'size_multiplier': 1.0, 'quality': 'original', 'ram': 'full'}
}
# For edge: Q4_K_M is the sweet spot
# 7B model: 7B * 4bit/8 = 3.5 GB in Q4 vs 14 GB in F16
def estimate_vram_gb(param_billions: float, quant: str) -> float:
multiplier = QUANT_GUIDE = {
'Q4_K_M': 0.45, 'Q8_0': 0.75, 'F16': 1.0
}
return round(param_billions * multiplier.get(quant, 0.5), 2)
print(f'Phi-3-mini Q4_K_M: {estimate_vram_gb(3.8, "Q4_K_M")} GB')
print(f'Phi-3-mini F16: {estimate_vram_gb(3.8, "F16")} GB')CPUでの推論速度の最適化
CPUのみのエッジデバイスでは、推論速度はスレッド数とプロンプトキャッシュに左右されます。スレッド数をCPUコア数に合わせ、システムプロンプトは短く保ちます（キャッシュされない場合、呼び出しのたびに再処理されます）。可能な場合は、繰り返し行うクエリをバッチ処理してください。
import os
import time
# Ollama environment variables for performance tuning
# Set before starting the Ollama service:
# OLLAMA_NUM_PARALLEL=1 (single request at a time on small devices)
# OLLAMA_MAX_LOADED_MODELS=1 (only keep one model in RAM)
def timed_inference(prompt: str, client, model: str = 'phi3:mini') -> dict:
start = time.time()
response = client.chat.completions.create(
model=model,
messages=[{'role': 'user', 'content': prompt}],
max_tokens=128
)
elapsed = time.time() - start
text = response.choices[0].message.content
tokens = len(text.split()) # approximate
return {
'text': text,
'elapsed_s': round(elapsed, 2),
'approx_tps': round(tokens / elapsed, 1)
}
result = timed_inference('Classify: temperature=45C, normal range 18-30C. Action?',
local_client)
print(result)ハイブリッドエッジ・クラウドアーキテクチャ
最適なアーキテクチャでは、高速でリスクの低い判断（分類やしきい値ロジック）にはエッジモデルを使用し、複雑で頻度が低い、またはリスクの高い判断にはクラウドモデルと同期します。エッジエージェントは複雑なクエリをキューに入れ、接続可能になった時点で送信します。
import anthropic
import time
class HybridAgent:
def __init__(self, edge_client, cloud_api_key: str):
self.edge = edge_client # Ollama local client
self.cloud = anthropic.Anthropic(api_key=cloud_api_key)
self.pending_cloud_queries = []
def decide(self, context: dict) -> str:
# Fast path: edge model for simple binary decisions
simple_prompt = (
f'Sensor: {context}. '
'Respond with exactly one word: NORMAL or ALERT.'
)
edge_result = self.edge.chat.completions.create(
model='phi3:mini',
messages=[{'role': 'user', 'content': simple_prompt}],
max_tokens=5
).choices[0].message.content.strip()
if edge_result == 'ALERT':
# Queue complex analysis for cloud
self.pending_cloud_queries.append(context)
return edge_result
def sync_to_cloud(self):
"""Call when online connectivity is available."""
for ctx in self.pending_cloud_queries:
result = self.cloud.messages.create(
model='claude-opus-4-5', max_tokens=512,
messages=[{'role': 'user', 'content':
f'Full analysis of: {ctx}'}]
)
print('Cloud analysis:', result.content[0].text[:100])
self.pending_cloud_queries.clear()クラウド同期用のオフラインキュー
エッジエージェントは、接続が断続的な環境で動作することがよくあります。クエリ、センサーの読み取り値、アクションログをローカルにバッファリングし、接続が復旧したらクラウドにフラッシュします。ローカルバッファにはSQLiteデータベースを使用してください。
import sqlite3
import json
from datetime import datetime
DB_PATH = '/home/pi/agent_buffer.db'
def init_buffer_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS sync_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
type TEXT NOT NULL,
payload TEXT NOT NULL,
created_at TEXT NOT NULL,
synced INTEGER DEFAULT 0
)
""")
conn.commit()
conn.close()
def buffer_for_sync(record_type: str, payload: dict):
conn = sqlite3.connect(DB_PATH)
conn.execute(
'INSERT INTO sync_queue (type, payload, created_at) VALUES (?, ?, ?)',
(record_type, json.dumps(payload), datetime.utcnow().isoformat())
)
conn.commit()
conn.close()
def flush_to_cloud(cloud_fn):
conn = sqlite3.connect(DB_PATH)
rows = conn.execute(
'SELECT id, type, payload FROM sync_queue WHERE synced=0 LIMIT 100'
).fetchall()
for row_id, r_type, payload in rows:
cloud_fn(r_type, json.loads(payload))
conn.execute('UPDATE sync_queue SET synced=1 WHERE id=?', (row_id,))
conn.commit()
conn.close()
print(f'Synced {len(rows)} records to cloud')
if __name__ == '__main__':
import os, tempfile
DB_PATH = os.path.join(tempfile.gettempdir(), 'agent_buffer_demo.db')
init_buffer_db()
buffer_for_sync('sensor_reading', {'temp': 22.5})
flush_to_cloud(lambda t, p: print(f'Synced to cloud: {t} -> {p}'))
エッジエージェントの健全性監視
エッジデバイスは過熱したり、RAMが不足したり、モデル推論が停止したりする可能性があります。CPU温度、空きRAM、推論レイテンシを確認し、健全性メトリクスをクラウドに公開するヘルスモニターを実装してください。
import subprocess
import psutil # pip install psutil
def get_edge_health() -> dict:
cpu_percent = psutil.cpu_percent(interval=1)
ram = psutil.virtual_memory()
disk = psutil.disk_usage('/')
# Read CPU temperature (Raspberry Pi specific)
try:
temp_output = subprocess.check_output(
['vcgencmd', 'measure_temp'], text=True
)
cpu_temp = float(temp_output.strip().replace("temp=", "").replace("'C", ""))
except Exception:
cpu_temp = -1.0 # not available on non-Pi hardware
return {
'cpu_percent': cpu_percent,
'cpu_temp_c': cpu_temp,
'ram_used_pct': ram.percent,
'ram_available_mb': round(ram.available / 1024 / 1024),
'disk_used_pct': disk.percent,
'timestamp': datetime.utcnow().isoformat()
}
health = get_edge_health()
print('Edge health:', health)エッジエージェントの推論キャッシュ
エッジモデルは処理が遅いため、同一の入力に対する応答をキャッシュしてください。センサー分類タスクでは、同じプロンプト（例："temperature=23.5, classify"）が繰り返されることがよくあります。TTL付きの単純なdictキャッシュにより、不要な推論呼び出しを回避できます。
from datetime import datetime, timedelta
import hashlib
class InferenceCache:
def __init__(self, ttl_seconds: int = 60):
self.ttl = timedelta(seconds=ttl_seconds)
self._cache: dict = {} # hash -> {'result', 'expires'}
def _key(self, prompt: str) -> str:
return hashlib.md5(prompt.encode()).hexdigest()
def get(self, prompt: str):
key = self._key(prompt)
entry = self._cache.get(key)
if entry and datetime.utcnow() < entry['expires']:
return entry['result']
return None
def set(self, prompt: str, result: str):
key = self._key(prompt)
self._cache[key] = {
'result': result,
'expires': datetime.utcnow() + self.ttl
}
cache = InferenceCache(ttl_seconds=60)
def cached_edge_inference(prompt: str, client) -> str:
cached = cache.get(prompt)
if cached:
print('Cache hit')
return cached
result = local_inference(prompt, client)
cache.set(prompt, result)
return resultデプロイ前チェックリスト
エッジエージェントを本番環境にデプロイする前に、モデルが20%の余裕を含めて利用可能なRAMに収まること、推論レイテンシがイベント応答要件を満たすこと、オフラインバッファがテスト済みであること、ヘルスモニタリングがメトリクスを公開していること、クラッシュ時の自動再起動が設定されていることを確認してください。
# systemd service file for auto-restart (save as /etc/systemd/system/edge-agent.service)
SYSTEMD_SERVICE = '''
[Unit]
Description=IoT Edge Agent
After=network.target ollama.service
Requires=ollama.service
[Service]
User=pi
WorkingDirectory=/home/pi/edge-agent
ExecStart=/usr/bin/python3 /home/pi/edge-agent/agent.py
Restart=always
RestartSec=5
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=multi-user.target
'''
# Enable:
# sudo systemctl enable edge-agent
# sudo systemctl start edge-agent
# sudo journalctl -u edge-agent -f (follow logs)
print('Deployment checklist:')
checklist = [
'Model RAM fits with 20% headroom',
'Inference latency < response time requirement',
'Offline SQLite buffer tested',
'Health metrics publishing to cloud',
'systemd auto-restart configured'
]
for item in checklist:
print(f' [ ] {item}')Raspberry Piのハードウェア設定
ソフトウェアをデプロイする前に、ハードウェアを準備します。Phi-3-miniを実行する場合、8 GB RAM搭載のRaspberry Pi 5が推奨される最小構成です。GPUメモリの割り当てを有効にし、スワップを無効にして（フラッシュストレージの性能が低下します）、安定したリモートアクセスのために固定IPを設定してください。
# Raspberry Pi setup notes (run manually over SSH):
# sudo raspi-config -> System -> GPU Memory -> 256
# Disable swap to protect SD card:
# sudo dphys-swapfile swapoff && sudo dphys-swapfile uninstall
def check_available_ram_mb():
try:
with open('/proc/meminfo') as f:
for line in f:
if line.startswith('MemAvailable'):
return int(line.split()[1]) // 1024
except FileNotFoundError:
return None
ram_mb = check_available_ram_mb()
if ram_mb is None:
print('Could not read /proc/meminfo on this OS — simulated Pi reading: MemAvailable ~ 850 MB')
else:
print(f'Available RAM: {ram_mb} MB')理解度チェック
言語モデルにおけるQ4_K_M量子化とは何を意味しますか？
要約：軽量エージェントのエッジデプロイ
このレッスンで学習した内容：
- モデル選定：エッジ向けのPhi-3-mini、Gemma-2B、TinyLlama、Q4_K_M量子化
- Ollama：ポート11434でOpenAI互換APIを提供するローカルLLMサーバー
- ハイブリッドアーキテクチャ：高速な判断にはエッジモデル、複雑な分析にはクラウド
- オフラインバッファ：再接続時にクラウドへフラッシュするSQLiteキュー
- 推論キャッシュ：TTLキャッシュにより、同一プロンプトに対する推論の繰り返しを回避
- デプロイ：クラッシュ時に自動再起動するsystemdサービス
次のコース：エージェントマーケットプレイスとプラグインシステム
AI チューターと学ぶ AI Agents — 無料
ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。
- コース
- 60
- レッスン
- 239
よくある質問
「軽量エージェントのエッジデプロイ」レッスンは無料ですか？
はい。「軽量エージェントのエッジデプロイ」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「軽量エージェントのエッジデプロイ」で何を学びますか？
Raspberry Pi やエッジデバイス上で小規模モデルを実行し、低遅延で応答させます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「軽量エージェントのエッジデプロイ」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- エージェント連携のための MQTT プロトコル
- エージェントにおける時系列データ処理
- センサーイベントへの自動応答
- 軽量エージェントのエッジデプロイ