github/home-llm
1
1
Fork 0
mirror of https://github.com/acon96/home-llm.git synced 2026-01-08 05:14:02 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add Anthropic API backend support

Browse Source 
Adds a new backend for Anthropic API alongside existing backends
(Ollama, OpenAI, llama.cpp, etc.).

Features:
- Support for official Anthropic API (api.anthropic.com)
- Custom base URL support for Anthropic-compatible servers
- Streaming responses via Messages API
- Tool calling support with Home Assistant integration
- Vision support for image attachments

Files changed:
- backends/anthropic.py: New AnthropicAPIClient implementation
- const.py: Added backend type and configuration constants
- __init__.py: Registered new backend
- config_flow.py: Added connection schema and options
- manifest.json: Added anthropic>=0.40.0 dependency
- translations/en.json: Added UI labels
This commit is contained in:
Thomas
2025-12-28 21:51:23 -05:00
parent 831ef9bfca
commit e84e4c21f8
6 changed files with 455 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
custom_components/llama_conversation/__init__.py
View File

@@ -44,6 +44,7 @@ from .const import (
BACKEND_TYPE_GENERIC_OPENAI_RESPONSES,
BACKEND_TYPE_LLAMA_CPP_SERVER,
BACKEND_TYPE_OLLAMA,
BACKEND_TYPE_ANTHROPIC,
BACKEND_TYPE_LLAMA_EXISTING_OLD,
BACKEND_TYPE_LLAMA_HF_OLD,
)
@@ -52,6 +53,7 @@ from .backends.llamacpp import LlamaCppClient
from .backends.generic_openai import GenericOpenAIAPIClient, GenericOpenAIResponsesAPIClient
from .backends.tailored_openai import TextGenerationWebuiClient, LlamaCppServerClient
from .backends.ollama import OllamaAPIClient
from .backends.anthropic import AnthropicAPIClient
from .utils import get_llama_cpp_python_version, download_model_from_hf
_LOGGER = logging.getLogger(__name__)
@@ -64,9 +66,10 @@ BACKEND_TO_CLS: dict[str, type[LocalLLMClient]] = {
BACKEND_TYPE_LLAMA_CPP: LlamaCppClient,
BACKEND_TYPE_GENERIC_OPENAI: GenericOpenAIAPIClient,
BACKEND_TYPE_GENERIC_OPENAI_RESPONSES: GenericOpenAIResponsesAPIClient,
BACKEND_TYPE_TEXT_GEN_WEBUI: TextGenerationWebuiClient,
BACKEND_TYPE_TEXT_GEN_WEBUI: TextGenerationWebuiClient,
BACKEND_TYPE_LLAMA_CPP_SERVER: LlamaCppServerClient,
BACKEND_TYPE_OLLAMA: OllamaAPIClient,
BACKEND_TYPE_ANTHROPIC: AnthropicAPIClient,
}
async def async_setup_entry(hass: HomeAssistant, entry: LocalLLMConfigEntry) -> bool:

353
custom_components/llama_conversation/backends/anthropic.py Normal file
View File

@@ -0,0 +1,353 @@
"""Defines the Anthropic API backend using the official Python SDK."""
from __future__ import annotations
import json
import logging
from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple
from anthropic import AsyncAnthropic, APIError, APIConnectionError, APITimeoutError
from homeassistant.components import conversation as conversation
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import llm
from voluptuous_openapi import convert as convert_to_openapi
from custom_components.llama_conversation.const import (
CONF_CHAT_MODEL,
CONF_MAX_TOKENS,
CONF_TEMPERATURE,
CONF_TOP_P,
CONF_TOP_K,
CONF_REQUEST_TIMEOUT,
CONF_ENABLE_LEGACY_TOOL_CALLING,
CONF_TOOL_RESPONSE_AS_STRING,
CONF_ANTHROPIC_API_KEY,
CONF_ANTHROPIC_BASE_URL,
DEFAULT_MAX_TOKENS,
DEFAULT_TEMPERATURE,
DEFAULT_TOP_P,
DEFAULT_TOP_K,
DEFAULT_REQUEST_TIMEOUT,
DEFAULT_ENABLE_LEGACY_TOOL_CALLING,
DEFAULT_TOOL_RESPONSE_AS_STRING,
DEFAULT_ANTHROPIC_BASE_URL,
RECOMMENDED_ANTHROPIC_MODELS,
)
from custom_components.llama_conversation.entity import LocalLLMClient, TextGenerationResult
_LOGGER = logging.getLogger(__name__)
def _convert_to_anthropic_messages(
conversation_messages: List[conversation.Content],
tool_result_to_str: bool = True,
) -> Tuple[str, List[Dict[str, Any]]]:
"""
Convert Home Assistant conversation format to Anthropic Messages API format.
Returns:
Tuple of (system_prompt, messages_list)
Note: Anthropic requires system prompt as a separate parameter, not in messages.
"""
system_prompt = ""
messages: List[Dict[str, Any]] = []
for message in conversation_messages:
if message.role == "system":
# Anthropic handles system prompts separately
system_prompt = message.content if hasattr(message, 'content') else str(message)
elif message.role == "user":
content = []
msg_content = message.content if hasattr(message, 'content') else str(message)
if msg_content:
content.append({"type": "text", "text": msg_content})
# Handle image attachments (Anthropic supports vision)
if hasattr(message, 'attachments') and message.attachments:
import base64
for attachment in message.attachments:
if hasattr(attachment, 'mime_type') and attachment.mime_type.startswith("image/"):
try:
with open(attachment.path, "rb") as f:
image_data = base64.b64encode(f.read()).decode("utf-8")
content.append({
"type": "image",
"source": {
"type": "base64",
"media_type": attachment.mime_type,
"data": image_data,
}
})
except Exception as e:
_LOGGER.warning("Failed to load image attachment: %s", e)
if content:
messages.append({"role": "user", "content": content})
elif message.role == "assistant":
content = []
msg_content = message.content if hasattr(message, 'content') else None
if msg_content:
content.append({"type": "text", "text": str(msg_content)})
# Handle tool calls (Anthropic's tool_use format)
if hasattr(message, 'tool_calls') and message.tool_calls:
for tool_call in message.tool_calls:
tool_id = getattr(tool_call, 'id', None) or f"toolu_{id(tool_call)}"
content.append({
"type": "tool_use",
"id": tool_id,
"name": tool_call.tool_name,
"input": tool_call.tool_args if isinstance(tool_call.tool_args, dict) else {},
})
if content:
messages.append({"role": "assistant", "content": content})
elif message.role == "tool_result":
# Anthropic expects tool results in user messages with tool_result content
tool_result = message.tool_result if hasattr(message, 'tool_result') else {}
if tool_result_to_str:
result_content = json.dumps(tool_result) if isinstance(tool_result, dict) else str(tool_result)
else:
result_content = str(tool_result)
tool_call_id = getattr(message, 'tool_call_id', None) or "unknown"
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_call_id,
"content": result_content,
}]
})
return system_prompt, messages
def _convert_tools_to_anthropic_format(
llm_api: llm.APIInstance,
) -> List[Dict[str, Any]]:
"""Convert Home Assistant LLM tools to Anthropic tool format."""
tools: List[Dict[str, Any]] = []
for tool in sorted(llm_api.tools, key=lambda t: t.name):
schema = convert_to_openapi(tool.parameters, custom_serializer=llm_api.custom_serializer)
tools.append({
"name": tool.name,
"description": tool.description or "",
"input_schema": schema,
})
return tools
class AnthropicAPIClient(LocalLLMClient):
"""Implements the Anthropic Messages API backend."""
api_key: str
base_url: str
def __init__(self, hass: HomeAssistant, client_options: dict[str, Any]) -> None:
super().__init__(hass, client_options)
self.api_key = client_options.get(CONF_ANTHROPIC_API_KEY, "")
self.base_url = client_options.get(CONF_ANTHROPIC_BASE_URL, DEFAULT_ANTHROPIC_BASE_URL)
def _build_client(self, timeout: float | None = None) -> AsyncAnthropic:
"""Build an async Anthropic client."""
effective_timeout = timeout or DEFAULT_REQUEST_TIMEOUT
# Only pass base_url if it's different from the default
kwargs: Dict[str, Any] = {
"api_key": self.api_key,
"timeout": effective_timeout,
}
if self.base_url and self.base_url != DEFAULT_ANTHROPIC_BASE_URL:
kwargs["base_url"] = self.base_url
return AsyncAnthropic(**kwargs)
@staticmethod
def get_name(client_options: dict[str, Any]) -> str:
base_url = client_options.get(CONF_ANTHROPIC_BASE_URL, DEFAULT_ANTHROPIC_BASE_URL)
if base_url == DEFAULT_ANTHROPIC_BASE_URL:
return "Anthropic API"
return f"Anthropic-compatible API at '{base_url}'"
@staticmethod
async def async_validate_connection(
hass: HomeAssistant, user_input: Dict[str, Any]
) -> str | None:
"""Validate connection to the Anthropic API."""
api_key = user_input.get(CONF_ANTHROPIC_API_KEY, "")
base_url = user_input.get(CONF_ANTHROPIC_BASE_URL, DEFAULT_ANTHROPIC_BASE_URL)
if not api_key:
return "API key is required"
try:
kwargs: Dict[str, Any] = {
"api_key": api_key,
"timeout": 10.0,
}
if base_url and base_url != DEFAULT_ANTHROPIC_BASE_URL:
kwargs["base_url"] = base_url
client = AsyncAnthropic(**kwargs)
# Test the connection with a minimal request
await client.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=1,
messages=[{"role": "user", "content": "hi"}],
)
return None
except APIConnectionError as err:
return f"Connection error: {err}"
except APITimeoutError:
return "Connection timed out"
except APIError as err:
if hasattr(err, 'status_code') and err.status_code == 401:
return "Invalid API key"
return f"API error: {err}"
except Exception as err:
return str(err)
async def async_get_available_models(self) -> List[str]:
"""Return available Anthropic models."""
# Anthropic doesn't have a models list endpoint, so return recommended models
return RECOMMENDED_ANTHROPIC_MODELS
def _supports_vision(self, entity_options: dict[str, Any]) -> bool:
"""Anthropic models support vision."""
return True
def _generate_stream(
self,
conversation: List[conversation.Content],
llm_api: llm.APIInstance | None,
agent_id: str,
entity_options: dict[str, Any],
) -> AsyncGenerator[TextGenerationResult, None]:
"""Generate streaming response using Anthropic's Messages API."""
model_name = entity_options.get(CONF_CHAT_MODEL, RECOMMENDED_ANTHROPIC_MODELS[0])
max_tokens = int(entity_options.get(CONF_MAX_TOKENS, DEFAULT_MAX_TOKENS))
temperature = entity_options.get(CONF_TEMPERATURE, DEFAULT_TEMPERATURE)
top_p = entity_options.get(CONF_TOP_P, DEFAULT_TOP_P)
top_k = entity_options.get(CONF_TOP_K, DEFAULT_TOP_K)
timeout = entity_options.get(CONF_REQUEST_TIMEOUT, DEFAULT_REQUEST_TIMEOUT)
enable_legacy_tool_calling = entity_options.get(
CONF_ENABLE_LEGACY_TOOL_CALLING, DEFAULT_ENABLE_LEGACY_TOOL_CALLING
)
tool_response_as_string = entity_options.get(
CONF_TOOL_RESPONSE_AS_STRING, DEFAULT_TOOL_RESPONSE_AS_STRING
)
# Convert conversation to Anthropic format
system_prompt, messages = _convert_to_anthropic_messages(
conversation, tool_result_to_str=tool_response_as_string
)
# Prepare tools if available and not using legacy tool calling
tools = None
if llm_api and not enable_legacy_tool_calling:
tools = _convert_tools_to_anthropic_format(llm_api)
_LOGGER.debug(
"Generating completion with model=%s, %d messages, and %d tools...",
model_name,
len(messages),
len(tools) if tools else 0,
)
async def anext_token() -> AsyncGenerator[Tuple[Optional[str], Optional[List[dict]]], None]:
client = self._build_client(timeout=timeout)
request_params: Dict[str, Any] = {
"model": model_name,
"max_tokens": max_tokens,
"messages": messages,
}
# Add optional parameters
if system_prompt:
request_params["system"] = system_prompt
if tools:
request_params["tools"] = tools
if temperature is not None:
request_params["temperature"] = temperature
if top_p is not None:
request_params["top_p"] = top_p
if top_k is not None and top_k > 0:
request_params["top_k"] = top_k
try:
current_tool_call: Dict[str, Any] | None = None
async with client.messages.stream(**request_params) as stream:
async for event in stream:
event_type = getattr(event, 'type', None)
if event_type == "content_block_start":
block = getattr(event, 'content_block', None)
if block and getattr(block, 'type', None) == "tool_use":
current_tool_call = {
"id": getattr(block, 'id', ''),
"name": getattr(block, 'name', ''),
"input": "",
}
elif event_type == "content_block_delta":
delta = getattr(event, 'delta', None)
if delta:
delta_type = getattr(delta, 'type', None)
if delta_type == "text_delta":
text = getattr(delta, 'text', '')
if text:
yield text, None
elif delta_type == "input_json_delta":
if current_tool_call:
partial_json = getattr(delta, 'partial_json', '')
current_tool_call["input"] += partial_json
elif event_type == "content_block_stop":
if current_tool_call:
# Parse the accumulated JSON and yield the tool call
try:
tool_args = json.loads(current_tool_call["input"]) if current_tool_call["input"] else {}
except json.JSONDecodeError:
tool_args = {}
tool_call_dict = {
"function": {
"name": current_tool_call["name"],
"arguments": tool_args,
},
"id": current_tool_call["id"],
}
yield None, [tool_call_dict]
current_tool_call = None
elif event_type == "message_stop":
break
except APITimeoutError as err:
raise HomeAssistantError(
"The generation request timed out! Please check your connection "
"settings, increase the timeout in settings, or decrease the "
"number of exposed entities."
) from err
except APIConnectionError as err:
raise HomeAssistantError(
f"Failed to connect to the Anthropic API: {err}"
) from err
except APIError as err:
raise HomeAssistantError(
f"Anthropic API error: {err}"
) from err
return self._async_stream_parse_completion(
llm_api, agent_id, entity_options, anext_token=anext_token()
)

94
custom_components/llama_conversation/config_flow.py
View File

@@ -152,6 +152,11 @@ from .const import (
BACKEND_TYPE_GENERIC_OPENAI_RESPONSES,
BACKEND_TYPE_LLAMA_CPP_SERVER,
BACKEND_TYPE_OLLAMA,
BACKEND_TYPE_ANTHROPIC,
CONF_ANTHROPIC_API_KEY,
CONF_ANTHROPIC_BASE_URL,
DEFAULT_ANTHROPIC_BASE_URL,
RECOMMENDED_ANTHROPIC_MODELS,
TEXT_GEN_WEBUI_CHAT_MODE_CHAT,
TEXT_GEN_WEBUI_CHAT_MODE_INSTRUCT,
TEXT_GEN_WEBUI_CHAT_MODE_CHAT_INSTRUCT,
@@ -186,7 +191,8 @@ def pick_backend_schema(backend_type=None, selected_language=None):
BACKEND_TYPE_GENERIC_OPENAI,
BACKEND_TYPE_GENERIC_OPENAI_RESPONSES,
BACKEND_TYPE_LLAMA_CPP_SERVER,
BACKEND_TYPE_OLLAMA
BACKEND_TYPE_OLLAMA,
BACKEND_TYPE_ANTHROPIC,
],
translation_key=CONF_BACKEND_TYPE,
multiple=False,
@@ -231,6 +237,18 @@ def remote_connection_schema(backend_type: str, *, host=None, port=None, ssl=Non
}
)
def anthropic_connection_schema(*, api_key=None, base_url=None):
"""Schema for Anthropic API connection configuration."""
return vol.Schema({
vol.Required(CONF_ANTHROPIC_API_KEY, default=api_key if api_key else ""): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
vol.Optional(
CONF_ANTHROPIC_BASE_URL,
default=base_url if base_url else DEFAULT_ANTHROPIC_BASE_URL
): TextSelector(TextSelectorConfig()),
})
class ConfigFlow(BaseConfigFlow, domain=DOMAIN):
"""Handle a config flow for Local LLM Conversation."""
@@ -297,8 +315,12 @@ class ConfigFlow(BaseConfigFlow, domain=DOMAIN):
)
else:
self.internal_step = "configure_connection"
if backend == BACKEND_TYPE_ANTHROPIC:
schema = anthropic_connection_schema()
else:
schema = remote_connection_schema(self.client_config[CONF_BACKEND_TYPE])
return self.async_show_form(
step_id="user", data_schema=remote_connection_schema(self.client_config[CONF_BACKEND_TYPE]), last_step=True
step_id="user", data_schema=schema, last_step=True
)
elif self.install_wheel_error:
errors["base"] = str(self.install_wheel_error)
@@ -339,31 +361,50 @@ class ConfigFlow(BaseConfigFlow, domain=DOMAIN):
self.internal_step = next_step
return self.async_show_progress_done(next_step_id="finish")
elif self.internal_step == "configure_connection":
backend = self.client_config[CONF_BACKEND_TYPE]
if user_input:
self.client_config.update(user_input)
hostname = user_input.get(CONF_HOST, "")
if not is_valid_hostname(hostname):
errors["base"] = "invalid_hostname"
else:
# validate remote connections
connect_err = await BACKEND_TO_CLS[self.client_config[CONF_BACKEND_TYPE]].async_validate_connection(self.hass, self.client_config)
# Anthropic doesn't use hostname - skip hostname validation
if backend == BACKEND_TYPE_ANTHROPIC:
connect_err = await BACKEND_TO_CLS[backend].async_validate_connection(self.hass, self.client_config)
if connect_err:
errors["base"] = "failed_to_connect"
description_placeholders["exception"] = str(connect_err)
else:
return await self.async_step_finish()
return self.async_show_form(
step_id="user",
data_schema=remote_connection_schema(
self.client_config[CONF_BACKEND_TYPE],
else:
hostname = user_input.get(CONF_HOST, "")
if not is_valid_hostname(hostname):
errors["base"] = "invalid_hostname"
else:
# validate remote connections
connect_err = await BACKEND_TO_CLS[backend].async_validate_connection(self.hass, self.client_config)
if connect_err:
errors["base"] = "failed_to_connect"
description_placeholders["exception"] = str(connect_err)
else:
return await self.async_step_finish()
# Use appropriate schema for the backend type
if backend == BACKEND_TYPE_ANTHROPIC:
schema = anthropic_connection_schema(
api_key=self.client_config.get(CONF_ANTHROPIC_API_KEY),
base_url=self.client_config.get(CONF_ANTHROPIC_BASE_URL),
)
else:
schema = remote_connection_schema(
backend,
host=self.client_config.get(CONF_HOST),
port=self.client_config.get(CONF_PORT),
ssl=self.client_config.get(CONF_SSL),
selected_path=self.client_config.get(CONF_GENERIC_OPENAI_PATH)
),
)
return self.async_show_form(
step_id="user",
data_schema=schema,
errors=errors,
description_placeholders=description_placeholders,
last_step=True
@@ -966,6 +1007,29 @@ def local_llama_config_option_schema(
default=DEFAULT_OLLAMA_KEEP_ALIVE_MIN,
): NumberSelector(NumberSelectorConfig(min=-1, max=1440, step=1, unit_of_measurement=UnitOfTime.MINUTES, mode=NumberSelectorMode.BOX)),
})
elif backend_type == BACKEND_TYPE_ANTHROPIC:
result.update({
vol.Required(
CONF_MAX_TOKENS,
description={"suggested_value": options.get(CONF_MAX_TOKENS)},
default=DEFAULT_MAX_TOKENS,
): NumberSelector(NumberSelectorConfig(min=1, max=8192, step=1)),
vol.Required(
CONF_TOP_K,
description={"suggested_value": options.get(CONF_TOP_K)},
default=DEFAULT_TOP_K,
): NumberSelector(NumberSelectorConfig(min=1, max=256, step=1)),
vol.Required(
CONF_TOP_P,
description={"suggested_value": options.get(CONF_TOP_P)},
default=DEFAULT_TOP_P,
): NumberSelector(NumberSelectorConfig(min=0, max=1, step=0.05)),
vol.Required(
CONF_REQUEST_TIMEOUT,
description={"suggested_value": options.get(CONF_REQUEST_TIMEOUT)},
default=DEFAULT_REQUEST_TIMEOUT,
): NumberSelector(NumberSelectorConfig(min=5, max=900, step=1, unit_of_measurement=UnitOfTime.SECONDS, mode=NumberSelectorMode.BOX)),
})
# sort the options
global_order = [

12
custom_components/llama_conversation/const.py
View File

@@ -115,7 +115,19 @@ BACKEND_TYPE_GENERIC_OPENAI = "generic_openai"
BACKEND_TYPE_GENERIC_OPENAI_RESPONSES = "generic_openai_responses"
BACKEND_TYPE_LLAMA_CPP_SERVER = "llama_cpp_server"
BACKEND_TYPE_OLLAMA = "ollama"
BACKEND_TYPE_ANTHROPIC = "anthropic"
DEFAULT_BACKEND_TYPE = BACKEND_TYPE_LLAMA_CPP
# Anthropic-specific configuration
CONF_ANTHROPIC_API_KEY = "anthropic_api_key"
CONF_ANTHROPIC_BASE_URL = "anthropic_base_url"
DEFAULT_ANTHROPIC_BASE_URL = "https://api.anthropic.com"
RECOMMENDED_ANTHROPIC_MODELS = [
"claude-sonnet-4-20250514",
"claude-3-5-sonnet-20241022",
"claude-3-5-haiku-20241022",
"claude-3-opus-20240229",
]
CONF_INSTALLED_LLAMACPP_VERSION = "installed_llama_cpp_version"
CONF_SELECTED_LANGUAGE = "selected_language"
CONF_SELECTED_LANGUAGE_OPTIONS = [ "en", "de", "fr", "es", "pl"]

3
custom_components/llama_conversation/manifest.json
View File

@@ -12,6 +12,7 @@
"requirements": [
"huggingface-hub>=0.23.0",
"webcolors>=24.8.0",
"ollama>=0.5.1"
"ollama>=0.5.1",
"anthropic>=0.40.0"
]
}

8
custom_components/llama_conversation/translations/en.json
View File

@@ -22,7 +22,9 @@
"openai_path": "API Path",
"text_generation_webui_admin_key": "Admin Key",
"text_generation_webui_preset": "Generation Preset/Character Name",
"text_generation_webui_chat_mode": "Chat Mode"
"text_generation_webui_chat_mode": "Chat Mode",
"anthropic_api_key": "Anthropic API Key",
"anthropic_base_url": "Anthropic API Base URL (leave default for official API)"
},
"description": "Please select the backend and/or provide the connection details to connect to the API that is hosting the model.",
"title": "Configure Backend"
@@ -376,8 +378,8 @@
"generic_openai_responses": "Generic OpenAPI Compatible Responses API",
"llama_cpp_server": "Llama.cpp Server",
"ollama": "Ollama API",
"text-generation-webui_api": "text-generation-webui API"
"text-generation-webui_api": "text-generation-webui API",
"anthropic": "Anthropic API (Claude)"
}
},
"text_generation_webui_chat_mode": {