home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2026-07-05 08:01:26 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
dev
core/tests/components/llama_cpp/test_conversation.py
T
Allen Porter 3aa56eab27 Add llama_cpp conversation integration (#175580)
2026-07-04 08:20:10 +02:00

573 lines
19 KiB
Python
Raw Permalink Blame History

"""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"
)
Reference in New Issue View Git Blame Copy Permalink