import logging import httpx import asyncio import re from typing import Dict, Any, List, Union, Optional from fastapi import HTTPException from modules.shared.configuration import APP_CONFIG # Configure logger logger = logging.getLogger(__name__) def loadConfigData(): """Load configuration data for LangDoc connector""" return { "apiKey": APP_CONFIG.get('Connector_AiLangdoc_API_SECRET'), "apiUrl": APP_CONFIG.get('Connector_AiLangdoc_API_URL'), "modelName": APP_CONFIG.get('Connector_AiLangdoc_MODEL_NAME'), "temperature": float(APP_CONFIG.get('Connector_AiLangdoc_TEMPERATURE')), "maxTokens": int(APP_CONFIG.get('Connector_AiLangdoc_MAX_TOKENS')) } class AiLangdoc: """Connector for communication with the LangDoc API (OpenAI-compatible).""" def __init__(self): # Load configuration self.config = loadConfigData() self.apiKey = self.config["apiKey"] self.apiUrl = self.config["apiUrl"] self.modelName = self.config["modelName"] # HttpClient for API calls self.httpClient = httpx.AsyncClient( timeout=120.0, # Longer timeout for complex requests headers={ "Authorization": f"Bearer {self.apiKey}", "Content-Type": "application/json" } ) logger.info(f"LangDoc Connector initialized with model: {self.modelName}") async def callAiBasic(self, messages: List[Dict[str, Any]], temperature: float = None, maxTokens: int = None) -> str: """ Calls the LangDoc API with the given messages. Args: messages: List of messages in OpenAI format (role, content) temperature: Temperature for response generation (0.0-1.0) maxTokens: Maximum number of tokens in the response Returns: The response from the LangDoc API Raises: HTTPException: For errors in API communication """ try: # Use parameters from configuration if none were overridden if temperature is None: temperature = self.config.get("temperature", 0.2) if maxTokens is None: maxTokens = self.config.get("maxTokens", 2000) payload = { "model": self.modelName, "messages": messages, "temperature": temperature, "max_tokens": maxTokens } response = await self.httpClient.post( self.apiUrl, json=payload ) if response.status_code != 200: error_detail = f"LangDoc API error: {response.status_code} - {response.text}" logger.error(error_detail) # Provide more specific error messages based on status code if response.status_code == 429: error_message = "Rate limit exceeded. Please wait before making another request." elif response.status_code == 401: error_message = "Invalid API key. Please check your LangDoc API configuration." elif response.status_code == 400: error_message = f"Invalid request to LangDoc API: {response.text}" else: error_message = f"LangDoc API error ({response.status_code}): {response.text}" raise HTTPException(status_code=500, detail=error_message) responseJson = response.json() content = responseJson["choices"][0]["message"]["content"] return content except Exception as e: logger.error(f"Error calling LangDoc API: {str(e)}") raise HTTPException(status_code=500, detail=f"Error calling LangDoc API: {str(e)}") async def callAiImage(self, prompt: str, imageData: Union[str, bytes], mimeType: str = None) -> str: """ Analyzes an image using LangDoc's vision capabilities. Args: imageData: Either a file path (str) or image data (bytes) mimeType: The MIME type of the image (optional, only for binary data) prompt: The prompt for analysis Returns: The analysis response as text """ try: # Distinguish between file path and binary data if isinstance(imageData, str): # It's a file path - import filehandling only when needed from modules import agentserviceFilemanager as fileHandler base64Data, autoMimeType = fileHandler.encodeFileToBase64(imageData) mimeType = mimeType or autoMimeType else: # It's binary data import base64 base64Data = base64.b64encode(imageData).decode('utf-8') # MIME type must be specified for binary data if not mimeType: # Fallback to generic image type mimeType = "image/png" # Prepare the payload for the Vision API messages = [ { "role": "user", "content": [ {"type": "text", "text": prompt}, { "type": "image_url", "image_url": { "url": f"data:{mimeType};base64,{base64Data}" } } ] } ] # Use the existing callAiBasic function response = await self.callAiBasic(messages) return response except Exception as e: logger.error(f"Error during image analysis: {str(e)}", exc_info=True) return f"[Error during image analysis: {str(e)}]" async def listModels(self) -> List[Dict[str, Any]]: """ Lists available models from the LangDoc API. Returns: List of available models with their details """ try: # LangDoc uses OpenAI-compatible endpoints modelsUrl = self.apiUrl.replace("/chat/completions", "/models") response = await self.httpClient.get(modelsUrl) if response.status_code != 200: error_detail = f"LangDoc API error listing models: {response.status_code} - {response.text}" logger.error(error_detail) raise HTTPException(status_code=500, detail=error_detail) responseJson = response.json() return responseJson.get("data", []) except Exception as e: logger.error(f"Error listing LangDoc models: {str(e)}") raise HTTPException(status_code=500, detail=f"Error listing LangDoc models: {str(e)}") async def getModelInfo(self, modelName: str = None) -> Dict[str, Any]: """ Gets information about a specific model. Args: modelName: Name of the model to get info for (uses default if None) Returns: Model information dictionary """ try: if modelName is None: modelName = self.modelName models = await self.listModels() for model in models: if model.get("id") == modelName: return model raise HTTPException(status_code=404, detail=f"Model {modelName} not found") except Exception as e: logger.error(f"Error getting LangDoc model info: {str(e)}") raise HTTPException(status_code=500, detail=f"Error getting LangDoc model info: {str(e)}") async def generateImage(self, prompt: str, size: str = "1024x1024", quality: str = "standard", style: str = "vivid") -> Dict[str, Any]: """ Generates an image using LangDoc's DALL-E 3 integration. Args: prompt: Text description of the image to generate size: Image size - "1024x1024", "1792x1024", or "1024x1792" quality: Image quality - "standard" or "hd" style: Image style - "vivid" or "natural" Returns: Dictionary containing the generated image data and metadata Raises: HTTPException: For errors in API communication """ try: # Use OpenAI-compatible images endpoint imagesUrl = self.apiUrl.replace("/chat/completions", "/images/generations") payload = { "model": "dall-e-3", "prompt": prompt, "size": size, "quality": quality, "style": style, "n": 1 } response = await self.httpClient.post( imagesUrl, json=payload ) if response.status_code != 200: error_detail = f"LangDoc Image Generation API error: {response.status_code} - {response.text}" logger.error(error_detail) # Provide more specific error messages if response.status_code == 429: error_message = "Rate limit exceeded for image generation. Please wait before making another request." elif response.status_code == 401: error_message = "Invalid API key for image generation. Please check your LangDoc API configuration." elif response.status_code == 400: error_message = f"Invalid request to LangDoc Image API: {response.text}" else: error_message = f"LangDoc Image API error ({response.status_code}): {response.text}" raise HTTPException(status_code=500, detail=error_message) responseJson = response.json() # Extract image data imageData = responseJson.get("data", []) if not imageData: raise HTTPException(status_code=500, detail="No image data returned from LangDoc API") imageInfo = imageData[0] return { "success": True, "image_url": imageInfo.get("url"), "revised_prompt": imageInfo.get("revised_prompt"), "size": size, "quality": quality, "style": style, "model": "dall-e-3", "created": responseJson.get("created"), "raw_response": responseJson } except Exception as e: logger.error(f"Error generating image with LangDoc: {str(e)}") raise HTTPException(status_code=500, detail=f"Error generating image with LangDoc: {str(e)}") async def generateImageWithVariations(self, prompt: str, variations: int = 1, size: str = "1024x1024", quality: str = "standard", style: str = "vivid") -> List[Dict[str, Any]]: """ Generates multiple image variations using LangDoc's DALL-E 3 integration. Args: prompt: Text description of the image to generate variations: Number of variations to generate (1-4) size: Image size - "1024x1024", "1792x1024", or "1024x1792" quality: Image quality - "standard" or "hd" style: Image style - "vivid" or "natural" Returns: List of dictionaries containing generated image data and metadata Raises: HTTPException: For errors in API communication """ try: # Limit variations to reasonable number variations = min(max(variations, 1), 4) # Use OpenAI-compatible images endpoint imagesUrl = self.apiUrl.replace("/chat/completions", "/images/generations") results = [] # Generate multiple variations by making multiple API calls for i in range(variations): # Add variation to prompt to get different results variationPrompt = f"{prompt} (variation {i+1})" payload = { "model": "dall-e-3", "prompt": variationPrompt, "size": size, "quality": quality, "style": style, "n": 1 } response = await self.httpClient.post( imagesUrl, json=payload ) if response.status_code != 200: logger.warning(f"Failed to generate variation {i+1}: {response.status_code} - {response.text}") continue responseJson = response.json() imageData = responseJson.get("data", []) if imageData: imageInfo = imageData[0] results.append({ "variation": i + 1, "image_url": imageInfo.get("url"), "revised_prompt": imageInfo.get("revised_prompt"), "size": size, "quality": quality, "style": style, "model": "dall-e-3", "created": responseJson.get("created") }) # Add small delay between requests to avoid rate limiting if i < variations - 1: await asyncio.sleep(1) return results except Exception as e: logger.error(f"Error generating image variations with LangDoc: {str(e)}") raise HTTPException(status_code=500, detail=f"Error generating image variations with LangDoc: {str(e)}") async def generateImageWithChat(self, prompt: str, size: str = "1024x1024", quality: str = "standard", style: str = "vivid") -> str: """ Generates an image using LangDoc's chat interface with image generation tools. This method uses the chat completions endpoint with image generation capabilities. Args: prompt: Text description of the image to generate size: Image size - "1024x1024", "1792x1024", or "1024x1792" quality: Image quality - "standard" or "hd" style: Image style - "vivid" or "natural" Returns: Response text from the chat model (may include image references) Raises: HTTPException: For errors in API communication """ try: # Create a prompt that requests image generation imagePrompt = f"Please generate an image with the following description: {prompt}. Size: {size}, Quality: {quality}, Style: {style}" messages = [ { "role": "user", "content": imagePrompt } ] # Use the chat completions endpoint response = await self.callAiBasic(messages) return response except Exception as e: logger.error(f"Error generating image with chat: {str(e)}") raise HTTPException(status_code=500, detail=f"Error generating image with chat: {str(e)}") async def _testConnection(self) -> bool: """ Tests the connection to the LangDoc API. Returns: True if connection is successful, False otherwise """ try: # Try to list models as a simple connection test await self.listModels() return True except Exception as e: logger.error(f"LangDoc connection test failed: {str(e)}") return False