importos\nimportjson\nimportlogging\nfromanthropicimportAnthropic\nfrombase_telegram_inference_botimportBaseTelegramInferenceBot\nfromtelegram_helperimportTelegramHelper\n\n# logging.basicConfig(level=logging.INFO) # Usually configured in main execution script\n\nclass AnthropicTelegramInferenceBot(BaseTelegramInferenceBot):\n def __init__(self):\n super().__init__()\n self.anthropic_client = Anthropic(api_key=os.environ.get(\"ANTHROPIC_API_KEY\"))\n # Note: default_headers for max_tokens with older models might be needed.\n # For Claude 3.5 Sonnet, max_tokens is a top-level param in messages.create\n \n # Configure model and tokens. Using Sonnet 3.5 as default.\n # ANTHROPIC_MODEL and ANTHROPIC_MAX_TOKENS would be new ENVs.\n self._configure_model_and_tokens(\n os.environ.get(\"ANTHROPIC_MODEL\", \"claude-3-5-sonnet-20240620\"),\n os.environ.get(\"ANTHROPIC_MAX_TOKENS\", \"4096\") # Default max tokens for Sonnet 3.5\n )\n\n def _configure_model_and_tokens(self, model_name, max_tokens_str, default_max_tokens=4096):\n self.model = model_name if model_name else \"claude-3-5-sonnet-20240620\"\n try:\n # Anthropic\'s max_tokens is an integer.\n self.max_tokens = int(max_tokens_str) if max_tokens_str is not None else default_max_tokens\n except ValueError:\n logging.error(f\"Invalid value for Anthropic max_tokens: {max_tokens_str}. Using default {default_max_tokens}.\")\n self.max_tokens = default_max_tokens\n logging.info(f\"Configured to use Anthropic model: {self.model} with max_tokens: {self.max_tokens}\")\n\n def get_system_prompt_description(self) -> str:\n system_prompt_path = os.getenv(\"SYSTEM_PROMPT_PATH\")\n if system_prompt_path and os.path.isfile(system_prompt_path):\n return f\"System Prompt File: {os.path.basename(system_prompt_path)}\"\n elif system_prompt_path:\n return f\"System Prompt File: {os.path.basename(system_prompt_path)} (Not found at path: {system_prompt_path})\"\n else:\n return \"System Prompt File: Not configured (SYSTEM_PROMPT_PATH not set).\"\n\n def get_llm_description(self) -> str:\n return f\"LLM: {self.model}, Max Tokens: {self.max_tokens}\"\n\n def get_chat_response(self, messages_history):\n # Anthropic expects messages in a specific format.\n # Filter out any non-compliant messages or transform if necessary.\n # For simplicity, assume messages_history is already compliant.\n \n # Ensure self.system_prompt is a string, not None.\n current_system_prompt = self.system_prompt if self.system_prompt else \"\"\n\n anthropic_tools = []\n if hasattr(self, \'functions\') and self.functions:\n anthropic_tools = [\n {\n \"name\": function[\'name\'],\n \"description\": function[\'description\'],\n \"input_schema\": function[\'parameters\'] if function[\'parameters\'] not in [None, {}] else {\"type\": \"object\", \"properties\": {}}\n }\n for function in self.functions\n ]\n \n try:\n response = self.anthropic_client.messages.create(\n model=self.model,\n system=current_system_prompt, # System prompt is a top-level parameter\n messages=messages_history, # User/assistant turns\n max_tokens=self.max_tokens, # Use configured max_tokens\n tools=anthropic_tools if anthropic_tools else None,\n tool_choice={\"type\": \"auto\"} if anthropic_tools else None\n )\n return response\n except Exception as e:\n logging.error(f\"Anthropic API call failed: {e}\")\n # Consider how to propagate this error: re-raise or return a specific error structure\n raise # Re-raising for now\n\n async def handle_message(self, use
