AI 教學 16 min read

LangGraph Conditional Edge + Reducer:讓 graph 會分流、會重試

線性 graph 不夠用時你需要兩個東西:Conditional Edge(跑完一個 node 看結果決定下一步去哪——過了往前、沒過退回重試、重試太多次直接退件)跟 Reducer(讓某個 state 欄位累加而不是覆蓋,最典型就是 retry 計數器)。這篇用 add_conditional_edges + router 函式 + Annotated[int, operator.add] 把這兩件事講透,並解釋一個關鍵設計決策:MAX_RETRY 該設多少、為什麼很多嚴肅場景答案是 0。「用真實專案學 LangGraph」系列 EP3。

#LangGraph #LangGraph 教學 #Conditional Edge +4 tags
本文目錄 · 10
TL;DR：線性 graph 不夠用的時候，你需要兩個東西：Conditional Edge（跑完一個 node，看結果決定下一步去哪——過了往前、沒過退回重試、重試太多次直接退件）跟 Reducer（讓某個 state 欄位「累加」而不是「覆蓋」，最典型就是 retry 計數器）。這篇用 add_conditional_edges + router 函式 + Annotated[int, operator.add] 把這兩件事講透，並解釋一個關鍵設計決策：MAX_RETRY 該設多少、為什麼很多時候答案是 0。這是「用真實專案學 LangGraph」系列 EP3。

EP2 教完線性流水線——一站接一站往前走。但真實流程不是直線：LLM 出的結果可能不合規，要退回重做；重試幾次還是不行，要直接放棄。這篇就把「會分流、會重試」的能力加上去。

📌 目錄

為什麼線性 graph 不夠用

接著上一篇的血脂決策 pipeline。現在多一個現實需求：LLM 出的建議要過一道業務規則驗證。比如：

  • 它說「升級用藥」，但理由欄位空白 → 不合格，退回重出
  • 它推薦的藥不在允許清單裡 → 不合格
  • 重試 N 次還是不合格 → 別卡死，直接退件回報

通過

不通過 & 還能重試

不通過 & 重試用完

LLM 出建議

業務規則驗證

輸出建議

退件

看到那個「分三條路」的 B 沒有？線性的 add_edge 做不到這個——它只能「A 跑完一定跑 B」。要讓「跑完 B 之後走哪條，由 B 的結果決定」，就要用 conditional edge。

⚡ Conditional Edge：跑完看結果決定去哪

add_conditional_edges 取代 add_edge，多帶兩個東西：一個 router 函式（看 state 決定回傳哪個標籤），跟一個 標籤到 node 的對照表：

g.add_conditional_edges(
    "validate",          # 跑完這個 node 後
    validate_router,     # 呼叫這個函式決定去哪
    {
        "ok": "adapt_response",   # 回 "ok" → 去這個 node
        "retry": "llm_decide",    # 回 "retry" → 退回 LLM node 重做
        "reject": END,            # 回 "reject" → 結束
    },
)

意思很直白：validate 這個 node 跑完，LangGraph 呼叫 validate_router(state)，拿它的回傳值去對照表裡查下一站。回 "retry" 就跳回 llm_decide——這條邊往回連，就形成了重試迴圈。

🧠 Router 函式：回傳一個字串就好

Router 函式跟 node 不一樣：node 回傳 dict（改 state），router 回傳字串（決定路線）。Router 不改 state，只讀 state 然後做決定：

from typing import Literal

MAX_RETRY = 1


def validate_router(state) -> Literal["ok", "retry", "reject"]:
    v = state.get("validation")
    if v and v["passed"]:
        return "ok"                          # 驗過了，往前
    if state.get("retry_count", 0) >= MAX_RETRY:
        return "reject"                      # 重試用完，退件
    return "retry"                           # 還能試，退回重做

回傳型別標 Literal["ok", "retry", "reject"] 是好習慣——它讓你（跟你的 IDE）一眼看出這個 router 只會吐這三個值，而且必須跟 add_conditional_edges 對照表的 key 完全對上。少一個、拼錯一個，跑到那條路就炸。

⚙️ Reducer：讓欄位累加而不是覆蓋

現在問題來了：retry_count 怎麼加上去？

按 EP2 教的，node 回傳 dict 會「合併」進 state——但合併是覆蓋。如果驗證失敗的 node 回傳 {"retry_count": 1}，那 state 裡的 retry_count 就被設成 1。再失敗一次，還是 1。永遠加不上去，重試迴圈跑不完。

解法是 reducer：告訴 LangGraph 這個欄位的更新方式不是「覆蓋」，而是「用某個函式合併」。對計數器來說，那個函式就是「加法」：

import operator
from typing import Annotated, TypedDict

class AnalyzeState(TypedDict, total=False):
    validation: dict
    retry_count: Annotated[int, operator.add]   # 關鍵：標上 reducer
    # ... 其他欄位

Annotated[int, operator.add] 的意思是：「這欄位是 int，每次 node 回傳新值時，用 operator.add 跟舊值合併」。所以驗證失敗的 node 回傳 {"retry_count": 1} 時：

  • 第一次失敗：0 + 1 = 1
  • 第二次失敗：1 + 1 = 2
它真的會累加了。這就是 reducer 的核心——改變某個欄位的合併語意。最常見的兩個用途：計數器（operator.add）跟訊息列表（把新訊息 append 進舊列表，而不是整串換掉）。

📝 完整範例：驗證 + 重試 + 退件

把 conditional edge + reducer 拼進 EP2 的 graph：

import operator
from typing import Annotated, Literal, TypedDict, Any
from langgraph.graph import StateGraph, START, END

MAX_RETRY = 1


class AnalyzeState(TypedDict, total=False):
    patient: dict[str, Any]
    decision: str
    validation: dict
    retry_count: Annotated[int, operator.add]
    advice: str


def llm_decide(state: AnalyzeState) -> dict:
    # 假裝這裡呼叫 LLM。第一次故意給沒理由的爛結果，重試後給好的
    attempt = state.get("retry_count", 0)
    if attempt == 0:
        return {"decision": "升級用藥", "validation": None}  # 缺理由
    return {"decision": "升級用藥（理由：LDL 未達標）"}


def validate(state: AnalyzeState) -> dict:
    ok = "理由" in state.get("decision", "")
    out = {"validation": {"passed": ok}}
    if not ok:
        out["retry_count"] = 1     # 失敗才 +1，靠 reducer 累加
    return out


def validate_router(state) -> Literal["ok", "retry", "reject"]:
    if state["validation"] and state["validation"]["passed"]:
        return "ok"
    if state.get("retry_count", 0) >= MAX_RETRY:
        return "reject"
    return "retry"


def adapt_response(state: AnalyzeState) -> dict:
    return {"advice": state["decision"]}


g = StateGraph(AnalyzeState)
g.add_node("llm_decide", llm_decide)
g.add_node("validate", validate)
g.add_node("adapt_response", adapt_response)


g.add_edge(START, "llm_decide")
g.add_edge("llm_decide", "validate")
g.add_conditional_edges(
    "validate",
    validate_router,
    {"ok": "adapt_response", "retry": "llm_decide", "reject": END},
)
g.add_edge("adapt_response", END)


app = g.compile()
result = app.invoke({"patient": {}})
print(result.get("advice", "（退件）"))
# → 升級用藥（理由：LDL 未達標）

跑這段：第一次 llm_decide 給沒理由的結果 → validate 不過、retry_count 變 1 → router 回 retry → 退回 llm_decide → 這次給有理由的 → validate 過 → 往 adapt_response → 印出最終建議。整個重試迴圈活起來了。

🎯 關鍵決策：MAX_RETRY 該設多少

這是設計層的決定，不是 code 問題，但它最影響系統行為。

直覺上「多重試幾次提高成功率」，但實務上很多嚴肅場景 MAX_RETRY 應該設 0——也就是「失敗一次就直接退件，不重試」。原因：

  • 重試會放大延遲跟成本：每次重試都是一次完整 LLM 呼叫，retry 3 次 = 4 倍 latency + 4 倍 token 費用。
  • 重試常常救不回來：如果 LLM 第一次就出系統性錯誤（prompt 設計問題、模型能力不足），重試只是換個說法錯一樣的。
  • fail fast 比較誠實：尤其醫療、金融這種高風險場景，與其讓系統「重試到看起來像對」，不如失敗就老實退件給人處理。
我自己那套醫療 pipeline 就是 MAX_RETRY = 0——驗證不過直接退件，噴 422 給上游。保留 "retry" 這條路只是讓 graph 結構完整，實際上不會被走到（因為 retry_count 一失敗就 ≥ 0）。

什麼時候才該 retry > 0？當失敗是隨機性的（網路抖動、偶發 JSON parse 失敗），重試真的可能成功。如果失敗是系統性的，retry 只是燒錢。先想清楚你的失敗是哪一種。

🔥 我踩過的坑

坑一：router 回傳的字串對不上對照表 key

def router(state) -> str:
    return "OK"     # ❌ 對照表 key 是小寫 "ok"

router 回傳的字串必須完全等於 add_conditional_edges 對照表的 key。大小寫、空白差一個就 KeyError。標 Literal[...] 回傳型別能讓這類錯誤在寫的時候就被 IDE 抓到。

坑二：忘記標 reducer，計數器永遠不累加

retry_count 沒標 Annotated[int, operator.add]，每次回傳 {"retry_count": 1} 都是覆蓋成 1，永遠到不了 MAX_RETRY，重試迴圈變無限迴圈。LangGraph 預設有遞迴上限（recursion_limit）會幫你踩煞車丟錯，但那是兜底，不是你想要的行為。需要累加的欄位，第一時間就標 reducer。

坑三：retry 邊沒接回正確的 node

重試是「退回去重做」，那條 conditional edge 要連回該重做的起點。常見錯誤是連回 validate 自己（那就只是反覆驗同一個爛結果，不會變好），或連到不相干的 node。想清楚「重試到底要重跑哪一段」。

❓ 常見問題 FAQ

Q：conditional edge 跟 add_edge 可以混用嗎？
A：可以，而且通常會混。大部分的線用 add_edge 直連，只有真的需要分流的那幾個點用 add_conditional_edges。不要每條線都搞成 conditional，那會讓流程難讀。

Q：router 函式可以改 state 嗎？
A：不該。router 的職責是「讀 state、決定路線」，改 state 是 node 的事。如果你發現 router 裡想改東西，那段邏輯應該搬到前一個 node 去。職責分清楚，graph 才好維護。

Q：reducer 只能用 operator.add 嗎？
A：不是。reducer 是任何「拿舊值跟新值、回傳合併值」的函式。operator.add 適合數字相加跟 list 串接；你也可以自己寫一個函式做更複雜的合併（例如去重後合併、取最大值）。LangGraph 對訊息列表還有專用的 add_messages reducer。

Q：怎麼避免重試迴圈跑成無限迴圈？
A：兩層保險。第一層是你自己的 MAX_RETRY 邏輯（router 判斷 retry_count 超過就走 reject）。第二層是 LangGraph 的 recursion_limit（compile 後 invoke 時可設），跑超過步數上限會丟 GraphRecursionError。第一層是你該做的，第二層是 LangGraph 幫你兜底。

Q：MAX_RETRY = 0 那為什麼還要保留 retry 那條邊？
A：因為 add_conditional_edges 的對照表必須涵蓋 router 可能回傳的所有值。即使 retry 實際上不會被走到，對照表少了那個 key，萬一 router 真的回了它就 KeyError。保留是為了結構完整跟防禦。

🔗 延伸資源

author
陳
陳彥彤

AI 工程師 · AI 顧問。Java 後端 8 年、AI 工程師 2 年。AI 內訓 · AI 導入顧問 · 前後端與雲端培訓。

support

覺得文章有用可以到 GitHub 給個 star，或是透過信箱聊聊 AI 內訓、AI 導入顧問或前後端 / 雲端培訓。

email github
related

相關文章

[AI 教學] · 17min
LangGraph 核心:State / Node / Edge 完整教學(第一個會跑的 graph)
LangGraph 的核心只有三個概念:State(跨步驟共用的狀態)、Node(收 state 回傳要改的欄位的函式)、Edge(把 node 串起來的線)。這篇用血脂決策 pipeline 的真實場景,帶你從 StateGraph 定義、add_node、add_edge 一路到 compile 與 invoke,給一個複製貼上就能跑的最小範例,並講三個新手最常踩的坑:回傳整份 state、忘記接 START/END、在 node 裡 mutate state。「用真實專案學 LangGraph」系列 EP2。
[AI 教學] · 13min
LangChain vs LangGraph 是什麼?2026 該學哪個(完整比較)
LangChain 跟 LangGraph 不是二選一,是上下層:LangGraph 是底層 runtime(管 state、流程、重試),LangChain 是建在它上面的高階 API。2025-10-22 兩者都上 1.0 正式版。這篇用比較表、三個情境判斷、三個真實踩坑,講清楚標準 agent loop 該用 LangChain、要精細控制流程分流與重試該用 LangGraph,以及完全新手該從哪個開始學。「用真實專案學 LangGraph」系列第一篇。
[AI 工具] · 12min
我原本用 Hermes Agent CLI,後來為什麼換成 LangGraph(遷移覆盤)
系列番外。我這套醫療決策 pipeline 最早不是用 LangGraph,是用 Hermes Agent CLI(NousResearch 2026 年初開源的自主 agent 框架)。Hermes 很強,但強在通用對話助理、會自我學習、接 20+ 平台;而我要的是確定性步驟為主、流程可畫成圖、每步可審計、出錯能精準退件的決策管線。方向不對再強也是錯的工具。這篇覆盤這個選型決策:Hermes 哪裡好、為什麼對我的場景不對、換到 LangGraph 解決了什麼。「用真實專案學 LangGraph」系列 EP6(完結)。
← all posts