| 69 | |
| 70 | |
| 71 | class ChatMessageHistory(BaseModel): |
| 72 | messages: List[BaseMessage] = [] |
| 73 | |
| 74 | def save_message(self, message: str, message_type: MessageType, **kwargs): |
| 75 | if message_type == MessageType.AIMessage: |
| 76 | self.messages.append(AIMessage(content=message)) |
| 77 | elif message_type == MessageType.UserMessage: |
| 78 | self.messages.append(UserMessage(content=message)) |
| 79 | elif message_type == MessageType.FunctionMessage: |
| 80 | self.messages.append( |
| 81 | FunctionMessage( |
| 82 | content=message, |
| 83 | name=kwargs["name"], |
| 84 | conversational_message=kwargs["conversational_message"], |
| 85 | ) |
| 86 | ) |
| 87 | elif message_type == MessageType.SystemMessage: |
| 88 | self.messages.append(SystemMessage(content=message)) |
| 89 | |
| 90 | def format_message(self): |
| 91 | string_messages = [] |
| 92 | if len(self.messages) > 0: |
| 93 | for m in self.messages: |
| 94 | if isinstance(m, FunctionMessage): |
| 95 | string_messages.append(f"Action: {m.conversational_message}") |
| 96 | continue |
| 97 | |
| 98 | if isinstance(m, UserMessage): |
| 99 | role = "User" |
| 100 | elif isinstance(m, AIMessage): |
| 101 | role = "Assistant" |
| 102 | elif isinstance(m, SystemMessage): |
| 103 | role = "System" |
| 104 | else: |
| 105 | continue |
| 106 | string_messages.append(f"{role}: {m.content}") |
| 107 | return "\n".join(string_messages) + "\n" |
| 108 | return "" |
| 109 | |
| 110 | def get_latest_user_message(self) -> UserMessage: |
| 111 | for message in reversed(self.messages): |
| 112 | if isinstance(message, UserMessage): |
| 113 | return message |
| 114 | return UserMessage(content="n/a") |
| 115 | |
| 116 | def clear(self) -> None: |
| 117 | self.messages = [] |
no outgoing calls