141 lines
4.8 KiB
Python
141 lines
4.8 KiB
Python
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(prompt, imageData, mimeType)
|
|
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)}"
|
|
|