Simple conversation management with a list of dialog turns.
| 49 | MAX_INPUT_TOKENS = 7500 # Safe threshold below 8192 token limit |
| 50 | |
| 51 | class Memory(adal.core.component.DataComponent): |
| 52 | """Simple conversation management with a list of dialog turns.""" |
| 53 | |
| 54 | def __init__(self): |
| 55 | super().__init__() |
| 56 | # Use our custom implementation instead of the original Conversation class |
| 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 | """ |
| 93 | Add a dialog turn to the conversation history. |
| 94 | |
| 95 | Args: |
| 96 | user_query: The user's query |
| 97 | assistant_response: The assistant's response |
| 98 | |
| 99 | Returns: |
| 100 | bool: True if successful, False otherwise |
| 101 | """ |
| 102 | try: |
| 103 | # Create a new dialog turn using our custom implementation |
| 104 | dialog_turn = DialogTurn( |
| 105 | id=str(uuid4()), |
| 106 | user_query=UserQuery(query_str=user_query), |
| 107 | assistant_response=AssistantResponse(response_str=assistant_response), |
| 108 | ) |