革新！提示工程架构师引领Agentic AI在智能设备应用革新
副标题：从提示工程到自主智能：构建下一代智能设备Agent系统的全栈指南
第一部分：引言与基础 (Introduction & Foundation)
1. 引人注目的标题 (Compelling Title)
革新！提示工程架构师引领Agentic AI在智能设备应用革新
副标题：从提示工程到自主智能：构建下一代智能设备Agent系统的全栈指南
2. 摘要/引言 (Abstract / Introduction)
问题陈述：当前智能设备AI的“被动困境”
当你唤醒智能音箱播放音乐时，它能否主动提醒“根据你的睡眠数据，今晚建议播放白噪音而非摇滚”？当智能手表检测到心率异常时，它能否自主联系家人、同步医疗记录并规划最近医院路线，而非仅震动提醒？
今天的智能设备AI，本质上是“被动响应的工具”：依赖用户显式指令、功能模块化割裂（音箱负责语音、手表负责健康、家电负责控制）、场景适应性差（换个房间可能无法联动设备）、个性化不足（千人一面的交互逻辑）。这一困境的核心在于：缺乏“自主性”与“目标导向能力”。
核心方案：Agentic AI + 提示工程，重构智能设备“主动智能”
解决之道在于引入Agentic AI（智能体AI）——一种具备“感知-规划-执行-反思”闭环能力的自主智能系统，并通过提示工程架构师的系统性设计，让Agent在智能设备的资源约束下（低算力、有限内存、实时性要求）实现高效运转。
本文提出的“智能设备Agent系统”架构，通过三层核心设计突破瓶颈：
- 提示工程架构层：动态生成适配设备状态的提示策略（如低电量时优先低功耗任务）；
- Agent能力层：集成多模态感知（传感器数据）、工具调用（设备控制接口）、记忆系统（用户偏好学习）；
- 边缘优化层：针对智能设备硬件特性优化LLM推理（量化压缩、上下文窗口裁剪）。
主要成果/价值：读者将获得什么？
读完本文后，你将掌握：
- Agentic AI在智能设备的落地范式：从需求分析到系统设计的全流程方法论；
- 提示工程架构师的核心能力：设计资源感知型提示策略、工具调用模板、反思机制；
- 全栈实现案例：基于Raspberry Pi + 边缘LLM（Llama 3-8B INT4）构建智能家居控制Agent，包含完整代码与性能优化方案；
- 避坑指南：解决智能设备Agent开发中的9大核心难题（如实时性、功耗控制、离线运行）。
文章导览
本文分为四部分：
- 基础篇：解析Agentic AI与提示工程的核心概念，以及智能设备的特殊约束；
- 设计篇：详解智能设备Agent系统架构与提示工程框架设计；
- 实现篇：从环境搭建到代码落地，手把手实现智能家居控制Agent；
- 优化与展望篇：性能调优、最佳实践、未来多Agent协同与边缘AI的融合方向。
3. 目标读者与前置知识 (Target Audience & Prerequisites)
目标读者
- AI工程师：希望将LLM与Agent能力落地到边缘设备的算法开发者；
- 嵌入式/智能设备工程师：负责智能硬件固件与应用开发，需集成AI功能的研发人员；
- 提示工程实践者：关注如何在资源受限场景下设计高效提示策略的技术人员；
- 智能设备产品经理：需理解Agentic AI技术边界与应用场景的产品设计者。
前置知识
- AI基础：了解LLM工作原理（如Transformer架构、上下文窗口）、提示工程基本概念（如零样本/少样本提示）；
- 编程能力：熟练掌握Python（核心开发语言），了解异步编程（设备事件处理）；
- 智能设备背景：了解边缘计算架构（如ARM嵌入式系统）、传感器数据采集（如I2C/SPI协议）、设备控制接口（如MQTT、蓝牙）；
- 工具链：了解Docker（环境隔离）、Git（版本控制）、模型量化工具（如llama.cpp）。
4. 文章目录 (Table of Contents)
第一部分：引言与基础
- 引人注目的标题
- 摘要/引言
- 目标读者与前置知识
- 文章目录
第二部分：核心内容
5. 问题背景与动机：智能设备AI的进化瓶颈与Agentic AI的破局
6. 核心概念与理论基础：从Agent定义到提示工程架构师的角色
7. 环境准备：开发工具链、硬件平台与依赖库配置
8. 分步实现：智能家居控制Agent的全栈开发（设计→提示→部署→集成）
9. 关键代码解析：提示工程架构、Agent决策循环与边缘LLM优化
第三部分：验证与扩展
10. 结果展示与验证：功能测试、性能指标与用户体验评估
11. 性能优化与最佳实践：算力、功耗、实时性三维度调优
12. 常见问题与解决方案：9大核心难题的避坑指南
13. 未来展望：多Agent协同、具身智能与边缘AI芯片的融合
第四部分：总结与附录
14. 总结：提示工程架构师的能力模型与技术路线图
15. 参考资料：论文、开源项目与工具链文档
16. 附录：完整代码仓库、硬件兼容性列表与提示模板库
第二部分：核心内容 (Core Content)
5. 问题背景与动机 (Problem Background & Motivation)
5.1 智能设备AI的三代进化与当前瓶颈
智能设备AI的发展可分为三代：
-
第一代（2015-2018）：规则驱动
代表：早期智能音箱（如Amazon Echo初代），通过关键词匹配执行固定指令（“播放音乐”→调用Spotify API）。
局限：无泛化能力，新增功能需重写规则。 -
第二代（2019-2022）：单一模型驱动
代表：搭载专用AI芯片的设备（如Apple Watch Series 6的心率监测AI），通过CNN/RNN处理特定任务。
局限：功能孤立（健康模型无法联动通知模型），依赖云端推理（断网即失效）。 -
第三代（2023-今）：LLM驱动
代表：支持自然对话的设备（如Google Pixel 8的AI助手），通过云端LLM理解复杂指令（“明天出差提醒我带充电器和护照”）。
局限：被动响应（需用户主动唤醒）、资源依赖（云端延迟500ms+）、无长期规划（无法执行“下班前预热烤箱”等跨时段任务）。
5.2 Agentic AI：突破第三代瓶颈的核心能力
Agentic AI（智能体AI）的定义可追溯至1995年MIT的《Intelligent Agents》论文：“能感知环境并自主执行动作以实现目标的实体”。与传统LLM应用相比，Agentic AI在智能设备中具备三大关键优势：
| 特性 | 传统LLM应用 | Agentic AI |
|---|---|---|
| 交互模式 | 被动响应（用户提问→回答） | 主动感知（环境变化→触发行动） |
 任务能力 | 单轮对话/单任务 | 多任务规划与执行（目标分解） |
| 环境交互 | 无工具调用能力 | 集成传感器、执行器、API工具 |
| 学习与记忆 | 对话内短期记忆 | 长期用户偏好学习与更新 |
例如，一个智能家居Agentic系统可实现：
- 感知：通过温湿度传感器发现室内湿度>70%；
- 规划：目标是“维持舒适环境”，分解为“关闭加湿器→打开除湿机→通知用户”；
- 执行：调用MQTT接口控制设备；
- 反思：1小时后检查湿度是否达标，若未达标则调整除湿机模式。
5.3 提示工程架构师：Agentic AI落地的“翻译官”与“指挥官”
将Agentic AI部署到智能设备，最大挑战是资源约束与智能需求的矛盾：设备算力（如Raspberry Pi 5的4核Cortex-A76）仅为云端GPU的1%，却需实现实时响应（<300ms）、低功耗（电池供电设备续航>24h）、离线运行（无网络时核心功能可用）。
此时，提示工程架构师的角色至关重要，其核心职责是：
[{"tool": "mqtt", "action": "set", "device": "dehumidifier", "params": {"mode": "high"}}]没有提示工程架构师的系统性设计，Agent要么因提示过长耗尽设备内存，要么因无法理解设备状态而执行错误动作（如在用户睡觉时打开强光）。
6. 核心概念与理论基础 (Core Concepts & Theoretical Foundation)
6.1 Agentic AI的核心架构：从“感知”到“反思”的闭环
智能设备Agent系统需包含五大核心模块（图1为架构图示意）：
图1：智能设备Agent系统架构图
┌─────────────────────────────────────────────────────────┐
│ 设备环境层 │
│ (传感器：温湿度/光照/心率 | 执行器：灯光/电机 | 网络) │
└───────────────────────┬─────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ 感知处理层 │
│ (数据清洗→特征提取→事件检测：如“异常心率”“高湿度”) │
└───────────────────────┬─────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ Agent核心层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 规划模块 │ │ 执行模块 │ │ 记忆模块 │ │ 反思模块 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└───────────────────────┬─────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ 提示工程架构层 (核心) │
│ (系统提示生成→工具调用提示→反思提示→上下文管理) │
└───────────────────────┬─────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ 边缘LLM层 │
│ (量化模型：如Llama 3-8B INT4 | 推理引擎：llama.cpp) │
└─────────────────────────────────────────────────────────┘
各模块功能详解：
- 感知处理层：将原始传感器数据转化为Agent可理解的“事件”（如加速度传感器数据→“用户摔倒”事件）；
- 规划模块：基于目标（如“节能”）和当前状态，分解任务（如“关闭闲置灯光→降低空调温度”）；
- 执行模块：调用工具（设备接口）执行子任务，并处理工具返回结果；
- 记忆模块：分为短期记忆（当前对话/任务）和长期记忆（用户偏好，如“用户讨厌冷色调灯光”）；
- 反思模块：评估任务执行效果（如“空调温度调整后用户是否感到舒适”），并优化后续策略。
6.2 提示工程架构设计：三大核心提示模板
提示工程架构师需设计三类关键提示模板，确保Agent在设备约束下高效运行：
6.2.1 系统提示模板（System Prompt Template）
系统提示定义Agent的“身份”“能力”“约束”，需动态注入设备状态变量。例如：
SYSTEM_PROMPT_TEMPLATE = """
你是智能家居控制Agent，运行在{device_model}上，当前状态：电量{ battery_level }%，网络{ network_status }。
=== 能力 ===
- 可调用工具：{available_tools}（格式：<|FunctionCallBegin|>[{"tool": "...", "params": {...}}]<|FunctionCallEnd|>）
- 记忆访问：可查询用户偏好（如“用户通常10点关灯”）
=== 约束 ===
1. 若电量<20%：禁用云端API调用，优先使用本地缓存数据；
2. 响应时间需<300ms，复杂任务可分步骤执行；
3. 避免控制用户明确禁用的设备（如“儿童房灯光”）。
当前目标：{user_goal}
"""
变量说明：
device_modelbattery_level/sys/class/power_supply/BAT0/capacitynetwork_statusping 8.8.8.8available_tools6.2.2 工具调用提示模板（Tool Call Template）
为避免LLM生成非结构化工具调用指令（增加解析难度），需设计严格的JSON格式模板，并在提示中明确示例：
TOOL_CALL_EXAMPLE = """
<|FunctionCallBegin|>
[
{{"tool": "mqtt", "action": "set", "device": "light_livingroom", "params": {{"brightness": 70, "color": "white"}}}},
{{"tool": "notification", "action": "send", "params": {{"text": "客厅灯光已调亮至70%"}}}}
]
<|FunctionCallEnd|>
"""
"retry": true"reason": "设备离线"6.2.3 反思提示模板（Reflection Prompt Template）
反思模块通过提示让Agent评估任务执行效果，例如：
REFLECTION_PROMPT_TEMPLATE = """
任务：{task_description}
执行结果：{execution_result}（成功/失败，附设备返回数据）
用户反馈：{user_feedback}（若有）
请回答：
1. 任务是否达成目标？（是/否，原因）
2. 若未达成，下一步应如何调整？（需调用工具请生成工具调用指令）
3. 是否需要更新用户偏好记忆？（如“用户不喜欢70%亮度”）
"""
6.3 边缘LLM选型与部署：平衡算力与智能
智能设备Agent的“大脑”是边缘LLM，选型需考虑三大因素：模型大小（影响内存占用）、推理速度（影响响应延迟）、量化级别（影响精度与算力需求）。
6.3.1 主流边缘LLM对比（2024年Q2数据）
| 模型 | 参数量 | 量化方式 | 推理速度（RPi 5） | 内存占用 | 对话质量（MT-Bench） |
|---|---|---|---|---|---|
| Llama 3-8B | 8B | INT4 | ~20 tokens/秒 | 4.5GB | 7.6/10 |
| Mistral-7B | 7B | INT4 | ~25 tokens/秒 | 4GB | 7.3/10 |
| Phi-2-2.7B | 2.7B | INT4 | ~50 tokens/秒 | 1.8GB | 6.8/10 |
| TinyLlama-1.1B | 1.1B | INT4 | ~100 tokens/秒 | 0.8GB | 5.5/10 |
选型建议：
- 高性能设备（如RPi 5、NVIDIA Jetson Nano）：优先Llama 3-8B INT4（平衡质量与速度）；
- 低功耗设备（如ESP32-S3）：选择TinyLlama-1.1B（需简化Agent能力，仅保留核心任务）；
- 折中方案：Mistral-7B（比Llama 3快，质量略低）。
6.3.2 量化与推理优化工具链
libllama.so7. 环境准备 (Environment Setup)
7.1 硬件平台选择
本文以Raspberry Pi 5（4GB内存版） 为开发板，其配置满足边缘LLM运行需求：
- CPU：4核Cortex-A76（2.4GHz）；
- 内存：4GB LPDDR4X；
- 存储：32GB microSD卡（建议UHS-I U3级别）；
- 扩展：GPIO接口（连接传感器）、USB 3.0（连接外置SSD提升模型加载速度）。
备选平台：
- 低成本：Raspberry Pi 4（2GB+，Llama 3-8B INT4可运行但速度较慢）；
- 高性能：NVIDIA Jetson Orin NX（支持更大模型如Llama 3-70B量化版）。
7.2 软件与依赖库配置
7.2.1 操作系统安装
- 下载Raspberry Pi OS Bookworm（64-bit，带桌面环境）；
- 使用Raspberry Pi Imager写入microSD卡，启用SSH（系统设置→接口选项）。
7.2.2 核心依赖库
创建Python虚拟环境并安装依赖：
# 更新系统
sudo apt update && sudo apt upgrade -y
# 安装系统依赖
sudo apt install -y git build-essential cmake python3-venv libopenblas-dev
# 创建虚拟环境
python3 -m venv agent-env
source agent-env/bin/activate
# 安装Python依赖
pip install --upgrade pip
pip install -r requirements.txt
requirements.txt# Agent框架
langchain==0.2.5
langchain-community==0.2.5
langgraph==0.0.61 # LangChain的Agent工作流库
# LLM推理
llama-cpp-python==0.2.75 # 调用llama.cpp
ctransformers==0.2.27 # 备用推理库
# 设备通信
paho-mqtt==2.1.0 # MQTT客户端（控制智能设备）
pyserial==3.5 # 串口通信（连接传感器）
# 数据处理
numpy==1.26.4
pandas==2.2.2
scikit-learn==1.4.2 # 简单用户偏好学习
# 系统监控
psutil==5.9.8 # 获取设备状态（电量、CPU）
netifaces==0.11.0 # 网络状态检测
7.2.3 边缘LLM模型下载与量化
以Llama 3-8B为例，下载原始模型并量化为INT4：
# 安装llama.cpp
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp && make
# 下载Llama 3-8B模型（需Hugging Face账号与访问权限）
huggingface-cli download meta-llama/Llama-3-8B-Instruct --include "original/*" --local-dir ./models/llama3-8b-original
# 转换为llama.cpp格式
python convert.py models/llama3-8b-original/original/ --outfile models/llama3-8b-f16.gguf
# 量化为INT4（4-bit）
./quantize models/llama3-8b-f16.gguf models/llama3-8b-int4.gguf q4_0
# 测试模型（生成“Hello World”）
./main -m models/llama3-8b-int4.gguf -p "Hello World" -n 10
量化后模型大小约4.5GB，加载时间（冷启动）约20秒，后续推理无需重新加载。
7.2.4 智能设备模拟环境
为避免实际硬件连接，可使用Docker容器模拟智能设备（如加湿器、除湿机）：
# 拉取MQTT broker（设备通信中枢）
docker run -d --name mqtt -p 1883:1883 eclipse-mosquitto
# 拉取设备模拟器（模拟温湿度传感器、执行器）
git clone https://github.com/smart-home-simulator/mqtt-devices
cd mqtt-devices && docker-compose up -d
模拟器提供Web界面（http://:8080），可手动调整传感器数据或观察设备状态变化。
8. 分步实现 (Step-by-Step Implementation)
8.1 步骤1：需求分析与Agent能力定义
目标：构建一个能维持“舒适家居环境”的Agent，具体需求：
- 感知：监测温度（18-26℃）、湿度（40-60%）、光照（白天/夜晚）；
- 执行：控制空调（温度调节）、加湿器/除湿机（湿度调节）、灯光（亮度/色温）；
- 主动行为：环境异常时（如湿度>60%）自动触发调节，无需用户指令；
- 个性化：学习用户偏好（如“用户喜欢晚上22点后灯光色温<3000K”）；
- 资源约束：响应时间<500ms，内存占用<6GB，离线时保留核心功能。
Agent能力矩阵：
| 能力模块 | 输入（感知） | 输出（执行） | 触发条件 |
|---|---|---|---|
| 温度调节 | 温度传感器数据 | 空调温度设置（MQTT指令） | 温度<18℃或>26℃ |
| 湿度调节 | 湿度传感器数据 | 加湿器/除湿机开关/模式 | 湿度<40%或>60% |
| 光照调节 | 光照传感器+时间 | 灯光亮度/色温 | 光照<300lux（白天）或>22点 |
| 用户提醒 | 环境异常持续10分钟未解决 | 推送通知（本地+手机APP） | 调节失败或需人工干预 |
8.2 步骤2：系统架构设计与模块划分
基于6.1节的核心架构，细化为可编码的模块：
smart-home-agent/
├── agent/ # Agent核心模块
│ ├── planner.py # 任务规划模块
│ ├── executor.py # 工具执行模块
│ ├── memory.py # 记忆系统（短期+长期）
│ └── reflector.py # 反思与优化模块
├── prompt/ # 提示工程架构
│ ├── system_prompt.py # 系统提示生成器
│ ├── tool_templates.py # 工具调用模板
│ └── context_manager.py # 上下文窗口管理
├── perception/ # 感知处理层
│ ├── sensors/ # 传感器驱动（温度、湿度等）
│ └── event_detector.py # 事件检测（如“高湿度事件”）
├── llm/ # 边缘LLM层
│ ├── model_loader.py # 模型加载与推理
│ └── config.py # LLM配置（量化方式、温度等）
├── tools/ # 工具接口
│ ├── mqtt_client.py # MQTT设备控制
│ └── notification.py # 通知发送
├── utils/ # 工具函数
│ ├── device_status.py # 设备状态获取（电量、网络）
│ └── logger.py # 日志系统
├── main.py # 主程序入口
└── requirements.txt # 依赖库
8.3 步骤3：感知层实现——传感器数据采集与事件检测
8.3.1 传感器连接与驱动
使用DHT22温湿度传感器（通过GPIO连接Raspberry Pi），代码如下：
# perception/sensors/dht22.py
import Adafruit_DHT # DHT传感器库
import time
class DHT22Sensor:
def __init__(self, pin=4): # GPIO4（Pin 7）
self.sensor = Adafruit_DHT.DHT22
self.pin = pin
def read(self):
"""读取温度（℃）和湿度（%），失败返回None"""
humidity, temperature = Adafruit_DHT.read_retry(self.sensor, self.pin)
if humidity is not None and temperature is not None:
return {"temperature": round(temperature, 1), "humidity": round(humidity, 1)}
else:
return None # 读取失败（传感器常见问题，需重试）
# 测试
if __name__ == "__main__":
sensor = DHT22Sensor()
while True:
data = sensor.read()
if data:
print(f"温度：{data['temperature']}℃，湿度：{data['humidity']}%")
time.sleep(2) # 每2秒读取一次
8.3.2 事件检测
将原始传感器数据转换为Agent可理解的“事件”：
# perception/event_detector.py
class EnvironmentEventDetector:
def __init__(self):
# 舒适区间配置（可通过记忆系统动态更新）
self.comfort_ranges = {
"temperature": (18.0, 26.0),
"humidity": (40.0, 60.0)
}
def detect(self, sensor_data):
"""输入传感器数据，输出事件列表"""
events = []
# 温度事件
temp = sensor_data.get("temperature")
if temp is not None:
if temp < self.comfort_ranges["temperature"][0]:
events.append({"type": "temperature_low", "value": temp, "message": f"温度过低：{temp}℃"})
elif temp > self.comfort_ranges["temperature"][1]:
events.append({"type": "temperature_high", "value": temp, "message": f"温度过高：{temp}℃"})
# 湿度事件
humi = sensor_data.get("humidity")
if humi is not None:
if humi < self.comfort_ranges["humidity"][0]:
events.append({"type": "humidity_low", "value": humi, "message": f"湿度过低：{humi}%"})
elif humi > self.comfort_ranges["humidity"][1]:
events.append({"type": "humidity_high", "value": humi, "message": f"湿度过高：{humi}%"})
return events
# 测试
if __name__ == "__main__":
detector = EnvironmentEventDetector()
sensor_data = {"temperature": 28.5, "humidity": 65.0}
print(detector.detect(sensor_data))
# 输出：[{"type": "temperature_high",...}, {"type": "humidity_high",...}]
8.4 步骤4：提示工程架构实现——动态提示生成
8.4.1 设备状态获取
实现获取设备状态（电量、网络）的工具函数：
# utils/device_status.py
import psutil
import netifaces
import time
def get_battery_level():
"""获取电量（仅适用于电池供电设备，如Raspberry Pi Zero W+电池扩展板）"""
try:
with open("/sys/class/power_supply/BAT0/capacity", "r") as f:
return int(f.read().strip())
except FileNotFoundError:
return 100 # 非电池设备默认100%
def get_network_status():
"""检查网络连接状态"""
try:
# 检查是否能ping通网关
gateway = netifaces.gateways()['default'][netifaces.AF_INET][0]
response = psutil.subprocess.run(
["ping", "-c", "1", "-W", "1", gateway],
stdout=psutil.subprocess.PIPE,
stderr=psutil.subprocess.PIPE
)
return "online" if response.returncode == 0 else "offline"
except KeyError:
return "offline"
# 测试
if __name__ == "__main__":
print(f"电量：{get_battery_level()}%")
print(f"网络：{get_network_status()}")
8.4.2 系统提示生成器
根据设备状态动态生成系统提示：
# prompt/system_prompt.py
from utils.device_status import get_battery_level, get_network_status
class SystemPromptGenerator:
def __init__(self, device_model="Raspberry Pi 5", available_tools=None):
self.device_model = device_model
self.available_tools = available_tools or ["mqtt", "notification", "memory"]
def generate(self, user_goal="维持舒适家居环境"):
# 获取设备状态
battery_level = get_battery_level()
network_status = get_network_status()
# 动态调整可用工具（如网络离线则移除依赖网络的工具）
current_tools = self.available_tools.copy()
if network_status == "offline" and "notification" in current_tools:
current_tools.remove("notification") # 假设通知需联网
# 生成系统提示
system_prompt = f"""
你是智能家居控制Agent，运行在{self.device_model}上，当前状态：电量{battery_level}%，网络{network_status}。
=== 能力 ===
- 可调用工具：{current_tools}（格式：<|FunctionCallBegin|>[{"tool": "...", "params": {...}}]<|FunctionCallEnd|>）
- 记忆访问：可查询用户偏好（如“用户通常10点关灯”）
=== 约束 ===
1. 若电量<20%：禁用云端API调用，优先使用本地缓存数据；
2. 响应时间需<500ms，复杂任务可分步骤执行；
3. 避免控制用户明确禁用的设备（如“儿童房灯光”）。
当前目标：{user_goal}
""".strip()
return system_prompt
# 测试
if __name__ == "__main__":
generator = SystemPromptGenerator()
print(generator.generate())
# 输出包含动态电量、网络状态、可用工具的系统提示
8.4.3 上下文窗口管理
边缘LLM的上下文窗口有限（如Llama 3-8B INT4约2k token），需管理对话历史、设备状态、任务列表的优先级：
# prompt/context_manager.py
import tiktoken # 计算token数
class ContextManager:
def __init__(self, max_tokens=2048, model_name="llama3"):
self.max_tokens = max_tokens
self.encoding = tiktoken.encoding_for_model(model_name)
self.context = {
"system_prompt": "",
"chat_history": [], # 列表：[{"role": "user", "content": "..."}]
"device_states": {}, # 设备状态：{"temperature": 25, "humidity": 50}
"current_tasks": [] # 当前任务列表
}
def add_chat_message(self, role, content):
"""添加对话历史，超出token限制时截断最早消息"""
self.context["chat_history"].append({"role": role, "content": content})
self._truncate_chat_history()
def update_device_states(self, states):
"""更新设备状态（如传感器数据）"""
self.context["device_states"].update(states)
def update_current_tasks(self, tasks):
"""更新当前任务列表"""
self.context["current_tasks"] = tasks
def get_prompt(self):
"""生成完整的LLM输入prompt"""
# 系统提示
prompt = self.context["system_prompt"] + "\n\n"
# 设备状态
prompt += "=== 当前设备状态 ===\n"
prompt += "\n".join([f"{k}: {v}" for k, v in self.context["device_states"].items()]) + "\n\n"
# 当前任务
prompt += "=== 当前任务 ===\n"
prompt += "\n".join([f"- {t}" for t in self.context["current_tasks"]]) + "\n\n"
# 对话历史
prompt += "=== 对话历史 ===\n"
for msg in self.context["chat_history"]:
prompt += f"{msg['role']}: {msg['content']}\n"
return prompt
def _truncate_chat_history(self):
"""截断对话历史以满足max_tokens限制"""
total_tokens = self._count_tokens(self.get_prompt())
while total_tokens > self.max_tokens and len(self.context["chat_history"]) > 0:
# 移除最早的用户/助手对话对
if len(self.context["chat_history"]) >= 2:
self.context["chat_history"].pop(0) # 移除用户消息
self.context["chat_history"].pop(0) # 移除助手消息
else:
self.context["chat_history"].pop(0) # 只剩一条时移除
total_tokens = self._count_tokens(self.get_prompt())
def _count_tokens(self, text):
"""计算文本的token数"""
return len(self.encoding.encode(text))
# 测试
if __name__ == "__main__":
manager = ContextManager(max_tokens=500)
manager.context["system_prompt"] = "你是测试Agent"
manager.add_chat_message("user", "你好！")
manager.add_chat_message("assistant", "你好！有什么可以帮你？")
manager.update_device_states({"temperature": 25, "humidity": 50})
print(manager.get_prompt())
print(f"Token数：{manager._count_tokens(manager.get_prompt())}")
8.5 步骤5：LLM推理层实现——边缘模型加载与调用
8.5.1 模型加载与推理封装
使用llama-cpp-python加载量化模型并实现推理接口：
# llm/model_loader.py
from llama_cpp import Llama
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class EdgeLLM:
def __init__(self, model_path, n_ctx=2048, n_threads=4, n_gpu_layers=0):
"""
初始化边缘LLM
:param model_path: 量化模型路径（如./models/llama3-8b-int4.gguf）
:param n_ctx: 上下文窗口大小（token数）
:param n_threads: 推理线程数（建议设为CPU核心数）
:param n_gpu_layers: GPU加速层数（0表示纯CPU推理）
"""
self.model = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_threads=n_threads,
n_gpu_layers=n_gpu_layers,
verbose=False # 禁用详细日志
)
logger.info(f"模型加载成功：{model_path}（上下文窗口：{n_ctx} token）")
def generate(self, prompt, max_tokens=200, temperature=0.7, stop=["<|FunctionCallEnd|>"]):
"""
生成文本
:param prompt: LLM输入提示
:param max_tokens: 最大生成token数
:param temperature: 随机性（0=确定性，1=随机性高）
:param stop: 停止词列表
:return: 生成的文本
"""
try:
output = self.model(
prompt=prompt,
max_tokens=max_tokens,
temperature=temperature,
stop=stop,
echo=False # 不回显输入prompt
)
return output["choices"][0]["text"].strip()
except Exception as e:
logger.error(f"LLM推理失败：{e}")
return ""
# 测试
if __name__ == "__main__":
llm = EdgeLLM(
model_path="./models/llama3-8b-int4.gguf",
n_ctx=2048,
n_threads=4
)
prompt = "介绍一下你自己，用10个字以内。"
print(llm.generate(prompt, max_tokens=10)) # 输出："我是智能家居Agent。"
8.6 步骤6：Agent核心模块实现——规划、执行、记忆、反思
8.6.1 任务规划模块（Planner）
规划模块将用户目标或环境事件分解为可执行的子任务，输入为“目标+当前状态”，输出为任务列表：
# agent/planner.py
from prompt.system_prompt import SystemPromptGenerator
from prompt.context_manager import ContextManager
from llm.model_loader import EdgeLLM
class Planner:
def __init__(self, llm: EdgeLLM):
self.llm = llm
self.system_prompt_generator = SystemPromptGenerator()
self.context_manager = ContextManager(max_tokens=1500) # 预留500 token给生成
def plan(self, goal, device_states, current_tasks=None):
"""
生成任务计划
:param goal: 目标（如“维持舒适家居环境”或事件“湿度过高”）
:param device_states: 当前设备状态（传感器数据）
:param current_tasks: 当前正在执行的任务（可选）
:return: 任务列表（如[{"task": "降低湿度", "steps": [...]}, ...]）
"""
# 1. 生成系统提示
system_prompt = self.system_prompt_generator.generate(user_goal=goal)
self.context_manager.context["system_prompt"] = system_prompt
# 2. 更新上下文（设备状态、当前任务）
self.context_manager.update_device_states(device_states)
self.context_manager.update_current_tasks(current_tasks or [])
# 3. 生成用户提示（要求LLM输出任务列表）
user_prompt = """
基于当前设备状态和目标，请生成详细的任务计划。
输出格式：
[
{"task_id": 1, "description": "任务描述", "steps": ["步骤1", "步骤2"]}
]
确保步骤可通过调用工具完成。
""".strip()
self.context_manager.add_chat_message("user", user_prompt)
# 4. 调用LLM生成任务计划
full_prompt = self.context_manager.get_prompt()
response = self.llm.generate(
prompt=full_prompt,
max_tokens=300,
temperature=0.5, # 降低随机性，确保任务可行性
stop=["</task_plan>"]
)
# 5. 解析响应（简化处理，实际需加入JSON解析与错误处理）
import json
try:
task_plan = json.loads(response)
return task_plan
except json.JSONDecodeError:
logger.error(f"任务计划解析失败：{response}")
return []
# 测试
if __name__ == "__main__":
llm = EdgeLLM(model_path="./models/llama3-8b-int4.gguf")
planner = Planner(llm)
device_states = {"temperature": 25.0, "humidity": 65.0} # 湿度偏高
goal = "湿度过高，需要降低湿度"
tasks = planner.plan(goal, device_states)
print(tasks)
# 预期输出：
# [{"task_id": 1, "description": "降低室内湿度", "steps": ["调用除湿机设置为高模式", "1小时后检查湿度"]}]
8.6.2 工具执行模块（Executor）
执行模块将任务步骤转换为工具调用指令，并调用实际设备接口：
# agent/executor.py
import json
import logging
from tools.mqtt_client import MQTTClient # 假设已实现MQTT客户端
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ToolExecutor:
def __init__(self):
# 初始化工具客户端
self.mqtt_client = MQTTClient(broker="localhost", port=1883)
self.mqtt_client.connect()
def execute(self, tool_calls):
"""
执行工具调用
:param tool_calls: 工具调用列表（LLM生成的JSON）
:return: 执行结果列表
"""
results = []
for call in tool_calls:
tool = call.get("tool")
params = call.get("params", {})
try:
if tool == "mqtt":
# 执行MQTT设备控制
device = params.get("device")
action = params.get("action")
device_params = params.get("params", {})
result = self.mqtt_client.control_device(device, action, device_params)
results.append({
"tool": tool,
"status": "success",
"result": result
})
elif tool == "notification":
# 发送通知（简化示例）
text = params.get("text", "")
# 实际实现：调用手机APP推送API
results.append({
"tool": tool,
"status": "success",
"result": f"通知发送成功：{text}"
})
else:
results.append({
"tool": tool,
"status": "error",
"result": "不支持的工具"
})
except Exception as e
