iT邦幫忙

2026 iThome 鐵人賽

DAY 13
0
AI Engineering

從 Stateless LLM 到 Agentic Memory:30 天打造會記憶的 AI Agent系列 第 13

Day 13|Structured Output:把 Memory 變成真正可以存的資料

  • 分享至 

  • xImage
  •  

昨天 Memora 已經可以自動分析 User Message,找出可能值得保存的資訊。

例如輸入:

我的英文程度是 B1,我想加強旅遊英文。

Memory Extractor 可能回傳:

MEMORY: 使用者的英文程度是 B1。
MEMORY: 使用者想加強旅遊英文。

接著 Python 再透過:

splitlines()
startswith("MEMORY:")

解析這段文字。

這個方法可以運作,但它建立在一個不太穩定的前提上:

模型必須完全按照我們約定的文字格式回答。

如果模型改成:

以下是值得保存的資訊:

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

人類仍然看得懂,程式卻可能無法解析。

今天要保留昨天的 Memory Extraction 流程,只替換其中最不穩定的一段:

Prompt 約定文字格式
        ↓
Python 手動解析

改成:

Pydantic Schema
        ↓
Structured Output
        ↓
Python Object

一、我們真正需要的不是一段「像 JSON 的文字」

假設我希望模型輸出:

{
    "should_remember": true,
    "memories": [
        {
            "content": "使用者的英文程度是 B1。"
        }
    ]
}

只在 Prompt 裡寫:

Please return JSON.

並不代表每次都能得到完全相同的結構。

模型可能輸出:

Here is the JSON:

再附上一段JSON;也可能漏掉:

should_remember

或把:

memories

從List變成單一字串。

即使輸出是合法JSON:

{
    "memory": "使用者的英文程度是 B1。"
}

它仍然不符合程式期待的欄位。

因此要分清楚三個層級:

方法 能保證什麼
Prompt 要求 JSON 模型盡量照做,但格式仍可能改變
JSON Mode 輸出是合法 JSON,但不保證符合指定 Schema
Structured Output 輸出遵守指定的 JSON Schema

OpenAI Docs 說明,Structured Outputs 會讓模型輸出符合提供的 JSON Schema;Python SDK 也可以直接使用 Pydantic 定義資料結構。OpenAI Structured Outputs 文件


二、先定義 Memory 的資料結構

今天先建立最小的 Memory Schema。在程式最上方加入:

from pydantic import BaseModel, Field

接著定義兩個 Class:

class MemoryCandidate(BaseModel):
    content: str = Field(
        description=(
            "A short, self-contained fact about the user "
            "that may be useful in future conversations."
        )
    )


class MemoryExtractionResult(BaseModel):
    should_remember: bool = Field(
        description=(
            "Whether the user message contains information "
            "worth remembering."
        )
    )

    memories: list[MemoryCandidate] = Field(
        description=(
            "The memories extracted from the user message. "
            "Return an empty list when should_remember is false."
        )
    )

這就是目前 Memory Extractor 必須遵守的結構。

最外層是:

MemoryExtractionResult

包含:

should_remember
memories

memories 裡的每一筆資料都是:

MemoryCandidate

目前只包含一個欄位:

content

我們先不加入:

Memory Type
Importance Score
Timestamp
Embedding

因為這些內容會在後面的章節逐步處理。今天只需要建立一個穩定、可由程式操作的基本資料單位。


三、沒有 Memory 時也要有固定結構

Day 12 沒有抽取到 Memory 時,模型會回傳:

NONE

今天不再需要特殊字串。

如果使用者輸入:

請解釋現在完成式。

Structured Output 應該回傳:

{
    "should_remember": false,
    "memories": []
}

如果使用者輸入:

我的英文程度是 B1。

則回傳:

{
    "should_remember": true,
    "memories": [
        {
            "content": "使用者的英文程度是 B1。"
        }
    ]
}

不論有沒有 Memory,程式收到的資料都有相同欄位。因此不再需要判斷:

if raw_output.upper() == "NONE":

也不需要逐行尋找:

MEMORY:

只要檢查:

result.should_remember

以及:

result.memories

就可以了。


四、修改 Memory Extraction Instructions

Day 12 的 Instructions 花了一部分篇幅要求輸出格式:

Return exactly:

NONE

or:

MEMORY: <memory>

今天格式已經交給 Structured Output 管理,所以可以移除這些文字。

將原本的 MEMORY_EXTRACTION_INSTRUCTIONS 改成:

MEMORY_EXTRACTION_INSTRUCTIONS = """
You extract potential long-term memories from a user message
for a personal English learning assistant.

Extract only information that may be useful in future
conversations, such as:
- The user's English level
- Long-term learning goals
- Stable learning preferences
- Recurring learning difficulties
- Relevant learning constraints

Do not extract:
- Greetings or casual conversation
- One-time requests
- Questions that only depend on the current context
- Information created by the assistant
- Uncertain assumptions
- Passwords, secret codes, or unnecessary sensitive data

Rewrite each memory as a short, self-contained statement.

If the message contains useful information:
- Set should_remember to true.
- Add each independent memory to memories.

If the message contains no useful information:
- Set should_remember to false.
- Return an empty memories list.

Do not invent information that the user did not state.
"""

這份 Instructions 現在只負責定義:

什麼內容值得抽取

至於:

輸出有哪些欄位
每個欄位是什麼型別
memories 是不是 List

則由 Pydantic Schema 負責。


五、替換昨天的 extract_memory_candidates()

Day 12 使用:

client.responses.create()

取得純文字,再自己解析。

今天把它替換成:

client.responses.parse()

完整 Function 如下:

def extract_memory_candidates(user_input):
    response = client.responses.parse(
        model=MODEL,
        input=[
            {
                "role": "system",
                "content": MEMORY_EXTRACTION_INSTRUCTIONS
            },
            {
                "role": "user",
                "content": user_input
            }
        ],
        text_format=MemoryExtractionResult
    )

    result = response.output_parsed

    if result is None:
        raise ValueError(
            "Memory Extraction 沒有產生可解析的結果。"
        )

    return result, response.usage.total_tokens

這次最重要的改變是:

text_format=MemoryExtractionResult

Python SDK 會根據這個 Pydantic Model 建立對應的 Schema,並把成功解析的結果放進:

response.output_parsed

所以 result 不再是一段字串,而是一個:

MemoryExtractionResult

我們可以直接使用:

result.should_remember

也可以取得:

result.memories

每一筆 Memory 也已經是:

MemoryCandidate

不需要再呼叫:

splitlines()

或自己移除:

MEMORY:

六、看看 output_parsed 長什麼樣子

假設使用者輸入:

我的英文程度是 B1,我想加強旅遊英文。

取得的 result 概念上會是:

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

因此可以直接寫:

if result.should_remember:
    for candidate in result.memories:
        print(candidate.content)

輸出:

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

如果需要轉回一般 Python Dictionary,可以使用:

result.model_dump()

得到:

{
    "should_remember": True,
    "memories": [
        {
            "content": "使用者的英文程度是 B1。"
        },
        {
            "content": "使用者想加強旅遊英文。"
        }
    ]
}

如果要查看 JSON,則可以使用:

result.model_dump_json(indent=2)

這會在後面把 Memory 寫進儲存系統時派上用場。


七、memory_candidates 也要一起修改

Day 12 的 memory_candidates 是:

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

裡面保存的是普通字串。

今天要改成:

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

初始化的程式不用改:

memory_candidates = []

但手動 remember 指令不能再直接加入字串。

將 Day 12 的:

memory_candidates.append(candidate)

改成:

manual_candidate = MemoryCandidate(
    content=candidate
)

memory_candidates.append(
    manual_candidate
)

查看 Memory 時,原本是:

print(f"{index}. {candidate}")

現在要改成:

print(
    f"{index}. {candidate.content}"
)

這樣手動與自動建立的 Memory Candidate 都會使用同一種資料結構。


八、修改自動抽取的程式

Day 12 原本會取得:

extracted_candidates
extraction_tokens
last_extraction_output

今天改成先取得完整的 Pydantic Object:

extraction_result, extraction_tokens = (
    extract_memory_candidates(user_input)
)

接著判斷:

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

    memory_candidates.extend(
        extracted_candidates
    )

最後可以把解析結果轉成 JSON,留給 Debug Command:

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

完整修改如下:

extracted_candidates = []
extraction_tokens = 0

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

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

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

            memory_candidates.extend(
                extracted_candidates
            )

        memory.add_token_usage(
            extraction_tokens
        )

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

顯示新 Memory 時也要改成:

if extracted_candidates:
    print("\n--- New Memory Candidates ---")

    for candidate in extracted_candidates:
        print("-", candidate.content)

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

九、Memora v0.10 的主程式

Day 10 的 ShortTermMemory 和 Context Management 保持不變。

Day 13 需要修改的部分是:

Import
Memory Schema
Extraction Instructions
extract_memory_candidates()
Memory Candidate 的讀寫方式

以下是更新後,從建立 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
)

memory_candidates = []
last_extraction_output = ""

print("Memora v0.10")
print(
    "Commands: history, context, summary, status, "
    "remember <text>, memories, extraction, exit"
)

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

    if not user_input:
        continue

    command = user_input.lower()
    should_auto_extract = True

    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}"
                )

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

    if command == "extraction":
        print("\n--- Last Extraction Output ---")
        print(last_extraction_output or "(not run)")
        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("---------------------")
        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
        )

        memory_candidates.append(
            manual_candidate
        )

        print(
            "Saved as Memory Candidate:",
            manual_candidate.content
        )

        user_input = candidate_text
        should_auto_extract = False
        last_extraction_output = (
            manual_candidate.model_dump_json(
                indent=2
            )
        )

    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 = []
    extraction_tokens = 0

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

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

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

                memory_candidates.extend(
                    extracted_candidates
                )

            memory.add_token_usage(
                extraction_tokens
            )

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

    print("Memora:", assistant_reply)

    if extracted_candidates:
        print("\n--- New Memory Candidates ---")

        for candidate in extracted_candidates:
            print("-", candidate.content)

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

    print("\n--- Memory Status ---")
    print(
        "Newly summarized messages:",
        memory_stats["newly_summarized_count"]
    )
    print(
        "Context messages sent:",
        len(memory_stats["context_messages"])
    )
    print(
        "Counted input tokens:",
        memory_stats["input_tokens"]
    )
    print(
        "Actual input tokens:",
        response.usage.input_tokens
    )
    print(
        "Output tokens:",
        response.usage.output_tokens
    )
    print(
        "Summary update tokens:",
        memory_stats["summary_token_usage"]
    )
    print(
        "Memory extraction tokens:",
        extraction_tokens
    )
    print(
        "Memory candidates:",
        len(memory_candidates)
    )
    print(
        "Session total tokens:",
        memory.session_total_tokens
    )
    print("---------------------")

如果環境中的 OpenAI SDK 或 Pydantic 版本較舊,可以先更新:

pip install -U openai pydantic

十、實際測試 Structured Output

輸入:

You:
我的英文程度是 B1,我希望三個月後 TOEIC 可以考到 850 分。

Memora 會正常回答,同時 Extractor 可能建立:

--- New Memory Candidates ---
- 使用者的英文程度是 B1。
- 使用者希望三個月後 TOEIC 達到 850 分。
-----------------------------

輸入:

extraction

可以查看:

{
  "should_remember": true,
  "memories": [
    {
      "content": "使用者的英文程度是 B1。"
    },
    {
      "content": "使用者希望三個月後 TOEIC 達到 850 分。"
    }
  ]
}

如果下一句是:

You:
請先給我五個單字。

Extractor 應該得到:

{
  "should_remember": false,
  "memories": []
}

程式不需要處理:

NONE

也不需要猜測模型使用了哪一種項目符號。


十一、Structured Output 解決了什麼?

Day 12 的程式依賴:

raw_output.splitlines()

以及:

line.startswith("MEMORY:")

Day 13 改成:

response.output_parsed

因此解決的是:

欄位名稱不固定
資料型別不固定
List 結構不固定
模型加入額外說明
手動解析文字容易失敗

現在 Application 可以明確知道:

result.should_remember

一定是 Boolean,而:

result.memories

是一個由 MemoryCandidate 組成的 List。

這讓 Memory 從「一段需要猜測格式的文字」,變成「程式可以直接操作的資料」。


十二、Structured Output 沒有解決什麼?

Structured Output 保證的是:

資料符合指定結構。

它不保證:

模型對內容的判斷一定正確。

例如使用者說:

我今天比較累,只想練習五分鐘。

模型仍可能錯誤抽取:

{
  "should_remember": true,
  "memories": [
    {
      "content": "使用者偏好每天只練習五分鐘。"
    }
  ]
}

這份資料的 Schema 完全正確,但內容把暫時狀態誤認成長期偏好。

此外,現在仍然沒有處理:

Memory 重複
Memory 更新
Memory 衝突
Memory 是否已經過期

因此可以把兩種問題分開:

Structured Output
解決資料格式是否可靠

Memory Engineering
處理資料內容是否值得相信與保存

後面的 Memory Policy、Importance、Decay 與 Contradiction,才會繼續處理第二類問題。


十三、現在可以存進 Database 了嗎?

從資料格式來看,現在已經比純文字更適合儲存。

例如:

candidate.model_dump()

可以得到:

{
    "content": "使用者的英文程度是 B1。"
}

這筆資料可以很容易轉成:

JSON
Database Record
API Payload

但是今天還不會立刻建立 Database。

因為下一步還要先解決:

當 Memory 越來越多時,怎麼找到和目前問題最相關的那一筆?

如果只依賴完全相同的文字搜尋:

英文程度

可能找不到:

使用者目前大約具備 B1 程度。

雖然兩句話的字不同,意思卻非常接近。

這就是下一篇要處理的問題。


Day 13 小結

昨天 Memora 的 Memory Extractor 會產生:

MEMORY: 使用者的英文程度是 B1。

Python 再靠字串處理找出其中內容。

今天我們加入:

MemoryCandidate
MemoryExtractionResult

並把:

client.responses.create()

改成:

client.responses.parse()

讓 Extractor 直接回傳:

response.output_parsed

目前的資料流程變成:

User Message
      ↓
Memory Extractor
      ↓
Structured Output
      ↓
MemoryExtractionResult
      ↓
MemoryCandidate List

因此 Memory Candidate 不再只是普通字串,而是具有明確 Schema 的 Python Object。

這還不是完整的 Long-term Memory Store,但資料已經從:

可以閱讀的文字

前進到:

可以驗證、處理與儲存的資料

Day 14|Embedding:AI 怎麼知道兩段記憶「很像」?

下一篇,我們會拿目前的:

candidate.content

例如:

使用者想加強旅遊英文。

轉換成一串數值向量。

接著比較:

我想練習機場和飯店會話。

和:

使用者想加強旅遊英文。

看看即使兩段文字沒有使用完全相同的詞,AI為什麼仍然可以判斷它們在語意上很接近。

Memora 將第一次從保存文字走向:

理解不同 Memory 之間的語意關係。


上一篇
Day 12|讓 LLM 自動抽取 Memory:從對話找到重要資訊
系列文
從 Stateless LLM 到 Agentic Memory:30 天打造會記憶的 AI Agent13
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言