浪链遇见家庭助理：释放智能家居中生成式人工智能的力量

本文描述了一个架构和设计家庭助理(HA) 集成称为本地生成代理。该项目使用浪链和郎图创建一个生成式人工智能代理与 HA 智能家居环境中的任务进行交互并自动执行任务。代理了解您家的环境，了解您的偏好，并与您和您的家互动，以完成您认为有价值的活动。主要功能包括创建自动化、分析图像以及使用各种 LLM（大型语言模型）管理家庭状态。该架构涉及基于云和基于边缘的模型，以实现最佳性能和成本效益。包括安装说明、配置详细信息以及有关项目架构和所使用的不同模型的信息，可以在家庭生成代理GitHub。该项目是开源的，欢迎贡献。

以下是当前支持的一些功能：

• 创建复杂的家庭助理自动化。
• 图像场景分析和理解。
• 实体、设备和区域的主状态分析。
• 代理对家中允许的实体进行完全控制。
• 使用语义搜索的短期和长期记忆。
• 自动汇总主状态以管理 LLM 上下文长度。

这是我的个人项目，也是我所说的一个例子学习导向的黑客攻击。该项目与我在 Amazon 的工作无关，也与负责 Home Assistant 或 LangChain/LangGraph 的组织没有任何关系。

创建代理来监视和控制您的家庭可能会导致意外操作，并可能因 LLM 幻觉和隐私问题而使您的家庭和您自己面临风险，特别是在将家庭状态和用户信息暴露给基于云的 LLM 时。我已经做出了合理的架构和设计选择来减轻这些风险，但它们无法完全消除。

一个关键的早期决定是依赖混合云边缘方法。这使得能够使用最复杂的推理和规划模型，这应该有助于减少幻觉。采用更简单、更注重任务的边缘模型来进一步减少 LLM 错误。

另一个关键决策是利用 LangChain 的功能，该功能可以对 LLM 工具隐藏敏感信息，并且仅在运行时提供。例如，工具逻辑可能需要使用发出请求的用户的 ID。然而，这些值通常不应由法学硕士控制。允许法学硕士操纵用户 ID 可能会带来安全和隐私风险。为了缓解这个问题，我利用了注入工具参数注解。

此外，使用基于云的大型法学硕士会产生巨大的云成本，并且运行法学硕士边缘模型所需的边缘硬件可能非常昂贵。目前，综合运营和安装成本对于普通用户来说可能令人望而却步。需要全行业共同努力“让法学硕士与 CNN 一样便宜”，才能将国内代理人推向大众市场。

重要的是要意识到这些风险并了解，尽管采取了这些缓解措施，但我们总体上仍处于该项目和家庭代理的早期阶段。要使这些代理成为真正有用且值得信赖的助手，还有大量工作要做。

下面是家庭生成代理架构的高级视图。

一般集成架构遵循最佳实践，如中所述家庭助理核心并符合家庭助理社区商店(HACS) 发布要求。

该代理使用 LangGraph 构建并使用 HA对话与用户交互的组件。该代理使用 Home Assistant LLM API 来获取家庭状态并了解其可用的 HA 本机工具。我使用 LangChain 实现了代理可用的所有其他工具。该代理采用了多个 LLM、用于高级推理的大型且非常准确的主要模型、用于相机图像分析的较小的专用辅助模型、主要模型上下文摘要以及用于长期语义搜索的嵌入生成。主要模型是基于云的，辅助模型是基于边缘的并在奥拉玛位于家中的计算机上的框架。

目前使用的模型总结如下。

• GPT-4o|开放人工智能云 |高级推理和规划
• 骆驼-3.2-vision-11b|奥拉马边缘 |图像场景分析
• 骆驼-3.2-vision-11b|奥拉马边缘 |主要模型上下文摘要
• mxbai-嵌入-大|奥拉马边缘 |用于语义搜索的嵌入生成

基于 LangGraph 的代理

LangGraph 为对话代理提供支持，使您能够利用 LLM 尽快创建有状态的多角色应用程序。它扩展了 LangChain 的功能，引入了创建和管理循环图的能力，这对于开发复杂的代理运行时至关重要。图表对代理工作流程进行了建模，如下图所示。

代理工作流程有五个节点，每个 Python 模块都会修改代理的状态（共享数据结构）。节点之间的边表示它们之间允许的转换，实线是无条件的，虚线是有条件的。节点完成工作，边缘告诉下一步做什么。这

__开始__和__结尾__节点通知图表从哪里开始和停止。这代理人节点运行主 LLM，如果它决定使用工具，则行动节点运行该工具，然后将控制权返回给代理人。这总结和修剪节点处理 LLM 的上下文以管理增长，同时保持准确性，如果代理人无调用工具且消息数量满足下述条件。

法学硕士情境管理

您需要仔细管理 LLM 的上下文长度，以平衡成本、准确性和延迟，并避免触发速率限制，例如 OpenAI 的每分钟令牌限制。系统通过两种方式控制主模型的上下文长度：如果上下文中的消息超过最大参数，则修剪上下文中的消息；一旦消息数量超过另一个参数，则对上下文进行汇总。这些参数可在常量.py;他们的描述如下。

• CONTEXT_MAX_MESSAGES|删除前应保持上下文的消息 |默认 = 100
• CONTEXT_SUMMARIZE_THRESHOLD|摘要生成之前上下文中的消息 |默认 = 20

这总结和修剪图中的节点仅在内容摘要后才会修剪消息。您可以在下面的代码片段中看到与此节点关联的 Python 代码。

异步 def _summarize_and_trim(
状态：状态，配置：RunnableConfig，*，存储：BaseStore
) -> 字典[str, 列表[AnyMessage]]:
"""用于总结和修剪消息历史记录的协程。"""
摘要 = state.get("摘要", "")
如果总结：
摘要消息 = SUMMARY_PROMPT_TEMPLATE.format(summary=summary)
别的：
摘要消息 = SUMMARY_INITIAL_PROMPT
消息 = (
[系统消息（内容=SUMMARY_SYSTEM_PROMPT）] +
状态[“消息”] +
[HumanMessage(内容=summary_message)]）
模型 = config["可配置"]["vlm_model"]
选项 = config["可配置"]["选项"]
model_with_config = model.with_config(
配置={
“模型”：选项.get(
CONF_VLM，
推荐_VLM，
),
“温度”：选项.get(
CONF_SUMMARIZATION_MODEL_TEMPERATURE，
RECOMMENDED_SUMMARIZATION_MODEL_TEMPERATURE，
),
"top_p": 选项.get(
CONF_SUMMARIZATION_MODEL_TOP_P，
RECOMMENDED_SUMMARIZATION_MODEL_TOP_P，
),
“num_predict”：VLM_NUM_PREDICT，
}
）
LOGGER.debug("摘要消息：%s", messages)
响应 = 等待 model_with_config.ainvoke(消息)
# 修剪消息历史记录以管理上下文窗口长度。
修剪消息 = 修剪消息（
消息=状态[“消息”]，
token_counter=len,
max_tokens=CONTEXT_MAX_MESSAGES,
策略=“最后”，
start_on=“人类”，
include_system=真，
）
messages_to_remove = [m for m in state["messages"] if m 不在trimmed_messages中]
LOGGER.debug("要删除的消息：%s", messages_to_remove)
remove_messages = [RemoveMessage(id=m.id) for messages_to_remove]
返回{“summary”：response.content，“messages”：remove_messages}
延迟

用户请求之间的延迟或代理代表用户及时采取行动之间的延迟对于您在设计中考虑至关重要。

我使用了多种技术来减少延迟，包括使用在边缘运行的专门的、较小的帮助器 LLM，并通过构建提示来放置静态内容（例如指令和示例）、前期和可变内容（例如用户特定的内容）来促进主要模型提示缓存信息在最后。这些技术还大大降低了主要模型的使用成本。

您可以在下面看到典型的延迟性能。

• HA 意图（例如，开灯）|< 1 秒
• 分析相机图像（初始请求）|< 3 秒
• 添加自动化 |< 1 秒
• 内存操作 |< 1 秒

工具

代理可以使用HA工具中指定的法学硕士API以及 LangChain 框架中内置的其他工具，如工具.py。此外，您还可以使用自己的工具扩展 LLM API。该代码为主要 LLM 提供了它可以调用的工具列表，以及在其系统消息和工具的 Python 函数定义的文档字符串中使用它们的说明。您可以在下面的代码片段中看到文档字符串指令的示例获取并分析相机图像工具。

@工具（parse_docstring = False）
异步 def get_and_analyze_camera_image( # noqa: D417
相机名称：str，
检测关键字：列表[str] |无=无，
*,
# 从模型中隐藏这些参数。
配置：带注释的[RunnableConfig，InjectedToolArg（）]，
) -> 字符串:
”“”
获取相机图像并对其进行场景分析。
参数：
camera_name：用于场景分析的相机名称。
detector_keywords：要在图像中查找的特定对象（如果有）。
例如，如果用户说“检查前廊摄像头是否有
盒子和狗”，检测关键字将是[“盒子”，“狗”]。
”“”
hass = config["可配置"]["hass"]
vlm_model = config["可配置"]["vlm_model"]
选项 = config["可配置"]["选项"]
图像=等待_get_camera_image（hass，相机名称）
返回等待_analyze_image（vlm_model，选项，图像，检测关键字）

如果代理决定使用工具，LangGraph 节点行动输入后，节点的代码就会运行该工具。该节点使用简单的错误恢复机制，如果出现错误，将要求代理尝试使用更正的参数再次调用该工具。下面的代码片段显示了与行动节点。

异步 def _call_tools(
状态：状态，配置：RunnableConfig，*，存储：BaseStore
) -> dict[str, list[ToolMessage]]:
"""调用 Home Assistant 或 langchain LLM 工具的协程。"""
# 工具调用将是状态中的最后一条消息。
tool_calls = 状态["消息"][-1].tool_calls
langchain_tools = config["可配置"]["langchain_tools"]
ha_llm_api = config["可配置"]["ha_llm_api"]
tool_responses: 列表[ToolMessage] = []
对于 tool_calls 中的 tool_call：
工具名称=工具调用[“名称”]
tool_args = tool_call["args"]
记录器.调试（
“工具调用：%s(%s)”，tool_name，tool_args）
def _handle_tool_error(err:str, name:str, tid:str) -> ToolMessage:
返回工具消息(
内容=TOOL_CALL_ERROR_TEMPLATE.format（错误=错误），
姓名=姓名，
tool_call_id=tid,
状态=“错误”，
）
# 调用了 langchain 工具。
如果 langchain_tools 中的工具名称：
lc_tool = langchain_tools[工具名称.lower()]
# 在运行时向工具提供隐藏参数。
tool_call_copy = copy.deepcopy(tool_call)
tool_call_copy["args"].update(
{
“商店”：商店，
“配置”：配置，
}
）
尝试：
tool_response = 等待 lc_tool.ainvoke(tool_call_copy)
except (HomeAssistantError, ValidationError) as e:
tool_response = _handle_tool_error(repr(e), tool_name, tool_call["id"])
# 调用了 Home Assistant 工具。
别的：
工具输入 = llm.ToolInput(
工具名称=工具名称，
工具参数=工具参数，
）
尝试：
响应 = 等待 ha_llm_api.async_call_tool(tool_input)
工具响应 = 工具消息(
内容=json.dumps(响应),
tool_call_id=tool_call["id"],
名称=工具名称，
）
except (HomeAssistantError, vol.Invalid) as e:
tool_response = _handle_tool_error(repr(e), tool_name, tool_call["id"])
LOGGER.debug("工具响应：%s", tool_response)
tool_responses.append(tool_response)
返回{“消息”：tool_responses}
LLM API 指示代理始终使用 HA 调用工具

内置意图控制 Home Assistant 时，使用意图“HassTurnOn”来锁定，使用“HassTurnOff”来解锁。意图描述了由用户操作产生的用户意图。

您可以在下面看到代理可以使用的 LangChain 工具列表。

• 获取并分析相机图像|对相机图像进行场景分析
• 更新插入内存|添加或更新内存
• 添加自动化|创建并注册 HA 自动化
• 获取实体历史记录|查询 HA 数据库以获取实体历史记录

硬件

我在具有 SSD 存储、Zigbee 和 LAN 连接的 Raspberry Pi 5 上构建了 HA 安装。我在一台基于 Ubuntu 的服务器上部署了 Ollama 下的边缘模型，该服务器配备 AMD 64 位 3.4 GHz CPU、Nvidia 3090 GPU 和 64 GB 系统 RAM。服务器与 Raspberry Pi 在同一 LAN 中。

我已经在家里使用这个项目几个星期了，发现它很有用，但在我将致力于解决的一些领域令人沮丧。以下是我与代理的经验的优缺点列表。

优点

• 相机图像场景分析非常有用且灵活，因为您几乎可以查询任何内容，而不必像传统的 ML 方法那样担心是否有正确的分类器。
• 自动化设置非常容易，但也可能相当复杂。初级法学硕士在生成符合 HA 的 YAML 方面的表现令人惊叹。
• 大多数情况下，延迟是可以接受的。
• 使用 LangChain 和 LangGraph 添加额外的 LLM 工具和图形状态非常容易。

缺点

• 相机图像分析似乎不如传统的机器学习方法准确。例如，模型很难检测部分被遮挡的包裹。
• 初级模型云成本较高。每 30 分钟运行一次单个包裹检测器的成本约为每天 2.50 美元。
• 对辅助法学硕士使用结构化模型输出，这将使下游法学硕士处理更容易，但会大大降低准确性。
• 代理人需要更加主动。向代理图添加规划步骤有望解决这个问题。

以下是您可以使用主生成代理 (HGA) 集成执行的操作的一些示例，如我在与 HA 安装交互期间截取的“协助”对话框的屏幕截图所示。

• 创建自动化。

• 创建定期运行的自动化。

下面的代码片段显示，根据代理生成并注册为 HA 自动化的内容，该代理能够熟练使用 YAML。

别名： 检查垃圾箱废物抽屉
触发器：
- 分钟：/30
触发器：时间模式
状况：
- 条件：数字状态
实体ID：sensor.litter_robot_4_waste_drawer
以上：90
行动：
- 数据：
消息：垃圾箱垃圾抽屉已满 90% 以上！
动作：notify.notify

• 检查相机。

• 检查多个摄像头（作者视频）。

https://github.com/user-attachments/assets/230baae5-8702-4375-a3f0-ffa981ee66a3

• 总结一下家乡的情况（视频作者）。

https://github.com/user-attachments/assets/96f834a8-58cc-4bd9-a899-4604c1103a98

• 具有语义搜索的长期记忆。

您可以看到代理正确生成了下面的自动化。

别名： 准备回家
描述：晚上 7:30 打开前廊灯并解锁车库门锁
模式：单人
触发器：
- 时间：“19:30:00”
触发：时间
行动：
- 目标：
实体_id：light.front_porch_light
动作：light.turn_on
数据： {}
- 目标：
实体_id：lock.garage_door_lock
动作：锁定.解锁
数据： {}

• 检查相机中是否有包裹。

下面是代理分析的相机图像，您可以看到两个包裹是可见的。

• 包裹递送的主动通知。

如果有任何盒子或包裹可见，下面是此自动化的通知示例。

家庭生成代理提供了一种有趣的方式，使您的家庭助理设置更加用户友好和直观。通过实现自然语言交互和简化自动化，它为日常智能家居使用提供了实用且有用的工具。

使用家庭生成代理会带来安全、隐私和成本风险，需要进一步缓解和创新，才能真正对大众市场有用。

无论您是 Home Assistant 的新手还是经验丰富的用户，这种集成都是增强系统功能并熟悉在家里使用生成式 AI 和代理的好方法。如果您有兴趣探索其潜力，请访问Home 生成代理 GitHub 页面今天就开始吧。

安装

1. 使用选择的工具，打开 HA 配置的目录（文件夹）（您可以在其中找到配置.yaml）。

2. 如果您没有`custom_components`目录（文件夹），则必须创建它。

3. 在自定义组件目录（文件夹），创建一个新文件夹，名为家庭生成代理。

4. 下载_全部_文件来自自定义组件/home_generative_agent/此存储库中的目录（文件夹）。

4. 将下载的文件放入您创建的新目录（文件夹）中。

6. 重启家庭助手

7. 在 HA UI 中，转到“配置”->“集成”，单击“+,”并搜索“Home Generative Agent”

8. 安装所有蓝图蓝图目录（文件夹）。

配置

配置是在 HA UI 中通过以下参数完成的常量.py。.