昨天我從 top-down 的角度把 Memory Structure 與 Thinking Net Structure 拆開，今天回到實作面，先完成 Memory 的最小語義介面，並提供第一個實作：MessagesMemory。

Memory 的核心理念

在我的設計裡，所有狀態都落在 Memory：包含目標（goal）、對話歷程（history）、當前可用工具清單、上一步的 Action 結果、以及流程是否已完成等旗標。
State 本身是 Stateless，只讀寫 Memory；Controller（例如 Brain）負責決定何時呼叫哪個 State、何時轉移、何時串流輸出。

這樣帶來三個好處：

1. 可預期：轉移權集中在 Controller，不會在 State 內偷偷改狀態（避免隱性轉移與 reentrancy 地雷）。
2. 可測試：State 可以用假 Memory 打單元測試；Memory 也能獨立測。
3. 可觀測：所有重要資訊都在 Memory，便於加上 metrics、trace、dump、replay。

最小語義介面（Minimal Interface）

今天先定義 Memory 的最小語義介面，滿足 ReAct/Reflexion 這類最常見的 Agent 流程；之後隨著其他 Net（如 Tree-of-Thought、Graph-of-Thought）再增補。

from abc import ABC, abstractmethod
from typing import Any

class Memory(ABC):
    def __init__(self, tools: list["BaseTool"]) -> None:
        self.done: bool = False
        self.next_action: "Action | None" = None
        self.available_tools: dict[str, "BaseTool"] = {tool.name: tool for tool in tools}

    def list_tools(self) -> str:
        return ",".join([str(tool.to_dict()) for tool in self.available_tools.values()])

    def get_tool(self, name: str) -> "BaseTool | None":
        return self.available_tools.get(name)

    # --- 語義介面（async 的理由見下文） ---
    @abstractmethod
    async def set_goal(self, goal: str) -> None: ...

    @abstractmethod
    async def update(self, messages: list["Message"]) -> None: ...

    @abstractmethod
    async def goal(self) -> list["Message"]: ...

    @abstractmethod
    async def dump(self) -> list["Message"]: ...

為何這 4 個方法？

• set_goal(goal: str)：將任務目標標準化放入 Memory；很多 Net 在第一步就需要它（prompt 編排、規劃）。
• update(messages: list[Message])：同步地把 State 的輸出、Tool 的結果等追加到 Memory。
• goal() -> list[Message]：以訊息格式取回目標，方便 prompt 組裝（比直接回字串更一致）。
• dump() -> list[Message]：吐出「目前可序列化的工作記錄」，便於觀測與回放（replay）。

為什麼要 async？
這些操作未來可能落到 I/O（如把 long-term 資料寫到 DB / vector store、或跨程序 pipelines）。若一開始就把介面定為 sync，之後改 async 會造成破壞性變更；反之先用 async，短期內在記憶體操作也沒負擔。

MessagesMemory：最暴力也最實用的第一步

第一個 Memory 實作是訊息列表型：萬用、直觀，也最容易做對照基準（baseline）。

class MessagesMemory(Memory):
    def __init__(self, tools: list["BaseTool"]) -> None:
        self._history: list["Message"] = []
        self._goal: "Message | None" = None
        super().__init__(tools)

    async def set_goal(self, goal: str) -> None:
        self._goal = Message(role=Role.USER, content=goal)

    async def update(self, messages: list["Message"]) -> None:
        self._history.extend(messages)

    async def goal(self) -> list["Message"]:
        return [self._goal] if self._goal else []

    async def dump(self) -> list["Message"]:
        return self._history

設計重點

• 保持極簡：沒有任何額外 indexing 或裁切策略（之後可加 sliding window/TTL）。
• 統一資料型別：Message 代表一切（user/assistant/tool），讓 Net 組 prompt 時只面對一種型別。
• 工具可觀測：list_tools() 把工具清單序列化，能直接塞入 system prompt 或 debug log。

** 加上 memory 的 states**

action state

class ActionState(State):
    async def run(self, memory: "Memory") -> AsyncIterator[str]:
        if not memory.next_action:
            return

        if tool := memory.get_tool(memory.next_action.name):
            result = await tool.execute(**memory.next_action.args)
            await memory.update([Message(role=Role.ACT, content=str(result))])
        yield "\n"

    async def next_state(self, memory: "Memory") -> Enum:
        return ReAct.REASONING

answer state

class AnswerState(State):
    async def run(self, memory: "Memory") -> AsyncIterator[str]:
        generated_response = ""
        messages = await get_messages(memory)
        async for chunk in llm.stream_response(
            model_name=ANSERING_MODEL, messages=messages
        ):
            if chunk:
                generated_response += chunk
                yield chunk

        await memory.update([Message(role=Role.ASSISTANT, content=generated_response)])

    async def next_state(self, memory: "Memory") -> Enum:
        memory.done = True
        return ReAct.ANSWER

reasoning state

class ReasoningState(State):
    async def run(self, memory: "Memory") -> AsyncIterator[str]:
        generated_response = ""
        messages = await get_messages(memory)
        async for chunk in stream_response(
            model_name=REASONING_MODEL, messages=messages
        ):
            if chunk:
                generated_response += chunk
                yield chunk

        await memory.update([Message(role=Role.ASSISTANT, content=generated_response)])

        if action := parse_json_response(generated_response):
            memory.next_action = Action(**action)
        else:
            memory.next_action = None

    async def next_state(self, memory: "Memory") -> Enum:
        if memory.next_action:
            return ReAct.ACTION
        return ReAct.ANSWER

介面設計抉擇與未來擴充

為什麼 list_tools() 與 get_tool() 在 Memory？

• 可觀測：工具清單本身就是 Prompt 的一部分（system/tools block），也常用於 debug。
• 可替換：未來 Memory 可以決定依任務不同提供不同工具子集合（白名單/ACL）。

明天來介紹設計 brain 的抽象介面