昨天我們使用 Structured Output，把抽取結果整理成：

MemoryCandidate(
    content="使用者想加強旅遊英文。"
)

現在每一筆 Memory Candidate 已經有穩定的資料結構，但如果未來累積了幾百筆 Memory，新的問題就會出現：

當使用者提出新問題時，程式要怎麼知道哪一筆 Memory 和它最相關？

假設 Memory 裡保存的是：

使用者想加強旅遊英文。

之後使用者問：

可以陪我練習在機場辦理登機的對話嗎？

如果只使用關鍵字搜尋，兩段文字甚至沒有完全相同的詞：

旅遊英文

機場辦理登機

但從語意上來看，它們明顯有關。

今天要替每一筆 Memory Candidate 加上 Embedding，讓文字不只是一段字串，也能變成可以計算的向量。

一、Embedding 是什麼？

Embedding 可以先理解成：

把一段文字轉換成一串能代表語意的浮點數。

例如：

使用者想加強旅遊英文。

經過 Embedding Model 後，可能變成：

[
    0.0124,
    -0.0281,
    0.0067,
    0.0412,
    ...
]

這串數值就叫做：

Embedding Vector

實際的 Vector 很長，不會只有四個數字。本文使用：

text-embedding-3-small

在預設設定下會產生 1536 維的 Vector。

OpenAI Docs 將 Embedding 定義為一串浮點數組成的向量，向量間的距離可以用來表示文字的相關程度，常見用途包括搜尋、分群、推薦與分類。OpenAI Embeddings 文件

二、Vector 的每個數字代表什麼？

看到：

[
    0.0124,
    -0.0281,
    0.0067,
    ...
]

可能會直覺地以為：

第一個數字代表旅遊
第二個數字代表英文
第三個數字代表學習

實際上不是這麼簡單。Embedding 的語意分散在整個 Vector 裡，單獨看其中一個 Dimension，通常無法直接解釋它對應哪個概念。真正有用的是比較兩個完整 Vector 的關係。

例如：

A：使用者想加強旅遊英文。

B：使用者想練習機場和飯店會話。

C：使用者喜歡用文法書學習。

轉換成 Embedding 後，A 和 B 的 Vector 應該比 A 和 C 更接近。

Embedding Model 不是直接回傳：

這兩句話很像。

它只負責把文字轉成 Vector。至於「有多像」，要由 Application 使用數學方法計算。

三、用 Cosine Similarity 比較兩個 Vector

今天使用：

Cosine Similarity

比較兩個 Embedding。

公式是：

$$
\text{cosine similarity}(A,B)

\frac{A \cdot B}
{\lVert A\rVert \lVert B\rVert}
$$

先不用被公式嚇到，它做的事情可以簡化成：

比較兩個 Vector 指向的方向有多接近。

數值越接近 1，通常代表方向越接近；分數越低，代表語意關係可能越弱。

但不要直接訂出：

0.8 以上一定相似
0.5 以下一定無關

Similarity Score 會受到模型、文字長度、資料內容與應用情境影響。真正的 Threshold 通常需要根據自己的資料測試。

四、先加入 Embedding Model

Day 13 已經有：

MODEL = "gpt-5-mini"

這個 Model 負責：

Chat Response
Conversation Summary
Memory Extraction

Embedding 需要使用另一種專門的 Model。

在下方新增：

EMBEDDING_MODEL = "text-embedding-3-small"

因此目前程式有兩種 Model：

MODEL = "gpt-5-mini"

EMBEDDING_MODEL = "text-embedding-3-small"

它們的用途不同：

Embedding Model 不會回答問題，也不會產生 Conversation Reply。

五、不要讓 LLM 自己產生 Embedding

Day 13 的 Structured Output Model 是：

class MemoryCandidate(BaseModel):
    content: str

可能會想直接改成：

class MemoryCandidate(BaseModel):
    content: str
    embedding: list[float]

但這樣會產生一個問題。

MemoryCandidate 是傳給：

client.responses.parse()

的 Structured Output Schema。如果把 embedding 放進去，就等於要求文字生成模型自己填入上千個浮點數。

Embedding 應該由：

Embedding Model

產生，不是由 Memory Extractor 猜出來。

所以保留原本的：

class MemoryCandidate(BaseModel):
    content: str

再另外新增：

class EmbeddedMemoryCandidate(BaseModel):
    content: str
    embedding: list[float]

兩者的責任如下：

MemoryCandidate
LLM 從 User Message 抽取的內容

EmbeddedMemoryCandidate
Application 加上 Embedding 後的資料

這個分離很重要，因為：

content

是由 LLM 抽取的語意資訊，而：

embedding

是由 Embedding API 計算出的衍生資料。

六、一次替多筆 Memory 產生 Embedding

一段 User Message 可能抽取出多筆 Memory：

[
    MemoryCandidate(
        content="使用者的英文程度是 B1。"
    ),
    MemoryCandidate(
        content="使用者想加強旅遊英文。"
    )
]

我們不需要分別呼叫兩次 Embedding API。

Embeddings API 的 input 可以接受多段文字，因此可以一次送出：

[
    "使用者的英文程度是 B1。",
    "使用者想加強旅遊英文。"
]

新增以下 Function：

def embed_memory_candidates(candidates):
    if not candidates:
        return [], 0

    texts = [
        candidate.content
        for candidate in candidates
    ]

    response = client.embeddings.create(
        model=EMBEDDING_MODEL,
        input=texts,
        encoding_format="float"
    )

    embedding_data = sorted(
        response.data,
        key=lambda item: item.index
    )

    embedded_candidates = []

    for candidate, data in zip(
        candidates,
        embedding_data
    ):
        embedded_candidates.append(
            EmbeddedMemoryCandidate(
                content=candidate.content,
                embedding=data.embedding
            )
        )

    return (
        embedded_candidates,
        response.usage.total_tokens
    )

這個 Function 會先取出所有 Memory Content：

texts = [
    candidate.content
    for candidate in candidates
]

接著呼叫：

client.embeddings.create()

最後把 Content 和對應的 Vector 組合成：

EmbeddedMemoryCandidate

七、為什麼要按照 index 排序？

Embeddings API 的每一筆結果都包含：

data.index

假設輸入順序是：

[
    "使用者的英文程度是 B1。",
    "使用者想加強旅遊英文。"
]

Response 裡的每個 Embedding 會帶有它對應的 Index。

因此程式先執行：

embedding_data = sorted(
    response.data,
    key=lambda item: item.index
)

再使用：

zip(candidates, embedding_data)

配對。

這樣可以確保：

第一筆 Memory
搭配第一筆 Embedding

第二筆 Memory
搭配第二筆 Embedding

否則一旦 Content 和 Vector 配錯，後面計算 Similarity 時就會得到完全錯誤的結果。

八、實作 Cosine Similarity

在程式最上方加入：

from math import sqrt

接著新增：

def cosine_similarity(vector_a, vector_b):
    if len(vector_a) != len(vector_b):
        raise ValueError(
            "兩個 Vector 的 Dimension 不一致。"
        )

    dot_product = sum(
        a * b
        for a, b in zip(vector_a, vector_b)
    )

    magnitude_a = sqrt(
        sum(value * value for value in vector_a)
    )

    magnitude_b = sqrt(
        sum(value * value for value in vector_b)
    )

    if magnitude_a == 0 or magnitude_b == 0:
        return 0.0

    return dot_product / (
        magnitude_a * magnitude_b
    )

其中：

dot_product

計算兩個 Vector 的 Dot Product。

magnitude_a
magnitude_b

則分別計算兩個 Vector 的長度。

最後得到：

similarity_score

今天只會用它比較兩筆指定的 Memory。

下一篇才會把一個 Query 和所有 Memory 比較，再按照分數排序。

九、修改手動 remember

Day 13 的手動 Memory 是：

manual_candidate = MemoryCandidate(
    content=candidate_text
)

memory_candidates.append(
    manual_candidate
)

今天不能直接加入，因為 memory_candidates 接下來要保存：

EmbeddedMemoryCandidate

因此改成：

manual_candidate = MemoryCandidate(
    content=candidate_text
)

try:
    (
        new_embedded_candidates,
        embedding_tokens
    ) = embed_memory_candidates(
        [manual_candidate]
    )

    memory_candidates.extend(
        new_embedded_candidates
    )

    memory.add_token_usage(
        embedding_tokens
    )

except Exception as error:
    embedding_error = str(error)

流程變成：

remember 後面的文字
        ↓
MemoryCandidate
        ↓
Embedding API
        ↓
EmbeddedMemoryCandidate
        ↓
memory_candidates

手動建立的 Memory 和 LLM 自動抽取的 Memory，最後會使用相同的資料結構。

十、修改自動 Memory Extraction

Day 13 的 Extractor 會產生：

extraction_result.memories

它是一個：

list[MemoryCandidate]

今天要在加入 memory_candidates 前，多做一步：

if extraction_result.should_remember:
    extracted_candidates = (
        extraction_result.memories
    )

    (
        new_embedded_candidates,
        embedding_tokens
    ) = embed_memory_candidates(
        extracted_candidates
    )

    memory_candidates.extend(
        new_embedded_candidates
    )

    memory.add_token_usage(
        embedding_tokens
    )

因此自動抽取流程會從：

User Message
      ↓
Memory Extraction
      ↓
MemoryCandidate

變成：

User Message
      ↓
Memory Extraction
      ↓
MemoryCandidate
      ↓
Embedding Model
      ↓
EmbeddedMemoryCandidate

原本的 Chat Response 和 Short-term Memory 流程都不需要改變。

十一、加入 vector 指令

Embedding 通常包含上千個數字，不適合每次全部印出。

所以新增：

vector <Memory Number>

只查看指定 Memory 的基本資訊與前八個數字。

if command == "vector":
    print("Usage: vector <memory number>")
    continue

if command.startswith("vector "):
    parts = command.split()

    try:
        memory_number = int(parts[1])
        candidate = memory_candidates[
            memory_number - 1
        ]

    except (
        ValueError,
        IndexError
    ):
        print("Invalid memory number.")
        continue

    print("\n--- Embedding Vector ---")
    print("Content:", candidate.content)
    print(
        "Dimensions:",
        len(candidate.embedding)
    )
    print(
        "Preview:",
        [
            round(value, 6)
            for value in candidate.embedding[:8]
        ]
    )
    print("------------------------")
    continue

例如：

You:
vector 1

可能看到：

--- Embedding Vector ---
Content: 使用者想加強旅遊英文。
Dimensions: 1536
Preview: [0.012451, -0.028103, 0.006712, ...]
------------------------

每次實際取得的數值可能不同，不需要特別記住其中任何一個數字。

十二、加入 similarity 指令

再新增：

similarity <Memory A> <Memory B>

用來比較兩筆 Memory：

if command == "similarity":
    print(
        "Usage: similarity "
        "<memory number 1> "
        "<memory number 2>"
    )
    continue

if command.startswith("similarity "):
    parts = command.split()

    try:
        first_number = int(parts[1])
        second_number = int(parts[2])

        first_memory = memory_candidates[
            first_number - 1
        ]

        second_memory = memory_candidates[
            second_number - 1
        ]

    except (
        ValueError,
        IndexError
    ):
        print("Invalid memory number.")
        continue

    score = cosine_similarity(
        first_memory.embedding,
        second_memory.embedding
    )

    print("\n--- Memory Similarity ---")
    print(
        "Memory 1:",
        first_memory.content
    )
    print(
        "Memory 2:",
        second_memory.content
    )
    print(
        "Cosine similarity:",
        round(score, 4)
    )
    print("-------------------------")
    continue

這個指令不會再次呼叫 API，因為兩筆 Memory 的 Embedding 已經保存於：

memory_candidates

程式只需要拿出兩個 Vector 進行數學計算。

十三、Memora v0.11 的主程式

Day 13 的 ShortTermMemory、MemoryExtractionResult 與 extract_memory_candidates() 全部保留。

新增以下內容：

from math import sqrt

EMBEDDING_MODEL = "text-embedding-3-small"

class EmbeddedMemoryCandidate(BaseModel):
    content: str
    embedding: list[float]

以及：

embed_memory_candidates()
cosine_similarity()

接著將 Day 13 從建立 memory 開始的主程式，更新成以下版本：

memory = ShortTermMemory(
    client=client,
    model=MODEL,
    system_prompt=SYSTEM_PROMPT,
    summary_instructions=SUMMARY_INSTRUCTIONS,
    max_input_tokens=MAX_INPUT_TOKENS,
    max_recent_turns=MAX_RECENT_TURNS
)

# 現在保存的是 EmbeddedMemoryCandidate。
memory_candidates = []

last_extraction_output = ""

print("Memora v0.11")
print(
    "Commands: history, context, summary, status, "
    "remember <text>, memories, extraction, "
    "vector <number>, similarity <a> <b>, exit"
)

while True:
    user_input = input("\nYou: ").strip()

    if not user_input:
        continue

    command = user_input.lower()

    should_auto_extract = True
    extraction_tokens = 0
    embedding_tokens = 0
    embedding_error = ""
    new_embedded_candidates = []

    if command == "exit":
        print("Bye!")
        break

    if command == "history":
        print_messages(
            "Full Conversation History",
            memory.history
        )
        continue

    if command == "context":
        print_messages(
            "Last Request Context",
            memory.last_context
        )
        continue

    if command == "summary":
        print("\n--- Conversation Summary ---")
        print(memory.summary or "(empty)")
        print("----------------------------")
        continue

    if command == "memories":
        print("\n--- Memory Candidates ---")

        if not memory_candidates:
            print("(empty)")
        else:
            for index, candidate in enumerate(
                memory_candidates,
                start=1
            ):
                print(
                    f"{index}. {candidate.content}"
                    f" ({len(candidate.embedding)} dimensions)"
                )

        print("-------------------------")
        continue

    if command == "extraction":
        print("\n--- Last Extraction Output ---")
        print(last_extraction_output or "(not run)")
        print("------------------------------")
        continue

    if command == "vector":
        print("Usage: vector <memory number>")
        continue

    if command.startswith("vector "):
        parts = command.split()

        try:
            memory_number = int(parts[1])

            if memory_number < 1:
                raise IndexError

            candidate = memory_candidates[
                memory_number - 1
            ]

        except (
            ValueError,
            IndexError
        ):
            print("Invalid memory number.")
            continue

        print("\n--- Embedding Vector ---")
        print("Content:", candidate.content)
        print(
            "Model:",
            EMBEDDING_MODEL
        )
        print(
            "Dimensions:",
            len(candidate.embedding)
        )
        print(
            "Preview:",
            [
                round(value, 6)
                for value in candidate.embedding[:8]
            ]
        )
        print("------------------------")
        continue

    if command == "similarity":
        print(
            "Usage: similarity "
            "<memory number 1> "
            "<memory number 2>"
        )
        continue

    if command.startswith("similarity "):
        parts = command.split()

        try:
            first_number = int(parts[1])
            second_number = int(parts[2])

            if first_number < 1 or second_number < 1:
                raise IndexError

            first_memory = memory_candidates[
                first_number - 1
            ]

            second_memory = memory_candidates[
                second_number - 1
            ]

        except (
            ValueError,
            IndexError
        ):
            print("Invalid memory number.")
            continue

        score = cosine_similarity(
            first_memory.embedding,
            second_memory.embedding
        )

        print("\n--- Memory Similarity ---")
        print(
            "Memory 1:",
            first_memory.content
        )
        print(
            "Memory 2:",
            second_memory.content
        )
        print(
            "Cosine similarity:",
            round(score, 4)
        )
        print("-------------------------")
        continue

    if command == "status":
        print("\n--- Memory Status ---")

        for name, value in memory.get_status().items():
            print(f"{name}: {value}")

        print(
            "Memory candidates:",
            len(memory_candidates)
        )
        print(
            "Embedding model:",
            EMBEDDING_MODEL
        )

        print("---------------------")
        continue

    if command == "remember":
        print("Usage: remember <text>")
        continue

    if command.startswith("remember "):
        candidate_text = user_input[
            len("remember "):
        ].strip()

        if not candidate_text:
            print("Usage: remember <text>")
            continue

        manual_candidate = MemoryCandidate(
            content=candidate_text
        )

        last_extraction_output = (
            manual_candidate.model_dump_json(
                indent=2
            )
        )

        try:
            (
                new_embedded_candidates,
                embedding_tokens
            ) = embed_memory_candidates(
                [manual_candidate]
            )

            memory_candidates.extend(
                new_embedded_candidates
            )

            memory.add_token_usage(
                embedding_tokens
            )

        except Exception as error:
            embedding_error = str(error)

        user_input = candidate_text
        should_auto_extract = False

    memory.add_user_message(user_input)

    try:
        memory_stats = memory.prepare_context()

        response = client.responses.create(
            model=MODEL,
            instructions=SYSTEM_PROMPT,
            input=memory_stats["context_messages"]
        )

    except Exception as error:
        memory.rollback_last_user_message()
        print("Request failed:", error)
        continue

    assistant_reply = response.output_text

    memory.finish_turn(
        assistant_reply=assistant_reply,
        context_messages=memory_stats[
            "context_messages"
        ],
        response_tokens=response.usage.total_tokens
    )

    extracted_candidates = []

    if should_auto_extract:
        try:
            (
                extraction_result,
                extraction_tokens
            ) = extract_memory_candidates(
                user_input
            )

            memory.add_token_usage(
                extraction_tokens
            )

            last_extraction_output = (
                extraction_result.model_dump_json(
                    indent=2
                )
            )

            if extraction_result.should_remember:
                extracted_candidates = (
                    extraction_result.memories
                )

        except Exception as error:
            last_extraction_output = (
                f"Extraction failed: {error}"
            )

    if extracted_candidates:
        try:
            (
                new_embedded_candidates,
                embedding_tokens
            ) = embed_memory_candidates(
                extracted_candidates
            )

            memory_candidates.extend(
                new_embedded_candidates
            )

            memory.add_token_usage(
                embedding_tokens
            )

        except Exception as error:
            embedding_error = str(error)

    print("Memora:", assistant_reply)

    if new_embedded_candidates:
        print("\n--- New Embedded Memories ---")

        for candidate in new_embedded_candidates:
            print("-", candidate.content)
            print(
                "  Dimensions:",
                len(candidate.embedding)
            )

        print("-----------------------------")

    if embedding_error:
        print(
            "\nEmbedding failed:",
            embedding_error
        )

    print("\n--- Memory Status ---")
    print(
        "Newly summarized messages:",
        memory_stats["newly_summarized_count"]
    )
    print(
        "Context messages sent:",
        len(memory_stats["context_messages"])
    )
    print(
        "Memory extraction tokens:",
        extraction_tokens
    )
    print(
        "Embedding input tokens:",
        embedding_tokens
    )
    print(
        "Memory candidates:",
        len(memory_candidates)
    )
    print(
        "Session total tokens:",
        memory.session_total_tokens
    )
    print("---------------------")

十四、實際比較三筆 Memory

可以先手動建立三筆容易觀察的 Memory：

You:
remember 使用者想加強旅遊英文。

You:
remember 使用者想練習機場和飯店會話。

You:
remember 使用者偏好透過文法題學習。

輸入：

memories

會看到：

1. 使用者想加強旅遊英文。 (1536 dimensions)
2. 使用者想練習機場和飯店會話。 (1536 dimensions)
3. 使用者偏好透過文法題學習。 (1536 dimensions)

接著比較：

similarity 1 2

再比較：

similarity 1 3

實際分數會依文字與模型結果而不同，但預期關係是：

Similarity(旅遊英文, 機場與飯店會話)
>
Similarity(旅遊英文, 文法題學習)

重點不是某一次得到多少分，而是 Embedding 讓我們可以比較語意，不必要求兩段文字擁有完全相同的關鍵字。

十五、比較 Embedding 時要注意什麼？

第一，兩個 Vector 必須具有相同 Dimension。

程式已經檢查：

if len(vector_a) != len(vector_b):
    raise ValueError(...)

更重要的是，它們應該由相同的 Embedding Model 與相同設定產生。

不要直接比較：

text-embedding-3-small 產生的 Vector

和：

另一個 Embedding Model 產生的 Vector

因為它們不一定位於相同的向量空間中。

第二，Embedding 不是原始文字的替代品。

我們仍然要保存：

content

因為 Vector 適合計算，卻不適合讓人閱讀。

所以每筆資料要同時具有：

EmbeddedMemoryCandidate(
    content="使用者想加強旅遊英文。",
    embedding=[...]
)

第三，Embedding 不會判斷 Memory 是否正確。

如果 Extractor 錯誤地建立：

使用者每天只想練習五分鐘。

Embedding 只會忠實地替這句錯誤內容產生 Vector，不會替我們修正它。

十六、現在還不是 Semantic Search

今天我們只能手動指定：

比較 Memory 1 和 Memory 2

也就是：

similarity 1 2

真正的 Semantic Search 還需要：

1. 把使用者的 Query 也轉成 Embedding。
2. 和所有 Memory Embedding 比較。
3. 計算每一筆 Cosine Similarity。
4. 按照分數排序。
5. 取回最相關的前幾筆結果。

例如使用者搜尋：

我想練習出國時會用到的對話。

程式應該自動找出：

使用者想加強旅遊英文。
使用者想練習機場和飯店會話。

而不是要求我們手動輸入 Memory Number。

這正是 Day 15 要實作的內容。

Day 14 小結

昨天的 Memory Candidate 只有：

MemoryCandidate(
    content="使用者想加強旅遊英文。"
)

今天我們沒有改掉 Structured Output，而是在它後面增加 Embedding：

EmbeddedMemoryCandidate(
    content="使用者想加強旅遊英文。",
    embedding=[...]
)

目前的資料流程變成：

User Message
      ↓
Structured Memory Extraction
      ↓
MemoryCandidate
      ↓
Embedding API
      ↓
EmbeddedMemoryCandidate

接著使用：

cosine_similarity()

比較兩筆 Memory 的 Vector。

因此 Memora 第一次可以透過數學計算判斷：

即使兩段文字使用不同的詞，它們在語意上仍然可能非常接近。

但現在仍然要手動指定要比較哪兩筆 Memory。

Day 15｜不用 Vector Database，自己實作一次 Semantic Search

下一篇，我們會直接沿用今天保存的：

memory_candidates

並加入：

semantic_search(query, memories)

當使用者輸入：

我想練習出國時會用到的英文。

程式會：

將 Query 轉成 Embedding
        ↓
和每一筆 Memory 計算 Cosine Similarity
        ↓
按照分數排序
        ↓
回傳最相關的 Memory

在使用 Vector Database 之前，我們先自己做一次最基本的 Semantic Search，真正看懂 Vector Search 背後做了什麼。