starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 02:24:51 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add LiteLLM integration (#172960)

Browse Source 
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Luís Alves
2026-07-17 12:01:53 +02:00
committed by GitHub
co-authored by Claude Opus 4.8
parent d1b8881603
commit 40e243a176
21 changed files with 2053 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.strict-typing
+1
View File
@@ -350,6 +350,7 @@ homeassistant.components.lifx.*
homeassistant.components.light.*
homeassistant.components.linkplay.*
homeassistant.components.litejet.*
homeassistant.components.litellm.*
homeassistant.components.litterrobot.*
homeassistant.components.llama_cpp.*
homeassistant.components.local_ip.*
CODEOWNERS
Generated
+2
View File
@@ -1034,6 +1034,8 @@ CLAUDE.md @home-assistant/core
/homeassistant/components/linux_battery/ @fabaff
/homeassistant/components/litejet/ @joncar
/tests/components/litejet/ @joncar
/homeassistant/components/litellm/ @luismalves
/tests/components/litellm/ @luismalves
/homeassistant/components/litterrobot/ @natekspencer @tkdrob
/tests/components/litterrobot/ @natekspencer @tkdrob
/homeassistant/components/livisi/ @StefanIacobLivisi @planbnet
homeassistant/components/litellm/__init__.py
+33
View File
@@ -0,0 +1,33 @@
"""The LiteLLM integration."""
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .coordinator import LiteLLMConfigEntry, LiteLLMDataUpdateCoordinator
PLATFORMS = [Platform.CONVERSATION]
async def async_setup_entry(hass: HomeAssistant, entry: LiteLLMConfigEntry) -> bool:
"""Set up LiteLLM from a config entry."""
coordinator = LiteLLMDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
async def _async_update_listener(
hass: HomeAssistant, entry: LiteLLMConfigEntry
) -> None:
"""Handle update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: LiteLLMConfigEntry) -> bool:
"""Unload LiteLLM."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
homeassistant/components/litellm/config_flow.py
+253
View File
@@ -0,0 +1,253 @@
"""Config flow for LiteLLM integration."""
import logging
from typing import Any, override
from openai import AsyncOpenAI, AuthenticationError, OpenAIError, PermissionDeniedError
import voluptuous as vol
from yarl import URL
from homeassistant.config_entries import (
SOURCE_USER,
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, CONF_MODEL, CONF_URL
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
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 (
CONF_PROMPT,
DOMAIN,
PLACEHOLDER_API_KEY,
RECOMMENDED_CONVERSATION_OPTIONS,
)
_LOGGER = logging.getLogger(__name__)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect to the proxy."""
class InvalidAuth(HomeAssistantError):
"""Error to indicate the API key is invalid."""
def _normalize_url(url: str) -> str:
"""Normalize the proxy URL, ensuring it ends with the OpenAI `/v1` path."""
parsed = URL(url.strip())
path = parsed.path.rstrip("/")
if not path.endswith("/v1"):
path = f"{path}/v1"
return str(parsed.with_path(path))
async def _get_models(hass: HomeAssistant, url: str, api_key: str | None) -> list[str]:
"""Fetch the available model names from the LiteLLM proxy.
Uses the OpenAI-compatible `/v1/models` endpoint, which a LiteLLM proxy
serves with the configured model names.
"""
client = AsyncOpenAI(
base_url=url,
api_key=api_key or PLACEHOLDER_API_KEY,
http_client=get_async_client(hass),
)
try:
return [
model.id async for model in client.with_options(timeout=10.0).models.list()
]
except (AuthenticationError, PermissionDeniedError) as err:
raise InvalidAuth from err
except OpenAIError as err:
raise CannotConnect from err
class LiteLLMConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for LiteLLM."""
VERSION = 1
@classmethod
@callback
@override
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this handler."""
return {"conversation": ConversationFlowHandler}
@override
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
url = _normalize_url(user_input[CONF_URL])
api_key = user_input.get(CONF_API_KEY)
self._async_abort_entries_match({CONF_URL: url})
try:
await _get_models(self.hass, url, api_key)
except InvalidAuth:
errors["base"] = "invalid_auth"
except CannotConnect:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
data = {CONF_URL: url}
if api_key:
data[CONF_API_KEY] = api_key
return self.async_create_entry(
title=URL(url).host or url,
data=data,
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_URL): str,
vol.Optional(CONF_API_KEY): str,
}
),
errors=errors,
)
class LiteLLMSubentryFlowHandler(ConfigSubentryFlow):
"""Handle subentry flow for LiteLLM."""
def __init__(self) -> None:
"""Initialize the subentry flow."""
self.models: list[str] = []
async def _fetch_models(self) -> None:
"""Fetch models from the LiteLLM proxy."""
entry = self._get_entry()
self.models = await _get_models(
self.hass, entry.data[CONF_URL], entry.data.get(CONF_API_KEY)
)
class ConversationFlowHandler(LiteLLMSubentryFlowHandler):
"""Handle conversation subentry flow."""
def __init__(self) -> None:
"""Initialize the subentry flow."""
super().__init__()
self.options: dict[str, Any] = {}
@property
def _is_new(self) -> bool:
"""Return if this is a new subentry."""
return self.source == SOURCE_USER
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_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle reconfiguration of a conversation agent."""
self.options = self._get_reconfigure_subentry().data.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 is not 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)
if self._is_new:
return self.async_create_entry(
title=user_input[CONF_MODEL], data=user_input
)
return self.async_update_and_abort(
self._get_entry(),
self._get_reconfigure_subentry(),
title=user_input[CONF_MODEL],
data=user_input,
)
try:
await self._fetch_models()
except InvalidAuth:
return self.async_abort(reason="invalid_auth")
except CannotConnect:
return self.async_abort(reason="cannot_connect")
except Exception:
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
options = [SelectOptionDict(value=model, label=model) for model in self.models]
hass_apis: list[SelectOptionDict] = [
SelectOptionDict(
label=api.name,
value=api.id,
)
for api in llm.async_get_apis(self.hass)
]
if suggested_llm_apis := self.options.get(CONF_LLM_HASS_API):
valid_api_ids = {api["value"] for api in hass_apis}
self.options[CONF_LLM_HASS_API] = [
api for api in suggested_llm_apis if api in valid_api_ids
]
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
CONF_MODEL, default=self.options.get(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/litellm/const.py
+18
View File
@@ -0,0 +1,18 @@
"""Constants for the LiteLLM integration."""
import logging
from homeassistant.const import CONF_LLM_HASS_API, CONF_PROMPT
from homeassistant.helpers import llm
DOMAIN = "litellm"
LOGGER = logging.getLogger(__package__)
# LiteLLM proxies may run without authentication. The OpenAI client requires a
# non-empty API key, so we send a placeholder when the user did not provide one.
PLACEHOLDER_API_KEY = "sk-no-key-required"
RECOMMENDED_CONVERSATION_OPTIONS = {
CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
CONF_PROMPT: llm.DEFAULT_INSTRUCTIONS_PROMPT,
}
homeassistant/components/litellm/conversation.py
+69
View File
@@ -0,0 +1,69 @@
"""Conversation support for LiteLLM."""
from typing import Literal, override
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 LiteLLMConfigEntry
from .const import DOMAIN
from .entity import LiteLLMEntity
async def async_setup_entry(
hass: HomeAssistant,
config_entry: LiteLLMConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up conversation entities."""
for subentry in config_entry.get_subentries_of_type("conversation"):
async_add_entities(
[LiteLLMConversationEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class LiteLLMConversationEntity(LiteLLMEntity, conversation.ConversationEntity):
"""LiteLLM conversation agent."""
_attr_name = None
def __init__(self, entry: LiteLLMConfigEntry, 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
@override
def supported_languages(self) -> list[str] | Literal["*"]:
"""Return a list of supported languages."""
return MATCH_ALL
@override
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)
homeassistant/components/litellm/coordinator.py
+74
View File
@@ -0,0 +1,74 @@
"""Coordinator for the LiteLLM integration."""
from datetime import timedelta
from typing import override
from openai import AsyncOpenAI, AuthenticationError, OpenAIError, PermissionDeniedError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_URL
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER, PLACEHOLDER_API_KEY
# Ping the proxy hourly while it is reachable, and back off to once a minute
# while it is down so entities recover quickly once it returns.
UPDATE_INTERVAL_CONNECTED = timedelta(hours=1)
UPDATE_INTERVAL_DISCONNECTED = timedelta(minutes=1)
type LiteLLMConfigEntry = ConfigEntry[LiteLLMDataUpdateCoordinator]
class LiteLLMDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""Own the OpenAI client and track LiteLLM proxy availability."""
config_entry: LiteLLMConfigEntry
def __init__(self, hass: HomeAssistant, config_entry: LiteLLMConfigEntry) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
LOGGER,
config_entry=config_entry,
name=config_entry.title,
update_interval=UPDATE_INTERVAL_CONNECTED,
always_update=False,
)
self.client = AsyncOpenAI(
base_url=config_entry.data[CONF_URL],
api_key=config_entry.data.get(CONF_API_KEY) or PLACEHOLDER_API_KEY,
http_client=get_async_client(hass),
)
@override
async def _async_update_data(self) -> None:
"""Ping the proxy to confirm it is reachable and authenticated."""
self.update_interval = UPDATE_INTERVAL_DISCONNECTED
try:
async for _ in self.client.with_options(timeout=10.0).models.list():
break
except (AuthenticationError, PermissionDeniedError) as err:
raise ConfigEntryAuthFailed from err
except OpenAIError as err:
raise UpdateFailed(err) from err
self.update_interval = UPDATE_INTERVAL_CONNECTED
@callback
@override
def async_set_updated_data(self, data: None) -> None:
"""Manually update data and reset to the connected interval."""
self.update_interval = UPDATE_INTERVAL_CONNECTED
super().async_set_updated_data(data)
@callback
def mark_connection_error(self) -> None:
"""Flag the proxy as unreachable and schedule a quick recheck."""
self.update_interval = UPDATE_INTERVAL_DISCONNECTED
if self.last_update_success:
self.last_update_success = False
self.async_update_listeners()
if self._listeners and not self.hass.is_stopping:
self._schedule_refresh()
homeassistant/components/litellm/entity.py
+211
View File
@@ -0,0 +1,211 @@
"""Base entity for LiteLLM."""
from collections.abc import AsyncGenerator, Callable
import json
from typing import Any, Literal
import openai
from openai.types.chat import (
ChatCompletionAssistantMessageParam,
ChatCompletionFunctionToolParam,
ChatCompletionMessage,
ChatCompletionMessageFunctionToolCallParam,
ChatCompletionMessageParam,
ChatCompletionSystemMessageParam,
ChatCompletionToolMessageParam,
ChatCompletionUserMessageParam,
)
from openai.types.chat.chat_completion_message_function_tool_call_param import Function
from openai.types.shared_params import FunctionDefinition
from voluptuous_openapi import convert
from homeassistant.components import conversation
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_MODEL
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, llm
from homeassistant.helpers.json import json_dumps
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, LOGGER
from .coordinator import LiteLLMConfigEntry, LiteLLMDataUpdateCoordinator
MAX_TOOL_ITERATIONS = 10
def _format_tool(
tool: llm.Tool,
custom_serializer: Callable[[Any], Any] | None,
) -> ChatCompletionFunctionToolParam:
"""Format tool specification."""
unsupported_keys = {"oneOf", "anyOf", "allOf"}
schema = convert(tool.parameters, custom_serializer=custom_serializer)
schema = {k: v for k, v in schema.items() if k not in unsupported_keys}
tool_spec = FunctionDefinition(
name=tool.name,
parameters=schema,
)
if tool.description:
tool_spec["description"] = tool.description
return ChatCompletionFunctionToolParam(type="function", function=tool_spec)
def _convert_content_to_chat_message(
content: conversation.Content,
) -> ChatCompletionMessageParam | None:
"""Convert any native chat message for this agent to the native format."""
LOGGER.debug("_convert_content_to_chat_message=%s", content)
if isinstance(content, conversation.ToolResultContent):
return ChatCompletionToolMessageParam(
role="tool",
tool_call_id=content.tool_call_id,
content=json_dumps(content.tool_result),
)
role: Literal["user", "assistant", "system"] = content.role
if role == "system" and content.content:
return ChatCompletionSystemMessageParam(role="system", content=content.content)
if role == "user" and content.content:
return ChatCompletionUserMessageParam(role="user", content=content.content)
if role == "assistant":
param = ChatCompletionAssistantMessageParam(
role="assistant",
content=content.content,
)
if isinstance(content, conversation.AssistantContent) and content.tool_calls:
param["tool_calls"] = [
ChatCompletionMessageFunctionToolCallParam(
type="function",
id=tool_call.id,
function=Function(
arguments=json_dumps(tool_call.tool_args),
name=tool_call.tool_name,
),
)
for tool_call in content.tool_calls
]
return param
LOGGER.warning("Could not convert message to Completions API: %s", content)
return None
def _decode_tool_arguments(arguments: str) -> Any:
"""Decode tool call arguments."""
try:
return json.loads(arguments)
except json.JSONDecodeError as err:
raise HomeAssistantError(f"Unexpected tool argument response: {err}") from err
async def _transform_response(
message: ChatCompletionMessage,
) -> AsyncGenerator[conversation.AssistantContentDeltaDict]:
"""Transform the LiteLLM message to a ChatLog format."""
data: conversation.AssistantContentDeltaDict = {
"role": message.role,
"content": message.content,
}
if message.tool_calls:
data["tool_calls"] = [
llm.ToolInput(
id=tool_call.id,
tool_name=tool_call.function.name,
tool_args=_decode_tool_arguments(tool_call.function.arguments),
)
for tool_call in message.tool_calls
if tool_call.type == "function"
]
yield data
class LiteLLMEntity(CoordinatorEntity[LiteLLMDataUpdateCoordinator]):
"""Base entity for LiteLLM."""
_attr_has_entity_name = True
def __init__(self, entry: LiteLLMConfigEntry, subentry: ConfigSubentry) -> None:
"""Initialize the entity."""
super().__init__(entry.runtime_data)
self.entry = entry
self.subentry = subentry
self.model = subentry.data[CONF_MODEL]
self._attr_unique_id = subentry.subentry_id
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, subentry.subentry_id)},
name=subentry.title,
entry_type=dr.DeviceEntryType.SERVICE,
)
async def _async_handle_chat_log(
self,
chat_log: conversation.ChatLog,
) -> None:
"""Generate an answer for the chat log."""
model_args = {
"model": self.model,
"user": chat_log.conversation_id,
}
tools: list[ChatCompletionFunctionToolParam] | None = None
if chat_log.llm_api:
tools = [
_format_tool(tool, chat_log.llm_api.custom_serializer)
for tool in chat_log.llm_api.tools
]
if tools:
model_args["tools"] = tools
model_args["messages"] = [
m
for content in chat_log.content
if (m := _convert_content_to_chat_message(content))
]
coordinator = self.entry.runtime_data
client = coordinator.client
for _iteration in range(MAX_TOOL_ITERATIONS):
try:
result = await client.chat.completions.create(**model_args)
except (openai.AuthenticationError, openai.PermissionDeniedError) as err:
# Re-check so the proxy is marked unavailable for the auth failure.
await coordinator.async_request_refresh()
LOGGER.error("Error talking to API: %s", err)
raise HomeAssistantError("Error talking to API") from err
except openai.APIConnectionError as err:
coordinator.mark_connection_error()
LOGGER.error("Error talking to API: %s", err)
raise HomeAssistantError("Error talking to API") from err
except openai.OpenAIError as err:
# Reachable but the request failed; keep the entity available.
coordinator.async_set_updated_data(None)
LOGGER.error("Error talking to API: %s", err)
raise HomeAssistantError("Error talking to API") from err
if not result.choices:
LOGGER.error("API returned empty choices")
raise HomeAssistantError("API returned empty response")
result_message = result.choices[0].message
model_args["messages"].extend(
[
msg
async for content in chat_log.async_add_delta_content_stream(
self.entity_id, _transform_response(result_message)
)
if (msg := _convert_content_to_chat_message(content))
]
)
if not chat_log.unresponded_tool_results:
coordinator.async_set_updated_data(None)
break
else:
LOGGER.warning(
"Stopped after %s tool iterations with unresolved tool calls",
MAX_TOOL_ITERATIONS,
)
homeassistant/components/litellm/manifest.json
+13
View File
@@ -0,0 +1,13 @@
{
"domain": "litellm",
"name": "LiteLLM",
"after_dependencies": ["assist_pipeline", "intent"],
"codeowners": ["@luismalves"],
"config_flow": true,
"dependencies": ["conversation"],
"documentation": "https://www.home-assistant.io/integrations/litellm",
"integration_type": "service",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["openai==2.45.0"]
}
homeassistant/components/litellm/quality_scale.yaml
+98
View File
@@ -0,0 +1,98 @@
rules:
# Bronze
action-setup:
status: exempt
comment: No actions are implemented
appropriate-polling:
status: done
comment: >-
the coordinator polls the proxy hourly for an availability check, backing
off to once a minute while it is unreachable
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: No actions are implemented
docs-conditions:
status: exempt
comment: This integration does not have any conditions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-triggers:
status: exempt
comment: This integration does not have any triggers.
entity-event-setup:
status: exempt
comment: the integration does not subscribe to events
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: the integration has no options
docs-installation-parameters: done
entity-unavailable:
status: done
comment: >-
the conversation entity follows the coordinator and is marked unavailable
when the proxy cannot be reached
integration-owner: done
log-when-unavailable: done
parallel-updates: todo
reauthentication-flow: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Service can't be discovered
discovery:
status: exempt
comment: Service can't be discovered
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: devices are created via subentries, not discovered dynamically
entity-category:
status: exempt
comment: the conversation entity does not use entity categories
entity-device-class:
status: exempt
comment: no suitable device class for the conversation entity
entity-disabled-by-default:
status: exempt
comment: only one conversation entity
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: the integration has no repairs
stale-devices:
status: exempt
comment: only one device per entry, is deleted with the entry.
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done
homeassistant/components/litellm/strings.json
+55
View File
@@ -0,0 +1,55 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"url": "[%key:common::config_flow::data::url%]"
},
"data_description": {
"api_key": "An optional LiteLLM API key or virtual key. Leave empty if your proxy does not require authentication.",
"url": "The base URL of your LiteLLM proxy, including the host and port"
}
}
}
},
"config_subentries": {
"conversation": {
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"entry_not_loaded": "The main integration entry is not loaded. Please ensure the integration is loaded before reconfiguring.",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"entry_type": "Conversation agent",
"initiate_flow": {
"reconfigure": "Reconfigure conversation agent",
"user": "Add conversation agent"
},
"step": {
"init": {
"data": {
"llm_hass_api": "[%key:common::config_flow::data::llm_hass_api%]",
"model": "[%key:common::generic::model%]",
"prompt": "[%key:common::config_flow::data::prompt%]"
},
"data_description": {
"llm_hass_api": "Select which tools the model can use to interact with your devices and entities.",
"model": "The model to use for the conversation agent",
"prompt": "Instruct how the LLM should respond. This can be a template."
},
"description": "Configure the conversation agent"
}
}
}
}
}
homeassistant/generated/config_flows.py
Generated
+1
View File
@@ -429,6 +429,7 @@ FLOWS = {
"lifx",
"linkplay",
"litejet",
"litellm",
"litterrobot",
"livisi",
"llama_cpp",
homeassistant/generated/integrations.json
+6
View File
@@ -3911,6 +3911,12 @@
"iot_class": "local_push",
"single_config_entry": true
},
"litellm": {
"name": "LiteLLM",
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling"
},
"litterrobot": {
"name": "Whisker",
"integration_type": "hub",
mypy.ini
Generated
+10
View File
@@ -3257,6 +3257,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.litellm.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.litterrobot.*]
check_untyped_defs = true
disallow_incomplete_defs = true
requirements_all.txt
Generated
+1
View File
@@ -1770,6 +1770,7 @@ open-garage==0.2.0
open-meteo==0.3.2
# homeassistant.components.cloud
# homeassistant.components.litellm
# homeassistant.components.llama_cpp
# homeassistant.components.open_router
# homeassistant.components.openai_conversation
tests/components/litellm/__init__.py
+25
View File
@@ -0,0 +1,25 @@
"""Tests for the LiteLLM integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
def get_subentry_id(mock_config_entry: MockConfigEntry, subentry_type: str) -> str:
"""Get the subentry ID for a given type."""
ids = [
subentry_id
for subentry_id, subentry in mock_config_entry.subentries.items()
if subentry.subentry_type == subentry_type
]
if not ids:
raise ValueError(f"No subentry found for type {subentry_type}")
return ids[0]
tests/components/litellm/conftest.py
+133
View File
@@ -0,0 +1,133 @@
"""Fixtures for LiteLLM integration tests."""
from collections.abc import AsyncGenerator, Generator
from typing import Any
from unittest.mock import AsyncMock, patch
from openai.types import CompletionUsage, Model
from openai.types.chat import ChatCompletion, ChatCompletionMessage
from openai.types.chat.chat_completion import Choice
import pytest
from homeassistant.components.litellm.const import CONF_PROMPT, DOMAIN
from homeassistant.config_entries import ConfigSubentryData
from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, CONF_MODEL, CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers import llm
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
TEST_URL = "http://localhost:4000/v1"
async def models_response(*model_ids: str) -> AsyncGenerator[Model]:
"""Yield models as the OpenAI client's `models.list()` would."""
for model_id in model_ids:
yield Model(id=model_id, created=0, object="model", owned_by="litellm")
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.litellm.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def enable_assist() -> bool:
"""Return whether the Assist LLM API is enabled for the conversation agent."""
return False
@pytest.fixture
def conversation_subentry_data(enable_assist: bool) -> dict[str, Any]:
"""Mock conversation subentry data."""
res: dict[str, Any] = {
CONF_MODEL: "gpt-3.5-turbo",
CONF_PROMPT: "You are a helpful assistant.",
}
if enable_assist:
res[CONF_LLM_HASS_API] = [llm.LLM_API_ASSIST]
return res
@pytest.fixture
def mock_config_entry(
hass: HomeAssistant,
conversation_subentry_data: dict[str, Any],
) -> MockConfigEntry:
"""Mock a config entry."""
return MockConfigEntry(
title="localhost:4000",
domain=DOMAIN,
data={
CONF_URL: TEST_URL,
CONF_API_KEY: "bla",
},
subentries_data=[
ConfigSubentryData(
data=conversation_subentry_data,
subentry_id="ABCDEF",
subentry_type="conversation",
title="gpt-3.5-turbo",
unique_id=None,
),
],
)
@pytest.fixture
async def mock_openai_client() -> AsyncGenerator[AsyncMock]:
"""Mock the OpenAI client used for chat completions."""
with patch(
"homeassistant.components.litellm.coordinator.AsyncOpenAI"
) as mock_client:
client = mock_client.return_value
client.chat.completions.create = AsyncMock(
return_value=ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
Choice(
finish_reason="stop",
index=0,
message=ChatCompletionMessage(
content="Hello, how can I help you?",
role="assistant",
function_call=None,
tool_calls=None,
),
)
],
created=1700000000,
model="gpt-3.5-turbo",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
)
)
yield client
@pytest.fixture
def mock_models() -> Generator[AsyncMock]:
"""Mock the OpenAI client the config flow uses to list proxy models."""
with patch(
"homeassistant.components.litellm.config_flow.AsyncOpenAI"
) as mock_client:
client = mock_client.return_value
client.with_options.return_value.models.list.side_effect = (
lambda *args, **kwargs: models_response("gpt-3.5-turbo", "gpt-4")
)
yield client
@pytest.fixture(autouse=True)
async def setup_ha(hass: HomeAssistant) -> None:
"""Set up Home Assistant."""
assert await async_setup_component(hass, "homeassistant", {})
tests/components/litellm/snapshots/test_conversation.ambr
+295
View File
@@ -0,0 +1,295 @@
# serializer version: 1
# name: test_all_entities[assist][conversation.gpt_3_5_turbo-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'conversation',
'entity_category': None,
'entity_id': 'conversation.gpt_3_5_turbo',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
'conversation': dict({
'should_expose': False,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'litellm',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ConversationEntityFeature: 1>,
'translation_key': None,
'unique_id': 'ABCDEF',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[assist][conversation.gpt_3_5_turbo-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'gpt-3.5-turbo',
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ConversationEntityFeature: 1>,
}),
'context': <ANY>,
'entity_id': 'conversation.gpt_3_5_turbo',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_all_entities[no_assist][conversation.gpt_3_5_turbo-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'conversation',
'entity_category': None,
'entity_id': 'conversation.gpt_3_5_turbo',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
'conversation': dict({
'should_expose': False,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'litellm',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'ABCDEF',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[no_assist][conversation.gpt_3_5_turbo-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'gpt-3.5-turbo',
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <ConversationEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'conversation.gpt_3_5_turbo',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_default_prompt
list([
dict({
'attachments': None,
'content': 'hello',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'content': 'Hello, how can I help you?',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
])
# ---
# name: test_function_call[True]
list([
dict({
'attachments': None,
'content': 'What time is it?',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'content': None,
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': list([
dict({
'external': True,
'id': 'mock_tool_call_id',
'tool_args': dict({
}),
'tool_name': 'HassGetCurrentTime',
}),
]),
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'tool_result',
'tool_call_id': 'mock_tool_call_id',
'tool_name': 'HassGetCurrentTime',
'tool_result': dict({
'data': dict({
'failed': list([
]),
'success': list([
]),
}),
'response_type': 'action_done',
'speech': dict({
'plain': dict({
'extra_data': None,
'speech': '12:00 PM',
}),
}),
'speech_slots': dict({
'time': datetime.time(12, 0),
}),
}),
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'content': '12:00 PM',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
dict({
'attachments': None,
'content': 'Please call the test function',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'content': None,
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': list([
dict({
'external': False,
'id': 'call_call_1',
'tool_args': dict({
'param1': 'call1',
}),
'tool_name': 'test_tool',
}),
]),
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'tool_result',
'tool_call_id': 'call_call_1',
'tool_name': 'test_tool',
'tool_result': 'value1',
}),
dict({
'agent_id': 'conversation.gpt_3_5_turbo',
'content': 'I have successfully called the function',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
])
# ---
# name: test_function_call[True].1
list([
dict({
'content': '''
You are a helpful assistant.
Only if the user wants to control a device, tell them to expose entities to their voice assistant in Home Assistant.
''',
'role': 'system',
}),
dict({
'content': 'What time is it?',
'role': 'user',
}),
dict({
'content': None,
'role': 'assistant',
'tool_calls': list([
dict({
'function': dict({
'arguments': '{}',
'name': 'HassGetCurrentTime',
}),
'id': 'mock_tool_call_id',
'type': 'function',
}),
]),
}),
dict({
'content': '{"speech":{"plain":{"speech":"12:00 PM","extra_data":null}},"response_type":"action_done","speech_slots":{"time":"12:00:00"},"data":{"success":[],"failed":[]}}',
'role': 'tool',
'tool_call_id': 'mock_tool_call_id',
}),
dict({
'content': '12:00 PM',
'role': 'assistant',
}),
dict({
'content': 'Please call the test function',
'role': 'user',
}),
dict({
'content': None,
'role': 'assistant',
'tool_calls': list([
dict({
'function': dict({
'arguments': '{"param1":"call1"}',
'name': 'test_tool',
}),
'id': 'call_call_1',
'type': 'function',
}),
]),
}),
dict({
'content': '"value1"',
'role': 'tool',
'tool_call_id': 'call_call_1',
}),
dict({
'content': 'I have successfully called the function',
'role': 'assistant',
}),
])
# ---
tests/components/litellm/test_config_flow.py
+397
View File
@@ -0,0 +1,397 @@
"""Test the LiteLLM config flow."""
from unittest.mock import AsyncMock, patch
import httpx
from openai import (
APIConnectionError,
APITimeoutError,
AuthenticationError,
PermissionDeniedError,
)
import pytest
from homeassistant.components.litellm.config_flow import CannotConnect, InvalidAuth
from homeassistant.components.litellm.const import CONF_PROMPT, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, CONF_MODEL, CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import get_subentry_id, setup_integration
from .conftest import TEST_URL, models_response
from tests.common import MockConfigEntry
CONVERSATION_MODEL_OPTIONS = [
{"value": "gpt-3.5-turbo", "label": "gpt-3.5-turbo"},
{"value": "gpt-4", "label": "gpt-4"},
]
@pytest.mark.usefixtures("mock_setup_entry", "mock_models")
@pytest.mark.parametrize(
"url_input",
["http://localhost:4000", "http://localhost:4000/", TEST_URL, f"{TEST_URL}/"],
)
async def test_full_flow(hass: HomeAssistant, url_input: str) -> None:
"""Test the full config flow normalizes the URL and stores the key."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_URL: url_input, CONF_API_KEY: "bla"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "localhost"
assert result["data"] == {CONF_URL: TEST_URL, CONF_API_KEY: "bla"}
@pytest.mark.usefixtures("mock_setup_entry", "mock_models")
async def test_full_flow_without_api_key(hass: HomeAssistant) -> None:
"""Test the config flow works without an API key."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_URL: "http://localhost:4000"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_URL: TEST_URL}
@pytest.mark.parametrize(
("exception", "error"),
[
(InvalidAuth, "invalid_auth"),
(CannotConnect, "cannot_connect"),
(Exception, "unknown"),
],
)
@pytest.mark.usefixtures("mock_setup_entry")
async def test_form_errors(
hass: HomeAssistant,
exception: Exception,
error: str,
) -> None:
"""Test we handle errors and can recover."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.litellm.config_flow._get_models",
new_callable=AsyncMock,
) as mock_get_models:
mock_get_models.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_URL: "http://localhost:4000", CONF_API_KEY: "bla"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
mock_get_models.side_effect = None
mock_get_models.return_value = {"gpt-3.5-turbo": {}}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_URL: "http://localhost:4000", CONF_API_KEY: "bla"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
def _status_error(
error: type[AuthenticationError | PermissionDeniedError], status_code: int
) -> AuthenticationError | PermissionDeniedError:
"""Build an OpenAI status error backed by a real httpx response."""
return error(
response=httpx.Response(
status_code=status_code, request=httpx.Request("GET", TEST_URL)
),
body=None,
message="error",
)
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("side_effect", "error"),
[
(_status_error(AuthenticationError, 401), "invalid_auth"),
(_status_error(PermissionDeniedError, 403), "invalid_auth"),
(APIConnectionError(request=httpx.Request("GET", TEST_URL)), "cannot_connect"),
(APITimeoutError(request=httpx.Request("GET", TEST_URL)), "cannot_connect"),
],
)
async def test_user_step_proxy_errors(
hass: HomeAssistant,
side_effect: Exception,
error: str,
) -> None:
"""Test the user step surfaces errors raised by the OpenAI client."""
with patch(
"homeassistant.components.litellm.config_flow.AsyncOpenAI"
) as mock_client:
mock_client.return_value.with_options.return_value.models.list.side_effect = (
side_effect
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_URL: "http://localhost:4000", CONF_API_KEY: "bla"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
@pytest.mark.usefixtures("mock_setup_entry")
async def test_duplicate_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test aborting the flow if an entry with the same URL already exists."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_URL: "http://localhost:4000", CONF_API_KEY: "other"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("mock_models")
async def test_create_conversation_agent(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a conversation agent."""
await setup_integration(hass, mock_config_entry)
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert (
result["data_schema"].schema["model"].config["options"]
== CONVERSATION_MODEL_OPTIONS
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_MODEL: "gpt-3.5-turbo",
CONF_PROMPT: "you are an assistant",
CONF_LLM_HASS_API: ["assist"],
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "gpt-3.5-turbo"
assert result["data"] == {
CONF_MODEL: "gpt-3.5-turbo",
CONF_PROMPT: "you are an assistant",
CONF_LLM_HASS_API: ["assist"],
}
@pytest.mark.usefixtures("mock_models")
async def test_create_conversation_agent_no_control(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a conversation agent without control over the LLM API."""
await setup_integration(hass, mock_config_entry)
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": SOURCE_USER},
)
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_MODEL: "gpt-3.5-turbo",
CONF_PROMPT: "you are an assistant",
CONF_LLM_HASS_API: [],
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_MODEL: "gpt-3.5-turbo",
CONF_PROMPT: "you are an assistant",
}
async def test_conversation_agent_model_options(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the model dropdown is populated from the proxy's model list."""
await setup_integration(hass, mock_config_entry)
with patch(
"homeassistant.components.litellm.config_flow.AsyncOpenAI"
) as mock_client:
mock_client.return_value.with_options.return_value.models.list.side_effect = (
lambda *args, **kwargs: models_response("gpt-4o", "gpt-5")
)
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["data_schema"].schema["model"].config["options"] == [
{"value": "gpt-4o", "label": "gpt-4o"},
{"value": "gpt-5", "label": "gpt-5"},
]
@pytest.mark.parametrize(
("exception", "reason"),
[
(InvalidAuth, "invalid_auth"),
(CannotConnect, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_subentry_exceptions(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
reason: str,
) -> None:
"""Test subentry flow aborts on errors fetching models."""
await setup_integration(hass, mock_config_entry)
with patch(
"homeassistant.components.litellm.config_flow._get_models",
new_callable=AsyncMock,
side_effect=exception,
):
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
@pytest.mark.usefixtures("mock_models")
async def test_reconfigure_conversation_agent(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring a conversation agent."""
await setup_integration(hass, mock_config_entry)
subentry_id = get_subentry_id(mock_config_entry, "conversation")
result = await mock_config_entry.start_subentry_reconfigure_flow(hass, subentry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_MODEL: "gpt-4",
CONF_PROMPT: "updated prompt",
CONF_LLM_HASS_API: ["assist"],
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
subentry = mock_config_entry.subentries[subentry_id]
assert subentry.title == "gpt-4"
assert subentry.data[CONF_MODEL] == "gpt-4"
assert subentry.data[CONF_PROMPT] == "updated prompt"
assert subentry.data[CONF_LLM_HASS_API] == ["assist"]
async def test_reconfigure_entry_not_loaded(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring aborts when the main entry is not loaded."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
@pytest.mark.parametrize(
("current_llm_apis", "suggested_llm_apis", "expected_options"),
[
(["assist"], ["assist"], ["assist"]),
(["non-existent"], [], ["assist"]),
(["assist", "non-existent"], ["assist"], ["assist"]),
],
)
@pytest.mark.usefixtures("mock_models")
async def test_reconfigure_conversation_subentry_llm_api_schema(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
current_llm_apis: list[str],
suggested_llm_apis: list[str],
expected_options: list[str],
) -> None:
"""Test llm_hass_api field values when reconfiguring a conversation subentry."""
await setup_integration(hass, mock_config_entry)
subentry_id = get_subentry_id(mock_config_entry, "conversation")
subentry = mock_config_entry.subentries[subentry_id]
hass.config_entries.async_update_subentry(
mock_config_entry,
subentry,
data={**subentry.data, CONF_LLM_HASS_API: current_llm_apis},
)
await hass.async_block_till_done()
result = await mock_config_entry.start_subentry_reconfigure_flow(hass, subentry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
schema = result["data_schema"].schema
key = next(k for k in schema if k == CONF_LLM_HASS_API)
assert key.default() == suggested_llm_apis
field_schema = schema[key]
assert field_schema.config
assert [
opt["value"] for opt in field_schema.config.get("options")
] == expected_options
tests/components/litellm/test_conversation.py
+297
View File
@@ -0,0 +1,297 @@
"""Tests for the LiteLLM conversation entity."""
import datetime
from unittest.mock import AsyncMock, patch
from freezegun import freeze_time
import httpx
import openai
from openai.types import CompletionUsage
from openai.types.chat import (
ChatCompletion,
ChatCompletionMessage,
ChatCompletionMessageFunctionToolCall,
)
from openai.types.chat.chat_completion import Choice
from openai.types.chat.chat_completion_message_function_tool_call_param import Function
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import conversation
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import Context, HomeAssistant
from homeassistant.helpers import entity_registry as er, intent
from homeassistant.helpers.llm import ToolInput
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
from tests.components.conversation import MockChatLog, mock_chat_log # noqa: F401
AGENT_ID = "conversation.gpt_3_5_turbo"
@pytest.fixture(autouse=True)
def freeze_the_time():
"""Freeze the time."""
with freeze_time("2024-05-24 12:00:00", tz_offset=0):
yield
@pytest.mark.parametrize("enable_assist", [True, False], ids=["assist", "no_assist"])
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test all entities."""
with patch(
"homeassistant.components.litellm.PLATFORMS",
[Platform.CONVERSATION],
):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_default_prompt(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
mock_openai_client: AsyncMock,
mock_chat_log: MockChatLog, # noqa: F811
) -> None:
"""Test that the default prompt works."""
await setup_integration(hass, mock_config_entry)
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id=AGENT_ID,
)
assert result.response.response_type is intent.IntentResponseType.ACTION_DONE
assert mock_chat_log.content[1:] == snapshot
call = mock_openai_client.chat.completions.create.call_args_list[0][1]
assert call["model"] == "gpt-3.5-turbo"
assert "extra_headers" not in call
async def test_empty_api_response(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_openai_client: AsyncMock,
mock_chat_log: MockChatLog, # noqa: F811
) -> None:
"""Test that an empty choices response raises an error."""
await setup_integration(hass, mock_config_entry)
mock_openai_client.chat.completions.create = AsyncMock(
return_value=ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[],
created=1700000000,
model="gpt-3.5-turbo",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(completion_tokens=0, prompt_tokens=8, total_tokens=8),
)
)
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id=AGENT_ID,
)
assert result.response.response_type is intent.IntentResponseType.ERROR
async def test_api_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_openai_client: AsyncMock,
mock_chat_log: MockChatLog, # noqa: F811
) -> None:
"""Test that an error talking to the API is handled gracefully."""
await setup_integration(hass, mock_config_entry)
mock_openai_client.chat.completions.create = AsyncMock(
side_effect=openai.OpenAIError("boom")
)
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id=AGENT_ID,
)
assert result.response.response_type is intent.IntentResponseType.ERROR
async def test_connection_error_availability(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_openai_client: AsyncMock,
mock_chat_log: MockChatLog, # noqa: F811
) -> None:
"""Test a connection error marks the entity unavailable until it recovers."""
await setup_integration(hass, mock_config_entry)
assert hass.states.get(AGENT_ID).state != STATE_UNAVAILABLE
mock_openai_client.chat.completions.create = AsyncMock(
side_effect=openai.APIConnectionError(
request=httpx.Request("POST", "http://localhost")
)
)
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id=AGENT_ID,
)
assert result.response.response_type is intent.IntentResponseType.ERROR
await hass.async_block_till_done()
assert hass.states.get(AGENT_ID).state == STATE_UNAVAILABLE
# A successful availability ping restores the entity.
await mock_config_entry.runtime_data.async_request_refresh()
await hass.async_block_till_done()
assert hass.states.get(AGENT_ID).state != STATE_UNAVAILABLE
@pytest.mark.parametrize("enable_assist", [True])
async def test_function_call(
hass: HomeAssistant,
mock_chat_log: MockChatLog, # noqa: F811
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
mock_openai_client: AsyncMock,
) -> None:
"""Test function call from the assistant."""
await setup_integration(hass, mock_config_entry)
mock_chat_log.async_add_user_content(
conversation.UserContent(content="What time is it?")
)
mock_chat_log.async_add_assistant_content_without_tools(
conversation.AssistantContent(
agent_id=AGENT_ID,
tool_calls=[
ToolInput(
tool_name="HassGetCurrentTime",
tool_args={},
id="mock_tool_call_id",
external=True,
)
],
)
)
mock_chat_log.async_add_assistant_content_without_tools(
conversation.ToolResultContent(
agent_id=AGENT_ID,
tool_call_id="mock_tool_call_id",
tool_name="HassGetCurrentTime",
tool_result={
"speech": {"plain": {"speech": "12:00 PM", "extra_data": None}},
"response_type": "action_done",
"speech_slots": {"time": datetime.time(12, 0)},
"data": {"success": [], "failed": []},
},
)
)
mock_chat_log.async_add_assistant_content_without_tools(
conversation.AssistantContent(
agent_id=AGENT_ID,
content="12:00 PM",
)
)
mock_chat_log.mock_tool_results(
{
"call_call_1": "value1",
"call_call_2": "value2",
}
)
mock_openai_client.chat.completions.create.side_effect = (
ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
Choice(
finish_reason="tool_calls",
index=0,
message=ChatCompletionMessage(
content=None,
role="assistant",
function_call=None,
tool_calls=[
ChatCompletionMessageFunctionToolCall(
id="call_call_1",
function=Function(
arguments='{"param1":"call1"}',
name="test_tool",
),
type="function",
)
],
),
)
],
created=1700000000,
model="gpt-4",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
),
ChatCompletion(
id="chatcmpl-1234567890ZYXWVUTSRQPONMLKJIH",
choices=[
Choice(
finish_reason="stop",
index=0,
message=ChatCompletionMessage(
content="I have successfully called the function",
role="assistant",
function_call=None,
tool_calls=None,
),
)
],
created=1700000000,
model="gpt-4",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
),
)
result = await conversation.async_converse(
hass,
"Please call the test function",
mock_chat_log.conversation_id,
Context(),
agent_id=AGENT_ID,
)
assert result.response.response_type is intent.IntentResponseType.ACTION_DONE
# Don't test the prompt, as it's not deterministic
assert mock_chat_log.content[1:] == snapshot
assert mock_openai_client.chat.completions.create.call_count == 2
assert (
mock_openai_client.chat.completions.create.call_args.kwargs["messages"]
== snapshot
)
tests/components/litellm/test_init.py
+61
View File
@@ -0,0 +1,61 @@
"""Tests for the LiteLLM integration setup."""
from unittest.mock import AsyncMock
import httpx
from openai import APIConnectionError, AuthenticationError
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_load_unload_entry(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test loading and unloading the integration."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("side_effect", "expected_state"),
[
(
AuthenticationError(
response=httpx.Response(
status_code=401, request=httpx.Request("GET", "http://localhost")
),
body=None,
message="invalid api key",
),
ConfigEntryState.SETUP_ERROR,
),
(APIConnectionError(request=None), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_error(
hass: HomeAssistant,
mock_openai_client: AsyncMock,
mock_config_entry: MockConfigEntry,
side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test that setup handles errors validating the connection."""
mock_openai_client.with_options.return_value.models.list.side_effect = side_effect
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is expected_state