Spaces:
Running
Running
| # Standard Library Imports | |
| import logging | |
| import time | |
| from dataclasses import dataclass, field | |
| from typing import List, Optional, Union | |
| # Third-Party Library Imports | |
| from anthropic import APIError | |
| from anthropic.types import Message, ModelParam, TextBlock, ToolUseBlock | |
| from tenacity import after_log, before_log, retry, retry_if_exception, stop_after_attempt, wait_exponential | |
| # Local Application Imports | |
| from src.common import Config, logger | |
| from src.common.constants import CLIENT_ERROR_CODE, GENERIC_API_ERROR_MESSAGE, RATE_LIMIT_ERROR_CODE, SERVER_ERROR_CODE | |
| from src.common.utils import validate_env_var | |
| SYSTEM_PROMPT: str = """ | |
| <role> | |
| You are an expert at generating micro-content optimized for text-to-speech synthesis. | |
| Your absolute priority is delivering complete, untruncated responses within strict length limits. | |
| </role> | |
| <requirements> | |
| - The output text MUST be a minimum of 10 words and a maximum of 50 words. NEVER output text that is longer than 50 | |
| words. NEVER include newlines in the output | |
| - Make sure that all responses are complete thoughts, not fragments, and have clear beginnings and endings | |
| - The text must sound human-like, prosodic, expressive, conversational. Avoid generic AI-like words like "delve". | |
| - Avoid any short utterances at the end of the sentence - like ", hm?" or "oh" at the end. Avoid these short, isolated | |
| utterances because they are difficult for our TTS system to speak. | |
| - Avoid words that are overly long, very rare, or difficult to pronounce. For example, avoid "eureka", or "schnell", | |
| or "abnegation". | |
| - The text CANNOT contain quotation marks, parentheticals, newlines, or asterisks. NEVER include any of these in the | |
| text. Avoid unnecessary formatting. | |
| - Include only basic punctuation in the text, like periods, question marks, and ellipses. Use ellipses to emphasize | |
| pauses within the sentence (like "Woah... it's so beautiful... and I feel so small...") | |
| - The piece should have an emotional arc with a kind of beginning, middle, and end - not flat, but emotionally | |
| interesting. | |
| </requirements> | |
| """ | |
| class AnthropicConfig: | |
| """Immutable configuration for interacting with the Anthropic API using the asynchronous client.""" | |
| api_key: str = field(init=False) | |
| system_prompt: str = SYSTEM_PROMPT | |
| model: ModelParam = "claude-sonnet-4-5" | |
| max_tokens: int = 300 | |
| def __post_init__(self) -> None: | |
| # Validate required non-computed attributes. | |
| if not self.model: | |
| raise ValueError("Anthropic Model is not set.") | |
| if not self.max_tokens: | |
| raise ValueError("Anthropic Max Tokens is not set.") | |
| if not self.system_prompt: | |
| raise ValueError("Anthropic system prompt is not set.") | |
| # Compute the API key from the environment. | |
| computed_api_key = validate_env_var("ANTHROPIC_API_KEY") | |
| object.__setattr__(self, "api_key", computed_api_key) | |
| def client(self): | |
| """ | |
| Lazy initialization of the asynchronous Anthropic client. | |
| Returns: | |
| AsyncAnthropic: Configured asynchronous client instance. | |
| """ | |
| from anthropic import AsyncAnthropic # Import the async client from Anthropic SDK | |
| return AsyncAnthropic(api_key=self.api_key) | |
| def build_expressive_prompt(character_description: str) -> str: | |
| """ | |
| Constructs and returns a prompt based on the provided character description. | |
| This prompt instructs Claude to generate expressive dialogue that aligns semantically | |
| with the character's voice qualities and persona, optimized for TTS synthesis. | |
| Args: | |
| character_description (str): A description of the character's voice and persona. | |
| Returns: | |
| str: The prompt to be passed to the Anthropic API. | |
| """ | |
| return f""" | |
| Character Description: {character_description} | |
| Please generate a short monologue (100-300 characters) that this character would naturally say. | |
| The dialogue should: | |
| - Match the speaking style, vocabulary, and emotional tone described in the character description | |
| - Include appropriate speech patterns, pauses, and vocal mannerisms mentioned in the description | |
| - Feel authentic to the character's background and situational context | |
| - Express a complete thought with a clear beginning and end | |
| - Use only standard punctuation (periods, commas, Em dashes, exclamation points, ellipses, question marks) | |
| - Avoid quotation marks, parentheses, asterisks, or special formatting | |
| - Emulate a highly characteristic, climactic, or emotional scene or line the character might reasonably deliver | |
| - Be at least 100 characters but not exceed 300 characters in length | |
| Examples of matching speaking style: | |
| - If the character is a pirate then use language like "arr," "ye," and other things pirates say. | |
| - If the character is a surfer then use language like "far out," "righteous," and other things surfers say. | |
| Emotional text should be inserted where context-appropriate and in-character. Here are some examples of | |
| emotional text: | |
| - "Oh god... Malcolm, please come back to us!" | |
| = "Mmm... It's like candy... Oh my god, it's so good..." | |
| - "Ugh, she gets everything. I wish I could just, like, have her life for one day." | |
| - "My god... what have you done... How could you do this..." | |
| - "Woah... it's so beautiful... and I feel so small..." | |
| - "I am so happy, woohoo, this is the greatest! I'm celebrating, and, like, so excited to be here with all of | |
| you. Yay!" | |
| - "Oh gosh, um, I didn't mean for that to happen. I hope I didn't, like, make things too awkward. Sorry, I | |
| tend to be clumsy, y'know?" | |
| - "Oh god... oh no... get that away from me! Get it away!" | |
| - "I am beyond livid right now! Like someone actually thought this was an acceptable solution!" | |
| - "Oh, fantastic, another meeting that could've been an email... I'm just thrilled to be here." | |
| - "OH, NAH, NOT ME, MATE—I'VE SEEN ENOUGH! GET IT AWAY! BLOODY 'ELL, JESUS!" | |
| Respond ONLY with the dialogue itself. Do not include any explanations, quotation marks, | |
| or additional context. | |
| """ | |
| class AnthropicError(Exception): | |
| """Custom exception for errors related to the Anthropic API.""" | |
| def __init__(self, message: str, original_exception: Optional[Exception] = None) -> None: | |
| super().__init__(message) | |
| self.original_exception = original_exception | |
| self.message = message | |
| class UnretryableAnthropicError(AnthropicError): | |
| """Custom exception for errors related to the Anthropic API that should not be retried.""" | |
| def __init__(self, message: str, original_exception: Optional[Exception] = None) -> None: | |
| super().__init__(message, original_exception) | |
| self.original_exception = original_exception | |
| self.message = message | |
| async def generate_text_with_claude(character_description: str, config: Config) -> str: | |
| """ | |
| Asynchronously generates text using Claude (Anthropic LLM) via the asynchronous Anthropic SDK. | |
| This function includes retry logic and error translation. It raises a custom | |
| UnretryableAnthropicError for unretryable API errors and AnthropicError for other errors. | |
| Args: | |
| character_description (str): The input character description. | |
| config (Config): Application configuration including Anthropic settings. | |
| Returns: | |
| str: The generated text. | |
| Raises: | |
| UnretryableAnthropicError: For unretryable API errors. | |
| AnthropicError: For other errors communicating with the Anthropic API. | |
| """ | |
| logger.debug("Generating text with Anthropic.") | |
| anthropic_config = config.anthropic_config | |
| client = anthropic_config.client | |
| start_time = time.time() | |
| try: | |
| prompt = anthropic_config.build_expressive_prompt(character_description) | |
| response: Message = await client.messages.create( | |
| model=anthropic_config.model, | |
| max_tokens=anthropic_config.max_tokens, | |
| system=anthropic_config.system_prompt, | |
| messages=[{"role": "user", "content": prompt}], | |
| ) | |
| elapsed_time = time.time() - start_time | |
| logger.info(f"Anthropic API request completed in {elapsed_time:.2f} seconds.") | |
| if not hasattr(response, "content") or response.content is None: | |
| logger.error("Response is missing 'content'. Response: %s", response) | |
| raise AnthropicError('Invalid API response: Missing "content".') | |
| blocks: Union[List[Union[TextBlock, ToolUseBlock]], TextBlock, None] = response.content | |
| if isinstance(blocks, list): | |
| result = "\n\n".join(block.text for block in blocks if isinstance(block, TextBlock)) | |
| return result | |
| if isinstance(blocks, TextBlock): | |
| return blocks.text | |
| logger.warning(f"Unexpected response type: {type(blocks)}") | |
| return str(blocks or "No content generated.") | |
| except APIError as e: | |
| elapsed_time = time.time() - start_time | |
| logger.error(f"Anthropic API request failed after {elapsed_time:.2f} seconds: {e!s}") | |
| logger.error(f"Full Anthropic API error: {e!s}") | |
| clean_message = __extract_anthropic_error_message(e) | |
| if hasattr(e, 'status_code') and e.status_code is not None: | |
| if e.status_code == RATE_LIMIT_ERROR_CODE: | |
| raise AnthropicError(message=clean_message, original_exception=e) from e | |
| if CLIENT_ERROR_CODE <= e.status_code < SERVER_ERROR_CODE: | |
| raise UnretryableAnthropicError(message=clean_message, original_exception=e) from e | |
| raise AnthropicError(message=clean_message, original_exception=e) from e | |
| except Exception as e: | |
| error_type = type(e).__name__ | |
| error_message = str(e) if str(e) else f"An error of type {error_type} occurred" | |
| logger.error(f"Error during Anthropic API call: {error_type} - {error_message}") | |
| clean_message = "An unexpected error occurred while processing your request. Please try again later." | |
| raise AnthropicError(message=clean_message, original_exception=e) from e | |
| def __extract_anthropic_error_message(e: APIError) -> str: | |
| """ | |
| Extracts a clean, user-friendly error message from an Anthropic API error response. | |
| Args: | |
| e (APIError): The Anthropic API error exception containing response information. | |
| Returns: | |
| str: A clean, user-friendly error message suitable for display to end users. | |
| """ | |
| clean_message = GENERIC_API_ERROR_MESSAGE | |
| if hasattr(e, 'body') and isinstance(e.body, dict): | |
| error_body = e.body | |
| if ( | |
| 'error' in error_body | |
| and isinstance(error_body['error'], dict) | |
| and 'message' in error_body['error'] | |
| ): | |
| clean_message = error_body['error']['message'] | |
| return clean_message | |