import logging from typing import Dict, Any, List, Union, Optional from modules.connectors.connectorAiOpenai import AiOpenai from modules.connectors.connectorAiAnthropic import AiAnthropic logger = logging.getLogger(__name__) class AiCalls: """Interface for AI service interactions""" def __init__(self): self.openaiService = AiOpenai() self.anthropicService = AiAnthropic() async def callAiTextBasic(self, prompt: str, context: Optional[str] = None) -> str: """ Basic text processing using OpenAI. Args: prompt: The user prompt to process context: Optional system context/prompt Returns: The AI response as text """ # Prepare messages in OpenAI format messages = [] # Add system message if context provided if context: messages.append({ "role": "system", "content": context }) # Add user message messages.append({ "role": "user", "content": prompt }) # Add language instruction for user-facing responses if hasattr(self, 'userLanguage') and self.userLanguage: ltext = f"Please respond in '{self.userLanguage}' language." if messages and messages[0]["role"] == "system": if "language" not in messages[0]["content"].lower(): messages[0]["content"] = f"{ltext} {messages[0]['content']}" else: messages.insert(0, { "role": "system", "content": ltext }) try: return await self.openaiService.callAiBasic(messages) except Exception as e: logger.error(f"Error in OpenAI call: {str(e)}") return f"Error: {str(e)}" async def callAiTextAdvanced(self, prompt: str, context: Optional[str] = None) -> str: """ Advanced text processing using Anthropic. Args: prompt: The user prompt to process context: Optional system context/prompt Returns: The AI response as text """ # Prepare messages in OpenAI format messages = [] # Add system message if context provided if context: messages.append({ "role": "system", "content": context }) # Add user message messages.append({ "role": "user", "content": prompt }) # Add language instruction for user-facing responses if hasattr(self, 'userLanguage') and self.userLanguage: ltext = f"Please respond in '{self.userLanguage}' language." if messages and messages[0]["role"] == "system": if "language" not in messages[0]["content"].lower(): messages[0]["content"] = f"{ltext} {messages[0]['content']}" else: messages.insert(0, { "role": "system", "content": ltext }) try: response = await self.anthropicService.callAiBasic(messages) return response["choices"][0]["message"]["content"] except Exception as e: logger.error(f"Error in Anthropic call: {str(e)}") return f"Error: {str(e)}" async def callAiImageBasic(self, prompt: str, imageData: Union[str, bytes], mimeType: str = None) -> str: """ Basic image processing using OpenAI. Args: prompt: The prompt for image analysis imageData: The image data (file path or bytes) mimeType: Optional MIME type of the image Returns: The AI response as text """ try: return await self.openaiService.callAiImage(imageData, mimeType, prompt) except Exception as e: logger.error(f"Error in OpenAI image call: {str(e)}") return f"Error: {str(e)}" async def callAiImageAdvanced(self, prompt: str, imageData: Union[str, bytes], mimeType: str = None) -> str: """ Advanced image processing using Anthropic. Args: prompt: The prompt for image analysis imageData: The image data (file path or bytes) mimeType: Optional MIME type of the image Returns: The AI response as text """ try: return await self.anthropicService.callAiImage(prompt, imageData, mimeType) except Exception as e: logger.error(f"Error in Anthropic image call: {str(e)}") return f"Error: {str(e)}"