2026 iThome 鐵人賽

DAY 27

Software Development

手刻 AI Agent！大一新生的 Python 實戰筆記系列第 27 篇

Day 27 - 上下文用量 - 下 + 中斷模型回答

18th鐵人賽 #ai #agent #code #python

leematthewcat

2026-10-09 23:51:17

12 瀏覽

分享至

今天，先來處理昨天做了一半的上下文用量，然後，來處理回答時 ctrl + c 終止模型回答的小功能。
馬上開始吧！

ollama 回傳的準確 token

還記得嗎？昨天有提到，我們會透過 ollama 來得到「準確」的 token 量。

取得

ollama 的 ollama.chat() 所回傳的 ChatResponse 物件中，有 prompt_eval_count 和 eval_count 兩個屬性可用來取得 token 用量。
前者是「輸入的 token 數」，這裡包含的是 system prompt + 此次呼叫 chat() 之前輸入的所有對話（使用者先前問題、模型調用工具、模型回答等） + 此次輸入到 ollama.chat() 的內容；
後者則是此次「模型生成答覆所消耗的 token 數」。

而在把 ollama.chat() 設為 stream=True 時，prompt_eval_count 和 eval_count 只在回傳最後一個 chunk 時輸出，透過 done 屬性即可判斷是否為最後一個 chunk。

# src/meowgent/providers/ollama_provider.py
...
class OllamaProvider(LLMProvider):
	def stream_generate(...) -> ...:
		...
		for chunk in response:
			...
			
			token = None
            if chunk.done: # 輸出完成時

                prompt_eval = getattr(chunk, "prompt_eval_count", 0) or 0
                eval_cnt = getattr(chunk, "eval_count", 0) or 0

                if prompt_eval or eval_cnt:
                    token = prompt_eval + eval_cnt
                    # 輸入的 token（含歷史）+ 輸出的 token

這裡用 getattr() 來取屬性，是為了防止在 ollama SDK 早期版本此屬性不存在。

而已經有 getattr() 的預設值了，為什麼後面還要 or 0 呢？
當模型生成中斷（記憶體不足、連線斷開、ctrl + c 強制終止）時，依然會輸出 done = True，而這時 prompt_eval_count 和 eval_count 還是 None。

計算出了 token 量，那要如何回傳呢？
我們為 stream_generate() 的回傳（StreamChunk）加一個屬性 - token：

# src/meowgent/providers/base.py
...
@dataclass
class StreamChunk:
    ...
    token: Optional[int] = None
    
...

這樣就可以回傳了！

# src/meowgent/providers/ollama_provider.py
...
class OllamaProvider(LLMProvider):
	def stream_generate(...) -> ...:
		...
		for chunk in response:
			...
			
			if chunk.done:
				...
				
			yield StreamChunk(..., token=token)

接收

在 Agent() 的 chat() 中，要接收回傳的 token：
把拿到的準確 token 寫入 self.true_token，並記錄此次呼叫所生效的 system prompt 至 self.last_system_prompt。

這是因為有了 self.true_token 後，下次呼叫 get_context_status_text() 就會進入 if self.true_token: 分支，並進行新舊提示詞的比對。

# src/meowgent/agent.py
...
class Agent():
	...
	def chat(...) -> ...:
		...
		while ...:
			...
			for chunk in response:
				if chunk.token:
                    self.true_token = chunk.token

                    self.last_system_prompt = temp_history_messages[0]["content"]
                    # 記下當前輪次的提示詞（放在索引 0）
                
                if chunk.thinking_chunk:
	                ...

這麼一來，get_context_status_text() 就可以拿到準確的 token 了！

接入到渲染

渲染的部分，靠的是 prompt() 裡的 rprompt 參數來做出在輸入框最右方的效果，方法大致如下：

... .prompt(..., rprompt="要顯示的字串")

但是，這樣單純只傳入字串的方式有一個限制，只有在畫面剛顯示出來當下才做更新，沒有辦法達到我想要的「即時」效果，所以又要用到「回呼函式」的概念了！
讓 get_input() 接收 rprompt 參數，且為「可呼叫物件」，
在內部定義的回呼函式中，取得 PromptSession 當前輸入緩衝區的文字（default_buffer.text），並傳給 rprompt 執行與回傳，這就相當於把當前文字即時交給了 get_context_status_text()：

# src/meowgent/cli/input_prompt.py
...
def get_input(rprompt: Optional[Callable] = None) -> ...:
	def _get_image_toolbar_text():
		...
	def _on_text_change(...):
		...
		
	def _dynamic_rprompt():
        """ 回呼函式，用來即時取得上下文佔用文字 """
        
        if not rprompt:
            return ""
            # 防止輸入時沒填 rprompt -> 直接不啟用顯示用量的功能
        
        return rprompt(_prompt_session.default_buffer.text)
        # 取得當前輸入（buffer 內）傳入 rprompt() 回傳

    ...
    user_input = _prompt_session.prompt("> ", rprompt=_dynamic_rprompt).strip()

這麼做，只要我們一打字更新，default_buffer.text 就會收到新文字，就會被更新上去。
邏輯大致如下：

使用者打字，文字進到緩衝區（buffer）。
buffer 內容改變，觸發重繪。

重繪包含了渲染出使用者打出的內容，以及 rprompt 的內容。

回呼被調用，buffer 的文字進到 get_context_status_text()，取得要顯示的內容。
透過 prompt() 顯示出內容。
這部分邏輯不斷重複，直到使用者按下 Enter 後定格。

但到這裡還沒結束，我們還要在 main.py 呼叫 get_input() 時把計算函式傳入：

# src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	try:
		while ...:
		
			user_input, images = get_input(rprompt=model.get_context_status_text)
			
			...

這麼一來就大功告成啦～快去試試有沒有效果吧！

捕捉模型意外中斷

在向 AI 提問時，送出後卻發現打錯字或想臨時補充，相信大家一定遇過這種情況；但到目前為止，我們還沒有辦法主動終止模型的回答。現在，我們就來處理這件事！
這邊，我們透過 ctrl + c 會觸發的 KeyboardInterrupt 來達成攔截。

首先，先了解一下例外的向上傳遞（也稱作冒泡）機制：
當報錯時，直譯器會先看當前的函式有沒有寫 try...except 把它抓下來，如果沒有抓，Python 就會立刻中斷該函式，退回給呼叫它的上一層函式。如果上一層函式同樣沒有攔截，就會繼續往上退，直到某一層有人把它接住為止。
而當底層向上傳遞到上一層時，我們會說「底層被穿透」，也就是冒泡的動作。

看一下整體拋出例外傳遞的過程：

 main.py               (最外層：有寫 except KeyboardInterrupt)
    ↑
 agent.py              (中間層：生成文字階段沒有抓 KeyboardInterrupt)
    ↑
 ollama_provider.py    (底層：只有一般的 try...except，沒有抓 KeyboardInterrupt)
    ↑
 網路套件 (httpx)       (正在等 ollama 送資料過來)

httpx 處完全沒有攔截，直接穿透。
來到 ollama_provider.py 中，也沒有任何攔截又穿透了。
來到 agent.py，在這邊會出現下面兩種情況：
1. 工具已經發起並正在執行的階段：在「工具正在執行」的 for t, tool_v in tools_result: 被攔截並進行處理，處理完後主動拋出例外，形成「穿透」。
2. 模型思考或文字生成的串流階段，中斷訊號直接「穿透」agent.py。
來到最頂層的 main.py 中進行終端動態更新並清理顯示暫存（回答時，還處在 Live 渲染，還未更新到 Console 的，把它固化），然後做出終止提示，continue 到下一輪迴圈讓使用者繼續輸入。

也就是說，只有第三點中的第一點，以及第四點要做例外的攔截，先來做前者：

補上工具調用

想像一個情況，你叫 agent 幫你改寫兩份文件，它同時調用了兩個工具來分別編輯文件，第一個先完成了，但第二個還沒好時，你按下 ctrl + c 想中斷。
這時如果不特別修改，會出現兩個問題：

模型預期發出了 <tool_call> 就要收到 <tool_response>，若沒補上回應，可能會被判定結構不完整而報錯。
模型沒有實體感官，它對外部環境的認知完全來自對話歷史。如果工具執行到一半被中斷，但沒有留下任何紀錄，模型可能會誤以為工具調用已經完成，或以為系統出錯忽略了它的工具調用而在下次對話時再次做調用。

所以我們要在被中斷處做「例外處理」：

這邊要注意，真正執行工具的不是在 if tools_to_execute: 區塊，它只是把任務丟進執行緒池（Thread Pool），它「不會等待工具執行完成」，而是在幾微秒內立刻回傳一個 Future 物件。
真正停下來等待工具執行的地方，是在 if tools_result: 裡的 tool_v.result()，所以攔截要做在此處。

首先，先設 interrupted = False，下面如果捕捉到例外，就要改為 True，
而這時會進到 if interrupted: 裡頭，這裡處理的就是上面第一點的部分，做出「成對的工具標籤」。

# src/meowgent/agent.py
...
class Agent():
	...
	def chat(...) -> ...:
		...
		while ...:
			...
			
			if tools_result: # 表示有 tool use 需求

                interrupted = False # 初始化判斷條件
                
                self.history_messages.append(...) # 模型在調用前說的話
                
                for t, tool_v in tools_result:
	                # t 為 ToolCall
	                # tool_v 為 Future 物件（執行中任務）或 str（被拒絕/出錯的提示字串）
	                
	                if interrupted: # 前面工具已取消 -> 後面工具也取消
                        self.history_messages.append({
                            "role": "user",
                            "content": f"<tool_response>\n[工具 {t.tool_name} 執行結果]：\n[系統提示] 前序操作已被使用者手動中止，此工具已取消執行。\n</tool_response>"
                        })
                        continue

然後接住 KeyboardInterrupt 例外：

# src/meowgent/agent.py
...
class Agent():
	...
	def chat(...) -> ...:
		...
		while ...:
			...
			if ...:
				...
				for t, tool_v in tools_result:
					...
					try:
                        ... # 正常的工具調用情況

                    except KeyboardInterrupt:

                        interrupted = True

                        result = "[系統提示] 工具執行已被使用者手動中止（KeyboardInterrupt）。"
                        
                        is_success = False
                    
                    # 下面這邊照舊
                    self.history_messages.append(...)

                    status = ...
                    yield LLMResponse(...)

然後在迴圈結束後（把所有還沒結束的工具結果寫上因前面取消，所以也取消的訊息），要做上面說的第二點，最後「重新拋出例外」，穿透 agent.py：

不重拋例外的話，程式會跟沒事一樣繼續在 while 迴圈裡調用模型。

# src/meowgent/agent.py
...
class Agent():
	...
	def chat(...) -> ...:
		...
		while ...:
			...
			if ...:
				...
				for t, tool_v in tools_result:
					...
				if interrupted:
                    self.history_messages.append({
                        "role": "assistant",
                        "content": "已停止執行後續操作。"
                    }) # 上一個訊息是 role 為 user 的 
                    # [系統提示] 工具執行已被使用者手動中止（KeyboardInterrupt）
                    # 防止接下來的使用者輸入跟它角色重疊

                    raise KeyboardInterrupt # 處理完了，重新拋出例外，否則會繼續下一個 while 迴圈

顯示清理以及刪除多輪殘留

顯示清理的部分剛剛稍微有提過了，而「多輪殘留」是怎麼回事？
當模型沒有做任何工具調用時，也就是它不用記得它曾經改過 ... 之類的，就可以直接當作這個回答不存在，把它刪掉。

而這部分用 history_messages 的長度來做比較，先記錄下呼叫模型前的長度：

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

            history_checkpoint = len(model.history_messages) # 記錄下當前數量

首先 clean() 需要做一些改動：

原本 live.update("") 是放在 if self.cache.strip() 裡，但在推理階段沒有 self.cache，這會導致推理內容被重複。

因為中斷時沒先清空動態畫布，導致中斷訊息插隊把未完成的思考文字推成了歷史紀錄，隨後畫面元件退出時又把最後一幀文字自動補印了一次。

加入隔線控制，因為我要在顯示出「已中斷當前操作」才畫出隔線。
其他都照舊：

# src/meowgent/cli/renderers.py
...
class ResponseStreamer:
	def clean(self, live: Live, rule: bool = True):
        """ 清除最後留在 live 的內容（轉為 console.print()）"""

        live.update("")

        if self.cache.strip():

            render_items = [Padding(Markdown(self.cache), (0, 0, 0, 2))]
            if rule:
                render_items.append(self.render.get_rule())

            self.console.print(Group(*render_items))

        self.cache = ""
        self.last_len = 0

另外，CLIRenderer 還需要實作 render_keyboard_interrupt_end()：

# src/meowgent/cli/renderers.py
class CLIRenderer:
	...
	
	def render_keyboard_interrupt_end(self):
        return Padding("[yellow]已中斷當前操作[/yellow]", (0, 0, 0, 2))

然後來捕捉例外，在 with Live(...) 內把生成迴圈包進 try，
例外捕捉內先執行 clean()，要注意傳入 rule=False。
接著判斷若使用者只傳入問題而尚未完成任何工具調用，則透過串列切片將該對話回滾刪除。
最後印出隔線然後 continue 就大功告成啦！

# src/meowgent/cli/main.py
...
if __name__ == "__main__":
	...
	try:
        while True:
	        ...
	        
			with Live(...) as live:
				try:
                    for stream_content ...:
	                    ...
	                    
	            except KeyboardInterrupt:

                    response_streamer.clean(live=live, rule=False) 

                    # 若中斷時連一輪工具都沒完成（只多了一則 user 訊息），將其回滾抹除
                    if len(model.history_messages) == history_checkpoint + 1:
                        model.history_messages = model.history_messages[:history_checkpoint]

                    cli.console.print(cli.render_keyboard_interrupt_end())

                    cli.console.print(cli.get_rule())

                    continue # 重新等待下一次使用者輸入

今天內容也是好多啊，完成了上下文用量的顯示，還順手做了 ctrl + c 中斷處理的部分。
明天，終於可以正式進入到上下文壓縮的部分了。
（本來想說把上下文顯示寫完就好，但我覺得好像有點少，寫著寫著又寫到一萬字啦，太扯了）