AG Day 6 手刻 Agent 迴圈:工具執行與多輪對話
執行需求:CPU+API key。在上一篇 AG Day 5(原文連結)中,我們透過 JSON Schema 與 Python 型別反射實作了標準的 Function calling(函式呼叫)註冊與派發架構,並驗證了單次回合的模型工具呼叫與結果回傳。然而,真實世界的研究任務絕不可能僅靠單一工具呼叫就能完成。當使用者提出「請針對 2026 年邊緣晶片架構進行產業研調,比較其功耗、算力與散熱瓶頸」時,代理必須自主決定先查詢知識庫、根據查到的線索再發起數值計算、若有資料缺口再進行二次檢索,最後綜合所有事實產出結構化的研究報告。今天,我們要在不依賴任何外部重型框架的前提下,使用純 Python 親手刻劃出一個具備完整生命週期的「自主代理迴圈(Agent Loop)」。本篇將深入拆解狀態維護、終止條件判定、最大步數熔斷機制,並支援 --dry-run 離線多輪模擬,讓大家徹底看清 Agent 運作的本質底層。
引言
在當前 AI 開發社群中,充斥著各式各樣高層次的 Agent 框架。許多初學者直接匯入框架封裝好的物件,點選執行後看到終端機跑出五花八門的日誌,便以為自己掌握了 Agent 工程。但這種「黑盒子」學習法極為危險:一旦進入真實生產環境,面對模型陷入重複呼叫、狀態記憶體異常膨脹、或是網路中斷導致流程卡死等棘手問題時,缺乏底層認知的人往往束手無策。
所有複雜的 Agent 框架,無論外表包裝得如何華麗,其靈魂核心永遠是一個「狀態驅動的執行迴圈(State-driven Execution Loop)」。這個迴圈的運作模式可以用「感知(Perception)— 推理決策(Reasoning)— 工具行動(Action)— 觀察回饋(Observation)— 狀態更新(State Update)」五大步驟來概括。只要目標尚未達成且步數未超標,迴圈就會反覆推進。今天我們將以純 Python 3.13 親手搭建出這個迴圈,同時將執行事件即時寫入我們在 Day 2 建立的 SQLite 知識庫資料表,為後續導入狀態圖框架建立最堅實的認知基礎。
原理/觀念
自主代理迴圈(Agent Loop)的五階段狀態機
在軟體架構中,手刻 Agent 迴圈本質上是一個有限狀態機(Finite State Machine, FSM)。迴圈的每一輪迭代(Step)都依循嚴謹的狀態流轉:
- 感知階段(Perceive):讀取當前累積的對話歷史與環境狀態,在傳入模型前透過 Day 3 實作的滑動視窗機制進行安全性修剪,確保在不破壞工具呼叫原子完整性的前提下控制 Token 總量。
- 推理階段(Reason):將修剪後的訊息陣列與可用的工具 JSON Schema 傳遞給大型語言模型,由模型根據上下文判斷當前進度並產出下一步決策。
- 分支判定(Branching):
- 分支 A(觸發工具):若模型輸出的回覆中包含
tool_calls,代表任務需要外部數據或計算支援,狀態機轉向行動階段。 - 分支 B(達成目標):若模型沒有輸出任何工具呼叫,而是直接產出完整的自然語言結論,代表模型判定研究任務已完成,狀態機標記
is_finished = True並退出迴圈。
- 分支 A(觸發工具):若模型輸出的回覆中包含
- 行動階段(Act):解析模型傳入的工具名稱與 JSON 引數,交由 Day 5 建立的
ToolRegistry進行執行時期型別檢查與本地函式派發。 - 觀察階段(Observe):將本地函式執行的回傳字串包裝為
role="tool"的訊息,連同上一輪模型的assistant決策訊息一併追加到狀態歷史中,完成一個完整的推理閉環。
最大步數熔斷機制(Circuit Breaker)的重要性
在開放式自動化推理中,大型語言模型存在著不可預測的非確定性。如果工具回傳的結果始終無法滿足模型的預期,或者模型對某個參數產生理解偏差,它很容易陷入重複呼叫同一個工具的「震盪迴圈(Oscillation Loop)」。
若缺乏安全熔斷防護,這種失控的迴圈不僅會無休止地消耗 API 費用(費用計量請依各廠商官方文件為準),更會導致伺服器執行緒被長期佔用。因此,在任何工業級的代理迴圈中,我們都必須設定由環境變數 RESEARCH_AGENT_MAX_STEPS 嚴格管制的計步器。一旦 step_count >= max_steps,無論任務是否完成,系統都必須強制熔斷跳出迴圈,並將異常狀態持久化儲存以供後續人工介入排查。
完整實作
接下來,我們將在 src/research_agent/agent_loop.py 中實作一個完整、健全的手刻代理迴圈。本實作串聯了我們先前打造的設定載入器、SQLite 事件儲存層、滑動修剪器以及工具註冊中心,並完整支援 --dry-run 離線多輪推理示範。
第一步:定義代理執行過程中的核心狀態結構 src/research_agent/agent_state.py:
# research-agent/src/research_agent/agent_state.py
from dataclasses import dataclass, field
from typing import List, Dict, Any
from research_agent.llm_types import ChatMessage
@dataclass
class AgentState:
"""代理執行生命週期之完整狀態容器"""
run_id: str
user_query: str
model_name: str
max_steps: int
step_count: int = 0
is_finished: bool = False
final_report: str | None = None
messages: List[ChatMessage] = field(default_factory=list)
intermediate_steps: List[Dict[str, Any]] = field(default_factory=list)
第二步:建立能支撐多輪研調推理的模擬工具集 src/research_agent/multi_step_tools.py:
# research-agent/src/research_agent/multi_step_tools.py
from research_agent.tools import ToolRegistry
research_registry = ToolRegistry()
@research_registry.register(
description="在產業研調庫中檢索特定技術的關鍵架構參數與硬體指標。"
)
def search_tech_specs(subject: str) -> str:
"""查詢架構規格"""
if "晶片" in subject or "邊緣" in subject:
return "【研調庫】:2026 年邊緣 AI 晶片普遍採用 4nm 先進製程,算力達到 120 TOPS,典型功耗為 15W。"
return f"【研調庫】:查無關於 [{subject}] 的規格紀錄。"
@research_registry.register(
description="計算運算設備的每瓦效能比(TOPS/W)與能效指標。"
)
def compute_energy_efficiency(tops: float, power_watts: float) -> str:
"""計算每瓦效能比"""
if power_watts <= 0:
return "計算錯誤:功耗必須大於零。"
efficiency = tops / power_watts
return f"運算密度分析:在 {tops} TOPS 算力與 {power_watts}W 功耗下,能效比為 {efficiency:.2f} TOPS/W。"
@research_registry.register(
description="在研調報告中標記已完成審核的重要技術事實。"
)
def mark_section_verified(section_name: str, verification_note: str) -> str:
"""標記審核狀態"""
return f"審核記錄成功:章節 [{section_name}] 已查驗,備註:{verification_note}"
第三步:實作智慧多輪模擬器,支援在 --dry-run 模式下精確呈現多輪推理的動態演進:
# research-agent/src/research_agent/multi_turn_simulator.py
import json
from typing import List, Dict, Any
from research_agent.llm_types import ChatMessage
def simulate_multi_turn_step(
messages: List[ChatMessage],
model_name: str,
step_count: int
) -> ChatMessage:
"""離線模擬多輪研調代理的思考與行動決策"""
last_msg = messages[-1]
# 步驟 1:初次接收任務,決定先查硬體規格
if step_count == 1:
return ChatMessage(
role="assistant",
content="收到研調目標。第一步:我需要先查詢邊緣晶片的算力與典型功耗數據。",
tool_calls=[{
"id": "call_step_101",
"type": "function",
"function": {
"name": "search_tech_specs",
"arguments": json.dumps({"subject": "邊緣 AI 晶片"}, ensure_ascii=False)
}
}]
)
# 步驟 2:取得規格後,決定計算每瓦效能比
elif step_count == 2:
return ChatMessage(
role="assistant",
content="已取得基礎算力(120 TOPS)與功耗(15W)指標。第二步:計算每瓦能效比。",
tool_calls=[{
"id": "call_step_102",
"type": "function",
"function": {
"name": "compute_energy_efficiency",
"arguments": json.dumps({"tops": 120.0, "power_watts": 15.0})
}
}]
)
# 步驟 3:數據計算完畢,產出最終研調結論並退出迴圈
else:
return ChatMessage(
role="assistant",
content=(
f"【研究助理研調總結報告(模型:{model_name})】:\n"
"1. 2026 年邊緣 AI 晶片採用 4nm 先進製程,算力達 120 TOPS。\n"
"2. 經精確運算,其能效比高達 8.00 TOPS/W,每瓦運算效率提升顯著。\n"
"3. 整體硬體架構在散熱與能耗間取得極佳平衡,具備高度量產可行性。"
),
tool_calls=None
)
第四步:在 src/research_agent/agent_loop.py 中編寫手刻代理迴圈的核心引擎,並包含 SQLite 步數事件持久化:
# research-agent/src/research_agent/agent_loop.py
import sqlite3
import json
from pathlib import Path
from research_agent.agent_state import AgentState
from research_agent.llm_types import ChatMessage
from research_agent.pruner import prune_chat_history
from research_agent.tools import ToolRegistry
from research_agent.multi_turn_simulator import simulate_multi_turn_step
def save_step_event(
db_path: Path,
run_id: str,
step_number: int,
event_type: str,
thought: str | None = None,
tool_name: str | None = None,
tool_args: str | None = None,
tool_result: str | None = None
) -> None:
"""將單步執行事件持久化儲存進 SQLite events 資料表"""
with sqlite3.connect(db_path) as conn:
conn.execute(
"""INSERT INTO events
(run_id, step_number, event_type, thought, tool_name, tool_args, tool_result)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(run_id, step_number, event_type, thought, tool_name, tool_args, tool_result)
)
conn.commit()
def run_autonomous_loop(
state: AgentState,
registry: ToolRegistry,
db_path: Path
) -> AgentState:
"""執行手刻 Agent 核心多輪迴圈"""
print(f"=== 啟動自主代理迴圈 [Run ID: {state.run_id}] ===")
print(f"研究目標:{state.user_query}")
print(f"配置模型變數:{state.model_name},安全步數上限:{state.max_steps}\n")
while not state.is_finished and state.step_count < state.max_steps:
state.step_count += 1
current_step = state.step_count
print(f"--- [輪次 {current_step} / {state.max_steps}] 模型思考與決策中 ---")
# 1. 歷史修剪防護(控制上下文視窗膨脹)
safe_messages = prune_chat_history(state.messages, max_turns=3)
# 2. 模型推理決策(離線模擬呼叫)
assistant_reply = simulate_multi_turn_step(
safe_messages,
state.model_name,
current_step
)
state.messages.append(assistant_reply)
# 3. 檢查是否達成終止條件(無 tool_calls 代表推理完成)
if not assistant_reply.tool_calls:
print("[決策判定]:模型未發起進一步工具呼叫,任務順利收斂!")
state.is_finished = True
state.final_report = assistant_reply.content
save_step_event(
db_path=db_path,
run_id=state.run_id,
step_number=current_step,
event_type="FINISH",
thought=assistant_reply.content
)
break
# 4. 處理工具呼叫分支
print(f"[模型思考]:{assistant_reply.content}")
for tool_call in assistant_reply.tool_calls:
call_id = tool_call["id"]
fn_name = tool_call["function"]["name"]
fn_args = tool_call["function"]["arguments"]
print(f" -> 執行工具呼叫:{fn_name},引數:{fn_args}")
# 執行本地函式
tool_output = registry.dispatch(fn_name, fn_args)
print(f" -> 工具執行回傳:{tool_output}")
# 5. 持久化事件軌跡至 SQLite
save_step_event(
db_path=db_path,
run_id=state.run_id,
step_number=current_step,
event_type="TOOL_CALL",
thought=assistant_reply.content,
tool_name=fn_name,
tool_args=fn_args,
tool_result=tool_output
)
# 6. 包裝為 tool 訊息追加回狀態歷史
state.messages.append(
ChatMessage(role="tool", content=tool_output, tool_call_id=call_id)
)
state.intermediate_steps.append({
"step": current_step,
"tool": fn_name,
"output": tool_output
})
print()
# 7. 檢查是否觸發安全步數熔斷
if not state.is_finished and state.step_count >= state.max_steps:
print("\n[安全警告]:代理執行已達最大安全步數限制,強制觸發熔斷中斷!")
state.final_report = "任務因超過最大步驟限制而終止,請檢視中間軌跡排查問題。"
save_step_event(
db_path=db_path,
run_id=state.run_id,
step_number=state.step_count,
event_type="CIRCUIT_BREAKER",
thought="達到最大步數熔斷上限"
)
return state
第五步:實作步數熔斷與防護機制驗證腳本 test_circuit_breaker.py,驗證當 max_steps 設定較小時系統是否能如期截斷:
# research-agent/test_circuit_breaker.py
from pathlib import Path
from research_agent.agent_state import AgentState
from research_agent.llm_types import ChatMessage
from research_agent.multi_step_tools import research_registry
from research_agent.storage import init_database
from research_agent.agent_loop import run_autonomous_loop
def test_breaker():
print("=== 驗證步數安全熔斷機制 ===")
test_db = Path("data/test_breaker.db")
init_database(test_db)
# 故意將 max_steps 設為 1,強制在第一輪工具呼叫後觸發熔斷
state = AgentState(
run_id="run_breaker_001",
user_query="測試超長任務是否會被安全中斷",
model_name="mock-model",
max_steps=1,
messages=[
ChatMessage(role="system", content="你是嚴格受步數控制的助理。"),
ChatMessage(role="user", content="請進行全面性深度研調。")
]
)
final_state = run_autonomous_loop(state, research_registry, test_db)
assert final_state.is_finished is False, "錯誤:超過步數卻未被標記為未完成!"
assert "超過最大步驟限制" in (final_state.final_report or "")
print("驗證通過:最大步數熔斷機制成功攔截無限迴圈!\n")
if test_db.exists():
test_db.unlink(missing_ok=True)
if __name__ == "__main__":
test_breaker()
第六步:撰寫端到端多輪研調主程式 run_agent_demo.py,完整執行三輪多步推理閉環:
# research-agent/run_agent_demo.py
from pathlib import Path
from research_agent.agent_state import AgentState
from research_agent.llm_types import ChatMessage
from research_agent.multi_step_tools import research_registry
from research_agent.storage import init_database
from research_agent.agent_loop import run_autonomous_loop
def main():
db_file = Path("data/knowledge.db")
init_database(db_file)
initial_messages = [
ChatMessage(
role="system",
content="你是由研發團隊打造的專業邊緣運算研究助理,遵循嚴謹的查證與運算流程。"
),
ChatMessage(
role="user",
content="請針對 2026 年邊緣 AI 晶片架構進行產業研調,並評估其每瓦效能指標。"
)
]
state = AgentState(
run_id="run_demo_882",
user_query="2026 年邊緣 AI 晶片架構研調與能效評估",
model_name="mock-research-agent",
max_steps=5,
messages=initial_messages
)
result_state = run_autonomous_loop(state, research_registry, db_file)
print("\n==========================================")
print(" 最終研調報告產出 ")
print("==========================================")
print(result_state.final_report)
print("==========================================")
print(f"總執行輪次:{result_state.step_count} 輪")
print(f"任務完成狀態:{'成功結束' if result_state.is_finished else '未完成'}")
if __name__ == "__main__":
main()
第七步:在終端機中執行多輪代理示範腳本:
python run_agent_demo.py
此時在終端機中將呈現完整的「感知—推理—行動—觀察」多步動態歷程(示範輸出):
=== 啟動自主代理迴圈 [Run ID: run_demo_882] ===
研究目標:2026 年邊緣 AI 晶片架構研調與能效評估
配置模型變數:mock-research-agent,安全步數上限:5
--- [輪次 1 / 5] 模型思考與決策中 ---
[模型思考]:收到研調目標。第一步:我需要先查詢邊緣晶片的算力與典型功耗數據。
-> 執行工具呼叫:search_tech_specs,引數:{"subject": "邊緣 AI 晶片"}
-> 工具執行回傳:【研調庫】:2026 年邊緣 AI 晶片普遍採用 4nm 先進製程,算力達到 120 TOPS,典型功耗為 15W。
--- [輪次 2 / 5] 模型思考與決策中 ---
[模型思考]:已取得基礎算力(120 TOPS)與功耗(15W)指標。第二步:計算每瓦能效比。
-> 執行工具呼叫:compute_energy_efficiency,引數:{"tops": 120.0, "power_watts": 15.0}
-> 工具執行回傳:運算密度分析:在 120.0 TOPS 算力與 15.0W 功耗下,能效比為 8.00 TOPS/W。
--- [輪次 3 / 5] 模型思考與決策中 ---
[決策判定]:模型未發起進一步工具呼叫,任務順利收斂!
==========================================
最終研調報告產出
==========================================
【研究助理研調總結報告(模型:mock-research-agent)】:
1. 2026 年邊緣 AI 晶片採用 4nm 先進製程,算力達 120 TOPS。
2. 經精確運算,其能效比高達 8.00 TOPS/W,每瓦運算效率提升顯著。
3. 整體硬體架構在散熱與能耗間取得極佳平衡,具備高度量產可行性。
==========================================
總執行輪次:3 輪
任務完成狀態:成功結束
常見錯誤與踩雷
在親手實作自主 Agent 迴圈時,有四個極為致命的工程地雷必須特別警戒:
- 對話歷史的就地修改(In-place Mutation)陷阱:在迴圈內部如果直接對
state.messages進行切片或 pop 操作,容易在非同步並發或異常回溯時破壞狀態的一致性。特別是當多個輔助函式共享同一個串列時,就地修改會引發難以重現的幽靈 Bug。最佳實務是在傳遞給修剪函式或模型客戶端時,傳遞淺拷貝或明確的防禦性複製品。 - 忽略模型輸出中的文字與工具混雜現象:部分模型在輸出
tool_calls的同時,仍會在文字內容(content)中夾帶思考歷程。如果程式碼直接寫成if assistant_reply.content: return,就會在模型只是在自言自語時誤判為「任務結束」,導致工具從未被執行。必須明確以tool_calls是否存在作為行動分支的第一判斷條件。 - 震盪迴圈與呼叫重複偵測:當外部工具回傳了非預期的錯誤訊息時,模型可能在下一輪以完全一模一樣的參數再次呼叫同一個工具,重複陷入死迴圈。高階的代理迴圈應當維護一個最近呼叫的雜湊指紋(Hash Fingerprint),若連續偵測到相同工具與相同參數,應主動在對話歷史中插入「警告:請勿重複傳入相同錯誤引數,請嘗試轉換思路」的系統提醒。
- 未在資料庫中記錄單步執行耗時:在生產維運中,當一個長任務耗時數分鐘時,工程師很難得知究竟是模型推理緩慢、還是某個本地工具在進行網路請求時卡住。在
events資料表中記錄每一步的啟動與結束時間戳記,是日後進行系統瓶頸分析的關鍵原料。
效能與實務提醒
在長時間多輪代理系統中,效能最佳化與資源控制至關重要:
第一,上下文視窗膨脹對延遲的指數級衝擊:在多輪對話中,隨著 messages 的長度遞增,模型每一步的首字回傳延遲(Time To First Token, TTFT)與運算成本會直線攀升。這正是為什麼我們在 Day 3 實作的 prune_chat_history() 必須在每一輪呼叫模型前強制執行,嚴格控制輸入佇列的長度。
第二,SQLite WAL 模式下的高並發寫入:在多代理協作或長時間背景執行中,頻繁將每一步的 events 寫入 SQLite。藉由在 Day 2 啟用的 WAL 模式,資料庫寫入操作不會阻塞其他模組的查詢請求,大幅提升整體吞吐表現。
第三,嚴格設定模型費用與資源預算:在開放式任務中,某些複雜題目的推演輪次難以預估。除設定最大步數限制外,實務上應搭配 Token 累計計量器,一旦累計 Token 費用突破預警門檻(計費依廠商官方文件為準),立即發出告警或降級為更輕量級的模型執行後續步驟。
小結
今天我們完成了 AI Agent 開發中最具里程碑意義的一步:以純 Python 親手刻劃出完整的自主執行迴圈。我們擺脫了任何黑盒子框架的遮蔽,建立了五階段狀態機,規範了嚴格的終止條件與最大步數安全熔斷機制,並將每一步的思考與工具回傳無縫持久化儲存進 SQLite 知識庫中。
以下整理本章節的核心概念與台灣用語對照表:
- 代理迴圈(Agent Loop):驅動模型進行多輪感知、推理、工具呼叫與觀察的自主驅動迴圈。
- 狀態機(State Machine):由明確狀態、事件與轉移規則所構成的軟體架構模型。
- 安全熔斷(Circuit Breaker):當系統步數或資源消耗超標時,主動中斷流程以防止災難性失控的保護機制。
- 終止條件(Termination Condition):判斷任務已成功收斂或需停止執行的邊界判斷規則。
- 執行軌跡(Execution Trajectory):完整記載代理歷次思考、呼叫與回饋細節的時間序列紀錄。
結語
有了手刻迴圈後,我們的研究助理已經能夠自主完成多步驟的調研工作。但在真實的網路世界中,環境是充滿不確定性的:外部 API 可能會逾時、網路連線可能隨時斷線、模型給出的參數偶爾會引發例外。如果一遇到工具拋錯整個迴圈就崩潰跳出,系統將完全無法在生產環境中生存。
明天,我們會進入「AG Day 7 錯誤處理:逾時、重試與工具失敗」,探討如何在代理迴圈中建立強大的工業級容錯防護網,涵蓋指數退避重試(Exponential Backoff)、非同步逾時控制,以及將工具失敗轉化為模型自我修復契機的高階除錯策略!
延伸資源
- LangChain 官方文件:The Agent Loop and Tool Calling Lifecycle。
- OpenAI 官方指南:Building Resilient Multi-turn Function Calling Agents。
- Python 官方文件:
concurrent.futures— Launching parallel tasks。 - SQLite 官方架構指南:Write-Ahead Logging Performance and Benchmarks。
留言
張貼留言