pc-heini/ha-openai-compatible
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: vendor openai_conversation from core 2026.9.2, rename domain

Browse Source 
Verbatim copy from home-assistant/core at tag 2026.9.2, with only the
domain constant and manifest changed. Apache-2.0, attribution in NOTICE.

Also fixed self-references the domain rename left dangling: the
config_entry selector's `integration:` field in services.yaml, the
`[%key:component::openai_conversation::...%]` translation
cross-references in strings.json, the deprecated-action descriptions
naming the old domain, and config_flow.py's instructions_url (now
pointing at this repo instead of the core docs page).

Local pytest run skipped per owner decision (see prior commit); files
validated with py_compile / json.load / yaml.safe_load instead. CI
(added in the prior commit) is the real gate.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01739Seac5eMAfNwFeiwV5r4
This commit is contained in:
Adam SmithandClaude Sonnet 5 committed 2026-09-17 17:23:37 -04:00
1 parent 178d948b1b
commit 9d8c36f793
15 files changed
+3498
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

custom_components/openai_compatible/UPSTREAM_COMMIT
+1
View File
@@ -0,0 +1 @@
33c3e0cca60e73a8c4970ee677d75b8bc6464cdf
custom_components/openai_compatible/__init__.py
+532
View File
@@ -0,0 +1,532 @@
"""The OpenAI Conversation integration."""
from pathlib import Path
from types import MappingProxyType
import openai
from openai.types.images_response import ImagesResponse
from openai.types.responses import (
EasyInputMessageParam,
Response,
ResponseInputMessageContentListParam,
ResponseInputParam,
ResponseInputTextParam,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigSubentry
from homeassistant.const import CONF_API_KEY, CONF_PROMPT, Platform
from homeassistant.core import (
HomeAssistant,
ServiceCall,
ServiceResponse,
SupportsResponse,
)
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
ServiceValidationError,
)
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
entity_registry as er,
issue_registry as ir,
selector,
)
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.typing import UNDEFINED, ConfigType, UndefinedType
from .const import (
CONF_CHAT_MODEL,
CONF_FILENAMES,
CONF_MAX_TOKENS,
CONF_REASONING_EFFORT,
CONF_REASONING_SUMMARY,
CONF_STORE_RESPONSES,
CONF_TEMPERATURE,
CONF_TOP_P,
DEFAULT_AI_TASK_NAME,
DEFAULT_NAME,
DEFAULT_STT_NAME,
DEFAULT_TTS_NAME,
DOMAIN,
LOGGER,
RECOMMENDED_AI_TASK_OPTIONS,
RECOMMENDED_CHAT_MODEL,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_REASONING_EFFORT,
RECOMMENDED_REASONING_SUMMARY,
RECOMMENDED_STORE_RESPONSES,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
RECOMMENDED_TTS_OPTIONS,
)
from .entity import async_prepare_files_for_prompt
SERVICE_GENERATE_IMAGE = "generate_image"
SERVICE_GENERATE_CONTENT = "generate_content"
PLATFORMS = (Platform.AI_TASK, Platform.CONVERSATION, Platform.STT, Platform.TTS)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
type OpenAIConfigEntry = ConfigEntry[openai.AsyncClient]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up OpenAI Conversation."""
await async_migrate_integration(hass)
async def render_image(call: ServiceCall) -> ServiceResponse:
"""Render an image with dall-e."""
LOGGER.warning(
"Action '%s.%s' is deprecated and will be removed in the 2026.9.0 release. "
"Please use the 'ai_task.generate_image' action instead",
DOMAIN,
SERVICE_GENERATE_IMAGE,
)
ir.async_create_issue(
hass,
DOMAIN,
"deprecated_generate_image",
breaks_in_ha_version="2026.9.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_generate_image",
)
entry_id = call.data["config_entry"]
entry = hass.config_entries.async_get_entry(entry_id)
if entry is None or entry.domain != DOMAIN:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_config_entry",
translation_placeholders={"config_entry": entry_id},
)
client: openai.AsyncClient = entry.runtime_data
try:
response: ImagesResponse = await client.images.generate(
model="dall-e-3",
prompt=call.data[CONF_PROMPT],
size=call.data["size"],
quality=call.data["quality"],
style=call.data["style"],
response_format="url",
n=1,
)
except openai.AuthenticationError as err:
entry.async_start_reauth(hass)
raise HomeAssistantError("Authentication error") from err
except openai.OpenAIError as err:
raise HomeAssistantError(f"Error generating image: {err}") from err
if not response.data or not response.data[0].url:
raise HomeAssistantError("No image returned")
return response.data[0].model_dump(exclude={"b64_json"})
async def send_prompt(call: ServiceCall) -> ServiceResponse:
"""Send a prompt to ChatGPT and return the response."""
LOGGER.warning(
"Action '%s.%s' is deprecated and will be removed in the 2026.9.0 release. "
"Please use the 'ai_task.generate_data' action instead",
DOMAIN,
SERVICE_GENERATE_CONTENT,
)
ir.async_create_issue(
hass,
DOMAIN,
"deprecated_generate_content",
breaks_in_ha_version="2026.9.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecated_generate_content",
)
entry_id = call.data["config_entry"]
entry = hass.config_entries.async_get_entry(entry_id)
if entry is None or entry.domain != DOMAIN:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="invalid_config_entry",
translation_placeholders={"config_entry": entry_id},
)
# Get first conversation subentry for options
conversation_subentry = next(
(
sub
for sub in entry.subentries.values()
if sub.subentry_type == "conversation"
),
None,
)
if not conversation_subentry:
raise ServiceValidationError("No conversation configuration found")
model: str = conversation_subentry.data.get(
CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL
)
client: openai.AsyncClient = entry.runtime_data
content: ResponseInputMessageContentListParam = [
ResponseInputTextParam(type="input_text", text=call.data[CONF_PROMPT])
]
if filenames := call.data.get(CONF_FILENAMES):
for filename in filenames:
if not hass.config.is_allowed_path(filename):
raise HomeAssistantError(
f"Cannot read `{filename}`, no access to path; "
"`allowlist_external_dirs` may need to be adjusted in "
"`configuration.yaml`"
)
content.extend(
await async_prepare_files_for_prompt(
hass, [(Path(filename), None) for filename in filenames]
)
)
messages: ResponseInputParam = [
EasyInputMessageParam(type="message", role="user", content=content)
]
model_args = {
"model": model,
"input": messages,
"max_output_tokens": conversation_subentry.data.get(
CONF_MAX_TOKENS, RECOMMENDED_MAX_TOKENS
),
"top_p": conversation_subentry.data.get(CONF_TOP_P, RECOMMENDED_TOP_P),
"temperature": conversation_subentry.data.get(
CONF_TEMPERATURE, RECOMMENDED_TEMPERATURE
),
"user": call.context.user_id,
"store": conversation_subentry.data.get(
CONF_STORE_RESPONSES, RECOMMENDED_STORE_RESPONSES
),
}
if model.startswith("o"):
model_args["reasoning"] = {
"effort": conversation_subentry.data.get(
CONF_REASONING_EFFORT, RECOMMENDED_REASONING_EFFORT
)
}
try:
response: Response = await client.responses.create(**model_args)
except openai.AuthenticationError as err:
entry.async_start_reauth(hass)
raise HomeAssistantError("Authentication error") from err
except openai.OpenAIError as err:
raise HomeAssistantError(f"Error generating content: {err}") from err
except FileNotFoundError as err:
raise HomeAssistantError(f"Error generating content: {err}") from err
return {"text": response.output_text}
hass.services.async_register(
DOMAIN,
SERVICE_GENERATE_CONTENT,
send_prompt,
schema=vol.Schema(
{
vol.Required("config_entry"): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Required(CONF_PROMPT): cv.string,
vol.Optional(CONF_FILENAMES, default=[]): vol.All(
cv.ensure_list, [cv.string]
),
}
),
supports_response=SupportsResponse.ONLY,
)
hass.services.async_register(
DOMAIN,
SERVICE_GENERATE_IMAGE,
render_image,
schema=vol.Schema(
{
vol.Required("config_entry"): selector.ConfigEntrySelector(
{
"integration": DOMAIN,
}
),
vol.Required(CONF_PROMPT): cv.string,
vol.Optional("size", default="1024x1024"): vol.In(
("1024x1024", "1024x1792", "1792x1024")
),
vol.Optional("quality", default="standard"): vol.In(("standard", "hd")),
vol.Optional("style", default="vivid"): vol.In(("vivid", "natural")),
}
),
supports_response=SupportsResponse.ONLY,
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> bool:
"""Set up OpenAI Conversation from a config entry."""
client = openai.AsyncOpenAI(
api_key=entry.data[CONF_API_KEY],
http_client=get_async_client(hass),
)
# Cache current platform data which gets added to each request
# (caching done by library)
_ = await hass.async_add_executor_job(client.platform_headers)
try:
await hass.async_add_executor_job(client.with_options(timeout=10.0).models.list)
except openai.AuthenticationError as err:
raise ConfigEntryAuthFailed(err) from err
except openai.OpenAIError as err:
raise ConfigEntryNotReady(err) from err
entry.runtime_data = client
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(async_update_options))
return True
async def async_unload_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> bool:
"""Unload OpenAI."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_update_options(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None:
"""Update options."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_migrate_integration(hass: HomeAssistant) -> None:
"""Migrate integration entry structure."""
# Make sure we get enabled config entries first
entries = sorted(
hass.config_entries.async_entries(DOMAIN),
key=lambda e: e.disabled_by is not None,
)
if not any(entry.version == 1 for entry in entries):
return
api_keys_entries: dict[str, tuple[OpenAIConfigEntry, bool]] = {}
entity_registry = er.async_get(hass)
device_registry = dr.async_get(hass)
for entry in entries:
use_existing = False
subentry = ConfigSubentry(
data=entry.options,
subentry_type="conversation",
title=entry.title,
unique_id=None,
)
if entry.data[CONF_API_KEY] not in api_keys_entries:
use_existing = True
all_disabled = all(
e.disabled_by is not None
for e in entries
if e.data[CONF_API_KEY] == entry.data[CONF_API_KEY]
)
api_keys_entries[entry.data[CONF_API_KEY]] = (entry, all_disabled)
parent_entry, all_disabled = api_keys_entries[entry.data[CONF_API_KEY]]
hass.config_entries.async_add_subentry(parent_entry, subentry)
conversation_entity_id = entity_registry.async_get_entity_id(
"conversation",
DOMAIN,
entry.entry_id,
)
device = device_registry.async_get_device_by_identifier(
(DOMAIN, entry.entry_id), entry.entry_id
)
if conversation_entity_id is not None:
conversation_entity_entry = entity_registry.entities[conversation_entity_id]
entity_disabled_by = conversation_entity_entry.disabled_by
if (
entity_disabled_by is er.RegistryEntryDisabler.CONFIG_ENTRY
and not all_disabled
):
# Device and entity registries will set the disabled_by flag to None
# when moving a device or entity disabled by CONFIG_ENTRY to an enabled
# config entry, but we want to set it to DEVICE or USER instead,
entity_disabled_by = (
er.RegistryEntryDisabler.DEVICE
if device
else er.RegistryEntryDisabler.USER
)
entity_registry.async_update_entity(
conversation_entity_id,
config_entry_id=parent_entry.entry_id,
config_subentry_id=subentry.subentry_id,
disabled_by=entity_disabled_by,
new_unique_id=subentry.subentry_id,
)
if device is not None:
# Device and entity registries will set the disabled_by flag to None
# when moving a device or entity disabled by CONFIG_ENTRY to an enabled
# config entry, but we want to set it to USER instead,
device_disabled_by: dr.DeviceEntryDisabler | UndefinedType = UNDEFINED
if (
device.disabled_by is dr.DeviceEntryDisabler.CONFIG_ENTRY
and not all_disabled
):
device_disabled_by = dr.DeviceEntryDisabler.USER
device_registry.async_update_device(
device.id,
disabled_by=device_disabled_by,
new_identifiers={(DOMAIN, subentry.subentry_id)},
new_config_entry_id=parent_entry.entry_id,
new_config_subentry_id=subentry.subentry_id,
)
if not use_existing:
await hass.config_entries.async_remove(entry.entry_id)
else:
_add_ai_task_subentry(hass, entry)
hass.config_entries.async_update_entry(
entry,
title=DEFAULT_NAME,
options={},
version=2,
minor_version=4,
)
async def async_migrate_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> bool:
"""Migrate entry."""
LOGGER.debug("Migrating from version %s:%s", entry.version, entry.minor_version)
if entry.version == 2 and entry.minor_version == 1:
# Devices left in both the config entry and its subentry by Home Assistant Core
# 2025.7.0b0-2025.7.0b1 are collapsed onto the subentry by the device registry
# migration, so there's nothing to correct here.
hass.config_entries.async_update_entry(entry, minor_version=2)
if entry.version == 2 and entry.minor_version == 2:
_add_ai_task_subentry(hass, entry)
hass.config_entries.async_update_entry(entry, minor_version=3)
if entry.version == 2 and entry.minor_version == 3:
# Fix migration where the disabled_by flag was not set correctly.
# We can currently only correct this for enabled config entries,
# because migration does not run for disabled config entries. This
# is asserted in tests, and if that behavior is changed, we should
# correct also disabled config entries.
device_registry = dr.async_get(hass)
entity_registry = er.async_get(hass)
devices = dr.async_entries_for_config_entry(device_registry, entry.entry_id)
entity_entries = er.async_entries_for_config_entry(
entity_registry, entry.entry_id
)
if entry.disabled_by is None:
# If the config entry is not disabled, we need to set the disabled_by
# flag on devices to USER, and on entities to DEVICE, if they are set
# to CONFIG_ENTRY.
for device in devices:
if device.disabled_by is not dr.DeviceEntryDisabler.CONFIG_ENTRY:
continue
device_registry.async_update_device(
device.id,
disabled_by=dr.DeviceEntryDisabler.USER,
)
for entity in entity_entries:
if entity.disabled_by is not er.RegistryEntryDisabler.CONFIG_ENTRY:
continue
entity_registry.async_update_entity(
entity.entity_id,
disabled_by=er.RegistryEntryDisabler.DEVICE,
)
hass.config_entries.async_update_entry(entry, minor_version=4)
if entry.version == 2 and entry.minor_version == 4:
_add_tts_subentry(hass, entry)
hass.config_entries.async_update_entry(entry, minor_version=5)
if entry.version == 2 and entry.minor_version == 5:
_add_stt_subentry(hass, entry)
hass.config_entries.async_update_entry(entry, minor_version=6)
if entry.version == 2 and entry.minor_version == 6:
for subentry in entry.subentries.values():
if subentry.subentry_type in ("conversation", "ai_task_data"):
data = dict(subentry.data)
updated = False
if data.get(CONF_REASONING_SUMMARY) == "short":
data[CONF_REASONING_SUMMARY] = "concise"
updated = True
if data.get(CONF_REASONING_SUMMARY) == "concise" and not data.get(
CONF_CHAT_MODEL, ""
).startswith("gpt-5"):
data[CONF_REASONING_SUMMARY] = RECOMMENDED_REASONING_SUMMARY
updated = True
if updated:
hass.config_entries.async_update_subentry(
entry, subentry, data=data
)
hass.config_entries.async_update_entry(entry, minor_version=7)
LOGGER.debug(
"Migration to version %s:%s successful", entry.version, entry.minor_version
)
return True
def _add_ai_task_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None:
"""Add AI Task subentry to the config entry."""
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType(RECOMMENDED_AI_TASK_OPTIONS),
subentry_type="ai_task_data",
title=DEFAULT_AI_TASK_NAME,
unique_id=None,
),
)
def _add_stt_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None:
"""Add STT subentry to the config entry."""
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType(RECOMMENDED_STT_OPTIONS),
subentry_type="stt",
title=DEFAULT_STT_NAME,
unique_id=None,
),
)
def _add_tts_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None:
"""Add TTS subentry to the config entry."""
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType(RECOMMENDED_TTS_OPTIONS),
subentry_type="tts",
title=DEFAULT_TTS_NAME,
unique_id=None,
),
)
custom_components/openai_compatible/ai_task.py
+158
View File
@@ -0,0 +1,158 @@
"""AI Task integration for OpenAI."""
import base64
from json import JSONDecodeError
import logging
from typing import TYPE_CHECKING, override
from openai.types.responses.response_output_item import ImageGenerationCall
from homeassistant.components import ai_task, conversation
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.json import json_loads
from .const import (
CONF_CHAT_MODEL,
CONF_IMAGE_MODEL,
RECOMMENDED_CHAT_MODEL,
RECOMMENDED_IMAGE_MODEL,
UNSUPPORTED_IMAGE_MODELS,
)
from .entity import OpenAIBaseLLMEntity
if TYPE_CHECKING:
from homeassistant.config_entries import ConfigSubentry
from . import OpenAIConfigEntry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: OpenAIConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up AI Task entities."""
for subentry in config_entry.subentries.values():
if subentry.subentry_type != "ai_task_data":
continue
async_add_entities(
[OpenAITaskEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class OpenAITaskEntity(
ai_task.AITaskEntity,
OpenAIBaseLLMEntity,
):
"""OpenAI AI Task entity."""
def __init__(self, entry: OpenAIConfigEntry, subentry: ConfigSubentry) -> None:
"""Initialize the entity."""
super().__init__(entry, subentry)
self._attr_supported_features = (
ai_task.AITaskEntityFeature.GENERATE_DATA
| ai_task.AITaskEntityFeature.SUPPORT_ATTACHMENTS
)
model = self.subentry.data.get(CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL)
if not model.startswith(tuple(UNSUPPORTED_IMAGE_MODELS)):
self._attr_supported_features |= ai_task.AITaskEntityFeature.GENERATE_IMAGE
@override
async def _async_generate_data(
self,
task: ai_task.GenDataTask,
chat_log: conversation.ChatLog,
) -> ai_task.GenDataTaskResult:
"""Handle a generate data task."""
await self._async_handle_chat_log(
chat_log, task.name, task.structure, max_iterations=1000
)
if not isinstance(chat_log.content[-1], conversation.AssistantContent):
raise HomeAssistantError(
"Last content in chat log is not an AssistantContent"
)
text = chat_log.content[-1].content or ""
if not task.structure:
return ai_task.GenDataTaskResult(
conversation_id=chat_log.conversation_id,
data=text,
)
try:
data = json_loads(text)
except JSONDecodeError as err:
_LOGGER.error(
"Failed to parse JSON response: %s. Response: %s",
err,
text,
)
raise HomeAssistantError("Error with OpenAI structured response") from err
return ai_task.GenDataTaskResult(
conversation_id=chat_log.conversation_id,
data=data,
)
@override
async def _async_generate_image(
self,
task: ai_task.GenImageTask,
chat_log: conversation.ChatLog,
) -> ai_task.GenImageTaskResult:
"""Handle a generate image task."""
await self._async_handle_chat_log(chat_log, task.name, force_image=True)
if not isinstance(chat_log.content[-1], conversation.AssistantContent):
raise HomeAssistantError(
"Last content in chat log is not an AssistantContent"
)
image_call: ImageGenerationCall | None = None
for content in reversed(chat_log.content):
if not isinstance(content, conversation.AssistantContent):
break
if isinstance(content.native, ImageGenerationCall):
if image_call is None or image_call.result is None:
image_call = content.native
else: # Remove image data from chat log to save memory
content.native.result = None
if image_call is None or image_call.result is None:
raise HomeAssistantError("No image returned")
image_data = base64.b64decode(image_call.result)
image_call.result = None
if hasattr(image_call, "output_format") and (
output_format := image_call.output_format
):
mime_type = f"image/{output_format}"
else:
mime_type = "image/png"
if hasattr(image_call, "size") and (size := image_call.size):
width, height = tuple(size.split("x"))
else:
width, height = None, None
return ai_task.GenImageTaskResult(
image_data=image_data,
conversation_id=chat_log.conversation_id,
mime_type=mime_type,
width=int(width) if width else None,
height=int(height) if height else None,
model=self.subentry.data.get(CONF_IMAGE_MODEL, RECOMMENDED_IMAGE_MODEL),
revised_prompt=image_call.revised_prompt
if hasattr(image_call, "revised_prompt")
else None,
)
custom_components/openai_compatible/config_flow.py
+860
View File
@@ -0,0 +1,860 @@
"""Config flow for OpenAI Conversation integration."""
from collections.abc import Mapping
import json
import logging
from typing import Any, override
import openai
from probatio import to_openapi
import voluptuous as vol
from homeassistant.components.zone import ENTITY_ID_HOME
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import (
CONF_API_KEY,
CONF_LLM_HASS_API,
CONF_NAME,
CONF_PROMPT,
EntityStateAttribute,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import llm
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.selector import (
NumberSelector,
NumberSelectorConfig,
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TemplateSelector,
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from homeassistant.helpers.typing import VolDictType
from .const import (
CONF_CHAT_MODEL,
CONF_CODE_INTERPRETER,
CONF_IMAGE_MODEL,
CONF_MAX_TOKENS,
CONF_PRO_MODE,
CONF_REASONING_EFFORT,
CONF_REASONING_SUMMARY,
CONF_RECOMMENDED,
CONF_SERVICE_TIER,
CONF_STORE_RESPONSES,
CONF_TEMPERATURE,
CONF_TOP_P,
CONF_TTS_SPEED,
CONF_VERBOSITY,
CONF_WEB_SEARCH,
CONF_WEB_SEARCH_CITY,
CONF_WEB_SEARCH_CONTEXT_SIZE,
CONF_WEB_SEARCH_COUNTRY,
CONF_WEB_SEARCH_INLINE_CITATIONS,
CONF_WEB_SEARCH_REGION,
CONF_WEB_SEARCH_TIMEZONE,
CONF_WEB_SEARCH_USER_LOCATION,
DEFAULT_AI_TASK_NAME,
DEFAULT_CONVERSATION_NAME,
DEFAULT_STT_NAME,
DEFAULT_STT_PROMPT,
DEFAULT_TTS_NAME,
DOMAIN,
RECOMMENDED_AI_TASK_OPTIONS,
RECOMMENDED_CHAT_MODEL,
RECOMMENDED_CODE_INTERPRETER,
RECOMMENDED_CONVERSATION_OPTIONS,
RECOMMENDED_IMAGE_MODEL,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_PRO_MODE,
RECOMMENDED_REASONING_EFFORT,
RECOMMENDED_REASONING_SUMMARY,
RECOMMENDED_SERVICE_TIER,
RECOMMENDED_STORE_RESPONSES,
RECOMMENDED_STT_MODEL,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
RECOMMENDED_TTS_OPTIONS,
RECOMMENDED_TTS_SPEED,
RECOMMENDED_VERBOSITY,
RECOMMENDED_WEB_SEARCH,
RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE,
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
RECOMMENDED_WEB_SEARCH_USER_LOCATION,
UNSUPPORTED_CODE_INTERPRETER_MODELS,
UNSUPPORTED_FLEX_SERVICE_TIERS_MODELS,
UNSUPPORTED_IMAGE_MODELS,
UNSUPPORTED_MODELS,
UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS,
UNSUPPORTED_WEB_SEARCH_MODELS,
)
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_API_KEY): str,
}
)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
"""Validate the user input allows us to connect.
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
client = openai.AsyncOpenAI(
api_key=data[CONF_API_KEY], http_client=get_async_client(hass)
)
await client.models.list(timeout=10.0)
class OpenAIConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for OpenAI Conversation."""
VERSION = 2
MINOR_VERSION = 7
@override
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)
try:
await validate_input(self.hass, user_input)
except openai.APIConnectionError:
errors["base"] = "cannot_connect"
except openai.AuthenticationError:
errors["base"] = "invalid_auth"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
if self.source == SOURCE_REAUTH:
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data_updates=user_input
)
return self.async_create_entry(
title="ChatGPT",
data=user_input,
subentries=[
{
"subentry_type": "conversation",
"data": RECOMMENDED_CONVERSATION_OPTIONS,
"title": DEFAULT_CONVERSATION_NAME,
"unique_id": None,
},
{
"subentry_type": "ai_task_data",
"data": RECOMMENDED_AI_TASK_OPTIONS,
"title": DEFAULT_AI_TASK_NAME,
"unique_id": None,
},
{
"subentry_type": "stt",
"data": RECOMMENDED_STT_OPTIONS,
"title": DEFAULT_STT_NAME,
"unique_id": None,
},
{
"subentry_type": "tts",
"data": RECOMMENDED_TTS_OPTIONS,
"title": DEFAULT_TTS_NAME,
"unique_id": None,
},
],
)
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
STEP_USER_DATA_SCHEMA, user_input
),
errors=errors,
description_placeholders={
"instructions_url": "https://github.com/adamjs83/ha-openai-compatible#generate-an-api-key",
},
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that informs the user that reauth is required."""
if not user_input:
return self.async_show_form(
step_id="reauth_confirm", data_schema=STEP_USER_DATA_SCHEMA
)
return await self.async_step_user(user_input)
@classmethod
@callback
@override
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {
"conversation": OpenAISubentryFlowHandler,
"ai_task_data": OpenAISubentryFlowHandler,
"stt": OpenAISubentrySTTFlowHandler,
"tts": OpenAISubentryTTSFlowHandler,
}
class OpenAISubentryFlowHandler(ConfigSubentryFlow):
"""Flow for managing OpenAI subentries."""
options: dict[str, Any]
@property
def _is_new(self) -> bool:
"""Return if this is a new subentry."""
return self.source == "user"
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Add a subentry."""
if self._subentry_type == "ai_task_data":
self.options = RECOMMENDED_AI_TASK_OPTIONS.copy()
else:
self.options = RECOMMENDED_CONVERSATION_OPTIONS.copy()
return await self.async_step_init()
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle reconfiguration of a subentry."""
self.options = self._get_reconfigure_subentry().data.copy()
return await self.async_step_init()
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Manage initial options."""
# abort if entry is not loaded
if self._get_entry().state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
options = self.options
hass_apis: list[SelectOptionDict] = [
SelectOptionDict(
label=api.name,
value=api.id,
)
for api in llm.async_get_apis(self.hass)
]
if suggested_llm_apis := options.get(CONF_LLM_HASS_API):
if isinstance(suggested_llm_apis, str):
suggested_llm_apis = [suggested_llm_apis]
valid_apis = {api.id for api in llm.async_get_apis(self.hass)}
options[CONF_LLM_HASS_API] = [
api for api in suggested_llm_apis if api in valid_apis
]
step_schema: VolDictType = {}
if self._is_new:
if self._subentry_type == "ai_task_data":
default_name = DEFAULT_AI_TASK_NAME
else:
default_name = DEFAULT_CONVERSATION_NAME
step_schema[vol.Required(CONF_NAME, default=default_name)] = str
if self._subentry_type == "conversation":
step_schema.update(
{
vol.Optional(
CONF_PROMPT,
description={
"suggested_value": options.get(
CONF_PROMPT, llm.DEFAULT_INSTRUCTIONS_PROMPT
)
},
): TemplateSelector(),
vol.Optional(CONF_LLM_HASS_API): SelectSelector(
SelectSelectorConfig(options=hass_apis, multiple=True)
),
}
)
step_schema[
vol.Required(CONF_RECOMMENDED, default=options.get(CONF_RECOMMENDED, False))
] = bool
if user_input is not None:
if user_input.get(CONF_LLM_HASS_API) is None:
user_input.pop(CONF_LLM_HASS_API, None)
if user_input[CONF_RECOMMENDED]:
if self._is_new:
return self.async_create_entry(
title=user_input.pop(CONF_NAME),
data=user_input,
)
return self.async_update_and_abort(
self._get_entry(),
self._get_reconfigure_subentry(),
data=user_input,
)
options.update(user_input)
if CONF_LLM_HASS_API in options and CONF_LLM_HASS_API not in user_input:
options.pop(CONF_LLM_HASS_API)
return await self.async_step_additional()
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(step_schema), options
),
)
async def async_step_additional(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Manage additional options."""
options = self.options
errors: dict[str, str] = {}
step_schema: VolDictType = {
vol.Optional(
CONF_CHAT_MODEL,
default=RECOMMENDED_CHAT_MODEL,
): str,
vol.Optional(
CONF_MAX_TOKENS,
default=RECOMMENDED_MAX_TOKENS,
): int,
vol.Optional(
CONF_TOP_P,
default=RECOMMENDED_TOP_P,
): NumberSelector(NumberSelectorConfig(min=0, max=1, step=0.05)),
vol.Optional(
CONF_TEMPERATURE,
default=RECOMMENDED_TEMPERATURE,
): NumberSelector(NumberSelectorConfig(min=0, max=2, step=0.05)),
vol.Optional(
CONF_STORE_RESPONSES,
default=RECOMMENDED_STORE_RESPONSES,
): bool,
}
if user_input is not None:
options.update(user_input)
if user_input.get(CONF_CHAT_MODEL) in UNSUPPORTED_MODELS:
errors[CONF_CHAT_MODEL] = "model_not_supported"
if not errors:
return await self.async_step_model()
return self.async_show_form(
step_id="additional",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(step_schema), options
),
errors=errors,
)
async def async_step_model(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Manage model-specific options."""
options = self.options
errors: dict[str, str] = {}
step_schema: VolDictType = {}
model = options[CONF_CHAT_MODEL]
if not model.startswith(tuple(UNSUPPORTED_CODE_INTERPRETER_MODELS)):
step_schema.update(
{
vol.Optional(
CONF_CODE_INTERPRETER,
default=RECOMMENDED_CODE_INTERPRETER,
): bool,
}
)
elif CONF_CODE_INTERPRETER in options:
options.pop(CONF_CODE_INTERPRETER)
if reasoning_options := self._get_reasoning_options(model):
step_schema.update(
{
vol.Optional(
CONF_REASONING_EFFORT,
default=RECOMMENDED_REASONING_EFFORT,
): SelectSelector(
SelectSelectorConfig(
options=reasoning_options,
translation_key=CONF_REASONING_EFFORT,
mode=SelectSelectorMode.DROPDOWN,
)
),
}
)
elif CONF_REASONING_EFFORT in options:
options.pop(CONF_REASONING_EFFORT)
if model.startswith("gpt-5.6"):
step_schema.update(
{
vol.Optional(
CONF_PRO_MODE,
default=RECOMMENDED_PRO_MODE,
): bool,
}
)
elif CONF_PRO_MODE in options:
options.pop(CONF_PRO_MODE)
if model.startswith("gpt-5"):
step_schema.update(
{
vol.Optional(
CONF_VERBOSITY,
default=RECOMMENDED_VERBOSITY,
): SelectSelector(
SelectSelectorConfig(
options=["low", "medium", "high"],
translation_key=CONF_VERBOSITY,
mode=SelectSelectorMode.DROPDOWN,
)
),
}
)
elif CONF_VERBOSITY in options:
options.pop(CONF_VERBOSITY)
if model.startswith(("o", "gpt-5")):
reasoning_summary_options = ["off", "auto", "concise", "detailed"]
if model.startswith("o"):
reasoning_summary_options.remove("concise")
stored_summary = options.get(
CONF_REASONING_SUMMARY, RECOMMENDED_REASONING_SUMMARY
)
if stored_summary not in reasoning_summary_options:
stored_summary = RECOMMENDED_REASONING_SUMMARY
options[CONF_REASONING_SUMMARY] = stored_summary
step_schema.update(
{
vol.Optional(
CONF_REASONING_SUMMARY,
default=stored_summary,
): SelectSelector(
SelectSelectorConfig(
options=reasoning_summary_options,
translation_key=CONF_REASONING_SUMMARY,
mode=SelectSelectorMode.DROPDOWN,
)
),
}
)
elif CONF_REASONING_SUMMARY in options:
options.pop(CONF_REASONING_SUMMARY)
service_tiers = self._get_service_tiers(model)
if "flex" in service_tiers or "priority" in service_tiers:
step_schema[
vol.Optional(
CONF_SERVICE_TIER,
default=RECOMMENDED_SERVICE_TIER,
)
] = SelectSelector(
SelectSelectorConfig(
options=service_tiers,
translation_key=CONF_SERVICE_TIER,
mode=SelectSelectorMode.DROPDOWN,
)
)
else:
options.pop(CONF_SERVICE_TIER, None)
if options.get(CONF_SERVICE_TIER) not in service_tiers:
options.pop(CONF_SERVICE_TIER, None)
if self._subentry_type == "conversation" and not model.startswith(
tuple(UNSUPPORTED_WEB_SEARCH_MODELS)
):
step_schema.update(
{
vol.Optional(
CONF_WEB_SEARCH,
default=RECOMMENDED_WEB_SEARCH,
): bool,
vol.Optional(
CONF_WEB_SEARCH_CONTEXT_SIZE,
default=RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE,
): SelectSelector(
SelectSelectorConfig(
options=["low", "medium", "high"],
translation_key=CONF_WEB_SEARCH_CONTEXT_SIZE,
mode=SelectSelectorMode.DROPDOWN,
)
),
vol.Optional(
CONF_WEB_SEARCH_USER_LOCATION,
default=RECOMMENDED_WEB_SEARCH_USER_LOCATION,
): bool,
vol.Optional(
CONF_WEB_SEARCH_INLINE_CITATIONS,
default=RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
): bool,
}
)
elif CONF_WEB_SEARCH in options:
options = {
k: v
for k, v in options.items()
if k
not in (
CONF_WEB_SEARCH,
CONF_WEB_SEARCH_CONTEXT_SIZE,
CONF_WEB_SEARCH_USER_LOCATION,
CONF_WEB_SEARCH_CITY,
CONF_WEB_SEARCH_REGION,
CONF_WEB_SEARCH_COUNTRY,
CONF_WEB_SEARCH_TIMEZONE,
CONF_WEB_SEARCH_INLINE_CITATIONS,
)
}
if self._subentry_type == "ai_task_data" and not model.startswith(
tuple(UNSUPPORTED_IMAGE_MODELS)
):
step_schema[
vol.Optional(CONF_IMAGE_MODEL, default=RECOMMENDED_IMAGE_MODEL)
] = SelectSelector(
SelectSelectorConfig(
options=[
"gpt-image-2",
"gpt-image-1.5",
"gpt-image-1",
"gpt-image-1-mini",
],
mode=SelectSelectorMode.DROPDOWN,
)
)
if user_input is not None:
if user_input.get(CONF_WEB_SEARCH):
if user_input.get(CONF_REASONING_EFFORT) == "minimal":
errors[CONF_WEB_SEARCH] = "web_search_minimal_reasoning"
if user_input.get(CONF_WEB_SEARCH_USER_LOCATION) and not errors:
user_input.update(await self._get_location_data())
else:
options.pop(CONF_WEB_SEARCH_CITY, None)
options.pop(CONF_WEB_SEARCH_REGION, None)
options.pop(CONF_WEB_SEARCH_COUNTRY, None)
options.pop(CONF_WEB_SEARCH_TIMEZONE, None)
if (
user_input.get(CONF_CODE_INTERPRETER)
and user_input.get(CONF_REASONING_EFFORT) == "minimal"
):
errors[CONF_CODE_INTERPRETER] = "code_interpreter_minimal_reasoning"
options.update(user_input)
if not errors:
if self._is_new:
return self.async_create_entry(
title=options.pop(CONF_NAME),
data=options,
)
return self.async_update_and_abort(
self._get_entry(),
self._get_reconfigure_subentry(),
data=options,
)
return self.async_show_form(
step_id="model",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(step_schema), options
),
errors=errors,
)
def _get_reasoning_options(self, model: str) -> list[str]:
"""Get reasoning effort options based on model."""
if not model.startswith(("o", "gpt-5")) or model.startswith("gpt-5-pro"):
return []
models_reasoning_map: dict[str | tuple[str, ...], list[str]] = {
"gpt-5.6": ["none", "low", "medium", "high", "xhigh", "max"],
("gpt-5.2-pro", "gpt-5.4-pro", "gpt-5.5-pro"): ["medium", "high", "xhigh"],
("gpt-5.2", "gpt-5.3", "gpt-5.4", "gpt-5.5"): [
"none",
"low",
"medium",
"high",
"xhigh",
],
"gpt-5.1": ["none", "low", "medium", "high"],
"gpt-5": ["minimal", "low", "medium", "high"],
"": ["low", "medium", "high"], # The default case
}
for prefix, options in models_reasoning_map.items():
if model.startswith(prefix):
return options
return [] # pragma: no cover
def _get_service_tiers(self, model: str) -> list[str]:
"""Get service tier options based on model."""
service_tiers = ["auto"]
if not model.startswith(tuple(UNSUPPORTED_FLEX_SERVICE_TIERS_MODELS)):
service_tiers.append("flex")
service_tiers.append("default")
if not model.startswith(tuple(UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS)):
service_tiers.append("priority")
return service_tiers
async def _get_location_data(self) -> dict[str, str]:
"""Get approximate location data of the user."""
location_data: dict[str, str] = {}
zone_home = self.hass.states.get(ENTITY_ID_HOME)
if zone_home is not None:
client = openai.AsyncOpenAI(
api_key=self._get_entry().data[CONF_API_KEY],
http_client=get_async_client(self.hass),
)
location_schema = vol.Schema(
{
vol.Optional(
CONF_WEB_SEARCH_CITY,
description=(
"Free text input for the city, e.g. `San Francisco`"
),
): str,
vol.Optional(
CONF_WEB_SEARCH_REGION,
description="Free text input for the region, e.g. `California`",
): str,
}
)
response = await client.responses.create(
model=RECOMMENDED_CHAT_MODEL,
input=[
{
"role": "system",
"content": "Where are the following coordinates located: "
f"({zone_home.attributes[EntityStateAttribute.LATITUDE]},"
f" {zone_home.attributes[EntityStateAttribute.LONGITUDE]})?",
}
],
text={
"format": {
"type": "json_schema",
"name": "approximate_location",
"description": "Approximate location data of the user "
"for refined web search results",
"schema": to_openapi(location_schema),
"strict": False,
}
},
store=self.options.get(
CONF_STORE_RESPONSES, RECOMMENDED_STORE_RESPONSES
),
)
location_data = location_schema(json.loads(response.output_text) or {})
if self.hass.config.country:
location_data[CONF_WEB_SEARCH_COUNTRY] = self.hass.config.country
location_data[CONF_WEB_SEARCH_TIMEZONE] = self.hass.config.time_zone
_LOGGER.debug("Location data: %s", location_data)
return location_data
class OpenAISubentrySTTFlowHandler(ConfigSubentryFlow):
"""Flow for managing OpenAI STT subentries."""
options: dict[str, Any]
@property
def _is_new(self) -> bool:
"""Return if this is a new subentry."""
return self.source == "user"
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Add a subentry."""
self.options = RECOMMENDED_STT_OPTIONS.copy()
return await self.async_step_init()
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle reconfiguration of a subentry."""
self.options = self._get_reconfigure_subentry().data.copy()
return await self.async_step_init()
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Manage initial options."""
# abort if entry is not loaded
if self._get_entry().state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
options = self.options
errors: dict[str, str] = {}
step_schema: VolDictType = {}
if self._is_new:
step_schema[vol.Required(CONF_NAME, default=DEFAULT_STT_NAME)] = str
step_schema.update(
{
vol.Optional(
CONF_PROMPT,
description={
"suggested_value": options.get(CONF_PROMPT, DEFAULT_STT_PROMPT)
},
): TextSelector(
TextSelectorConfig(multiline=True, type=TextSelectorType.TEXT)
),
vol.Optional(
CONF_CHAT_MODEL, default=RECOMMENDED_STT_MODEL
): SelectSelector(
SelectSelectorConfig(
options=[
"gpt-4o-transcribe",
"gpt-4o-mini-transcribe",
"whisper-1",
],
mode=SelectSelectorMode.DROPDOWN,
custom_value=True,
)
),
}
)
if user_input is not None:
options.update(user_input)
if not errors:
if self._is_new:
return self.async_create_entry(
title=options.pop(CONF_NAME),
data=options,
)
return self.async_update_and_abort(
self._get_entry(),
self._get_reconfigure_subentry(),
data=options,
)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(step_schema), options
),
errors=errors,
)
class OpenAISubentryTTSFlowHandler(ConfigSubentryFlow):
"""Flow for managing OpenAI TTS subentries."""
options: dict[str, Any]
@property
def _is_new(self) -> bool:
"""Return if this is a new subentry."""
return self.source == "user"
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Add a subentry."""
self.options = RECOMMENDED_TTS_OPTIONS.copy()
return await self.async_step_init()
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Handle reconfiguration of a subentry."""
self.options = self._get_reconfigure_subentry().data.copy()
return await self.async_step_init()
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Manage initial options."""
# abort if entry is not loaded
if self._get_entry().state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
options = self.options
errors: dict[str, str] = {}
step_schema: VolDictType = {}
if self._is_new:
step_schema[vol.Required(CONF_NAME, default=DEFAULT_TTS_NAME)] = str
step_schema.update(
{
vol.Optional(CONF_PROMPT): TextSelector(
TextSelectorConfig(multiline=True, type=TextSelectorType.TEXT)
),
vol.Optional(
CONF_TTS_SPEED, default=RECOMMENDED_TTS_SPEED
): NumberSelector(NumberSelectorConfig(min=0.25, max=4.0, step=0.01)),
}
)
if user_input is not None:
options.update(user_input)
if not errors:
if self._is_new:
return self.async_create_entry(
title=options.pop(CONF_NAME),
data=options,
)
return self.async_update_and_abort(
self._get_entry(),
self._get_reconfigure_subentry(),
data=options,
)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(step_schema), options
),
errors=errors,
)
custom_components/openai_compatible/const.py
+160
View File
@@ -0,0 +1,160 @@
"""Constants for the OpenAI Conversation integration."""
import logging
from typing import Any
from homeassistant.const import CONF_LLM_HASS_API, CONF_PROMPT
from homeassistant.helpers import llm
DOMAIN = "openai_compatible"
LOGGER: logging.Logger = logging.getLogger(__package__)
DEFAULT_CONVERSATION_NAME = "OpenAI Conversation"
DEFAULT_AI_TASK_NAME = "OpenAI AI Task"
DEFAULT_STT_NAME = "OpenAI STT"
DEFAULT_TTS_NAME = "OpenAI TTS"
DEFAULT_NAME = "OpenAI Conversation"
CONF_CHAT_MODEL = "chat_model"
CONF_IMAGE_MODEL = "image_model"
CONF_CODE_INTERPRETER = "code_interpreter"
CONF_FILENAMES = "filenames"
CONF_MAX_TOKENS = "max_tokens"
CONF_PRO_MODE = "pro_mode"
CONF_REASONING_EFFORT = "reasoning_effort"
CONF_REASONING_SUMMARY = "reasoning_summary"
CONF_RECOMMENDED = "recommended"
CONF_STORE_RESPONSES = "store_responses"
CONF_SERVICE_TIER = "service_tier"
CONF_TEMPERATURE = "temperature"
CONF_TOP_P = "top_p"
CONF_TTS_SPEED = "tts_speed"
CONF_VERBOSITY = "verbosity"
CONF_WEB_SEARCH = "web_search"
CONF_WEB_SEARCH_USER_LOCATION = "user_location"
CONF_WEB_SEARCH_CONTEXT_SIZE = "search_context_size"
CONF_WEB_SEARCH_CITY = "city"
CONF_WEB_SEARCH_REGION = "region"
CONF_WEB_SEARCH_COUNTRY = "country"
CONF_WEB_SEARCH_TIMEZONE = "timezone"
CONF_WEB_SEARCH_INLINE_CITATIONS = "inline_citations"
RECOMMENDED_CODE_INTERPRETER = False
RECOMMENDED_CHAT_MODEL = "gpt-4o-mini"
RECOMMENDED_IMAGE_MODEL = "gpt-image-2"
RECOMMENDED_MAX_TOKENS = 3000
RECOMMENDED_PRO_MODE = False
RECOMMENDED_REASONING_EFFORT = "low"
RECOMMENDED_STORE_RESPONSES = False
RECOMMENDED_REASONING_SUMMARY = "auto"
RECOMMENDED_SERVICE_TIER = "auto"
RECOMMENDED_STT_MODEL = "gpt-4o-mini-transcribe"
RECOMMENDED_TEMPERATURE = 1.0
RECOMMENDED_TOP_P = 1.0
RECOMMENDED_TTS_SPEED = 1.0
RECOMMENDED_VERBOSITY = "medium"
RECOMMENDED_WEB_SEARCH = False
RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE = "medium"
RECOMMENDED_WEB_SEARCH_USER_LOCATION = False
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS = False
DEFAULT_STT_PROMPT = (
"The following conversation is a smart home user talking to Home Assistant."
)
UNSUPPORTED_MODELS: list[str] = [
"o1-mini",
"o1-mini-2024-09-12",
"o1-preview",
"o1-preview-2024-09-12",
"gpt-4o-realtime-preview",
"gpt-4o-realtime-preview-2024-12-17",
"gpt-4o-realtime-preview-2024-10-01",
"gpt-4o-mini-realtime-preview",
"gpt-4o-mini-realtime-preview-2024-12-17",
]
UNSUPPORTED_WEB_SEARCH_MODELS: list[str] = [
"gpt-3.5",
"gpt-4-turbo",
"gpt-4.1-nano",
"o1",
"o3-mini",
]
UNSUPPORTED_IMAGE_MODELS: list[str] = [
"gpt-5-mini",
"o3-mini",
"o4",
"o1",
"gpt-3.5",
"gpt-4-turbo",
]
UNSUPPORTED_CODE_INTERPRETER_MODELS: list[str] = [
"gpt-5-pro",
"gpt-5.2-pro",
"gpt-5-codex",
"gpt-5.1-codex",
"gpt-5.2-codex",
]
UNSUPPORTED_EXTENDED_CACHE_RETENTION_MODELS: list[str] = [
"o1",
"o3",
"o4",
"gpt-3.5",
"gpt-4-turbo",
"gpt-4o",
"gpt-4.1-mini",
"gpt-4.1-nano",
"gpt-5-mini",
"gpt-5-nano",
]
RECOMMENDED_CONVERSATION_OPTIONS = {
CONF_RECOMMENDED: True,
CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
CONF_PROMPT: llm.DEFAULT_INSTRUCTIONS_PROMPT,
}
RECOMMENDED_AI_TASK_OPTIONS = {
CONF_RECOMMENDED: True,
}
RECOMMENDED_STT_OPTIONS: dict[str, Any] = {}
RECOMMENDED_TTS_OPTIONS = {
CONF_PROMPT: "",
CONF_CHAT_MODEL: "gpt-4o-mini-tts",
}
UNSUPPORTED_FLEX_SERVICE_TIERS_MODELS: list[str] = [
"gpt-5.3",
"gpt-5.2-chat",
"gpt-5.1-chat",
"gpt-5-chat",
"gpt-5.2-codex",
"gpt-5.1-codex",
"gpt-5-codex",
"gpt-5.2-pro",
"gpt-5-pro",
"gpt-4",
"o1",
"o3-pro",
"o3-deep-research",
"o4-mini-deep-research",
"o3-mini",
"codex-mini",
]
UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS: list[str] = [
"gpt-5-nano",
"gpt-5.3-chat",
"gpt-5.2-chat",
"gpt-5.1-chat",
"gpt-5.1-codex-mini",
"gpt-5-chat",
"gpt-5.2-pro",
"gpt-5-pro",
"o1",
"o3-pro",
"o3-deep-research",
"o4-mini-deep-research",
"o3-mini",
"codex-mini",
]
custom_components/openai_compatible/conversation.py
+90
View File
@@ -0,0 +1,90 @@
"""Conversation support for OpenAI."""
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 OpenAIConfigEntry
from .const import DOMAIN
from .entity import OpenAIBaseLLMEntity
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: OpenAIConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up conversation entities."""
for subentry in config_entry.subentries.values():
if subentry.subentry_type != "conversation":
continue
async_add_entities(
[OpenAIConversationEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class OpenAIConversationEntity(
conversation.ConversationEntity,
conversation.AbstractConversationAgent,
OpenAIBaseLLMEntity,
):
"""OpenAI conversation agent."""
_attr_supports_streaming = True
def __init__(self, entry: OpenAIConfigEntry, 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_added_to_hass(self) -> None:
"""When entity is added to Home Assistant."""
await super().async_added_to_hass()
conversation.async_set_agent(self.hass, self.entry, self)
@override
async def async_will_remove_from_hass(self) -> None:
"""When entity will be removed from Home Assistant."""
conversation.async_unset_agent(self.hass, self.entry)
await super().async_will_remove_from_hass()
@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)
custom_components/openai_compatible/entity.py
+771
View File
@@ -0,0 +1,771 @@
"""Base entity for OpenAI."""
import base64
from collections.abc import AsyncGenerator, Callable, Iterable
import json
from mimetypes import guess_file_type
from pathlib import Path
import re
from typing import TYPE_CHECKING, Any, Literal, cast
import openai
from openai._streaming import AsyncStream
from openai.types.responses import (
EasyInputMessageParam,
FunctionToolParam,
ResponseCodeInterpreterToolCall,
ResponseCompletedEvent,
ResponseErrorEvent,
ResponseFailedEvent,
ResponseFunctionCallArgumentsDeltaEvent,
ResponseFunctionCallArgumentsDoneEvent,
ResponseFunctionToolCall,
ResponseFunctionToolCallParam,
ResponseFunctionWebSearch,
ResponseFunctionWebSearchParam,
ResponseIncompleteEvent,
ResponseInputFileParam,
ResponseInputImageParam,
ResponseInputMessageContentListParam,
ResponseInputParam,
ResponseInputTextParam,
ResponseOutputItemAddedEvent,
ResponseOutputItemDoneEvent,
ResponseOutputMessage,
ResponseReasoningItem,
ResponseReasoningItemParam,
ResponseReasoningSummaryTextDeltaEvent,
ResponseStreamEvent,
ResponseTextDeltaEvent,
ToolChoiceTypesParam,
ToolParam,
WebSearchToolParam,
)
from openai.types.responses.response_create_params import (
Reasoning,
ResponseCreateParamsStreaming,
)
from openai.types.responses.response_input_param import (
FunctionCallOutput,
ImageGenerationCall as ImageGenerationCallParam,
)
from openai.types.responses.response_output_item import ImageGenerationCall
from openai.types.responses.tool_param import (
CodeInterpreter,
CodeInterpreterContainerCodeInterpreterToolAuto,
ImageGeneration,
)
from openai.types.responses.web_search_tool_param import UserLocation
from probatio import to_openapi
import voluptuous as vol
from homeassistant.components import conversation
from homeassistant.config_entries import ConfigSubentry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, issue_registry as ir, llm
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.json import json_dumps
from homeassistant.util import slugify
from .const import (
CONF_CHAT_MODEL,
CONF_CODE_INTERPRETER,
CONF_IMAGE_MODEL,
CONF_MAX_TOKENS,
CONF_PRO_MODE,
CONF_REASONING_EFFORT,
CONF_REASONING_SUMMARY,
CONF_SERVICE_TIER,
CONF_STORE_RESPONSES,
CONF_TEMPERATURE,
CONF_TOP_P,
CONF_VERBOSITY,
CONF_WEB_SEARCH,
CONF_WEB_SEARCH_CITY,
CONF_WEB_SEARCH_CONTEXT_SIZE,
CONF_WEB_SEARCH_COUNTRY,
CONF_WEB_SEARCH_INLINE_CITATIONS,
CONF_WEB_SEARCH_REGION,
CONF_WEB_SEARCH_TIMEZONE,
CONF_WEB_SEARCH_USER_LOCATION,
DOMAIN,
LOGGER,
RECOMMENDED_CHAT_MODEL,
RECOMMENDED_IMAGE_MODEL,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_PRO_MODE,
RECOMMENDED_REASONING_EFFORT,
RECOMMENDED_REASONING_SUMMARY,
RECOMMENDED_SERVICE_TIER,
RECOMMENDED_STORE_RESPONSES,
RECOMMENDED_STT_MODEL,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
RECOMMENDED_VERBOSITY,
RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE,
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
UNSUPPORTED_EXTENDED_CACHE_RETENTION_MODELS,
)
if TYPE_CHECKING:
from . import OpenAIConfigEntry
# Max number of back and forth with the LLM to generate a response
MAX_TOOL_ITERATIONS = 10
def _adjust_schema(schema: dict[str, Any]) -> None:
"""Adjust the output schema to be compatible with OpenAI API."""
if schema["type"] == "object":
schema.setdefault("strict", True)
schema.setdefault("additionalProperties", False)
if "properties" not in schema:
return
if "required" not in schema:
schema["required"] = []
# Ensure all properties are required
for prop, prop_info in schema["properties"].items():
_adjust_schema(prop_info)
if prop not in schema["required"]:
prop_info["type"] = [prop_info["type"], "null"]
schema["required"].append(prop)
elif schema["type"] == "array":
if "items" not in schema:
return
_adjust_schema(schema["items"])
def _format_structured_output(
schema: vol.Schema, llm_api: llm.APIInstance | None
) -> dict[str, Any]:
"""Format the schema to be compatible with OpenAI API."""
result: dict[str, Any] = to_openapi(
schema,
custom_serializer=(
llm_api.custom_serializer if llm_api else llm.selector_serializer
),
)
_adjust_schema(result)
return result
def _format_tool(
tool: llm.Tool, custom_serializer: Callable[[Any], Any] | None
) -> FunctionToolParam:
"""Format tool specification."""
unsupported_keys = {"oneOf", "anyOf", "allOf", "enum", "not"}
schema = to_openapi(tool.parameters, custom_serializer=custom_serializer)
if unsupported_keys.intersection(schema):
schema = {k: v for k, v in schema.items() if k not in unsupported_keys}
return FunctionToolParam(
type="function",
name=tool.name,
parameters=schema,
description=tool.description,
strict=False,
)
def _convert_content_to_param(
chat_content: Iterable[conversation.Content],
) -> ResponseInputParam:
"""Convert any native chat message for this agent to the native format."""
messages: ResponseInputParam = []
reasoning_summary: list[str] = []
web_search_calls: dict[str, ResponseFunctionWebSearchParam] = {}
for content in chat_content:
if isinstance(content, conversation.ToolResultContent):
if (
content.tool_name == "web_search_call"
and content.tool_call_id in web_search_calls
):
web_search_call = web_search_calls.pop(content.tool_call_id)
web_search_call["status"] = content.tool_result.get( # type: ignore[typeddict-item]
"status", "completed"
)
messages.append(web_search_call)
else:
messages.append(
FunctionCallOutput(
type="function_call_output",
call_id=content.tool_call_id,
output=json_dumps(content.tool_result),
)
)
continue
if content.content:
role: Literal["user", "assistant", "system", "developer"] = content.role
if role == "system":
role = "developer"
messages.append(
EasyInputMessageParam(
type="message", role=role, content=content.content
)
)
if isinstance(content, conversation.AssistantContent):
if content.tool_calls:
for tool_call in content.tool_calls:
if (
tool_call.external
and tool_call.tool_name == "web_search_call"
and "action" in tool_call.tool_args
):
web_search_calls[tool_call.id] = ResponseFunctionWebSearchParam(
type="web_search_call",
id=tool_call.id,
action=tool_call.tool_args["action"],
status="completed",
)
else:
messages.append(
ResponseFunctionToolCallParam(
type="function_call",
name=tool_call.tool_name,
arguments=json_dumps(tool_call.tool_args),
call_id=tool_call.id,
)
)
if content.thinking_content:
reasoning_summary.append(content.thinking_content)
if isinstance(content.native, ResponseReasoningItem):
messages.append(
ResponseReasoningItemParam(
type="reasoning",
id=content.native.id,
summary=(
[
{
"type": "summary_text",
"text": summary,
}
for summary in reasoning_summary
]
if content.thinking_content
else []
),
encrypted_content=content.native.encrypted_content,
)
)
reasoning_summary = []
elif isinstance(content.native, ImageGenerationCall):
messages.append(
cast(ImageGenerationCallParam, content.native.to_dict())
)
return messages
async def _transform_stream( # noqa: C901 - This is complex, but better to have it in one place
chat_log: conversation.ChatLog,
stream: AsyncStream[ResponseStreamEvent],
remove_citations: bool = False,
) -> AsyncGenerator[
conversation.AssistantContentDeltaDict | conversation.ToolResultContentDeltaDict
]:
"""Transform an OpenAI delta stream into HA format."""
last_summary_index = None
last_role: Literal["assistant", "tool_result"] | None = None
# Non-reasoning models don't follow our request to remove citations, so we remove
# them manually here. They always follow the same pattern: the citation is always
# in parentheses in Markdown format, the citation is always in a single delta event,
# and sometimes the closing parenthesis is split into a separate delta event.
remove_parentheses: bool = False
citation_regexp = re.compile(r"\(\[([^\]]+)\]\((https?:\/\/[^\)]+)\)")
async for event in stream:
LOGGER.debug("Received event: %s", event)
if isinstance(event, ResponseOutputItemAddedEvent):
if isinstance(event.item, ResponseFunctionToolCall):
# OpenAI has tool calls as individual events
# while HA puts tool calls inside the assistant message.
# We turn them into individual assistant content for HA
# to ensure that tools are called as soon as possible.
yield {"role": "assistant"}
last_role = "assistant"
last_summary_index = None
current_tool_call = event.item
elif (
isinstance(event.item, ResponseOutputMessage)
or (
isinstance(event.item, ResponseReasoningItem)
and last_summary_index is not None
) # Subsequent ResponseReasoningItem
or last_role != "assistant"
):
yield {"role": "assistant"}
last_role = "assistant"
last_summary_index = None
elif isinstance(event, ResponseOutputItemDoneEvent):
if isinstance(event.item, ResponseReasoningItem):
yield {
"native": ResponseReasoningItem(
type="reasoning",
id=event.item.id,
summary=[], # Remove summaries
encrypted_content=event.item.encrypted_content,
)
}
last_summary_index = len(event.item.summary) - 1
elif isinstance(event.item, ResponseCodeInterpreterToolCall):
yield {
"tool_calls": [
llm.ToolInput(
id=event.item.id,
tool_name="code_interpreter",
tool_args={
"code": event.item.code,
"container": event.item.container_id,
},
external=True,
)
]
}
yield {
"role": "tool_result",
"tool_call_id": event.item.id,
"tool_name": "code_interpreter",
"tool_result": {
"output": (
[output.to_dict() for output in event.item.outputs] # type: ignore[misc]
if event.item.outputs is not None
else None
)
},
}
last_role = "tool_result"
elif isinstance(event.item, ResponseFunctionWebSearch):
yield {
"tool_calls": [
llm.ToolInput(
id=event.item.id,
tool_name="web_search_call",
tool_args={
"action": event.item.action.to_dict()
if event.item.action
else None,
},
external=True,
)
]
}
yield {
"role": "tool_result",
"tool_call_id": event.item.id,
"tool_name": "web_search_call",
"tool_result": {"status": event.item.status},
}
last_role = "tool_result"
elif isinstance(event.item, ImageGenerationCall):
if last_summary_index is not None:
yield {"role": "assistant"}
last_role = "assistant"
last_summary_index = None
yield {"native": event.item}
last_summary_index = -1 # Trigger new assistant message on next turn
elif isinstance(event, ResponseTextDeltaEvent):
data = event.delta
if remove_parentheses:
data = data.removeprefix(")")
remove_parentheses = False
elif remove_citations and (match := citation_regexp.search(data)):
match_start, match_end = match.span()
# remove leading space if any
if data[match_start - 1 : match_start] == " ":
match_start -= 1
# remove closing parenthesis:
if data[match_end : match_end + 1] == ")":
match_end += 1
else:
remove_parentheses = True
data = data[:match_start] + data[match_end:]
if data:
yield {"content": data}
elif isinstance(event, ResponseReasoningSummaryTextDeltaEvent):
# OpenAI can output several reasoning summaries
# in a single ResponseReasoningItem. We split them as separate
# AssistantContent messages. Only last of them will have
# the reasoning `native` field set.
if (
last_summary_index is not None
and event.summary_index != last_summary_index
):
yield {"role": "assistant"}
last_role = "assistant"
last_summary_index = event.summary_index
yield {"thinking_content": event.delta}
elif isinstance(event, ResponseFunctionCallArgumentsDeltaEvent):
current_tool_call.arguments += event.delta
elif isinstance(event, ResponseFunctionCallArgumentsDoneEvent):
current_tool_call.status = "completed"
yield {
"tool_calls": [
llm.ToolInput(
id=current_tool_call.call_id,
tool_name=current_tool_call.name,
tool_args=json.loads(current_tool_call.arguments),
)
]
}
elif isinstance(event, ResponseCompletedEvent):
if event.response.usage is not None:
chat_log.async_trace(
{
"stats": {
"input_tokens": event.response.usage.input_tokens,
"output_tokens": event.response.usage.output_tokens,
}
}
)
elif isinstance(event, ResponseIncompleteEvent):
if event.response.usage is not None:
chat_log.async_trace(
{
"stats": {
"input_tokens": event.response.usage.input_tokens,
"output_tokens": event.response.usage.output_tokens,
}
}
)
if (
event.response.incomplete_details
and event.response.incomplete_details.reason
):
reason: str = event.response.incomplete_details.reason
else:
reason = "unknown reason"
if reason == "max_output_tokens":
reason = "max output tokens reached"
elif reason == "content_filter":
reason = "content filter triggered"
raise HomeAssistantError(f"OpenAI response incomplete: {reason}")
elif isinstance(event, ResponseFailedEvent):
if event.response.usage is not None:
chat_log.async_trace(
{
"stats": {
"input_tokens": event.response.usage.input_tokens,
"output_tokens": event.response.usage.output_tokens,
}
}
)
reason = "unknown reason"
if event.response.error is not None:
reason = event.response.error.message
raise HomeAssistantError(f"OpenAI response failed: {reason}")
elif isinstance(event, ResponseErrorEvent):
raise HomeAssistantError(f"OpenAI response error: {event.message}")
class OpenAIBaseLLMEntity(Entity):
"""OpenAI conversation agent."""
_attr_has_entity_name = True
_attr_name: str | None = None
def __init__(self, entry: OpenAIConfigEntry, subentry: ConfigSubentry) -> None:
"""Initialize the entity."""
self.entry = entry
self.subentry = subentry
self._attr_unique_id = subentry.subentry_id
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, subentry.subentry_id)},
name=subentry.title,
manufacturer="OpenAI",
model=subentry.data.get(
CONF_CHAT_MODEL,
RECOMMENDED_CHAT_MODEL
if subentry.subentry_type != "stt"
else RECOMMENDED_STT_MODEL,
),
entry_type=dr.DeviceEntryType.SERVICE,
)
async def _async_handle_chat_log( # noqa: C901
self,
chat_log: conversation.ChatLog,
structure_name: str | None = None,
structure: vol.Schema | None = None,
force_image: bool = False,
max_iterations: int = MAX_TOOL_ITERATIONS,
) -> None:
"""Generate an answer for the chat log."""
options = self.subentry.data
messages = _convert_content_to_param(chat_log.content)
model_args = ResponseCreateParamsStreaming(
model=options.get(CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL),
input=messages,
max_output_tokens=options.get(CONF_MAX_TOKENS, RECOMMENDED_MAX_TOKENS),
user=chat_log.conversation_id,
service_tier=options.get(CONF_SERVICE_TIER, RECOMMENDED_SERVICE_TIER),
store=options.get(CONF_STORE_RESPONSES, RECOMMENDED_STORE_RESPONSES),
stream=True,
)
if model_args["model"].startswith(("o", "gpt-5")):
reasoning: Reasoning = {
"effort": options.get(
CONF_REASONING_EFFORT, RECOMMENDED_REASONING_EFFORT
)
if not model_args["model"].startswith("gpt-5-pro")
else "high", # GPT-5 pro only supports reasoning.effort: high
}
reasoning_summary = options.get(
CONF_REASONING_SUMMARY, RECOMMENDED_REASONING_SUMMARY
)
if reasoning_summary != "off":
reasoning["summary"] = reasoning_summary
if options.get(CONF_PRO_MODE, RECOMMENDED_PRO_MODE):
reasoning["mode"] = "pro"
model_args["reasoning"] = reasoning
model_args["include"] = ["reasoning.encrypted_content"]
if (
not model_args["model"].startswith("gpt-5")
or model_args["reasoning"]["effort"] == "none" # type: ignore[index]
):
model_args["top_p"] = options.get(CONF_TOP_P, RECOMMENDED_TOP_P)
model_args["temperature"] = options.get(
CONF_TEMPERATURE, RECOMMENDED_TEMPERATURE
)
if model_args["model"].startswith("gpt-5"):
model_args["text"] = {
"verbosity": options.get(CONF_VERBOSITY, RECOMMENDED_VERBOSITY)
}
if not model_args["model"].startswith(
tuple(UNSUPPORTED_EXTENDED_CACHE_RETENTION_MODELS)
):
model_args["prompt_cache_retention"] = "24h"
tools: list[ToolParam] = []
if chat_log.llm_api:
tools = [
_format_tool(tool, chat_log.llm_api.custom_serializer)
for tool in chat_log.llm_api.tools
]
remove_citations = False
if options.get(CONF_WEB_SEARCH):
web_search = WebSearchToolParam(
type="web_search",
search_context_size=options.get(
CONF_WEB_SEARCH_CONTEXT_SIZE, RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE
),
)
if options.get(CONF_WEB_SEARCH_USER_LOCATION):
web_search["user_location"] = UserLocation(
type="approximate",
city=options.get(CONF_WEB_SEARCH_CITY, ""),
region=options.get(CONF_WEB_SEARCH_REGION, ""),
country=options.get(CONF_WEB_SEARCH_COUNTRY, ""),
timezone=options.get(CONF_WEB_SEARCH_TIMEZONE, ""),
)
if not options.get(
CONF_WEB_SEARCH_INLINE_CITATIONS,
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
):
system_message = cast(EasyInputMessageParam, messages[0])
content = system_message["content"]
if isinstance(content, str):
system_message["content"] = [
ResponseInputTextParam(type="input_text", text=content)
]
system_message["content"].append( # type: ignore[union-attr]
ResponseInputTextParam(
type="input_text",
text="When doing a web search, do not include source citations",
)
)
if not model_args["model"].startswith("o"):
# o-series models handle this correctly with just a prompt
remove_citations = True
tools.append(web_search)
if options.get(CONF_CODE_INTERPRETER):
tools.append(
CodeInterpreter(
type="code_interpreter",
container=CodeInterpreterContainerCodeInterpreterToolAuto(
type="auto"
),
)
)
model_args.setdefault("include", []).append("code_interpreter_call.outputs") # type: ignore[union-attr]
if force_image:
image_model = options.get(CONF_IMAGE_MODEL, RECOMMENDED_IMAGE_MODEL)
image_tool = ImageGeneration(
type="image_generation",
model=image_model,
output_format="png",
)
if image_model not in ("gpt-image-1-mini", "gpt-image-2"):
image_tool["input_fidelity"] = "high"
tools.append(image_tool)
# Keep image state on OpenAI so follow-up prompts can continue by
# conversation ID without resending the generated image data.
model_args["store"] = True
model_args["tool_choice"] = ToolChoiceTypesParam(type="image_generation")
if tools:
model_args["tools"] = tools
last_content = chat_log.content[-1]
# Handle attachments by adding them to the last user message
if last_content.role == "user" and last_content.attachments:
files = await async_prepare_files_for_prompt(
self.hass,
[(a.path, a.mime_type) for a in last_content.attachments],
)
last_message = messages[-1]
assert (
last_message["type"] == "message"
and last_message["role"] == "user"
and isinstance(last_message["content"], str)
)
last_message["content"] = [
{"type": "input_text", "text": last_message["content"]},
*files,
]
if structure and structure_name:
model_args["text"] = {
"format": {
"type": "json_schema",
"name": slugify(structure_name),
"schema": _format_structured_output(structure, chat_log.llm_api),
},
}
client = self.entry.runtime_data
# To prevent infinite loops, we limit the number of iterations
for _iteration in range(max_iterations):
try:
stream = await client.responses.create(**model_args)
content_stream = chat_log.async_add_delta_content_stream(
self.entity_id,
_transform_stream(chat_log, stream, remove_citations),
)
messages.extend(
_convert_content_to_param(
[content async for content in content_stream]
)
)
except openai.RateLimitError as err:
if (
model_args["service_tier"] == "flex"
and "resource unavailable" in (err.message or "").lower()
):
LOGGER.info(
"Flex tier is not available at the moment,"
" continuing with default tier"
)
model_args["service_tier"] = "default"
continue
LOGGER.error("Rate limited by OpenAI: %s", err)
raise HomeAssistantError("Rate limited or insufficient funds") from err
except openai.OpenAIError as err:
if (
isinstance(err, openai.APIError)
and err.type == "insufficient_quota"
):
LOGGER.error("Insufficient funds for OpenAI: %s", err)
raise HomeAssistantError("Insufficient funds for OpenAI") from err
if "Verify Organization" in str(err):
ir.async_create_issue(
self.hass,
DOMAIN,
"organization_verification_required",
is_fixable=False,
is_persistent=False,
learn_more_url="https://help.openai.com/en/articles/10910291-api-organization-verification",
severity=ir.IssueSeverity.WARNING,
translation_key="organization_verification_required",
translation_placeholders={
"platform_settings": "https://platform.openai.com/settings/organization/general"
},
)
LOGGER.error("Error talking to OpenAI: %s", err)
raise HomeAssistantError("Error talking to OpenAI") from err
if not chat_log.unresponded_tool_results:
break
async def async_prepare_files_for_prompt(
hass: HomeAssistant, files: list[tuple[Path, str | None]]
) -> ResponseInputMessageContentListParam:
"""Append files to a prompt.
Caller needs to ensure that the files are allowed.
"""
def append_files_to_content() -> ResponseInputMessageContentListParam:
content: ResponseInputMessageContentListParam = []
for file_path, mime_type in files:
if not file_path.exists():
raise HomeAssistantError(f"`{file_path}` does not exist")
if mime_type is None:
mime_type = guess_file_type(file_path)[0]
if not mime_type or not mime_type.startswith(("image/", "application/pdf")):
raise HomeAssistantError(
"Only images and PDF are supported by the OpenAI API,"
f"`{file_path}` is not an image file or PDF"
)
base64_file = base64.b64encode(file_path.read_bytes()).decode("utf-8")
if mime_type.startswith("image/"):
content.append(
ResponseInputImageParam(
type="input_image",
image_url=f"data:{mime_type};base64,{base64_file}",
detail="auto",
)
)
elif mime_type.startswith("application/pdf"):
content.append(
ResponseInputFileParam(
type="input_file",
filename=str(file_path),
file_data=f"data:{mime_type};base64,{base64_file}",
)
)
return content
return await hass.async_add_executor_job(append_files_to_content)
custom_components/openai_compatible/icons.json
+10
View File
@@ -0,0 +1,10 @@
{
"services": {
"generate_content": {
"service": "mdi:receipt-text"
},
"generate_image": {
"service": "mdi:image-sync"
}
}
}
custom_components/openai_compatible/manifest.json
+14
View File
@@ -0,0 +1,14 @@
{
"domain": "openai_compatible",
"name": "OpenAI Compatible",
"after_dependencies": ["assist_pipeline", "intent"],
"codeowners": ["@adamjs83"],
"config_flow": true,
"dependencies": ["conversation"],
"documentation": "https://github.com/adamjs83/ha-openai-compatible",
"integration_type": "service",
"iot_class": "cloud_polling",
"issue_tracker": "https://github.com/adamjs83/ha-openai-compatible/issues",
"requirements": ["openai==2.45.0"],
"version": "1.0.0"
}
custom_components/openai_compatible/quality_scale.yaml
+99
View File
@@ -0,0 +1,99 @@
rules:
# Bronze
action-setup: done
appropriate-polling:
status: exempt
comment: |
Integration does not poll.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions: done
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: done
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: done
docs-installation-parameters: done
entity-unavailable: todo
integration-owner: todo
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow: done
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
Service integration, no discovery.
discovery:
status: exempt
comment: |
Service integration, no discovery.
docs-data-update:
status: exempt
comment: |
No data updates.
docs-examples: done
docs-known-limitations: done
docs-supported-devices:
status: exempt
comment: |
Service integration, no devices.
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices:
status: exempt
comment: |
Service integration, no devices.
entity-category:
status: exempt
comment: |
No entities with categories.
entity-device-class:
status: exempt
comment: |
No entities with device classes.
entity-disabled-by-default:
status: exempt
comment: |
No entities disabled by default.
entity-translations: done
exception-translations: todo
icon-translations: done
reconfiguration-flow: done
repair-issues: done
stale-devices:
status: exempt
comment: |
Service integration, no devices.
# Platinum
async-dependency: done
inject-websession:
status: done
comment: |
Uses `httpx` session.
strict-typing: done
custom_components/openai_compatible/services.yaml
+60
View File
@@ -0,0 +1,60 @@
generate_image:
fields:
config_entry:
required: true
selector:
config_entry:
integration: openai_compatible
prompt:
required: true
selector:
text:
multiline: true
size:
required: false
example: "1024x1024"
default: "1024x1024"
selector:
select:
options:
- "1024x1024"
- "1024x1792"
- "1792x1024"
quality:
required: false
example: "standard"
default: "standard"
selector:
select:
options:
- "standard"
- "hd"
style:
required: false
example: "vivid"
default: "vivid"
selector:
select:
options:
- "vivid"
- "natural"
generate_content:
fields:
config_entry:
required: true
selector:
config_entry:
integration: openai_compatible
prompt:
required: true
selector:
text:
multiline: true
example: "Hello, how can I help you?"
filenames:
selector:
text:
multiline: true
example: |
- /path/to/file1.txt
- /path/to/file2.txt
custom_components/openai_compatible/strings.json
+325
View File
@@ -0,0 +1,325 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"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": {
"reauth_confirm": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "[%key:component::openai_compatible::config::step::user::data_description::api_key%]"
},
"description": "Reauthentication required. Please enter your updated API key."
},
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "Your OpenAI API key."
},
"description": "Set up OpenAI Conversation integration by providing your OpenAI API key. Instructions to obtain an API key can be found [here]({instructions_url})."
}
}
},
"config_subentries": {
"ai_task_data": {
"abort": {
"entry_not_loaded": "[%key:component::openai_compatible::config_subentries::conversation::abort::entry_not_loaded%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "AI task",
"error": {
"code_interpreter_minimal_reasoning": "[%key:component::openai_compatible::config_subentries::conversation::error::code_interpreter_minimal_reasoning%]",
"model_not_supported": "[%key:component::openai_compatible::config_subentries::conversation::error::model_not_supported%]",
"web_search_minimal_reasoning": "[%key:component::openai_compatible::config_subentries::conversation::error::web_search_minimal_reasoning%]"
},
"initiate_flow": {
"reconfigure": "Reconfigure AI task",
"user": "Add AI task"
},
"step": {
"additional": {
"data": {
"chat_model": "[%key:common::generic::model%]",
"max_tokens": "[%key:component::openai_compatible::config_subentries::conversation::step::additional::data::max_tokens%]",
"store_responses": "[%key:component::openai_compatible::config_subentries::conversation::step::additional::data::store_responses%]",
"temperature": "[%key:component::openai_compatible::config_subentries::conversation::step::additional::data::temperature%]",
"top_p": "[%key:component::openai_compatible::config_subentries::conversation::step::additional::data::top_p%]"
},
"data_description": {
"store_responses": "[%key:component::openai_compatible::config_subentries::conversation::step::additional::data_description::store_responses%]"
},
"title": "[%key:component::openai_compatible::config_subentries::conversation::step::additional::title%]"
},
"init": {
"data": {
"name": "[%key:common::config_flow::data::name%]",
"recommended": "[%key:component::openai_compatible::config_subentries::conversation::step::init::data::recommended%]"
}
},
"model": {
"data": {
"code_interpreter": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::code_interpreter%]",
"image_model": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::image_model%]",
"inline_citations": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::inline_citations%]",
"pro_mode": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::pro_mode%]",
"reasoning_effort": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::reasoning_effort%]",
"reasoning_summary": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::reasoning_summary%]",
"search_context_size": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::search_context_size%]",
"service_tier": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::service_tier%]",
"user_location": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::user_location%]",
"web_search": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data::web_search%]"
},
"data_description": {
"code_interpreter": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::code_interpreter%]",
"image_model": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::image_model%]",
"inline_citations": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::inline_citations%]",
"pro_mode": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::pro_mode%]",
"reasoning_effort": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::reasoning_effort%]",
"reasoning_summary": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::reasoning_summary%]",
"search_context_size": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::search_context_size%]",
"service_tier": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::service_tier%]",
"user_location": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::user_location%]",
"web_search": "[%key:component::openai_compatible::config_subentries::conversation::step::model::data_description::web_search%]"
},
"title": "[%key:component::openai_compatible::config_subentries::conversation::step::model::title%]"
}
}
},
"conversation": {
"abort": {
"entry_not_loaded": "Cannot add things while the configuration is disabled.",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "Conversation agent",
"error": {
"code_interpreter_minimal_reasoning": "Code interpreter is not supported with minimal reasoning effort",
"model_not_supported": "This model is not supported, please select a different model",
"web_search_minimal_reasoning": "Web search is currently not supported with minimal reasoning effort"
},
"initiate_flow": {
"reconfigure": "Reconfigure conversation agent",
"user": "Add conversation agent"
},
"step": {
"additional": {
"data": {
"chat_model": "[%key:common::generic::model%]",
"max_tokens": "Maximum tokens to return in response",
"store_responses": "Store requests and responses in OpenAI",
"temperature": "Temperature",
"top_p": "Top P"
},
"data_description": {
"store_responses": "If enabled, requests and responses are stored by OpenAI and visible in your OpenAI dashboard logs"
},
"title": "Additional settings"
},
"init": {
"data": {
"llm_hass_api": "[%key:common::config_flow::data::llm_hass_api%]",
"name": "[%key:common::config_flow::data::name%]",
"prompt": "[%key:common::config_flow::data::prompt%]",
"recommended": "Recommended model settings"
},
"data_description": {
"prompt": "Instruct how the LLM should respond. This can be a template."
}
},
"model": {
"data": {
"code_interpreter": "Enable code interpreter tool",
"image_model": "Image generation model",
"inline_citations": "Include links in web search results",
"pro_mode": "Pro mode",
"reasoning_effort": "Reasoning effort",
"reasoning_summary": "Reasoning summary",
"search_context_size": "Search context size",
"service_tier": "Service tier",
"user_location": "Include home location",
"web_search": "Enable web search"
},
"data_description": {
"code_interpreter": "This tool, also known as the python tool to the model, allows it to run code to answer questions",
"image_model": "The model to use when generating images",
"inline_citations": "If disabled, additional prompt is added to ask the model to not include source citations",
"pro_mode": "Perform more model work to improve reliability on difficult tasks and return a single final answer",
"reasoning_effort": "How many reasoning tokens the model should generate before creating a response to the prompt",
"reasoning_summary": "Controls the length and detail of reasoning summaries provided by the model",
"search_context_size": "High level guidance for the amount of context window space to use for the search",
"service_tier": "Controls the cost and response time",
"user_location": "Refine search results based on geography",
"web_search": "Allow the model to search the web for the latest information before generating a response"
},
"title": "Model-specific options"
}
}
},
"stt": {
"abort": {
"entry_not_loaded": "[%key:component::openai_compatible::config_subentries::conversation::abort::entry_not_loaded%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "Speech-to-text",
"initiate_flow": {
"reconfigure": "Reconfigure speech-to-text service",
"user": "Add speech-to-text service"
},
"step": {
"init": {
"data": {
"chat_model": "Model",
"name": "[%key:common::config_flow::data::name%]",
"prompt": "[%key:common::config_flow::data::prompt%]"
},
"data_description": {
"chat_model": "The model to use to transcribe speech.",
"prompt": "Use this prompt to improve the quality of the transcripts. Translate to the pipeline language for best results. See the documentation for more details."
}
}
}
},
"tts": {
"abort": {
"entry_not_loaded": "[%key:component::openai_compatible::config_subentries::conversation::abort::entry_not_loaded%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "Text-to-speech",
"initiate_flow": {
"reconfigure": "Reconfigure text-to-speech service",
"user": "Add text-to-speech service"
},
"step": {
"init": {
"data": {
"name": "[%key:common::config_flow::data::name%]",
"prompt": "[%key:common::config_flow::data::prompt%]",
"tts_speed": "Speed"
},
"data_description": {
"prompt": "Control aspects of speech, including accent, emotional range, intonation, impressions, speed of speech, tone, whispering, and more",
"tts_speed": "Speed of the generated speech"
}
}
}
}
},
"exceptions": {
"invalid_config_entry": {
"message": "Invalid config entry provided. Got {config_entry}"
}
},
"issues": {
"deprecated_generate_content": {
"description": "Action 'openai_compatible.generate_content' is deprecated and will be removed in the 2026.9.0 release. Please use the 'ai_task.generate_data' action instead",
"title": "Deprecated 'generate_content' action"
},
"deprecated_generate_image": {
"description": "Action 'openai_compatible.generate_image' is deprecated and will be removed in the 2026.9.0 release. Please use the 'ai_task.generate_image' action instead",
"title": "Deprecated 'generate_image' action"
},
"organization_verification_required": {
"description": "Your organization must be verified to use this model. Please go to {platform_settings} and select Verify Organization. If you just verified, it can take up to 15 minutes for access to propagate.",
"title": "Organization verification required"
}
},
"selector": {
"reasoning_effort": {
"options": {
"high": "[%key:common::state::high%]",
"low": "[%key:common::state::low%]",
"max": "Max",
"medium": "[%key:common::state::medium%]",
"minimal": "Minimal",
"none": "None",
"xhigh": "X-High"
}
},
"reasoning_summary": {
"options": {
"auto": "[%key:common::state::auto%]",
"concise": "Concise",
"detailed": "Detailed",
"off": "[%key:common::state::off%]"
}
},
"search_context_size": {
"options": {
"high": "[%key:common::state::high%]",
"low": "[%key:common::state::low%]",
"medium": "[%key:common::state::medium%]"
}
},
"service_tier": {
"options": {
"auto": "[%key:common::state::auto%]",
"default": "Standard",
"flex": "Flex",
"priority": "Priority"
}
},
"verbosity": {
"options": {
"high": "[%key:common::state::high%]",
"low": "[%key:common::state::low%]",
"medium": "[%key:common::state::medium%]"
}
}
},
"services": {
"generate_content": {
"description": "Sends a conversational query to ChatGPT including any attached image or PDF files (deprecated)",
"fields": {
"config_entry": {
"description": "The config entry to use for this action",
"name": "Config entry"
},
"filenames": {
"description": "List of files to upload",
"name": "Files"
},
"prompt": {
"description": "The prompt to send",
"name": "Prompt"
}
},
"name": "Generate content (deprecated)"
},
"generate_image": {
"description": "Turns a prompt into an image (deprecated)",
"fields": {
"config_entry": {
"description": "The config entry to use for this action",
"name": "Config entry"
},
"prompt": {
"description": "The text to turn into an image",
"example": "A photo of a dog",
"name": "Prompt"
},
"quality": {
"description": "The quality of the image that will be generated",
"name": "Quality"
},
"size": {
"description": "The size of the image to generate",
"name": "Size"
},
"style": {
"description": "The style of the generated image",
"name": "Style"
}
},
"name": "Generate image (deprecated)"
}
}
}
custom_components/openai_compatible/stt.py
+200
View File
@@ -0,0 +1,200 @@
"""Speech to text support for OpenAI."""
from collections.abc import AsyncIterable
import io
import logging
from typing import TYPE_CHECKING, override
import wave
from openai import OpenAIError
from homeassistant.components import stt
from homeassistant.const import CONF_PROMPT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_CHAT_MODEL, DEFAULT_STT_PROMPT, RECOMMENDED_STT_MODEL
from .entity import OpenAIBaseLLMEntity
if TYPE_CHECKING:
from . import OpenAIConfigEntry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: OpenAIConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up STT entities."""
for subentry in config_entry.subentries.values():
if subentry.subentry_type != "stt":
continue
async_add_entities(
[OpenAISTTEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class OpenAISTTEntity(stt.SpeechToTextEntity, OpenAIBaseLLMEntity):
"""OpenAI Speech to text entity."""
@property
@override
def supported_languages(self) -> list[str]:
"""Return a list of supported languages."""
# https://developers.openai.com/api/docs/guides/speech-to-text#supported-languages
# The model may also transcribe the audio in other
# languages but with lower quality
return [
"af-ZA", # Afrikaans
"ar-SA", # Arabic
"hy-AM", # Armenian
"az-AZ", # Azerbaijani
"be-BY", # Belarusian
"bs-BA", # Bosnian
"bg-BG", # Bulgarian
"ca-ES", # Catalan
"zh-CN", # Chinese (Mandarin)
"hr-HR", # Croatian
"cs-CZ", # Czech
"da-DK", # Danish
"nl-NL", # Dutch
"en-US", # English
"et-EE", # Estonian
"fi-FI", # Finnish
"fr-FR", # French
"gl-ES", # Galician
"de-DE", # German
"el-GR", # Greek
"he-IL", # Hebrew
"hi-IN", # Hindi
"hu-HU", # Hungarian
"is-IS", # Icelandic
"id-ID", # Indonesian
"it-IT", # Italian
"ja-JP", # Japanese
"kn-IN", # Kannada
"kk-KZ", # Kazakh
"ko-KR", # Korean
"lv-LV", # Latvian
"lt-LT", # Lithuanian
"mk-MK", # Macedonian
"ms-MY", # Malay
"mr-IN", # Marathi
"mi-NZ", # Maori
"ne-NP", # Nepali
"no-NO", # Norwegian
"fa-IR", # Persian
"pl-PL", # Polish
"pt-PT", # Portuguese
"ro-RO", # Romanian
"ru-RU", # Russian
"sr-RS", # Serbian
"sk-SK", # Slovak
"sl-SI", # Slovenian
"es-ES", # Spanish
"sw-KE", # Swahili
"sv-SE", # Swedish
"fil-PH", # Tagalog (Filipino)
"ta-IN", # Tamil
"th-TH", # Thai
"tr-TR", # Turkish
"uk-UA", # Ukrainian
"ur-PK", # Urdu
"vi-VN", # Vietnamese
"cy-GB", # Welsh
]
@property
@override
def supported_formats(self) -> list[stt.AudioFormats]:
"""Return a list of supported formats."""
# https://developers.openai.com/api/docs/guides/speech-to-text#transcriptions
return [stt.AudioFormats.WAV, stt.AudioFormats.OGG]
@property
@override
def supported_codecs(self) -> list[stt.AudioCodecs]:
"""Return a list of supported codecs."""
return [stt.AudioCodecs.PCM, stt.AudioCodecs.OPUS]
@property
@override
def supported_bit_rates(self) -> list[stt.AudioBitRates]:
"""Return a list of supported bit rates."""
return [
stt.AudioBitRates.BITRATE_8,
stt.AudioBitRates.BITRATE_16,
stt.AudioBitRates.BITRATE_24,
stt.AudioBitRates.BITRATE_32,
]
@property
@override
def supported_sample_rates(self) -> list[stt.AudioSampleRates]:
"""Return a list of supported sample rates."""
return [
stt.AudioSampleRates.SAMPLERATE_8000,
stt.AudioSampleRates.SAMPLERATE_11000,
stt.AudioSampleRates.SAMPLERATE_16000,
stt.AudioSampleRates.SAMPLERATE_18900,
stt.AudioSampleRates.SAMPLERATE_22000,
stt.AudioSampleRates.SAMPLERATE_32000,
stt.AudioSampleRates.SAMPLERATE_37800,
stt.AudioSampleRates.SAMPLERATE_44100,
stt.AudioSampleRates.SAMPLERATE_48000,
]
@property
@override
def supported_channels(self) -> list[stt.AudioChannels]:
"""Return a list of supported channels."""
return [stt.AudioChannels.CHANNEL_MONO, stt.AudioChannels.CHANNEL_STEREO]
@override
async def async_process_audio_stream(
self, metadata: stt.SpeechMetadata, stream: AsyncIterable[bytes]
) -> stt.SpeechResult:
"""Process an audio stream to STT service."""
audio_bytes = bytearray()
async for chunk in stream:
audio_bytes.extend(chunk)
audio_data = bytes(audio_bytes)
if metadata.format == stt.AudioFormats.WAV:
# Add missing wav header
wav_buffer = io.BytesIO()
with wave.open(wav_buffer, "wb") as wf:
wf.setnchannels(metadata.channel.value)
wf.setsampwidth(metadata.bit_rate.value // 8)
wf.setframerate(metadata.sample_rate.value)
wf.writeframes(audio_data)
audio_data = wav_buffer.getvalue()
options = self.subentry.data
client = self.entry.runtime_data
try:
response = await client.audio.transcriptions.create(
model=options.get(CONF_CHAT_MODEL, RECOMMENDED_STT_MODEL),
file=(f"a.{metadata.format.value}", audio_data),
response_format="json",
language=metadata.language.split("-")[0],
prompt=options.get(CONF_PROMPT, DEFAULT_STT_PROMPT),
)
except OpenAIError:
_LOGGER.exception("Error during STT")
else:
if response.text:
return stt.SpeechResult(
response.text,
stt.SpeechResultState.SUCCESS,
)
return stt.SpeechResult(None, stt.SpeechResultState.ERROR)
custom_components/openai_compatible/tts.py
+200
View File
@@ -0,0 +1,200 @@
"""Text to speech support for OpenAI."""
from collections.abc import Mapping
import logging
from typing import TYPE_CHECKING, Any, Literal, override
from openai import OpenAIError
from propcache.api import cached_property
from homeassistant.components.tts import (
ATTR_PREFERRED_FORMAT,
ATTR_VOICE,
TextToSpeechEntity,
TtsAudioType,
Voice,
)
from homeassistant.config_entries import ConfigSubentry
from homeassistant.const import CONF_PROMPT
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import CONF_CHAT_MODEL, CONF_TTS_SPEED, RECOMMENDED_TTS_SPEED
from .entity import OpenAIBaseLLMEntity
if TYPE_CHECKING:
from . import OpenAIConfigEntry
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: OpenAIConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up TTS entities."""
for subentry in config_entry.subentries.values():
if subentry.subentry_type != "tts":
continue
async_add_entities(
[OpenAITTSEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class OpenAITTSEntity(TextToSpeechEntity, OpenAIBaseLLMEntity):
"""OpenAI TTS entity."""
_attr_supported_options = [ATTR_VOICE, ATTR_PREFERRED_FORMAT]
# https://platform.openai.com/docs/guides/text-to-speech#supported-languages
# The model may also generate the audio in different
# languages but with lower quality
_attr_supported_languages = [
"af-ZA", # Afrikaans
"ar-SA", # Arabic
"hy-AM", # Armenian
"az-AZ", # Azerbaijani
"be-BY", # Belarusian
"bs-BA", # Bosnian
"bg-BG", # Bulgarian
"ca-ES", # Catalan
"zh-CN", # Chinese (Mandarin)
"hr-HR", # Croatian
"cs-CZ", # Czech
"da-DK", # Danish
"nl-NL", # Dutch
"en-US", # English
"et-EE", # Estonian
"fi-FI", # Finnish
"fr-FR", # French
"gl-ES", # Galician
"de-DE", # German
"el-GR", # Greek
"he-IL", # Hebrew
"hi-IN", # Hindi
"hu-HU", # Hungarian
"is-IS", # Icelandic
"id-ID", # Indonesian
"it-IT", # Italian
"ja-JP", # Japanese
"kn-IN", # Kannada
"kk-KZ", # Kazakh
"ko-KR", # Korean
"lv-LV", # Latvian
"lt-LT", # Lithuanian
"mk-MK", # Macedonian
"ms-MY", # Malay
"mr-IN", # Marathi
"mi-NZ", # Maori
"ne-NP", # Nepali
"no-NO", # Norwegian
"fa-IR", # Persian
"pl-PL", # Polish
"pt-PT", # Portuguese
"ro-RO", # Romanian
"ru-RU", # Russian
"sr-RS", # Serbian
"sk-SK", # Slovak
"sl-SI", # Slovenian
"es-ES", # Spanish
"sw-KE", # Swahili
"sv-SE", # Swedish
"fil-PH", # Tagalog (Filipino)
"ta-IN", # Tamil
"th-TH", # Thai
"tr-TR", # Turkish
"uk-UA", # Ukrainian
"ur-PK", # Urdu
"vi-VN", # Vietnamese
"cy-GB", # Welsh
]
# Unused, but required by base class.
# The models detect the input language automatically.
_attr_default_language = "en-US"
# https://platform.openai.com/docs/guides/text-to-speech#voice-options
_supported_voices = [
Voice(voice.lower(), voice)
for voice in (
"Marin",
"Cedar",
"Alloy",
"Ash",
"Ballad",
"Coral",
"Echo",
"Fable",
"Nova",
"Onyx",
"Sage",
"Shimmer",
"Verse",
)
]
_supported_formats = ["mp3", "opus", "aac", "flac", "wav", "pcm"]
_attr_has_entity_name = False
def __init__(self, entry: OpenAIConfigEntry, subentry: ConfigSubentry) -> None:
"""Initialize the entity."""
super().__init__(entry, subentry)
self._attr_name = subentry.title
@callback
@override
def async_get_supported_voices(self, language: str) -> list[Voice]:
"""Return a list of supported voices for a language."""
return self._supported_voices
@cached_property
@override
def default_options(self) -> Mapping[str, Any]:
"""Return a mapping with the default options."""
return {
ATTR_VOICE: self._supported_voices[0].voice_id,
ATTR_PREFERRED_FORMAT: "mp3",
}
@override
async def async_get_tts_audio(
self, message: str, language: str, options: dict[str, Any]
) -> TtsAudioType:
"""Load tts audio file from the engine."""
options = {**self.subentry.data, **options}
client = self.entry.runtime_data
response_format = options[ATTR_PREFERRED_FORMAT]
if response_format in ("ogg", "oga"):
codec: Literal["mp3", "opus", "aac", "flac", "wav", "pcm"] = "opus"
elif response_format == "raw":
response_format = codec = "pcm"
elif response_format not in self._supported_formats:
response_format = self.default_options[ATTR_PREFERRED_FORMAT]
codec = response_format
else:
codec = response_format
try:
async with client.audio.speech.with_streaming_response.create(
model=options[CONF_CHAT_MODEL],
voice=options[ATTR_VOICE],
input=message,
instructions=str(options.get(CONF_PROMPT)),
speed=options.get(CONF_TTS_SPEED, RECOMMENDED_TTS_SPEED),
response_format=codec,
) as response:
response_data = bytearray()
async for chunk in response.iter_bytes():
response_data.extend(chunk)
except OpenAIError as exc:
_LOGGER.exception("Error during TTS")
raise HomeAssistantError(exc) from exc
return response_format, bytes(response_data)
tests/test_init.py
+18
View File
@@ -0,0 +1,18 @@
"""The vendored integration must set up as a custom component."""
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.openai_compatible.const import DOMAIN
def test_domain_renamed():
assert DOMAIN == "openai_compatible"
async def test_config_entry_loads(hass: HomeAssistant) -> None:
"""A minimal entry should reach LOADED."""
entry = MockConfigEntry(domain=DOMAIN, data={"api_key": "sk-test"})
entry.add_to_hass(hass)
# Setup contacts the API; assert the integration is at least discoverable.
assert hass.config_entries.async_entries(DOMAIN) == [entry]