iT邦幫忙

2026 iThome 鐵人賽

DAY 14
0

哇!我已經更新了兩個禮拜了,為我自己鼓掌一下哈哈。
在先前的 13 篇,我們陸續完成了**「模型提供」、「工具調用」、「模型調度」、「system prompt 注入」、「輸入」、「渲染」、「ollama 管理」這七大項目**,今天這篇,我們終於要來把它們給組裝起來了,也就是要來做最重要的 main.py 啦!


初始化

初始化部分,首先我們先建立 Agent 物件,建立前我們先回顧一下 Agent 類別:

# src/meowgent/agent.py
...
class Agent():
    def __init__(self, provider: LLMProvider, max_turns: int = 20):
	    ...

要傳入的有 LLMProvider 物件,以及最大工具調用次數(這裡直接採用預設),接著回顧一下 OllamaProvider 物件:

# src/meowgent/providers/ollama_provider.py
...
class OllamaProvider(LLMProvider):
    def __init__(self, model_name: str):
        ...

這邊要傳入的則是模型名稱,由於我們還沒做模型選擇的邏輯,這裡直接指定一個就好,所以我們的 Agent 物件 - model 這樣建立:

#src/meowgent/cli/main.py
from agent import Agent
from providers import OllamaProvider
from cli import get_input, get_tool_approval, CLIRenderer, start_ollama, clean_ollama
import sys

if __name__ == "__main__":
    
    model = Agent(OllamaProvider(model_name="qwen-3.8"))

然後,要來建立渲染器物件,看一下 CLIRenderer 類別:

# src/meowgent/cli/renderers.py
...
class CLIRenderer:
	def __init__(self):
		...

太好啦!不需要傳入任何東西,建立好渲染器物件以後我們來用 initialization() 初始化介面:

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	cli = CLIRenderer() 

    cli.initialization() # 初始介面

再來要來處理 ollama 的進程:
這裡上一篇才說過,應該還蠻熟悉的,start_ollama() 回傳了是否有被啟用的布林值以及可選的 Popen 物件。

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	success, process = start_ollama() # 啟動 ollama 進程

    if not success:
        cli.console.print(cli.render_end("ollama 未被正常啟動"))

        sys.exit(1) # 異常退出

接收到的狀態,我們判斷如果進程未啟用,調用 render_end() 來顯示結束渲染,並用 sys.exit(),傳入的「 1 」指的是「異常結束」。

如果是 sys.exit(0) 則是「正常結束」,而其實,不是只有 1 代表異常,異常的代碼非常廣泛,從 1 ~ 255 都是,而其中 1 ~ 63 屬於我們開發者可以自行定義的,其他則是有對應的錯誤的,有興趣可以自己去查查,這裡就不細說了。

而這個結束代碼是在做什麼的呢?
這個代碼事實上不是給人看,而是給電腦看的,舉個例子,還記得上一篇我們有講到「終端語法」吧!
但我們說的是管道 - |,這裡來介紹另外兩個 - && 和 ||,前者是「前面的指令成功才執行後面的」,而後者則相反,是「前面的指令失敗才執行後面的」。
舉例來說:

uv run python -m cli.main && echo "任務圓滿完成!"

就是在返回代碼 0 時在終端輸出(echo 就相當於 print())。

uv run python -m cli.main || echo "警告:意外終止!"

這個則是在輸出代碼 1 時輸出。


工具審核

看到 chat() 的定義:

# src/meowgent/agent.py
...
def chat(
	self,
	user_input: str,
	tool_approval: Callable[[str, dict], bool]
) -> Iterator[LLMResponse]:
	...

要求要傳入一個「函數」,函數中要傳入字串(工具名)和字典(參數)的,回傳布林值(審核結果),而這個函數就是用來呼叫審核渲染(get_tool_approval)並回傳審核結果的。
我們來實作這個函數吧!

首先,先把 Live 清空(分隔線的部分),接著需要暫停 Live 的渲染。

因為 Live 和 Questionary 在同一時間只能有一人接管螢幕,所以必須「暫時」讓出 Live 的控制權。

然後就可以呼叫 get_tool_approval() 獲取審核結果,最後重啟 Live 並回傳結果。

而為什麼這裡要用 try...finally 呢?
這裡先來說 finally 是什麼,它跟 try...except 語法是類似的,在功能上,它保證了在任何情況下一定會執行 finally 縮排內的代碼,也就是說,就算是程式崩潰了,也會把清理的部分給跑完。

這是因為,在 Questionary 問問題的過程中,可以 Ctrl - C 來終止問答(預設直接拒絕調用),而這樣會觸發 KeyboardInterrupt 例外,如果沒用 finally,而是直接把重啟 Live 放在下面,它就會被直接跳過,也就是 Live 沒有被重新打開,就無法繼續正常渲染了。

而關於 Ctrl - C 終止問答的部分,這邊也深入講一下,Ctrl - C 的捕捉是在 .ask() 內部的,而它會直接回傳 None,也就是 approval 收到的是 None,但沒關係,後面再判定時,會被當作 False 看。

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	    def ask_tool_approval(tool_name: str, tool_args: dict) -> bool:
        live.update("") # 清除分隔線,否則調用許可會出現在分隔線下方
        live.stop() # 停止 live 更新

        try:
            approval = get_tool_approval(tool_name, tool_args) # 呼叫取得輸入

        finally:

            live.start() # 重啟 live 更新

        return approval

進入對話迴圈

進程關閉

進入迴圈前還記得上一篇除了用來啟動進程的 start_ollama() 還做了 clean_ollama() 嗎?
想想看 clean_ollama() 要放在哪裡呢?沒錯,就是最後面!直覺上可能會想要這樣放:

...

... = start_ollama() # 這裡啟動了程序

... # 中間的渲染部分

clean_ollama(process) # 清理

但這會出現一個問題,如果運行途中,程式突然被停止了呢(例如手動 ctrl-C 或者是意外崩潰)不就清不到了嗎,也就形成了之前說過的「孤兒進程」。
這時我們就又要用到 try...finally 了。

所以整個對話迴圈都會被包在 try 底下,而清理在 finally 下:

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	try:
		# 等等要實作的渲染部分
		
	finally:
		clean_ollama(process) # 清理

輸入對話及 Live 物件的建立

關於對話的迴圈,使用的是 while True: 這樣簡單的迴圈,並沒有在迴圈上做結束程式的控制。

結束的控制我們會在下一篇用「指令」的方式來實現。

進入迴圈後,第一件事就是「獲取用戶的輸入」,調用的是先前寫好的 get_input(),並用 strip() 把輸入清理乾淨,然後我們需要做一個邏輯判斷,防止空白的輸入也傳入到模型中浪費算力。

strip() 用於清除頭尾的空格、換行符號、製表符(做表格用的)。

接著還要建立 ResponseStreamer 物件,用的是 get_response_streamer()。

這裡建立的 ResponseStreamer 物件我們不在「初始化」的部分做,是因為 ResponseStreamer 類別中涉及到了 self.cache 及 self.last_len 的「重置」問題。
雖然每次對話結束後都會進行 clean() 來做重置,但是建立物件的耗時是非常短的,對效能也不影響,所以這裡我更偏向於每輪迴圈都重新建立一個物件,如果未來我們在這部分加入了其他功能(例如模型回覆中途 Ctrl-C 停止模型輸出)也不用對原有邏輯做太多改動,是「防禦性編程」的展現。

當然,如果你真的想放在初始化的部分建立物件,對於「現階段」來說是沒有問題的!

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	try:

		 while True: # 對話迴圈

            user_input = get_input().strip()

            if not user_input:
                continue # 空的 -> 跳過此輪
            
            response_streamer = cli.get_response_streamer()
    
    finally:
	    ...

建立 ResponseStreamer 物件的時機放在判斷空輸入後面,可以省得根本沒有輸入,白白建立了物件。

接下來,透過 with 來建立 Live 物件:

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	try:

		 while True:
			...
			
			with Live(
				console=cli.self.console,
				refresh_per_second=10,
				vertical_overflow="crop"
			) as live:
	
	finally:
		...

刷新率設定每秒 10 次(如果你的電腦非常強,每秒可以產出 10 個以上的 chunk 當然可以調高點),而之前有聊過,溢出時直接裁切,不用 Live 渲染,結束後用 Console。


分類做渲染

來回顧一下,Agent 物件的 chat() 回傳的資料類別 - LLMResponse:

# src/meowgent/providers/base.py
...
@dataclass
class LLMResponse:
    status: Literal[
	    "response",
	    "thinking",
	    "thinking_done",
	    "tool_calling",
	    "tool_executed",
	    "tool_rejected"
    ]
    ...

六種狀態分別對應到了 update_content()、render_single_line_streamer()、render_thinking_summary()、render_single_line_streamer()、render_tool_approval_result()("tool_executed" 和 "tool_rejected" 共用)。

其中 update_content() 需透過 get_response_streamer() 建立的 ResponseStreamer 物件 - response_streamer 呼叫,也就是我們剛才上面做過的 response_streamer = cli.get_response_streamer(),其餘的都是直接透過 cli 物件呼叫。

調用 Agent 類別的 chat() 來獲取模型輸出的 chunk,依照以上六種狀態對應到的函數做分配,最後記得 clean():

#src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	
	try:
		 while True:
			...
			
			with Live ...:
			
				for stream_content in model.chat(
					user_input=user_input,
					tool_approval=ask_tool_approval
				):
                    # ask_tool_approval() 的執行權在 agent.py 上
					
					# 輸出模型回答內容
					if stream_content.status == "response":
                        response_streamer.update_content(
	                        full_text=stream_content.content,
	                        live=live
                        )
                        
                    # 輸出推理內容    
                    elif stream_content.status == "thinking":
                        live.update(
	                        cli.render_single_line_streamer(stream_content.content)
                        )
                        
					# 清除推理內容,輸出推理總結
                    elif stream_content.status == "thinking_done":
                        live.update("") # 清除推理內容及隔線
                        cli.console.print(
	                        cli.render_thinking_summary(stream_content.think_time)
                        )
                        
                    # 輸出工具參數生成跑馬燈    
                    elif stream_content.status == "tool_calling":
                        live.update(
	                        cli.render_single_line_streamer(stream_content.content)
                        )

                    elif stream_content.status == "tool_executed": # 輸出工具調用成功
                        response_streamer.reset(live=live) # 歸零長度記帳,文字不被吞掉
                        cli.console.print(
	                        cli.render_tool_approval_result(
		                        stream_content.tool_name,
		                        True
	                        )
                        )
                        
					# 輸出工具調用失敗
                    elif stream_content.status == "tool_rejected":
                        response_streamer.reset(live=live) # 歸零長度記帳,文字不被吞掉
                        cli.console.print(
	                        cli.render_tool_approval_result(
		                        stream_content.tool_name,
		                        False
	                        )
                        )

                response_streamer.clean(live=live)
	finally:
		...

看到比較特別的地方是 "thinking_done" 時,我們要先清除一下 Live 版面的內容(推理內容)。

還有是在 "tool_executed" 和 "tool_rejected" 要重置紀錄(因為輸出的工具內容會被算進 self.last_len 記進 self.cache)。


太棒啦!到這裡,我們終於可以正式地讓程式跑起來啦~
不過,這離一個真正好用的 Agent 還有一段距離,下一篇我們來做幾個簡單的「指令功能」。


上一篇
Day 13 - ollama 進程的管理
下一篇
Day 15 - 輸入指令來改東西
系列文
手刻 AI Agent!大一新生的 Python 實戰筆記 共 15 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言