ovhcloud ai endpoints: initial commit

Author: Crocmagnon

.strict-typing

@@ -428,6 +428,7 @@ homeassistant.components.otp.*
 homeassistant.components.ouman_eh_800.*
 homeassistant.components.overkiz.*
 homeassistant.components.overseerr.*
+homeassistant.components.ovhcloud_ai_endpoints.*
 homeassistant.components.p1_monitor.*
 homeassistant.components.paj_gps.*
 homeassistant.components.panel_custom.*

CODEOWNERS

@@ -0,0 +1,56 @@
+"""The OVHcloud AI Endpoints integration."""
+
+from openai import AsyncOpenAI, OpenAIError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_KEY, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.httpx_client import get_async_client
+
+from .const import BASE_URL
+
+PLATFORMS = [Platform.CONVERSATION]
+
+type OVHcloudAIEndpointsConfigEntry = ConfigEntry[AsyncOpenAI]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: OVHcloudAIEndpointsConfigEntry
+) -> bool:
+    """Set up OVHcloud AI Endpoints from a config entry."""
+    client = AsyncOpenAI(
+        base_url=BASE_URL,
+        api_key=entry.data[CONF_API_KEY],
+        http_client=get_async_client(hass),
+    )
+
+    try:
+        # Unfortunately I couldn't find an endpoint that would authenticate the key
+        # without calling an LLM. This always succeeds regardless of auth.
+        async for _ in client.with_options(timeout=10.0).models.list():
+            break
+    except OpenAIError as err:
+        raise ConfigEntryNotReady(err) from err
+
+    entry.runtime_data = client
+
+    entry.async_on_unload(entry.add_update_listener(async_update_entry))
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_update_entry(
+    hass: HomeAssistant, entry: OVHcloudAIEndpointsConfigEntry
+) -> None:
+    """Reload the entry when its data or subentries change."""
+    await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: OVHcloudAIEndpointsConfigEntry
+) -> bool:
+    """Unload OVHcloud AI Endpoints."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/ovhcloud_ai_endpoints/__init__.py

@@ -0,0 +1,173 @@
+"""Config flow for the OVHcloud AI Endpoints integration."""
+
+import logging
+from typing import Any
+
+from openai import AsyncOpenAI, OpenAIError
+import voluptuous as vol
+
+from homeassistant.config_entries import (
+    ConfigEntry,
+    ConfigEntryState,
+    ConfigFlow,
+    ConfigFlowResult,
+    ConfigSubentryFlow,
+    SubentryFlowResult,
+)
+from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, CONF_MODEL
+from homeassistant.core import callback
+from homeassistant.helpers import llm
+from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.helpers.selector import (
+    SelectOptionDict,
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
+    TemplateSelector,
+)
+
+from .const import BASE_URL, CONF_PROMPT, DOMAIN, RECOMMENDED_CONVERSATION_OPTIONS
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class OVHcloudAIEndpointsConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for OVHcloud AI Endpoints."""
+
+    VERSION = 1
+    MINOR_VERSION = 1
+
+    @classmethod
+    @callback
+    def async_get_supported_subentry_types(
+        cls, config_entry: ConfigEntry
+    ) -> dict[str, type[ConfigSubentryFlow]]:
+        """Return subentries supported by this handler."""
+        return {"conversation": ConversationFlowHandler}
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            self._async_abort_entries_match(user_input)
+            client = AsyncOpenAI(
+                base_url=BASE_URL,
+                api_key=user_input[CONF_API_KEY],
+                http_client=get_async_client(self.hass),
+            )
+            try:
+                # Unfortunately I couldn't find an endpoint that would authenticate the key
+                # without calling an LLM. This always succeeds regardless of auth.
+                async for _ in client.with_options(timeout=10.0).models.list():
+                    break
+            except OpenAIError:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title="OVHcloud AI Endpoints",
+                    data=user_input,
+                )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_API_KEY): str,
+                }
+            ),
+            errors=errors,
+        )
+
+
+class ConversationFlowHandler(ConfigSubentryFlow):
+    """Handle conversation subentry flow."""
+
+    def __init__(self) -> None:
+        """Initialize the subentry flow."""
+        self.models: list[str] = []
+        self.options: dict[str, Any] = {}
+
+    async def _get_models(self) -> None:
+        """Fetch models from OVHcloud AI Endpoints."""
+        client: AsyncOpenAI = self._get_entry().runtime_data
+        self.models = [
+            model.id async for model in client.with_options(timeout=10.0).models.list()
+        ]
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """User flow to create a conversation agent."""
+        self.options = RECOMMENDED_CONVERSATION_OPTIONS.copy()
+        return await self.async_step_init(user_input)
+
+    async def async_step_init(
+        self, user_input: dict[str, Any] | None = None
+    ) -> SubentryFlowResult:
+        """Manage conversation agent configuration."""
+        if self._get_entry().state != ConfigEntryState.LOADED:
+            return self.async_abort(reason="entry_not_loaded")
+
+        if user_input is not None:
+            if not user_input.get(CONF_LLM_HASS_API):
+                user_input.pop(CONF_LLM_HASS_API, None)
+            return self.async_create_entry(
+                title=user_input[CONF_MODEL], data=user_input
+            )
+
+        try:
+            await self._get_models()
+        except OpenAIError:
+            return self.async_abort(reason="cannot_connect")
+        except Exception:
+            _LOGGER.exception("Unexpected exception")
+            return self.async_abort(reason="unknown")
+
+        options = [
+            SelectOptionDict(value=model_id, label=model_id) for model_id in self.models
+        ]
+
+        hass_apis: list[SelectOptionDict] = [
+            SelectOptionDict(
+                label=api.name,
+                value=api.id,
+            )
+            for api in llm.async_get_apis(self.hass)
+        ]
+
+        return self.async_show_form(
+            step_id="init",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_MODEL): SelectSelector(
+                        SelectSelectorConfig(
+                            options=options,
+                            mode=SelectSelectorMode.DROPDOWN,
+                            sort=True,
+                        ),
+                    ),
+                    vol.Optional(
+                        CONF_PROMPT,
+                        description={
+                            "suggested_value": self.options.get(
+                                CONF_PROMPT,
+                                RECOMMENDED_CONVERSATION_OPTIONS[CONF_PROMPT],
+                            )
+                        },
+                    ): TemplateSelector(),
+                    vol.Optional(
+                        CONF_LLM_HASS_API,
+                        default=self.options.get(
+                            CONF_LLM_HASS_API,
+                            RECOMMENDED_CONVERSATION_OPTIONS[CONF_LLM_HASS_API],
+                        ),
+                    ): SelectSelector(
+                        SelectSelectorConfig(options=hass_apis, multiple=True)
+                    ),
+                }
+            ),
+        )

homeassistant/components/ovhcloud_ai_endpoints/config_flow.py

@@ -0,0 +1,16 @@
+"""Constants for the OVHcloud AI Endpoints integration."""
+
+import logging
+
+from homeassistant.const import CONF_LLM_HASS_API, CONF_PROMPT
+from homeassistant.helpers import llm
+
+DOMAIN = "ovhcloud_ai_endpoints"
+LOGGER = logging.getLogger(__package__)
+
+BASE_URL = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1"
+
+RECOMMENDED_CONVERSATION_OPTIONS = {
+    CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+    CONF_PROMPT: llm.DEFAULT_INSTRUCTIONS_PROMPT,
+}

homeassistant/components/ovhcloud_ai_endpoints/const.py

@@ -0,0 +1,73 @@
+"""Conversation support for OVHcloud AI Endpoints."""
+
+from typing import Literal
+
+from homeassistant.components import conversation
+from homeassistant.config_entries import ConfigSubentry
+from homeassistant.const import CONF_LLM_HASS_API, CONF_PROMPT, MATCH_ALL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from . import OVHcloudAIEndpointsConfigEntry
+from .const import DOMAIN
+from .entity import OVHcloudAIEndpointsEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: OVHcloudAIEndpointsConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up conversation entities."""
+    for subentry_id, subentry in config_entry.subentries.items():
+        async_add_entities(
+            [OVHcloudAIEndpointsConversationEntity(config_entry, subentry)],
+            config_subentry_id=subentry_id,
+        )
+
+
+class OVHcloudAIEndpointsConversationEntity(
+    OVHcloudAIEndpointsEntity, conversation.ConversationEntity
+):
+    """OVHcloud AI Endpoints conversation agent."""
+
+    _attr_name = None
+
+    def __init__(
+        self,
+        entry: OVHcloudAIEndpointsConfigEntry,
+        subentry: ConfigSubentry,
+    ) -> None:
+        """Initialize the agent."""
+        super().__init__(entry, subentry)
+        if self.subentry.data.get(CONF_LLM_HASS_API):
+            self._attr_supported_features = (
+                conversation.ConversationEntityFeature.CONTROL
+            )
+
+    @property
+    def supported_languages(self) -> list[str] | Literal["*"]:
+        """Return a list of supported languages."""
+        return MATCH_ALL
+
+    async def _async_handle_message(
+        self,
+        user_input: conversation.ConversationInput,
+        chat_log: conversation.ChatLog,
+    ) -> conversation.ConversationResult:
+        """Process the user input and call the API."""
+        options = self.subentry.data
+
+        try:
+            await chat_log.async_provide_llm_data(
+                user_input.as_llm_context(DOMAIN),
+                options.get(CONF_LLM_HASS_API),
+                options.get(CONF_PROMPT),
+                user_input.extra_system_prompt,
+            )
+        except conversation.ConverseError as err:
+            return err.as_conversation_result()
+
+        await self._async_handle_chat_log(chat_log)
+
+        return conversation.async_get_result_from_chat_log(user_input, chat_log)