Pydantic AI Graph Beta API：條件分支 (Decisions)

在設計 AI 代理人工作流時，除了平行執行，流程中也常需要「條件判斷」與「動態路由」。在舊版的 Graph API 中，我們必須在 run() 方法內部寫 if/else 並 return 特定的節點實例。

在 Beta API 的 Builder 模式下，你可以利用特定的決策機制（如 g.decision()）來決定下一個連線方向。這使得你可以把判斷邏輯完全隔離在連線規則中，讓主流程的程式碼更加乾淨。

決策分支範例

你可以讓一個步驟回傳特定的標籤（例如 Literal 型別），然後在宣告 Graph 連線時，使用 g.decision() 搭配 .branch() 根據回傳值的型別來決定走向：

from typing import Literal
from pydantic_graph.beta import GraphBuilder, StepContext, TypeExpression

# 讓步驟根據條件回傳不同的標籤 (字串)
@g.step
async def check_quality(ctx: StepContext[None, None, str]) -> Literal['pass', 'fail']:
    # 模擬業務判斷或是 AI 評估的結果
    if "good" in ctx.inputs:
        return 'pass'
    return 'fail'

# 在定義 Graph 時，根據回傳值的型別進行條件分流
g.add(
    g.edge_from(g.start_node).to(check_quality),

    # 使用 g.decision() 與 TypeExpression 來配對回傳值
    g.edge_from(check_quality).to(
        g.decision()
        .branch(g.match(TypeExpression[Literal['pass']]).to(success_step))
        .branch(g.match(TypeExpression[Literal['fail']]).to(retry_step))
    )
)

透過這種設計，流程的判斷點 (Router) 與實際執行任務的節點被完美解耦，修改判斷規則時不再需要動到任務本身的程式碼。

分支優先權與預設分支 (Branch Priority & Catch-All)

在 g.decision() 中，分支的評估是由上而下依序進行的（類似 if-elif-else），系統會走入第一個符合條件的分支。因此，如果你需要一個類似 else 的「預設 (Fallback) 分支」，可以使用 TypeExpression[object] 放在最後面捕捉所有未匹配的結果：

g.add(
    g.edge_from(check_quality).to(
        g.decision()
        .branch(g.match(TypeExpression[Literal['pass']]).to(success_step))
        # 當前面所有條件都不符合時，就會走這條預設路徑 (Catch-All)
        .branch(g.match(TypeExpression[object]).to(default_step))
    )
)

自訂判斷邏輯 (Custom Matchers)

除了比對資料型別以外，g.match() 還提供了一個強大的 matches 參數，允許你傳入自訂的 lambda 函式，直接對數值或內容進行邏輯判斷：

@g.step
async def return_score(ctx: StepContext[None, None, None]) -> int:
    return 85

g.add(
    g.edge_from(g.start_node).to(return_score),
    g.edge_from(return_score).to(
        g.decision()
        # 根據回傳的分數大小，動態決定走向，且越上方的分支優先權越高
        .branch(g.match(TypeExpression[int], matches=lambda x: x >= 90).to(grade_a_step))
        .branch(g.match(TypeExpression[int], matches=lambda x: x >= 60).to(grade_b_step))
        .branch(g.match(TypeExpression[int]).to(fail_step)) # 上述皆非的 fallback
    )
)

這種做法極大地提升了路由設計的彈性，讓你可以把複雜的商業判斷規則集中且清晰地宣告在 Graph 的連線邏輯中！