iT邦幫忙

2026 iThome 鐵人賽

DAY 10
0

昨天我們替 Memora 加上 Conversation Summary。當較舊的訊息離開 Sliding Window 時,程式不再直接丟掉它們,而是把重要內容壓縮成摘要,再和最近幾輪對話一起放進下一次 Request。

到目前為止,我們已經有:

  • 完整保存的 conversation_history
  • 最近幾輪的原始訊息
  • 較舊對話的 conversation_summary
  • Input Token Budget
  • 每次真正送進模型的 Context

今天不再加入新的 Memory 技術,而是把這些功能整理成一個完整的 Short-term Memory 元件,完成 Chapter 2 的 Stateful Chatbot。


一、什麼叫做「Short-term Memory 完成」?

這裡的完成是指:

在同一次程式執行、同一段 Conversation 裡,Memora 能保存對話狀態,並在有限的 Context Budget 中維持多輪對話。

它需要做到幾件事:

保存完整 Conversation History

保留最近幾輪原始訊息

把較舊訊息壓縮成 Summary

控制下一次 Request 的 Input Tokens

讓每次 Request 都能取得需要的 Context

這些能力前幾天已經分別完成。

今天要解決的是另一個問題:目前和記憶有關的變數與函式散落在整支程式裡。

例如 Day 9 有:

conversation_history = []
conversation_summary = ""
summarized_message_count = 0
last_context_messages = []
session_total_tokens = 0

也有:

count_input_tokens()
update_summary()
create_context_messages()
prepare_context()

它們其實都在處理同一件事情:

管理目前這段 Conversation 的 Short-term Memory。

所以今天要把它們整理到同一個元件裡。


二、建立 ShortTermMemory

我們新增一個 Class:

class ShortTermMemory:
    ...

這不是替 Memora 加入新的記憶能力,也不是把 Python List 換成某種特殊的 Memory Framework。

它只是把昨天已經完成的功能整理在一起。

程式的對應關係如下:

Day 9 Day 10
conversation_history memory.history
conversation_summary memory.summary
summarized_message_count memory.summarized_message_count
last_context_messages memory.last_context
session_total_tokens memory.session_total_tokens
prepare_context() memory.prepare_context()

整理之後,Chatbot 主程式只需要負責:

  1. 接收 User Input。
  2. 交給 Short-term Memory。
  3. 取得這一輪要送出的 Context。
  4. 呼叫 LLM。
  5. 把 Assistant Reply 存回 Memory。

至於哪些訊息要保留、哪些要摘要,以及目前使用多少 Input Tokens,都由 ShortTermMemory 處理。


三、每一輪對話的執行順序

整理之後,每一輪會經過以下流程:

接收 User Message
        ↓
加入完整 History
        ↓
建立這一輪的 Context
        ↓
必要時更新 Summary
        ↓
檢查 Input Token Budget
        ↓
呼叫 LLM
        ↓
保存 Assistant Message
        ↓
等待下一輪

這裡有兩份資料要繼續分清楚:

history

保存這次 Conversation 的完整訊息。

context_messages

代表這一輪實際提供給模型的內容。

完整 History 可以持續累積,但不代表每一次都要全部送進模型。


四、完整的 Memora v0.7

以下程式直接從 Day 9 的版本整理而來。

原本的 Sliding Window、Conversation Summary、Token Counting 和 Debug Command 都還在,只是被收進 ShortTermMemory

from openai import OpenAI

client = OpenAI()

MODEL = "gpt-5-mini"

SYSTEM_PROMPT = """
You are Memora, a personal English learning assistant.

Your goal is to help the user learn English clearly and efficiently.

Guidelines:
- Explain concepts in simple language.
- Keep answers focused on the user's question.
- Use short examples when helpful.
- Avoid unnecessary technical grammar terminology.
"""

SUMMARY_INSTRUCTIONS = """
You maintain a compact summary of an ongoing conversation.

Requirements:
- Merge the existing summary with the new messages.
- Preserve user facts, preferences, goals, corrections,
  decisions, and unresolved questions.
- Remove greetings, repetition, and unimportant wording.
- Do not invent information.
- If newer information corrects older information,
  keep the newer information.
- Keep the summary under 150 words.
- Use the same language as the conversation when possible.
- Return only the updated summary.
"""

# 這是 Memora 自己設定的 Input Budget,
# 不是模型真正的最大 Context Window。
MAX_INPUT_TOKENS = 1000

# 優先保留最近三輪的原始 User / Assistant Message。
MAX_RECENT_TURNS = 3


class ShortTermMemory:
    def __init__(
        self,
        client,
        model,
        system_prompt,
        summary_instructions,
        max_input_tokens,
        max_recent_turns
    ):
        self.client = client
        self.model = model
        self.system_prompt = system_prompt
        self.summary_instructions = summary_instructions
        self.max_input_tokens = max_input_tokens
        self.max_recent_turns = max_recent_turns

        self.history = []
        self.summary = ""
        self.summarized_message_count = 0
        self.last_context = []
        self.session_total_tokens = 0

    def add_user_message(self, content):
        self.history.append(
            {
                "role": "user",
                "content": content
            }
        )

    def rollback_last_user_message(self):
        if self.history and self.history[-1]["role"] == "user":
            self.history.pop()

    def format_messages(self, messages):
        return "\n".join(
            f"{message['role']}: {message['content']}"
            for message in messages
        )

    def count_input_tokens(self, messages):
        count = self.client.responses.input_tokens.count(
            model=self.model,
            instructions=self.system_prompt,
            input=messages
        )

        return count.input_tokens

    def update_summary(self, new_messages):
        existing_summary = self.summary or "(empty)"

        summary_input = f"""
Existing summary:
{existing_summary}

New messages:
{self.format_messages(new_messages)}
"""

        response = self.client.responses.create(
            model=self.model,
            instructions=self.summary_instructions,
            input=summary_input
        )

        self.summary = response.output_text.strip()

        summary_tokens = response.usage.total_tokens
        self.session_total_tokens += summary_tokens

        return summary_tokens

    def create_context_messages(
        self,
        recent_messages,
        current_user_message
    ):
        context_messages = []

        if self.summary:
            context_messages.append(
                {
                    "role": "developer",
                    "content": (
                        "The following is a compact summary of earlier "
                        "conversation. Use it as background context. "
                        "If it conflicts with recent messages, prefer "
                        "the recent messages.\n\n"
                        + self.summary
                    )
                }
            )

        context_messages.extend(recent_messages)
        context_messages.append(current_user_message)

        return context_messages

    def prepare_context(self):
        if not self.history:
            raise ValueError("Conversation History 是空的。")

        if self.history[-1]["role"] != "user":
            raise ValueError(
                "建立 Context 前,最後一則訊息必須是 User Message。"
            )

        completed_messages = self.history[:-1]
        current_user_message = self.history[-1]

        summary_token_usage = 0
        newly_summarized_count = 0

        recent_message_limit = self.max_recent_turns * 2

        target_summarized_count = max(
            0,
            len(completed_messages) - recent_message_limit
        )

        # 先把已經離開最近訊息範圍的內容加入摘要。
        if target_summarized_count > self.summarized_message_count:
            new_messages = completed_messages[
                self.summarized_message_count:
                target_summarized_count
            ]

            summary_token_usage += self.update_summary(new_messages)
            newly_summarized_count += len(new_messages)

            self.summarized_message_count = (
                target_summarized_count
            )

        recent_messages = completed_messages[
            self.summarized_message_count:
        ]

        # 即使保留的輪數沒有超標,也要檢查實際 Token。
        while True:
            context_messages = self.create_context_messages(
                recent_messages,
                current_user_message
            )

            input_tokens = self.count_input_tokens(
                context_messages
            )

            if input_tokens <= self.max_input_tokens:
                return {
                    "context_messages": context_messages,
                    "input_tokens": input_tokens,
                    "summary_token_usage": summary_token_usage,
                    "newly_summarized_count": (
                        newly_summarized_count
                    )
                }

            # 沒有更多完整的 User / Assistant Pair 可以壓縮。
            if len(recent_messages) < 2:
                raise ValueError(
                    "目前的 User Message 與摘要已超過 "
                    "MAX_INPUT_TOKENS。"
                )

            oldest_turn = recent_messages[:2]

            summary_token_usage += self.update_summary(
                oldest_turn
            )

            newly_summarized_count += len(oldest_turn)
            self.summarized_message_count += len(oldest_turn)

            recent_messages = recent_messages[2:]

    def finish_turn(
        self,
        assistant_reply,
        context_messages,
        response_tokens
    ):
        self.history.append(
            {
                "role": "assistant",
                "content": assistant_reply
            }
        )

        self.last_context = [
            message.copy()
            for message in context_messages
        ]

        self.session_total_tokens += response_tokens

    def get_status(self):
        return {
            "History messages": len(self.history),
            "Summarized messages": (
                self.summarized_message_count
            ),
            "Last context messages": len(self.last_context),
            "Summary available": bool(self.summary),
            "Session total tokens": self.session_total_tokens
        }


def print_messages(title, messages):
    print(f"\n--- {title} ---")

    if not messages:
        print("(empty)")
    else:
        for message in messages:
            print(
                f"{message['role']}: "
                f"{message['content']}"
            )

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


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
)

print("Memora v0.7")
print("Commands: history, context, summary, status, exit")

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

    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 == "status":
        print("\n--- Short-term Memory Status ---")

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

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

    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
    )

    print("Memora:", assistant_reply)

    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(
        "Session total tokens:",
        memory.session_total_tokens
    )
    print("---------------------")

程式仍然使用 Responses API 的 Token Counting Endpoint,在送出主要 Request 前檢查 instructionsinput 的 Token 數量。官方文件也示範了 client.responses.input_tokens.count() 可以接收 Message List。OpenAI Token Counting 文件


五、prepare_context() 是 Short-term Memory 的核心

整理成 Class 之後,最重要的方法是:

memory.prepare_context()

它會依序完成:

找出已完成的對話

更新較舊訊息的 Summary

保留最近幾輪原始訊息

加入目前的 User Message

計算 Input Tokens

必要時繼續壓縮舊訊息

回傳這一輪真正要使用的 Context

因此主程式不需要知道 Summary 應該何時更新,也不需要自己操作 Sliding Window。

它只需要取得:

memory_stats["context_messages"]

然後交給模型:

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

六、為什麼要在 API 成功後才完成這一輪?

主程式先保存 User Message:

memory.add_user_message(user_input)

但 Assistant Message 必須等 API 成功後才加入:

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

如果 API 呼叫失敗,程式會執行:

memory.rollback_last_user_message()

把剛才尚未完成的 User Message 移除。

否則 History 可能變成:

User
Assistant
User
Assistant
User

最後一個 User Message 沒有對應的 Assistant Reply,下一輪處理訊息時就可能破壞原本的 User/Assistant Pair。

這是一個不太顯眼,但對 Conversation State 很重要的細節。


七、用 status 檢查 Short-term Memory

Day 10 新增了一個指令:

status

它不會呼叫模型,而是直接顯示目前的 Memory 狀態:

--- Short-term Memory Status ---
History messages: 12
Summarized messages: 6
Last context messages: 7
Summary available: True
Session total tokens: 2841
--------------------------------

這些數字代表不同的事情。

History messages

目前完整保存了多少則 User/Assistant Message。

Summarized messages

其中多少則較舊的訊息已經被合併進 Summary。

Last context messages

上一個 Request 實際送出了多少則 Message。

Summary available

目前是否已經建立 Conversation Summary。

Session total tokens

這次程式執行期間,主要回答與摘要 Request 累積使用的 Token。

因此,即使:

History messages: 40

上一個 Request 也不一定送出了 40 則訊息。

這表示完整紀錄和模型目前使用的 Context 已經被分開管理。


八、怎麼確認 Short-term Memory 真的有工作?

可以用以下方式驗收目前的 Memora。

先在對話前段提供一項資訊:

You:
我的英文程度是 B1,我最近想練習旅遊英文。

接著聊幾輪其他內容,直到早期訊息被移入 Summary,再問:

You:
請根據我的程度和目標安排一個練習。

如果 Summary 正確保留了先前資訊,Memora 應該能產生適合 B1、偏向旅遊情境的內容。

接著分別輸入:

history

確認完整對話仍然存在。

輸入:

summary

確認較舊資訊已經被壓縮。

輸入:

context

確認真正送入上一個 Request 的不是完整 History,而是:

Conversation Summary
+
Recent Messages
+
Current User Message

最後輸入:

status

確認 Input Context 沒有跟著完整 History 無限成長。


九、現在為什麼可以叫做 Stateful Chatbot?

OpenAI 官方文件把「將額外 Messages 放進後續 Request」列為手動管理 Conversation State 的方式。也就是說,即使個別 Request 可以互相獨立,Application 仍然能透過保存並重新提供資料,建立多輪對話。OpenAI Conversation State 文件

目前 Memora 每完成一輪,都會更新:

Full History
Conversation Summary
Summarized Message Position
Last Request Context
Token Usage

下一輪 Request 的內容,會受到這些既有 State 影響。

因此現在的架構不再是:

Current User Message
        ↓
       LLM

而是由 Application 根據目前狀態建立 Context:

Short-term Memory
        +
Current User Message
        ↓
Context Management
        ↓
       LLM

所以從 Application 的角度來看,現在的 Memora 已經是:

Stateful Chatbot

Class 本身並沒有讓程式變得 Stateful;真正讓它 Stateful 的,是其中保存的資料會跨 Request 更新,並影響下一輪輸入。


十、但這還不是 Long-term Memory

現在把 Python 程式關掉:

exit

重新執行後:

memory = ShortTermMemory(...)

裡面的狀態仍然會重新開始:

history = []
summary = ""

所以目前的記憶範圍仍然是:

同一次 Python Process
+
同一段 Conversation

此外,Conversation Summary 的用途是壓縮目前這段聊天,不是挑選值得跨 Conversation 保存的資訊。

即使我們現在直接把完整 History 存進 Database,也只能先稱為:

Conversation Persistence

因為「資料被保存」不等於「系統知道哪些內容值得記住,也知道未來何時應該取回」。

這正是 Chapter 3 要開始處理的問題。


Day 10 小結

Chapter 2 到這裡完成了。

今天沒有加入新的 Memory 演算法,而是把前幾天完成的功能整理成:

ShortTermMemory

目前它負責:

保存完整 Conversation History

維護 Conversation Summary

保留最近幾輪原始訊息

建立下一次 Request 的 Context

控制 Input Token Budget

記錄目前 Session 的 Token Usage

Memora 的演進也來到:

Day 06
Conversation History

Day 07
Token 與 Context Limit

Day 08
Sliding Window

Day 09
Conversation Summary

Day 10
Short-term Memory

現在它已經不只是把所有訊息塞回模型,而是開始由 Application 主動管理:

這一輪的模型到底需要看到什麼?

但目前所有資料仍然只服務於「這次對話」。

下一步,我們要第一次離開單一 Conversation,開始思考一個完全不同的問題:

一段對話裡,哪些事情值得讓 AI 在未來繼續記得?

Day 11|什麼事情值得 AI 記住?從聊天紀錄到 Memory

下一篇不會立刻把全部 Conversation History 存進 Database。

我們會先區分:

聊天紀錄

和:

真正值得保存的 Memory

因為如果 AI 把使用者說過的每一句話都當成長期記憶,它得到的很可能不是更好的 Memory,而是一個越來越混亂的資料庫。


上一篇
Day 09|AI 也要做筆記:用 Summarization 壓縮對話
下一篇
Day 11|什麼事情值得 AI 記住?從聊天紀錄到 Memory
系列文
從 Stateless LLM 到 Agentic Memory:30 天打造會記憶的 AI Agent12
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言