(self, messages: List[Dict])
| 92 | return _func |
| 93 | |
| 94 | def filter_messages(self, messages: List[Dict]) -> List[Dict]: |
| 95 | try: |
| 96 | assert len(messages) % 2 == 1 |
| 97 | for idx, message in enumerate(messages): |
| 98 | assert isinstance(message, dict) |
| 99 | assert "role" in message and "content" in message |
| 100 | assert isinstance(message["role"], str) |
| 101 | assert isinstance(message["content"], str) |
| 102 | assert message["role"] in ["user", "agent"] |
| 103 | if idx % 2 == 0: |
| 104 | assert message["role"] == "user" |
| 105 | else: |
| 106 | assert message["role"] == "agent" |
| 107 | except: |
| 108 | raise SessionExeption("Invalid messages") |
| 109 | threashold_segments = 3500 |
| 110 | return_messages = [] |
| 111 | # only include the latest {threashold_segments} segments |
| 112 | |
| 113 | segments = self._calc_segments(messages[0]["content"]) |
| 114 | |
| 115 | for message in messages[:0:-1]: |
| 116 | segments += self._calc_segments(message["content"]) |
| 117 | if segments >= threashold_segments: |
| 118 | break |
| 119 | return_messages.append(message) |
| 120 | |
| 121 | if len(return_messages) > 0 and return_messages[-1]["role"] == "user": |
| 122 | return_messages.pop() |
| 123 | |
| 124 | instruction = messages[0]["content"] |
| 125 | |
| 126 | omit = len(messages) - len(return_messages) - 1 |
| 127 | |
| 128 | if omit > 0: |
| 129 | instruction += f"\n\n[NOTICE] {omit} messages are omitted." |
| 130 | print(f"Warning: {omit} messages are omitted.") |
| 131 | |
| 132 | return_messages.append({ |
| 133 | "role": "user", |
| 134 | "content": instruction |
| 135 | }) |
| 136 | |
| 137 | return_messages.reverse() |
| 138 | return return_messages |
| 139 | |
| 140 | |
| 141 |
no test coverage detected