File size: 3,675 Bytes
4556150 65f7c11 4556150 65f7c11 4556150 65f7c11 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 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
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, field_validator
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
@field_validator("id", mode="before")
@classmethod
def validate_credits(cls, v):
if isinstance(v, str):
return v
return str(v)
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
@field_validator("id", mode="before")
@classmethod
def validate_credits(cls, v):
if isinstance(v, str):
return v
return str(v)
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)
|