diff --git a/custom_components/webhook_conversation/ai_task.py b/custom_components/webhook_conversation/ai_task.py index a878117..ccc4ce8 100644 --- a/custom_components/webhook_conversation/ai_task.py +++ b/custom_components/webhook_conversation/ai_task.py @@ -4,6 +4,7 @@ import base64 import logging +from typing import Any import anyio from voluptuous_openapi import convert @@ -14,6 +15,7 @@ from homeassistant.helpers import llm from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from .const import CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD from .entity import WebhookConversationLLMBaseEntity from .models import WebhookConversationBinaryObject @@ -75,14 +77,19 @@ async def _async_generate_data( task.structure.schema, custom_serializer=llm.selector_serializer ) + reply: Any if self._streaming_enabled: - reply_parts = [ - content_chunk - async for content_chunk in self._send_payload_streaming(payload) - ] + reply_parts = [] + async for chunk_data in self._send_payload_streaming(payload): + if chunk_data.get("type") == "item" and "content" in chunk_data: + reply_parts.append(chunk_data["content"]) reply = "".join(reply_parts) else: - reply = await self._send_payload(payload) + output_field: str = self._subentry.data.get( + CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD + ) + result = await self._send_payload(payload) + reply = result.get(output_field) if not task.structure: text = reply if isinstance(reply, str) else str(reply) diff --git a/custom_components/webhook_conversation/conversation.py b/custom_components/webhook_conversation/conversation.py index af62d0a..3f493c7 100644 --- a/custom_components/webhook_conversation/conversation.py +++ b/custom_components/webhook_conversation/conversation.py @@ -15,15 +15,18 @@ area_registry as ar, device_registry as dr, entity_registry as er, + llm, ) from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .const import DOMAIN +from .const import CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD, DOMAIN from .entity import WebhookConversationLLMBaseEntity from .models import WebhookConversationPayload _LOGGER = logging.getLogger(__name__) +MAX_TOOL_ITERATIONS = 10 + async def async_setup_entry( hass: HomeAssistant, @@ -137,21 +140,56 @@ def set_default(obj: Any) -> Any: payload["user_id"] = user_input.context.user_id payload["user_name"] = user.name if user else None - if self._streaming_enabled: - async for _ in chat_log.async_add_delta_content_stream( - self.entity_id, - self._transform_webhook_stream(payload), - ): - pass - else: - reply = await self._send_payload(payload) - async for _ in chat_log.async_add_assistant_content( - conversation.AssistantContent( + output_field: str = self._subentry.data.get( + CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD + ) + + for _iteration in range(MAX_TOOL_ITERATIONS): + if self._streaming_enabled: + async for _ in chat_log.async_add_delta_content_stream( self.entity_id, - reply, + self._transform_webhook_stream(payload), + ): + pass + else: + result = await self._send_payload(payload, allow_tool_only=True) + reply = result.get(output_field) + tool_calls = _parse_tool_calls(result.get("tool_calls")) + async for _ in chat_log.async_add_assistant_content( + conversation.AssistantContent( + agent_id=self.entity_id, + content=reply, + tool_calls=tool_calls or None, + ) + ): + pass + + if not chat_log.unresponded_tool_results: + break + + payload = self._build_payload(chat_log, include_last=True) + payload["query"] = user_messages[-1]["content"] + payload["agent_id"] = user_input.agent_id + payload["device_id"] = user_input.device_id + payload["device_info"] = ( + ( + device.dict_repr + if (device := device_registry.async_get(user_input.device_id)) + else None ) - ): - pass + if user_input.device_id + else None + ) + payload["exposed_entities"] = json.dumps( + self._get_exposed_entities(), default=set_default + ) + payload["language"] = user_input.language + payload["user_id"] = user_input.context.user_id + payload["user_name"] = user.name if user else None + else: + raise HomeAssistantError( + f"Webhook tool call loop exceeded {MAX_TOOL_ITERATIONS} iterations" + ) async def _transform_webhook_stream( self, payload: WebhookConversationPayload @@ -159,9 +197,14 @@ async def _transform_webhook_stream( """Transform webhook streaming content into HA format.""" yield {"role": "assistant"} - async for content_delta in self._send_payload_streaming(payload): - _LOGGER.debug("Webhook streaming response: %s", content_delta) - yield {"content": content_delta} + async for chunk_data in self._send_payload_streaming(payload): + _LOGGER.debug("Webhook streaming response: %s", chunk_data) + if chunk_data.get("type") == "item" and "content" in chunk_data: + yield {"content": chunk_data["content"]} + elif chunk_data.get("type") == "tool_calls" and "tool_calls" in chunk_data: + tool_calls = _parse_tool_calls(chunk_data["tool_calls"]) + if tool_calls: + yield {"tool_calls": tool_calls} def _get_exposed_entities(self) -> list[dict[str, Any]]: states = [ @@ -212,3 +255,20 @@ def _get_exposed_entities(self) -> list[dict[str, Any]]: } ) return exposed_entities + + +def _parse_tool_calls( + raw_tool_calls: list[dict[str, Any]] | None, +) -> list[llm.ToolInput] | None: + """Parse tool calls from webhook response into ToolInput objects.""" + if not raw_tool_calls: + return None + return [ + llm.ToolInput( + tool_name=tool_call["name"], + tool_args=tool_call.get("arguments", {}), + id=tool_call.get("id", ""), + ) + for tool_call in raw_tool_calls + if "name" in tool_call + ] or None diff --git a/custom_components/webhook_conversation/entity.py b/custom_components/webhook_conversation/entity.py index d268ac9..9ef1126 100644 --- a/custom_components/webhook_conversation/entity.py +++ b/custom_components/webhook_conversation/entity.py @@ -91,7 +91,11 @@ def __init__(self, config_entry: ConfigEntry, subentry: ConfigSubentry) -> None: CONF_ENABLE_STREAMING, DEFAULT_ENABLE_STREAMING ) - async def _send_payload(self, payload: WebhookConversationPayload) -> Any: + async def _send_payload( + self, + payload: WebhookConversationPayload, + allow_tool_only: bool = False, + ) -> dict[str, Any]: """Send the payload to the webhook.""" _LOGGER.debug( "Webhook request: %s", @@ -118,15 +122,18 @@ async def _send_payload(self, payload: WebhookConversationPayload) -> Any: output_field: str = self._subentry.data.get( CONF_OUTPUT_FIELD, DEFAULT_OUTPUT_FIELD ) - if not isinstance(result, dict) or output_field not in result: + if not isinstance(result, dict) or ( + output_field not in result + and not (allow_tool_only and "tool_calls" in result) + ): raise HomeAssistantError(f"Invalid webhook response: {result}") _LOGGER.debug("Webhook response: %s", result) - return result.get(output_field) + return result async def _send_payload_streaming( self, payload: WebhookConversationPayload - ) -> AsyncGenerator[str]: + ) -> AsyncGenerator[dict[str, Any]]: """Send the payload to the webhook and stream the response.""" _LOGGER.debug("Webhook streaming request: %s", payload) @@ -153,15 +160,14 @@ async def _send_payload_streaming( try: chunk_data = json.loads(line_str) chunk_type = chunk_data.get("type") - if chunk_type == "item" and "content" in chunk_data: - yield chunk_data["content"] - elif chunk_type == "error": + if chunk_type == "error": raise HomeAssistantError( f"n8n error: {chunk_data.get('message', chunk_data)}" ) # We don't break on "end" because n8n can send multiple # begin/end blocks when using tools or intermediate steps. # We keep reading until the stream actually closes. + yield chunk_data except json.JSONDecodeError: _LOGGER.warning( "Failed to parse streaming response chunk: %s", line_str @@ -169,16 +175,17 @@ async def _send_payload_streaming( continue def _build_payload( - self, chat_log: conversation.ChatLog + self, chat_log: conversation.ChatLog, include_last: bool = False ) -> WebhookConversationPayload: """Create a base payload from the chat log for webhook calls.""" system_message = chat_log.content[0] if not isinstance(system_message, conversation.SystemContent): raise TypeError("First message must be a system message") + end_idx = len(chat_log.content) if include_last else -1 messages = [ self._convert_content_to_param(content) - for content in chat_log.content[1:-1] + for content in chat_log.content[1:end_idx] ] return WebhookConversationPayload( @@ -195,11 +202,35 @@ def _convert_content_to_param( self, content: conversation.Content ) -> WebhookConversationMessage: """Convert native chat content into a simple dict.""" + if isinstance(content, conversation.ToolResultContent): + return WebhookConversationMessage( + { + "role": content.role, + "content": json.dumps(content.tool_result) + if content.tool_result is not None + else "", + "tool_call_id": content.tool_call_id, + "tool_name": content.tool_name, + } + ) + if isinstance(content, conversation.AssistantContent) and content.tool_calls: + return WebhookConversationMessage( + { + "role": content.role, + "content": content.content or "", + "tool_calls": [ + { + "id": tool_call.id, + "name": tool_call.tool_name, + "arguments": tool_call.tool_args, + } + for tool_call in content.tool_calls + ], + } + ) return WebhookConversationMessage( { "role": content.role, - "content": str(content.tool_result or "") - if isinstance(content, conversation.ToolResultContent) - else content.content or "", + "content": content.content or "", } ) diff --git a/custom_components/webhook_conversation/models.py b/custom_components/webhook_conversation/models.py index 74b7add..40aa3b1 100644 --- a/custom_components/webhook_conversation/models.py +++ b/custom_components/webhook_conversation/models.py @@ -8,11 +8,22 @@ MessageRole = Literal["assistant", "system", "tool_result", "user"] +class WebhookConversationToolCall(TypedDict): + """A single tool call item.""" + + id: str + name: str + arguments: dict[str, Any] + + class WebhookConversationMessage(TypedDict): """A single message item.""" role: MessageRole content: str + tool_calls: NotRequired[list[WebhookConversationToolCall]] + tool_call_id: NotRequired[str] + tool_name: NotRequired[str] class WebhookConversationBinaryObject(TypedDict):