(role: Literal['assistant', 'tool'],
round_message: Dict[str, Any])
| 206 | |
| 207 | @staticmethod |
| 208 | def convert_message(role: Literal['assistant', 'tool'], |
| 209 | round_message: Dict[str, Any]) -> Dict[str, Any]: |
| 210 | |
| 211 | if role == 'assistant': |
| 212 | res_msg: Dict[str, Any] = { |
| 213 | 'role': 'assistant', |
| 214 | 'content': round_message.get('content', ''), |
| 215 | 'tool_calls': [], |
| 216 | } |
| 217 | |
| 218 | tmp_tool_calls = [] |
| 219 | for tool_call in round_message['tool_calls']: |
| 220 | if isinstance(tool_call, ChoiceDeltaToolCall): |
| 221 | if not tool_call.id: |
| 222 | tool_call.id = f'tc_{uuid.uuid4().hex}' |
| 223 | tool_call = tool_call.model_dump( |
| 224 | include=['id', 'index', 'type', 'function']) |
| 225 | else: |
| 226 | raise ValueError( |
| 227 | f'Unsupported tool call type: {type(tool_call)}. Expected ChoiceDeltaToolCall.' |
| 228 | ) |
| 229 | tmp_tool_calls.append(tool_call) |
| 230 | |
| 231 | res_msg['tool_calls'] = tmp_tool_calls |
| 232 | |
| 233 | elif role == 'tool': |
| 234 | # TODO: tbd ... |
| 235 | raise ValueError( |
| 236 | '`tool message` is to be implemented in the future.') |
| 237 | |
| 238 | else: |
| 239 | raise ValueError( |
| 240 | f"Unsupported role: {role}. Supported roles are 'assistant' and 'tool' for now." |
| 241 | ) |
| 242 | |
| 243 | return res_msg |
| 244 | |
| 245 | def chat_stream_mt(self, |
| 246 | messages: List[Dict[str, Any]], |
no test coverage detected