lagent/llms/openai.py

import asyncio
import json
import os
import time
import traceback
import warnings
from concurrent.futures import ThreadPoolExecutor
from logging import getLogger
from threading import Lock
from typing import AsyncGenerator, Dict, List, Optional, Union

import aiohttp
import requests

from ..schema import ModelStatusCode
from ..utils import filter_suffix
from .base_api import AsyncBaseAPILLM, BaseAPILLM

warnings.simplefilter('default')

OPENAI_API_BASE = 'https://api.openai.com/v1/chat/completions'


class GPTAPI(BaseAPILLM):
    """Model wrapper around OpenAI's models.

    Args:
        model_type (str): The name of OpenAI's model.
        retry (int): Number of retires if the API call fails. Defaults to 2.
        key (str or List[str]): OpenAI key(s). In particular, when it
            is set to "ENV", the key will be fetched from the environment
            variable $OPENAI_API_KEY, as how openai defaults to be. If it's a
            list, the keys will be used in round-robin manner. Defaults to
            'ENV'.
        org (str or List[str], optional): OpenAI organization(s). If not
            specified, OpenAI uses the default organization bound to each API
            key. If specified, the orgs will be posted with each request in
            round-robin manner. Defaults to None.
        meta_template (Dict, optional): The model's meta prompt
            template if needed, in case the requirement of injecting or
            wrapping of any meta instructions.
        api_base (str): The base url of OpenAI's API. Defaults to
            'https://api.openai.com/v1/chat/completions'.
        gen_params: Default generation configuration which could be overridden
            on the fly of generation.
    """

    is_api: bool = True

    def __init__(self,
                 model_type: str = 'gpt-3.5-turbo',
                 retry: int = 2,
                 json_mode: bool = False,
                 key: Union[str, List[str]] = 'ENV',
                 org: Optional[Union[str, List[str]]] = None,
                 meta_template: Optional[Dict] = [
                     dict(role='system', api_role='system'),
                     dict(role='user', api_role='user'),
                     dict(role='assistant', api_role='assistant'),
                     dict(role='environment', api_role='system')
                 ],
                 api_base: str = OPENAI_API_BASE,
                 proxies: Optional[Dict] = None,
                 **gen_params):
        if 'top_k' in gen_params:
            warnings.warn('`top_k` parameter is deprecated in OpenAI APIs.',
                          DeprecationWarning)
            gen_params.pop('top_k')
        super().__init__(
            model_type=model_type,
            meta_template=meta_template,
            retry=retry,
            **gen_params)
        self.gen_params.pop('top_k')
        self.logger = getLogger(__name__)

        if isinstance(key, str):
            self.keys = [os.getenv('OPENAI_API_KEY') if key == 'ENV' else key]
        else:
            self.keys = key

        # record invalid keys and skip them when requesting API
        # - keys have insufficient_quota
        self.invalid_keys = set()

        self.key_ctr = 0
        if isinstance(org, str):
            self.orgs = [org]
        else:
            self.orgs = org
        self.org_ctr = 0
        self.url = api_base
        self.model_type = model_type
        self.proxies = proxies
        self.json_mode = json_mode

    def chat(
        self,
        inputs: Union[List[dict], List[List[dict]]],
        **gen_params,
    ) -> Union[str, List[str]]:
        """Generate responses given the contexts.

        Args:
            inputs (Union[List[dict], List[List[dict]]]): a list of messages
                or list of lists of messages
            gen_params: additional generation configuration

        Returns:
            Union[str, List[str]]: generated string(s)
        """
        assert isinstance(inputs, list)
        if 'max_tokens' in gen_params:
            raise NotImplementedError('unsupported parameter: max_tokens')
        gen_params = {**self.gen_params, **gen_params}
        with ThreadPoolExecutor(max_workers=20) as executor:
            tasks = [
                executor.submit(self._chat,
                                self.template_parser._prompt2api(messages),
                                **gen_params)
                for messages in (
                    [inputs] if isinstance(inputs[0], dict) else inputs)
            ]
        ret = [task.result() for task in tasks]
        return ret[0] if isinstance(inputs[0], dict) else ret

    def stream_chat(
        self,
        inputs: List[dict],
        **gen_params,
    ):
        """Generate responses given the contexts.

        Args:
            inputs (List[dict]): a list of messages
            gen_params: additional generation configuration

        Returns:
            str: generated string
        """
        assert isinstance(inputs, list)
        if 'max_tokens' in gen_params:
            raise NotImplementedError('unsupported parameter: max_tokens')
        gen_params = self.update_gen_params(**gen_params)
        gen_params['stream'] = True

        resp = ''
        finished = False
        stop_words = gen_params.get('stop_words')
        if stop_words is None:
            stop_words = []
        # mapping to role that openai supports
        messages = self.template_parser._prompt2api(inputs)
        for text in self._stream_chat(messages, **gen_params):
            if self.model_type.lower().startswith('qwen'):
                resp = text
            else:
                resp += text
            if not resp:
                continue
            # remove stop_words
            for sw in stop_words:
                if sw in resp:
                    resp = filter_suffix(resp, stop_words)
                    finished = True
                    break
            yield ModelStatusCode.STREAM_ING, resp, None
            if finished:
                break
        yield ModelStatusCode.END, resp, None

    def _chat(self, messages: List[dict], **gen_params) -> str:
        """Generate completion from a list of templates.

        Args:
            messages (List[dict]): a list of prompt dictionaries
            gen_params: additional generation configuration

        Returns:
            str: The generated string.
        """
        assert isinstance(messages, list)

        header, data = self.generate_request_data(
            model_type=self.model_type,
            messages=messages,
            gen_params=gen_params,
            json_mode=self.json_mode)

        max_num_retries, errmsg = 0, ''
        while max_num_retries < self.retry:
            with Lock():
                if len(self.invalid_keys) == len(self.keys):
                    raise RuntimeError('All keys have insufficient quota.')

                # find the next valid key
                while True:
                    self.key_ctr += 1
                    if self.key_ctr == len(self.keys):
                        self.key_ctr = 0

                    if self.keys[self.key_ctr] not in self.invalid_keys:
                        break

                key = self.keys[self.key_ctr]
                header['Authorization'] = f'Bearer {key}'

            if self.orgs:
                with Lock():
                    self.org_ctr += 1
                    if self.org_ctr == len(self.orgs):
                        self.org_ctr = 0
                header['OpenAI-Organization'] = self.orgs[self.org_ctr]

            response = dict()
            try:
                raw_response = requests.post(
                    self.url,
                    headers=header,
                    data=json.dumps(data),
                    proxies=self.proxies)
                response = raw_response.json()
                return response['choices'][0]['message']['content'].strip()
            except requests.ConnectionError:
                errmsg = 'Got connection error ' + str(traceback.format_exc())
                self.logger.error(errmsg)
                continue
            except requests.JSONDecodeError:
                errmsg = 'JsonDecode error, got ' + str(raw_response.content)
                self.logger.error(errmsg)
                continue
            except KeyError:
                if 'error' in response:
                    if response['error']['code'] == 'rate_limit_exceeded':
                        time.sleep(1)
                        continue
                    elif response['error']['code'] == 'insufficient_quota':
                        self.invalid_keys.add(key)
                        self.logger.warn(f'insufficient_quota key: {key}')
                        continue

                    errmsg = 'Find error message in response: ' + str(
                        response['error'])
                    self.logger.error(errmsg)
            except Exception as error:
                errmsg = str(error) + '\n' + str(traceback.format_exc())
                self.logger.error(errmsg)
            max_num_retries += 1

        raise RuntimeError('Calling OpenAI failed after retrying for '
                           f'{max_num_retries} times. Check the logs for '
                           f'details. errmsg: {errmsg}')

    def _stream_chat(self, messages: List[dict], **gen_params) -> str:
        """Generate completion from a list of templates.

        Args:
            messages (List[dict]): a list of prompt dictionaries
            gen_params: additional generation configuration

        Returns:
            str: The generated string.
        """

        def streaming(raw_response):
            for chunk in raw_response.iter_lines(
                    chunk_size=8192, decode_unicode=False, delimiter=b'\n'):
                if chunk:
                    decoded = chunk.decode('utf-8')
                    if decoded.startswith('data: [DONE]'):
                        return
                    if decoded[:5] == 'data:':
                        decoded = decoded[5:]
                        if decoded[0] == ' ':
                            decoded = decoded[1:]
                    else:
                        print(decoded)
                        continue
                    try:
                        response = json.loads(decoded)
                        if 'code' in response and response['code'] == -20003:
                            # Context exceeds maximum length
                            yield ''
                            return
                        if self.model_type.lower().startswith('qwen'):
                            choice = response['output']['choices'][0]
                            yield choice['message']['content']
                            if choice['finish_reason'] == 'stop':
                                return
                        else:
                            choice = response['choices'][0]
                            if choice['finish_reason'] == 'stop':
                                return
                            yield choice['delta'].get('content', '')
                    except Exception as exc:
                        msg = f'response {decoded} lead to exception of {str(exc)}'
                        self.logger.error(msg)
                        raise Exception(msg) from exc

        assert isinstance(messages, list)

        header, data = self.generate_request_data(
            model_type=self.model_type,
            messages=messages,
            gen_params=gen_params,
            json_mode=self.json_mode)

        max_num_retries, errmsg = 0, ''
        while max_num_retries < self.retry:
            if len(self.invalid_keys) == len(self.keys):
                raise RuntimeError('All keys have insufficient quota.')

            # find the next valid key
            while True:
                self.key_ctr += 1
                if self.key_ctr == len(self.keys):
                    self.key_ctr = 0

                if self.keys[self.key_ctr] not in self.invalid_keys:
                    break

            key = self.keys[self.key_ctr]
            header['Authorization'] = f'Bearer {key}'

            if self.orgs:
                self.org_ctr += 1
                if self.org_ctr == len(self.orgs):
                    self.org_ctr = 0
                header['OpenAI-Organization'] = self.orgs[self.org_ctr]

            response = dict()
            try:
                raw_response = requests.post(
                    self.url,
                    headers=header,
                    data=json.dumps(data),
                    proxies=self.proxies)
                return streaming(raw_response)
            except requests.ConnectionError:
                errmsg = 'Got connection error ' + str(traceback.format_exc())
                self.logger.error(errmsg)
                continue
            except requests.JSONDecodeError:
                errmsg = 'JsonDecode error, got ' + str(raw_response.content)
                self.logger.error(errmsg)
                continue
            except KeyError:
                if 'error' in response:
                    if response['error']['code'] == 'rate_limit_exceeded':
                        time.sleep(1)
                        continue
                    elif response['error']['code'] == 'insufficient_quota':
                        self.invalid_keys.add(key)
                        self.logger.warn(f'insufficient_quota key: {key}')
                        continue

                    errmsg = 'Find error message in response: ' + str(
                        response['error'])
                    self.logger.error(errmsg)
            except Exception as error:
                errmsg = str(error) + '\n' + str(traceback.format_exc())
                self.logger.error(errmsg)
            max_num_retries += 1

        raise RuntimeError('Calling OpenAI failed after retrying for '
                           f'{max_num_retries} times. Check the logs for '
                           f'details. errmsg: {errmsg}')

    def generate_request_data(self,
                              model_type,
                              messages,
                              gen_params,
                              json_mode=False):
        """
        Generates the request data for different model types.

        Args:
            model_type (str): The type of the model (e.g., 'gpt', 'internlm', 'qwen').
            messages (list): The list of messages to be sent to the model.
            gen_params (dict): The generation parameters.
            json_mode (bool): Flag to determine if the response format should be JSON.

        Returns:
            tuple: A tuple containing the header and the request data.
        """
        # Copy generation parameters to avoid modifying the original dictionary
        gen_params = gen_params.copy()

        # Hold out 100 tokens due to potential errors in token calculation
        max_tokens = min(gen_params.pop('max_new_tokens'), 4096)
        if max_tokens <= 0:
            return '', ''

        # Initialize the header
        header = {
            'content-type': 'application/json',
        }

        # Common parameters processing
        gen_params['max_tokens'] = max_tokens
        if 'stop_words' in gen_params:
            gen_params['stop'] = gen_params.pop('stop_words')
        if 'repetition_penalty' in gen_params:
            gen_params['frequency_penalty'] = gen_params.pop(
                'repetition_penalty')

        # Model-specific processing
        data = {}
        if model_type.lower().startswith('gpt'):
            if 'top_k' in gen_params:
                warnings.warn(
                    '`top_k` parameter is deprecated in OpenAI APIs.',
                    DeprecationWarning)
                gen_params.pop('top_k')
            gen_params.pop('skip_special_tokens', None)
            gen_params.pop('session_id', None)
            data = {
                'model': model_type,
                'messages': messages,
                'n': 1,
                **gen_params
            }
            if json_mode:
                data['response_format'] = {'type': 'json_object'}
        elif model_type.lower().startswith('internlm'):
            data = {
                'model': model_type,
                'messages': messages,
                'n': 1,
                **gen_params
            }
            if json_mode:
                data['response_format'] = {'type': 'json_object'}
        elif model_type.lower().startswith('qwen'):
            header['X-DashScope-SSE'] = 'enable'
            gen_params.pop('skip_special_tokens', None)
            gen_params.pop('session_id', None)
            if 'frequency_penalty' in gen_params:
                gen_params['repetition_penalty'] = gen_params.pop(
                    'frequency_penalty')
            gen_params['result_format'] = 'message'
            data = {
                'model': model_type,
                'input': {
                    'messages': messages
                },
                'parameters': {
                    **gen_params
                }
            }
        else:
            raise NotImplementedError(
                f'Model type {model_type} is not supported')

        return header, data

    def tokenize(self, prompt: str) -> list:
        """Tokenize the input prompt.

        Args:
            prompt (str): Input string.

        Returns:
            list: token ids
        """
        import tiktoken
        self.tiktoken = tiktoken
        enc = self.tiktoken.encoding_for_model(self.model_type)
        return enc.encode(prompt)


class AsyncGPTAPI(AsyncBaseAPILLM):
    """Model wrapper around OpenAI's models.

    Args:
        model_type (str): The name of OpenAI's model.
        retry (int): Number of retires if the API call fails. Defaults to 2.
        key (str or List[str]): OpenAI key(s). In particular, when it
            is set to "ENV", the key will be fetched from the environment
            variable $OPENAI_API_KEY, as how openai defaults to be. If it's a
            list, the keys will be used in round-robin manner. Defaults to
            'ENV'.
        org (str or List[str], optional): OpenAI organization(s). If not
            specified, OpenAI uses the default organization bound to each API
            key. If specified, the orgs will be posted with each request in
            round-robin manner. Defaults to None.
        meta_template (Dict, optional): The model's meta prompt
            template if needed, in case the requirement of injecting or
            wrapping of any meta instructions.
        api_base (str): The base url of OpenAI's API. Defaults to
            'https://api.openai.com/v1/chat/completions'.
        gen_params: Default generation configuration which could be overridden
            on the fly of generation.
    """

    is_api: bool = True

    def __init__(self,
                 model_type: str = 'gpt-3.5-turbo',
                 retry: int = 2,
                 json_mode: bool = False,
                 key: Union[str, List[str]] = 'ENV',
                 org: Optional[Union[str, List[str]]] = None,
                 meta_template: Optional[Dict] = [
                     dict(role='system', api_role='system'),
                     dict(role='user', api_role='user'),
                     dict(role='assistant', api_role='assistant')
                 ],
                 api_base: str = OPENAI_API_BASE,
                 proxies: Optional[Dict] = None,
                 **gen_params):
        if 'top_k' in gen_params:
            warnings.warn('`top_k` parameter is deprecated in OpenAI APIs.',
                          DeprecationWarning)
            gen_params.pop('top_k')
        super().__init__(
            model_type=model_type,
            meta_template=meta_template,
            retry=retry,
            **gen_params)
        self.gen_params.pop('top_k')
        self.logger = getLogger(__name__)

        if isinstance(key, str):
            self.keys = [os.getenv('OPENAI_API_KEY') if key == 'ENV' else key]
        else:
            self.keys = key

        # record invalid keys and skip them when requesting API
        # - keys have insufficient_quota
        self.invalid_keys = set()

        self.key_ctr = 0
        if isinstance(org, str):
            self.orgs = [org]
        else:
            self.orgs = org
        self.org_ctr = 0
        self.url = api_base
        self.model_type = model_type
        self.proxies = proxies or {}
        self.json_mode = json_mode

    async def chat(
        self,
        inputs: Union[List[dict], List[List[dict]]],
        session_ids: Union[int, List[int]] = None,
        **gen_params,
    ) -> Union[str, List[str]]:
        """Generate responses given the contexts.

        Args:
            inputs (Union[List[dict], List[List[dict]]]): a list of messages
                or list of lists of messages
            gen_params: additional generation configuration

        Returns:
            Union[str, List[str]]: generated string(s)
        """
        assert isinstance(inputs, list)
        if 'max_tokens' in gen_params:
            raise NotImplementedError('unsupported parameter: max_tokens')
        gen_params = {**self.gen_params, **gen_params}
        tasks = [
            self._chat(messages, **gen_params) for messages in (
                [inputs] if isinstance(inputs[0], dict) else inputs)
        ]
        ret = await asyncio.gather(*tasks)
        return ret[0] if isinstance(inputs[0], dict) else ret

    async def stream_chat(
        self,
        inputs: List[dict],
        **gen_params,
    ):
        """Generate responses given the contexts.

        Args:
            inputs (List[dict]): a list of messages
            gen_params: additional generation configuration

        Returns:
            str: generated string
        """
        assert isinstance(inputs, list)
        if 'max_tokens' in gen_params:
            raise NotImplementedError('unsupported parameter: max_tokens')
        gen_params = self.update_gen_params(**gen_params)
        gen_params['stream'] = True

        resp = ''
        finished = False
        stop_words = gen_params.get('stop_words')
        if stop_words is None:
            stop_words = []
        # mapping to role that openai supports
        messages = self.template_parser._prompt2api(inputs)
        async for text in self._stream_chat(messages, **gen_params):
            if self.model_type.lower().startswith('qwen'):
                resp = text
            else:
                resp += text
            if not resp:
                continue
            # remove stop_words
            for sw in stop_words:
                if sw in resp:
                    resp = filter_suffix(resp, stop_words)
                    finished = True
                    break
            yield ModelStatusCode.STREAM_ING, resp, None
            if finished:
                break
        yield ModelStatusCode.END, resp, None

    async def _chat(self, messages: List[dict], **gen_params) -> str:
        """Generate completion from a list of templates.

        Args:
            messages (List[dict]): a list of prompt dictionaries
            gen_params: additional generation configuration

        Returns:
            str: The generated string.
        """
        assert isinstance(messages, list)

        header, data = self.generate_request_data(
            model_type=self.model_type,
            messages=messages,
            gen_params=gen_params,
            json_mode=self.json_mode)

        max_num_retries, errmsg = 0, ''
        while max_num_retries < self.retry:
            if len(self.invalid_keys) == len(self.keys):
                raise RuntimeError('All keys have insufficient quota.')

            # find the next valid key
            while True:
                self.key_ctr += 1
                if self.key_ctr == len(self.keys):
                    self.key_ctr = 0

                if self.keys[self.key_ctr] not in self.invalid_keys:
                    break

            key = self.keys[self.key_ctr]
            header['Authorization'] = f'Bearer {key}'

            if self.orgs:
                self.org_ctr += 1
                if self.org_ctr == len(self.orgs):
                    self.org_ctr = 0
                header['OpenAI-Organization'] = self.orgs[self.org_ctr]

            response = dict()
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(
                            self.url,
                            headers=header,
                            json=data,
                            proxy=self.proxies.get(
                                'https', self.proxies.get('http'))) as resp:
                        response = await resp.json()
                        return response['choices'][0]['message'][
                            'content'].strip()
            except aiohttp.ClientConnectionError:
                errmsg = 'Got connection error ' + str(traceback.format_exc())
                self.logger.error(errmsg)
                continue
            except aiohttp.ClientResponseError as e:
                errmsg = 'Response error, got ' + str(e)
                self.logger.error(errmsg)
                continue
            except json.JSONDecodeError:
                errmsg = 'JsonDecode error, got ' + (await resp.text(
                    errors='replace'))
                self.logger.error(errmsg)
                continue
            except KeyError:
                if 'error' in response:
                    if response['error']['code'] == 'rate_limit_exceeded':
                        time.sleep(1)
                        continue
                    elif response['error']['code'] == 'insufficient_quota':
                        self.invalid_keys.add(key)
                        self.logger.warn(f'insufficient_quota key: {key}')
                        continue

                    errmsg = 'Find error message in response: ' + str(
                        response['error'])
                    self.logger.error(errmsg)
            except Exception as error:
                errmsg = str(error) + '\n' + str(traceback.format_exc())
                self.logger.error(errmsg)
            max_num_retries += 1

        raise RuntimeError('Calling OpenAI failed after retrying for '
                           f'{max_num_retries} times. Check the logs for '
                           f'details. errmsg: {errmsg}')

    async def _stream_chat(self, messages: List[dict],
                           **gen_params) -> AsyncGenerator[str, None]:
        """Generate completion from a list of templates.

        Args:
            messages (List[dict]): a list of prompt dictionaries
            gen_params: additional generation configuration

        Returns:
            str: The generated string.
        """

        async def streaming(raw_response):
            async for chunk in raw_response.content:
                if chunk:
                    decoded = chunk.decode('utf-8')
                    if decoded.startswith('data: [DONE]'):
                        return
                    if decoded[:5] == 'data:':
                        decoded = decoded[5:]
                        if decoded[0] == ' ':
                            decoded = decoded[1:]
                    else:
                        print(decoded)
                        continue
                    try:
                        response = json.loads(decoded)
                        if 'code' in response and response['code'] == -20003:
                            # Context exceeds maximum length
                            yield ''
                            return
                        if self.model_type.lower().startswith('qwen'):
                            choice = response['output']['choices'][0]
                            yield choice['message']['content']
                            if choice['finish_reason'] == 'stop':
                                return
                        else:
                            choice = response['choices'][0]
                            if choice['finish_reason'] == 'stop':
                                return
                            yield choice['delta'].get('content', '')
                    except Exception as exc:
                        msg = f'response {decoded} lead to exception of {str(exc)}'
                        self.logger.error(msg)
                        raise Exception(msg) from exc

        assert isinstance(messages, list)

        header, data = self.generate_request_data(
            model_type=self.model_type,
            messages=messages,
            gen_params=gen_params,
            json_mode=self.json_mode)

        max_num_retries, errmsg = 0, ''
        while max_num_retries < self.retry:
            if len(self.invalid_keys) == len(self.keys):
                raise RuntimeError('All keys have insufficient quota.')

            # find the next valid key
            while True:
                self.key_ctr += 1
                if self.key_ctr == len(self.keys):
                    self.key_ctr = 0

                if self.keys[self.key_ctr] not in self.invalid_keys:
                    break

            key = self.keys[self.key_ctr]
            header['Authorization'] = f'Bearer {key}'

            if self.orgs:
                self.org_ctr += 1
                if self.org_ctr == len(self.orgs):
                    self.org_ctr = 0
                header['OpenAI-Organization'] = self.orgs[self.org_ctr]

            response = dict()
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(
                            self.url,
                            headers=header,
                            json=data,
                            proxy=self.proxies.get(
                                'https',
                                self.proxies.get('http'))) as raw_response:
                        async for msg in streaming(raw_response):
                            yield msg
                        return
            except aiohttp.ClientConnectionError:
                errmsg = 'Got connection error ' + str(traceback.format_exc())
                self.logger.error(errmsg)
                continue
            except aiohttp.ClientResponseError as e:
                errmsg = 'Response error, got ' + str(e)
                self.logger.error(errmsg)
                continue
            except KeyError:
                if 'error' in response:
                    if response['error']['code'] == 'rate_limit_exceeded':
                        time.sleep(1)
                        continue
                    elif response['error']['code'] == 'insufficient_quota':
                        self.invalid_keys.add(key)
                        self.logger.warn(f'insufficient_quota key: {key}')
                        continue

                    errmsg = 'Find error message in response: ' + str(
                        response['error'])
                    self.logger.error(errmsg)
            except Exception as error:
                errmsg = str(error) + '\n' + str(traceback.format_exc())
                self.logger.error(errmsg)
            max_num_retries += 1

        raise RuntimeError('Calling OpenAI failed after retrying for '
                           f'{max_num_retries} times. Check the logs for '
                           f'details. errmsg: {errmsg}')

    def generate_request_data(self,
                              model_type,
                              messages,
                              gen_params,
                              json_mode=False):
        """
        Generates the request data for different model types.

        Args:
            model_type (str): The type of the model (e.g., 'gpt', 'internlm', 'qwen').
            messages (list): The list of messages to be sent to the model.
            gen_params (dict): The generation parameters.
            json_mode (bool): Flag to determine if the response format should be JSON.

        Returns:
            tuple: A tuple containing the header and the request data.
        """
        # Copy generation parameters to avoid modifying the original dictionary
        gen_params = gen_params.copy()

        # Hold out 100 tokens due to potential errors in token calculation
        max_tokens = min(gen_params.pop('max_new_tokens'), 4096)
        if max_tokens <= 0:
            return '', ''

        # Initialize the header
        header = {
            'content-type': 'application/json',
        }

        # Common parameters processing
        gen_params['max_tokens'] = max_tokens
        if 'stop_words' in gen_params:
            gen_params['stop'] = gen_params.pop('stop_words')
        if 'repetition_penalty' in gen_params:
            gen_params['frequency_penalty'] = gen_params.pop(
                'repetition_penalty')

        # Model-specific processing
        data = {}
        if model_type.lower().startswith('gpt'):
            if 'top_k' in gen_params:
                warnings.warn(
                    '`top_k` parameter is deprecated in OpenAI APIs.',
                    DeprecationWarning)
                gen_params.pop('top_k')
            gen_params.pop('skip_special_tokens', None)
            gen_params.pop('session_id', None)
            data = {
                'model': model_type,
                'messages': messages,
                'n': 1,
                **gen_params
            }
            if json_mode:
                data['response_format'] = {'type': 'json_object'}
        elif model_type.lower().startswith('internlm'):
            data = {
                'model': model_type,
                'messages': messages,
                'n': 1,
                **gen_params
            }
            if json_mode:
                data['response_format'] = {'type': 'json_object'}
        elif model_type.lower().startswith('qwen'):
            header['X-DashScope-SSE'] = 'enable'
            gen_params.pop('skip_special_tokens', None)
            gen_params.pop('session_id', None)
            if 'frequency_penalty' in gen_params:
                gen_params['repetition_penalty'] = gen_params.pop(
                    'frequency_penalty')
            gen_params['result_format'] = 'message'
            data = {
                'model': model_type,
                'input': {
                    'messages': messages
                },
                'parameters': {
                    **gen_params
                }
            }
        else:
            raise NotImplementedError(
                f'Model type {model_type} is not supported')

        return header, data

    def tokenize(self, prompt: str) -> list:
        """Tokenize the input prompt.

        Args:
            prompt (str): Input string.

        Returns:
            list: token ids
        """
        import tiktoken
        self.tiktoken = tiktoken
        enc = self.tiktoken.encoding_for_model(self.model_type)
        return enc.encode(prompt)