OpenCompass/opencompass/models/pangu_api.py

from concurrent.futures import ThreadPoolExecutor
from typing import Dict, List, Optional, Union

import requests

from opencompass.utils.prompt import PromptList

from .base_api import BaseAPIModel

PromptType = Union[PromptList, str]


class PanGu(BaseAPIModel):
    """Model wrapper around PanGu.

    Args:
        path (str): The name of Pangu model.
            e.g. `pangu`
        access_key (str): provided access_key
        secret_key (str): secretkey in order to obtain access_token
        url (str): provide url for requests
        token_url (str): url of token server
        project_name (str): project name for generate the token
        query_per_second (int): The maximum queries allowed per second
            between two consecutive calls of the API. Defaults to 1.
        max_seq_len (int): Unused here.
        meta_template (Dict, optional): The model's meta prompt
            template if needed, in case the requirement of injecting or
            wrapping of any meta instructions.
        retry (int): Number of retires if the API call fails. Defaults to 2.
    """

    def __init__(
        self,
        path: str,
        access_key: str,
        secret_key: str,
        url: str,
        token_url: str,
        project_name: str,
        query_per_second: int = 2,
        max_seq_len: int = 2048,
        meta_template: Optional[Dict] = None,
        retry: int = 2,
    ):
        super().__init__(path=path,
                         max_seq_len=max_seq_len,
                         query_per_second=query_per_second,
                         meta_template=meta_template,
                         retry=retry)

        self.access_key = access_key
        self.secret_key = secret_key
        self.url = url
        self.token_url = token_url
        self.project_name = project_name
        self.model = path

    def generate(
        self,
        inputs: List[str or PromptList],
        max_out_len: int = 512,
    ) -> List[str]:
        """Generate results given a list of inputs.

        Args:
            inputs (List[str or PromptList]): A list of strings or PromptDicts.
                The PromptDict should be organized in OpenCompass'
                API format.
            max_out_len (int): The maximum length of the output.

        Returns:
            List[str]: A list of generated strings.
        """
        with ThreadPoolExecutor() as executor:
            results = list(
                executor.map(self._generate, inputs,
                             [max_out_len] * len(inputs)))
        self.flush()
        return results

    def _get_token(self):
        url = self.token_url
        payload = {
            'auth': {
                'identity': {
                    'methods': ['hw_ak_sk'],
                    'hw_ak_sk': {
                        'access': {
                            'key': self.access_key
                        },
                        'secret': {
                            'key': self.secret_key
                        }
                    }
                },
                'scope': {
                    'project': {
                        'name': self.project_name
                    }
                }
            }
        }
        headers = {'Content-Type': 'application/json'}

        response = requests.request('POST', url, headers=headers, json=payload)
        return response

    def _generate(
        self,
        input: str or PromptList,
        max_out_len: int = 512,
    ) -> str:
        """Generate results given an input.

        Args:
            inputs (str or PromptList): A string or PromptDict.
                The PromptDict should be organized in OpenCompass'
                API format.
            max_out_len (int): The maximum length of the output.

        Returns:
            str: The generated string.
        """
        assert isinstance(input, (str, PromptList))

        if isinstance(input, str):
            messages = [{'role': 'user', 'content': input}]
        else:
            messages = []
            for item in input:
                msg = {'content': item['prompt']}
                if item['role'] == 'HUMAN':
                    msg['role'] = 'user'
                elif item['role'] == 'BOT':
                    msg['role'] = 'system'

                messages.append(msg)

        data = {'messages': messages, 'stream': False}

        token_response = self._get_token()
        if token_response.status_code == 201:
            token = token_response.headers['X-Subject-Token']
            print('请求成功！')
        else:
            msg = 'token生成失败'
            print(msg)
            return ''

        headers = {'Content-Type': 'application/json', 'X-Auth-Token': token}

        max_num_retries = 0
        while max_num_retries < self.retry:
            self.acquire()
            raw_response = requests.request('POST',
                                            url=self.url,
                                            headers=headers,
                                            json=data)
            response = raw_response.json()
            self.release()

            if response is None:
                print('Connection error, reconnect.')
                # if connect error, frequent requests will casuse
                # continuous unstable network, therefore wait here
                # to slow down the request
                self.wait()
                continue
            if raw_response.status_code == 200:
                # msg = json.load(response.text)
                # response
                msg = response['choices'][0]['message']['content']
                return msg

            if (raw_response.status_code != 200):
                print(response['error_msg'])
                return ''
            print(response)
            max_num_retries += 1

        raise RuntimeError(response['error_msg'])