from abc import abstractmethod from typing import Any, Dict, List, Tuple from evalscope.perf.arguments import Arguments class ApiPluginBase: def __init__(self, model_path: str) -> None: self.model_path = model_path @abstractmethod def build_request(self, messages: List[Dict], param: Arguments) -> Dict: """Build a api request body. Args: messages (List[Dict]): The messages generated by dataset. param (QueryParameters): The query parameters. Raises: NotImplementedError: Not implemented. Returns: Dict: The api request body. """ raise NotImplementedError @abstractmethod def parse_responses(self, responses: List, request: Any = None, **kwargs: Any) -> Tuple[int, int]: """Parser responses and return number of request and response tokens. Args: responses (List[bytes]): List of http response body, for stream output, there are multiple responses, each is bytes, for general only one. request (Any): The request body. Returns: Tuple: (Number of prompt_tokens and number of completion_tokens). """ raise NotImplementedError @staticmethod def replace_values(input_json: Any, model: str, prompt: str): if isinstance(input_json, dict): for key, value in input_json.items(): if isinstance(value, str): input_json[key] = value.replace('%m', model).replace('%p', prompt) else: ApiPluginBase.replace_values(value, model, prompt) elif isinstance(input_json, list): for idx, item in enumerate(input_json): if isinstance(item, str): input_json[idx] = item.replace('%m', model).replace('%p', prompt) else: ApiPluginBase.replace_values(item, model, prompt) elif isinstance(input_json, str): input_json = input_json.replace('%m', model).replace('%p', prompt) else: pass