def validate_provider_credentials(self, credentials: dict) -> None: """ Validate provider credentials by making a test API call Parameters: credentials: Provider credentials as defined in `provider_credential_schema` Raises: CredentialsValidateFailedError: If validation fails """ try: # Example implementation: validate using an LLM model instance model_instance = self.get_model_instance(ModelType.LLM) model_instance.validate_credentials( model="example-model", credentials=credentials ) except Exception as ex: logger.exception(f"Credential validation failed") raise CredentialsValidateFailedError(f"Invalid credentials: {str(ex)}")
class XinferenceProvider(Provider): def validate_provider_credentials(self, credentials: dict) -> None: """ For custom-only model providers, a simple implementation is sufficient as validation happens at the model level """ pass
def validate_credentials(self, model: str, credentials: dict) -> None: """ Validate that the provided credentials work with the specified model Parameters: model: The specific model identifier (e.g., "gpt-4") credentials: Authentication details for the model Raises: CredentialsValidateFailedError: If validation fails """ try: # Make a lightweight API call to verify credentials # Example: List available models or check account status response = self._api_client.validate_api_key(credentials["api_key"]) # Verify the specific model is available if applicable if model not in response.get("available_models", []): raise CredentialsValidateFailedError(f"Model {model} is not available") except ApiException as e: raise CredentialsValidateFailedError(str(e))
def get_num_tokens( self, model: str, credentials: dict, prompt_messages: list[PromptMessage], tools: Optional[list[PromptMessageTool]] = None) -> int: """ Calculate the number of tokens in the prompt """ # Convert prompt_messages to the format expected by the tokenizer text = self._convert_messages_to_text(prompt_messages) try: # Use the appropriate tokenizer for this model tokenizer = self._get_tokenizer(model) return len(tokenizer.encode(text)) except Exception: # Fall back to a generic tokenizer return self._get_num_tokens_by_gpt2(text)
def get_customizable_model_schema( self, model: str, credentials: dict) -> Optional[AIModelEntity]: """ Get parameter schema for custom models """ # For fine-tuned models, you might return the base model's schema if model.startswith("ft:"): base_model = self._extract_base_model(model) return self._get_predefined_model_schema(base_model) # For standard models, return None to use the predefined schema return None
def get_num_tokens( self, model: str, credentials: dict, texts: list[str]) -> int: """ Calculate the number of tokens in the texts to be embedded """ # Join all texts to estimate token count combined_text = " ".join(texts) try: # Use the appropriate tokenizer for this model tokenizer = self._get_tokenizer(model) return len(tokenizer.encode(combined_text)) except Exception: # Fall back to a generic tokenizer return self._get_num_tokens_by_gpt2(combined_text)
def _invoke( self, model: str, credentials: dict, query: str, docs: list[str], score_threshold: Optional[float] = None, top_n: Optional[int] = None, user: Optional[str] = None) -> RerankResult: """ Rerank documents based on relevance to the query """ # Set up API client with credentials client = self._get_client(credentials) # Prepare request data request_data = { "query": query, "documents": docs, } # Call reranking API endpoint response = client.rerank( model=model, **request_data, user=user ) # Process results ranked_results = [] for i, result in enumerate(response.results): # Create RerankDocument for each result doc = RerankDocument( index=result.document_index, # Original index in docs list text=docs[result.document_index], # Original text score=result.relevance_score # Relevance score ) ranked_results.append(doc) # Sort by score in descending order ranked_results.sort(key=lambda x: x.score, reverse=True) # Apply score threshold filtering if specified if score_threshold is not None: ranked_results = [doc for doc in ranked_results if doc.score >= score_threshold] # Apply top_n limit if specified if top_n is not None and top_n > 0: ranked_results = ranked_results[:top_n] return RerankResult( model=model, docs=ranked_results )
def _invoke( self, model: str, credentials: dict, file: IO[bytes], user: Optional[str] = None) -> str: """ Convert speech audio to text """ # Set up API client with credentials client = self._get_client(credentials) try: # Determine the file format file_format = self._detect_audio_format(file) # Prepare the file for API submission # Most APIs require either a file path or binary data audio_data = file.read() # Call the speech-to-text API response = client.audio.transcriptions.create( model=model, file=("audio.mp3", audio_data), # Adjust filename based on actual format user=user ) # Extract and return the transcribed text return response.text except Exception as e: # Map to appropriate error type self._handle_api_error(e) finally: # Reset file pointer for potential reuse file.seek(0)
def _detect_audio_format(self, file: IO[bytes]) -> str: """ Detect the audio format based on file header """ # Read the first few bytes to check the file signature header = file.read(12) file.seek(0) # Reset file pointer # Check for common audio format signatures if header.startswith(b'RIFF') and header[8:12] == b'WAVE': return 'wav' elif header.startswith(b'ID3') or header.startswith(b'\xFF\xFB'): return 'mp3' elif header.startswith(b'OggS'): return 'ogg' elif header.startswith(b'fLaC'): return 'flac' else: # Default or additional format checks return 'mp3' # Default assumption
def _invoke( self, model: str, credentials: dict, text: str, user: Optional[str] = None) -> bool: """ Analyze text for harmful content Returns: bool: False if the text is safe, True if it contains harmful content """ # Set up API client with credentials client = self._get_client(credentials) try: # Call moderation API response = client.moderations.create( model=model, input=text, user=user ) # Check if any categories were flagged result = response.results[0] # Return True if flagged in any category, False if safe return result.flagged except Exception as e: # Log the error but default to safe if there's an API issue. # This is a conservative approach; production systems might want # different fallback behavior. logger.error(f"Moderation API error: {str(e)}") return False
def _invoke( self, model: str, credentials: dict, text: str, user: Optional[str] = None) -> bool: """ Analyze text for harmful content with detailed category checking """ # Set up API client with credentials client = self._get_client(credentials) try: # Call moderation API response = client.moderations.create( model=model, input=text, user=user ) # Get detailed category results result = response.results[0] categories = result.categories # Check specific categories based on your application's needs # For example, you might want to flag certain categories but not others critical_violations = [ categories.harassment, categories.hate, categories.self_harm, categories.sexual, categories.violence ] # Flag content if any critical category is violated return any(critical_violations) except Exception as e: self._handle_api_error(e) # Default to safe in case of error return False
class TextPromptMessageContent(PromptMessageContent): """ Model class for text prompt message content. """ type: PromptMessageContentType = PromptMessageContentType.TEXT
class ImagePromptMessageContent(PromptMessageContent): """ Model class for image prompt message content. """ class DETAIL(Enum): LOW = 'low' HIGH = 'high' type: PromptMessageContentType = PromptMessageContentType.IMAGE detail: DETAIL = DETAIL.LOW # Resolution
class PromptMessage(ABC, BaseModel): """ Model class for prompt message. """ role: PromptMessageRole # Message role content: Optional[str | list[PromptMessageContent]] = None # Either a string or a content list; the list form supports multimodal input, see PromptMessageContent name: Optional[str] = None # Optional name
class AssistantPromptMessage(PromptMessage): """ Model class for assistant prompt message. """ class ToolCall(BaseModel): """ Model class for assistant prompt message tool call. """ class ToolCallFunction(BaseModel): """ Model class for assistant prompt message tool call function. """ name: str # Tool name arguments: str # Tool parameters id: str # Tool call ID; only meaningful for OpenAI tool calls. Uniquely identifies one invocation, since the same tool can be called multiple times type: str # Defaults to "function" function: ToolCallFunction # Tool call information role: PromptMessageRole = PromptMessageRole.ASSISTANT tool_calls: list[ToolCall] = [] # Model's tool call results (only returned when tools are passed in and the model decides to call them)
class ToolPromptMessage(PromptMessage): """ Model class for tool prompt message. """ role: PromptMessageRole = PromptMessageRole.TOOL tool_call_id: str # Tool call ID; if the provider doesn't support OpenAI tool calls, you can pass the tool name instead
class LLMResult(BaseModel): """ Model class for llm result. """ model: str # Model actually used prompt_messages: list[PromptMessage] # Prompt message list message: AssistantPromptMessage # Reply message usage: LLMUsage # Token usage and cost information system_fingerprint: Optional[str] = None # Request fingerprint; see OpenAI's parameter definition
class LLMResultChunkDelta(BaseModel): """ Model class for llm result chunk delta. """ index: int # Sequence number message: AssistantPromptMessage # Reply message usage: Optional[LLMUsage] = None # Token usage and cost information; only returned in the last chunk finish_reason: Optional[str] = None # Completion reason; only returned in the last chunk
class LLMResultChunk(BaseModel): """ Model class for llm result chunk. """ model: str # Model actually used prompt_messages: list[PromptMessage] # Prompt message list system_fingerprint: Optional[str] = None # Request fingerprint; see OpenAI's parameter definition delta: LLMResultChunkDelta # Content changes in this chunk
class LLMUsage(ModelUsage): """ Model class for llm usage. """ prompt_tokens: int # Tokens used by the prompt prompt_unit_price: Decimal # Prompt unit price prompt_price_unit: Decimal # Prompt price unit, i.e., the number of tokens the unit price applies to prompt_price: Decimal # Prompt cost completion_tokens: int # Tokens used by the completion completion_unit_price: Decimal # Completion unit price completion_price_unit: Decimal # Completion price unit, i.e., the number of tokens the unit price applies to completion_price: Decimal # Completion cost total_tokens: int # Total tokens used total_price: Decimal # Total cost currency: str # Currency unit latency: float # Request latency in seconds
class TextEmbeddingResult(BaseModel): """ Model class for text embedding result. """ model: str # Model actually used embeddings: list[list[float]] # Embedding vectors, in the same order as the input texts usage: EmbeddingUsage # Usage information
class EmbeddingUsage(ModelUsage): """ Model class for embedding usage. """ tokens: int # Tokens used total_tokens: int # Total tokens used unit_price: Decimal # Unit price price_unit: Decimal # Price unit, i.e., the number of tokens the unit price applies to total_price: Decimal # Total cost currency: str # Currency unit latency: float # Request latency in seconds
class RerankResult(BaseModel): """ Model class for rerank result. """ model: str # Model actually used docs: list[RerankDocument] # List of reranked documents
class RerankDocument(BaseModel): """ Model class for rerank document. """ index: int # Index in the original docs list text: str # Document text score: float # Relevance score
