Kerwin AI Information

為 AI Agent 建長期記憶：Rust 實戰

為 AI Agent 建長期記憶：Rust 實戰

作者:

kerwin77106

分類:

📌 本文重點

  • 單純丟向量庫不足以支撐可運維的長期記憶
  • 將 Events / Sessions / Facts 結構化並分層檢索
  • 以 Rust 記憶服務提供 vendor-agnostic 的統一記憶層

AI Agent 開到一定規模後，「把聊天記錄丟進向量資料庫」很快就不夠用：

  • 記憶膨脹導致成本失控、查詢變慢
  • 多個 Agent、不同 LLM 共用資料時互相污染
  • 想換模型或供應商時，歷史記憶幾乎不能重用

這篇從記憶層設計問題拆解起，用 Rust 開源專案 akitaonrails/ai-memory 當主線，示範如何把「長期記憶」升級成可運維的基礎設施，而不是一堆臨時向量。

重點說明：把「記憶」變成明確的基礎設施

💡 關鍵： 把記憶從「單一向量庫」拆成 Events / Sessions / Facts，可以同時兼顧語義檢索與結構化查詢，讓長期記憶真正可維護、可重用。

1. 資料結構：從 chat log 到 events / sessions / facts

粗糙作法：

  • 每輪對話做 embedding → 丟進向量庫 → 用 semantic search 回撈

問題：

  • 事件順序感消失：只剩相似度，沒有「先後」與「上下文」
  • 無法表達持久事實：像使用者偏好、系統狀態，只是散落在多個對話片段

較好的設計是拆成三類：

  • Events：原始互動紀錄
  • 例：user_message, tool_call, agent_decision
  • 保留時間戳與來源，適合做 audit 和重播
  • Sessions：一次任務或一段對話的邊界
  • 例：session_id、agent_id、status=completed
  • 讓你知道「這次訂房流程」完整發生了什麼
  • Facts：可重用、可更新的持久知識
  • 例：使用者偏好、系統配置、業務規則
  • 需要版本與狀態（active, deprecated）

這樣一來，semantic search 用在 Events/Facts 的內容檢索，structured query 用在 Sessions/Facts 的條件篩選，組出可靠的上下文再餵給 LLM，而不是全靠相似度。

2. 檢索策略與 vendor-agnostic 介面

實際上你會需要兩層 API：

  • 語義檢索層：
  • 例如：search_events(query, top_k)、search_facts(query, filters)

  • 背後可接不同 embedding provider（OpenAI, local model 等），但對上層 Agent 暴露的是穩定的介面

  • 結構化查詢層：

  • 例如：get_session(session_id)、list_facts(owner_id, kind="preference")
  • 通常走資料庫索引（Postgres, SQLite），不牽涉向量搜尋

akitaonrails/ai-memory 正是要提供這種「供應商無關的記憶層」，讓你可以：

  • 今天用 OpenAI，明天換到本地模型
  • 多個 Agent framework（LangChain, LlamaIndex, 自家的 SDK）共用同一套記憶服務

3. Rust 記憶服務 + RAG / 工具調用整合

長期記憶一旦變成基礎設施，就有三個技術要求：

  • 效能：高頻讀寫、多 Agent 並行
  • 安全：記憶裡必然有 PII 和敏感業務資訊
  • 跨語言可接：Python、Node、Go、甚至 CLI agent 都要用

Rust 在這裡的角色：

  • 提供一個高效、型別安全的記憶核心（ai-memory）
  • 對外以 FFI / HTTP / IPC 三種方式暴露 API
  • Agent 框架只需要呼叫類似 store_event / query_memory 的介面，就能把記憶接進 RAG pipeline 或工具調用流程

實作範例：用 ai-memory 建一個共用長期記憶服務

以下以虛構的 ai-memory 介面示意，重點放在設計思路而不是精確函式名稱。

1）定義記憶 schema，接上現有 Agent

先在 Rust 端定義核心結構：

// memory_schema.rs

#[derive(Debug, Clone)]
pub enum EventKind {
    UserMessage,
    AgentReply,
    ToolCall,
    ToolResult,
}

#[derive(Debug, Clone)]
pub struct Event {
    pub id: String,
    pub session_id: String,
    pub agent_id: String,
    pub kind: EventKind,
    pub content: String,
    pub metadata: serde_json::Value,
    pub created_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Debug, Clone)]
pub struct Fact {
    pub id: String,
    pub owner_id: String,    // user_id 或 system
    pub kind: String,        // preference, policy, profile
    pub content: String,
    pub embedding: Option<Vec<f32>>,  // for semantic search
    pub version: i32,
    pub status: String,      // active, deprecated
}

對 Python Agent 來說，只需要一個薄封裝，把每輪互動寫入記憶：

# agent_memory.py

class MemoryClient:
    def __init__(self, base_url: str):
        self.base_url = base_url

    def store_event(self, session_id, agent_id, kind, content, metadata=None):
        payload = {
            "session_id": session_id,
            "agent_id": agent_id,
            "kind": kind,
            "content": content,
            "metadata": metadata or {},
        }
        requests.post(f"{self.base_url}/events", json=payload)

    def search_facts(self, query, owner_id=None, kind=None, top_k=5):
        params = {
            "query": query,
            "owner_id": owner_id,
            "kind": kind,
            "top_k": top_k,
        }
        resp = requests.get(f"{self.base_url}/facts/search", params=params)
        return resp.json()["results"]

Agent loop 中的使用方式：

memory = MemoryClient(base_url="http://localhost:8080")

# 每輪對話寫入 event
memory.store_event(
    session_id=session_id,
    agent_id="support-bot-v2",
    kind="UserMessage",
    content=user_input,
)

# 在回應前查詢長期偏好
facts = memory.search_facts(
    query="使用者的語言偏好與通知設定",
    owner_id=user_id,
    kind="preference",
    top_k=3,
)

context = format_facts_for_prompt(facts)
response = llm.chat(prompt=build_prompt(user_input, context))

這樣你的 Agent 邏輯完全不需要知道底層用的是哪家向量資料庫，也不綁在單一 LLM provider。

2）Rust 記憶服務：FFI / HTTP / IPC 三種接法

假設 ai-memory 提供核心 crate ai_memory_core，我們可以用三種方式包出去。

HTTP 服務（最通用）

優點：任何語言都能接；缺點：有網路 overhead。

// http_server.rs

use ai_memory_core::{MemoryStore, SearchQuery};
use axum::{routing::get, routing::post, Json, Router};

async fn create_event(Json(payload): Json<CreateEventRequest>) -> Json<EventResponse> {
    let mut store = MemoryStore::global();
    let event = store.store_event(payload.into())?;
    Json(EventResponse::from(event))
}

async fn search_facts(Json(query): Json<SearchFactsRequest>) -> Json<SearchFactsResponse> {
    let store = MemoryStore::global();
    let results = store.search_facts(SearchQuery::from(query))?;
    Json(SearchFactsResponse { results })
}

pub fn app() -> Router {
    Router::new()
        .route("/events", post(create_event))
        .route("/facts/search", get(search_facts))
}

命令列 Agent 或後端服務只要跑一個共用 memory server，就可以用 HTTP 存取。

FFI（嵌入到 Python / Node 進程）

優點：效能好、延遲低；缺點：需維護 binding。適用高頻工具調用型 Agent。

// lib.rs (Rust)

#[no_mangle]
pub extern "C" fn store_event_ffi(json_payload: *const c_char) -> *const c_char {
    // 解析 JSON，呼叫 MemoryStore，回傳 JSON 字串
}

Python 側使用 ctypes 或 pyo3 包一層，暴露同樣的 store_event / search_facts 介面，對應前面的 MemoryClient。

IPC（同機不同進程，高安全場景）

可以用 Unix socket + protobuf 或 Cap’n Proto：

  • 優點：比 HTTP 更輕量，適合同機多服務共用記憶
  • 缺點：部署稍複雜，需額外 tooling

設計上只要確保三種接法都共用同一套核心 API（MemoryStore），就能在不同專案中任意選擇實作方式，而不改動 Agent 邏輯。

3）版本升級、資料遷移、多模型共用記憶庫

長期記憶真正困難在於 「持續演化」。

Schema 版本管理

在 Fact 結構上掛版本欄位：

pub struct Fact {
    pub id: String,
    pub owner_id: String,
    pub kind: String,
    pub content: String,
    pub version: i32,       // schema/version
    pub status: String,
    pub embedding: Option<Vec<f32>>,  
}

當你需要新增欄位或改變結構時：

  • 新寫入用 version = 2
  • 舊資料由 migration job 緩慢升級
  • 查詢 API 接受 min_version / max_version 作為過渡策略

多模型共用同一記憶庫

不同 LLM 對同一條記錄的解讀可能不同，所以要在 metadata 明確標注來源模型：

pub struct Event {
    pub model_name: Option<String>,   // gpt-4o, llama-3-70b 等
    pub agent_id: String,
    // ...
}

策略上：

  • Facts 儘量由工具或人類決策產生，減少「模型幻覺」寫入持久記憶
  • 檢索時可加 filter：model_name in ["gpt-4o", "internal-rule-engine"]，避免用某些品質較差模型產生的事件來推論

建議與注意事項：把坑提前填好

💡 關鍵： 控制 embedding 範圍、處理 PII、標記 embedding 模型，是讓長期記憶在成本、隱私與準確度間取得平衡的三個關鍵。

1. 記憶膨脹與成本控制

問題：

  • 所有對話都 embedding → 向量庫數量爆炸，成本跟查詢延遲一起上升

建議：

  • 只對重要 Events / Facts 做 embedding，例如：完成一個任務時產生 summary fact
  • 針對長期 session，定期做 conversation summarization，保留摘要而不是 raw log
  • 對向量庫設計 TTL 或冷/熱層級：冷資料只保留摘要向量

2. 隱私與合規（PII / 敏感資料）

問題：

  • 長期記憶通常含姓名、電話、訂單資訊等 PII

建議：

  • 設計 PII-aware schema：把 PII 拆出獨立欄位，便於加密與 masking
  • 在寫入前跑簡單的 PII 檢測 rule：
  • 例：電話號碼、email pattern 直接用工具抽出 → 存入安全欄位
  • 對查詢 API 加上角色權限（例如 owner_id＋role=internal_support）

3. 語義檢索漂移與模型更新

問題：

  • 換 embedding 模型後，舊向量的語義分佈不同，semantic search 精度變差

簡單防線：

  • 在向量旁存 embedding_model 欄位
  • 新模型上線後：
  • 新增資料用新 embedding
  • 舊資料分批 re-embed，或只重算活躍 Facts
  • 用 offline evaluation：定義一組標準 query + expected hits，監控 semantic search 成效

4. 不同模型對同一紀錄的解讀不一致

問題：

  • Model A 把某次對話解讀為「使用者喜歡簡訊通知」，Model B 覺得是「偏好 email」

建議：

  • 把「推論」與「事實」分開存：
  • Fact：明確的、可驗證的偏好（使用者自行設定）
  • Inference：模型的猜測，需經多次交叉驗證或人工確認才升級為 Fact
  • 在 prompt 中區分：
  • 「已確認偏好」 vs. 「推測偏好」

結語：把長期記憶升級成基礎設施

對成熟的 AI 專案來說，長期記憶不再是「多塞幾個向量」的 hack，而是需要設計、版本與治理的系統。利用像 akitaonrails/ai-memory 這種以 Rust 實作的開源方案，你可以：

  • 為所有 Agent 建立一個 供應商無關、可演化的記憶層
  • 清楚區分 Events / Sessions / Facts，同時用語義與結構化查詢做精準檢索
  • 以 HTTP / FFI / IPC 等方式，在不同語言與框架中重用同一記憶庫

這些設計一開始多花一點功夫，換到的是更穩定的成本、更易維護的架構，以及在多 Agent、多模型環境裡，真正能「記住」使用者與業務脈絡的系統。

🚀 你現在可以做的事

  • 到 GitHub 搜尋並閱讀 akitaonrails/ai-memory 專案原始碼與文件
  • 在現有 Agent 專案中先導入 Events / Sessions / Facts 基本 schema，重構記憶寫入流程
  • 實作一個簡單的 HTTP memory server，讓至少兩個不同語言或框架的 Agent 共用同一記憶層

留言

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *

更多文章