在這篇文章中，我們要處理一個較為棘手的任務：從流式輸出的模型回覆（chunk.content_chunk）中，精準「分離出工具調用請求」。
你可能覺得很奇怪，回答就回答，怎麼還混進了工具調用呢？
如果是最單純的調用方式（如下所示），直接交由 Ollama SDK 處理，它會自動幫我們完成工具分離的動作：
response = ollama.chat(
tools=tools_list,
...
)
然而，當模型生成那一長串工具調用的 JSON 格式時，SDK 沒辦法把內容流式的交到 response 裡，就導致了一旦要調用工具時，畫面都會有數秒甚至數十秒的停頓，最後才顯示出工具已被調用。
工具需要的參數越長，停頓也就會越長，像是要請模型寫出一篇冗長的文件時，畫面就會卡住非常久。
所以麻煩了點，我們自行處理「分離工具調用」、「工具調用解析」兩個部分，今天來搞定前者！
我們要來製作一個函數，傳入模型輸出的串流內容，解析並輸出「模型回答內容」、「還未完整的工具調用」、「已完整的工具調用」。
這裡的完不完整指的是是否出現了
</tool_call>，還沒出現前表示還沒輸出完成。
回答內容跟工具分離可能還好理解，那為什麼工具還要分是否完整呢？
前面有說過，在輸出工具調用時，我們會對其進行顯示，而若已經完成輸出了，就不需要進行顯示了，要進入下方的邏輯進行工具的解析。這就是需要分流的原因。
我們在 Agent() 類別外建立函數：
這個函數接收目前為止累積的模型全文（full_text），並透過迴圈與底層演算法加持的 find()，在全文中搜尋是否出現 <tool_call>
這裡的演算法是 Fastsearch ，它非常厲害，不需要把每個字都遍歷，就能搜尋到目標索引。
搜索方式是用「跳的」，從目標樣式的最後一個字元比對，具體過程如以下解釋：
目標樣式為<tool_call>，長度 11，
- 起始索引為 0，樣式涵蓋索引 0 ~ 10，看到索引 10 為「要」，而不是「>」，往後跳 11 格（新起始索引為 0 + 11 = 11）。
- 起始索引 11，涵蓋索引 11 ~ 21，索引 21 依然不是「>」，而是「c」，但是 c 「有出現在樣式中」，對應到的「樣式內索引」為 6，
演算法將這兩個 c 做對齊，再移動 10 - 6 = 4 格，新索引來到 11 + 4 =15- 起始索引 15，涵蓋 15 ~ 25，索引 25 剛好是「>」，找到後從尾部依序向前核對目標樣式和文本是否完整對齊。
- 確認完整對齊，得到答案為目標起始索引為 15。
文本：關 於 這 個 問 題 ，這 裡 需 要 調 用 工 具 < t o o l _ c a l l > 索引： 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 樣式：< t o o l _ c a l l > 索引：0 1 2 3 4 5 6 7 8 9 10
# src/meowgent/agent.py
from ...
from typing import Tuple, Optional, List
def _extract_safe_text(
full_text: str,
start_tag: str = "<tool_call>",
end_tag: str = "</tool_call>"
) -> Tuple[str, Optional[str], List[str]]:
safe_parts = [] # 已經提取出的模型回覆
completed_tools = [] # 已完整的工具調用請求
current_tool_text = None # 尚未完整的工具調用請求
start_pos = 0 # 目前進行到的索引（此輪的起點索引）
full_len = len(full_text) # 目前全文長
while start_pos < full_len: # 遍歷完全文
start_target_idx = full_text.find(start_tag, start_pos)
...
這裡的
start_pos是為了要在「前一個工具已經完整閉環，接著繼續輸出了第二個工具的情況」時從第一個工具結束後開始搜尋的。
那你可能會想，剛剛搜尋過的內容，再搜一遍不是很浪費效能嗎？
這部分，我請 AI 幫我跑了測試，目前專案內的搜尋模式，在 2,452 個字元，內有三個<tool_call>的情況下，單次僅需 0.003175 毫秒，換算下來每秒可以處理 7 億個字。
所以，從頭搜尋一遍根本不成問題，而且還能大幅精簡代碼，
如果要記錄上次搜到哪，還要多一個記錄的參數傳入函數，並且必須寫一大堆複雜的邏輯去記錄：「上次是不是剛好停在<to中間？這次新來的字是不是ol_call>？」。
沒有找到 <tool_call> 時會有兩種情況：「真的沒有工具調用」跟「疑似有工具調用」。
疑似的部分指的是只出現例如
"<tool_ca"這樣寫到一半的標籤，而真的沒有就是在遍歷完後結尾處連<都沒有。
首先，利用 full_text[start_pos:] 切出此次調用多出來的內容，trimmed_len 做計數，
我們著重來看一下 range() 的部分，這裡我們要搜尋的目標是如下面的形式：
起點 "<tool_call" -> i = 10 -> len(start_tag) - 1 = 11 - 1
| "<tool_cal"
| "<tool_ca"
| ...
| "<t"
終點 "<" ---------> i = 1
這就是為什麼起點是 len(start_tag) - 1。
這裡的語法為
range(起點, 終點, 步長)，而步長可以理解為是：每輪迴圈結束後，要對起點 + 步長，
所以如果我們把起點終點對調（起點大終點小），步長為負數，可以達到從後向前遍歷的效果。
而在遍歷的過程，我們用 endswith() 檢查結尾是否為目標值。
因為切片終點不包含尾部，剛好
start_tag[:i]能切出該輪迴圈要尋找的不完整標籤。
接下來就分為有找到和沒找到兩種狀況了：
1. 一開始：
remaining = "你好<tool"
trimmed_len = len(remaining) = 7 (總共有 7 個字)
2. 迴圈開始檢查：
當 i = 5 時，start_tag[:5] 是 "<tool"
-> remaining.endswith("<tool") 比對成功！(命中碎片，長度 i = 5)
3. 執行 trimmed_len -= i：
trimmed_len = 7 - 5 = 2 (扣掉這 5 個字，只剩下 2 個字的安全長度！)
4. 最終切片輸出：
remaining[:trimmed_len] ──> remaining[:2]
-> 剛好只取出前 2 個字："你好"！
-> 尾巴的 "<tool" (5 個字) 成功被砍掉，完全不會洩漏到畫面上！
trimmed_len 維持原長度，切片會將剩餘文字完整保留並加入安全回覆列表。最後，把這兩種情況切片切出來的值保存到 safe_parts 串列中。
# src/meowgent/agent.py
...
def _extract_safe_text(...) -> ...:
...
while start_pos < full_len:
...
if start_target_idx == -1:
# 後續沒有完整的 start_tag，檢查剩餘文字尾端是否有正在成形的前綴（如 "<tool"）
remaining = full_text[start_pos:] # 切出新內容
trimmed_len = len(remaining) # 用於下方迴圈計數
for i in range(len(start_tag) - 1, 0, -1):
if remaining.endswith(start_tag[:i]):
trimmed_len -= i
break # 退出 for
safe_parts.append(remaining[:trimmed_len])
break # 退出 while
...
找到標籤後，我們要做的就是分離回答以及工具了，切出文字內容，放入 safe_parts 串列中：
# src/meowgent/agent.py
...
def _extract_safe_text(...) -> ...:
...
while start_pos < full_len:
...
if start_target_idx == -1:
...
else:
# 收集 <tool_call> 之前的安全文字
safe_parts.append(full_text[start_pos:start_target_idx])
...
為了提高搜尋效率，我們將結束標籤 </tool_call> 的搜尋起點設在 <tool_call> 之後（略過起始標籤本身，避免無效比對）**：
<tool_call>
{ -> 此為索引 0
"name": "read_file",
"arguments": {"file_path": "src/meowgent/agent.py"}
}
</tool_call>
同樣分成了有搜尋到跟沒有兩種情況：
current_tool_text。completed_tools，並且更新起始索引，繼續 while 迴圈搜尋是否有第二個工具調用。而要注意的是，我們需要的僅是 {...}，不需要 <tool_call> 標籤：
{
"name": "read_file",
"arguments": {"file_path": "src/meowgent/agent.py"}
}
最後，回傳拼接完的字串、未結束和已結束的工具調用。
# src/meowgent/agent.py
...
def _extract_safe_text(...) -> ...:
...
while start_pos < full_len:
...
else:
...
# 尋找對應的 </tool_call>
content_start = start_target_idx + len(start_tag)
end_idx = full_text.find(end_tag, content_start)
if end_idx == -1:
# 工具調用尚未閉合
current_tool_text = full_text[content_start:].strip()
break
else:
completed_tools.append(full_text[content_start:end_idx].strip())
start_pos = end_idx + len(end_tag) # 更新索引
return "".join(safe_parts), current_tool_text, completed_tools
...
搞定了這個函數，我們就可以來將其接入 chat() 中了！
把未完成的工具調用和模型回答分別回傳，而完成的工具我們下一篇會對其進行處理。
# src/meowgent/agent.py
...
class Agent():
...
def chat(...) -> Iterator[LLMResponse]:
...
while turns < self.max_turns:
...
for chunk in response:
...
if chunk.content_chunk:
...
result_full_text ...
# 分離
safe_text, tool_text, completed_tools = _extract_safe_text(
result_full_text
)
if tool_text is not None:
# 還未完成的工具調用參數
yield LLMResponse(status="tool_calling", content=tool_text)
elif safe_text:
# 模型回答
yield LLMResponse(status="response", content=safe_text)
這裡
completed_tools在下一篇用到。
我們辛辛苦苦的把 completed_tools 給提取出來了，但這可還沒結束呢！
下一篇，要來把這個 JSON 格式（用字串來表示的）的工具調用給解析出來，判斷出是要調用哪個工具函數、要傳入什麼參數、是否需要審核，等等的。
iThome鐵人賽
看影片追技術