LLM 性能测试实践

from evalscope.perf.main import run_perf_benchmark
import os

task_cfg = {"url": "https://api.deepseek.com/chat/completions",
   "api_key": os.getenv("DEEPSEEK_API_KEY"),
            "parallel": 5,
            "model": "deepseek-chat",
            "number": 20,
            "api": "openai",
            "dataset": "openqa",
            "stream": True}
run_perf_benchmark(task_cfg)

(evalscope) (base) [windealli@VM-52-29-tencentos llm-benchmark]$ python evalscope/deepseek_perf.py

import os
from evalscope.perf.main import run_perf_benchmark
import sbsagent.custom

query_template = """{
    "query_id": "{query_id}",
    "agent_session_id": "{agent_session_id}",
    "query": "%p" ,
    "forward_service": "%m", 
    "stream": true
}
"""
task_cfg = {
    "url": "https://127.0.0.1:8080/v1/chat/complete",
    "api_key": os.getenv("SSV_AI_GW_API_KEY"),
 ...
 "api": "sbs-agent-api",
    "query_template": query_template,
   "no_test_connection": True,
}

run_perf_benchmark(task_cfg)

# import os
import json
import time
from abc import abstractmethod
from typing import Any, Dict, List, Tuple
from evalscope.perf.arguments import Arguments
from evalscope.perf.plugin.api.base import ApiPluginBase
from evalscope.perf.plugin.registry import register_api
from evalscope.utils.logger import get_logger

logger = get_logger()

# from evalscope.perf.plugin.registry import register_api
@register_api('sbs-agent-api')
class CustomPlugin(ApiPluginBase):
    def __init__(self, model_path: str) -> None:
        print("CustomPlugin initialized.=============")
        self.model_path = model_path
        self.invalid_output_count = 0
        self.request_count = 0
        if self.model_path is not None:
            from modelscope import AutoTokenizer
            self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        else:
            self.tokenizer = None
        
    @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.
        """
        # print(messages, param)
        print("======message=====")
        print(messages)
        # print("======param=====")
        # print(param)

        try:
            query = json.loads(param.query_template)
            if not isinstance(query, dict):
                raise ValueError('Prompt must be a dict.')
            ApiPluginBase.replace_values(query, param.model, messages)
            #将query中的{query_id}替换成当前时间戳
            # query['query_id'] = time.strftime("%Y%m%d%H%M%S", time.localtime())
            query['query_id'] = "query_id-" + str(self.request_count % 100)
            print("======", query)
            return query
        except Exception as e:
            logger.exception(e)
            logger.error('Prompt: %s invalidate!' % messages)
            return None
        # 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).
        """
        full_response_content = ''
        delta_contents = {}
        input_tokens = None
        output_tokens = None

        # Parse responses to extract and concatenate data content
        # parsed_responses_result = []
        full_response_result = '' 
        for response in responses:
            try:
            # Decode the response and split by lines
                lines = response.decode('utf-8').splitlines()
                for line in lines:
                    if line.startswith("data:"):
                    # Extract the JSON part after "data:" and parse it
                        data = json.loads(line[5:].strip())
                    if "retcode" not in data or data["retcode"] != 0:
                        raise Exception(f"Error code {data['retcode']}: {data['message']}")

                    if "result" in data:
                        full_response_result = full_response_result + data["result"]
                    # parsed_responses.append(data)
                    if "usage" in data:
                        if "prompt_tokens" in data["usage"]:
                            input_tokens = input_tokens + data["usage"]["prompt_tokens"]
                        if "completion_tokens" in data["usage"]:
                            output_tokens = output_tokens + data["usage"]["completion_tokens"]
            except Exception as e:
                logger.exception(f"Failed to parse response: {response}")
        if input_tokens is None or output_tokens is None:
            logger.warning('No usage info get.')
            if self.tokenizer is not None:
                input_tokens = len(self.tokenizer.encode(request["query"]))
                output_tokens = len(self.tokenizer.encode(full_response_result))
            else:
                input_tokens = len(request["query"])
                output_tokens = len(full_response_result)


        return input_tokens, output_tokens
        
        print("Parsed Responses:", parsed_responses)
        # return parsed_responses
        # 解析响应

# Automatically initialize the module when imported
def _initialize_module():
    print("Initializing CustomPlugin module...")
    # Perform any necessary setup or registration here

_initialize_module()

import sqlite3
import base64
import pickle
import json
import xlsxwriter

result_db_path = 'outputs/20250425_142407/xxxx/benchmark_data.db'
result_xlws_path = 'outputs/20250425_142407/xxxx/benchmark_results.xlsx'
con = sqlite3.connect(result_db_path)
query_sql = "SELECT request, response_messages, prompt_tokens, completion_tokens \
                FROM result WHERE success='1'"
# how to save base64.b64encode(pickle.dumps(benchmark_data["request"])).decode("ascii"), 
with con:

    # Initialize Excel workbook and worksheet
    workbook = xlsxwriter.Workbook(result_xlws_path)
    worksheet = workbook.add_worksheet()

    # Write headers
    headers = ['Prompt', 'Prompt Tokens', 'Completion', 'Completion Tokens']
    for col_num, header in enumerate(headers):
        worksheet.write(0, col_num, header)
    row_num = 1

    rows = con.execute(query_sql).fetchall()
    if len(rows) > 0:
        for row in rows:
            request = row[0]
            responses = row[1]
            request = base64.b64decode(request)
            request = pickle.loads(request)
            responses = base64.b64decode(responses)
            responses = pickle.loads(responses)
            print('request: %s' % request)
            print('responses: %s' % responses)
            response_content = ''
            for response in responses:
                print(response)
                response = json.loads(response)
                # if not response['result']:
                #    print(response)
                #    continue
                if 'content' not in response['choices'][0]['delta']:
                    print(response)
                    continue
                # response_content += response['result']
                response_content += response['choices'][0]['delta']['content']
            print('prompt: %s, tokens: %s, completion: %s, tokens: %s' %
                  (request['query'], row[2], response_content,
                   row[3]))
            worksheet.write(row_num, 0, request['query'])
            worksheet.write(row_num, 1, row[2])  # Prompt tokens
            worksheet.write(row_num, 2, response_content)
            worksheet.write(row_num, 3, row[3])  # Completion tokens
            row_num += 1
        
        # Close the workbook
        workbook.close()