pipeline home assistant

[nathan-skynet]

The best way to integrate a home-assistant pipeline... ?

[Fusseldieb]

Without further information we can't help you.

[nathan-skynet]

i'd like a pipeline that responds to my requests via open-webui a home assistant like this
user : “turn on the kitchen light”
system: “Of course
or something like
user: “turn on the kitchen light”.
system: “The kitchen light is already on”.

[rootcli]

In Home Assistant Jun release, they integrated LLMs to the platform, but haven't got the chance to explore it yet

[atgehrhardt]

I have put together a function calling pipeline with basic Home Assistant Light control. It's a work in progress. Wondering if I should make a PR?

Right now you can list all lights, and turn lights on and off with fuzzy matching. Will work on improving matching logic as well as ability to turn multiple lights on and off at once (right now It can treat light groups as one light so that works well).

Works really well with llama3 and llama3:instruct

After lights, I will work on adding logic for other entity types:

import os
import requests
from typing import Literal, Optional
from datetime import datetime
import pytz
from difflib import get_close_matches

from blueprints.function_calling_blueprint import Pipeline as FunctionCallingBlueprint

class Pipeline(FunctionCallingBlueprint):
    class Valves(FunctionCallingBlueprint.Valves):
        HOME_ASSISTANT_URL: str = ""
        HOME_ASSISTANT_TOKEN: str = ""

    class Tools:
        def __init__(self, pipeline) -> None:
            self.pipeline = pipeline

        def get_current_time(self) -> str:
            """
            Get the current time in EST.

            :return: The current time in EST.
            """
            now_est = datetime.now(pytz.timezone('US/Eastern'))  # Get the current time in EST
            current_time = now_est.strftime("%I:%M %p")  # %I for 12-hour clock, %M for minutes, %p for am/pm
            print(f"Current Time = {current_time}")
            return f"Current Time = {current_time}"

        def get_all_lights(self) -> dict:
            """
            Lists my lights.
            Shows me my lights.
            Get a dictionary of all lights in my home.

            :return: A dictionary of light entity names and their IDs.
            """
            if not self.pipeline.valves.HOME_ASSISTANT_URL or not self.pipeline.valves.HOME_ASSISTANT_TOKEN:
                return {"error": "Home Assistant URL or token not set, ask the user to set it up."}
            else:
                url = f"{self.pipeline.valves.HOME_ASSISTANT_URL}/api/states"
                headers = {
                    "Authorization": f"Bearer {self.pipeline.valves.HOME_ASSISTANT_TOKEN}",
                    "Content-Type": "application/json",
                }

                response = requests.get(url, headers=headers)
                response.raise_for_status()  # Raises an HTTPError for bad responses
                data = response.json()

                lights = {entity["attributes"]["friendly_name"]: entity["entity_id"]
                          for entity in data if entity["entity_id"].startswith("light.")}

                return lights

        def control_light(self, name: str, state: Literal['on', 'off']) -> str:
            """
            Turn a light on or off based on its name.

            :param name: The friendly name of the light.
            :param state: The desired state ('on' or 'off').
            :return: The result of the operation.
            """
            if not self.pipeline.valves.HOME_ASSISTANT_URL or not self.pipeline.valves.HOME_ASSISTANT_TOKEN:
                return "Home Assistant URL or token not set, ask the user to set it up."

            # Normalize the light name by converting to lowercase and stripping extra spaces
            normalized_name = " ".join(name.lower().split())

            # Get a dictionary of all lights
            lights = self.get_all_lights()
            if "error" in lights:
                return lights["error"]

            # Find the closest matching light name
            light_names = list(lights.keys())
            closest_matches = get_close_matches(normalized_name, light_names, n=1, cutoff=0.6)

            if not closest_matches:
                return f"Light named '{name}' not found."

            best_match = closest_matches[0]
            light_id = lights[best_match]

            url = f"{self.pipeline.valves.HOME_ASSISTANT_URL}/api/services/light/turn_{state}"
            headers = {
                "Authorization": f"Bearer {self.pipeline.valves.HOME_ASSISTANT_TOKEN}",
                "Content-Type": "application/json",
            }
            payload = {
                "entity_id": light_id
            }

            response = requests.post(url, headers=headers, json=payload)
            if response.status_code == 200:
                return f"ONLY RESPOND 'Will do' TO THE USER. DO NOT SAY ANYTHING ELSE!"
            else:
                return f"ONLY RESPOND 'Couldn't find light' TO THE USER. DO NOT SAY ANYTHING ELSE!"

    def __init__(self):
        super().__init__()
        self.name = "My Tools Pipeline"
        self.valves = self.Valves(
            **{
                **self.valves.model_dump(),
                "pipelines": ["*"],  # Connect to all pipelines
            },
        )
        self.tools = self.Tools(self)

[tjbck]

Yes! This is awesome! 🙌

[atgehrhardt]

Sweet I'll make a PR this week when I have a more solid version.

I am attempting to employ a mechanism that allows for chaining function calls together without using an agent pipeline (for speed).

If anyone has had successful strategies around this please feel free to weigh in. The only thing i'm not considering is using nexusravenv2 because it can easily support this well, BUT it would require writing the pipeline specifically for it and I want it to be as universal as possible.

As is it can pretty consistently execute one function at a time no problem.

[atgehrhardt]

Anybody interesting in trying current version can grab the code from the PR:

#95