From Zero to Cyber Wife! 5 - Summary and Persistent Memory Loading

Cover: PIXIV 82860854

In the last article, we learned how to use MongoDB to store every conversation record. Next, we will attempt to use these chat records to provide our AI with a more long-term memory, and we will automate this process.

Introduction

During interactions with AI, we cannot be engaged every moment. To manage chat histories and memory clusters more effectively, we will use a “session” mechanism to manage each interaction. Note that this differs from the interaction concept of the ChatGPT web version. The distinction is that ChatGPT offers a user-selectable session backtrack mechanism, allowing us to restore previous dialogues and continue from where we left off. But why did I not adopt this approach here? Because a personal assistant differs from a single LLM tool; it possesses temporal and spatial attributes. One could argue that our AI assistant can automate the backtrack process of the session, which might represent a “more human-like” quality. In this context, dialogues can become more natural and fluid. Additionally, I believe that the ChatGPT web version is also balancing costs; after all, automating session databases for numerous users would be incredibly expensive.

You may notice that the previous paragraph mentioned the concept of session multiple times; how should we understand it?

Session

Session refers to a conversation or communication between two or more participants. In computing and networking, a session typically refers to a period of interaction between a user and a system. For instance, when you log into a website and interact with it (like browsing pages or sending messages), that period is known as a session.

Session Timeout

Session timeout refers to the state in which a system automatically ends a session after a certain period of inactivity. This is usually for security purposes, such as preventing others from accessing your account when you leave the computer. For example, if you remain on a website for too long without clicking anything, the system may automatically log you out; this process is called session timeout.

In short:

Session is the period during which you interact with the system.
Session timeout is the process of automatically ending that period due to prolonged inactivity.

The functionality we aim to implement today is to set up a session system where, upon session timeout, the system automatically summarizes all content from the recent session and stores it in the MongoDB database for future reference.

Timer Module

For the timer module, we have two options: thread timer and asynchronous timer. Both timers have their advantages and disadvantages. However, considering that this tutorial serves as a beginner’s guide, and because using an asynchronous timer requires all asynchronous operations to run within the event loop, which may hinder readers’ understanding, we will temporarily adopt the thread timer as the session timeout timer. To avoid misleading, it’s pointed out upfront that there are no absolute rights or wrongs, only trade-offs based on different scenarios.

In the previous article, we introduced custom modules. Today, we will still create a custom module.

We will create a cyberaitimer.py file, which will look like this in the file tree:

your_project/
│
├── main.py
└── cyberaimodules/
    ├── __init__.py
    ├── cyberaimongo.py 
    └── cyberaitimer.py # Timer module

Here’s our module code:

# Timer module
import threading

class CyberaiTimer:
    """
    The CyberaiTimer class creates a timer that executes a callback function after a specified timeout period.

    Attributes:
        timeout (float): Timer's timeout period (in seconds).
        callback (function): Callback function to execute after the timer times out.
        timer (threading.Timer): threading.Timer object implementing the timer functionality.
    """

    def __init__(self, timeout, callback):
        """
        Initializes the CyberaiTimer instance.

        Args:
            timeout (float): Timer's timeout period (in seconds).
            callback (function): Callback function to execute after the timer times out.
        """
        self.timeout = timeout
        self.callback = callback
        self.timer = None

    def start_timer(self):
        """
        Starts the timer. If the timer is already running, it cancels the current timer and restarts it.
        """
        if self.timer:
            self.timer.cancel()
        self.timer = threading.Timer(self.timeout, self.callback)
        self.timer.start()

    def stop_timer(self):
        """
        Stops the timer. If the timer is running, it cancels the timer and sets it to None.
        """
        if self.timer:
            self.timer.cancel()
            self.timer = None

The design concept of this module is as follows:

Initialization function (__init__):
- When we create a new timer, we need to inform it two things: a) How long to wait (timeout) b) What to do when the time is up (callback)
- This is similar to setting an alarm clock; you set the ringing time and decide what to do when the alarm rings.
Start timer (start_timer):
- This function is like pressing the start button on the alarm clock.
- If the alarm is already running, we first turn it off and then reset it to ensure we don’t have two alarms running simultaneously.
Stop timer (stop_timer):
- This is like turning off the alarm before it rings.
- If the timer is running, we cancel it and set it to None (indicating no timer is currently running).
Using threading.Timer:
- Python provides a built-in timer tool called threading.Timer.
- This operates like the internal mechanism of an alarm clock. We don’t need to know precisely how it works, just how to use it.
Why use a class?:
- Classes allow us to organize all things related to the timer (timeout duration, callback function, start, stop) together.
- Thus, each time we need a new timer, we can create a new instance of this class without having to copy and paste code all over.
Flexibility:
- This design allows us to create multiple different timers, each with its own timeout period and tasks to execute.

Session (session_id) System

For the current user, there is only one valid session ID at any moment, which must also call back to the corresponding function when the timer times out. In this case, using a global variable might be a suitable choice…

(Yes, you might feel that global variables are inappropriate; if you think so, you are certainly not a newbie, so you can use whatever you like—context managers like contextmanager, threading.local(), or contextvars.ContextVar, etc.)

After session timeout, the global variable resets, creating a new session ID, while the previous session ID is stored in the database for later retrieval.

We need to add some content to the previous main.py:

session_id = None
chat_id_pool = []

Then, add content in the get_response_from_llm function that updates session_id and chat_id_pool:

global session_id
if session_id is None:
    session_id = str(uuid.uuid4()).replace('-', '')
    print(f"A new session ID has been generated: {session_id}")

# Insert the absolute ID of this conversation into the chat ID list
chat_id_pool.append(chat_id)

The logic here is as follows: when we begin chatting with AI, check for the presence of a timer; if there is none, create a new timer to start counting. When the get_response_from_llm detects that session_id is empty, this indicates that there is currently no session, prompting the creation of a new session_id. After each AI response is completed, the chat_id of this conversation (noting that this refers to a back-and-forth exchange of two messages) is inserted into the chat_id_pool list for summarization purposes, while also refreshing the timer to restart the session count.

When we have not spoken with the AI for a long time and reach the timeout duration, the timer triggers the closure session function end_session_and_start_new, which is responsible for the following tasks:

Read all chat_id values in the chat_id_pool and use a for loop to retrieve all corresponding chat records from the database. These records represent the content of the current chat; the end_session_and_start_new function summarizes this content.
Write all chat_id values from chat_id_pool, along with the summarized content, into the MongoDB database, referencing a data structure like this:

'timestamp': timestamp_str, # Timestamp
'session_id': session_id_json_data, # Session ID
'session_summary': session_summary, # Session summary
'session_id_pool': session_id_pool_json_data # Corresponding chat record list

To implement these functionalities, we will need to modify several components. Here’s a list to aid understanding:

Modify the cyberaimongo module to add generate_session_id_json and insert_session functions; the former generates session data in JSON format, and the latter writes the session into the database.
Create a function end_session_and_start_new in main.py, as mentioned above.
Import the timer module we created at the beginning of the article into main.py. After creating the end_session_and_start_new function, we will initialize this timer module.
Add timer reset and chat_id_pool related operations in get_response_from_llm, updating the new session_id with each successful reply, ensuring that every session_id is unique and different.

We will go step by step… (This was not written by AI! This is not COT!!!)

Modify the `cyberaimongo` module

We will add two new functions:

`generate_session_id_json`

def generate_session_id_json(self, chat_id_pool: list, session_summary: str, session_id: str, timestamp: datetime = None) -> str:
    """
    Generate a JSON string for the Session ID.

    :param chat_id_pool: Chat ID pool
    :param session_summary: Session summary
    :param session_id: Session ID
    :param timestamp: Timestamp
    :return: JSON string for the session ID
    """
    if timestamp is None:
        timestamp = datetime.now()
    timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M")

    # Convert ObjectId objects to strings
    chat_id_pool = [str(id) for id in chat_id_pool]

    chat_id_pool_json_data = json.dumps(chat_id_pool)
    session_id_json_data = json.dumps(str(session_id))  # Convert ObjectId object to string

    chat_data = {
        'timestamp': timestamp_str,
        'session_id': session_id_json_data,
        'session_summary': session_summary,
        'chat_id_pool': chat_id_pool_json_data,
    }
    return json.dumps(chat_data)

`insert_session`

def insert_session(self, collection_name: str, chat_id_pool: list, session_summary: str, session_id: str, timestamp: datetime = None) -> str:
    """
    Insert the session record into the specified collection.

    :param collection_name: Collection name
    :param chat_id_pool: Session ID pool
    :param session_summary: Session summary
    :param session_id: Session ID
    :param timestamp: Timestamp
    :return: ID of the inserted record
    """
    json_data = self.generate_session_id_json(chat_id_pool, session_summary, session_id, timestamp)
    collection = self.db[collection_name]
    if isinstance(json_data, str):
        json_data = json.loads(json_data)
    return collection.insert_one(json_data).inserted_id

`end_session_and_start_new` function

def end_session_and_start_new():
    """
    Ends the current session and starts a new one
    """
    # Initialize an empty string to store all metadata
    metadata_str = ''
    global session_id

    # If the session ID is not empty
    for chat_id in chat_id_pool:
        # Retrieve the current session’s chat records
        chat_data = mongo_manager.get_data_by_id(
            collection_name='daily',
            id=chat_id
        )
        # Extract the required fields and concatenate them into a string
        if chat_data:
            timestamp = chat_data.get('timestamp', '')
            ai_reply = chat_data.get('ai_reply', '')
            human_message = chat_data.get('human_message', '')
            metadata_str += f'{timestamp} \n User: {human_message} \n CyberAi: {ai_reply} \n\n'

    if not metadata_str.strip():
        print('No chat records, no summary needed')
        return None
  
    # Generate session summary
    session_summary = summarize_chat_history(metadata_str)

    # Insert the session summary into the database
    session_id = mongo_manager.insert_session(
        collection_name='session-history',
        session_id_pool=chat_id_pool,
        session_summary=session_summary,
        session_id=session_id
    )
    print(f"Session summary generated")

    # Clear the chat records and session ID
    chat_id_pool.clear()
    chat_history.clear()
    print('Cleared chat history cache')
    session_id = None
    print('Cleared session ID')

Initialize Timer

# Add at the beginning of main.py:
from cyberaimodules import cyberaitimer
session_timeout_time = 60  # Define session timeout period (in seconds)

# After creating the end_session_and_start_new function, initialize the timer:
session_timer = cyberaitimer.CyberaiTimer(timeout=session_timeout_time, callback=end_session_and_start_new)

Modify the `get_response_from_llm` function

Modifications include:

Eliminating the previous context summary memory, leaving only window memory, and using SUMMARY_THRESHOLD to decide the number of dialogue turns.
Adding user input and LLM response to the chat records, previously handled outside the function.
Starting the session timer.
Temporarily commenting out the database query for loading memory, as we will soon add the persistent memory loading feature.
Resetting or generating functionality related to session_id, as discussed earlier.

The complete code looks like this:

def get_response_from_llm(question: str) -> str:
    """
    Get the response from LLM
    :param question: User's question
    :return: LLM's response
    """

    global chat_history
    # Add user input to chat history
    chat_history.append(('human', question))


    # Fetch recent chat session summaries
    recent_chat_session_history = mongo_manager.get_recent_mem(n=2, collection_name='session-history', datapart='session_summary')
    yesterday_chat_session_history = mongo_manager.get_yesterday_mem(collection_name='session-history', datapart='session_summary')

    # Get the recent chat history window
    chat_history_window = "\n".join([f"{role}: {content}" for role, content in chat_history[-2*SUMMARY_THRESHOLD:-1]])
    chat_history_prompt = f"Here is yesterday's memory: {yesterday_chat_session_history}\n Here is the recent memory: {recent_chat_session_history} \n Here is the current chat history:\n {chat_history_window}"  

    print(f"chat_history_prompt: {chat_history_prompt}\n")
    # Create message list
    message = [
        {"role": "system", "content": "You are a catgirl! Output in Chinese."},
        {"role": "assistant", "content": chat_history_prompt},
        {"role": "user", "content": question},
    ]

    # Call LLM to get response
    response = chat_model.chat.completions.create(
        model='gpt-4o',
        messages=message,
        temperature=0.7,
    )
  
    print(f"message: {message}")
    # Get the response content
    response_str = response.choices[0].message.content

    # Add the LLM's response to the chat history
    chat_history.append(('ai', response_str))

    # Start the session timer
    session_timer.start_timer()


    # Insert the conversation record into the database
    chat_id = mongo_manager.insert_chat(
        collection_name='daily',
        ai_reply=response_str,
        human_message=question,
        session_id=str(uuid.uuid4()),
    )

    # query = mongo_manager.get_data_by_id(collection_name='daily', id=chat_id)
  
    global session_id
    if session_id is None:
        session_id = str(uuid.uuid4()).replace('-', '')
        print(f"A new session ID has been generated: {session_id}")

    # Insert the absolute ID of this conversation into the chat ID database
    chat_id_pool.append(chat_id)

    return response_str

Memory Loading

In the previous two sections, we have completed writing the chat records and session memory; now we will add two simple functions: loading the last few sessions and loading all sessions from yesterday. Here’s a brief introduction to the logic behind persistent memory loading: in practical usage, occurrences that have recently happened or been mentioned are more frequently referenced than those from long ago. Considering cost control, we can let the LLM load summaries from recent sessions each time, achieving a balance between memory and cost. If there is no strict requirement for response speed, a memory manager can also be added to manage loading more memory or recalling specific session periods. If quick responses are required, keyword triggers or a history preference library (needing prior triggered data support) could determine whether to activate memory loading. Due to space limitations, we won’t elaborate further; that’s the general idea.

We need to modify the cyberaimongo module to add two functions: get_recent_mem and get_yesterday_mem, which fetch the summaries of the last n sessions and all of yesterday’s summaries, respectively.

def get_recent_mem(self, n: int = 1, collection_name: str = 'session-history', datapart: str = 'session_summary') -> str:
    """
    Get the last n records. If the record count is less than n, return the actual number of records.
  
    :param n: Record count
    :param collection_name: Collection name
    :param datapart: Data part
    :return: Summary string of records
    """
    collection = self.db[collection_name]
    cursor = collection.find().sort('timestamp', -1).limit(n)
    summaries = [[item['timestamp'], item.get(datapart, ''), item.get('tags', [])] for item in cursor]
    if not summaries:
        return ''
    summaries_str = '\n'.join([str(summary) for summary in summaries])
    return summaries_str
  
def get_yesterday_mem(self, collection_name: str = 'session-history', datapart: str = 'session_summary') -> str:
    """
    Get all records from yesterday. If no records exist, return an empty string.
  
    :param collection_name: Collection name
    :param datapart: Data part
    :return: Summary string of records
    """
    now = datetime.now()
    start = datetime(now.year, now.month, now.day) - timedelta(days=1)
    end = start + timedelta(days=1)
    sessions = self.get_mem_in_time_range(collection_name, start, end)
    if not sessions:
        return ''
    summaries = [f"{session['timestamp']} {session.get(datapart, '')}" for session in sessions]
    summaries_str = '\n'.join(summaries)
    return summaries_str

Next, we need to add the loading logic in the get_response_from_llm function. This part will be quite interesting, as there are many ways to trigger memory recall, each with its pros and cons; there is no one-size-fits-all solution, so choose what suits you best. Another small detail: upon acquiring memory, should it be placed in the system or assistant prompts? There are various circumstances; I personally prefer placing it in the assistant prompt, as it won’t significantly impact the foundational prompts. Meanwhile, placing it in system might lead to ambiguity or strange bugs if you have feedback on subsequent tool calls.~~(Of course, if it’s Claude, that’s less of a concern, so specific cases should be analyzed; I recommend conducting your experiments to verify… )~~

We add a few lines simply:

recent_chat_session_history = mongo_manager.get_recent_mem(n=2, collection_name='session-history', datapart='session_summary')
yesterday_chat_session_history = mongo_manager.get_yesterday_mem(collection_name='session-history', datapart='session_summary')
chat_history_prompt = f"Here is memory from yesterday: {yesterday_chat_session_history}\n Here is recent memory: {recent_chat_session_history} \n Here is the current chat history:\n {chat_history_window}"

If you don’t want it to trigger every single time, you can implement a secondary trigger or use keyword triggering and tweak n to set how many past session contents to load.

Here’s the complete main.py:

# main.py
import os
import uuid
from openai import OpenAI

# Import previously created cyberaimodules
from cyberaimodules import cyberaimongo
from cyberaimodules import cyberaitimer

mongo_host = os.getenv('MONGO_HOST', 'localhost')
mongo_port = int(os.getenv('MONGO_PORT', 27017))
mongo_user = os.getenv('MONGO_USER', 'admin')
mongo_password = os.getenv('MONGO_PASSWORD', 'secret')
# The database is named chat_history
mongo_db_name = os.getenv('MONGO_DB_NAME', 'chat_history')

# Initialize MongoDB client
mongo_manager = cyberaimongo.MongoManager(
    host=mongo_host,
    port=mongo_port,
    username=mongo_user,
    password=mongo_password,
    db_name=mongo_db_name,
)

# Initialize OpenAI model
chat_model = OpenAI(
    base_url="https://api.openai.com/v1/",  # Replace with your backend API address
    api_key="sk-SbmHyhKJHt3378h9dn1145141919810D1Fbcd12d"  # This is used for authentication
)

# Chat records and summary lists
chat_history = []
SUMMARY_THRESHOLD = 4  # Define the context window length
session_timeout_time = 600  # Define the session timeout period (in seconds)
session_id = None
chat_id_pool = []

def summarize_chat_history(chat_history_window):
    """
    Summarizes the recent chat records.
    :param chat_history_window: Recent chat records
    :return: Summary string
    """
    # Create summary prompt
    summary_prompt = f"请总结以下对话内容：\n{chat_history_window}"

    print(f"Generating summary for: {chat_history_window}")
    # Call LLM to generate summary
    summary_response = chat_model.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{"role": "user", "content": summary_prompt}],
        temperature=0.7,
    )
    # Get the summary content
    summary_str = summary_response.choices[0].message.content
    print(f"Generated summary: {summary_str}")
    return summary_str

def end_session_and_start_new():
    """
    Ends the current session and starts a new one
    """
    # Initialize an empty string to store all metadata
    metadata_str = ''
    global session_id

    # If the session ID is not empty
    for chat_id in chat_id_pool:
        # Retrieve the current session's chat records
        chat_data = mongo_manager.get_data_by_id(
            collection_name='daily',
            id=chat_id
        )
        # Extract required fields and concatenate them into a string
        if chat_data:
            timestamp = chat_data.get('timestamp', '')
            ai_reply = chat_data.get('ai_reply', '')
            human_message = chat_data.get('human_message', '')
            metadata_str += f'{timestamp} \n User: {human_message} \n CyberAi: {ai_reply} \n\n'

    if not metadata_str.strip():
        print('No chat records, no summary needed')
        return None
  
    # Generate session summary
    session_summary = summarize_chat_history(metadata_str)
    print(f"Session summary: {session_summary}")

    # Insert the session summary into the database
    session_id = mongo_manager.insert_session(
        collection_name='session-history',
        chat_id_pool=chat_id_pool,
        session_summary=session_summary,
        session_id=session_id
    )
    print(f"Session summary generated")

    # Clear the chat records and session ID
    chat_id_pool.clear()
    chat_history.clear()
    print('Cleared chat history cache')
    session_id = None
    print('Cleared session ID')

session_timer = cyberaitimer.CyberaiTimer(timeout=session_timeout_time, callback=end_session_and_start_new)

def get_response_from_llm(question: str) -> str:
    """
    Gets the response from LLM
    :param question: User's question
    :return: LLM's response
    """
    global chat_history
    # Add user input to chat history
    chat_history.append(('human', question))

    recent_chat_session_history = mongo_manager.get_recent_mem(n=2, collection_name='session-history', datapart='session_summary')
    yesterday_chat_session_history = mongo_manager.get_yesterday_mem(collection_name='session-history', datapart='session_summary')

    # Get the recent chat history window
    chat_history_window = "\n".join([f"{role}: {content}" for role, content in chat_history[-2*SUMMARY_THRESHOLD:-1]])
    chat_history_prompt = f"Here is yesterday's memory: {yesterday_chat_session_history}\n Here is recent memory: {recent_chat_session_history} \n Here is the current chat history:\n {chat_history_window}"  

    print(f"chat_history_prompt: {chat_history_prompt}\n")
    # Create message list
    message = [
        {"role": "system", "content": "You are a catgirl! Output in Chinese."},
        {"role": "assistant", "content": chat_history_prompt},
        {"role": "user", "content": question},
    ]

    # Call LLM to get response
    response = chat_model.chat.completions.create(
        model='gpt-4o',
        messages=message,
        temperature=0.7,
    )
  
    print(f"message: {message}")
    # Retrieve response content
    response_str = response.choices[0].message.content

    # Add the LLM's response to the chat history
    chat_history.append(('ai', response_str))

    # Start the session timer
    session_timer.start_timer()

    # Insert the conversation record into the database
    chat_id = mongo_manager.insert_chat(
        collection_name='daily',
        ai_reply=response_str,
        human_message=question,
        session_id=str(uuid.uuid4()),
    )

    # query = mongo_manager.get_data_by_id(collection_name='daily', id=chat_id)
  
    global session_id
    if session_id is None:
        session_id = str(uuid.uuid4()).replace('-', '')
        print(f"A new session ID has been generated: {session_id}")

    # Insert the absolute ID of this conversation into the chat ID database
    chat_id_pool.append(chat_id)

    return response_str

if __name__ == "__main__":
    while True:
        user_input = input("\nInput question or type 'exit' to quit: ")
        if user_input.lower() == 'exit':
            print("Goodbye")
            break 
      
        # Get LLM's response
        response = get_response_from_llm(user_input)
      
        # Print LLM's response
        print(response)

Here’s the complete cyberaimongo module:

# cyberaimongo.py - MongoDB Management Module
import json
import uuid
from datetime import datetime, timedelta
from pymongo import MongoClient

class MongoManager:
    def __init__(self, host: str = 'localhost', port: int = 27017, username: str = 'admin', password: str = 'secret', db_name: str = 'chat_history'):
        """
        Initializes MongoManager class, connecting to the MongoDB database.

        :param host: MongoDB host address
        :param port: MongoDB port number
        :param username: MongoDB username
        :param password: MongoDB password
        :param db_name: Database name
        """
        self.client = MongoClient(f'mongodb://{username}:{password}@{host}:{port}/')
        self.db = self.client[db_name]

    def generate_chat_json(self, ai_reply: str, human_message: str, session_id: str, timestamp: datetime = None) -> str:
        """
        Generates a JSON string for the chat record.

        :param ai_reply: AI reply content
        :param human_message: Human message content
        :param session_id: Session ID
        :param timestamp: Timestamp
        :return: JSON string for the chat record
        """
        if timestamp is None:
            timestamp = datetime.now()
        timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M")
        chat_data = {
            'message_id': str(uuid.uuid4()),
            'ai_reply': ai_reply,
            'human_message': human_message,
            'timestamp': timestamp_str,
            'session_id': session_id
        }
        return json.dumps(chat_data, ensure_ascii=False, indent=4)
  
    def generate_session_id_json(self, chat_id_pool: list, session_summary: str, session_id: str, timestamp: datetime = None) -> str:
        """
        Generates a JSON string for the Session ID.

        :param chat_id_pool: Chat ID pool
        :param session_summary: Session summary
        :param session_id: Session ID
        :param timestamp: Timestamp
        :return: JSON string for the session ID
        """
        if timestamp is None:
            timestamp = datetime.now()
        timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M")

        # Convert ObjectId objects to strings
        chat_id_pool = [str(id) for id in chat_id_pool]

        chat_id_pool_json_data = json.dumps(chat_id_pool)
        session_id_json_data = json.dumps(str(session_id))  # Convert ObjectId object to string

        chat_data = {
            'timestamp': timestamp_str,
            'session_id': session_id_json_data,
            'session_summary': session_summary,
            'chat_id_pool': chat_id_pool_json_data,
        }
        return json.dumps(chat_data)
  
    def insert_chat(self, collection_name: str, ai_reply: str, human_message: str, session_id: str, timestamp: datetime = None) -> str:
        """
        Inserts chat records into the specified collection.

        :param collection_name: Collection name
        :param ai_reply: AI reply content
        :param human_message: Human message content
        :param session_id: Session ID
        :param timestamp: Timestamp
        :return: ID of the inserted record
        """
        json_data = self.generate_chat_json(ai_reply, human_message, session_id, timestamp)
        collection = self.db[collection_name]
        if isinstance(json_data, str):
            json_data = json.loads(json_data)
        return collection.insert_one(json_data).inserted_id

    def insert_session(self, collection_name: str, chat_id_pool: list, session_summary: str, session_id: str, timestamp: datetime = None) -> str:
        """
        Inserts the session record into the specified collection.

        :param collection_name: Collection name
        :param chat_id_pool: Session ID pool
        :param session_summary: Session summary
        :param session_id: Session ID
        :param timestamp: Timestamp
        :return: ID of the inserted record
        """
        json_data = self.generate_session_id_json(chat_id_pool, session_summary, session_id, timestamp)
        collection = self.db[collection_name]
        if isinstance(json_data, str):
            json_data = json.loads(json_data)
        return collection.insert_one(json_data).inserted_id

    def get_mem_in_time_range(self, collection_name: str, start_time: datetime, end_time: datetime) -> list:
        """
        Gets records within a specified time range.

        :param collection_name: Collection name
        :param start_time: Start time
        :param end_time: End time
        :return: List of records
        """
        collection = self.db[collection_name]
        cursor = collection.find({
            'timestamp': {
                '$gte': start_time.isoformat(),
                '$lte': end_time.isoformat()
            }
        })
        return list(cursor)

    def get_data_by_id(self, collection_name: str, id: str) -> dict:
        """
        Retrieves records by ID.

        :param collection_name: Collection name
        :param id: Record ID
        :return: Record data
        """
        collection = self.db[collection_name]
        data = collection.find_one({'_id': id})
        return data
  
    def get_recent_mem(self, n: int = 1, collection_name: str = 'session-history', datapart: str = 'session_summary') -> str:
        """
        Gets the last n records. If the record count is less than n, return the actual number of records.

        :param n: Record count
        :param collection_name: Collection name
        :param datapart: Data part
        :return: Summary string of records
        """
        try:
            collection = self.db[collection_name]
            cursor = collection.find().sort('timestamp', -1).limit(n)
            summaries = [[item['timestamp'], item[datapart]] for item in cursor]
          
            if not summaries:
                return ''
          
            summaries_str = '\n'.join([str(summary) for summary in summaries])
            return summaries_str
      
        except Exception as e:
            print(f"An error occurred: {e}")
            return ''

  
    def get_yesterday_mem(self, collection_name: str = 'session-history', datapart: str = 'session_summary') -> str:
        """
        Gets all records from yesterday. If no records exist, return an empty string.
  
        :param collection_name: Collection name
        :param datapart: Data part
        :return: Summary string of records
        """
        now = datetime.now()
        start = datetime(now.year, now.month, now.day) - timedelta(days=1)
        end = start + timedelta(days=1)
        sessions = self.get_mem_in_time_range(collection_name, start, end)
        if not sessions:
            return ''
        summaries = [f"{session['timestamp']} {session[datapart]}" for session in sessions]
        summaries_str = '\n'.join(summaries)
        return summaries_str

With this setup, running the main program should allow for automatic summarization and session memory/loading functionalities. In practical use, it is recommended to adopt a hybrid triggering mechanism, also known as a dynamic scheme; when you need a quick reply, just use keywords. If speed is less critical, develop a dedicated memory manager to trigger and manage memory reads and writes.

Further Exploration

How about trying to write the previously discussed TTS functions as a module and integrate it into the existing code? It could be a lot of fun!

Introduction#

Session#

Session Timeout#

Timer Module#

Session (session_id) System#

Modify the cyberaimongo module#

generate_session_id_json#

insert_session#

end_session_and_start_new function#

Initialize Timer#

Modify the get_response_from_llm function#

Memory Loading#

Further Exploration#