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)