0Pricing
AI Agents · 课时

轻量级智能体的边缘部署

在 Raspberry Pi 和边缘设备上运行小型模型，实现低延迟响应

轻量级智能体的边缘部署 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

边缘 AI Agent

边缘智能体直接运行在物联网设备或本地网关上，例如 Raspberry Pi、Jetson Nano 或工业 PC，而不是运行在云端。优点包括：延迟更低（无需往返通信）、可离线工作以及带宽成本更低。代价是：计算能力和 RAM 有限，因此需要更小、更高效的模型。

选择用于边缘部署的模型

边缘设备无法运行 GPT-4o 或 Claude Opus。可以在 Raspberry Pi 5（8 GB RAM）上运行的小型模型包括：Phi-3-mini（38 亿参数）、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"]}')

在 Raspberry Pi 上安装 Ollama

Ollama 是在本地硬件上运行小型 LLM 的最简单方式。它负责管理模型下载、量化以及与 OpenAI SDK 兼容的本地 REST API。一个命令即可安装，另一个命令即可拉取并启动模型。

# 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 位混合量化——相比 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 的简单字典缓存可以避免重复调用推理。

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 硬件设置

部署软件之前，请先准备硬件。运行 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 兼容接口的本地 LLM 服务器
  • 混合架构：边缘模型用于快速决策，云端用于复杂分析
  • 离线缓冲区：重新连接时将 SQLite 队列发送到云端
  • 推理缓存：TTL 缓存避免对相同提示重复推理
  • 部署：使用 systemd 服务在崩溃后自动重启

下一课程：代理市场与插件系统。

常见问题解答

「轻量级智能体的边缘部署」课时是免费的吗？

是的 — 「轻量级智能体的边缘部署」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「轻量级智能体的边缘部署」这节课中我会学到什么？

在 Raspberry Pi 和边缘设备上运行小型模型，实现低延迟响应 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「轻量级智能体的边缘部署」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 用于智能体集成的 MQTT 协议
  2. 智能体中的时间序列数据处理
  3. 对传感器事件的自动响应
  4. 轻量级智能体的边缘部署
← 返回 AI Agents