Развёртывание лёгких агентов на периферии
Запуск небольших моделей на Raspberry Pi и периферийных устройствах для быстрого ответа.
«Развёртывание лёгких агентов на периферии» — бесплатный урок AI Agents на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.
Периферийные ИИ-агенты
Периферийный агент работает непосредственно на устройстве IoT или локальном шлюзе — Raspberry Pi, Jetson Nano или промышленном PC — вместо облака. Преимущества: меньшая задержка (не требуется обмен данными туда и обратно), работа без подключения к сети и меньшие затраты на пропускную способность. Компромисс: ограниченные вычислительные ресурсы и RAM требуют меньших и более эффективных моделей.
Выбор модели для периферийного развёртывания
Периферийные устройства не могут запускать GPT-4o или Claude Opus. Небольшие модели, которые помещаются на Raspberry Pi 5 с 8 GB RAM: Phi-3-mini (3,8 млрд параметров), Gemma-2B, TinyLlama-1.1B. После квантования до 4 бит этим моделям требуется 1–3 GB RAM, а на 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"]}')Установка Ollama на Raspberry Pi
Ollama — самый простой способ запускать небольшие LLM на локальном оборудовании. Она управляет загрузкой и квантованием моделей, а также локальным REST API, совместимым с SDK OpenAI. Одна команда устанавливает Ollama, а другая загружает и запускает модель.
# 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) — это формат файлов, который используют llama.cpp и Ollama для квантованных моделей. Q4_K_M означает смешанное 4-битное квантование — примерно на 75% меньше, чем fp32, при минимальной потере качества в задачах рассуждения на периферийных устройствах.
# 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 позволяет избежать повторных вызовов вывода.
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Контрольный список перед развертыванием
Перед развертыванием периферийного агента в рабочей среде проверьте следующее: модель помещается в доступную RAM с запасом 20%, задержка вывода соответствует требованию к времени отклика на событие, автономный буфер протестирован, мониторинг состояния публикует данные, а автоматический перезапуск после сбоя настроен.
# 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
Перед развертыванием программного обеспечения подготовьте оборудование. Raspberry Pi 5 с 8 GB RAM — это минимально рекомендуемый вариант для запуска Phi-3-mini. Включите распределение памяти 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: локальный сервер LLM с API, совместимым с OpenAI, на порту 11434
- Гибридная архитектура: периферийная модель для быстрых решений, облако для сложного анализа
- Автономный буфер: очередь SQLite, отправляемая в облако после восстановления подключения
- Кэш вывода: кэш с TTL предотвращает повторный вывод для одинаковых запросов
- Развертывание: служба systemd для автоматического перезапуска после сбоя
Следующий курс: маркетплейс агентов и системы плагинов.
Изучай AI Agents с ИИ-репетитором — бесплатно
Пиши и запускай код прямо в браузере, получай мгновенную помощь от ИИ-репетитора 24/7 и продолжи учиться на сайте или в приложении.
- Курсы
- 60
- Уроки
- 239
Часто задаваемые вопросы
Урок «Развёртывание лёгких агентов на периферии» бесплатный?
Да — полный текст урока «Развёртывание лёгких агентов на периферии» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.
Чему я научусь в уроке «Развёртывание лёгких агентов на периферии»?
Запуск небольших моделей на Raspberry Pi и периферийных устройствах для быстрого ответа. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать AI Agents?
Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.
Сколько времени занимает урок «Развёртывание лёгких агентов на периферии»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке AI Agents?
Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Протокол MQTT для интеграции агентов
- Обработка временных рядов агентами
- Автоматическая реакция на события датчиков
- Развёртывание лёгких агентов на периферии