封面:PIXIV 82860854

@やたぬき圭

我们在上一篇已经了解了如何使用MongoDB存入每一次的对话记录。接下来我们将尝试使用这些聊天记录为我们的AI提供更加久远的记忆,我们将要自动化这一过程。

前言

在与AI的交互过程中,我们不可能每一分每一秒都在交互,要实现更加合理地管理聊天记录和记忆簇,我们将会使用 “会话” 机制管理每一次交互。注意,这与 ChatGPT 网页版交互理念是不一样的。区别在于,Chat GPT 提供了用户可选的 会话回溯 机制,我们可以找回之前的对话并继续之前的内容。但为什么在这里我没有采用这个方式呢?因为个人助理 有别于 单个 LLM 工具,她是具有 时空属性 的。你既可以理解为,我们的AI助理可以自动化回溯会话这一过程,也可以理解为这是一种“更拟人”的体现。在这种情况下,对话相对会更加自然而流畅,同时我认为 ChatGPT 网页版也是权衡成本,毕竟这么多用户全都自动化会话数据库那开销就要上天了。

你会发现,上一段话多次提到了会话这一概念,那我们如何理解呢?

会话

会话指的是两个或多个参与者之间的交谈或交流。在计算机和网络领域,会话通常指的是用户与系统之间的一段互动时间。例如,当你登录一个网站并与其进行交互(如浏览页面、发送消息等)时,这段时间就被称为一个会话。

会话超时

会话超时则是指在一段时间内没有任何活动后,系统自动结束该会话的状态。这通常是为了安全考虑,比如防止他人在你离开电脑时访问你的账户。比如,如果你在某个网站上停留了太久没有点击任何东西,系统可能会自动注销你,这个过程就叫做会话超时。

简而言之:

  • 会话是你和系统互动的时间段。
  • 会话超时是因为长时间没有互动而自动结束这个时间段的过程。

我们今天要实现的功能就是,设置一个会话系统,当会话超时了,系统会自动把刚刚的会话的所有内容进行总结,并存到 MongoDB 数据库里,便于后续我们回溯。

计时器模块

计时器模块我们有两个选择:线程计时器异步计时器 。这两种计时器各有优劣,但是考虑到本教程是从零教程,同时由于使用异步计时器需要所有的异步操作都需要在事件循环中运行,这可能不利于读者理解。再加之不是所有的函数都是异步函数,故本篇文章暂时采用线程计时器作为会话超时的计时器使用。这里避免误导提前说明,没有对错,只有对于不同场景下的权衡利弊的选择。

我们上一篇文章为大家介绍了自定义模块,今天我们仍然会创建自定义模块。

我们创建一个 cyberaitimer.py 文件,它在文件树里面看起来应该是这样的:

your_project/
│
├── main.py
└── cyberaimodules/
    ├── __init__.py
    ├── cyberaimongo.py 
    └── cyberaitimer.py # 计时器模块

下面是我们的模块代码:

# 计时器模块
import threading

class CyberaiTimer:
    """
    CyberaiTimer 类用于创建一个定时器,在指定的超时时间后执行回调函数。

    Attributes:
        timeout (float): 定时器的超时时间(以秒为单位)。
        callback (function): 定时器超时后要执行的回调函数。
        timer (threading.Timer): threading.Timer 对象,用于实现定时器功能。
    """

    def __init__(self, timeout, callback):
        """
        初始化 CyberaiTimer 实例。

        Args:
            timeout (float): 定时器的超时时间(以秒为单位)。
            callback (function): 定时器超时后要执行的回调函数。
        """
        self.timeout = timeout
        self.callback = callback
        self.timer = None

    def start_timer(self):
        """
        启动定时器。如果定时器已经在运行,则先取消当前定时器,然后重新启动。
        """
        if self.timer:
            self.timer.cancel()
        self.timer = threading.Timer(self.timeout, self.callback)
        self.timer.start()

    def stop_timer(self):
        """
        停止定时器。如果定时器正在运行,则取消定时器并将其设为 None。
        """
        if self.timer:
            self.timer.cancel()
            self.timer = None

这个模块的设计思路是这样的:

  1. 初始化函数 (init):
    • 当我们创建一个新的计时器时,我们需要告诉它两件事: a) 要等多长时间(timeout) b) 时间到了要做什么(callback)
    • 这就像设置一个闹钟,你要设定响铃时间,并决定闹铃响时要做什么。
  2. 启动计时器 (start_timer):
    • 这个函数就像按下闹钟的启动按钮。
    • 如果闹钟已经在运行,我们先关掉它,然后重新设置。这确保我们不会有两个闹钟同时运行。
  3. 停止计时器 (stop_timer):
    • 这就像在闹铃响之前把闹钟关掉。
    • 如果计时器正在运行,我们取消它并将其设置为None(表示没有正在运行的计时器)。
  4. 使用 threading.Timer:
    • Python提供了一个内置的计时器工具叫 threading.Timer
    • 这就像闹钟的内部机制。我们不需要知道它具体如何工作,只需要知道如何使用它。
  5. 为什么使用类?
    • 使用类可以把所有与计时器相关的东西(超时时间、回调函数、开始、停止)组织在一起。
    • 这样,每次需要一个新的计时器时,我们只需创建这个类的一个新实例,而不用到处复制粘贴代码。
  6. 灵活性:
    • 这个设计允许我们创建多个不同的计时器,每个都有自己的超时时间和要执行的任务。

会话 (session_id) 系统

对于当前用户,有且只有一个会话ID是当前有效的,同时还要能够在计时器到时间的时候回调 (call_back) 给相对应的函数,这种情况下用一个全局变量可能是一个比较好的选择……

会话超时后,全局变量擦除,生成新的会话ID,之前的会话ID存入数据库便于以后通过ID来回溯之前的会话。

我们要在之前的 main.py 里面加上一些内容:

session_id = None
chat_id_pool = []

然后在 get_response_from_llm 函数里加上用于更新 session_id 和 chat_id_pool的内容:

global session_id
if session_id is None:
  session_id = str(uuid.uuid4()).replace('-', '')
  print(f"已生成新的会话ID:{session_id}")

# 把本次对话的记录的绝对ID插入到数据库
chat_id_pool.append(chat_id)

(作者正在忙着开发 LICO!! 的 APP 语音交互功能以及折腾一大堆硬件, 教程的事情就先鸽啦~ 如果有任何问题可以去找我问或者私聊都可以……)