前言：先「可控」再「自主」

上一篇我們認識了 LangGraph 的核心積木 — StateGraph、Node、Edge、State。這一篇要把積木組裝起來，蓋出第一種真正能上線的東西：Workflow（工作流）。

在往下走之前，有一個觀念一定要先建立清楚，因為它會決定你接下來每一個地端 AI 專案的設計方向：

• Workflow（工作流）：程式路徑是「事先定義好」的，LLM 只負責在既定的節點上完成任務，執行順序、分支條件都由開發者掌控。

• Agent（智能體）：LLM 自己決定要呼叫哪個工具、走哪條路、要不要繼續執行，路徑是「動態產生」的。

對於地端 AI 系統來說，Workflow 幾乎永遠是第一個該學、也是第一個該上線的形式。原因很現實：

1. 可預測、可除錯：地端環境往往沒有雲端那麼寬裕的算力去跑大量重試與探索，流程固定代表行為可預期，出錯時容易定位是哪個節點壞了。

2. 成本與延遲可控：每個節點呼叫幾次模型、用什麼模型，都是寫死的，不會因為 LLM「自由發揮」而暴增 token 用量或呼叫次數。

3. 容易導入審核與人工介入點：既定路徑上要插入「人工確認」「規則檢查」都很自然，這在企業內部系統（例如公文分類、報表產生、內部知識庫問答）特別重要。

所以這篇會完整拆解 LangGraph 官方文件中列出的幾種經典 Workflow 模式，並用地端場景改寫範例，讓你看完就能直接動手蓋一個屬於自己的 Workflow。

事前準備

沿用前幾篇的環境設定即可：

pip install langchain_core langgraph langchain-ollama

因為這系列的主軸是「地端 AI」，範例中會用 ChatOllama 取代雲端 API，你只要本機或內網已經跑起 Ollama 服務即可：

from langchain_ollama import ChatOllama

llm = ChatOllama(model="ornith-1.5:9b", temperature=0)

小提醒：Workflow 的節點如果需要「結構化輸出」（例如路由判斷、分類結果），模型的 instruction following 能力會直接影響穩定度。

LangGraph 提供兩種寫法：Graph API（用 StateGraph 明確畫出節點與邊）與 Functional API（用 @task / @entrypoint 裝飾一般函式）。這篇以 Graph API 為主，因為圖形化的節點/邊結構對「可控性」的呈現最直觀，也方便之後搭配 LangGraph Studio 除錯與視覺化。

模式一：Prompt Chaining（提示鏈）

概念：把一個大任務拆成好幾個小步驟，讓多次 LLM 呼叫依序接力，前一步的輸出是下一步的輸入。每一步都可以驗證、也可以插入「品質關卡」，不合格就轉去修正、合格就往下走。

適合場景：文件翻譯後再校對、報告先生成草稿再潤飾、公文先摘要再分類定級 — 任何可以拆成「可驗證子步驟」的任務。

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    raw_text: str
    summary: str
    is_qualified: str
    final_report: str


def summarize(state: State):
    """第一步：生成摘要"""
    msg = llm.invoke(f"請將以下內容整理成三點摘要：\n{state['raw_text']}")
    return {"summary": msg.content}


def quality_gate(state: State):
    """關卡：檢查摘要是否包含具體數字或結論"""
    if any(ch.isdigit() for ch in state["summary"]):
        return "通過"
    return "退回"


def refine(state: State):
    """未通過關卡時，補強摘要"""
    msg = llm.invoke(f"這份摘要缺乏具體數據，請補上關鍵數字：\n{state['summary']}")
    return {"summary": msg.content}


def finalize(state: State):
    """定稿"""
    return {"final_report": f"【最終摘要】\n{state['summary']}"}


builder = StateGraph(State)
builder.add_node("summarize", summarize)
builder.add_node("refine", refine)
builder.add_node("finalize", finalize)

builder.add_edge(START, "summarize")
builder.add_conditional_edges(
    "summarize", quality_gate, {"退回": "refine", "通過": "finalize"}
)
builder.add_edge("refine", "finalize")
builder.add_edge("finalize", END)

chain = builder.compile()
result = chain.invoke({"raw_text": "本季地端 GPU 使用率提升，但未附上數字..."})
print(result["final_report"])

可控性重點：quality_gate 是純程式邏輯，不靠 LLM 判斷，這代表這個關卡 100% 可預測、可單元測試。這正是 Workflow 相對 Agent 最大的優勢 — 關鍵決策點你可以選擇「要不要交給 LLM」。

模式二：Parallelization（平行化）

概念：把同一份輸入同時丟給多個獨立任務平行處理，最後再彙總結果。適合「互不依賴」的子任務，或是「同一件事跑多次、取多個角度」來提高信心。

適合場景：同時對一份合約做「風險條款掃描」、「用詞合規檢查」、「摘要生成」三件事，互不影響，最後合併成一份審閱報告。

class ReviewState(TypedDict):
    document: str
    risk_notes: str
    compliance_notes: str
    summary: str
    review_report: str


def check_risk(state: ReviewState):
    msg = llm.invoke(f"找出以下合約中的風險條款：\n{state['document']}")
    return {"risk_notes": msg.content}


def check_compliance(state: ReviewState):
    msg = llm.invoke(f"檢查以下合約用詞是否符合公司規範：\n{state['document']}")
    return {"compliance_notes": msg.content}


def summarize_doc(state: ReviewState):
    msg = llm.invoke(f"用一段話摘要以下合約重點：\n{state['document']}")
    return {"summary": msg.content}


def aggregate(state: ReviewState):
    report = (
        f"# 合約審閱報告\n\n## 摘要\n{state['summary']}\n\n"
        f"## 風險條款\n{state['risk_notes']}\n\n"
        f"## 合規檢查\n{state['compliance_notes']}"
    )
    return {"review_report": report}


builder = StateGraph(ReviewState)
builder.add_node("check_risk", check_risk)
builder.add_node("check_compliance", check_compliance)
builder.add_node("summarize_doc", summarize_doc)
builder.add_node("aggregate", aggregate)

builder.add_edge(START, "check_risk")
builder.add_edge(START, "check_compliance")
builder.add_edge(START, "summarize_doc")
builder.add_edge("check_risk", "aggregate")
builder.add_edge("check_compliance", "aggregate")
builder.add_edge("summarize_doc", "aggregate")
builder.add_edge("aggregate", END)

parallel_workflow = builder.compile()

可控性重點：三個節點各自獨立、互不干擾，其中任何一個換模型、換 prompt，都不會影響其他節點的行為，這對地端多模型混搭（例如小模型跑分類、大模型跑摘要）非常實用，也方便日後個別汰換。

模式三：Routing（路由）

概念：先用一個節點判斷「這個輸入該交給誰處理」，再依判斷結果導向不同的專責節點。這是把「單一大而全的 prompt」拆解成多個「小而專精的 prompt」的關鍵技巧。

適合場景：內部客服工單先分類（帳務 / 技術 / 一般諮詢），再導向對應的處理節點；地端知識庫問答先判斷問題類型，再決定要不要觸發 RAG 檢索。

from typing_extensions import Literal
from pydantic import BaseModel, Field
from langchain.messages import HumanMessage, SystemMessage


class Route(BaseModel):
    category: Literal["帳務", "技術", "一般諮詢"] = Field(
        description="這張工單應該被分派到哪一類"
    )


router = llm.with_structured_output(Route)


class TicketState(TypedDict):
    ticket: str
    category: str
    reply: str


def classify(state: TicketState):
    decision = router.invoke(
        [
            SystemMessage(content="請判斷這張客服工單屬於哪一類。"),
            HumanMessage(content=state["ticket"]),
        ]
    )
    return {"category": decision.category}


def handle_billing(state: TicketState):
    msg = llm.invoke(f"請以帳務客服的身份回覆：\n{state['ticket']}")
    return {"reply": msg.content}


def handle_tech(state: TicketState):
    msg = llm.invoke(f"請以技術支援的身份回覆：\n{state['ticket']}")
    return {"reply": msg.content}


def handle_general(state: TicketState):
    msg = llm.invoke(f"請以一般客服的身份回覆：\n{state['ticket']}")
    return {"reply": msg.content}


def route_decision(state: TicketState):
    return {"帳務": "handle_billing", "技術": "handle_tech", "一般諮詢": "handle_general"}[
        state["category"]
    ]


builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("handle_billing", handle_billing)
builder.add_node("handle_tech", handle_tech)
builder.add_node("handle_general", handle_general)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_decision,
    {"handle_billing": "handle_billing", "handle_tech": "handle_tech", "handle_general": "handle_general"},
)
builder.add_edge("handle_billing", END)
builder.add_edge("handle_tech", END)
builder.add_edge("handle_general", END)

router_workflow = builder.compile()

可控性重點：路由節點雖然由 LLM 判斷，但用了 with_structured_output 限制輸出只能是三個固定選項之一，把「LLM 的不確定性」框在一個有限、可驗證的範圍內，這是在 Workflow 裡引入 LLM 判斷、又不失控的標準做法。

模式四：Orchestrator Worker（協調工作者）

概念：與 Parallelization 不同的地方在於，子任務的「數量」與「內容」事先不知道，要由一個「協調者」節點動態規劃出來，再分派給多個「工作者」節點平行執行，最後彙整。

適合場景：地端知識庫的批次更新（不知道有多少份文件要改）、報告生成（章節數量依主題而定）、跨多檔案的程式碼修改。

LangGraph 對這種「動態產生節點」的情境提供了 Send API：

import operator
from typing import Annotated, List
from langgraph.types import Send


class Section(BaseModel):
    name: str = Field(description="報告章節名稱")
    description: str = Field(description="這個章節要涵蓋的重點")


class Sections(BaseModel):
    sections: List[Section]


planner = llm.with_structured_output(Sections)


class ReportState(TypedDict):
    topic: str
    sections: list[Section]
    completed_sections: Annotated[list, operator.add]
    final_report: str


class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, operator.add]


def orchestrator(state: ReportState):
    plan = planner.invoke(f"請為主題「{state['topic']}」規劃報告章節。")
    return {"sections": plan.sections}


def write_section(state: WorkerState):
    msg = llm.invoke(
        f"請撰寫報告章節。\n章節名稱：{state['section'].name}\n重點：{state['section'].description}"
    )
    return {"completed_sections": [msg.content]}


def synthesizer(state: ReportState):
    return {"final_report": "\n\n---\n\n".join(state["completed_sections"])}


def assign_workers(state: ReportState):
    return [Send("write_section", {"section": s}) for s in state["sections"]]


builder = StateGraph(ReportState)
builder.add_node("orchestrator", orchestrator)
builder.add_node("write_section", write_section)
builder.add_node("synthesizer", synthesizer)

builder.add_edge(START, "orchestrator")
builder.add_conditional_edges("orchestrator", assign_workers, ["write_section"])
builder.add_edge("write_section", "synthesizer")
builder.add_edge("synthesizer", END)

orchestrator_worker = builder.compile()

可控性重點：即使子任務數量是動態的，整體流程「規劃 → 分派 → 彙整」三個階段依然固定不變，Send 只是讓同一個節點被平行實例化多次；這跟 Agent 那種「LLM 自己決定要不要規劃、要不要分派」有本質上的不同。

模式五：Evaluator Optimizer（評估優化器）

概念：一個節點負責產出，另一個節點負責「用固定標準」評估產出，不合格就帶著回饋重新產出，直到通過為止。這是把「反覆修改」這個人類常見的工作習慣，變成可控的迴圈。

適合場景：翻譯品質迭代、地端生成內容的合規檢查、程式碼產生後的自動測試回饋迴圈。

class DraftState(TypedDict):
    topic: str
    draft: str
    feedback: str
    verdict: str


class Feedback(BaseModel):
    verdict: Literal["合格", "不合格"] = Field(description="是否符合標準")
    feedback: str = Field(description="不合格時，說明需要修改的地方")


evaluator = llm.with_structured_output(Feedback)


def generate(state: DraftState):
    if state.get("feedback"):
        msg = llm.invoke(
            f"請根據以下回饋重寫內容：\n主題：{state['topic']}\n回饋：{state['feedback']}"
        )
    else:
        msg = llm.invoke(f"請寫一段關於「{state['topic']}」的說明文字。")
    return {"draft": msg.content}


def evaluate(state: DraftState):
    result = evaluator.invoke(f"請評估以下內容是否合格：\n{state['draft']}")
    return {"verdict": result.verdict, "feedback": result.feedback}


def route_verdict(state: DraftState):
    return "通過" if state["verdict"] == "合格" else "重寫"


builder = StateGraph(DraftState)
builder.add_node("generate", generate)
builder.add_node("evaluate", evaluate)

builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")
builder.add_conditional_edges(
    "evaluate", route_verdict, {"通過": END, "重寫": "generate"}
)

optimizer_workflow = builder.compile()

可控性重點：務必替這種迴圈設一個「最大重試次數」的保護機制（可以在 State 裡加一個 retry_count 欄位並在條件邊裡檢查），避免模型一直卡在「不合格」的無窮迴圈裡，這在地端環境尤其重要，因為每一次重試都是實打實的算力消耗。

五種模式怎麼選？

實務上這五種模式很少單獨存在，更常見的是互相嵌套，例如 Routing 之後接一段 Prompt Chaining，或是 Orchestrator-Worker 裡的每個 Worker 內部又是一個 Evaluator-Optimizer 迴圈。重點是：先想清楚哪些決策一定要交給 LLM、哪些可以用程式邏輯鎖死，這條界線畫得越清楚，你的地端 AI 系統就越穩、越好除錯。

小結

這篇把 LangGraph 官方文件中的五種經典 Workflow 模式，逐一改寫成貼近地端場景的範例：提示鏈、平行化、路由、協調工作者、評估優化器。它們的共通點是：流程骨架永遠是開發者寫死的，LLM 只在被指定的節點上發揮作用。