學習如何創建一個能理解您家居環境的代理人,學習您的偏好,並與您和您的家互動,以完成您認為有價值的活動。
這篇文章介紹了一個名為 home-generative-agent 的家庭助手 (HA) 整合架構與設計。這個專案使用 LangChain 和 LangGraph 創建了一個生成式 AI 代理人,能在 HA 智能家居環境中互動和自動化任務。這個代理人能理解您家的情境,學習您的喜好,並與您和您的家互動,完成您認為有價值的活動。主要功能包括創建自動化、分析圖像以及使用各種大型語言模型 (LLM) 管理家庭狀態。這個架構結合了雲端和邊緣模型,以達到最佳性能和成本效益。安裝說明、配置細節以及專案架構和使用的不同模型的資訊都可以在 home-generative-agent 的 GitHub 上找到。這個專案是開源的,歡迎貢獻。
目前支持的一些功能包括:
- 創建複雜的家庭助手自動化。
- 圖像場景分析和理解。
- 實體、設備和區域的家庭狀態分析。
- 對家中允許的實體進行全面的代理控制。
- 使用語義搜索的短期和長期記憶。
- 自動總結家庭狀態,以管理 LLM 上下文長度。
這是我個人的專案,示範了我所稱的學習導向駭客行為。這個專案與我在亞馬遜 (Amazon) 的工作無關,也與負責家庭助手或 LangChain/LangGraph 的組織沒有任何關聯。
創建一個監控和控制您家的代理人可能會導致意外行為,並可能因為 LLM 的幻覺和隱私問題而使您和您的家面臨風險,特別是在將家庭狀態和用戶資訊暴露給雲端 LLM 時。我已經做出合理的架構和設計選擇來減輕這些風險,但這些風險無法完全消除。
一個重要的早期決策是依賴混合雲邊緣方法。這使得能夠使用最先進的推理和規劃模型,應該能幫助減少幻覺。更簡單、更專注於任務的邊緣模型被用來進一步減少 LLM 錯誤。
另一個關鍵決策是利用 LangChain 的能力,這使得敏感資訊可以在運行時隱藏,僅在需要時提供。例如,工具邏輯可能需要使用發出請求的用戶 ID。然而,這些值通常不應由 LLM 控制。允許 LLM 操作用戶 ID 可能會帶來安全和隱私風險。為了減輕這一點,我使用了 InjectedToolArg 註解。
此外,使用大型雲端 LLM 會產生顯著的雲端成本,而運行 LLM 邊緣模型所需的邊緣硬體可能很昂貴。綜合的運營和安裝成本對於普通用戶來說可能是不可負擔的。需要業界共同努力“讓 LLM 的成本像 CNN 一樣便宜”,以將家庭代理人推向大眾市場。
重要的是要意識到這些風險,並理解儘管有這些減輕措施,我們仍然處於這個專案和家庭代理人的一個早期階段。還有大量工作需要完成,才能使這些代理人成為真正有用和可靠的助手。
以下是 home-generative-agent 架構的高層次概述。
一般的整合架構遵循家庭助手核心的最佳實踐,並符合家庭助手社區商店 (HACS) 的發布要求。
這個代理人是使用 LangGraph 構建的,並使用 HA 對話組件與用戶互動。代理人使用家庭助手 LLM API 獲取家庭狀態,並理解其可用的 HA 原生工具。我使用 LangChain 實現了所有其他可用於代理人的工具。代理人使用幾個 LLM,一個大型且非常準確的主要模型用於高層次推理,較小的專門輔助模型用於相機圖像分析、主要模型上下文總結和長期語義搜索的嵌入生成。主要模型是雲端的,而輔助模型是邊緣的,運行在位於家庭中的電腦上,並在 Ollama 框架下運行。
目前使用的模型總結如下。
基於 LangGraph 的代理人
LangGraph 驅動對話代理,使您能夠快速創建有狀態的多角色應用程序,利用 LLM 的能力。它擴展了 LangChain 的功能,引入了創建和管理循環圖的能力,這對於開發複雜的代理運行時至關重要。圖模型化了代理的工作流程,如下圖所示。
代理工作流程有五個節點,每個 Python 模組修改代理的狀態,這是一個共享數據結構。節點之間的邊表示允許的轉換,實線為無條件,虛線為有條件。節點執行工作,邊告訴接下來要做什麼。
__start__ 和 __end__ 節點告訴圖從哪裡開始和停止。代理節點運行主要 LLM,如果它決定使用工具,則動作節點運行該工具,然後將控制權返回給代理。summarize_and_trim 節點處理 LLM 的上下文,以管理增長,同時保持準確性,如果代理沒有工具可以調用,並且消息數量滿足以下條件。
LLM 上下文管理
您需要仔細管理 LLM 的上下文長度,以平衡成本、準確性和延遲,並避免觸發速率限制,例如 OpenAI 的每分鐘令牌限制。系統以兩種方式控制主要模型的上下文長度:如果消息超過最大參數,則修剪上下文中的消息,並且一旦消息數量超過另一個參數,則對上下文進行總結。這些參數可以在 const.py 中配置;其描述如下。
- CONTEXT_MAX_MESSAGES | 在刪除之前要保留的上下文消息 | 默認 = 100
- CONTEXT_SUMMARIZE_THRESHOLD | 在生成摘要之前的上下文消息 | 默認 = 20
圖中的 summarize_and_trim 節點將僅在內容總結後修剪消息。您可以在下面的代碼片段中看到與此節點相關的 Python 代碼。
async def _summarize_and_trim(state: State, config: RunnableConfig, *, store: BaseStore) -> dict[str, list[AnyMessage]:”””協程以總結和修剪消息歷史。”””summary = state.get(“summary”, “”)
if summary:summary_message = SUMMARY_PROMPT_TEMPLATE.format(summary=summary)else:summary_message = SUMMARY_INITIAL_PROMPT
messages = ([SystemMessage(content=SUMMARY_SYSTEM_PROMPT)] +state[“messages”] +[HumanMessage(content=summary_message)])
model = config[“configurable”][“vlm_model”]options = config[“configurable”][“options”]model_with_config = model.with_config(config=”model”: options.get(CONF_VLM,RECOMMENDED_VLM,),”temperature”: options.get(CONF_SUMMARIZATION_MODEL_TEMPERATURE,RECOMMENDED_SUMMARIZATION_MODEL_TEMPERATURE,),”top_p”: options.get(CONF_SUMMARIZATION_MODEL_TOP_P,RECOMMENDED_SUMMARIZATION_MODEL_TOP_P,),”num_predict”: VLM_NUM_PREDICT,)
LOGGER.debug(“Summary messages: %s”, messages)response = await model_with_config.ainvoke(messages)
# 修剪消息歷史以管理上下文窗口長度.trimmed_messages = trim_messages(messages=state[“messages”],token_counter=len,max_tokens=CONTEXT_MAX_MESSAGES,strategy=”last”,start_on=”human”,include_system=True,)messages_to_remove = [m for m in state[“messages”] if m not in trimmed_messages]LOGGER.debug(“Messages to remove: %s”, messages_to_remove)remove_messages = [RemoveMessage(id=m.id) for m in messages_to_remove]
return “summary”: response.content, “messages”: remove_messages
延遲
用戶請求之間的延遲或代理人及時為用戶採取行動的能力是您在設計中需要考慮的關鍵。我使用了幾種技術來減少延遲,包括使用運行在邊緣的專門較小的輔助 LLM,並通過結構化提示來促進主要模型的提示緩存,將靜態內容(如指令和示例)放在前面,將可變內容(如用戶特定資訊)放在後面。這些技術還顯著降低了主要模型的使用成本。
您可以在下面看到典型的延遲性能。
- HA 意圖(例如,打開燈) | < 1 秒
- 分析相機圖像(初始請求) | < 3 秒
- 添加自動化 | < 1 秒
- 記憶操作 | < 1 秒
工具
代理人可以使用 HA 工具,如 LLM API 中所指定的,以及在 tools.py 中定義的 LangChain 框架中內建的其他工具。此外,您還可以擴展 LLM API,添加您自己的工具。代碼會將代理人可調用的工具列表提供給主要 LLM,並在其系統消息和工具的 Python 函數定義的文檔字符串中提供使用說明。您可以在下面的代碼片段中看到 get_and_analyze_camera_image 工具的文檔字符串說明示例。
@tool(parse_docstring=False)async def get_and_analyze_camera_image( # noqa: D417camera_name: str,detection_keywords: list[str] | None = None,*,# 隱藏這些參數不讓模型看到.config: Annotated[RunnableConfig, InjectedToolArg()],) -> str:”””獲取相機圖像並對其進行場景分析。
Args:camera_name: 用於場景分析的相機名稱.detection_keywords: 圖像中要查找的特定物體(如果有)。例如,如果用戶說“檢查前廊相機是否有箱子和狗”,則 detection_keywords 將是 [“boxes”, “dogs”]。
“””hass = config[“configurable”][“hass”]vlm_model = config[“configurable”][“vlm_model”]options = config[“configurable”][“options”]image = await _get_camera_image(hass, camera_name)return await _analyze_image(vlm_model, options, image, detection_keywords)
如果代理人決定使用工具,則進入 LangGraph 節點動作,並且該節點的代碼運行該工具。該節點使用簡單的錯誤恢復機制,會要求代理人使用更正的參數再次嘗試調用工具。下面的代碼片段顯示了與動作節點相關的 Python 代碼。
async def _call_tools(state: State, config: RunnableConfig, *, store: BaseStore) -> dict[str, list[ToolMessage]]:”””協程以調用家庭助手或 langchain LLM 工具。”””# 工具調用將是狀態中的最後一條消息.tool_calls = state[“messages”][-1].tool_calls
langchain_tools = config[“configurable”][“langchain_tools”]ha_llm_api = config[“configurable”][“ha_llm_api”]
tool_responses: list[ToolMessage] = []for tool_call in tool_calls:tool_name = tool_call[“name”]tool_args = tool_call[“args”]
LOGGER.debug(“Tool call: %s(%s)”, tool_name, tool_args)
def _handle_tool_error(err:str, name:str, tid:str) -> ToolMessage:return ToolMessage(content=TOOL_CALL_ERROR_TEMPLATE.format(error=err),name=name,tool_call_id=tid,status=”error”,)
# 調用了 langchain 工具.if tool_name in langchain_tools:lc_tool = langchain_tools[tool_name.lower()]
# 在運行時向工具提供隱藏參數.tool_call_copy = copy.deepcopy(tool_call)tool_call_copy[“args”].update(“store”: store,”config”: config,)
try:tool_response = await lc_tool.ainvoke(tool_call_copy)except (HomeAssistantError, ValidationError) as e:tool_response = _handle_tool_error(repr(e), tool_name, tool_call[“id”])# 調用了家庭助手工具.else:tool_input = llm.ToolInput(tool_name=tool_name,tool_args=tool_args,)
try:response = await ha_llm_api.async_call_tool(tool_input)
tool_response = ToolMessage(content=json.dumps(response),tool_call_id=tool_call[“id”],name=tool_name,)except (HomeAssistantError, vol.Invalid) as e:tool_response = _handle_tool_error(repr(e), tool_name, tool_call[“id”])
LOGGER.debug(“Tool response: %s”, tool_response)tool_responses.append(tool_response)return “messages”: tool_responses
LLM API 指示代理人始終使用 HA 內建意圖來控制家庭助手,並使用意圖 `HassTurnOn` 鎖定和 `HassTurnOff` 解鎖鎖。意圖描述用戶行為生成的用戶意圖。
您可以在下面看到代理人可以使用的 LangChain 工具列表。
- get_and_analyze_camera_image | 對相機圖像進行場景分析
- upsert_memory | 添加或更新記憶
- add_automation | 創建並註冊 HA 自動化
- get_entity_history | 查詢 HA 數據庫中的實體歷史
硬體
我在 Raspberry Pi 5 上構建了 HA 安裝,配備 SSD 存儲、Zigbee 和 LAN 連接。我在一台擁有 AMD 64 位 3.4 GHz CPU、Nvidia 3090 GPU 和 64 GB 系統 RAM 的 Ubuntu 伺服器上部署了邊緣模型。該伺服器與 Raspberry Pi 在同一 LAN 上。
我在家中使用這個專案已經幾週,發現它很有用,但在某些方面也令人沮喪,我將努力解決這些問題。以下是我與代理人經驗的優缺點列表。
優點
- 相機圖像場景分析非常有用且靈活,因為您幾乎可以查詢任何內容,而不必擔心擁有正確的分類器,這與傳統的機器學習方法不同。
- 自動化設置非常簡單,且可以相當複雜。主要 LLM 在生成符合 HA 的 YAML 方面表現非常出色。
- 在大多數情況下,延遲是可以接受的。
- 使用 LangChain 和 LangGraph 添加額外的 LLM 工具和圖狀態非常簡單。
缺點
- 相機圖像分析似乎不如傳統的機器學習方法準確。例如,檢測部分遮擋的包裹對模型來說非常困難。
- 主要模型的雲端成本很高。每 30 分鐘運行一次包檢測器的成本約為 2.50 美元。
- 使用結構化模型輸出來輔助 LLM,這會使下游 LLM 處理變得更容易,但會顯著降低準確性。
- 代理人需要更主動。希望在代理圖中添加一個規劃步驟來解決這個問題。
以下是一些示例,展示您可以使用 home-generative-agent (HGA) 整合完成的操作,並附上我在與 HA 安裝互動過程中拍攝的 Assist 對話的截圖。
創建一個定期運行的自動化。
下面的片段顯示了代理人基於其生成和註冊的 HA 自動化,流利地使用 YAML。
alias: 檢查貓砂盒廢物抽屜triggers:- minutes: /30trigger: time_patternconditions:- condition: numeric_stateentity_id: sensor.litter_robot_4_waste_drawerabove: 90actions:- data:message: 貓砂盒廢物抽屜已超過 90% 滿!action: notify.notify
檢查多個相機(作者的視頻)。
總結家庭狀態(作者的視頻)。
長期記憶與語義搜索。
本文由 AI 台灣 運用 AI 技術編撰,內容僅供參考,請自行核實相關資訊。
歡迎加入我們的 AI TAIWAN 台灣人工智慧中心 FB 社團,
隨時掌握最新 AI 動態與實用資訊!