class\LLM-Memory-Learn\W3-VectorMemory\lecture.md

本文中將會用本地 Embedding + FAISS 做語意檢索記憶，最後整合三週積木，做出混合記憶助理 memchat

準備工作

conda activate mem
pip install sentence-transformers faiss-cpu numpy

40961-ebflhpga57.png

本地向量庫

用 sentence-transformers 做 Embedding，FAISS 建索引，實作相似度檢索

EmbeddingStore

實現邏輯如下：

import numpy as np

_model = None
def _get_model():
    """延遲載入 embedding 模型（第一次呼叫才載入）。"""
    global _model
    if _model is None:
        from sentence_transformers import SentenceTransformer
        _model = SentenceTransformer("all-MiniLM-L6-v2")
    return _model

def embed(texts):
    """把字串（或字串陣列）轉成正規化向量（float32）。"""
    if isinstance(texts, str):
        texts = [texts]
    vecs = _get_model().encode(texts, normalize_embeddings=True)
    return np.asarray(vecs, dtype="float32")

class EmbeddingStore:
    """存 (原文, 向量)，支援 Top-K 餘弦相似度檢索（內積索引，向量已正規化）。"""

    def __init__(self, dim=384):
        import faiss
        self.dim = dim
        self.index = faiss.IndexFlatIP(dim)   # 內積 = 正規化向量的餘弦相似度
        self.texts = []                        # 與索引平行的原文對照表

    def add(self, text):
        vec = embed(text)
        self.index.add(vec)
        self.texts.append(text)

    def search(self, query, k=3):
        """回傳 [(text, score), ...]，score 為餘弦相似度。"""
        if self.index.ntotal == 0:
            return []
        vec = embed(query)
        k = min(k, self.index.ntotal)
        scores, idxs = self.index.search(vec, k)
        return [(self.texts[i], float(s)) for s, i in zip(scores[0], idxs[0])]

    def __len__(self):
        return self.index.ntotal

驗證語意檢索

執行測試
python demos/w3_vector_demo.py

import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from memory.vector import EmbeddingStore

store = EmbeddingStore()
facts = [
    "我對甲殼類過敏，不能吃蝦蟹。",
    "我最喜歡的顏色是藍色。",
    "我在學做義大利麵。",
    "我住在京都車站附近的旅館。",
    "我的每餐預算是一千日圓。",
    "我養了一隻叫咪咪的貓。",
]
for f in facts:
    store.add(f)

for q in ["我不能吃什麼海鮮？", "我家有寵物嗎？", "我住哪裡？"]:
    print(f"\n 查詢：{q}")
    for text, score in store.search(q, k=2):
        print(f"   [{score:.3f}] {text}")

執行結果：

(py3.10) J:\git\mathTalk\class\LLM-Memory-Learn\src>python demos/w3_vector_demo.py
Skipping import of cpp extensions due to incompatible torch version. Please upgrade to torch >= 2.11.0 (found 2.10.0+cu126).
W0816 22:06:45.295000 30816 site-packages\torch\distributed\elastic\multiprocessing\redirects.py:29] NOTE: Redirects are currently not supported in Windows or MacOs.
Loading weights: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 103/103 [00:00<00:00, 5636.40it/s]
BertModel LOAD REPORT from: sentence-transformers/all-MiniLM-L6-v2
Key                     | Status     | Details
------------------------+------------+--------
embeddings.position_ids | UNEXPECTED |

Notes:
- UNEXPECTED:   can be ignored when loading from different task/architecture; not ok if you expect identical arch.

 查詢：我不能吃什麼海鮮？
   [0.732] 我對甲殼類過敏，不能吃蝦蟹。
   [0.704] 我養了一隻叫咪咪的貓。

 查詢：我家有寵物嗎？
   [0.630] 我最喜歡的顏色是藍色。
   [0.612] 我養了一隻叫咪咪的貓。

 查詢：我住哪裡？
   [0.799] 我最喜歡的顏色是藍色。
   [0.775] 我對甲殼類過敏，不能吃蝦蟹。

跑出來的結果很奇怪，查了下發現 all-MiniLM-L6-v2 是英文模型，需要替換成中文，不然 token 計算那邊都是錯誤的，難怪搜索出來的結果是錯的，修正如下：

_model = SentenceTransformer(
    "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

向量記憶

把每輪對話存進向量庫，新問題先檢索 Top-K 相關歷史，再拼進 Prompt

VectorMemory

實現 VectorMemory 如下，將對話内容存入向量庫，主要目的就是在回答前可以進行檢索

class VectorMemory:
    """每輪對話存入向量庫；回答前檢索 Top-K 相關歷史注入 system。"""
    def __init__(self, system=None, k=3, min_score=0.25):
        self.system = system
        self.k = k
        self.min_score = min_score       # 相似度門檻，濾掉不相關雜訊
        self.store = EmbeddingStore()
        self._messages = []
        self._pending_user = None

    def add_user(self, text):
        self._messages.append({"role": "user", "content": text})
        self._pending_user = text

    def add_assistant(self, text):
        self._messages.append({"role": "assistant", "content": text})
        # 把剛完成的一輪（Q+A）存進向量庫，作為長期記憶
        if self._pending_user is not None:
            self.store.add(f"使用者: {self._pending_user}\n助理: {text}")
            self._pending_user = None

    def retrieve(self, query):
        hits = self.store.search(query, k=self.k)
        return [t for t, s in hits if s >= self.min_score]

    def build_system(self, query):
        base = self.system or ""
        hits = self.retrieve(query)
        if hits:
            joined = "\n---\n".join(hits)
            return f"{base}\n\n【檢索到的相關歷史記憶】\n{joined}"
        return base

    def build_messages(self, query):
        """向量記憶只帶『當前問題』；歷史靠檢索注入 system。"""
        return [{"role": "user", "content": query}]

VectorMemory 的 messages 只放當前問題，所有歷史都靠檢索注入 system。在實務落地項目中會保留少量近期原文。

驗證跨時間召回

在這需要驗證，先鋪陳一堆事實對話、中間插大量無關閒聊，最後問一個久遠的事實，觀察向量記憶能否召回。驗證程式如下：


from memory.llm import chat
from memory.vector import VectorMemory

mem = VectorMemory(system="你是助理，用繁體中文簡短回答。", k=3)
script = [
    "我對甲殼類過敏。",
    "今天天氣不錯。", "你喜歡貓還是狗？", "推薦一部電影。",   # 無關閒聊
    "再聊聊音樂。", "週末有什麼安排？",
    "我剛剛說我對什麼過敏？",       # 久遠事實，靠檢索召回
]
for user in script:
    mem.add_user(user)
    reply, _ = chat(mem.build_messages(user), system=mem.build_system(user))
    mem.add_assistant(reply)
    print(f"你 > {user}\nAI > {reply}\n")

執行程式
python demos\w3_vector_probe.py

結果如下：


 等待 21 秒，避免超過 API RPM 限制…
你 > 今天天氣不錯。
AI > 謝謝分享，希望你有個愉快的一天！

 等待 21 秒，避免超過 API RPM 限制…
你 > 你喜歡貓還是狗？
AI > 作為AI，我沒有情感，所以我不會喜歡貓或狗。我旨在協助解答問題。

 等待 21 秒，避免超過 API RPM 限制…
你 > 推薦一部電影。
AI > 推薦電影《肖申克的救贖》（The Shawshank Redemption），這是一部深受觀眾喜愛的經典作品，講述了希望、友誼和自由的故事。

 等待 21 秒，避免超過 API RPM 限制…
你 > 再聊聊音樂。
AI > 當然可以。音樂是一種藝術形式，它通過組織聲音來表達情感、講述故事或創造氛圍。人們透過演奏、聆聽和創作音樂來享受和表達自己。音樂有著豐富的歷史和文化背景，風格多樣，從古典音樂到流行音樂，從爵士樂到電子音樂，各種風格皆有其獨特魅力。你可以告訴我你喜歡哪種風格的音樂，或者想要了解有關音樂的哪些方面。

 等待 21 秒，避免超過 API RPM 限制…
你 > 週末有什麼安排？
AI > 週末的安排有很多選擇，可以根據你個人的興趣和喜好來決定，例如郊遊、運動、會友、看電影、閱讀或學習新技能等。你打算怎麼安排呢？

 等待 21 秒，避免超過 API RPM 限制…
你 > 我剛剛說我對什麼過敏？
AI > 你剛剛說你對甲殼類過敏。

無標籤

關注作者：

新增評論