File size: 3,322 Bytes
4556150 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
import uuid
from datetime import datetime
from typing import Any, Optional, Union
from metagpt.actions.action import Action
from metagpt.actions.action_output import ActionOutput
from pydantic import BaseModel, Field
from message_enum import SentenceType
class SentenceValue(BaseModel):
answer: str
class Sentence(BaseModel):
type: str
id: Optional[str] = None
value: SentenceValue
is_finished: Optional[bool] = None
class Sentences(BaseModel):
id: Optional[str] = None
action: Optional[str] = None
role: Optional[str] = None
skill: Optional[str] = None
description: Optional[str] = None
timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
status: str
contents: list[dict]
class NewMsg(BaseModel):
"""Chat with MetaGPT"""
query: str = Field(description="Problem description")
config: dict[str, Any] = Field(description="Configuration information")
class LLMAPIkeyTest(BaseModel):
"""APIkey"""
api_key: str = Field(description="API Key")
llm_type: str = Field(description="Model Type")
class ErrorInfo(BaseModel):
error: str = None
traceback: str = None
class ThinkActStep(BaseModel):
id: str
status: str
title: str
timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
description: str
content: Sentence = None
class ThinkActPrompt(BaseModel):
message_id: int = None
timestamp: str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f%z"))
step: ThinkActStep = None
skill: Optional[str] = None
role: Optional[str] = None
def update_think(self, tc_id, action: Action):
self.step = ThinkActStep(
id=str(tc_id),
status="running",
title=action.desc,
description=action.desc,
)
def update_act(self, message: Union[ActionOutput, str], is_finished: bool = True):
if is_finished:
self.step.status = "finish"
self.step.content = Sentence(
type=SentenceType.TEXT.value,
id=str(1),
value=SentenceValue(answer=message.content if is_finished else message),
is_finished=is_finished,
)
@staticmethod
def guid32():
return str(uuid.uuid4()).replace("-", "")[0:32]
@property
def prompt(self):
return self.json(exclude_unset=True)
class MessageJsonModel(BaseModel):
steps: list[Sentences]
qa_type: str
created_at: datetime = Field(default_factory=datetime.now)
query_time: datetime = Field(default_factory=datetime.now)
answer_time: datetime = Field(default_factory=datetime.now)
score: Optional[int] = None
feedback: Optional[str] = None
def add_think_act(self, think_act_prompt: ThinkActPrompt):
s = Sentences(
action=think_act_prompt.step.title,
skill=think_act_prompt.skill,
description=think_act_prompt.step.description,
timestamp=think_act_prompt.timestamp,
status=think_act_prompt.step.status,
contents=[think_act_prompt.step.content.dict()],
)
self.steps.append(s)
@property
def prompt(self):
return self.json(exclude_unset=True)
|