Return the conversation history as a dictionary.
(self)
| 57 | self.current_conversation = CustomConversation() |
| 58 | |
| 59 | def call(self) -> Dict: |
| 60 | """Return the conversation history as a dictionary.""" |
| 61 | all_dialog_turns = {} |
| 62 | try: |
| 63 | # Check if dialog_turns exists and is a list |
| 64 | if hasattr(self.current_conversation, 'dialog_turns'): |
| 65 | if self.current_conversation.dialog_turns: |
| 66 | logger.info(f"Memory content: {len(self.current_conversation.dialog_turns)} turns") |
| 67 | for i, turn in enumerate(self.current_conversation.dialog_turns): |
| 68 | if hasattr(turn, 'id') and turn.id is not None: |
| 69 | all_dialog_turns[turn.id] = turn |
| 70 | logger.info(f"Added turn {i+1} with ID {turn.id} to memory") |
| 71 | else: |
| 72 | logger.warning(f"Skipping invalid turn object in memory: {turn}") |
| 73 | else: |
| 74 | logger.info("Dialog turns list exists but is empty") |
| 75 | else: |
| 76 | logger.info("No dialog_turns attribute in current_conversation") |
| 77 | # Try to initialize it |
| 78 | self.current_conversation.dialog_turns = [] |
| 79 | except Exception as e: |
| 80 | logger.error(f"Error accessing dialog turns: {str(e)}") |
| 81 | # Try to recover |
| 82 | try: |
| 83 | self.current_conversation = CustomConversation() |
| 84 | logger.info("Recovered by creating new conversation") |
| 85 | except Exception as e2: |
| 86 | logger.error(f"Failed to recover: {str(e2)}") |
| 87 | |
| 88 | logger.info(f"Returning {len(all_dialog_turns)} dialog turns from memory") |
| 89 | return all_dialog_turns |
| 90 | |
| 91 | def add_dialog_turn(self, user_query: str, assistant_response: str) -> bool: |
| 92 | """ |
nothing calls this directly
no test coverage detected