| 61 | |
| 62 | |
| 63 | class ChatData(BaseModel): |
| 64 | messages: List[Message] |
| 65 | |
| 66 | class Config: |
| 67 | json_schema_extra = { |
| 68 | "example": { |
| 69 | "messages": [ |
| 70 | { |
| 71 | "role": "user", |
| 72 | "content": "What standards for letters exist?", |
| 73 | } |
| 74 | ] |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | @validator("messages") |
| 79 | def messages_must_not_be_empty(cls, v): |
| 80 | if len(v) == 0: |
| 81 | raise ValueError("Messages must not be empty") |
| 82 | return v |
| 83 | |
| 84 | def get_last_message_content(self) -> str: |
| 85 | """ |
| 86 | Get the content of the last message along with the data content if available. Fallback to use data content from previous messages |
| 87 | """ |
| 88 | if len(self.messages) == 0: |
| 89 | raise ValueError("There is not any message in the chat") |
| 90 | last_message = self.messages[-1] |
| 91 | message_content = last_message.content |
| 92 | for message in reversed(self.messages): |
| 93 | if message.role == MessageRole.USER and message.annotations is not None: |
| 94 | annotation_contents = ( |
| 95 | annotation.to_content() for annotation in message.annotations |
| 96 | ) |
| 97 | annotation_text = "\n".join(annotation_contents) |
| 98 | message_content = f"{message_content}\n{annotation_text}" |
| 99 | break |
| 100 | return message_content |
| 101 | |
| 102 | def get_history_messages(self) -> List[Message]: |
| 103 | """ |
| 104 | Get the history messages |
| 105 | """ |
| 106 | return [ |
| 107 | ChatMessage(role=message.role, content=message.content) |
| 108 | for message in self.messages[:-1] |
| 109 | ] |
| 110 | |
| 111 | def is_last_message_from_user(self) -> bool: |
| 112 | return self.messages[-1].role == MessageRole.USER |
| 113 | |
| 114 | |
| 115 | class SourceNodes(BaseModel): |
nothing calls this directly
no outgoing calls
no test coverage detected