「沒有對話記憶的 AI，只是單次問答機器；具備長效記憶與自動摘要能力的 Agent，才能在漫長對話中持續保持脈絡一致。」

在 Day 05 中，我們介紹了 RAG 知識檢索引擎（rag_engine.py）如何提供外部金融知識。今天我們將深入探討 Angelina AI Agent 的長期記憶樞紐——app/services/conversation_memory.py。

為了在 Self-hosted 環境下達成 $0 元維運 與 零外部服務依賴，我們採用輕量級的 SQLite 搭配 aiosqlite 實現全非同步持久化儲存，並設計了 Auto-Summarization，確保對話在無限延伸時不會超出 LLM 的 上下文視窗。

本篇重點摘要

1. 為什麼選用 SQLite + aiosqlite 作為 Agent 的持久化記憶庫呢？
2. 拆解 conversations 資料表 DDL 與高效率複合索引設定 (session_id, created_at)。
3. 實作非同步對話讀寫 (save_turn, load_context) 與嚴格 10,000 字元長度限制。
4. 解析 Auto-Summarization 機制：當輪數過百時，如何將舊對話原子化壓縮（Atomics）為摘要。

一、記憶架構設計與 SQLite Schema 規劃

在 ConversationMemory 中，所有對話記錄均儲存於本機資料庫 data/conversations.db 中。為了支援高併發非同步存取並避免阻塞 FastAPI 的 Event Loop，我們全程採用 aiosqlite 進行操作。

資料庫 DDL 與索引設計

  為了確保在累積數萬筆對話時依然能在 500 ms 以內 完成 context 載入，我們建立了複合索引 idx_session_created：

SQL
-- conversations 資料表 schema
CREATE TABLE IF NOT EXISTS conversations (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id  TEXT    NOT NULL,
    role        TEXT    NOT NULL CHECK(role IN ('user', 'assistant', 'summary')),
    content     TEXT    NOT NULL CHECK(length(content) <= 10000),
    created_at  TEXT    NOT NULL DEFAULT (datetime('now', 'utc')),
    is_summary  INTEGER NOT NULL DEFAULT 0
);

-- 高效率 Session + 時間倒序複合索引
CREATE INDEX IF NOT EXISTS idx_session_created
    ON conversations(session_id, created_at DESC);

二、app/services/conversation_memory.py 核心邏輯解析

1. 歷史對話載入與排序邏輯
   在呼叫 Gemini 進行推理前，系統需載入該 Session 最近的 20 筆對話。資料庫查詢時以時間 倒序（DESC） 抓取最新記錄以發揮索引最大效能，隨後在記憶體中 Reversed，恢復成符合人類對話習慣的時間順序：

Python
async def load_context(self, session_id: str, limit: int = 20) -> list[Turn]:
    """讀取最近 limit 筆對話記錄，並按時間正序排序傳回"""
    async with aiosqlite.connect(self._db_path) as db:
        db.row_factory = aiosqlite.Row
        async with db.execute(
            """
            SELECT id, session_id, role, content, created_at, is_summary
            FROM   conversations
            WHERE  session_id = ?
            ORDER  BY created_at DESC, id DESC
            LIMIT  ?
            """,
            (session_id, limit),
        ) as cursor:
            rows = await cursor.fetchall()

    # 資料庫取回最新 20 筆 (DESC)；反轉為正序 (ASC) 供 LLM 理解上下文
    turns: list[Turn] = []
    for row in reversed(rows):
        turns.append(
            Turn(
                id=row["id"],
                session_id=row["session_id"],
                role=row["role"],
                content=row["content"],
                created_at=datetime.fromisoformat(row["created_at"]),
                is_summary=bool(row["is_summary"]),
            )
        )
    return turns

2. 對話自動摘要與歷史壓縮
   當單一 Session 的歷史輪數累積達到 101 輪以上 時，若直接將所有對話傳給 LLM，不僅會拉高 API 延遲，還可能超出 Token 限制。

   我的想法是：保留最新的 20 輪精確對話，將其餘舊對話壓縮成 1 筆 summary 輪次，使總輪數精準控制在 21 輪（1 筆 Summary + 20 筆最新對話）。

   這整個過程在 SQLite 中以 Transaction 原子化操作 進行（先刪除舊對話，再寫入摘要）：

Python
async def summarize_if_needed(self, session_id: str) -> None:
    """當對話記錄 >= 101 輪時，將舊對話壓縮為單一 summary 記錄"""
    count = await self.get_turn_count(session_id=session_id)
    if count < 101:
        return

    # 計算需被壓縮的舊對話筆數
    turns_to_summarize = count - 20

    async with aiosqlite.connect(self._db_path) as db:
        db.row_factory = aiosqlite.Row
        async with db.execute(
            """
            SELECT id, role, content, created_at
            FROM   conversations
            WHERE  session_id = ?
            ORDER  BY created_at ASC, id ASC
            LIMIT  ?
            """,
            (session_id, turns_to_summarize),
        ) as cursor:
            oldest_rows = await cursor.fetchall()

        if not oldest_rows:
            return

        oldest_ids = [row["id"] for row in oldest_rows]
        
        # 拼接舊對話內容並截斷至 10,000 字元限制內
        summary_parts = [f"[{row['role']}] {row['content']}" for row in oldest_rows]
        summary_text = ("\n".join(summary_parts))[:10_000]
        created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        placeholders = ",".join("?" * len(oldest_ids))
        try:
            # 原子化事務：刪除被壓縮的舊對話，插入 summary 記錄
            await db.execute(f"DELETE FROM conversations WHERE id IN ({placeholders})", oldest_ids)
            await db.execute(
                """
                INSERT INTO conversations (session_id, role, content, created_at, is_summary)
                VALUES (?, 'summary', ?, ?, 1)
                """,
                (session_id, summary_text, created_at),
            )
            await db.commit()
            logger.info("Summarised %d turns for session %s", len(oldest_ids), session_id)
        except aiosqlite.Error as exc:
            await db.rollback()
            logger.error("Failed to summarise conversation for session %s: %s", session_id, exc)
            raise

三、今日總結與架構優勢

透過 conversation_memory.py 的實作，我們為 Angelina Agent 打造了兼具穩定與效能的記憶模組：

  1. 高效能非同步：aiosqlite 配合 SQLite 複合索引，保證多使用者併發存取時非阻塞、高響應。
  2. Context Window 保護機制：透過 summarize_if_needed 自動壓縮機制，徹底解決長對話造成的 Token 爆表與 API 成本上升問題。
  3. 資料持久化與容錯：所有對話落盤至本機 data/conversations.db，即使 VM 重啟或容器更新，對話記憶也不會遺失。

明天（Day 07）我們將進入 app/services/learning_module.py，探討 Agent 如何透過背景任務，在對話結束後非同步自動提煉金融知識點並寫回向量資料庫，達成真正的「越用越聰明」！

明日預告：【Day 07】非同步背景學習：LearningModule 自動提煉金融知識與向量庫去重