支持modelscope直接拉起推理服务

Link: https://code.alibaba-inc.com/Ali-MaaS/MaaS-lib/codereview/14702876

* add inference server code

* add server requirement

* fix import issue

* debug

* add command line

* add llmpipeline support

* modify port to int

* add serer usage

* remove unused code

* fix lint issue

* add inference server code

* upgrade env to VLLM_USE_MODELSCOPE

docs/source/server.md

@@ -0,0 +1,41 @@
+# modelscope server使用
+## 1. 通用服务
+modelscope库基于fastapi开发一个简单模型服务，可以通过一条命令拉起绝大多数模型
+使用方法：
+
+```bash
+modelscope server --model_id=modelscope/Llama-2-7b-chat-ms --revision=v1.0.5
+```
+我们提供的官方镜像中也可以一个命令启动(镜像还未完成)
+```bash
+docker run --rm --name maas_dev --shm-size=50gb --gpus='"device=0"' -e MODELSCOPE_CACHE=/modelscope_cache -v /host_path_to_modelscope_cache:/modelscope_cache -p 8000:8000 reg.docker.alibaba-inc.com/modelscope/modelscope:ubuntu22.04-cuda11.8.0-py310-torch2.1.0-tf2.14.0-1.9.5-server modelscope server --model_id=modelscope/Llama-2-7b-chat-ms --revision=v1.0.5
+```
+服务默认监听8000端口，您也可以通过--port改变端口，默认服务提供两个接口，接口文档您可以通过
+http://ip:port/docs查看
+通过describe接口，可以获取服务输入输出信息以及输入sample数据，如下图：
+![describe](https://modelscope.oss-cn-beijing.aliyuncs.com/resource/describe.jpg)
+服务调用接口，可以直接拷贝describe接口example示例数据，如下图：
+![call](https://modelscope.oss-cn-beijing.aliyuncs.com/resource/call.jpg)
+
+## 2. vllm大模型推理
+对于LLM我们提供了vllm推理支持，目前只有部分模型支持vllm。
+
+### 2.1 vllm直接支持modelscope模型
+可以通过设置环境变量使得vllm从www.modelscope.cn下载模型。
+
+启动普通server
+```bash
+VLLM_USE_MODELSCOPE=True python -m vllm.entrypoints.api_server  --model="damo/nlp_gpt2_text-generation_english-base" --revision="v1.0.0"
+```
+启动openai兼容接口
+```bash
+VLLM_USE_MODELSCOPE=True python -m vllm.entrypoints.openai.api_server  --model="damo/nlp_gpt2_text-generation_english-base" --revision="v1.0.0"
+```
+
+如果模型在modelscope cache目录已经存在，则会直接使用cache中的模型，否则会从www.modelscope.cn下载模型。
+
+通过modelscope官方镜像启动vllm，指定端口为9090
+
+```bash
+docker run --rm --name maas_dev --shm-size=50gb --gpus='"device=0"' -e MODELSCOPE_CACHE=/modelscope_cache -v /host_path_to_modelscope_cache:/modelscope_cache -p 9090:9090 reg.docker.alibaba-inc.com/modelscope/modelscope:ubuntu22.04-cuda11.8.0-py310-torch2.1.0-tf2.14.0-1.9.5-server python -m vllm.entrypoints.api_server --model "modelscope/Llama-2-7b-chat-ms" --revision "v1.0.5" --port 9090
+```

modelscope/cli/cli.py

@@ -6,6 +6,7 @@ from modelscope.cli.download import DownloadCMD
 from modelscope.cli.modelcard import ModelCardCMD
 from modelscope.cli.pipeline import PipelineCMD
 from modelscope.cli.plugins import PluginsCMD
+from modelscope.cli.server import ServerCMD
 
 
 def run_cmd():
@@ -17,6 +18,7 @@ def run_cmd():
     PluginsCMD.define_args(subparsers)
     PipelineCMD.define_args(subparsers)
     ModelCardCMD.define_args(subparsers)
+    ServerCMD.define_args(subparsers)
 
     args = parser.parse_args()
 

modelscope/cli/server.py

@@ -0,0 +1,40 @@
+# Copyright (c) Alibaba, Inc. and its affiliates.
+import os
+from argparse import ArgumentParser
+from string import Template
+
+import uvicorn
+
+from modelscope.cli.base import CLICommand
+from modelscope.server.api_server import add_server_args, get_app
+from modelscope.utils.logger import get_logger
+
+logger = get_logger()
+
+current_path = os.path.dirname(os.path.abspath(__file__))
+template_path = os.path.join(current_path, 'template')
+
+
+def subparser_func(args):
+    """ Function which will be called for a specific sub parser.
+    """
+    return ServerCMD(args)
+
+
+class ServerCMD(CLICommand):
+    name = 'server'
+
+    def __init__(self, args):
+        self.args = args
+
+    @staticmethod
+    def define_args(parsers: ArgumentParser):
+        """ define args for create pipeline template command.
+        """
+        parser = parsers.add_parser(ServerCMD.name)
+        add_server_args(parser)
+        parser.set_defaults(func=subparser_func)
+
+    def execute(self):
+        app = get_app(self.args)
+        uvicorn.run(app, host=self.args.host, port=self.args.port)

modelscope/server/api/routers/health.py

@@ -0,0 +1,14 @@
+from faulthandler import disable
+from http import HTTPStatus
+from typing import Any, Dict
+
+from fastapi import APIRouter
+
+from modelscope.server.models.output import ApiResponse
+
+router = APIRouter()
+
+
+@router.get('', response_model=ApiResponse[Dict], status_code=200)
+def health() -> Any:
+    return ApiResponse[Dict](Data={}, Code=HTTPStatus.OK, Success=True)

modelscope/server/api/routers/model_router.py

@@ -0,0 +1,45 @@
+from fastapi import APIRouter, Body
+from pydantic import BaseModel
+from starlette.requests import Request
+
+from modelscope.utils.input_output import \
+    pipeline_output_to_service_base64_output  # noqa E125
+from modelscope.utils.input_output import call_pipeline_with_json
+
+router = APIRouter()
+
+
+@router.post('/call')
+async def inference(
+    request: Request,
+    body: BaseModel = Body(examples=[{
+        'usage': 'copy body from describe'
+    }])):  # noqa E125
+    """Inference general interface.
+
+    For image, video, audio etc binary data, need encoded with base64.
+
+    Args:
+        request (Request): The request object.
+        request_info (ModelScopeRequest): The post body.
+
+    Returns:
+        ApiResponse: For binary field, encoded with base64
+    """
+    pipeline_service = request.app.state.pipeline
+    pipeline_info = request.app.state.pipeline_info
+    request_json = await request.json()
+    result = call_pipeline_with_json(pipeline_info, pipeline_service,
+                                     request_json)
+    # convert output to json, if binary field, we need encoded.
+    output = pipeline_output_to_service_base64_output(
+        pipeline_info['task_name'], result)
+    return output
+
+
+@router.get('/describe')
+async def describe(request: Request):
+    info = {}
+    info['schema'] = request.app.state.pipeline_info
+    info['sample'] = request.app.state.pipeline_sample
+    return info

modelscope/server/api/routers/router.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+from starlette.routing import Route, WebSocketRoute
+
+from modelscope.server.api.routers import health, model_router
+
+api_router = APIRouter()
+api_router.include_router(model_router.router, tags=['prediction'], prefix='')
+api_router.include_router(health.router, tags=['health'], prefix='/health')

modelscope/server/api_server.py

@@ -0,0 +1,45 @@
+import argparse
+
+import uvicorn
+from fastapi import FastAPI
+
+from modelscope.server.api.routers.router import api_router
+from modelscope.server.core.event_handlers import (start_app_handler,
+                                                   stop_app_handler)
+
+
+def get_app(args) -> FastAPI:
+    app = FastAPI(
+        title='modelscope_server',
+        version='0.1',
+        debug=True,
+        swagger_ui_parameters={'tryItOutEnabled': True})
+    app.state.args = args
+    app.include_router(api_router)
+
+    app.add_event_handler('startup', start_app_handler(app))
+    app.add_event_handler('shutdown', stop_app_handler(app))
+    return app
+
+
+def add_server_args(parser):
+    parser.add_argument(
+        '--model_id', required=True, type=str, help='The target model id')
+    parser.add_argument(
+        '--revision', required=True, type=str, help='Model revision')
+    parser.add_argument('--host', default='0.0.0.0', help='Host to listen')
+    parser.add_argument('--port', type=int, default=8000, help='Server port')
+    parser.add_argument('--debug', default='debug', help='Set debug level.')
+    parser.add_argument(
+        '--llm_first',
+        type=bool,
+        default=True,
+        help='Use LLMPipeline first for llm models.')
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser('modelscope_server')
+    add_server_args(parser)
+    args = parser.parse_args()
+    app = get_app(args)
+    uvicorn.run(app, host=args.host, port=args.port)

modelscope/server/core/event_handlers.py

@@ -0,0 +1,47 @@
+from typing import Callable
+
+from fastapi import FastAPI
+
+from modelscope.utils.input_output import (  # yapf: disable
+    create_pipeline, get_pipeline_information_by_pipeline,
+    get_task_input_examples, get_task_schemas)
+from modelscope.utils.logger import get_logger
+
+# control the model start stop
+
+logger = get_logger()
+
+
+def _startup_model(app: FastAPI) -> None:
+    logger.info('download model and create pipeline')
+    app.state.pipeline = create_pipeline(app.state.args.model_id,
+                                         app.state.args.revision,
+                                         app.state.args.llm_first)
+    info = {}
+    info['task_name'] = app.state.pipeline.group_key
+    info['schema'] = get_task_schemas(app.state.pipeline.group_key)
+    app.state.pipeline_info = info
+    app.state.pipeline_sample = get_task_input_examples(
+        app.state.pipeline.group_key)
+    logger.info('pipeline created.')
+
+
+def _shutdown_model(app: FastAPI) -> None:
+    app.state.pipeline = None
+    logger.info('shutdown model service')
+
+
+def start_app_handler(app: FastAPI) -> Callable:
+
+    def startup() -> None:
+        _startup_model(app)
+
+    return startup
+
+
+def stop_app_handler(app: FastAPI) -> Callable:
+
+    def shutdown() -> None:
+        _shutdown_model(app)
+
+    return shutdown

modelscope/server/models/input.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class ModelScopeRequest(BaseModel):
+
+    def __init__(self, input: object, parameters: object):
+        self.input = input
+        self.parameters = parameters

modelscope/server/models/output.py

@@ -0,0 +1,34 @@
+import datetime
+from http import HTTPStatus
+from typing import Generic, Optional, Type, TypeVar
+
+import json
+from pydantic.generics import GenericModel
+
+ResultType = TypeVar('ResultType')
+
+
+class ApiResponse(GenericModel, Generic[ResultType]):
+    Code: Optional[int] = HTTPStatus.OK
+    Success: Optional[bool] = True
+    RequestId: Optional[str] = ''
+    Message: Optional[str] = 'success'
+    Data: Optional[ResultType] = {}
+    """
+        ResultType (_type_): The response data type.
+        Failed: {'Code': 10010101004, 'Message': 'get model info failed, err: unauthorized permission',
+                 'RequestId': '', 'Success': False}
+        Success: {'Code': 200, 'Data': {}, 'Message': 'success', 'RequestId': '', 'Success': True}
+
+
+
+    def set_data(self, data=Type[ResultType]):
+        self.Data = data
+
+    def set_message(self, message):
+        self.Message = message
+
+    def toJSON(self):
+        return json.dumps(self, default=lambda o: o.isoformat() if (isinstance(o, datetime.datetime))
+                          else o.__dict__, sort_keys=True, indent=4)
+    """

modelscope/utils/input_output.py

@@ -36,16 +36,18 @@ decodes relevant fields.
 Example:
     # create pipeine instance and pipeline information, save it to app
     pipeline_instance = create_pipeline('damo/cv_gpen_image-portrait-enhancement', 'v1.0.0')
+    # get pipeline information, input,output, request example.
     pipeline_info = get_pipeline_information_by_pipeline(pipeline_instance)
+    # save the pipeline and info to the app for use in subsequent request processing
     app.state.pipeline = pipeline_instance
     app.state.pipeline_info = pipeline_info
 
-    # for service schema request.
-    pipeline_info = request.app.state.pipeline_info
-    return pipeline_info.schema
-
-    # for service call request.
-    def inference(request: Request):
+    # for inference request, use call_pipeline_with_json to decode input and
+    # call pipeline, call pipeline_output_to_service_base64_output
+    # to encode necessary fields, and return the result.
+    # request and response are json format.
+    @router.post('/call')
+    async def inference(request: Request):
         pipeline_service = request.app.state.pipeline
         pipeline_info = request.app.state.pipeline_info
         request_json = await request.json()
@@ -55,19 +57,30 @@ Example:
         # convert output to json, if binary field, we need encoded.
         output = pipeline_output_to_service_base64_output(pipeline_info.task_name, result)
         return output
+
+    # Inference service input and output and sample information can be obtained through the docs interface
+    @router.get('/describe')
+    async def index(request: Request):
+        pipeline_info = request.app.state.pipeline_info
+        return pipeline_info.schema
+
 Todo:
     * Support more service input type, such as form.
 
 """
 
 
-def create_pipeline(model_id: str, revision: str):
+def create_pipeline(model_id: str, revision: str, llm_first: bool = True):
     model_configuration_file = model_file_download(
         model_id=model_id,
         file_path=ModelFile.CONFIGURATION,
         revision=revision)
     cfg = Config.from_file(model_configuration_file)
-    return pipeline(task=cfg.task, model=model_id, model_revision=revision)
+    return pipeline(
+        task=cfg.task,
+        model=model_id,
+        model_revision=revision,
+        llm_first=llm_first)
 
 
 def get_class_user_attributes(cls):
@@ -632,7 +645,7 @@ def call_pipeline_with_json(pipeline_info: PipelineInfomation,
     #     result = pipeline(**pipeline_inputs)
     # else:
     pipeline_inputs, parameters = service_base64_input_to_pipeline_input(
-        pipeline_info.task_name, body)
+        pipeline_info['task_name'], body)
     result = pipeline(pipeline_inputs, **parameters)
 
     return result

requirements/svr.txt

@@ -0,0 +1,4 @@
+fastapi
+requests
+sse-starlette
+uvicorn