0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Python と SQLite で AI の長期記憶を履歴から現在状態へ再構成する

0
Posted at

AI エージェントへ長期記憶を持たせると、同じ属性について古い値と新しい値が同時に保存されます。たとえば勤務先が Company A から Company B へ変わった場合、過去の問い合わせには Company A が必要ですが、現在の問い合わせでは Company B を選ぶ必要があります。

既稿では、AI エージェントが長く動くほど「情報を保持すること」と「どの情報が現在も有効かを管理すること」を分ける必要があると整理しました[1]。RD-Forget も、保存する情報と回答時に利用する情報を分離し、現在状態の問い合わせでは置換済みの値を抑制しながら、過去についての問い合わせでは以前の値を再び利用する構成を取っています[2]。

本稿では、このうち時間的有効性による状態選択だけを Python と SQLite で実装します。値を上書きせず履歴として残し、as_of() で指定時点の状態を再構成し、current() で現在値を取得します。

実装する状態モデル

同じ属性について、次の履歴があるとします。

2026-04-01  employer = Company A
2026-09-01  employer = Company B

現在値だけを保存すると、2026 年 6 月時点の勤務先を取得できません。反対に両方を無条件に検索対象へ入れると、現在の勤務先を聞いたときに古い値も候補へ残ります。

そこで、各値に有効期間を持たせます。

Company A  [2026-04-01, 2026-09-01)
Company B  [2026-09-01, ∞)

区間は半開区間 [valid_from, valid_to) とします。valid_from は値が有効になる時刻、valid_to は値が有効でなくなる時刻です。valid_to = NULL は終了時刻が未定であることを表します。

Fortunate Recall は、長期記憶のライフサイクル管理として slot-key supersession と event-time validity を扱っています[3]。Temporal Semantic Memory も、会話が行われた時刻だけでなく、情報そのものが意味的に有効な時間を扱う必要性を示しています[4]。

本稿では、この考え方を次の 5 列へ縮小します。

列 役割
subject 誰または何についての記憶か
attribute どの属性か
value 属性の値
valid_from 値が有効になる時刻
valid_to 値が有効でなくなる時刻

同じ属性は (subject, attribute) で識別します。

実行環境

実装は Python 3.12 以上を前提とします。データベースには標準ライブラリの sqlite3 を使います。SQLite は別プロセスのデータベースサーバーを必要とせず、Python 標準の sqlite3 から接続できます[5]。

テストには pytest を使います。

python -m pip install pytest

ファイルは次の 2 個です。

memory_history.py
test_memory_history.py

履歴テーブルを作る

まず、履歴を保存するテーブルを定義します。

memory_history.py
from __future__ import annotations

import sqlite3
from pathlib import Path
from time import time


SCHEMA = """
CREATE TABLE IF NOT EXISTS memory_history (
    id INTEGER PRIMARY KEY,
    subject TEXT NOT NULL,
    attribute TEXT NOT NULL,
    value TEXT NOT NULL,
    valid_from INTEGER NOT NULL,
    valid_to INTEGER,
    CHECK (valid_to IS NULL OR valid_from < valid_to),
    UNIQUE (subject, attribute, valid_from)
);

CREATE INDEX IF NOT EXISTS idx_memory_history_lookup
    ON memory_history(subject, attribute, valid_from DESC);
"""


def connect(path: str | Path = ":memory:") -> sqlite3.Connection:
    con = sqlite3.connect(path, isolation_level=None)
    con.executescript(SCHEMA)
    return con

時刻は UTC の Unix time を INTEGER で保存します。

CHECK 制約では、有効期間が閉じている場合に valid_from < valid_to を保証します。UNIQUE (subject, attribute, valid_from) によって、同じ属性へ同一開始時刻の履歴が重複して入ることも防ぎます。

検索では (subject, attribute) を指定したうえで新しい履歴から確認するため、同じ並びのインデックスを作ります。

新しい値は旧値を削除せず有効期間を閉じる

次に、値を追加する remember() を実装します。

memory_history.py
def remember(
    con: sqlite3.Connection,
    subject: str,
    attribute: str,
    value: str,
    valid_from: int,
) -> None:
    con.execute("BEGIN IMMEDIATE")
    try:
        latest = con.execute(
            """
            SELECT id, valid_from, valid_to
            FROM memory_history
            WHERE subject = ? AND attribute = ?
            ORDER BY valid_from DESC
            LIMIT 1
            """,
            (subject, attribute),
        ).fetchone()

        if latest is not None:
            latest_id, latest_from, latest_to = latest

            if valid_from <= latest_from:
                raise ValueError(
                    "valid_from must be newer than the latest history"
                )

            if latest_to is not None and valid_from < latest_to:
                raise ValueError(
                    "valid_from overlaps the latest history"
                )

            if latest_to is None:
                con.execute(
                    """
                    UPDATE memory_history
                    SET valid_to = ?
                    WHERE id = ?
                    """,
                    (valid_from, latest_id),
                )

        con.execute(
            """
            INSERT INTO memory_history(
                subject,
                attribute,
                value,
                valid_from,
                valid_to
            )
            VALUES (?, ?, ?, ?, NULL)
            """,
            (subject, attribute, value, valid_from),
        )
        con.commit()
    except Exception:
        con.rollback()
        raise

この実装では、同じ (subject, attribute) について履歴を時系列順に追加します。

たとえば Company A が有効な状態で Company B を追加すると、Company A の行は削除されず、その valid_to が Company B の valid_from で閉じられます。

Company A | valid_from=2026-04-01 | valid_to=2026-09-01
Company B | valid_from=2026-09-01 | valid_to=NULL

旧値の終了と新値の追加は一つのトランザクションで処理します。SQLite では読み書きはトランザクション内で行われ、BEGIN IMMEDIATE は書き込みトランザクションを明示的に開始します[6]。

本稿の実装は履歴の後方追加を対象とするため、既存の最新履歴より古い valid_from は ValueError にします。過去へ遡って履歴を訂正する処理を追加する場合は、前後の区間を再計算する別の更新規則が必要です。

指定時点の値を as_of() で取得する

履歴から状態を再構成する中心処理が as_of() です。

memory_history.py
def as_of(
    con: sqlite3.Connection,
    subject: str,
    attribute: str,
    at: int,
) -> str | None:
    row = con.execute(
        """
        SELECT value
        FROM memory_history
        WHERE subject = ?
          AND attribute = ?
          AND valid_from <= ?
          AND (valid_to IS NULL OR ? < valid_to)
        ORDER BY valid_from DESC
        LIMIT 1
        """,
        (subject, attribute, at, at),
    ).fetchone()

    return None if row is None else row[0]

選択条件は次です。

valid_from <= at < valid_to

valid_to IS NULL の行は、終了時刻がまだ設定されていない履歴として扱います。

たとえば次の履歴がある場合、

Company A  [2026-04-01, 2026-09-01)
Company B  [2026-09-01, ∞)

2026 年 6 月を指定すると Company A、2026 年 9 月 1 日以後を指定すると Company B が返ります。

検索時に「最新の行を返す」のではなく、「指定時点で有効な行を返す」点が重要です。将来の日付で有効になる値を先に登録しても、切替日前には現在の値を取得できます。

current() は as_of() の特殊形にする

現在値専用の判定規則は増やさず、current() を as_of() の特殊形として実装します。

memory_history.py
def current(
    con: sqlite3.Connection,
    subject: str,
    attribute: str,
    now: int | None = None,
) -> str | None:
    return as_of(
        con,
        subject,
        attribute,
        int(time()) if now is None else now,
    )

通常は現在時刻を使います。テストでは now を明示して、結果を時刻に依存せず再現できるようにします。

これで現在状態、過去状態、将来予定を同じ判定規則で扱えます。

撤回は履歴を削除せず valid_to を設定する

値を別の値へ置換するだけでなく、ある時点から値そのものが無効になる場合もあります。

たとえば次の履歴です。

2026-09-01  allow_external_upload = true
2026-09-15  revoke

この場合は行を削除せず、valid_to を 2026-09-15 に設定します。

memory_history.py
def revoke(
    con: sqlite3.Connection,
    subject: str,
    attribute: str,
    at: int,
) -> None:
    con.execute("BEGIN IMMEDIATE")
    try:
        latest = con.execute(
            """
            SELECT id, valid_from, valid_to
            FROM memory_history
            WHERE subject = ? AND attribute = ?
            ORDER BY valid_from DESC
            LIMIT 1
            """,
            (subject, attribute),
        ).fetchone()

        if latest is None:
            raise KeyError((subject, attribute))

        latest_id, latest_from, latest_to = latest

        if latest_to is not None:
            raise ValueError("latest history is already closed")

        if at <= latest_from:
            raise ValueError(
                "revoke time must be newer than valid_from"
            )

        con.execute(
            """
            UPDATE memory_history
            SET valid_to = ?
            WHERE id = ?
            """,
            (at, latest_id),
        )
        con.commit()
    except Exception:
        con.rollback()
        raise

この状態では 2026 年 9 月 10 日を指定すると true が返り、9 月 20 日を指定すると値なしになります。

履歴行は残るため、「過去には有効だった」という事実も取得できます。

4 ケースを pytest で確認する

テストでは時刻を固定するため、ISO 形式の文字列を UTC の Unix time へ変換します。

test_memory_history.py
from datetime import datetime, timezone

from memory_history import as_of, connect, current, remember, revoke


def ts(value: str) -> int:
    return int(
        datetime.fromisoformat(value)
        .replace(tzinfo=timezone.utc)
        .timestamp()
    )

過去値と現在値を分けて取得する

test_memory_history.py
def test_reconstructs_past_and_current_value() -> None:
    con = connect()

    remember(
        con,
        "user",
        "employer",
        "Company A",
        ts("2026-04-01T00:00:00"),
    )
    remember(
        con,
        "user",
        "employer",
        "Company B",
        ts("2026-09-01T00:00:00"),
    )

    assert (
        as_of(
            con,
            "user",
            "employer",
            ts("2026-06-01T00:00:00"),
        )
        == "Company A"
    )

    assert (
        as_of(
            con,
            "user",
            "employer",
            ts("2026-09-01T00:00:00"),
        )
        == "Company B"
    )

同じ履歴表から問い合わせ時点によって別の値を取得できれば、過去値を保存したまま現在値を選択できます。

撤回後は値なし、撤回前は旧値を返す

test_memory_history.py
def test_revoke_closes_validity_without_deleting_history() -> None:
    con = connect()

    remember(
        con,
        "user",
        "allow_external_upload",
        "true",
        ts("2026-09-01T00:00:00"),
    )
    revoke(
        con,
        "user",
        "allow_external_upload",
        ts("2026-09-15T00:00:00"),
    )

    assert (
        as_of(
            con,
            "user",
            "allow_external_upload",
            ts("2026-09-10T00:00:00"),
        )
        == "true"
    )

    assert (
        as_of(
            con,
            "user",
            "allow_external_upload",
            ts("2026-09-20T00:00:00"),
        )
        is None
    )

    rows = con.execute(
        """
        SELECT value, valid_from, valid_to
        FROM memory_history
        WHERE subject = ? AND attribute = ?
        """,
        ("user", "allow_external_upload"),
    ).fetchall()

    assert len(rows) == 1
    assert rows[0][2] == ts("2026-09-15T00:00:00")

最後の 2 行で、撤回後も履歴自体が残っていることを確認します。

将来予定は開始時刻から現在値になる

test_memory_history.py
def test_future_value_becomes_current_at_valid_from() -> None:
    con = connect()

    remember(
        con,
        "project",
        "release_status",
        "stable",
        ts("2026-09-01T00:00:00"),
    )
    remember(
        con,
        "project",
        "release_status",
        "maintenance",
        ts("2026-10-01T00:00:00"),
    )

    assert (
        current(
            con,
            "project",
            "release_status",
            now=ts("2026-09-30T23:59:59"),
        )
        == "stable"
    )

    assert (
        current(
            con,
            "project",
            "release_status",
            now=ts("2026-10-01T00:00:00"),
        )
        == "maintenance"
    )

将来値を先に登録しても、current() は現在時刻が属する区間だけを選びます。

時系列に逆行する追加を拒否する

test_memory_history.py
def test_rejects_out_of_order_history() -> None:
    con = connect()

    remember(
        con,
        "user",
        "employer",
        "Company B",
        ts("2026-09-01T00:00:00"),
    )

    try:
        remember(
            con,
            "user",
            "employer",
            "Company A",
            ts("2026-04-01T00:00:00"),
        )
    except ValueError:
        pass
    else:
        raise AssertionError(
            "out-of-order history must be rejected"
        )

この制約によって、単純な後方追加モデルのまま有効期間の重複を防ぎます。

テストを実行する

2 ファイルを保存したら、次のコマンドを実行します。

python -m pytest -q

結果は次のようになります。

....                                                                     [100%]
4 passed

4 ケースが通れば、少なくとも次の性質を確認できます。

確認対象 確認できること
値の置換 古い値を残したまま新しい値へ切り替えられる
過去時点の問い合わせ 指定時点で有効だった値を取得できる
撤回 履歴を残したまま、ある時点以後を値なしにできる
将来予定 開始時刻へ到達した時点で新しい値を現在値として選べる
追加順序 単純な履歴モデルで扱えない逆順追加を検出できる

LLM へ渡す前に現在状態を確定する

この実装で扱っているのは、長期記憶システム全体のうち次の部分です。

保存済みの構造化履歴
        |
        v
指定時点で有効な状態を選択
        |
        v
LLM へ渡す状態

自然言語から subject、attribute、value を抽出する処理や、意味的類似度による検索は別の層になります。本稿の実装では、保存済みの構造化履歴から時間条件だけで状態を確定します。

この分離には実務上の利点があります。LLM 自身へ「A と B のどちらが現在の値か」を毎回推論させる代わりに、データベース側で valid_from <= at < valid_to を満たす値を一意に選べます。LLM へ渡す前に現在状態を確定できるため、履歴の保存と回答時の利用を別々に管理できます。

一方、同じ時点について複数の値が競合する場合、過去へ遡って事実を訂正する場合、複数の書き込み主体が同時更新する場合には、追加の競合解決規則が必要です。ここまで扱うなら、単純な 1 本の有効期間だけでなく、事実の確度、取得元、記録時刻、訂正関係などもデータモデルへ追加する必要があります。

今回の実装から得られる実務上の帰結は明確です。長期記憶へ履歴を保存することと、その履歴から現在の回答に使う状態を選ぶことは別の処理です。値に有効期間を持たせ、問い合わせ時点で状態を再構成すれば、過去を保持したまま現在値を一意に選択できます。

参考文献

  1. id774, AI に仕事を任せるには、モデルの外側を設計する(2026-08-21). https://blog.id774.net/entry/2026/08/21/5544/
  2. Yuhang Li, Yuchen Li, What Should an Agent Forget? Separating What Is Stored from What Is Used(2026-09-09). https://arxiv.org/abs/2609.10263
  3. Ansuman Mullick, Eray Tüzün, Fortunate Recall: Ontology-Driven Memory Lifecycle Management for Persistent Coherence in LLMs(2026-09-09). https://arxiv.org/abs/2609.10413
  4. Miao Su, Yucan Guo, Zhongni Hou, Long Bai, Zixuan Li, Yufei Zhang, Guojun Yin, Wei Lin, Xiaolong Jin, Jiafeng Guo, Xueqi Cheng, Beyond Dialogue Time: Temporal Semantic Memory for Personalized LLM Agents(2026-01-12). https://arxiv.org/abs/2601.07468
  5. Python Software Foundation, sqlite3 — DB-API 2.0 interface for SQLite databases. https://docs.python.org/3/library/sqlite3.html
  6. SQLite, Transaction. https://www.sqlite.org/lang_transaction.html
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?