昨天，我們讓 Memory 的 Retrieval Score 加入了 Recency，也提供：

Soft Forgetting
Hard Forgetting

但時間只能回答：

這筆 Memory 最近有沒有被使用？

它不能回答：

這筆資料是不是另一筆的重複版本？
使用者的狀態是不是已經改變？
兩筆看似矛盾的 Memory，是否其實發生在不同時間？

例如 Long-term Memory Store 裡可能同時出現：

使用者的英文程度是 B1。
使用者目前的英文程度是 B2。

如果只是把兩筆都保存，之後 Retrieval 可能同時把 B1 和 B2 放進 Context。模型即使看得懂矛盾，也不代表 Store 裡的資料已經被正確維護。所以今天要從 Day 23 的程式繼續修改，在新 Memory 寫入 Chroma 前加入一個新的步驟：

Memory Reconciliation

讓 Memora 判斷這次應該建立、略過、更新，還是保留兩筆資料。

一、今天修改的是 Write Path

昨天已經有兩條主要路徑：

Read Path
Semantic Search
→ Relevance Threshold
→ Relevance + Importance + Recency
→ 放入 Context

Write Path
Memory Extraction
→ Memory Policy
→ Importance Score
→ Embedding
→ LongTermMemoryStore.add()

今天不修改前面的 Retrieval 流程，也不移除 touch()、delete() 與 forget Command。要修改的是最後一段 Write Path：

Memory Extraction
→ Memory Policy
→ Importance Score
→ Embedding
→ 搜尋相近的 Existing Memories
→ Reconciliation
→ add / skip / update / review

也就是把判斷放在：

LongTermMemoryStore.add()

之前，而不是等錯誤資料進入 Store 後才處理。

二、先分清楚五種結果

兩段文字相近，不一定代表它們應該被合併。

這裡新增 review，是因為不是每一個 Contradiction 都適合自動覆蓋。當 Application 無法確認 Incoming Memory 是新的狀態、過去事件，還是抽取錯誤時，先不寫入，也不破壞舊資料，會比猜一個答案安全。

三、Semantic Similarity 只負責找候選資料

Day 15 已經實作過 Semantic Search。今天可以重用它找出可能需要比較的 Existing Memories，但不能直接寫成：

if similarity > 0.8:
    skip_memory()

因為下面兩句通常非常相似：

使用者的英文程度是 B1。
使用者的英文程度是 B2。

它們談的是同一個屬性，所以 Embedding 距離很近；但它們不是重複資料，而是可能需要更新的狀態。

另一方面：

使用者昨天完成機場英文練習。
使用者今天完成餐廳英文練習。

兩筆也很相近，卻可以同時成立。因此今天採用兩階段判斷：

Embedding
→ 找出可能有關的 Candidates

LLM
→ 判斷 Candidates 和 Incoming Memory 的關係

Similarity 是搜尋工具，不是最終裁判。

四、替 Memory 加入 updated_at

Day 23 已經保存：

created_at
last_accessed_at

今天會原地更新既有 Record，因此還需要：

updated_at

三個欄位分別表示：

先修改 Day 23 的 LongTermMemoryStore.add()。建立新資料時，三個時間先使用同一個值：

created_at = datetime.now(
    timezone.utc
).isoformat()

metadatas=[
    {
        "source": source,
        "created_at": created_at,
        "updated_at": created_at,
        "last_accessed_at": created_at,
        "memory_type": memory.memory_type,
        "importance_score": (
            memory.importance_score
        )
    }
    for memory in memories
]

接著在 Day 23 的兩個 Read Models 中加入：

class StoredMemory(BaseModel):
    memory_id: str
    content: str
    memory_type: StoredMemoryType
    importance_score: int = Field(
        ge=1,
        le=5
    )
    source: str
    created_at: str
    updated_at: str
    last_accessed_at: str


class MemorySearchResult(BaseModel):
    memory_id: str
    content: str
    memory_type: StoredMemoryType
    importance_score: int = Field(
        ge=1,
        le=5
    )
    source: str
    created_at: str
    updated_at: str
    last_accessed_at: str
    distance: float
    score: float

為了相容舊資料，再加入：

def read_created_at(
    metadata: dict
) -> str:
    created_at = metadata.get(
        "created_at"
    )

    if (
        isinstance(created_at, str)
        and parse_utc_timestamp(
            created_at
        ) is not None
    ):
        return created_at

    return "unknown"


def read_updated_at(
    metadata: dict
) -> str:
    updated_at = metadata.get(
        "updated_at"
    )

    if (
        isinstance(updated_at, str)
        and parse_utc_timestamp(
            updated_at
        ) is not None
    ):
        return updated_at

    return read_created_at(metadata)

最後在 list_all() 與 search() 建立 Model 時加入：

updated_at=read_updated_at(metadata)

read_created_at() 沿用原本讀取 created_at 的 Helper；舊 Record 沒有 updated_at 時，就暫時回退到建立時間。

Day 23 的 memories Debug Output 也補上：

print(
    f"   Updated at: "
    f"{stored_memory.updated_at}"
)

五、更新過的 Memory 也應該重新變得新鮮

Day 23 的 Recency 只看 last_accessed_at。但如果今天把一筆兩個月前的 B1 更新成 B2，新的內容不應該立刻繼承兩個月前的低 Recency。

同時，也不應該把內容更新假裝成一次回答中的 Access。

因此保留 last_accessed_at 的原始意義，再讓 Recency 使用下面兩者中較新的時間：

updated_at
last_accessed_at

新增：

def calculate_memory_recency_score(
    last_accessed_at: str,
    updated_at: str,
    now: datetime | None = None
) -> float:
    timestamps = [
        parse_utc_timestamp(
            last_accessed_at
        ),
        parse_utc_timestamp(
            updated_at
        )
    ]

    valid_timestamps = [
        timestamp
        for timestamp in timestamps
        if timestamp is not None
    ]

    if not valid_timestamps:
        return DEFAULT_RECENCY_SCORE

    latest_activity_at = max(
        valid_timestamps
    )

    return calculate_recency_score(
        latest_activity_at.isoformat(),
        now=now
    )

接著只修改 Day 23 的 calculate_retrieval_score() 其中一段：

recency_score = (
    calculate_memory_recency_score(
        last_accessed_at=(
            result.last_accessed_at
        ),
        updated_at=result.updated_at,
        now=now
    )
)

原本的 Relevance、Importance 與三組權重都不需要改。

memories Command 顯示 Recency 時，也改用同一個 Helper：

recency_score = (
    calculate_memory_recency_score(
        last_accessed_at=(
            stored_memory.last_accessed_at
        ),
        updated_at=(
            stored_memory.updated_at
        )
    )
)

六、定義 Reconciliation 的 Structured Output

Day 13 已經讓 LLM 回傳可驗證的 Structured Output。今天沿用同一種做法，不解析自由格式文字。

先新增：

from typing import Literal


MemoryWriteAction = Literal[
    "create",
    "skip",
    "update",
    "keep_both",
    "review"
]


class MemoryReconciliationDecision(
    BaseModel
):
    action: MemoryWriteAction
    target_memory_id: str | None = None
    reason: str = Field(
        min_length=1,
        max_length=200
    )

target_memory_id 只在下列動作需要：

skip
update
review

這樣 Application 才知道 LLM 指的是哪一筆 Existing Memory，而不是根據 List 順序猜測。

reason 只用於 Debug，不直接存成新的 Memory。

七、告訴模型什麼時候可以 Update

建立新的 Prompt：

MEMORY_RECONCILIATION_PROMPT = """
You maintain long-term memory for an English
learning assistant.

Compare one incoming memory with the existing
candidate memories.

Choose exactly one action:

- create: The incoming memory is unrelated to
  the candidates and should become a new record.
- skip: An existing candidate already expresses
  the same fact or event.
- update: The incoming memory clearly replaces
  the same mutable current state in one existing
  semantic memory.
- keep_both: The memories are related, but both
  can remain true, especially distinct episodic
  events or facts about different times.
- review: There is a possible contradiction, but
  there is not enough evidence for a safe update.

Rules:
- Similar wording alone does not mean duplicate.
- Do not overwrite one event with another event.
- Use update only when the incoming memory is a
  clear correction or newer current state.
- For skip, update, or review, return the exact
  target_memory_id from the candidates.
- For create or keep_both, target_memory_id must
  be null.
- Do not invent facts.
- Keep reason to one short sentence.
"""

這裡特別限制：

Episodic Memory
不應該只因內容相似就互相覆蓋

例如昨天程度是 B1、今天程度是 B2，若兩者都被寫成「目前狀態」，適合更新；但昨天完成 Lesson 1、今天完成 Lesson 2，則是兩個不同事件。

八、用既有 Embedding 搜尋可能衝突的資料

先設定搜尋範圍：

RECONCILIATION_TOP_K = 3
RECONCILIATION_MIN_SCORE = 0.78

這個 Threshold 只用來排除明顯無關的資料，不直接決定 skip 或 update。

新增：

def find_reconciliation_candidates(
    incoming: EmbeddedMemoryCandidate
) -> list[MemorySearchResult]:
    candidates = long_term_memory.search(
        query_embedding=incoming.embedding,
        top_k=RECONCILIATION_TOP_K
    )

    return [
        candidate
        for candidate in candidates
        if candidate.score
        >= RECONCILIATION_MIN_SCORE
    ]

這裡直接使用 Incoming Memory 已經建立好的：

incoming.embedding

所以不用為 Reconciliation 再呼叫一次 Embeddings API。

這次搜尋屬於 Write Path 的內部檢查，不是把 Memory 放進回答 Context，因此不呼叫 Day 23 的：

long_term_memory.touch()

九、完全相同的內容先直接略過

有些重複資料不需要再請 LLM 判斷，例如大小寫或多餘空白不同的同一句話。

新增：

def normalize_memory_text(
    content: str
) -> str:
    return " ".join(
        content.casefold().split()
    )


def find_exact_duplicate(
    incoming: EmbeddedMemoryCandidate,
    candidates: list[MemorySearchResult]
) -> MemorySearchResult | None:
    incoming_text = normalize_memory_text(
        incoming.content
    )

    for candidate in candidates:
        if normalize_memory_text(
            candidate.content
        ) == incoming_text:
            return candidate

    return None

這只是 Cheap Check：

完全相同
→ 直接 skip

語意相同但文字不同
→ 交給 LLM 判斷

十、讓 LLM 判斷 Candidate 之間的關係

接著加入：

def decide_memory_reconciliation(
    incoming: EmbeddedMemoryCandidate,
    candidates: list[MemorySearchResult]
) -> tuple[
    MemoryReconciliationDecision,
    int
]:
    payload = {
        "incoming_memory": {
            "content": incoming.content,
            "memory_type": (
                incoming.memory_type
            ),
            "importance_score": (
                incoming.importance_score
            )
        },
        "existing_candidates": [
            {
                "memory_id": item.memory_id,
                "content": item.content,
                "memory_type": (
                    item.memory_type
                ),
                "importance_score": (
                    item.importance_score
                ),
                "created_at": item.created_at,
                "updated_at": item.updated_at,
                "similarity_score": item.score
            }
            for item in candidates
        ]
    }

    response = client.responses.parse(
        model="gpt-5.6",
        instructions=(
            MEMORY_RECONCILIATION_PROMPT
        ),
        input=json.dumps(
            payload,
            ensure_ascii=False
        ),
        text_format=(
            MemoryReconciliationDecision
        )
    )

    decision = response.output_parsed

    candidate_ids = {
        item.memory_id
        for item in candidates
    }

    if decision.action in {
        "skip",
        "update",
        "review"
    }:
        if (
            decision.target_memory_id
            not in candidate_ids
        ):
            raise ValueError(
                "Invalid reconciliation target."
            )
    elif decision.target_memory_id is not None:
        raise ValueError(
            "This action must not have a target."
        )

    return (
        decision,
        response.usage.total_tokens
    )

除了讓 Pydantic 驗證資料型別，Application 還要再確認：

target_memory_id
確實存在於這次提供的 Candidates

Structured Output 可以限制格式，但不能取代業務規則驗證。

十一、替 LongTermMemoryStore 加入 update()

現在 Store 已經有：

add()
search()
touch()
delete()

今天在同一個 Class 裡新增：

def update(
    self,
    memory_id: str,
    memory: EmbeddedMemoryCandidate,
    source: str
) -> bool:
    records = self.collection.get(
        ids=[memory_id],
        include=["metadatas"]
    )

    record_ids = records.get("ids", [])

    if not record_ids:
        return False

    record_metadatas = (
        records.get("metadatas")
        or [{}]
    )

    metadata = dict(
        record_metadatas[0] or {}
    )

    updated_at = datetime.now(
        timezone.utc
    ).isoformat()

    metadata.setdefault(
        "created_at",
        updated_at
    )
    metadata.setdefault(
        "last_accessed_at",
        metadata["created_at"]
    )

    metadata.update(
        {
            "source": source,
            "updated_at": updated_at,
            "memory_type": (
                memory.memory_type
            ),
            "importance_score": (
                memory.importance_score
            )
        }
    )

    self.collection.update(
        ids=[memory_id],
        documents=[memory.content],
        embeddings=[memory.embedding],
        metadatas=[metadata]
    )

    return True

這個方法會：

保留原本 memory_id
保留 created_at
保留 last_accessed_at
更新 Document
更新對應的 Embedding
更新 memory_type 與 importance_score
寫入新的 updated_at

Document 變了，Embedding 也必須一起更新，否則之後 Semantic Search 使用的向量仍然代表舊內容。

Chroma 的 update() 可以更新 Record 的 Document、Embedding 與 Metadata。這裡明確傳入新的 Embedding，也讓目前由 Application 管理 Embedding 的架構保持一致。

十二、把 Add、Skip 與 Update 接在一起

新增一個 Helper，負責處理單筆 Incoming Memory：

def reconcile_and_store_memory(
    incoming: EmbeddedMemoryCandidate,
    source: str
) -> tuple[list[str], int]:
    candidates = (
        find_reconciliation_candidates(
            incoming
        )
    )

    exact_duplicate = find_exact_duplicate(
        incoming=incoming,
        candidates=candidates
    )

    if exact_duplicate is not None:
        print(
            "Memory reconciliation: skip "
            "(exact duplicate)"
        )
        return [], 0

    if not candidates:
        memory_ids = long_term_memory.add(
            memories=[incoming],
            source=source
        )
        return memory_ids, 0

    (
        decision,
        reconciliation_tokens
    ) = decide_memory_reconciliation(
        incoming=incoming,
        candidates=candidates
    )

    print(
        "Memory reconciliation:",
        decision.action,
        "-",
        decision.reason
    )

    if decision.action in {
        "create",
        "keep_both"
    }:
        memory_ids = long_term_memory.add(
            memories=[incoming],
            source=source
        )
        return (
            memory_ids,
            reconciliation_tokens
        )

    if decision.action == "skip":
        return [], reconciliation_tokens

    if decision.action == "review":
        print(
            "Memory was not changed. "
            "Manual review is required."
        )
        return [], reconciliation_tokens

    target_memory_id = (
        decision.target_memory_id
    )

    if target_memory_id is None:
        raise ValueError(
            "Update action requires a target."
        )

    updated = long_term_memory.update(
        memory_id=target_memory_id,
        memory=incoming,
        source=source
    )

    if not updated:
        raise ValueError(
            "Reconciliation target no longer exists."
        )

    return (
        [target_memory_id],
        reconciliation_tokens
    )

這裡沒有在 Update 失敗後自動建立新 Record。因為 Target 突然消失代表 Store 狀態已經和判斷時不同，直接建立可能製造新的重複或衝突，應該讓本次寫入明確失敗。

十三、修改 Day 22 的 store_accepted_memories()

Day 22 已經把 Importance Scoring、Embedding 與寫入包在：

store_accepted_memories()

Day 23 沒有改掉這個入口。今天也不建立另一條寫入流程，只把最後原本直接呼叫 add() 的地方換掉：

def store_accepted_memories(
    candidates: list[MemoryCandidate],
    source: str
) -> tuple[list[str], int]:
    (
        scored_memories,
        importance_tokens
    ) = score_memory_importance(
        candidates
    )

    if not scored_memories:
        return [], importance_tokens

    memory_contents = [
        memory_item.content
        for memory_item in scored_memories
    ]

    (
        memory_embeddings,
        embedding_tokens
    ) = create_embeddings(
        memory_contents
    )

    embedded_memories = [
        EmbeddedMemoryCandidate(
            content=memory_item.content,
            memory_type=(
                memory_item.memory_type
            ),
            importance_score=(
                memory_item.importance_score
            ),
            embedding=embedding
        )
        for memory_item, embedding in zip(
            scored_memories,
            memory_embeddings
        )
    ]

    affected_memory_ids = []
    reconciliation_tokens = 0

    for embedded_memory in embedded_memories:
        (
            memory_ids,
            decision_tokens
        ) = reconcile_and_store_memory(
            incoming=embedded_memory,
            source=source
        )

        affected_memory_ids.extend(
            memory_ids
        )
        reconciliation_tokens += (
            decision_tokens
        )

    total_tokens = (
        importance_tokens
        + embedding_tokens
        + reconciliation_tokens
    )

    return (
        affected_memory_ids,
        total_tokens
    )

原本的處理順序仍然保留：

Policy 通過後
→ 評估 Importance
→ 建立 Embedding

只是在真正寫入前，改成逐筆 Reconcile。逐筆處理的好處是：同一輪抽出的第二筆 Memory，也能看見第一筆剛建立或更新的結果。

原本自動抽取 Memory 與手動 remember 都已經呼叫這個函式，所以兩條路徑會一起獲得 Deduplication 與 Update，不需要各寫一份邏輯。

十四、完整的 Write Path 現在長什麼樣子？

目前一筆新 Memory 會依序經過：

Conversation
→ Memory Extraction
→ Memory Policy
→ Importance Score
→ Embedding
→ Semantic Candidate Search

如果沒有相近 Candidate：

create
→ LongTermMemoryStore.add()

如果找到相近 Candidate：

完全相同
→ skip

語意可能相關
→ Structured Reconciliation
   ├─ create
   ├─ skip
   ├─ update
   ├─ keep_both
   └─ review

這是 Memory Maintenance 第一次正式出現在寫入流程中。

十五、測試 Duplicate、Update 與 Event

測試一：重複內容

先輸入兩次：

remember semantic 使用者偏好簡短的英文例句。
remember semantic 使用者偏好簡短的英文例句。

第二次應該直接得到：

skip (exact duplicate)

memories 中仍然只有一筆。

測試二：同一狀態更新

先建立：

remember semantic 使用者目前的英文程度是 B1。

再輸入：

remember semantic 使用者目前的英文程度是 B2。

如果 Reconciliation 判斷為 update，再次輸入：

memories

應該看到：

memory_id 沒有改變
content 變成 B2
created_at 保留
updated_at 更新
embedding 已重新寫入

測試三：不同事件都要留下

依序輸入：

remember episodic 使用者昨天完成機場英文練習。
remember episodic 使用者今天完成餐廳英文練習。

它們雖然都在描述「完成英文練習」，卻是不同時間、不同內容的事件，因此應該是：

keep_both

測試四：不確定的矛盾

如果 Incoming Memory 只說：

使用者可能不再準備 IELTS。

而 Existing Memory 是：

使用者的長期目標是準備 IELTS。

在沒有明確更新資訊時，應該回傳 review，不自動刪除或覆蓋任何一筆資料。

十六、Update 和 Forget 不一樣

Day 23 的：

forget <memory_id>

會永久刪除指定 Record。

今天的 update() 則是：

保留 Record Identity
替換目前內容與 Embedding
留下 created_at
記錄 updated_at

如果 B1 只是過去發生過的事，而不是一筆錯誤的目前狀態，那麼直接 Update 會失去歷史意義。這時比較合理的設計可能是：

B1 是過去狀態
B2 是目前狀態
→ 保留兩筆具有時間資訊的 Memory

因此 update 只適合「同一筆目前狀態被取代」，不能拿來代替所有形式的歷史變化。這也是為什麼 memory_type 和時間描述必須一起參與判斷。

十七、Long-term Memory 更新不等於 User Profile 更新

Day 19 已經把：

Long-term Memory
User Profile

分成兩個不同 Store。

因此今天更新 Chroma 中的 B1 Memory，不會自動修改：

user_profile.json

如果目前 Profile 也保存：

english_level = B1

仍然要明確執行：

profile level B2

目前的 Context 規則也繼續保留：Profile 與 Retrieved Memory 衝突時，以 Profile 表示的目前狀態為優先。

Day 24 維護的是 Long-term Memory Records，不越過 Day 19 已經建立的 Store Boundary。

十八、目前的 Reconciliation 仍然有成本

今天不是每一筆 Memory 都一定增加一次 LLM Request。

沒有相近 Candidate
→ 直接 create

完全相同
→ 直接 skip

有相近但關係不明確
→ 呼叫 LLM Reconciliation

因此只有第三種情況會增加 Reconciliation Tokens。

不過 Threshold 仍然需要依實際資料調整：

設得太高
→ 漏掉 Paraphrase 與 Contradiction

設得太低
→ 太多無關資料進入 LLM 判斷

今天的 0.78 是起始值，不是所有 Embedding Model 與資料集都通用的答案。

十九、這還不是 Agent 自己管理 Memory

現在的 Memora 已經能在固定的 Application Flow 中執行：

何時搜尋相近 Memory
何時呼叫 Reconciliation
哪些 Action 可以執行

但決定流程的仍然是 Python 程式：

reconcile_and_store_memory()

LLM 只能在程式預先提供的選項中輸出 Decision，還不能主動決定：

現在要不要使用 Memory Tool？
要先查詢、更新，還是刪除？
執行結果是否足以完成任務？

之後的章節會開始修改。

Day 24 小結

今天直接從 Day 23 的 LongTermMemoryStore 與 Write Path 繼續修改，沒有重寫原本的 Retrieval、Recency 或 Forgetting。

新增的是：

updated_at
MemoryReconciliationDecision
find_reconciliation_candidates()
find_exact_duplicate()
decide_memory_reconciliation()
LongTermMemoryStore.update()
reconcile_and_store_memory()

Memory 寫入流程現在變成：

Extract
→ Policy
→ Importance
→ Embedding
→ 找出相近 Candidates
→ 判斷 Memory 關係
→ create / skip / update / keep_both / review

今天最重要的觀念是：

Similarity 只能告訴我們兩筆 Memory 可能有關，不能直接決定它們是重複、更新，還是兩個都應該保留。可靠的 Memory Store 必須同時維護 Record 的一致性與歷史意義。

到這裡，Chapter 4 已經完成了一輪基本的 Memory Maintenance：

Memory 與 User Profile 的邊界
Semantic 與 Episodic Memory
Memory Policy
Importance
Recency 與 Forgetting
Deduplication、Update 與 Contradiction

目前這些能力仍然由 Application 依固定順序呼叫。下一章開始，我們要讓 LLM 不只回傳文字，而是能要求 Application 執行一個明確的動作。

Day 25｜Tool Calling 是什麼？讓 LLM 不只會回答

下一篇我們會先把 Memory 暫時放在背景，單獨理解 Tool Calling 的基本結構：

User Request
→ LLM 選擇 Tool
→ Application 執行 Function
→ Tool Result 回到模型
→ Final Response

接著才會在 Day 28 把 Tool、Memory 與 Agent 串在一起，讓目前由 Python 固定控制的 Memory Flow，逐步走向 Agentic Memory。

參考資料

• Chroma Docs：Update Records
• Chroma Docs：Update Data
• Chroma Docs：Delete Records
• Day 22｜Importance Score：哪些記憶比較重要？