我們昨天介紹了 State pattern, 今天來看一下該怎麼將 state pattern 應用在 LLM Agent，並且有什麼好處吧。

將 State pattern 套到 LLM Agent

以 LLM Agent 中的經典，ReAct 來說好了，我們可以將其拆解為3個不同的 state 如下
State 分成 3 大部分，Reasoning / Action / Answer

1. Reasoning:
   思考的狀態，起始的 state 沒意外的話也是 reasoning state，他主要有兩種行為:

   1. 需要使用工具 -> 選好工具，並且指定下一個 state 為 action state
   2. 不需要使用工具 -> 代表可直接回答，指定下一個 state 為 answer state

2. Action
   呼叫工具，並且更新 memoryEnv

3. Answer
   根據 memoryEnv 統整的資料進行回答

好處？

由於我們是用 State pattern 實作，如果之後要加入更複雜的處理 e.g., 把 reflection 與 continuous thinking 帶進來，我們不必使用一堆 if else 來做，可以利用 state pattern 的特性，對某些特定的 state 做修改即可

先 focus 在 ReAct 的 implementation

State implementation

由於我們用到了很多 pattern，先抽了一個 State 的 ABC (Abstract Base Class)

class State(ABC):
    async def run(self, env: MemoryEnv) -> AsyncGenerator[str]:
        async for chunk in self._run(env):
            yield chunk

        await self._transition(env)

    @abstractmethod
    async def _run(self, env: MemoryEnv) -> AsyncGenerator[str]: ...

    @abstractmethod
    async def _transition(self, env: MemoryEnv) -> None: ...

並且繼承此 State 的都有兩個 abstractmethod 需要實作

1. _run (input: MemoryEnv, output: AsyncGenerator[str])
   目的: 此 State 被呼叫時的實際 behavior
   input 的 MemoryEnv 為目前 env 的狀態，主要包含歷史的資訊，與目前狀態 ; 之後再詳細講 memory env 的設計
   output 用 AsyncGenerator[str]，這是為了讓 agent-brain 能隨時透過 streaming output 的方式，告知大家目前進度

2. _transition
   目的: 在這裡設計此 state 結束後該往哪個 state 的 transition

reasoning

class ReasoningState(State):
    def __init__(self) -> None:
        self.counter = 0

    async def _run(self, env: MemoryEnv) -> AsyncGenerator[str]:
        # TODO: implement thinking and parse action logic
        yield f"reasoning {self.counter}"

    async def _transition(self, env: MemoryEnv) -> None:
        self.counter += 1
        if self.counter >= 2:
            env.set_state(StateType.ANSWER)
        else:
            env.set_state(StateType.ACTION)

reasoning 的 _run 預計會 call llm 並且 parse action
transition 的部分:
1. 當 LLM 有 predict action: 那就往 action state 轉換
2. 沒有 predict action 或者超過 loop limite，走入 answer state

action

class ActionState(State):
    def __init__(self) -> None:
        self.counter = 0

    async def _run(self, env: MemoryEnv) -> AsyncGenerator[str]:
        yield f"action {self.counter}"

    async def _transition(self, env: MemoryEnv) -> None:
        self.counter += 1
        env.set_state(StateType.REASONING)

action 的 _run 會執行 reasoning state 想呼叫的 action
transition 的部分 由於目前沒有 reflection state，所以直接回到 reasoning state

answer

class AnswerState(State):
    async def _run(self, env: MemoryEnv) -> AsyncGenerator[str]:
        yield "answer"

    async def _transition(self, env: MemoryEnv) -> None:
        env.done = True

answer 的 _run 會根據(參考) memory env 來進行最後的回答
transition 的部分直接結束

如果要做到 man in a loop 的話，可能要多紀錄一下這個 state

明天來繼續完成 env 與 agent-brain 的設計