AG Day 13 條件邊:讓流程學會分岔
執行需求:CPU 可跑。本篇文章聚焦於 LangGraph 條件邊(Conditional Edges)的動態路由機制,所有範例皆在本地 CPU 環境下運算,不需要外部模型 API 金鑰,即可完整演練路由函式撰寫、路徑對照字典(Path Mapping)、迴圈重試防護與分支追蹤驗證。
引言
在上一篇 AG Day 12(原文連結)中,我們深入拆解了 LangGraph 的狀態聚合心臟,掌握了透過 Annotated 與 add_messages 縮減器管理多輪對話訊息累積與識別碼就地更新的技巧。然而,回顧我們在 AG Day 11(原文連結)建立的圖結構,節點之間的流轉完全是由固定邊(Normal Edges)所硬性串聯的線性流程:從規劃、收集到報告,每一步的執行軌跡在編譯期就已完全注定。
但現實世界的代理任務往往充滿了不確定性與情境依賴。試想一個嚴謹的研究助理系統:當使用者提出一個定義明確的基礎名詞查詢時,系統應該直接產出簡答,而不必興師動眾地呼叫搜尋工具;當文獻萃取節點發現檢索到的資料互相矛盾或置信度過低時,系統必須能夠退回檢索節點重新生成關鍵字;當執行達到最大容許輪數時,系統則必須強制熔斷並產出降級警語。如果流程只能單向直行,AI Agent 就失去了最重要的「環境適應性與自主決策力」。
讓靜態管線晉升為智慧代理的關鍵樞紐,正是 LangGraph 的條件邊(Conditional Edges)。條件邊賦予了圖在執行時期(Runtime)依據當前狀態動態評估並選擇分支路徑的能力。本文將帶領大家深入條件邊的底層邏輯,從純函式路由設計、嚴格的路徑對照映射(Path Mapping)、迴圈重試的熔斷控制,一路到官方內建的 tools_condition 工具路由,徹底解鎖狀態圖動態分岔與自主決策的強大威力。
條件邊架構原理:狀態評估與動態路由
在圖論狀態機中,固定邊代表「當節點 A 完成後,永遠流向節點 B」;而條件邊則代表「當節點 A 完成後,檢視當前狀態,依據特定規則動態決定下一個目的地可能是節點 B、節點 C 或直接前往終點 END」。
在 LangGraph 1.x 中,宣告一條條件邊主要由三個正交要素所構成:
- 來源節點(Source Node):分支發生的起點。當該節點執行完畢並將狀態增量寫入系統後,圖排程器不會立即跳轉,而是會攔截流程並啟動路由評估。
- 路由函式(Router Function):一個只讀系統狀態的純函式(
route_fn(state: State) -> str)。路由函式接收當前的完整狀態作為輸入,經過條件判斷(例如檢查模型回傳內容、評估分數門檻或檢查錯誤旗標)後,回傳一個代表決策結果的字串鍵值(Key)。 - 路徑對照字典(Path Mapping):一個將路由函式回傳鍵值映射至圖中實際目標節點名稱的對照表(
Dict[str, str])。例如{"need_search": "search_node", "direct_reply": "draft_node", "abort": END}。
初學者常問:為什麼 LangGraph 不直接讓路由函式回傳目標節點名稱,而要多此一舉地引入路徑對照字典?這正體現了工業級框架的防禦性設計思維:路徑對照字典建立了嚴格的「型別邊界與靜態拓撲約束」。編譯器能夠在圖構建階段就驗證對照表中宣告的所有目標節點是否存在,避免因為路由函式筆誤拼錯字串而導致執行階段發生未定義的幽靈跳轉;同時,對照表清晰地展示了該分支的所有可能走向,讓視覺化引擎(如 Mermaid)能精準繪製出菱形條件決策節點與多條分支線路。
避免無窮迴圈:步數防護與遞迴上限
當條件邊的目標路徑指向了拓撲上方的先前節點時,圖中便誕生了「迴圈(Cycle)」——這正是 Agent 實現自我反思、多輪重試與錯誤修復的技術基石。然而,有迴圈就伴隨著「無窮死迴圈」的致命風險。例如,審查節點若始終判定文獻不合格,且未配置最大重試次數限制,代理將會無止境地在研究與審查節點之間反覆橫跳,直到伺服器資源枯竭。
在 LangGraph 工程實踐中,防範死迴圈必須實施「雙重防護體系」:
- 業務層狀態累加計數器(State Step Guard):在狀態架構中宣告
retry_count欄位。每次迴圈時由節點或縮減器累加計數。路由函式在評估品質門檻時,必須強制加上if state["retry_count"] >= MAX_RETRIES: return "terminate"的熔斷檢查。 - 框架層全域遞迴上限(Recursion Limit):LangGraph 在執行圖時(呼叫
.invoke()或.stream()),預設配置了recursion_limit=25(可透過執行設定config={"recursion_limit": N}進行動態微調)。當任何一次執行走訪的節點總次數突破該上限時,排程器會立即拋出GraphRecursionError,從底層強制中斷執行,防止災難性的無限迴圈。
完整實作:為 research-agent 打造具備品質反思的自律研究圖
現在,我們進入具體工程實作。我們將在 research-agent 專案的 src/research_agent/graph.py 中,建構一個具備品質自動評估、動態重試與條件分岔的高階研究狀態圖。該圖包含四個運算節點與兩條關鍵條件邊:
query_classifier:意圖分類節點,判斷主題是否需要深入檢索。evidence_search:資料檢索節點,收集技術論點。quality_gate:品質評估節點,計算論點完備度分數。report_publisher:終端發布節點,格式化輸出最終結論。
第一步,宣告包含分支標記、重試計數與品質分數的狀態架構 ReflectiveResearchState:
"""src/research_agent/graph.py:定義具備條件分支與反思機制的狀態圖。"""
from typing import TypedDict, List, Optional, Dict
from langgraph.graph import StateGraph, START, END
class ReflectiveResearchState(TypedDict):
"""研究助理在反思與條件分支圖中流轉的狀態結構。"""
topic: str
is_complex: bool
evidence_list: List[str]
quality_score: float
retry_count: int
final_report: Optional[str]
第二步,實作四個純函式節點,負責執行局部運算與增量狀態產生:
"""定義四個運算節點函式。"""
def query_classifier_node(state: ReflectiveResearchState) -> dict:
"""節點 1:分析主題複雜度,決定是否需要發動深度檢索。"""
topic = state.get("topic", "")
print(f"[節點: classifier] 正在評估研究主題: '{topic}'")
# 簡單啟發式規則:主題字元長度大於 10 或包含特定詞彙視為複雜主題
is_complex = len(topic) > 8 or "架構" in topic or "原理" in topic
return {"is_complex": is_complex}
def evidence_search_node(state: ReflectiveResearchState) -> dict:
"""節點 2:檢索文獻資料,並依當前輪數擴充佐證內容。"""
current_retries = state.get("retry_count", 0)
topic = state.get("topic", "")
print(f"[節點: search] 執行第 {current_retries + 1} 輪資料檢索...")
new_evidence = list(state.get("evidence_list", []))
if current_retries == 0:
new_evidence.append(f"基礎論據:{topic} 具備高確定性語意。")
else:
new_evidence.append(f"深度佐證:{topic} 在高並行壓測下展現優異吞吐。")
return {"evidence_list": new_evidence}
def quality_gate_node(state: ReflectiveResearchState) -> dict:
"""節點 3:品質守門員,評估資料完備度並累加審查次數。"""
evidence = state.get("evidence_list", [])
current_retries = state.get("retry_count", 0)
print(f"[節點: quality_gate] 審查目前累積的 {len(evidence)} 筆佐證品質...")
# 模擬品質評定:若佐證達到 2 筆以上給予高分,否則給予低分需重試
score = 0.92 if len(evidence) >= 2 else 0.65
return {
"quality_score": score,
"retry_count": current_retries + 1,
}
def report_publisher_node(state: ReflectiveResearchState) -> dict:
"""節點 4:終端發布節點,產出完整研究摘要。"""
topic = state.get("topic", "")
evidence = state.get("evidence_list", [])
score = state.get("quality_score", 0.0)
print("[節點: publisher] 通過審查,正在生成最終定稿報告...")
report = (
f"【研究成果報告】主題:{topic}\n"
f"品質審查得分:{score:.2f} | 總探索輪數:{state.get('retry_count', 0)}\n"
f"核心論證要點:\n" + "\n".join(f" * {e}" for e in evidence)
)
return {"final_report": report}
第三步,實作兩個關鍵的純函式路由邏輯(Router Functions):
"""定義純函式路由評估邏輯。"""
def route_query_type(state: ReflectiveResearchState) -> str:
"""路由 1:依據複雜度決定走深入檢索還是快速發布。"""
if state.get("is_complex", False):
print(" --> [分岔判定] 判斷為複雜主題,進入深入檢索流程。")
return "need_search"
print(" --> [分岔判定] 判斷為常規查詢,跳過檢索直接結案。")
return "fast_track"
def route_quality_verdict(state: ReflectiveResearchState) -> str:
"""路由 2:依據品質得分與重試上限,決定是迴圈重試還是放行發布。"""
score = state.get("quality_score", 0.0)
retries = state.get("retry_count", 0)
# 滿足品質門檻直接放行
if score >= 0.85:
print(f" --> [審查判定] 得分 {score:.2f} 達標,放行至報告發布。")
return "pass"
# 未達標但已達最大重試限制(2次),強制終止防範死迴圈
if retries >= 2:
print(f" --> [審查判定] 已達最大重試上限 ({retries}),強制放行。")
return "pass"
print(f" --> [審查判定] 得分 {score:.2f} 未達標,觸發回溯重新檢索。")
return "retry"
第四步,組裝狀態圖並宣告條件邊。請注意 add_conditional_edges 的語法結構與明確的路徑對照字典:
"""組裝具備條件邊的 LangGraph 實體並編譯。"""
def build_conditional_research_graph():
"""建置具備動態分岔與反思迴圈的狀態圖。"""
workflow = StateGraph(ReflectiveResearchState)
# 1. 註冊四個核心節點
workflow.add_node("classifier", query_classifier_node)
workflow.add_node("searcher", evidence_search_node)
workflow.add_node("quality_gate", quality_gate_node)
workflow.add_node("publisher", report_publisher_node)
# 2. 起點固定連接至分類器
workflow.add_edge(START, "classifier")
# 3. 條件邊 1:分類器依據主題複雜度動態分岔
workflow.add_conditional_edges(
source="classifier",
path=route_query_type,
path_map={
"need_search": "searcher",
"fast_track": "publisher",
},
)
# 4. 固定邊:檢索完畢後固定流向品質審查節點
workflow.add_edge("searcher", "quality_gate")
# 5. 條件邊 2:品質審查節點依據得分決定迴圈回溯或結案發布
workflow.add_conditional_edges(
source="quality_gate",
path=route_quality_verdict,
path_map={
"retry": "searcher", # 構成反思迴圈!
"pass": "publisher",
},
)
# 6. 發布完畢導向終點
workflow.add_edge("publisher", END)
return workflow.compile()
第五步,實作執行入口,分別測試「複雜主題觸發檢索與回溯重試」以及「簡單主題快速通道」兩種截然不同的執行路徑:
"""執行具備條件邊的狀態圖並檢視動態躍遷軌跡。"""
from src.research_agent.graph import build_conditional_research_graph
app = build_conditional_research_graph()
# 測試案例 A:複雜主題,將觸發檢索、第一次審查不合格回溯、第二次審查合格放行
print("=" * 20, "案例 A:複雜架構主題(觸發反思迴圈)", "=" * 20)
complex_input = {
"topic": "LangGraph 條件邊與狀態機架構",
"is_complex": False,
"evidence_list": [],
"quality_score": 0.0,
"retry_count": 0,
"final_report": None,
}
res_a = app.invoke(complex_input)
print("\n--- 最終產出 ---")
print(res_a["final_report"])
# 測試案例 B:極短名詞,將走快速通道直接發布
print("\n" + "=" * 20, "案例 B:簡易詞彙查詢(快速通道)", "=" * 20)
simple_input = {
"topic": "狀態機",
"is_complex": False,
"evidence_list": ["常規定義:系統在有限狀態集合間躍遷。"],
"quality_score": 1.0,
"retry_count": 0,
"final_report": None,
}
res_b = app.invoke(simple_input)
print("\n--- 最終產出 ---")
print(res_b["final_report"])
# 輸出(範例輸出):
# ==================== 案例 A:複雜架構主題(觸發反思迴圈) ====================
# [節點: classifier] 正在評估研究主題: 'LangGraph 條件邊與狀態機架構'
# --> [分岔判定] 判斷為複雜主題,進入深入檢索流程。
# [節點: search] 執行第 1 輪資料檢索...
# [節點: quality_gate] 審查目前累積的 1 筆佐證品質...
# --> [審查判定] 得分 0.65 未達標,觸發回溯重新檢索。
# [節點: search] 執行第 2 輪資料檢索...
# [節點: quality_gate] 審查目前累積的 2 筆佐證品質...
# --> [審查判定] 得分 0.92 達標,放行至報告發布。
# [節點: publisher] 通過審查,正在生成最終定稿報告...
#
# --- 最終產出 ---
# 【研究成果報告】主題:LangGraph 條件邊與狀態機架構
# 品質審查得分:0.92 | 總探索輪數:2
# 核心論證要點:
# * 基礎論據:LangGraph 條件邊與狀態機架構 具備高確定性語意。
# * 深度佐證:LangGraph 條件邊與狀態機架構 在高並行壓測下展現優異吞吐。
第六步,我們為這個條件分岔圖撰寫完整的自動化單元測試 tests/test_conditional_edges.py,透過 pytest 針對分支邏輯、回溯次數與熔斷機制進行嚴格驗證:
"""tests/test_conditional_edges.py:驗證條件邊動態分岔與重試邏輯。"""
import pytest
from src.research_agent.graph import (
build_conditional_research_graph,
route_query_type,
route_quality_verdict,
)
def test_router_query_type_classification():
"""測試路由 1 依據主題長度與關鍵字的精確分岔。"""
assert route_query_type({"is_complex": True}) == "need_search"
assert route_query_type({"is_complex": False}) == "fast_track"
def test_router_quality_verdict_loop_and_fuse():
"""測試路由 2 在低分時重試、高分時通過、超限時熔斷。"""
# 低分且輪數低 -> 重試
assert route_quality_verdict({"quality_score": 0.5, "retry_count": 1}) == "retry"
# 高分達標 -> 通過
assert route_quality_verdict({"quality_score": 0.9, "retry_count": 1}) == "pass"
# 低分但達上限 2 輪 -> 強制通過熔斷
assert route_quality_verdict({"quality_score": 0.4, "retry_count": 2}) == "pass"
def test_graph_full_execution_cycle():
"""測試端到端執行時回溯迴圈成功被執行兩次後定稿。"""
app = build_conditional_research_graph()
result = app.invoke({
"topic": "複雜演算法架構評估",
"is_complex": False,
"evidence_list": [],
"quality_score": 0.0,
"retry_count": 0,
"final_report": None,
})
assert result["retry_count"] == 2
assert result["quality_score"] == 0.92
assert len(result["evidence_list"]) == 2
assert "品質審查得分:0.92" in result["final_report"]
讀者可以在命令列中執行測試以確保所有邊界條件均通過審查:
# 執行條件邊單元測試
uv run pytest tests/test_conditional_edges.py -v
常見錯誤與踩雷
條件邊在賦予系統高度靈活性的同時,也帶來了更高的除錯複雜度。以下四個常見陷阱值得高度警惕:
- 路由函式回傳了未在
path_map中宣告的鍵值:若route_quality_verdict回傳了字串"unknown",但path_map中僅有{"retry": "...", "pass": "..."},LangGraph 在執行階段會直接拋出ValueError,指出找不到對應目標。務必確保路由函式的每一個return分支都百分之百涵蓋在path_map的鍵清單中。 - 在路由函式內部偷改狀態或引發副作用:路由函式(Router Function)的職責是純粹的「決策者」。它只能讀取傳入的
state,絕對不能修改它,更不可在內部執行耗時的 I/O 呼叫。任何狀態修改必須且只能在「節點」中完成。破壞這項原則會導致檢查點紀錄失去因果一致性。 - 遺漏遞迴步數保護導致圖執行超限崩潰:當設計包含回溯邊的狀態機時,若忘了在狀態中累加重試計數,或忘記在路由中設定上限跳出條件,一旦品質評估持續未通過,圖執行將一路狂飆直到觸發 LangGraph 預設的
GraphRecursionError: Recursion limit of 25 reached。所有環狀圖都應具備清楚的終止保證。 - 混淆
add_edge與add_conditional_edges:固定邊add_edge(source, target)接受的是兩個字串;而條件邊add_conditional_edges(source, path, path_map)的第二個參數必須是一個「可呼叫的函式物件(Callable)」,第三個參數才是字典。若把字串直接傳入path,編譯器會立即報錯。
效能與實務提醒
在生產系統中運用條件邊,需要兼顧路由效率與架構開銷:
第一點是「確定性路由 vs 模型路由」的成本權衡。很多初學者習慣在每個分岔點都呼叫一次 LLM 來做決策(例如讓 GPT-4o 判斷是否需要搜尋)。然而,每次呼叫模型都會增加 500 到 1,500 毫秒的延遲與金錢開銷。在實務上,應該優先使用高效率的啟發式規則(Heuristic Rules)、正則表達式或狀態指標來做程式化路由;僅在面對語意高度模糊或需要複雜意圖理解的關鍵節點,才呼叫輕量級模型進行語意分類。
第二點是全域遞迴上限(Recursion Limit)的合理配置。在預設情況下,LangGraph 限制單次執行最多走訪 25 個節點。對於簡單的 ReAct 或三步迴圈而言,這個限制非常合適;但對於需要深度長程檢索、走訪十數篇論文的研究任務,25 個節點可能在半途就被耗盡。在此類場景下,可於呼叫時透過配置字典放大配額,例如:app.invoke(inputs, config={"recursion_limit": 50}),同時確保業務程式碼內部具備清晰的計數熔斷機制。
第三點是善用官方內建的 tools_condition。在後續建構工具調度代理(ReAct)時,我們不需要自己手刻判斷模型是否有提出工具呼叫請求的路由函式。LangGraph 在 langgraph.prebuilt 中直接提供了開箱即用的 tools_condition,它能自動檢查最後一則訊息是否包含 tool_calls,若有則無縫路由至 "tools" 節點,否則轉向 END。這能大幅精簡核心業務圖的樣板程式碼。
小結
今天我們深入探討了 LangGraph 的動態神經中樞——條件邊(Conditional Edges)。我們從靜態管線的僵化局限出發,掌握了路由函式與路徑對照字典(Path Mapping)的標準語法语意。我們在 research-agent 專案中實作了一個具備複雜度意圖分流、佐證品質審查與動態回溯重試的自律狀態圖。
透過步數累加防護與遞迴限制,我們為系統築起了防範無窮迴圈的堅固堡壘。至此,我們的研究助理已經徹底學會了「依據環境資訊自主抉擇路徑」,從被動的單向工作流正式躍升為具備反思與適應能力的智慧狀態機。
結語
回顧 Day 11 到 Day 13,我們從零開始掌握了 LangGraph 的 StateGraph 拓撲架構、TypedDict 狀態縮減器、固定邊與條件分岔。然而在實際開發中,如果每次要讓代理呼叫工具,我們都必須手動連線模型、解析 tool_calls、配置條件邊並指向工具節點,樣板程式碼依然略嫌繁瑣。
為了解決這個最經典的工程模式,LangGraph 提供了一個極具代表性的高階原語——create_react_agent。它將「思考-行動-觀察(Reason + Act)」的經典代理迴圈高度濃縮,讓我們僅需幾行程式碼就能組裝出功能齊備的對話工具代理。
明天,我們會進入「AG Day 14 ReAct Agent:create_react_agent 實戰」,正式引進預建構的 ReAct 代理工廠,學習如何將我們的研究工具與模型無縫縫合,並在保持圖靈活度的同時大幅提升工程開發效率。
延伸資源
- LangGraph 條件邊與分支官方指南:
https://langchain-ai.github.io/langgraph/how-tos/branching/。深入查閱動態路由的最佳實踐模式。 - LangGraph Recursion Limit 官方文件:
https://langchain-ai.github.io/langgraph/concepts/low_level/#recursion-limit。了解遞迴上限與死迴圈防護機制。 - LangGraph Prebuilt 工具路由規格:
https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.tools_condition。查閱官方內建工具條件邊的使用方式。 - ReAct 代理架構經典論文(Yao et al., 2022):
https://arxiv.org/abs/2210.03629。理解推理與行動交織的學術理論源頭。
留言
張貼留言