AG Day 14 ReAct Agent:create_react_agent 實戰
執行需求:CPU+API key。如果暫時沒有 API key,本篇文章提供完整的離線 mock 與 dry-run 模式,不需要外部連線也能在本機完整演練 ReAct 狀態流轉、工具呼叫、訊息交接與結果審查流程。
引言
在過去幾天的探索中,我們一步一腳印地建立了 LangGraph 的完整心智模型。從 AG Day 11(原文連結)的 StateGraph 與固定邊、AG Day 12(原文連結)的 TypedDict 狀態縮減器,到 AG Day 13(原文連結)的條件邊與動態路由,我們已經完全具備了親手編排任何複雜狀態圖的能力。
然而,回顧我們在 AG Day 6(原文連結)手刻的第一個自主代理迴圈,以及 AG Day 5(原文連結)的函式呼叫機制,工程師在實務中最常建構的拓撲結構,其實是由普林斯頓大學提出的經典範式——ReAct(Reasoning + Acting,推理與行動)。在 ReAct 模式下,模型觀察當前對話歷史、推理下一步行動(思考)、發起工具呼叫(行動)、取得工具執行結果(觀察),並反覆迭代直到達成目標或給予最終解答。
如果每一次要建構 ReAct 代理,我們都得從頭手寫 StateGraph、宣告 messages 鍵值、掛載工具節點、設定 tools_condition 條件邊並手動拉回線路,樣板程式碼將不可避免地大幅增加。為了解決這個最常見的高頻場景,LangGraph 在 langgraph.prebuilt 模組中提供了開箱即用的原語——create_react_agent。在本文中,我們將深入剖析 create_react_agent 的內部運作機制,為我們的貫穿專案 research-agent 搭建具備學術文獻檢索與統計計算能力的 ReAct 研究代理,並建立完整的離線降級測試架構。
ReAct 範式演進:從文字提示詞到工具狀態機
理解 ReAct 的技術演進史,有助於我們看清現代代理架構的本質。在 2022 年 Yao 等人發表 ReAct 論文的初期,大型語言模型尚未具備原生的 Function Calling 能力。當時的做法純粹依賴「提示詞模板工程」:在系統提示詞中規定模型必須輸出特定格式的文字,例如:
# 早期 Prompt-based ReAct 格式範例
Thought: 我需要先查詢 Transformer 的參數量。
Action: Search[Transformer parameter count]
Observation: 原始 Transformer 基礎模型約包含 6500 萬個參數。
Thought: 我已經取得資訊,可以回答使用者。
Final Answer: Transformer 基礎模型約有 65M 參數。
應用程式透過正則表達式解析這段文字,攔截 Action: 關鍵字並執行 Python 函式,再將 Observation: 拼接入提示詞送回模型。這種早期文字解析方案極端脆弱,模型常常拼錯動作格式或漏掉換行,導致解析器頻繁崩潰。
到了 2024 年之後,隨著各大模型供應商將工具呼叫(Tool Calling)深度內化至權杖解碼層,ReAct 範式經歷了徹底的「狀態機革命」。現代的 ReAct 不再依賴脆弱的文字解析,而是轉化為嚴謹的訊息狀態躍遷圖(Message State Transition Graph):
- 推理階段(Reasoning):模型接收包含歷史訊息的狀態,產出帶有
tool_calls結構化屬性的AIMessage。 - 行動階段(Acting):狀態機偵測到
tool_calls,自動將其路由至工具節點,並行或依序執行本機 Python 工具。 - 觀察階段(Observation):工具執行完畢後產出
ToolMessage,並將其追加回狀態清單中。 - 匯流閉環:流程無條件流回模型節點,由模型綜合先前的思考與最新的觀察資料,決定是發動下一輪工具呼叫,還是產出最終結論。
拆解 create_react_agent:預建構狀態圖的解剖學
LangGraph 的 create_react_agent 絕非一個封閉的黑盒子,它本質上是一個優雅的狀態圖工廠函式(Graph Factory)。如果我們將其內部解剖,它實際上在底層自動執行了以下標準圖構建步驟:
- 宣告狀態架構:預設採用
MessagesState,其內部核心宣告正是messages: Annotated[Sequence[BaseMessage], add_messages]。 - 建構 Agent 節點:將使用者傳入的工具清單透過
model.bind_tools(tools)綁定至語言模型,並定義一個名為"agent"的節點函式,負責呼叫模型並回傳{"messages": [response]}。 - 建構 Tools 節點:使用
ToolNode(tools)建立一個名為"tools"的節點函式,負責解析AIMessage.tool_calls、動態調度對應的 Python 函式並產出ToolMessage。 - 佈置拓撲邊界:
- 固定邊:
workflow.add_edge(START, "agent")。 - 條件邊:
workflow.add_conditional_edges("agent", tools_condition)。若最後一則訊息包含tool_calls則分岔至"tools",否則流向END。 - 閉環邊:
workflow.add_edge("tools", "agent"),將工具觀察結果送回模型進行下一輪思考。
- 固定邊:
- 編譯應用程式:最後呼叫
workflow.compile(checkpointer=...)並回傳編譯後的圖執行個體。
換言之,create_react_agent 是站在標準 StateGraph 之上的一層極薄、極度符合人體工學的高階語法糖。它產出的物件本質上就是一個 CompiledGraph,完全繼承了 LangGraph 的串流輸出、檢查點記憶與狀態檢查能力。
完整實作:為 research-agent 打造多工具 ReAct 研究代理
現在,我們進入具體工程實作。我們將在 research-agent 專案中,定義兩組專業的研究工具:學術文獻檢索工具 search_academic_database 以及統計指標計算工具 calculate_statistical_metrics,並利用 create_react_agent 打造一個能自主調查並計算數據的智慧代理。
第一步,在 src/research_agent/tools.py 中實作研究工具。請注意,我們使用 @tool 裝飾器,並撰寫詳盡的型別標註與說明字串(Docstrings),因為這份說明字串將直接成為模型理解工具用途的最高準則:
"""src/research_agent/tools.py:定義研究助理核心工具庫。"""
import statistics
from typing import List
from langchain_core.tools import tool
@tool
def search_academic_database(query: str, max_results: int = 3) -> str:
"""檢索學術論文庫與技術白皮書,回傳相關研究的關鍵發現與實測數據。"""
print(f" [工具執行: search_academic_database] 查詢語句: '{query}'")
# 模擬學術資料庫檢索結果
mock_db = {
"Transformer": [
"論文 A:Transformer-XL 在長文本注意力延遲降低了 35.5%",
"論文 B:FlashAttention-3 在 Hopper 架構上實現了 750 TFLOPs 吞吐",
"論文 C:Multi-Head Latent Attention 節省了 62.0% 的 KV 快取記憶體",
],
"LangGraph": [
"白皮書 1:LangGraph 狀態機在多輪對話中的狀態一致性達 99.9%",
"白皮書 2:SQLite Checkpointer 的單步持久化耗時平均為 4.2 毫秒",
],
}
results = []
for key, entries in mock_db.items():
if key.lower() in query.lower():
results.extend(entries[:max_results])
if not results:
return f"在資料庫中未找到關於 '{query}' 的精確紀錄,建議放寬檢索詞。"
return "\n".join(results)
@tool
def calculate_statistical_metrics(numbers: List[float]) -> str:
"""計算給定浮點數清單的平均值(Mean)與母體標準差(Standard Deviation)。"""
print(f" [工具執行: calculate_statistical_metrics] 輸入數列: {numbers}")
if not numbers:
return "錯誤:傳入數列為空,無法計算統計指標。"
mean_val = statistics.mean(numbers)
stdev_val = statistics.pstdev(numbers) if len(numbers) > 1 else 0.0
return (
f"樣本總數: {len(numbers)}, "
f"平均值: {mean_val:.2f}, "
f"母體標準差: {stdev_val:.2f}"
)
第二步,在 src/research_agent/agents/react.py 中建構代理組裝函式。我們實作雙軌機制:支援讀者配置 OPENAI_API_KEY 進行即時推論,同時提供無金鑰時的 dry_run 模擬代理,確保離線環境下依然能完整驗證:
"""src/research_agent/agents/react.py:ReAct 代理組裝與離線降級。"""
import os
from typing import List, Any
from langchain_core.messages import AIMessage, ToolMessage, HumanMessage
from langgraph.prebuilt import create_react_agent
from src.research_agent.tools import search_academic_database, calculate_statistical_metrics
class MockChatModel:
"""在缺乏 API Key 時模擬模型自主發起 tool_calls 與最終回答的行為。"""
def __init__(self):
self._call_count = 0
def bind_tools(self, tools: List[Any]):
return self
def invoke(self, messages: List[Any], **kwargs):
self._call_count += 1
# 第一輪推理:決定呼叫搜尋工具
if self._call_count == 1:
return AIMessage(
content="我需要先檢索 Transformer 的最新效能指標。",
tool_calls=[{
"name": "search_academic_database",
"args": {"query": "Transformer", "max_results": 2},
"id": "call_search_01",
"type": "tool_call",
}],
)
# 第二輪推理:分析搜尋結果,決定呼叫數值計算工具
elif self._call_count == 2:
return AIMessage(
content="我已經檢索到相關數據(35.5 與 62.0),現在計算這兩筆數據的統計指標。",
tool_calls=[{
"name": "calculate_statistical_metrics",
"args": {"numbers": [35.5, 62.0]},
"id": "call_calc_02",
"type": "tool_call",
}],
)
# 第三輪推理:產出最終結論
else:
return AIMessage(
content="【研究結論】針對 Transformer 效能指標,長文本延遲降低 35.5%、KV 快取節省 62.0%,平均改善幅度達 48.75%(標準差 13.25)。"
)
def get_research_react_agent(dry_run: bool = False):
"""組裝 ReAct 代理應用程式,支援環境變數切換與離線模擬。"""
tools = [search_academic_database, calculate_statistical_metrics]
api_key = os.getenv("OPENAI_API_KEY")
model_name = os.getenv("RESEARCH_AGENT_MODEL", "gpt-4o-mini")
if dry_run or not api_key:
print("[提示] 啟用 ReAct 離線 Mock 模式,使用模擬模型排程。")
llm = MockChatModel()
else:
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model=model_name, temperature=0.0)
system_prompt = (
"你是一位嚴謹的高階研究助理。"
"面對使用者的技術查詢,請先呼叫檢索工具收集客觀證據;"
"若涉及數值分析,必須使用統計計算工具進行精確驗證,切勿自行捏造計算結果。"
)
agent_app = create_react_agent(
model=llm,
tools=tools,
prompt=system_prompt,
)
return agent_app
第三步,實作執行腳本,透過 stream() 單步觀察 ReAct 代理在「模型思考 -> 工具執行 -> 模型再思考」之間的狀態躍遷:
"""執行 ReAct 代理並即時觀測多輪思考與工具呼叫過程。"""
from langchain_core.messages import HumanMessage
from src.research_agent.agents.react import get_research_react_agent
agent = get_research_react_agent(dry_run=True)
query = "請幫我調查 Transformer 的相關最佳化數據,並計算其平均改善指標。"
initial_messages = {"messages": [HumanMessage(content=query)]}
print(f"=== 使用者提問: {query} ===\n")
# 透過 stream 觀察每一步狀態更新
step_index = 0
for event in agent.stream(initial_messages, stream_mode="updates"):
step_index += 1
for node_name, node_update in event.items():
print(f"\n>>> [步驟 {step_index}] 節點名稱: {node_name}")
for msg in node_update.get("messages", []):
if hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
print(f" [發起工具呼叫] {tc['name']}(參數: {tc['args']})")
elif isinstance(msg, ToolMessage):
print(f" [工具回傳結果] {msg.content}")
else:
print(f" [助理終端回答] {msg.content}")
# 輸出(範例輸出):
# === 使用者提問: 請幫我調查 Transformer 的相關最佳化數據,並計算其平均改善指標。 ===
#
# >>> [步驟 1] 節點名稱: agent
# [發起工具呼叫] search_academic_database(參數: {'query': 'Transformer', 'max_results': 2})
# [工具執行: search_academic_database] 查詢語句: 'Transformer'
#
# >>> [步驟 2] 節點名稱: tools
# [工具回傳結果] 論文 A:Transformer-XL 在長文本注意力延遲降低了 35.5%
# 論文 B:FlashAttention-3 在 Hopper 架構上實現了 750 TFLOPs 吞吐
#
# >>> [步驟 3] 節點名稱: agent
# [發起工具呼叫] calculate_statistical_metrics(參數: {'numbers': [35.5, 62.0]})
# [工具執行: calculate_statistical_metrics] 輸入數列: [35.5, 62.0]
#
# >>> [步驟 4] 節點名稱: tools
# [工具回傳結果] 樣本總數: 2, 平均值: 48.75, 母體標準差: 13.25
#
# >>> [步驟 5] 節點名稱: agent
# [助理終端回答] 【研究結論】針對 Transformer 效能指標,長文本延遲降低 35.5%、KV 快取節省 62.0%,平均改善幅度達 48.75%(標準差 13.25)。
第四步,展示直接使用 invoke() 取得最終對話歷史串列,並檢查整個對話紀錄的完整性:
"""使用 invoke 執行完整生命週期並印出最終狀態。"""
final_res = agent.invoke({"messages": [HumanMessage(content="請分析 Transformer 效能數據。")]})
print("\n=== 最終狀態訊息歷史總結 ===")
print(f"對話歷史中共有 {len(final_res['messages'])} 則訊息:")
for idx, m in enumerate(final_res["messages"], 1):
role = m.__class__.__name__
preview = m.content[:45].replace("\n", " ")
print(f" [{idx}] {role}: {preview}...")
# 輸出(範例輸出):
# === 最終狀態訊息歷史總結 ===
# 對話歷史中共有 6 則訊息:
# [1] HumanMessage: 請分析 Transformer 效能數據。...
# [2] AIMessage: 我需要先檢索 Transformer 的最新效能指標。...
# [3] ToolMessage: 論文 A:Transformer-XL 在長文本注意力延遲降低了 35.5%...
# [4] AIMessage: 我已經檢索到相關數據(35.5 與 62.0),現在計算這兩筆數據的統計指標。...
# [5] ToolMessage: 樣本總數: 2, 平均值: 48.75, 母體標準差: 13.25...
# [6] AIMessage: 【研究結論】針對 Transformer 效能指標,長文本延遲降低 35.5%、KV 快取節省 62.0%,平均改善幅度達 48.75%(標準差 13.25)。...
第五步,我們為本篇實作的工具函式撰寫正式的單元測試 tests/test_react_tools.py,驗證學術檢索過濾邏輯與統計計算邊界條件:
"""tests/test_react_tools.py:針對自訂工具的嚴格單元測試。"""
import pytest
from src.research_agent.tools import search_academic_database, calculate_statistical_metrics
def test_search_academic_database_hit():
"""測試資料庫關鍵字命中與結果筆數限制。"""
res = search_academic_database.invoke({"query": "Transformer", "max_results": 2})
assert "Transformer-XL" in res
assert "FlashAttention-3" in res
# 驗證筆數不超過 2 行
assert len(res.strip().split("\n")) == 2
def test_search_academic_database_miss():
"""測試資料庫未命中時的優雅回退提示。"""
res = search_academic_database.invoke({"query": "QuantumComputing2099"})
assert "未找到關於" in res
assert "建議放寬檢索詞" in res
def test_calculate_statistical_metrics_calculation():
"""測試數值清單的平均值與標準差計算。"""
res = calculate_statistical_metrics.invoke({"numbers": [10.0, 20.0, 30.0]})
assert "平均值: 20.00" in res
assert "母體標準差: 8.16" in res
def test_calculate_statistical_metrics_empty():
"""測試空數列的錯誤防禦處理。"""
res = calculate_statistical_metrics.invoke({"numbers": []})
assert "錯誤:傳入數列為空" in res
第六步,我們為 ReAct 代理的圖狀態流轉編寫自動化測試 tests/test_react_agent.py,確認代理在離線 Mock 模式下能如期經歷完整的 5 步狀態躍遷:
"""tests/test_react_agent.py:驗證 ReAct 代理的端到端狀態流轉。"""
import pytest
from langchain_core.messages import HumanMessage
from src.research_agent.agents.react import get_research_react_agent
def test_mock_react_agent_full_flow():
"""測試離線 Mock 模式下的多輪工具調度與收斂。"""
agent = get_research_react_agent(dry_run=True)
res = agent.invoke({"messages": [HumanMessage(content="進行深度測試")]})
messages = res["messages"]
# 期望訊息鏈:Human -> AI(tool) -> Tool -> AI(tool) -> Tool -> AI(final)
assert len(messages) == 6
assert messages[0].content == "進行深度測試"
assert len(messages[1].tool_calls) == 1
assert messages[1].tool_calls[0]["name"] == "search_academic_database"
assert len(messages[3].tool_calls) == 1
assert messages[3].tool_calls[0]["name"] == "calculate_statistical_metrics"
assert "【研究結論】" in messages[5].content
讀者可以在命令列中執行測試套件以確認所有單元測試皆順利通過:
# 執行 ReAct 工具與代理流程單元測試
uv run pytest tests/test_react_tools.py tests/test_react_agent.py -v
常見錯誤與踩雷
在實務中使用 create_react_agent 時,有幾個極易引發執行期例外的踩雷點:
- 工具函式缺少型別標註或 Docstring:語言模型依據工具的 JSON Schema 來決定何時呼叫、以及該傳入什麼引數。如果使用
@tool時未給函式參數加上型別標註(例如寫成def search(q)),或者省略了函式上方的說明字串,LangChain 將無法生成合法的 JSON Schema,或者生成的欄位說明為空,導致模型在推論時完全不知道該工具的功用而拒絕呼叫。 - 將工具回傳值誤設為非字串型別:工具函式的回傳值最終會被封裝為
ToolMessage(content=...)送回給模型閱讀。因此,工具函式應該始終回傳字串(str)或可序列化的 JSON 字串。若直接回傳複雜的自訂 Python 物件或二進位緩衝區,在序列化時會拋出例外,或者被粗暴地轉為難以閱讀的記憶體位址表示法(如<CustomObject at 0x7f...>)。 - 誤以為
create_react_agent回傳的是字串:呼叫agent.invoke(...)回傳的不是一段文字,而是一個完整的狀態字典{"messages": [...]}。初學者常直接寫下print(agent.invoke("問題")),導致印出長篇的字典與訊息物件結構。正確提取最終答案的方法是讀取res["messages"][-1].content。 - 工具執行拋出未攔截的例外導致圖崩潰:如果在工具內部發生網路斷線、除以零或鍵值不存在等例外,而沒有進行
try-except捕捉,這個例外會直接穿透工具節點並導致整個 LangGraph 圖執行中斷。在生產環境中,我們需要更專業的工具錯誤攔截與回饋機制,這正是下一篇將深入展開的重點。
效能與實務提醒
ReAct 代理雖然靈活強大,但在生產部署時必須隨時警惕其資源消耗特徵:
第一點是工具宣告帶來的提示詞權杖負擔(Tool Schema Overhead)。每當你向 create_react_agent 傳入一個工具,該工具的名稱、說明字串與所有參數的 JSON Schema 就會被完整注入到系統提示詞中。如果你一口氣掛載了 20 個工具,每一次呼叫模型的「基本輸入權杖」可能就高達數千權杖。此外,在 ReAct 的多輪對話中,每一次工具回傳的 ToolMessage 都會持續保留在 messages 清單中,導致輸入權杖隨著輪數直線暴增。因此,工具箱應維持精簡專一,避免過度膨脹。
第二點是控制思考發散與死迴圈熔斷。有些時候模型在取得工具回傳後,會因為搜尋結果不理想而不斷換關鍵字重試,演變成無休止的探索。除了 LangGraph 內建的 recursion_limit 之外,在系統提示詞中明確規範「若連續兩次未檢索到滿意資料,請向使用者誠實說明並停止搜尋」,能顯著提高代理的收斂速度並節省 API 費用。
第三點是溫度參數(Temperature)的設定。在驅動 ReAct 工具代理時,強烈建議將模型的 temperature 設定為 0.0。這能最大程度地抑制幻覺,保證工具呼叫引數的精確性與嚴格遵循格式,避免因隨機採樣導致引數名稱或型別出錯。
小結
今天我們深入實戰了 LangGraph 最具代表性的高階原語——create_react_agent。我們回顧了 ReAct 範式從脆弱的純文字模板演進至現代訊息狀態機的歷程,徹底拆解了預建構代理在狀態定義、模型綁定、條件路由與工具節點上的底層拓撲。
我們在 research-agent 專案中實作了專業的學術檢索與統計計算工具,建立了支援 --dry-run 模擬降級的雙軌執行架構,展示了串流步階追蹤與最終訊息鏈審查,並透過多組自動化測試鞏固了工具與代理的可靠度。透過 create_react_agent,我們用極其精煉的程式碼,擁有了具備工業級確定性的自主研究助理核心。
結語
雖然 create_react_agent 幫我們省去了大量手動建圖的繁瑣細節,但現實環境往往充滿了各種突發意外:外部檢索 API 可能遭遇逾時、網路瞬斷、或是模型傳入了非法型別的引數導致工具拋出未預期的崩潰異常。如果工具直接拋出例外,整套代理流程就會立即中斷死亡。
在生產系統中,工具節點必須具備強大的自我防禦與容錯自癒能力:當工具失敗時,它不該崩潰,而是應該將錯誤訊息包裝為合法的觀察結果回傳給模型,讓模型依據錯誤原因自主嘗試修復或調整策略。明天,我們會進入「AG Day 15 ToolNode:工具節點與錯誤處理」,深入拆解 ToolNode 的底層機制與 handle_tool_errors 容錯策略,為我們的研究助理鍛造一副刀槍不入的容錯裝甲。
延伸資源
- LangGraph create_react_agent 官方指南:
https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/。掌握預建構代理的進階參數配置。 - ReAct: Synergizing Reasoning and Acting in Language Models(ICLR 2023):
https://arxiv.org/abs/2210.03629。經典論文原文研讀。 - LangChain Tool 裝飾器與自訂工具規格:
https://python.langchain.com/docs/concepts/tools/。深入了解如何宣告結構化工具。 - OpenAI Function Calling 與 Tool Use 官方指南:
https://platform.openai.com/docs/guides/function-calling。理解模型端解碼工具呼叫的底層通訊協定。
留言
張貼留言