| from typing import Dict, List, Any
|
| from lmdeploy import pipeline
|
| from lmdeploy.vl import load_image
|
| from lmdeploy.messages import TurbomindEngineConfig
|
|
|
| class EndpointHandler():
|
| def __init__(self, path):
|
|
|
| backend_config = TurbomindEngineConfig(model_name ="deepseek-ai/deepseek-coder-33b-instruct",model_format='hf',tp=1)
|
| self.pipe = pipeline(f"{path}", backend_config=backend_config, log_level='INFO')
|
|
|
| def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
| """
|
| data args:
|
| inputs (:obj: `str`)
|
| kwargs
|
| Return:
|
| A :obj:`str`| `Dict`: will be serialized and returned
|
| """
|
| query = data.get('query')
|
| if not query:
|
| return [{'error': 'No query provided'}]
|
|
|
| response = self.pipe([query])
|
| return {'response': response.text} |
