基于LangGraph集成Home Assistant MCP的智能语音管家

SmartDeng2025/7/13...大约 4 分钟

基于LangGraph集成Home Assistant MCP的智能语音管家

在智能家居领域，语音控制和自然语言交互正变得越来越重要。本文将详细介绍如何基于LangGraph框架，将Home Assistant的MCP（Model Context Protocol）服务集成进智能语音管家Moss，实现真正的自然语言家居控制。文章将以实用代码为主，帮助你快速上手。

GitHub链接：mawwalker/Moss，欢迎star和交流。

前言

传统的智能家居控制方式往往依赖固定指令，用户体验有限。随着大语言模型（LLM）和LangGraph等Agent框架的发展，我们可以让语音助手理解自然语言，通过MCP协议与Home Assistant深度集成，实现灵活、强大的家居控制。

MCP（Model Context Protocol）是一个开放标准，允许大语言模型安全地连接到外部数据源和工具。在本项目中，Moss智能管家通过MCP服务器，直接调用Home Assistant的各种功能，实现设备状态查询、场景控制等操作。

一、MCP集成架构概览

本项目的核心目标是让LangGraph Agent通过MCP协议与Home Assistant无缝对接。整体架构如下：

LangGraph Agent：负责自然语言理解与推理
MCP工具集成：通过langchain_mcp_tools将Home Assistant暴露为标准工具
异步资源管理：采用异步上下文管理器，确保MCP连接的正确初始化与释放

二、Home Assistant MCP工具集成

1. 构建HASS MCP工具

我们通过langchain_mcp_tools包，将Home Assistant的MCP服务转为LangChain可用的工具。核心代码如下：

from langchain_mcp_tools import convert_mcp_to_langchain_tools
from loguru import logger
import os
from config.conf import hass_config, PROJECT_ROOT

async def build_hass_tools() -> tuple:
    try:
        hass_mcp_config = {
            "homeassistant": {
                "transport": "stdio",
                "command": "uv",
                "args": [
                    "run",
                    "mcp",
                    "run",
                    "hass-mcp/app/server.py",
                ],
                "env": {
                    "HA_URL": hass_config["hass_url"],
                    "HA_TOKEN": hass_config["token"],
                    "PYTHONPATH": os.path.join(PROJECT_ROOT, "hass-mcp"),
                }
            }
        }
        tools, cleanup = await convert_mcp_to_langchain_tools(hass_mcp_config)
        return tools, cleanup
    except Exception as e:
        logger.error(f"Failed to build HASS tools: {e}")
        return [], lambda: None

说明：

通过配置hass_mcp_config，指定Home Assistant的MCP服务启动方式、环境变量（如HA_URL、HA_TOKEN等）。
convert_mcp_to_langchain_tools会自动将MCP服务暴露的所有功能转为LangChain标准工具。
返回的tools可直接用于Agent，cleanup用于资源释放。

2. 工具初始化与Agent集成

在Agent初始化时，动态加载HASS工具，并支持资源自动清理：

from agent_core.hass_mcp import build_hass_tools
from config.conf import hass_config
from loguru import logger

async def init_tools():
    tools = []
    cleanup_funcs = []
    if hass_config["enable"]:
        logger.debug("Initializing HASS tools...")
        try:
            hass_tools, cleanup = await build_hass_tools()
            tools.extend(hass_tools)
            cleanup_funcs.append(cleanup)
        except Exception as e:
            logger.error(f"Failed to initialize HASS tools: {e}")
    logger.info(f"Initialized tools: {[tool.name for tool in tools]}")
    return tools, cleanup_funcs

三、LangGraph Agent与MCP工具协作

MossAgent采用异步上下文管理器，确保MCP连接的生命周期安全。Agent初始化时自动加载HASS工具，支持流式响应和动态系统提示词。

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from agent_core.tools import init_tools
from agent_core.prompts import agent_additional_prompts
from langchain_core.runnables import RunnableConfig
from loguru import logger
import asyncio
from datetime import datetime

class MossAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="your-model",
            temperature=0.1,
            api_key="your-api-key",
            base_url="your-base-url",
        )
        self.agent = None
        self._initialized = False
        self._cleanup_funcs = []
    async def _initialize(self):
        if not self._initialized:
            tools, cleanup_funcs = await init_tools()
            self._cleanup_funcs.extend(cleanup_funcs)
            self.agent = create_react_agent(
                model=self.llm,
                tools=tools,
                prompt=self.dynamic_system_prompt
            )
            self._initialized = True
    async def cleanup(self):
        for cleanup_func in self._cleanup_funcs:
            try:
                if asyncio.iscoroutinefunction(cleanup_func):
                    await cleanup_func()
                else:
                    cleanup_func()
            except Exception as e:
                logger.error(f"Error during cleanup: {e}")
        self._cleanup_funcs.clear()
    async def __aenter__(self):
        await self._initialize()
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.cleanup()
    @staticmethod
    def dynamic_system_prompt(state, config):
        current_date = config.get("configurable", {}).get("current_date", "")
        language = config.get("configurable", {}).get("language", "zh")
        system_msg = agent_additional_prompts.format(
            current_date=current_date,
            language=language
        )
        return [{"role": "system", "content": system_msg}] + state["messages"]
    async def run(self, messages, config=None):
        await self._initialize()
        result = self.agent.astream(
            {"messages": messages},
            stream_mode=["updates", "messages", "custom"],
            config=config,
        )
        async for stream_mode, chunk in result:
            if stream_mode == "messages":
                content = chunk[0].content
                yield content

亮点说明：

支持异步上下文管理，自动清理MCP连接
动态系统提示词，支持多语言和上下文自适应
流式响应，提升用户体验

四、使用示例

假设你已经配置好Home Assistant和MCP服务，以下是一个简单的调用示例：

import asyncio
from datetime import datetime
from langchain_core.runnables import RunnableConfig

messages = [
    {"role": "user", "content": "关闭床头灯"},
]

async def main():
    configurable = {
        "current_date": datetime.now().strftime("%Y-%m-%d"),
        "language": "Chinese, Mandarin"
    }
    async with MossAgent() as agent:
        async for chunk in agent.run(messages, config=RunnableConfig(configurable=configurable)):
            print(chunk)

asyncio.run(main())

五、MCP集成优势

标准化接口：MCP为LLM与Home Assistant之间提供了统一协议，极大简化了集成难度。
实时通信：支持双向实时通信，设备状态变化可即时反馈。
安全可靠：内置安全机制，保障数据与操作安全。
高扩展性：后续如需扩展Home Assistant功能，仅需在MCP服务端添加，无需修改Agent核心代码。

六、总结

通过LangGraph与MCP协议的深度集成，Moss智能语音管家实现了真正意义上的自然语言家居控制。无论是设备开关、场景切换还是状态查询，都可以通过自然语言流畅完成。MCP协议的引入让系统更标准、更安全、更易扩展。

如果你也在构建智能家居语音助手，强烈推荐采用LangGraph+MCP的架构。欢迎交流、提问和star项目！