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)}")
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 _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
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
当传入文本和图片时，图片需要构造成此实体作为 content 列表的一部分。
data 可以是 url 或图片的 base64 编码字符串。
class PromptMessage(ABC, BaseModel): """ Model class for prompt message. """ role: PromptMessageRole # Message role content: Optional[str | list[PromptMessageContent]] = None # Supports two types: string and content list. The content list is for multimodal needs, see PromptMessageContent for details. name: Optional[str] = None # Name, optional.
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 ID, only effective for OpenAI tool call, a unique ID for tool invocation, the same tool can be called multiple times type: str # Default is 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 OpenAI tool call is not supported, you can also pass in the tool name
class LLMResult(BaseModel): """ Model class for llm result. """ model: str # Actually used model prompt_messages: list[PromptMessage] # Prompt message list message: AssistantPromptMessage # Reply message usage: LLMUsage # Tokens used and cost information system_fingerprint: Optional[str] = None # Request fingerprint, refer to OpenAI parameter definition
class LLMResultChunkDelta(BaseModel): """ Model class for llm result chunk delta. """ index: int # Sequence number message: AssistantPromptMessage # Reply message usage: Optional[LLMUsage] = None # Tokens used and cost information, only returned in the last message finish_reason: Optional[str] = None # Completion reason, only returned in the last message
class LLMResultChunk(BaseModel): """ Model class for llm result chunk. """ model: str # Actually used model prompt_messages: list[PromptMessage] # Prompt message list system_fingerprint: Optional[str] = None # Request fingerprint, refer to OpenAI parameter definition delta: LLMResultChunkDelta # Changes in content for each iteration
class LLMUsage(ModelUsage): """ Model class for llm usage. """ prompt_tokens: int # Tokens used by prompt prompt_unit_price: Decimal # Prompt unit price prompt_price_unit: Decimal # Prompt price unit, i.e., unit price based on how many tokens prompt_price: Decimal # Prompt cost completion_tokens: int # Tokens used by completion completion_unit_price: Decimal # Completion unit price completion_price_unit: Decimal # Completion price unit, i.e., unit price based on how many tokens completion_price: Decimal # Completion cost total_tokens: int # Total tokens used total_price: Decimal # Total cost currency: str # Currency unit latency: float # Request time (s)
class TextEmbeddingResult(BaseModel): """ Model class for text embedding result. """ model: str # Actually used model embeddings: list[list[float]] # Embedding vector list, corresponding to the input texts list 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., unit price based on how many tokens total_price: Decimal # Total cost currency: str # Currency unit latency: float # Request time (s)
class RerankResult(BaseModel): """ Model class for rerank result. """ model: str # Actually used model docs: list[RerankDocument] # List of reranked segments
class RerankDocument(BaseModel): """ Model class for rerank document. """ index: int # Original sequence number text: str # Segment text content score: float # Score
