Deserialize conversation history from JSON request data. Converts a list of message dictionaries (from the frontend) into Message objects suitable for passing to chat_with_database(). Args: history_data: List of dicts with 'role' and 'content' keys, and optionally '
(
history_data: list[dict]
)
| 353 | |
| 354 | |
| 355 | def deserialize_history( |
| 356 | history_data: list[dict] |
| 357 | ) -> list[Message]: |
| 358 | """Deserialize conversation history from JSON request data. |
| 359 | |
| 360 | Converts a list of message dictionaries (from the frontend) into |
| 361 | Message objects suitable for passing to chat_with_database(). |
| 362 | |
| 363 | Args: |
| 364 | history_data: List of dicts with 'role' and 'content' keys, |
| 365 | and optionally 'tool_calls' and 'tool_results'. |
| 366 | |
| 367 | Returns: |
| 368 | List of Message objects. |
| 369 | """ |
| 370 | if not isinstance(history_data, list): |
| 371 | return [] |
| 372 | |
| 373 | messages = [] |
| 374 | for item in history_data: |
| 375 | if not isinstance(item, dict): |
| 376 | continue |
| 377 | |
| 378 | role_str = item.get('role', '') |
| 379 | content = item.get('content', '') |
| 380 | |
| 381 | try: |
| 382 | role = Role(role_str) |
| 383 | except ValueError: |
| 384 | continue # Skip unknown roles |
| 385 | |
| 386 | # Reconstruct tool calls if present |
| 387 | tool_calls = [] |
| 388 | for tc_data in item.get('tool_calls') or []: |
| 389 | if not isinstance(tc_data, dict): |
| 390 | continue |
| 391 | tool_calls.append(ToolCall( |
| 392 | id=tc_data.get('id', ''), |
| 393 | name=tc_data.get('name', ''), |
| 394 | arguments=tc_data.get('arguments', {}) |
| 395 | )) |
| 396 | |
| 397 | # Reconstruct tool results if present |
| 398 | from pgadmin.llm.models import ToolResult |
| 399 | tool_results = [] |
| 400 | for tr_data in item.get('tool_results') or []: |
| 401 | if not isinstance(tr_data, dict): |
| 402 | continue |
| 403 | tool_results.append(ToolResult( |
| 404 | tool_call_id=tr_data.get('tool_call_id', ''), |
| 405 | content=tr_data.get('content', ''), |
| 406 | is_error=tr_data.get('is_error', False) |
| 407 | )) |
| 408 | |
| 409 | messages.append(Message( |
| 410 | role=role, |
| 411 | content=content, |
| 412 | tool_calls=tool_calls, |