67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
import json
|
|
import openai
|
|
from typing import Optional, List, Dict, Any, Tuple
|
|
|
|
|
|
class AIMessageBase(object):
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def __str__(self) -> str:
|
|
return json.dumps(vars(self))
|
|
|
|
|
|
class AIMessage(AIMessageBase):
|
|
def __init__(self, user: str, message: str) -> None:
|
|
self.user = user
|
|
self.message = message
|
|
|
|
|
|
class AIResponse(AIMessageBase):
|
|
def __init__(self, answer: str, answer_needed: bool, staff: Optional[str], picture: Optional[str], hack: bool) -> None:
|
|
self.answer = answer
|
|
self.answer_needed = answer_needed
|
|
self.staff = staff
|
|
self.picture = picture
|
|
self.hack = hack
|
|
|
|
|
|
class AIResponder(object):
|
|
def __init__(self, config: Dict[str, Any]) -> None:
|
|
self.config = config
|
|
self.history: List[Dict[str, Any]] = []
|
|
|
|
def _message(self, message: AIMessage, limit: Optional[int] = None) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
|
|
messages = []
|
|
messages.append({"role": "system", "content": self.config["system"]})
|
|
if limit is None:
|
|
history = self.history[:]
|
|
else:
|
|
history = self.history[-limit:]
|
|
history.append({"role": "user", "content": str(message)})
|
|
for msg in history:
|
|
messages.append(msg)
|
|
return messages, history
|
|
|
|
async def send(self, message: AIMessage) -> AIResponse:
|
|
limit = self.config["history-limit"]
|
|
while True:
|
|
messages, history = self._message(message, limit)
|
|
try:
|
|
result = await openai.ChatCompletion.acreate(model=self.config["model"],
|
|
messages=messages,
|
|
temperature=self.config["temperature"],
|
|
top_p=self.config["top-p"],
|
|
presence_penalty=self.config["presence-penalty"],
|
|
frequency_penalty=self.config["frequency-penalty"])
|
|
except openai.error.InvalidRequestError as err:
|
|
if 'maximum context length is' in str(err) and limit > 2:
|
|
limit -= 1
|
|
continue
|
|
raise err
|
|
answer = result['choices'][0]['message']
|
|
response = json.loads(answer['content'])
|
|
history.append(answer)
|
|
self.history = history
|
|
return response
|