前情提要：在 Day 15，我們看見市府如何透過「策略模式」彈性切換週末與平日的交通管制演算法。今天，我們要從「策略切換」進入「責任鏈分工」，探討一個更棘手的問題：當一個案件需要多個單位依序處理時，如何避免公文旅行和權責不清？我們來看看案件經理 Tess 和流程編排工程師 Kaito 如何聯手拆解這個難題。
週末深夜，市民服務中心 App 湧入一則棘手的陳情案：「夜間施工噪音擾鄰，而且施工圍籬還佔用了人行道！」這案件橫跨了工務局（圍籬）、環保局（噪音）、警察局（秩序），甚至還可能牽涉到社福單位（弱勢住戶安寧）。
舊有的前台工單系統設計得相當天真，只有一個「一鍵轉派」功能。想當然耳，這則陳情案就像一顆燙手山芋，在各局處間被踢來踢去，眼看就要超過 SLA（服務等級協定）的處理時限，卻無人能下決定。
週一早晨，案件經理 Tess 接手了這個爛攤子。她的目標非常明確：在不修改市民報案 App（呼叫端）的前提下，建立一個自動化的處理鏈。案件一進來，就應該像接力賽一樣，沿著指定的路線跑完所有程序：受理/去識別 → 轄區判定 → 噪音證據檢核 → 法規比對 → 罰則與勸導 → 現勘派遣 → 案結回報。
🎯 目標驗收 (Given/When/Then):
Given: 一筆複合型陳情案件（含噪音與圍籬外溢），附有地點、時段、錄音檔和照片。
When: 市府系統切換為「夜間加急處理鏈」，並因應媒體資料的敏感性，臨時在流程中插入「隱私去識別」節點；此外，當案件地點為「非本轄區」時，處理鏈必須能短路 (short-circuit) 並自動轉交。
Then: 整起案件應在 30 分鐘內完成責任歸屬與現勘派遣；各局處節點之間不可窺探或改動彼此的內部實作；案件的流轉由處理鏈自動接棒，而非由一個巨大的中央控制器寫死流程。
🧭 術語卡（今日會用到）
GoF｜Chain of Responsibility：將請求沿著物件鏈進行傳遞，直到鏈上有物件處理此請求為止。鏈上的每個節點都可以處理請求、將請求往後傳，或直接中斷（短路）流程。
EIP/EDA：企業整合模式/事件驅動架構。這裡是 Filter（過濾器）、Pipeline（管道）與 Content-based Router（內容路由）的組合應用，利用訊息的屬性來決定下一站該由哪個節點處理。
MAS：多代理系統。每個處理者（Handler Agents）都可以在 DF (Directory Facilitator，黃頁) 註冊自己的能力，並依循標準協定來交接任務。
讓我們倒帶看看，舊系統的 CaseService.handle() 方法裡藏著什麼祕密。噢，天啊！是一個長達 200 行的 if/elif/else 巨獸：
Before:
def handle_case(case):
if case.type == "noise_complaint":
# 處理噪音... -> 接著比對法規... -> 最後派遣人員...
elif case.type == "construction_issue":
# 處理圍籬...
elif case.type == "night_emergency" and case.hasMedia:
# 臨時加的去識別邏輯...
# ...
if not in_jurisdiction(case.location):
forward_case()
return
這種寫法的後果是組合爆炸、測試地獄、以及高度耦合。
After (概念骨架):
# 流程被封裝進鏈中，呼叫端變得乾淨
chain = build_chain(night=True, is_event=False)
chain.handle(case)
# 鏈的組裝邏輯
first = IntakeHandler()
(first.set_next(RedactPIIHandler())
.set_next(JurisdictionHandler())
.set_next(EvidenceCheckHandler())
.set_next(LegalCheckHandler())
.set_next(DispatchHandler()))
Kaito 提出的責任鏈模式，給了 Tess 一個優雅的解法。
一句話解釋：把流程順序與「能否處理」的判斷邏輯，一起封裝到一個個獨立的處理節點（Handler），以鏈式交棒；呼叫端只須把請求丟給第一個節點。
規約：
handle(ctx) -> Optional[Result]。回None＝放行；回Result＝節點接手並短路（包含成功終止或異常終止）。
✅ 何時用？
逐級審核流程：需要經過層層關卡放行、攔截或短路的場景，如：請假單審核、客服問題升級、風控系統的白名單→灰名單→黑名單檢查、API 權限核實等。
處理節點需動態配置：當處理流程需要依據不同情境（如夜間版、活動版、災防應變版）動態增減、替換或重排節點時。
⛔️ 何時不要用？
流程骨架固定，僅實作不同：如果處理的「大步驟」是固定的，只是每個步驟的「做法」有細微差異，那更適合用 Template Method（樣板方法模式）。
僅需在「同一步驟」切換演算法：如果只是想在某個環節替換不同的計算或處理策略，用 Strategy（策略模式） 就夠了。
需要協調多方複雜互動：如果物件之間是網狀的複雜溝通，而非線性的鏈式傳遞，那應該考慮 Mediator（中介者模式） 來建立一個中央協調中心。
與鄰近模式的分工邊界，再次劃重點：
Strategy: 在「某個步驟」選用不同演算法。
Template Method: 固定流程「骨架」，讓子類填空。
Mediator: 網狀溝通的「協調中心」。
Chain of Responsibility: 線性流程的「逐級接棒與短路」。當遇到非線性的分流時，CoR 應搭配 Router 或 Composite 模式，而非硬湊出多條鏈。
如何閱讀三層並置圖（先說為什麼）
目的：把同一概念在三個縮放層次對齊，避免只停在類圖而忽略「訊息怎麼流」「角色怎麼協作」。
順序：① 微觀 GoF → ② 中觀 EIP/EDA → ③ 宏觀 MAS。
在最微觀的層次，我們定義了 Handler 介面與一系列實作它的具體處理節點。每個節點都持有下一個節點的參考（next），形成一條鏈，鏈的終點為 null。
從訊息流的角度看，案件就像一個訊息，在一個 Pipeline（管道）中流動。途中的節點會根據訊息內容（例如是否為夜間、是否為本轄）決定流向，甚至提前結束流程（短路）。
註：若需多條平行處理鏈（如依案件類型分流），請由 Content-based Router 先行處理。
在宏觀的城市治理層面，Tess 就像一位任務調度官。她不用知道每個代理（Agent）的實作細節，只需向 DF（黃頁）查詢符合「夜間處理鏈」這個能力的代理清單，然後依序將案件委派下去。
Kaito 迅速地用 Python 刻畫出一個更工程化的責任鏈結構，引入了日誌、上下文、以及不可變資料物件。
觀測性提示：若上雲端正式環境，建議把
logging換成結構化日誌，並搭配 Metrics 監控指標（例如：各節點延遲、短路率）。
import abc
import logging
from dataclasses import dataclass, replace
from typing import Optional, Dict, Any, List
# --- 基礎建設 ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
@dataclass(frozen=True)
class Case:
"""不可變的案件資料"""
case_id: str
type: str
payload: Dict[str, Any]
meta: Dict[str, Any]
@dataclass(frozen=True)
class Result:
"""不可變的處理結果"""
code: int
reason: str
class Context:
"""可變的上下文，用於追蹤流程"""
def __init__(self, case: Case):
self.case = case
self.trace: List[str] = []
# --- 處理者介面與基底 ---
class Handler(abc.ABC):
"""定義所有處理節點的契約"""
@abc.abstractmethod
def set_next(self, nxt: "Handler") -> "Handler": ...
@abc.abstractmethod
def handle(self, context: Context) -> Optional[Result]: ...
class BaseHandler(Handler):
"""提供鏈結能力的基底類別（每個實例各自持有 next）"""
def __init__(self) -> None:
self._next: Optional[Handler] = None
def set_next(self, nxt: Handler) -> Handler:
self._next = nxt
return nxt
def _pass(self, context: Context) -> Optional[Result]:
"""將案件傳遞給下一個節點，若無則終止"""
if self._next:
return self._next.handle(context)
return None
# --- 具體處理節點 ---
class IntakeHandler(BaseHandler):
def handle(self, context: Context) -> Optional[Result]:
log_msg = f"1. [Intake] 接收案件 {context.case.case_id}..."
context.trace.append(self.__class__.__name__)
logging.info(log_msg)
return self._pass(context)
class RedactPIIHandler(BaseHandler):
def handle(self, context: Context) -> Optional[Result]:
case = context.case
if case.meta.get("night") and case.payload.get("media"):
log_msg = f"2. [RedactPII] 為案件 {case.case_id} 進行去識別化..."
context.trace.append(self.__class__.__name__)
logging.info(log_msg)
new_payload = case.payload.copy()
new_payload["media"] = "REDACTED_FOR_PRIVACY"
context.case = replace(case, payload=new_payload)
return self._pass(context)
class JurisdictionHandler(BaseHandler):
def handle(self, context: Context) -> Optional[Result]:
case = context.case
context.trace.append(self.__class__.__name__)
if not case.meta.get("inJurisdiction", True):
log_msg = f"3. [Jurisdiction] 案件 {case.case_id} 非本轄，短路！"
logging.info(log_msg)
return Result(code=200, reason="FORWARDED")
log_msg = f"3. [Jurisdiction] 案件 {case.case_id} 確認為本轄。"
logging.info(log_msg)
return self._pass(context)
class EvidenceCheckHandler(BaseHandler): # Stub
def handle(self, context: Context) -> Optional[Result]:
log_msg = f"4. [Evidence] 檢查案件 {context.case.case_id} 證據..."
context.trace.append(self.__class__.__name__)
logging.info(log_msg)
return self._pass(context)
class LegalCheckHandler(BaseHandler): # Stub
def handle(self, context: Context) -> Optional[Result]:
log_msg = f"5. [Legal] 檢查案件 {context.case.case_id} 法規..."
context.trace.append(self.__class__.__name__)
logging.info(log_msg)
return self._pass(context)
class DispatchHandler(BaseHandler):
def handle(self, context: Context) -> Optional[Result]:
log_msg = f"6. [Dispatch] 派遣人員處理案件 {context.case.case_id}。"
context.trace.append(self.__class__.__name__)
logging.info(log_msg)
# 鏈的成功終點
return Result(code=200, reason="DISPATCHED")
# --- 動態裝配與模擬執行 ---
def build_chain(night: bool) -> Handler:
first = IntakeHandler()
current = first
if night:
current = current.set_next(RedactPIIHandler())
# 完整串接
(current.set_next(JurisdictionHandler())
.set_next(EvidenceCheckHandler())
.set_next(LegalCheckHandler())
.set_next(DispatchHandler()))
return first
# 模擬執行
logging.info("--- 執行夜間、本轄案件 ---")
night_context = Context(Case("ID-001", "noise", {"media": "clip.mp3"}, {"night": True, "inJurisdiction": True}))
night_chain = build_chain(night=True)
result1 = night_chain.handle(night_context)
logging.info(f"最終結果: {result1}, 處理軌跡: {night_context.trace}\n")
logging.info("--- 執行日間、非本轄案件 ---")
day_context = Context(Case("ID-002", "construction", {}, {"night": False, "inJurisdiction": False}))
day_chain = build_chain(night=False)
result2 = day_chain.handle(day_context)
logging.info(f"最終結果: {result2}, 處理軌跡: {day_context.trace}")
當你看到以下跡象時，可能代表責任鏈被誤用了：
🚩 萬能處理器：一個 Handler 做了所有事，判斷案件類型、處理A、處理B...，這就退化成了 if/elif 巨獸，鏈的意義蕩然無存。
🚩 在呼叫端硬編排流程：如果在 main 函式裡用 if/else 決定 handlerA.handle() 之後要不要呼叫 handlerB.handle()，代表你根本沒把流程封裝進鏈裡。
🚩 沒有短路機制：如果每個節點都「一定會」把請求往下傳，那它就只是一個單純的 Pipeline，違背了 CoR 可隨時攔截、接手的精髓。
🚩 跨節點共享可變狀態：不同 Handler 偷偷去修改一個共用的全域變數或 Context 物件，會導致難以追蹤的副作用，也破壞了節點的獨立性。（註：我們範例中的 Context 雖然可變，但其職責被嚴格限定於追蹤與傳遞不可變的 Case，避免了這個問題。）
EIP/EDA (中觀)：我們可以把每個 Handler 視為一個 Filter（過濾器）。在案件進來時，可以先通過一個 Content-based Router（內容路由器），根據案件類型（噪音、施工）將其分派到不同的專屬 Pipeline（處理鏈）中。
Actor/MAS (宏觀)：在多代理系統的視角下，每個 Handler 都是一個獨立的代理人（Agent），擁有特定技能（如法規比對、證據檢核）。它們在 DF（黃頁）上註冊自己。Tess 作為管理者，只需要向 DF 查詢「能處理夜間緊急案件的代理清單」，然後按協定依序交付任務即可。
現在，讓我們用 Kaito 設計的責任鏈來實際演練開場的棘手案件。Tess 透過 build_chain(night=True) 取得一條夜間加急鏈。
情境演練：
案件進來：包含地點、錄音檔、照片的複合型陳情案。
IntakeHandler 接收，建立追蹤 ID。
RedactPIIHandler 檢查到 night=True 且有媒體檔，自動去識別化。
JurisdictionHandler 檢查地點...
狀況A（本轄）：確認為本轄案件，放行！請求繼續傳遞。驗收通過：30 分鐘內完成派遣。
真實 Log 軌跡：
['IntakeHandler', 'RedactPIIHandler', 'JurisdictionHandler', 'EvidenceCheckHandler', 'LegalCheckHandler', 'DispatchHandler']
最終結果: Result(code=200, reason='DISPATCHED')
狀況B（非本轄）：觸發短路！JurisdictionHandler 立刻接手，回傳 Result，流程終止。驗收通過：Tess 僅收到轉交成功的 ack。
真實 Log 軌跡：
['IntakeHandler', 'JurisdictionHandler']
最終結果: Result(code=200, reason='FORWARDED')
驗收通過，Tess 不再需要手動追蹤進度，流程變得自動、透明且可擴展。
要確保這條責任鏈穩固可靠，Kaito 規劃了幾種測試：
契約測試 (Contract Testing)：確保每一個 Handler 都確實實作了 handle 和 set_next 介面。並且，它們的輸出語意一致（回傳 None 或 Result），不應洩漏各自內部的資料型別。
重排測試 (Reordering Test)：驗證當我們在鏈中插入一個新節點（如此次的 RedactPIIHandler）後，原有的呼叫端程式碼完全不需要修改，且整體驗收依然通過。
短路測試 (Short-circuit Test)：建立一個 inJurisdiction=False 的測試案例，斷言後續的 EvidenceCheckHandler 等節點的 handle 方法「從未被呼叫」。
偽造資料測試 (Bad Data Test)：提供一個缺少必要欄位的案件，斷言 IntakeHandler 應直接回傳錯誤或中斷流程。
嘿，身為 Codetopia 的一員，換你動動腦了！
✍️ 動手題：
市府週末要舉辦大型音樂祭，需要一條「活動週末版」的處理鏈。請在不修改任何既有 Handler 的前提下，設計一個新的 CrowdControlHandler（群眾管制規則檢查），並將它插入到 JurisdictionHandler 和 EvidenceCheckHandler 之間。同時，請撰寫一個「重排測試」來驗證這個新節點有確實被調用。
🤔 二選一：
如果遇到一個「證據不足，但同一個市民在 1 小時內重複投訴 3 次」的案件，你會選擇：
A： 在 EvidenceCheckHandler 中偵測到重複投訴模式，即刻短路，並回退要求市民補件。
B： 允許案件繼續流到 LegalCheckHandler，由法規專家節點來裁決，是否先以「警示勸導單」的形式暫時了結此案？
請在下方留言你選 A 或 B，並附上一句你的理由！
📩 範例留言：我選 B｜理由：讓法規節點決定「警示單」更符合權責分工。
今天我們見證了責任鏈如何將一團亂麻的跨局處流程，梳理成一條清晰、可動態調整的處理管道。
二十字摘要：逐級接棒可短路，責任清晰易重排，夜間加急穩交付。
當案件可以被順暢地分派後，下一個問題來了：這些「派遣」的動作本身，是否也能被更好地管理？例如，如何支援撤銷、重做，或甚至排程執行？
明日預告：Day 17｜Command（派工命令）—— 把派遣與撤銷/重做包成命令，進佇列好排程。
為了確保在不支援 Mermaid 渲染的環境中也能正常閱讀，以下提供文中圖表的 ASCII 替代版本：
┌─────────────────┐
│ Handler │
│ <<interface>> │
│ │
│ +setNext(h) │
│ +handle(c) │
└─────────┬───────┘
│
▲
│ implements
┌─────────────────────┼─────────────────────┐
│ │ │
┌────▼────┐ ┌─────▼─────┐ ┌──────▼──────┐
│ Intake │ │ RedactPII │ │Jurisdiction │
│Handler │────────▶│ Handler │───────▶│ Handler │
└─────────┘ └───────────┘ └─────┬───────┘
│
▼
┌─────────────┐
│ Evidence │
│ Handler │───┐
└─────────────┘ │
▲ │
│ ▼
┌─────────────┐ │
│ Legal │ │
│ Handler │◄──┘
└─────┬───────┘
│
▼
┌─────────────┐
│ Dispatch │
│ Handler │
│ (end) │
└─────────────┘
案件輸入
│
▼
┌──────────┐ 接收案件
│ Intake │ ──────────────┐
│ Handler │ │
└────┬─────┘ │
│ ▼
▼ 有媒體檔?
┌──────────┐ 且夜間?
│RedactPII │ │
│ Handler │ ◄────────────┤
└────┬─────┘ 是 │ 否
│ │
▼ ▼
┌──────────┐ ┌──────────┐
│Jurisdic- │ │Jurisdic- │
│tion │ │tion │
│ Handler │ │ Handler │
└────┬─────┘ └────┬─────┘
│ │
▼ ▼
本轄? 本轄?
│ │
┌───┴───┐ ┌───┴───┐
│ │ │ │
是│ │否 是│ │否
│ │ │ │
│ ┌───▼───┐ │ ┌───▼────┐
│ │短路： │ │ │ 短路： │
│ │轉交 │ │ │ 轉交 │
│ │STOP! │ │ │ STOP! │
│ └───────┘ │ └────────┘
│ │
▼ ▼
┌──────────┐ ┌──────────┐
│Evidence │ │Evidence │
│Handler │ ────────│Handler │
└────┬─────┘ └────┬─────┘
│ │
▼ ▼
┌──────────┐ ┌──────────┐
│ Legal │ │ Legal │
│ Handler │ ────────│ Handler │
└────┬─────┘ └────┬─────┘
│ │
▼ ▼
┌──────────┐ ┌──────────┐
│Dispatch │ │Dispatch │
│Handler │ │Handler │
└────┬─────┘ └────┬─────┘
│ │
▼ ▼
通知&結案 通知&結案
Tess案件經理 DF黃頁服務 管轄代理 法規代理 派遣代理
│ │ │ │ │
│ lookup(夜間鏈) │ │ │ │
├──────────────▶ │ │ │ │
│ │ │ │ │
│ ◄─────────────┤ │ │ │
│ [Intake,Redact,Jurisdiction,Evidence,Legal,Dispatch]
│ │ │ │ │
│ handle(案件) │ │ │ │
├─────────────────────────▶ │ │ │
│ │ │ │ │
│ ◄─────────────────────────┤ │ │
│ ok/forward │ │ │
│ │ │ │ │
│ handle(案件) │ │ │ │
├───────────────────────────────────────▶ │
│ │ │ │ │
│ ◄─────────────────────────────────────┤ │
│ ok │ │
│ │ │ │ │
│ dispatch(案件) │ │ │ │
├─────────────────────────────────────────────────▶ │
│ │ │ │ │
│ ◄───────────────────────────────────────────────┤ │
│ ack │
│ │ │ │ │
情境A：夜間本轄案件
案件ID-001 ──┬─▶ [Intake] ──┬─▶ [RedactPII] ──┬─▶ [Jurisdiction] ──┬─▶ [Evidence] ──┬─▶ [Legal] ──┬─▶ [Dispatch]
│ │ │ │ │ │
▼ ▼ ▼ ▼ ▼ ▼
接收OK 去識別OK 確認本轄 檢查證據OK 法規OK 派遣成功
│
▼
Result(200, 'DISPATCHED')
情境B：非本轄案件（短路）
案件ID-002 ──┬─▶ [Intake] ──┬─▶ [Jurisdiction] ──┬─▶ ✋ 短路！
│ │ │
▼ ▼ ▼
接收OK 非本轄 Result(200, 'FORWARDED')
│
▼
[Evidence] ── 未執行
│
▼
[Legal] ── 未執行
│
▼
[Dispatch] ── 未執行
夜間模式組裝：
┌─────────────────────────────────────────────────────────────┐
│ build_chain(night=True) │
│ │
│ first ─┐ │
│ │ │
│ ▼ ▼ ▼ ▼ ▼ │
│ [Intake] ──▶ [RedactPII] ──▶ [Jurisdiction] ──▶ ... ──▶ │
│ ▲ │
│ │ │
│ if night=True │
│ 自動插入 │
└─────────────────────────────────────────────────────────────┘
日間模式組裝：
┌─────────────────────────────────────────────────────────────┐
│ build_chain(night=False) │
│ │
│ first ─┐ │
│ │ │
│ ▼ ▼ ▼ ▼ │
│ [Intake] ──────────────▶ [Jurisdiction] ──▶ ... ──▶ │
│ │
│ 跳過 RedactPII │
│ (無夜間需求) │
└─────────────────────────────────────────────────────────────┘
iThome鐵人賽
看影片追技術