luni/ha-groq-cloud-api
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #16 from Thyraz/history

Browse Source 
Change handling of conversation history to use the new HA improvements
This commit is contained in:
luni
2025-12-22 16:10:46 +01:00
committed by GitHub
parent 45aa4992d0 f9192b1cad
commit f87aecaefa
1 changed files with 147 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/groq_cloud_api/conversation.py
+146 -159
View File
@@ -8,8 +8,6 @@ import groq
from groq._types import NOT_GIVEN
from groq.types.chat import (
ChatCompletionAssistantMessageParam,
ChatCompletionMessage,
ChatCompletionMessageParam,
ChatCompletionMessageToolCallParam,
ChatCompletionSystemMessageParam,
ChatCompletionToolMessageParam,
@@ -18,18 +16,24 @@ from groq.types.chat import (
)
from groq.types.chat.chat_completion_message_tool_call_param import Function
from groq.types.shared_params import FunctionDefinition
import voluptuous as vol
from voluptuous_openapi import convert
from homeassistant.components import assist_pipeline, conversation
from homeassistant.components.conversation import trace
from homeassistant.components import conversation
from homeassistant.components.conversation import (
AssistantContent,
ConverseError,
SystemContent,
ToolResultContent,
UserContent,
async_get_result_from_chat_log,
trace,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_LLM_HASS_API, MATCH_ALL
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, TemplateError
from homeassistant.helpers import device_registry as dr, intent, llm, template
from homeassistant.helpers import device_registry as dr, intent, llm
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import ulid
from homeassistant.helpers.json import json_dumps
from . import GroqConfigEntry
from .const import (
@@ -73,6 +77,75 @@ def _format_tool(
return ChatCompletionToolParam(type="function", function=tool_spec)
def _assistant_content_to_message(
content: AssistantContent,
) -> ChatCompletionAssistantMessageParam:
"""Convert AssistantContent to a Groq assistant message."""
tool_calls: list[ChatCompletionMessageToolCallParam] = []
if content.tool_calls:
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
]
assistant_message = ChatCompletionAssistantMessageParam(
role="assistant",
content=content.content,
)
if tool_calls:
assistant_message["tool_calls"] = tool_calls
return assistant_message
def _chat_log_to_messages(
chat_log: conversation.ChatLog,
) -> list[
ChatCompletionSystemMessageParam
| ChatCompletionUserMessageParam
| ChatCompletionAssistantMessageParam
| ChatCompletionToolMessageParam
]:
"""Convert chat log content to Groq chat completion messages."""
messages: list[
ChatCompletionSystemMessageParam
| ChatCompletionUserMessageParam
| ChatCompletionAssistantMessageParam
| ChatCompletionToolMessageParam
] = []
for content in chat_log.content:
if isinstance(content, SystemContent):
messages.append(
ChatCompletionSystemMessageParam(
role="system", content=content.content
)
)
elif isinstance(content, UserContent):
messages.append(
ChatCompletionUserMessageParam(role="user", content=content.content)
)
elif isinstance(content, AssistantContent):
messages.append(_assistant_content_to_message(content))
elif isinstance(content, ToolResultContent):
messages.append(
ChatCompletionToolMessageParam(
role="tool",
tool_call_id=content.tool_call_id,
content=json_dumps(content.tool_result),
)
)
return messages
class GroqConversationEntity(
conversation.ConversationEntity, conversation.AbstractConversationAgent
):
@@ -84,7 +157,6 @@ class GroqConversationEntity(
def __init__(self, entry: GroqConfigEntry) -> None:
"""Initialize the agent."""
self.entry = entry
self.history: dict[str, list[ChatCompletionMessageParam]] = {}
self._attr_unique_id = entry.entry_id
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
@@ -116,111 +188,34 @@ class GroqConversationEntity(
conversation.async_unset_agent(self.hass, self.entry)
await super().async_will_remove_from_hass()
async def async_process(
self, user_input: conversation.ConversationInput
async def _async_handle_message(
self,
user_input: conversation.ConversationInput,
chat_log: conversation.ChatLog,
) -> conversation.ConversationResult:
"""Process a sentence."""
"""Process the user input and call the API."""
options = self.entry.options
intent_response = intent.IntentResponse(language=user_input.language)
llm_api: llm.APIInstance | None = None
tools: list[ChatCompletionToolParam] | None = None
user_name: str | None = None
llm_context = llm.LLMContext(
platform=DOMAIN,
context=user_input.context,
language=user_input.language,
assistant=conversation.DOMAIN,
device_id=user_input.device_id,
)
if options.get(CONF_LLM_HASS_API):
try:
llm_api = await llm.async_get_api(
self.hass,
options[CONF_LLM_HASS_API],
llm_context,
)
except HomeAssistantError as err:
LOGGER.error("Error getting LLM API: %s", err)
intent_response.async_set_error(
intent.IntentResponseErrorCode.UNKNOWN,
f"Error preparing LLM API: {err}",
)
return conversation.ConversationResult(
response=intent_response, conversation_id=user_input.conversation_id
)
try:
await chat_log.async_provide_llm_data(
llm_context=user_input.as_llm_context(DOMAIN),
user_llm_hass_api=options.get(CONF_LLM_HASS_API),
user_llm_prompt=options.get(
CONF_PROMPT, llm.DEFAULT_INSTRUCTIONS_PROMPT
),
user_extra_system_prompt=user_input.extra_system_prompt,
)
except ConverseError as err:
return err.as_conversation_result()
llm_api = chat_log.llm_api
tools: list[ChatCompletionToolParam] | None = None
if llm_api:
tools = [
_format_tool(tool, llm_api.custom_serializer) for tool in llm_api.tools
]
if user_input.conversation_id is None:
conversation_id = ulid.ulid_now()
messages = []
elif user_input.conversation_id in self.history:
conversation_id = user_input.conversation_id
messages = self.history[conversation_id]
else:
# Conversation IDs are ULIDs. We generate a new one if not provided.
# If an old OLID is passed in, we will generate a new one to indicate
# a new conversation was started. If the user picks their own, they
# want to track a conversation and we respect it.
try:
ulid.ulid_to_bytes(user_input.conversation_id)
conversation_id = ulid.ulid_now()
except ValueError:
conversation_id = user_input.conversation_id
messages = []
if (
user_input.context
and user_input.context.user_id
and (
user := await self.hass.auth.async_get_user(user_input.context.user_id)
)
):
user_name = user.name
try:
prompt_parts = [
template.Template(
llm.DATE_TIME_PROMPT
+ options.get(CONF_PROMPT, llm.DEFAULT_INSTRUCTIONS_PROMPT),
self.hass,
).async_render(
{
"ha_name": self.hass.config.location_name,
"user_name": user_name,
"llm_context": llm_context,
},
parse_result=False,
)
]
except TemplateError as err:
LOGGER.error("Error rendering prompt: %s", err)
intent_response = intent.IntentResponse(language=user_input.language)
intent_response.async_set_error(
intent.IntentResponseErrorCode.UNKNOWN,
f"Sorry, I had a problem with my template: {err}",
)
return conversation.ConversationResult(
response=intent_response, conversation_id=conversation_id
)
if llm_api:
prompt_parts.append(llm_api.api_prompt)
prompt = "\n".join(prompt_parts)
# Create a copy of the variable because we attach it to the trace
messages = [
ChatCompletionSystemMessageParam(role="system", content=prompt),
*messages[1:],
ChatCompletionUserMessageParam(role="user", content=user_input.text),
]
messages = _chat_log_to_messages(chat_log)
LOGGER.debug("Prompt: %s", messages)
LOGGER.debug("Tools: %s", tools)
@@ -241,7 +236,7 @@ class GroqConversationEntity(
max_tokens=options.get(CONF_MAX_TOKENS, RECOMMENDED_MAX_TOKENS),
top_p=options.get(CONF_TOP_P, RECOMMENDED_TOP_P),
temperature=options.get(CONF_TEMPERATURE, RECOMMENDED_TEMPERATURE),
user=conversation_id,
user=chat_log.conversation_id,
)
except groq.GroqError as err:
intent_response = intent.IntentResponse(language=user_input.language)
@@ -250,75 +245,67 @@ class GroqConversationEntity(
f"Sorry, I had a problem talking to Groq: {err}",
)
return conversation.ConversationResult(
response=intent_response, conversation_id=conversation_id
response=intent_response, conversation_id=chat_log.conversation_id
)
LOGGER.debug("Response %s", result)
response = result.choices[0].message
def message_convert(
message: ChatCompletionMessage,
) -> ChatCompletionMessageParam:
"""Convert from class to TypedDict."""
tool_calls: list[ChatCompletionMessageToolCallParam] = []
if message.tool_calls:
tool_calls = [
ChatCompletionMessageToolCallParam(
id=tool_call.id,
function=Function(
arguments=tool_call.function.arguments,
name=tool_call.function.name,
),
type=tool_call.type,
)
for tool_call in message.tool_calls
]
param = ChatCompletionAssistantMessageParam(
role=message.role,
content=message.content,
)
if tool_calls:
param["tool_calls"] = tool_calls
return param
messages.append(message_convert(response))
tool_calls = response.tool_calls
if not tool_calls or not llm_api:
break
for tool_call in tool_calls:
tool_input = llm.ToolInput(
groq_tool_calls = response.tool_calls or []
assistant_tool_calls = [
llm.ToolInput(
id=tool_call.id,
tool_name=tool_call.function.name,
tool_args=json.loads(tool_call.function.arguments),
)
for tool_call in groq_tool_calls
]
assistant_content = AssistantContent(
agent_id=self.entity_id,
content=response.content,
tool_calls=assistant_tool_calls or None,
)
messages.append(_assistant_content_to_message(assistant_content))
if not assistant_tool_calls:
chat_log.async_add_assistant_content_without_tools(assistant_content)
break
for tool_call in assistant_tool_calls:
LOGGER.debug(
"Tool call: %s(%s)", tool_input.tool_name, tool_input.tool_args
"Tool call: %s(%s)", tool_call.tool_name, tool_call.tool_args
)
try:
tool_response = await llm_api.async_call_tool(tool_input)
except (HomeAssistantError, vol.Invalid) as e:
tool_response = {"error": type(e).__name__}
if str(e):
tool_response["error_text"] = str(e)
if llm_api is None:
intent_response = intent.IntentResponse(language=user_input.language)
intent_response.async_set_error(
intent.IntentResponseErrorCode.UNKNOWN,
"Tool call requested but no LLM API configured",
)
return conversation.ConversationResult(
response=intent_response,
conversation_id=chat_log.conversation_id,
)
LOGGER.debug("Tool response: %s", tool_response)
async for tool_result_content in chat_log.async_add_assistant_content(
assistant_content
):
LOGGER.debug(
"Tool response: %s -> %s",
tool_result_content.tool_name,
tool_result_content.tool_result,
)
messages.append(
ChatCompletionToolMessageParam(
role="tool",
tool_call_id=tool_call.id,
content=json.dumps(tool_response),
tool_call_id=tool_result_content.tool_call_id,
content=json_dumps(tool_result_content.tool_result),
)
)
self.history[conversation_id] = messages
intent_response = intent.IntentResponse(language=user_input.language)
intent_response.async_set_speech(response.content or "")
return conversation.ConversationResult(
response=intent_response, conversation_id=conversation_id
)
return async_get_result_from_chat_log(user_input, chat_log)
async def _async_entry_update_listener(
self, hass: HomeAssistant, entry: ConfigEntry