| from abc import ABC, abstractmethod |
| from typing import Optional, List |
| import traceback |
| from collections import deque |
| from queue import Queue |
| from threading import Thread |
|
|
| import torch |
| import transformers |
| from models.loader import LoaderCheckPoint |
|
|
|
|
| class AnswerResult: |
| """ |
| 消息实体 |
| """ |
| history: List[List[str]] = [] |
| llm_output: Optional[dict] = None |
|
|
|
|
| class BaseAnswer(ABC): |
| """上层业务包装器.用于结果生成统一api调用""" |
|
|
| @property |
| @abstractmethod |
| def _check_point(self) -> LoaderCheckPoint: |
| """Return _check_point of llm.""" |
|
|
| @property |
| @abstractmethod |
| def _history_len(self) -> int: |
| """Return _history_len of llm.""" |
|
|
| @abstractmethod |
| def set_history_len(self, history_len: int) -> None: |
| """Return _history_len of llm.""" |
|
|
| def generatorAnswer(self, prompt: str, |
| history: List[List[str]] = [], |
| streaming: bool = False): |
| pass |
|
|