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

Add llama_cpp conversation integration (#175580)

Browse Source
This commit is contained in:
Allen Porter
2026-07-04 08:20:10 +02:00
committed by GitHub
parent 24774798f1
commit 3aa56eab27
21 changed files with 2926 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.strict-typing
+1
View File
@@ -347,6 +347,7 @@ homeassistant.components.light.*
homeassistant.components.linkplay.*
homeassistant.components.litejet.*
homeassistant.components.litterrobot.*
homeassistant.components.llama_cpp.*
homeassistant.components.local_ip.*
homeassistant.components.local_todo.*
homeassistant.components.lock.*
CODEOWNERS
Generated
+2
View File
@@ -1026,6 +1026,8 @@ CLAUDE.md @home-assistant/core
/tests/components/litterrobot/ @natekspencer @tkdrob
/homeassistant/components/livisi/ @StefanIacobLivisi @planbnet
/tests/components/livisi/ @StefanIacobLivisi @planbnet
/homeassistant/components/llama_cpp/ @allenporter
/tests/components/llama_cpp/ @allenporter
/homeassistant/components/llm/ @home-assistant/core
/tests/components/llm/ @home-assistant/core
/homeassistant/components/local_calendar/ @allenporter
homeassistant/components/llama_cpp/__init__.py
+60
View File
@@ -0,0 +1,60 @@
"""The llama.cpp integration."""
import logging
import openai
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
)
from .api import async_create_client, async_list_models
_LOGGER = logging.getLogger(__name__)
PLATFORMS = (Platform.CONVERSATION,)
type LlamaCppConfigEntry = ConfigEntry[openai.AsyncOpenAI]
async def async_setup_entry(hass: HomeAssistant, entry: LlamaCppConfigEntry) -> bool:
"""Set up llama.cpp from a config entry."""
client = await async_create_client(hass, entry.data)
# Validate the connection by listing models
try:
await async_list_models(client)
except HomeAssistantError as err:
if err.translation_key == "invalid_auth":
raise ConfigEntryAuthFailed(
translation_domain=err.translation_domain,
translation_key=err.translation_key,
translation_placeholders=err.translation_placeholders,
) from err
raise ConfigEntryNotReady(
translation_domain=err.translation_domain,
translation_key=err.translation_key,
translation_placeholders=err.translation_placeholders,
) 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: LlamaCppConfigEntry) -> bool:
"""Unload llama.cpp."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_update_options(hass: HomeAssistant, entry: LlamaCppConfigEntry) -> None:
"""Update options."""
await hass.config_entries.async_reload(entry.entry_id)
homeassistant/components/llama_cpp/api.py
+192
View File
@@ -0,0 +1,192 @@
"""API client helper for llama.cpp integration.
This module contains thin wrappers around the OpenAI completions APIs used
to simplify Home Assistant integration and configuration. It handles client
setup, model validation, and API error handling.
"""
from collections.abc import Generator, Mapping
from contextlib import contextmanager
import logging
from typing import Any, cast
import openai
from openai._streaming import AsyncStream
from openai.types.chat import (
ChatCompletionChunk,
ChatCompletionMessageParam,
ChatCompletionToolParam,
)
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.httpx_client import get_async_client
from .const import (
CONF_BASE_URL,
DEFAULT_API_KEY,
DEFAULT_MODEL,
DOMAIN,
RECOMMENDED_CHAT_MODELS,
)
_LOGGER = logging.getLogger(__name__)
# Simple prompt to test model basic chat completion capability. We send tools
# to ensure the model and server correctly supports tool calling. We set a
# minimal max_tokens to consume few resources.
_TEST_MESSAGES: list[ChatCompletionMessageParam] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is the capital of France?"},
]
_TEST_TOOLS: list[ChatCompletionToolParam] = [
{
"type": "function",
"function": {
"name": "test_function",
"description": "Test function.",
"parameters": {"type": "object", "properties": {}},
},
}
]
_TEST_MAX_TOKENS = 3
async def async_create_client(
hass: HomeAssistant, config_entry_data: Mapping[str, Any]
) -> openai.AsyncOpenAI:
"""Create a new OpenAI client."""
api_key = config_entry_data.get(CONF_API_KEY) or DEFAULT_API_KEY
client = openai.AsyncOpenAI(
api_key=api_key,
base_url=config_entry_data[CONF_BASE_URL],
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)
return client
async def async_list_models(client: openai.AsyncOpenAI) -> list[str]:
"""Return a list of models supported by the client."""
with api_error_handler():
page = await client.with_options(timeout=10.0).models.list()
return [model.id async for model in page]
async def async_validate_completions(
client: openai.AsyncOpenAI,
model: str,
stream: bool = False,
) -> None:
"""Validate that we can speak to the model over the completions API."""
with api_error_handler():
result = await client.chat.completions.create(
model=model,
messages=_TEST_MESSAGES,
tools=_TEST_TOOLS,
max_tokens=_TEST_MAX_TOKENS,
stream=stream,
)
if stream:
stream_result = cast(AsyncStream[ChatCompletionChunk], result)
async for event in stream_result:
if not event.choices:
continue
if event.choices[0].finish_reason is not None:
continue
def recommended_model(models: list[str] | None) -> str:
"""Return the selected model from user input."""
if not models:
return DEFAULT_MODEL
for model in RECOMMENDED_CHAT_MODELS:
if model in models:
return model
return models[0]
def model_name_to_title(model_id: str) -> str:
"""Convert a model ID into a human-readable title (inverse slugification).
Examples:
- "deepseek-v4-flash" -> "Deepseek V4 Flash"
- "gpt-4" -> "Gpt 4"
- "llama-3.2-3b-instruct" -> "Llama 3.2 3b Instruct"
- "anthropic/claude-fable-5" -> "Anthropic Claude Fable 5"
"""
words = model_id.replace("-", " ").replace("_", " ").replace("/", " ").split()
return " ".join(word.capitalize() for word in words)
def _extract_error_message(err: openai.APIStatusError) -> str:
"""Extract a clean error message from an APIStatusError response or message."""
error_message = ""
if err.response is not None:
try:
json_data = err.response.json()
if isinstance(json_data, dict) and "error" in json_data:
error_message = json_data["error"].get("message") or ""
except ValueError:
pass
return error_message or err.message or str(err)
@contextmanager
def api_error_handler() -> Generator[None]:
"""Context manager to handle API errors and translate them to HomeAssistantErrors."""
try:
yield
except openai.APITimeoutError as err:
_LOGGER.error("Timeout talking to API: %s", err)
error_message = err.message or str(err)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="timeout",
translation_placeholders={"message": error_message},
) from err
except openai.APIConnectionError as err:
_LOGGER.error("Connection error talking to API: %s", err)
error_message = err.message or str(err)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"message": error_message},
) from err
except openai.AuthenticationError as err:
_LOGGER.error("Authentication error talking to API: %s", err)
error_message = _extract_error_message(err)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"message": error_message},
) from err
except openai.APIStatusError as err:
_LOGGER.error("Status error talking to API: %s", err)
error_message = _extract_error_message(err)
if err.status_code == 402:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="quota_exceeded",
translation_placeholders={"message": error_message},
) from err
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="api_error",
translation_placeholders={"message": error_message},
) from err
except openai.OpenAIError as err:
_LOGGER.error("Generic error talking to API: %s", err)
error_message = getattr(err, "message", None) or str(err)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="api_error",
translation_placeholders={"message": error_message},
) from err
homeassistant/components/llama_cpp/config_flow.py
+383
View File
@@ -0,0 +1,383 @@
"""Config flow for llama.cpp integration."""
import logging
from typing import Any, cast, override
import openai
import voluptuous as vol
from homeassistant.config_entries import (
ConfigEntry,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
SubentryFlowResult,
)
from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, CONF_PROMPT
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import llm
from homeassistant.helpers.selector import (
NumberSelector,
NumberSelectorConfig,
SelectOptionDict,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TemplateSelector,
)
from .api import (
async_create_client,
async_list_models,
async_validate_completions,
model_name_to_title,
recommended_model,
)
from .const import (
CONF_BASE_URL,
CONF_CHAT_MODEL,
CONF_MAX_TOKENS,
CONF_RECOMMENDED,
CONF_STREAMING,
CONF_TEMPERATURE,
CONF_TOP_P,
DEFAULT_BASE_URL,
DOMAIN,
LOGGER,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
)
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_BASE_URL, default=DEFAULT_BASE_URL): str,
vol.Optional(CONF_API_KEY): str,
}
)
class LlamaCppConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for llama.cpp."""
VERSION = 1
data: dict[str, Any] | None = None
client: openai.AsyncOpenAI | None = None
models: list[str] | None = None
@override
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
self._async_abort_entries_match(user_input)
try:
self.client = await async_create_client(self.hass, user_input)
self.models = await async_list_models(self.client)
except HomeAssistantError as err:
LOGGER.error("Connection validation failed: %s", err)
errors["base"] = err.translation_key or "unknown"
except Exception: # pylint: disable=broad-except # noqa: BLE001
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self.data = user_input
return await self.async_step_model()
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
async def async_step_model(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle selecting a model."""
assert self.client is not None
assert self.models is not None
assert self.data is not None
errors = {}
if user_input is not None:
model = user_input[CONF_CHAT_MODEL]
try:
await async_validate_completions(
self.client,
model=model,
stream=False,
)
except HomeAssistantError as err:
LOGGER.error("Model completion validation failed: %s", err)
errors["base"] = err.translation_key or "unknown"
else:
stream_support = True
try:
await async_validate_completions(
self.client,
model=model,
stream=True,
)
except HomeAssistantError:
stream_support = False
base_options = {
**user_input,
}
return self.async_create_entry(
title=self.data[CONF_BASE_URL],
data={
**self.data,
CONF_STREAMING: stream_support,
},
subentries=[
{
"subentry_type": "conversation",
"data": {
CONF_RECOMMENDED: True,
CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
**base_options,
},
"title": model_name_to_title(model),
"unique_id": None,
},
],
)
return self.async_show_form(
step_id="model",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Optional(
CONF_CHAT_MODEL,
): SelectSelector(
SelectSelectorConfig(
options=self.models,
translation_key=CONF_CHAT_MODEL,
mode=SelectSelectorMode.DROPDOWN,
custom_value=True,
),
),
}
),
{
CONF_CHAT_MODEL: (user_input or {}).get(
CONF_CHAT_MODEL, recommended_model(self.models)
),
},
),
errors=errors,
)
@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": ConversationSubentryFlowHandler,
}
class ConversationSubentryFlowHandler(ConfigSubentryFlow):
"""Flow for managing conversation subentries."""
last_rendered_recommended = False
options: dict[str, Any] | None = None
models: list[str] | None = None
@property
def _openai_client(self) -> openai.AsyncOpenAI:
"""Return the OpenAI client."""
return cast(openai.AsyncOpenAI, self._get_entry().runtime_data)
async def _get_models(self) -> list[str] | None:
"""Return the list of models."""
if self.models is None:
self.models = await async_list_models(self._openai_client)
return self.models
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> SubentryFlowResult:
"""Add a subentry."""
if self._get_entry().state is not ConfigEntryState.LOADED:
return self.async_abort(reason="entry_not_loaded")
try:
models = await self._get_models()
except HomeAssistantError:
return self.async_abort(reason="cannot_connect")
self.options = {
CONF_RECOMMENDED: True,
CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
CONF_CHAT_MODEL: recommended_model(models),
}
self.last_rendered_recommended = cast(
bool, self.options.get(CONF_RECOMMENDED, False)
)
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."""
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")
if self.options is None:
self.options = self._get_reconfigure_subentry().data.copy()
self.last_rendered_recommended = cast(
bool, self.options.get(CONF_RECOMMENDED, False)
)
try:
models = await self._get_models()
except HomeAssistantError:
return self.async_abort(reason="cannot_connect")
options = self.options
if user_input is not None:
model = user_input[CONF_CHAT_MODEL]
try:
await async_validate_completions(
self._openai_client,
model=model,
stream=self._get_entry().data.get(CONF_STREAMING, False),
)
except HomeAssistantError as err:
LOGGER.error("Model completion validation failed: %s", err)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
llama_cpp_config_option_schema(self.hass, options, models)
),
user_input,
),
errors={"base": err.translation_key or "unknown"},
)
if user_input[CONF_RECOMMENDED] == self.last_rendered_recommended:
if self.source == "user":
return self.async_create_entry(
title=model_name_to_title(user_input[CONF_CHAT_MODEL]),
data=user_input,
)
return self.async_update_and_abort(
self._get_entry(),
self._get_reconfigure_subentry(),
data=user_input,
title=model_name_to_title(user_input[CONF_CHAT_MODEL]),
)
self.last_rendered_recommended = user_input[CONF_RECOMMENDED]
options = {
CONF_RECOMMENDED: user_input[CONF_RECOMMENDED],
CONF_PROMPT: user_input[CONF_PROMPT],
CONF_CHAT_MODEL: user_input[CONF_CHAT_MODEL],
CONF_LLM_HASS_API: user_input.get(CONF_LLM_HASS_API, []),
}
schema = llama_cpp_config_option_schema(self.hass, options, models)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(schema), options
),
)
def llama_cpp_config_option_schema(
hass: HomeAssistant,
options: dict[str, Any],
models: list[str] | None = None,
) -> dict:
"""Return a schema for llama.cpp completion options."""
hass_apis: list[SelectOptionDict] = [
SelectOptionDict(
label=api.name,
value=api.id,
)
for api in llm.async_get_apis(hass)
]
LOGGER.debug("Available LLM APIs: %s", hass_apis)
schema: dict[vol.Required | vol.Optional, Any] = {}
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)),
}
)
schema.update(
{
vol.Optional(
CONF_CHAT_MODEL,
description={"suggested_value": options.get(CONF_CHAT_MODEL)},
default=options.get(CONF_CHAT_MODEL, recommended_model(models)),
): SelectSelector(
SelectSelectorConfig(
options=models or [],
translation_key=CONF_CHAT_MODEL,
mode=SelectSelectorMode.DROPDOWN,
custom_value=True,
),
),
vol.Required(
CONF_RECOMMENDED, default=options.get(CONF_RECOMMENDED, False)
): bool,
}
)
if options.get(CONF_RECOMMENDED):
return schema
schema.update(
{
vol.Optional(
CONF_MAX_TOKENS,
description={"suggested_value": options.get(CONF_MAX_TOKENS)},
default=RECOMMENDED_MAX_TOKENS,
): int,
vol.Optional(
CONF_TOP_P,
description={"suggested_value": options.get(CONF_TOP_P)},
default=RECOMMENDED_TOP_P,
): NumberSelector(NumberSelectorConfig(min=0, max=1, step=0.05)),
vol.Optional(
CONF_TEMPERATURE,
description={"suggested_value": options.get(CONF_TEMPERATURE)},
default=RECOMMENDED_TEMPERATURE,
): NumberSelector(NumberSelectorConfig(min=0, max=2, step=0.05)),
}
)
return schema
homeassistant/components/llama_cpp/const.py
+30
View File
@@ -0,0 +1,30 @@
"""Constants for the llama.cpp integration."""
import logging
DOMAIN = "llama_cpp"
LOGGER = logging.getLogger(__package__)
DEFAULT_CONVERSATION_NAME = "llama.cpp Conversation"
CONF_CHAT_MODEL = "chat_model"
CONF_MAX_TOKENS = "max_tokens"
CONF_TEMPERATURE = "temperature"
CONF_TOP_P = "top_p"
CONF_BASE_URL = "base_url"
CONF_RECOMMENDED = "recommended"
CONF_STREAMING = "streaming"
# Some servers set placeholder model names which we can use as a default
DEFAULT_MODEL = "gpt-3.5-turbo"
RECOMMENDED_CHAT_MODELS = [
DEFAULT_MODEL,
"gpt-4",
"local-model",
]
RECOMMENDED_MAX_TOKENS = 3000
RECOMMENDED_TEMPERATURE = 0.7
RECOMMENDED_TOP_P = 1.0
DEFAULT_BASE_URL = "http://localhost:8080/v1"
DEFAULT_API_KEY = "sk-0000000000000000000"
homeassistant/components/llama_cpp/conversation.py
+83
View File
@@ -0,0 +1,83 @@
"""Conversation support for llama.cpp."""
from typing import Literal, override
from homeassistant.components import conversation
from homeassistant.config_entries import ConfigEntry, 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 LlamaCppConfigEntry
from .const import DOMAIN
from .entity import LlamaCppBaseLLMEntity
async def async_setup_entry(
hass: HomeAssistant,
config_entry: LlamaCppConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up conversation entities."""
for subentry in config_entry.subentries.values():
async_add_entities(
[LlamaCppConversationEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class LlamaCppConversationEntity(
conversation.ConversationEntity,
conversation.AbstractConversationAgent,
LlamaCppBaseLLMEntity,
):
"""llama.cpp conversation agent."""
def __init__(self, entry: ConfigEntry, 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 a sentence."""
options = self.subentry.data
try:
await chat_log.async_provide_llm_data(
user_input.as_llm_context(DOMAIN),
options.get(CONF_LLM_HASS_API),
options.get(CONF_PROMPT),
user_input.extra_system_prompt,
)
except conversation.ConverseError as err:
return err.as_conversation_result()
await self._async_handle_chat_log(chat_log)
return conversation.async_get_result_from_chat_log(user_input, chat_log)
homeassistant/components/llama_cpp/entity.py
+457
View File
@@ -0,0 +1,457 @@
"""Base entity for llama.cpp Conversation."""
import base64
from collections.abc import AsyncGenerator, Callable
import json
import logging
import mimetypes
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
from openai import AsyncOpenAI
from openai._streaming import AsyncStream
from openai._types import Omit
from openai.types.chat import (
ChatCompletion,
ChatCompletionAssistantMessageParam,
ChatCompletionChunk,
ChatCompletionContentPartParam,
ChatCompletionContentPartTextParam,
ChatCompletionFunctionToolParam,
ChatCompletionMessage,
ChatCompletionMessageFunctionToolCall,
ChatCompletionMessageParam,
ChatCompletionMessageToolCallParam,
ChatCompletionSystemMessageParam,
ChatCompletionToolMessageParam,
ChatCompletionUserMessageParam,
)
from openai.types.chat.chat_completion_message_function_tool_call_param import Function
from openai.types.shared_params import FunctionDefinition, ResponseFormatJSONSchema
import voluptuous as vol
from voluptuous_openapi import convert
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, llm
from homeassistant.helpers.entity import Entity
from .api import api_error_handler
from .const import (
CONF_CHAT_MODEL,
CONF_MAX_TOKENS,
CONF_STREAMING,
CONF_TEMPERATURE,
CONF_TOP_P,
DEFAULT_MODEL,
DOMAIN,
LOGGER,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
)
if TYPE_CHECKING:
from . import LlamaCppConfigEntry
# Max number of back and forth with the LLM to generate a response
MAX_TOOL_ITERATIONS = 10
_LOGGER = logging.getLogger(__name__)
def _format_structured_output(
name: str, structure: vol.Schema, llm_api: llm.APIInstance | None
) -> ResponseFormatJSONSchema:
"""Format structured output specification."""
schema = convert(
structure, custom_serializer=llm_api.custom_serializer if llm_api else None
)
return ResponseFormatJSONSchema(
type="json_schema",
json_schema={
"name": name,
"strict": True,
"schema": cast(dict[str, object], schema),
},
)
def _format_tool(
tool: llm.Tool,
custom_serializer: Callable[[Any], Any] | None,
) -> ChatCompletionFunctionToolParam:
"""Format tool specification."""
tool_spec = FunctionDefinition(
name=tool.name,
parameters=convert(tool.parameters, custom_serializer=custom_serializer),
)
if tool.description:
tool_spec["description"] = tool.description
return ChatCompletionFunctionToolParam(type="function", function=tool_spec)
def _convert_content_to_chat_message(
content: conversation.Content,
) -> ChatCompletionMessageParam | None:
"""Convert any native chat message for this agent to the native format."""
_LOGGER.debug("_convert_content_to_chat_message=%s", content)
if isinstance(content, conversation.ToolResultContent):
return ChatCompletionToolMessageParam(
role="tool",
tool_call_id=content.tool_call_id,
content=json.dumps(content.tool_result),
)
role: Literal["user", "assistant", "system"] = content.role
if role == "system" and content.content:
return ChatCompletionSystemMessageParam(role="system", content=content.content)
if role == "user" and content.content:
return ChatCompletionUserMessageParam(role="user", content=content.content)
if role == "assistant":
param = ChatCompletionAssistantMessageParam(
role="assistant",
content=content.content,
)
if isinstance(content, conversation.AssistantContent) and content.tool_calls:
param["tool_calls"] = [
ChatCompletionMessageToolCallParam(
type="function",
id=tool_call.id,
function=Function(
arguments=json.dumps(tool_call.tool_args),
name=tool_call.tool_name,
),
)
for tool_call in content.tool_calls
]
return param
LOGGER.warning("Could not convert message to OpenAI API: %s", content)
return None
def _decode_tool_arguments(arguments: str) -> Any:
"""Decode tool call arguments."""
try:
return json.loads(arguments)
except json.JSONDecodeError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="json_parse_error",
translation_placeholders={"message": str(err)},
) from err
async def _transform_response(
message: ChatCompletionMessage,
) -> AsyncGenerator[conversation.AssistantContentDeltaDict]:
"""Transform the OpenAI API message to a ChatLog format."""
data: conversation.AssistantContentDeltaDict = {
"role": message.role,
"content": message.content,
}
if message.tool_calls:
data["tool_calls"] = [
llm.ToolInput(
id=tool_call.id,
tool_name=tool_call.function.name,
tool_args=_decode_tool_arguments(tool_call.function.arguments),
)
for tool_call in message.tool_calls
if isinstance(tool_call, ChatCompletionMessageFunctionToolCall)
]
yield data
def _convert_content_to_param(
content: conversation.Content,
) -> ChatCompletionMessageParam:
"""Convert any native chat message for this agent to the native format."""
if isinstance(content, conversation.ToolResultContent):
return ChatCompletionToolMessageParam(
role="tool",
tool_call_id=content.tool_call_id,
content=json.dumps(content.tool_result),
)
if not isinstance(content, conversation.AssistantContent) or not content.tool_calls:
if isinstance(content, conversation.SystemContent):
return ChatCompletionSystemMessageParam(
role="system",
content=content.content or "",
)
return cast(
ChatCompletionMessageParam,
{"role": content.role, "content": content.content or ""},
)
return ChatCompletionAssistantMessageParam(
role="assistant",
content=content.content,
tool_calls=[
ChatCompletionMessageToolCallParam(
id=tool_call.id,
function=Function(
arguments=json.dumps(tool_call.tool_args),
name=tool_call.tool_name,
),
type="function",
)
for tool_call in content.tool_calls
],
)
async def _transform_stream(
result: AsyncStream[ChatCompletionChunk],
) -> AsyncGenerator[conversation.AssistantContentDeltaDict]:
"""Transform an OpenAI delta stream into HA format."""
current_tool_call: dict[str, Any] | None = None
yielded_role = False
async for chunk in result:
LOGGER.debug("Received chunk: %s", chunk)
if not chunk.choices:
continue
choice = chunk.choices[0]
if choice.finish_reason:
if current_tool_call:
yield {
"tool_calls": [
llm.ToolInput(
id=current_tool_call["id"],
tool_name=current_tool_call["tool_name"],
tool_args=_decode_tool_arguments(
current_tool_call["tool_args"]
)
if current_tool_call["tool_args"]
else {},
)
]
}
break
delta = choice.delta
if current_tool_call is None and not delta.tool_calls:
yield_dict: conversation.AssistantContentDeltaDict = {}
if not yielded_role and delta.role == "assistant":
yield_dict["role"] = "assistant"
yielded_role = True
if delta.content is not None:
yield_dict["content"] = delta.content
if yield_dict:
yield yield_dict
continue
if (
not delta.tool_calls
or not (delta_tool_call := delta.tool_calls[0])
or not delta_tool_call.function
):
continue
if current_tool_call and delta_tool_call.index == current_tool_call["index"]:
current_tool_call["tool_args"] += delta_tool_call.function.arguments or ""
continue
if current_tool_call:
yield {
"tool_calls": [
llm.ToolInput(
id=current_tool_call["id"],
tool_name=current_tool_call["tool_name"],
tool_args=_decode_tool_arguments(
current_tool_call["tool_args"]
),
)
]
}
current_tool_call = {
"index": delta_tool_call.index,
"id": delta_tool_call.id,
"tool_name": delta_tool_call.function.name,
"tool_args": delta_tool_call.function.arguments or "",
}
class LlamaCppBaseLLMEntity(Entity):
"""llama.cpp base LLM entity."""
_attr_has_entity_name = True
_attr_name = None
def __init__(self, entry: LlamaCppConfigEntry, 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="llama.cpp",
model=subentry.data.get(CONF_CHAT_MODEL, DEFAULT_MODEL),
entry_type=dr.DeviceEntryType.SERVICE,
)
async def _async_handle_chat_log(
self,
chat_log: conversation.ChatLog,
structure_name: str | None = None,
structure: vol.Schema | None = None,
) -> None:
"""Generate an answer for the chat log."""
options = self.subentry.data
tools: list[ChatCompletionFunctionToolParam] | None = None
if chat_log.llm_api:
tools = [
_format_tool(tool, chat_log.llm_api.custom_serializer)
for tool in chat_log.llm_api.tools
]
model: str = options.get(CONF_CHAT_MODEL, DEFAULT_MODEL)
messages = [
m
for content in chat_log.content
if (m := _convert_content_to_chat_message(content))
]
response_format: ResponseFormatJSONSchema | Omit = Omit()
if structure and structure_name:
response_format = _format_structured_output(
structure_name, structure, chat_log.llm_api
)
last_content = chat_log.content[-1]
if (
isinstance(last_content, conversation.UserContent)
and last_content.attachments
):
files = await async_prepare_files_for_prompt(
self.hass,
[a.path for a in last_content.attachments],
)
for i in range(len(messages) - 1, -1, -1):
if messages[i]["role"] == "user":
user_msg = cast(ChatCompletionUserMessageParam, messages[i])
current_content = user_msg.get("content")
if isinstance(current_content, str):
user_msg["content"] = [
ChatCompletionContentPartTextParam(
type="text", text=current_content
),
*files,
]
break
client: AsyncOpenAI = self.entry.runtime_data
streaming = bool(
self.entry.data.get(CONF_STREAMING, options.get(CONF_STREAMING, False))
)
for _iteration in range(MAX_TOOL_ITERATIONS):
with api_error_handler():
result = await client.chat.completions.create(
messages=messages,
model=model,
tools=tools or Omit(),
response_format=response_format,
max_tokens=cast(
int, options.get(CONF_MAX_TOKENS, RECOMMENDED_MAX_TOKENS)
),
top_p=cast(float, options.get(CONF_TOP_P, RECOMMENDED_TOP_P)),
temperature=cast(
float, options.get(CONF_TEMPERATURE, RECOMMENDED_TEMPERATURE)
),
user=chat_log.conversation_id,
stream=cast(Any, streaming),
)
convert_message: Callable[[Any], Any]
async_generator: AsyncGenerator[conversation.AssistantContentDeltaDict]
if streaming:
convert_message = _convert_content_to_param
async_generator = _transform_stream(
cast(AsyncStream[ChatCompletionChunk], result)
)
else:
convert_message = _convert_content_to_chat_message
async_generator = _transform_response(
cast(ChatCompletion, result).choices[0].message
)
messages.extend(
[
msg
async for content in chat_log.async_add_delta_content_stream(
self.entity_id, async_generator
)
if (msg := convert_message(content))
]
)
if not chat_log.unresponded_tool_results:
break
async def async_prepare_files_for_prompt(
hass: HomeAssistant, files: list[Path]
) -> list[ChatCompletionContentPartParam]:
"""Prepare files for OpenAI-compatible API.
Caller needs to ensure that the files are allowed.
"""
def guess_file_type(file_path: Path) -> tuple[str | None, str | None]:
"""Guess the file type based on the file extension."""
return mimetypes.guess_type(str(file_path))
def append_files_to_content() -> list[ChatCompletionContentPartParam]:
content: list[ChatCompletionContentPartParam] = []
for file_path in files:
if not file_path.exists():
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="file_not_found",
translation_placeholders={"file_path": str(file_path)},
)
mime_type, _ = guess_file_type(file_path)
if not mime_type or not mime_type.startswith(("image/", "application/pdf")):
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unsupported_file_type",
translation_placeholders={"file_path": str(file_path)},
)
base64_file = base64.b64encode(file_path.read_bytes()).decode("utf-8")
if mime_type.startswith("image/"):
content.append(
{
"type": "image_url",
"image_url": {
"url": f"data:{mime_type};base64,{base64_file}",
"detail": "auto",
},
}
)
elif mime_type.startswith("application/pdf"):
content.append(
{
"type": "text",
"text": f"[File: {file_path.name}]\nContent: {base64_file}",
}
)
return content
return await hass.async_add_executor_job(append_files_to_content)
homeassistant/components/llama_cpp/manifest.json
+13
View File
@@ -0,0 +1,13 @@
{
"domain": "llama_cpp",
"name": "llama.cpp",
"after_dependencies": ["assist_pipeline", "intent"],
"codeowners": ["@allenporter"],
"config_flow": true,
"dependencies": ["conversation"],
"documentation": "https://www.home-assistant.io/integrations/llama_cpp",
"integration_type": "service",
"iot_class": "local_polling",
"quality_scale": "bronze",
"requirements": ["openai==2.21.0"]
}
homeassistant/components/llama_cpp/quality_scale.yaml
+104
View File
@@ -0,0 +1,104 @@
rules:
# Bronze
action-setup:
status: exempt
comment: No service actions are registered by this integration.
appropriate-polling:
status: exempt
comment: The integration does not poll and is push-based.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: No service actions are registered by this integration.
docs-conditions:
status: exempt
comment: No custom conditions are supported by this integration.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
docs-triggers:
status: exempt
comment: No custom triggers are supported by this integration.
entity-event-setup:
status: exempt
comment: No event entities or helper events are supported by this integration.
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:
status: exempt
comment: No service actions are registered by this integration.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable:
status: exempt
comment: Conversation entities do not have an unavailable state.
integration-owner: done
log-when-unavailable:
status: exempt
comment: Conversation entities do not have an unavailable state.
parallel-updates:
status: exempt
comment: No periodic updates are performed by this integration.
reauthentication-flow: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: The integration does not support discovery.
discovery:
status: exempt
comment: The integration does not support discovery.
docs-data-update:
status: exempt
comment: No periodic data updates are performed by this integration.
docs-examples: done
docs-known-limitations: done
docs-supported-devices:
status: exempt
comment: The integration does not support physical devices.
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices:
status: exempt
comment: No physical devices are supported.
entity-category:
status: exempt
comment: Conversation entity does not require an entity category.
entity-device-class:
status: exempt
comment: Conversation entity does not require a device class.
entity-disabled-by-default:
status: exempt
comment: Conversation entity should be enabled by default.
entity-translations: done
exception-translations: done
icon-translations:
status: exempt
comment: No icons are defined for this integration.
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: No repair issues are defined for this integration.
stale-devices:
status: exempt
comment: No physical devices are supported.
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done
homeassistant/components/llama_cpp/strings.json
+98
View File
@@ -0,0 +1,98 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
},
"error": {
"api_error": "[%key:common::config_flow::error::unknown%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"quota_exceeded": "Your account or API key has insufficient credits.",
"timeout": "Connection timed out.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"model": {
"data": {
"chat_model": "[%key:common::generic::model%]"
},
"data_description": {
"chat_model": "Select the model to use."
}
},
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"base_url": "URL"
},
"data_description": {
"api_key": "API key for the server (optional).",
"base_url": "Base URL of your running OpenAI-compatible server (e.g. http://localhost:8080/v1)."
}
}
}
},
"config_subentries": {
"conversation": {
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"entry_not_loaded": "Cannot add things while the configuration is disabled.",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"entry_type": "Conversation agent",
"initiate_flow": {
"user": "Add conversation agent"
},
"step": {
"init": {
"data": {
"chat_model": "[%key:common::generic::model%]",
"llm_hass_api": "Control Home Assistant",
"max_tokens": "Maximum tokens to return in response",
"name": "[%key:common::config_flow::data::name%]",
"prompt": "Instructions",
"recommended": "Recommended model settings",
"temperature": "Temperature",
"top_p": "Top P"
},
"data_description": {
"chat_model": "Select the model to use.",
"llm_hass_api": "Select the level of control over Home Assistant.",
"max_tokens": "Select the maximum number of tokens to return.",
"prompt": "Instruct how the LLM should respond. This can be a template.",
"recommended": "Select whether to use recommended model settings.",
"temperature": "Select the temperature for response variability.",
"top_p": "Select the top P value for response diversity."
}
}
}
}
},
"exceptions": {
"api_error": {
"message": "API error: {message}."
},
"cannot_connect": {
"message": "Cannot connect to the server: {message}."
},
"file_not_found": {
"message": "File does not exist: {file_path}."
},
"invalid_auth": {
"message": "Invalid authentication: {message}."
},
"json_parse_error": {
"message": "Unexpected tool argument response: {message}."
},
"quota_exceeded": {
"message": "Your account or API key has insufficient credits: {message}."
},
"timeout": {
"message": "Connection timed out: {message}."
},
"unsupported_file_type": {
"message": "Only images and PDF are supported by the OpenAI API, {file_path} is not an image file or PDF."
}
}
}
homeassistant/generated/config_flows.py
Generated
+1
View File
@@ -424,6 +424,7 @@ FLOWS = {
"litejet",
"litterrobot",
"livisi",
"llama_cpp",
"local_calendar",
"local_file",
"local_ip",
homeassistant/generated/integrations.json
+6
View File
@@ -3887,6 +3887,12 @@
"config_flow": true,
"iot_class": "local_polling"
},
"llama_cpp": {
"name": "llama.cpp",
"integration_type": "service",
"config_flow": true,
"iot_class": "local_polling"
},
"llamalab_automate": {
"name": "LlamaLab Automate",
"integration_type": "hub",
mypy.ini
Generated
+10
View File
@@ -3227,6 +3227,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.llama_cpp.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.local_ip.*]
check_untyped_defs = true
disallow_incomplete_defs = true
requirements_all.txt
Generated
+1
View File
@@ -1752,6 +1752,7 @@ open-garage==0.2.0
open-meteo==0.3.2
# homeassistant.components.cloud
# homeassistant.components.llama_cpp
# homeassistant.components.open_router
# homeassistant.components.openai_conversation
# homeassistant.components.ovhcloud_ai_endpoints
tests/components/llama_cpp/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Tests for the llama.cpp integration."""
tests/components/llama_cpp/conftest.py
+167
View File
@@ -0,0 +1,167 @@
"""Fixtures for llama.cpp integration tests."""
from collections.abc import AsyncGenerator, Generator
from dataclasses import dataclass, field
import logging
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components import conversation
from homeassistant.components.llama_cpp.const import (
CONF_BASE_URL,
DEFAULT_BASE_URL,
DEFAULT_CONVERSATION_NAME,
DOMAIN,
)
from homeassistant.config_entries import ConfigSubentryData
from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import chat_session, llm
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
_LOGGER = logging.getLogger(__name__)
CONFIG_ENTRY_DATA = {
CONF_API_KEY: "sk-0000000000000000000",
CONF_BASE_URL: DEFAULT_BASE_URL,
}
ASSIST_OPTIONS = {CONF_LLM_HASS_API: llm.LLM_API_ASSIST}
@pytest.fixture(autouse=True)
async def setup_home_assistant(hass: HomeAssistant) -> None:
"""Enable dependencies."""
assert await async_setup_component(hass, "homeassistant", {})
@pytest.fixture(name="platforms")
def mock_platforms() -> list[Platform]:
"""Fixture for platforms loaded by the integration."""
return []
@pytest.fixture(name="setup_integration")
async def mock_setup_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
platforms: list[Platform],
) -> AsyncGenerator[None]:
"""Set up the integration."""
with patch(f"homeassistant.components.{DOMAIN}.PLATFORMS", platforms):
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
yield
@pytest.fixture(name="config_entry_data")
def config_entry_data_fixture() -> dict[str, Any]:
"""Fixture to add data to the config entry."""
return {}
@pytest.fixture(name="config_entry_options")
def config_entry_options_fixture() -> dict[str, Any]:
"""Fixture to add options to the config entry."""
return {}
@pytest.fixture(name="mock_config_entry")
def mock_config_entry_fixture(
hass: HomeAssistant,
config_entry_data: dict[str, Any],
config_entry_options: dict[str, Any],
) -> MockConfigEntry:
"""Fixture to create a configuration entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="llama.cpp",
data={
**CONFIG_ENTRY_DATA,
**config_entry_data,
},
version=1,
minor_version=1,
subentries_data=[
ConfigSubentryData(
data={**config_entry_options},
subentry_type="conversation",
title=DEFAULT_CONVERSATION_NAME,
unique_id=None,
),
],
)
config_entry.add_to_hass(hass)
return config_entry
@dataclass
class MockChatLog(conversation.ChatLog):
"""Mock chat log."""
_mock_tool_results: dict[str, Any] = field(default_factory=dict)
def mock_tool_results(self, results: dict[str, Any]) -> None:
"""Set tool results."""
self._mock_tool_results = results
@property
def llm_api(self) -> llm.APIInstance | None:
"""Return LLM API."""
return self._llm_api
@llm_api.setter
def llm_api(self, value: llm.APIInstance | None) -> None:
"""Set LLM API."""
self._llm_api = value
if not value:
return
async def async_call_tool(tool_input: llm.ToolInput) -> llm.ToolResult:
"""Call tool."""
if tool_input.id not in self._mock_tool_results:
raise ValueError(
f"Tool {tool_input.id} not found ({self._mock_tool_results})"
)
return self._mock_tool_results[tool_input.id]
self._llm_api.async_call_tool = async_call_tool
@pytest.fixture
def mock_chat_log(hass: HomeAssistant) -> Generator[conversation.ChatLog]:
"""Return mock chat logs."""
# pylint: disable-next=contextmanager-generator-missing-cleanup
with (
patch(
"homeassistant.components.conversation.chat_log.ChatLog",
MockChatLog,
),
chat_session.async_get_chat_session(hass, "mock-conversation-id") as session,
conversation.async_get_chat_log(hass, session) as chat_log,
):
yield chat_log
@pytest.fixture(autouse=True)
def mock_models_list() -> Generator[AsyncMock]:
"""Initialize integration."""
with patch(
"openai.resources.models.AsyncModels.list",
new_callable=AsyncMock,
) as mock_list:
yield mock_list
@pytest.fixture(name="mock_completion", autouse=True)
def mock_openai_client_fixture() -> Generator[AsyncMock]:
"""Fixture to mock the OpenAI client."""
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
) as mock_create:
yield mock_create
tests/components/llama_cpp/snapshots/test_conversation.ambr
+92
View File
@@ -0,0 +1,92 @@
# serializer version: 1
# name: test_conversation_entity
list([
dict({
'attachments': None,
'content': 'hello',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.llama_cpp_conversation',
'content': 'Hello, how can I help you?',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
])
# ---
# name: test_function_call[config_entry_options0]
list([
dict({
'attachments': None,
'content': 'Please call the test function',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.llama_cpp_conversation',
'content': None,
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': list([
dict({
'external': False,
'id': 'call_call_1',
'tool_args': dict({
'param1': 'call1',
}),
'tool_name': 'test_tool',
}),
]),
}),
dict({
'agent_id': 'conversation.llama_cpp_conversation',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'tool_result',
'tool_call_id': 'call_call_1',
'tool_name': 'test_tool',
'tool_result': 'value1',
}),
dict({
'agent_id': 'conversation.llama_cpp_conversation',
'content': 'I have successfully called the function',
'created': HAFakeDatetime(2024, 5, 24, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
])
# ---
# name: test_function_exception[-config_entry_options0]
'Unexpected tool argument response: Expecting value: line 1 column 1 (char 0)'
# ---
# name: test_function_exception[{"para-config_entry_options0]
'Unexpected tool argument response: Unterminated string starting at: line 1 column 2 (char 1)'
# ---
# name: test_unknown_hass_api[config_entry_options0]
dict({
'continue_conversation': False,
'conversation_id': <ANY>,
'response': dict({
'card': dict({
}),
'data': dict({
'code': 'unknown',
}),
'language': 'en',
'response_type': 'error',
'speech': dict({
'plain': dict({
'extra_data': None,
'speech': 'Error preparing LLM API',
}),
}),
}),
})
# ---
tests/components/llama_cpp/test_config_flow.py
+585
View File
@@ -0,0 +1,585 @@
"""Tests for the llama.cpp config flow."""
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import openai
import pytest
from homeassistant import config_entries
from homeassistant.components.llama_cpp.const import (
CONF_BASE_URL,
CONF_CHAT_MODEL,
CONF_MAX_TOKENS,
CONF_RECOMMENDED,
CONF_STREAMING,
CONF_TEMPERATURE,
CONF_TOP_P,
DEFAULT_MODEL,
DOMAIN,
)
from homeassistant.const import CONF_API_KEY, CONF_LLM_HASS_API, CONF_PROMPT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import llm
from tests.common import MockConfigEntry
RECOMMENDED_OPTIONS = {
CONF_RECOMMENDED: True,
CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
CONF_CHAT_MODEL: DEFAULT_MODEL,
}
@pytest.fixture(name="mock_setup")
def mock_setup(hass: HomeAssistant) -> Generator[AsyncMock]:
"""Mock the setup of the integration."""
with patch(
f"homeassistant.components.{DOMAIN}.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
async def test_config_flow(
hass: HomeAssistant,
mock_setup: AsyncMock,
) -> None:
"""Test selecting a model in the configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert not result.get("errors")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "sk-0000000000000000000",
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result.get("type") is FlowResultType.FORM
assert not result.get("errors")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_CHAT_MODEL: "gpt-4",
},
)
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "http://localhost:8080/v1"
assert result.get("data") == {
CONF_API_KEY: "sk-0000000000000000000",
CONF_BASE_URL: "http://localhost:8080/v1",
CONF_STREAMING: True,
}
assert result["options"] == {}
assert result["subentries"] == [
{
"subentry_type": "conversation",
"data": {
**RECOMMENDED_OPTIONS,
CONF_CHAT_MODEL: "gpt-4",
},
"title": "Gpt 4",
"unique_id": None,
},
]
assert len(mock_setup.mock_calls) == 1
@pytest.mark.parametrize(
("side_effect", "expected_error"),
[
(
openai.APIConnectionError(request=httpx.Request(method="POST", url="test")),
"cannot_connect",
),
(
openai.AuthenticationError(
message="Invalid key",
response=httpx.Response(
status_code=401,
request=httpx.Request(method="POST", url="test"),
),
body=None,
),
"invalid_auth",
),
(
openai.OpenAIError("Generic error"),
"api_error",
),
],
)
async def test_config_flow_fail_completion(
hass: HomeAssistant,
mock_setup: AsyncMock,
mock_completion: AsyncMock,
side_effect: Exception,
expected_error: str,
) -> None:
"""Test config flow where the API completion validation fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert not result.get("errors")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "sk-0000000000000000000",
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result.get("type") is FlowResultType.FORM
assert not result.get("errors")
mock_completion.side_effect = side_effect
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_CHAT_MODEL: "gpt-4",
},
)
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {"base": expected_error}
assert len(mock_setup.mock_calls) == 0
async def test_config_flow_no_streaming(
hass: HomeAssistant,
mock_setup: AsyncMock,
mock_completion: AsyncMock,
) -> None:
"""Test config flow where the API does not support streaming."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert not result.get("errors")
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_API_KEY: "sk-0000000000000000000",
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result.get("type") is FlowResultType.FORM
assert not result.get("errors")
def fail_streaming(stream: bool | None = None, **kwargs: Any) -> None:
"""Allow first check to succeed by fail streaming."""
if stream:
raise openai.OpenAIError("Invalid request")
mock_completion.side_effect = fail_streaming
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_CHAT_MODEL: "gpt-4",
},
)
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "http://localhost:8080/v1"
assert result.get("data") == {
CONF_API_KEY: "sk-0000000000000000000",
CONF_BASE_URL: "http://localhost:8080/v1",
CONF_STREAMING: False,
}
assert result["subentries"] == [
{
"subentry_type": "conversation",
"data": {
**RECOMMENDED_OPTIONS,
CONF_CHAT_MODEL: "gpt-4",
},
"title": "Gpt 4",
"unique_id": None,
},
]
assert len(mock_setup.mock_calls) == 1
@pytest.mark.usefixtures("setup_integration")
async def test_creating_conversation_subentry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a conversation subentry."""
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert not result["errors"]
result2 = await hass.config_entries.subentries.async_configure(
result["flow_id"],
RECOMMENDED_OPTIONS,
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "Gpt 3.5 Turbo"
assert result2["data"] == RECOMMENDED_OPTIONS
@pytest.mark.usefixtures("setup_integration")
async def test_creating_conversation_subentry_not_loaded(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a conversation subentry when entry is not loaded."""
await hass.config_entries.async_unload(mock_config_entry.entry_id)
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
return_value=[],
):
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
@pytest.mark.usefixtures("setup_integration")
async def test_creating_conversation_subentry_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a conversation subentry handles connection errors."""
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
side_effect=openai.APIConnectionError(request=None),
):
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@pytest.mark.usefixtures("setup_integration")
async def test_creating_conversation_subentry_advanced(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a conversation subentry with custom/advanced settings."""
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
# Toggle recommended to False to show advanced options
result2 = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_RECOMMENDED: False,
CONF_CHAT_MODEL: "gpt-4",
CONF_PROMPT: "Custom instructions",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "init"
# Now configure the advanced options
result3 = await hass.config_entries.subentries.async_configure(
result2["flow_id"],
{
CONF_RECOMMENDED: False,
CONF_CHAT_MODEL: "gpt-4",
CONF_PROMPT: "Custom instructions",
CONF_MAX_TOKENS: 500,
CONF_TEMPERATURE: 0.5,
CONF_TOP_P: 0.9,
},
)
await hass.async_block_till_done()
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["title"] == "Gpt 4"
assert result3["data"] == {
CONF_RECOMMENDED: False,
CONF_CHAT_MODEL: "gpt-4",
CONF_PROMPT: "Custom instructions",
CONF_MAX_TOKENS: 500,
CONF_TEMPERATURE: 0.5,
CONF_TOP_P: 0.9,
}
async def test_config_flow_model_selection_fallbacks(
hass: HomeAssistant,
mock_setup: AsyncMock,
) -> None:
"""Test model selection fallback options through the config flow."""
# 1. Test empty list fallback (should fallback to DEFAULT_MODEL)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
async def mock_empty_list(*args, **kwargs):
return
yield
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
side_effect=mock_empty_list,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "model"
schema = result2["data_schema"].schema
chat_model_key = next(k for k in schema if k == CONF_CHAT_MODEL)
assert chat_model_key.description["suggested_value"] == DEFAULT_MODEL
# 2. Test no recommended models match fallback (should select first model in the list)
result_custom = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
model1 = MagicMock()
model1.id = "my-custom-model-1"
model2 = MagicMock()
model2.id = "my-custom-model-2"
async def mock_custom_list(*args, **kwargs):
yield model1
yield model2
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
side_effect=mock_custom_list,
):
result3 = await hass.config_entries.flow.async_configure(
result_custom["flow_id"],
{
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result3["type"] is FlowResultType.FORM
assert result3["step_id"] == "model"
schema = result3["data_schema"].schema
chat_model_key = next(k for k in schema if k == CONF_CHAT_MODEL)
assert chat_model_key.description["suggested_value"] == "my-custom-model-1"
async def test_config_flow_connection_errors(
hass: HomeAssistant,
mock_setup: AsyncMock,
) -> None:
"""Test config flow handles connection validation errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
# 1. Test AuthenticationError
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
side_effect=openai.AuthenticationError(
message="Invalid Key",
response=httpx.Response(
status_code=401,
request=httpx.Request(method="GET", url="test"),
),
body=None,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
# 2. Test APIConnectionError
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
side_effect=openai.APIConnectionError(
request=httpx.Request(method="GET", url="test")
),
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {"base": "cannot_connect"}
# 3. Test OpenAIError (Generic API errors)
with patch(
"homeassistant.components.llama_cpp.config_flow.openai.resources.models.AsyncModels.list",
side_effect=openai.OpenAIError("generic error"),
):
result4 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result4["type"] is FlowResultType.FORM
assert result4["errors"] == {"base": "api_error"}
@pytest.mark.usefixtures("setup_integration")
async def test_reconfiguring_conversation_subentry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring an existing conversation subentry."""
subentry = list(mock_config_entry.subentries.values())[0]
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": "reconfigure", "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result2 = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_RECOMMENDED: False,
CONF_CHAT_MODEL: "gpt-4",
CONF_PROMPT: "New prompt",
},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reconfigure_successful"
updated_subentry = list(mock_config_entry.subentries.values())[0]
assert updated_subentry.title == "Gpt 4"
assert updated_subentry.data[CONF_CHAT_MODEL] == "gpt-4"
assert updated_subentry.data[CONF_PROMPT] == "New prompt"
assert CONF_STREAMING not in updated_subentry.data
async def test_subentry_options_entry_not_loaded(
hass: HomeAssistant,
setup_integration: None,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test options flow aborts if config entry is not loaded."""
subentry = list(mock_config_entry.subentries.values())[0]
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": "reconfigure", "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "entry_not_loaded"
async def test_reconfiguring_conversation_subentry_connection_error(
hass: HomeAssistant,
setup_integration: None,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring subentry aborts if model listing fails."""
subentry = list(mock_config_entry.subentries.values())[0]
with patch(
"openai.resources.models.AsyncModels.list",
side_effect=openai.APIConnectionError(request=None),
):
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": "reconfigure", "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
async def test_reconfiguring_conversation_subentry_validation_error(
hass: HomeAssistant,
setup_integration: None,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reconfiguring subentry shows form with error if model validation fails."""
subentry = list(mock_config_entry.subentries.values())[0]
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "conversation"),
context={"source": "reconfigure", "subentry_id": subentry.subentry_id},
)
assert result["type"] is FlowResultType.FORM
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
side_effect=openai.OpenAIError("generic error"),
):
result2 = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
CONF_RECOMMENDED: False,
CONF_CHAT_MODEL: "gpt-4",
CONF_PROMPT: "New prompt",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "api_error"}
async def test_config_flow_unexpected_exception(
hass: HomeAssistant,
) -> None:
"""Test user step handles unexpected exception by showing unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.llama_cpp.config_flow.async_create_client",
side_effect=RuntimeError("Unexpected error"),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BASE_URL: "http://localhost:8080/v1",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "unknown"}
tests/components/llama_cpp/test_conversation.py
+572
View File
@@ -0,0 +1,572 @@
"""Tests for the llama.cpp conversation platform."""
from collections.abc import AsyncGenerator, Generator
from typing import Any
from unittest.mock import AsyncMock, patch
from freezegun import freeze_time
import httpx
import openai
from openai.types.chat import (
ChatCompletion,
ChatCompletionChunk,
ChatCompletionMessage,
ChatCompletionMessageToolCall,
)
from openai.types.chat.chat_completion import Choice
from openai.types.chat.chat_completion_chunk import Choice as ChunkChoice, ChoiceDelta
from openai.types.chat.chat_completion_message_tool_call import Function
from openai.types.completion_usage import CompletionUsage
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components import conversation
from homeassistant.components.llama_cpp.const import CONF_STREAMING
from homeassistant.const import CONF_LLM_HASS_API
from homeassistant.core import Context, HomeAssistant
from homeassistant.helpers import intent
from homeassistant.setup import async_setup_component
from .conftest import ASSIST_OPTIONS, MockChatLog
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
def freeze_the_time() -> Generator[None]:
"""Freeze the time."""
with freeze_time("2024-05-24 12:00:00", tz_offset=0):
yield
@pytest.fixture(autouse=True)
def mock_ulid() -> Generator[AsyncMock]:
"""Mock the ulid library."""
with patch("homeassistant.helpers.llm.ulid_now") as mock_ulid_now:
mock_ulid_now.return_value = "mock-ulid"
yield mock_ulid_now
@pytest.fixture(autouse=True)
async def mock_setup_integration_fixture(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Setup the integration."""
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
async def test_conversation_entity(
hass: HomeAssistant,
mock_chat_log: MockChatLog,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Verify the conversation entity is loaded."""
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
return_value=ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
Choice(
finish_reason="stop",
index=0,
message=ChatCompletionMessage(
content="Hello, how can I help you?",
role="assistant",
function_call=None,
tool_calls=None,
),
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
),
):
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ACTION_DONE
assert mock_chat_log.content[1:] == snapshot
@pytest.mark.parametrize(("config_entry_options"), [ASSIST_OPTIONS])
async def test_function_call(
hass: HomeAssistant,
mock_chat_log: MockChatLog,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test function call from the assistant."""
mock_chat_log.mock_tool_results(
{
"call_call_1": "value1",
}
)
def completion_result(
*args: Any, messages: list[dict[str, Any]] | list[Any], **kwargs: Any
) -> ChatCompletion:
for message in messages:
role = message["role"] if isinstance(message, dict) else message.role
if role == "tool":
return ChatCompletion(
id="chatcmpl-1234567890ZYXWVUTSRQPONMLKJIH",
choices=[
Choice(
finish_reason="stop",
index=0,
message=ChatCompletionMessage(
content="I have successfully called the function",
role="assistant",
function_call=None,
tool_calls=None,
),
)
],
created=1700000000,
model="gpt-4-1106-preview",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
)
return ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
Choice(
finish_reason="tool_calls",
index=0,
message=ChatCompletionMessage(
content=None,
role="assistant",
function_call=None,
tool_calls=[
ChatCompletionMessageToolCall(
id="call_call_1",
function=Function(
arguments='{"param1":"call1"}',
name="test_tool",
),
type="function",
)
],
),
)
],
created=1700000000,
model="gpt-4-1106-preview",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
)
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
side_effect=completion_result,
):
result = await conversation.async_converse(
hass,
"Please call the test function",
mock_chat_log.conversation_id,
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ACTION_DONE
assert mock_chat_log.content[1:] == snapshot
@pytest.mark.parametrize(("config_entry_options"), [ASSIST_OPTIONS])
@pytest.mark.parametrize(
("tool_arguments"),
[
(""),
('{"para'),
],
)
async def test_function_exception(
hass: HomeAssistant,
mock_chat_log: MockChatLog,
mock_config_entry: MockConfigEntry,
tool_arguments: str,
snapshot: SnapshotAssertion,
) -> None:
"""Test function call with exception."""
def completion_result(
*args: Any, messages: list[dict[str, Any]] | list[Any], **kwargs: Any
) -> ChatCompletion:
for message in messages:
role = message["role"] if isinstance(message, dict) else message.role
if role == "tool":
return ChatCompletion(
id="chatcmpl-1234567890ZYXWVUTSRQPONMLKJIH",
choices=[
Choice(
finish_reason="stop",
index=0,
message=ChatCompletionMessage(
content="There was an error calling the function",
role="assistant",
function_call=None,
tool_calls=None,
),
)
],
created=1700000000,
model="gpt-4-1106-preview",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
)
return ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
Choice(
finish_reason="tool_calls",
index=0,
message=ChatCompletionMessage(
content=None,
role="assistant",
function_call=None,
tool_calls=[
ChatCompletionMessageToolCall(
id="call_AbCdEfGhIjKlMnOpQrStUvWx",
function=Function(
arguments=tool_arguments,
name="test_tool",
),
type="function",
)
],
),
)
],
created=1700000000,
model="gpt-4-1106-preview",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
)
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
side_effect=completion_result,
):
result = await conversation.async_converse(
hass,
"Please call the test function",
"conversation-id",
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ERROR
assert result.response.speech["plain"]["speech"] == snapshot
@pytest.mark.parametrize(("config_entry_options"), [ASSIST_OPTIONS])
async def test_assist_api_tools_conversion(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test that we are able to convert actual tools from Assist API."""
for component in (
"intent",
"todo",
"light",
"shopping_list",
"humidifier",
"climate",
"media_player",
"vacuum",
"cover",
"weather",
):
assert await async_setup_component(hass, component, {})
agent_id = mock_config_entry.entry_id
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
return_value=ChatCompletion(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
Choice(
finish_reason="stop",
index=0,
message=ChatCompletionMessage(
content="Hello, how can I help you?",
role="assistant",
function_call=None,
tool_calls=None,
),
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion",
system_fingerprint=None,
usage=CompletionUsage(
completion_tokens=9, prompt_tokens=8, total_tokens=17
),
),
) as mock_create:
await conversation.async_converse(hass, "hello", None, None, agent_id=agent_id)
tools = mock_create.mock_calls[0][2]["tools"]
assert tools
@pytest.mark.parametrize(("config_entry_options"), [{CONF_STREAMING: True}])
async def test_streaming_response(
hass: HomeAssistant,
mock_chat_log: MockChatLog,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test streaming response from the assistant."""
async def mock_stream() -> AsyncGenerator[ChatCompletionChunk]:
yield ChatCompletionChunk.model_construct(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
ChunkChoice.model_construct(
index=0,
delta=ChoiceDelta(role="assistant", content="Hello"),
finish_reason=None,
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion.chunk",
)
yield ChatCompletionChunk.model_construct(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
ChunkChoice.model_construct(
index=0,
delta=ChoiceDelta(content=" world"),
finish_reason=None,
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion.chunk",
)
yield ChatCompletionChunk(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
ChunkChoice(
index=0,
delta=ChoiceDelta(),
finish_reason="stop",
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion.chunk",
)
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
return_value=mock_stream(),
):
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ACTION_DONE
assert result.response.speech["plain"]["speech"] == "Hello world"
content = mock_chat_log.content[1:]
assert len(content) == 2
assert content[0].role == "user"
assert content[0].content == "hello"
assert content[1].role == "assistant"
assert content[1].content == "Hello world"
@pytest.mark.parametrize(("config_entry_options"), [{CONF_STREAMING: True}])
async def test_streaming_response_redundant_role(
hass: HomeAssistant,
mock_chat_log: MockChatLog,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test streaming response where every chunk redundantly includes the role."""
async def mock_stream() -> AsyncGenerator[ChatCompletionChunk]:
yield ChatCompletionChunk.model_construct(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
ChunkChoice.model_construct(
index=0,
delta=ChoiceDelta(role="assistant", content="Hello"),
finish_reason=None,
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion.chunk",
)
yield ChatCompletionChunk.model_construct(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
ChunkChoice.model_construct(
index=0,
delta=ChoiceDelta(role="assistant", content=" world"),
finish_reason=None,
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion.chunk",
)
yield ChatCompletionChunk(
id="chatcmpl-1234567890ABCDEFGHIJKLMNOPQRS",
choices=[
ChunkChoice(
index=0,
delta=ChoiceDelta(role="assistant"),
finish_reason="stop",
)
],
created=1700000000,
model="gpt-3.5-turbo-0613",
object="chat.completion.chunk",
)
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
new_callable=AsyncMock,
return_value=mock_stream(),
):
result = await conversation.async_converse(
hass,
"hello",
mock_chat_log.conversation_id,
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ACTION_DONE
assert result.response.speech["plain"]["speech"] == "Hello world"
content = mock_chat_log.content[1:]
assert len(content) == 2
assert content[0].role == "user"
assert content[0].content == "hello"
assert content[1].role == "assistant"
assert content[1].content == "Hello world"
@pytest.mark.parametrize(
("config_entry_options"), [{CONF_LLM_HASS_API: ["non-existing"]}]
)
async def test_unknown_hass_api(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test when we reference an API that no longer exists."""
result = await conversation.async_converse(
hass, "hello", "conversation-id", Context(), agent_id=mock_config_entry.entry_id
)
assert result.as_dict() == snapshot
async def test_conversation_agent_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test handling of OpenAI API connection errors in conversation entity."""
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
side_effect=openai.APIConnectionError(
request=httpx.Request(method="POST", url="test")
),
):
result = await conversation.async_converse(
hass,
"hello",
"conversation-id",
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ERROR
assert (
result.response.speech["plain"]["speech"]
== "Cannot connect to the server: Connection error."
)
async def test_conversation_agent_structured_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test handling of OpenAI API structured errors in conversation entity."""
response = httpx.Response(
status_code=402,
request=httpx.Request(
method="POST", url="https://api.openai.com/v1/chat/completions"
),
json={
"error": {
"message": "Insufficient Balance",
"type": "unknown_error",
"param": None,
"code": "invalid_request_error",
}
},
)
err = openai.APIStatusError(
message="Error code: 402 - {'error': {'message': 'Insufficient Balance'}}",
response=response,
body=response.json(),
)
with patch(
"openai.resources.chat.completions.AsyncCompletions.create",
side_effect=err,
):
result = await conversation.async_converse(
hass,
"hello",
"conversation-id",
Context(),
agent_id="conversation.llama_cpp_conversation",
)
assert result.response.response_type == intent.IntentResponseType.ERROR
assert (
result.response.speech["plain"]["speech"]
== "Your account or API key has insufficient credits: Insufficient Balance"
)
tests/components/llama_cpp/test_init.py
+68
View File
@@ -0,0 +1,68 @@
"""Tests for llama.cpp integration setup."""
from unittest.mock import AsyncMock, patch
import httpx
import openai
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_setup_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test setting up and unloading llama.cpp entry."""
with patch(
"openai.resources.models.AsyncModels.list",
new_callable=AsyncMock,
):
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("side_effect", "expected_state"),
[
(
openai.AuthenticationError(
message="Invalid API key",
response=httpx.Response(
status_code=401,
request=httpx.Request(method="GET", url="test"),
),
body=None,
),
ConfigEntryState.SETUP_ERROR,
),
(
openai.APIConnectionError(request=None),
ConfigEntryState.SETUP_RETRY,
),
],
)
async def test_setup_entry_failures(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
side_effect: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup entry failure handling."""
with patch(
"openai.resources.models.AsyncModels.list",
side_effect=side_effect,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is expected_state