Day 07: Function Calling 實戰 (下)：處理複雜參數、錯誤捕捉與 Tool Chaining (Function Calling Part 2)

本日核心價值 (Core Focus): 把昨日單次查詢 loop 升級成可驗證、可失敗、可熔斷的 Tool Chaining：nested 參數、enum 檢核、timeout，以及統一錯誤 JSON {ok,error,retryable}。示範鏈：lookup customer → list open orders → create ticket，最多 5 hop，並加 circuit breaker。

概念說明與實戰情境 (Overview)

客服場景很少只打一支 API。實際鏈是：用 email 找顧客 → 列出未結訂單 → 開 ticket。任一步 timeout、enum 填錯、或下游 500，都不該讓 Python traceback 冒進 Chat Completions。約定：每個 tool 只回 JSON；成功 {ok:true,data}，失敗 {ok:false,error,retryable}。程式在進 I/O 前做 enum / required 檢核，I/O 加 timeout；連續失敗熔斷。模型看到 retryable:true 可改參數再試，看到 false 應停止並向使用者說明。昨日的 loop 仍然適用，本日補上參數形狀、錯誤契約與 hop / 熔斷這三道閘。

關鍵操作與範例 (Implementation & Example)

三支 tool 的職責切清楚，避免一個函式包整條業務：

priority / category / region / status 用 JSON Schema enum，執行前再驗證一次——Schema 不能替代 runtime 檢核。

錯誤契約固定，不要有時回字串有時回 dict：

{"ok": false, "error": "timeout after 2.0s calling list_open_orders", "retryable": true}

retryable: true：timeout、429、暫時 5xx。false：enum 非法、顧客不存在、必要欄位缺失（再試也沒用）。把這份契約當成 tool 的 HTTP status：可重試等價於 503，不可重試等價於 400。模型沒有 TCP，它只看 JSON。

Nested 參數的目的是分層，不是炫技。query 屬於查找條件，ticket 屬於寫入命令；扁平化成十個 top-level 欄位時，模型更容易漏 region 或把 priority 填進 status。Schema 寫 nested 之後，runtime 仍要確認 query / ticket 真的是 object，否則 fn(**args) 會把字串拆成非法 keyword。

"""tool_chaining.py
依賴: pip install openai
環境: OPENAI_API_KEY；OPENAI_MODEL 預設 gpt-4o（可換）
執行: python tool_chaining.py
"""
from __future__ import annotations

import json
import os
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout
from typing import Any, Callable

from openai import OpenAI

MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o")
MAX_HOPS = 5
TOOL_TIMEOUT_SEC = 2.0
BREAKER_THRESHOLD = 3

REGION = {"TW", "JP", "US"}
ORDER_STATUS = {"open", "pending_payment"}
PRIORITY = {"low", "medium", "high", "urgent"}
CATEGORY = {"shipping", "billing", "product", "other"}

CUSTOMERS = {
    ("ada@example.com", "TW"): {"customer_id": "C-001", "name": "Ada"},
}
ORDERS = {
    "C-001": [
        {"order_id": "ORD-1001", "status": "open", "sku": "SKU-WIDGET-M"},
        {"order_id": "ORD-1002", "status": "shipped", "sku": "SKU-GADGET-L"},
    ]
}
TICKETS: list[dict[str, Any]] = []

TOOLS: list[dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "lookup_customer",
            "description": "用 email + region 查找顧客。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "object",
                        "properties": {
                            "email": {"type": "string"},
                            "region": {"type": "string", "enum": ["TW", "JP", "US"]},
                        },
                        "required": ["email", "region"],
                        "additionalProperties": False,
                    }
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_open_orders",
            "description": "列出顧客未結訂單。",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "status": {"type": "string", "enum": ["open", "pending_payment"]},
                },
                "required": ["customer_id", "status"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "create_ticket",
            "description": "為指定訂單建立客服 ticket。",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "order_id": {"type": "string"},
                    "ticket": {
                        "type": "object",
                        "properties": {
                            "priority": {
                                "type": "string",
                                "enum": ["low", "medium", "high", "urgent"],
                            },
                            "category": {
                                "type": "string",
                                "enum": ["shipping", "billing", "product", "other"],
                            },
                            "subject": {"type": "string"},
                            "body": {"type": "string"},
                        },
                        "required": ["priority", "category", "subject", "body"],
                        "additionalProperties": False,
                    },
                },
                "required": ["customer_id", "order_id", "ticket"],
                "additionalProperties": False,
            },
        },
    },
]


def fail(error: str, retryable: bool) -> dict[str, Any]:
    return {"ok": False, "error": error, "retryable": retryable}


def ok(data: Any) -> dict[str, Any]:
    return {"ok": True, "data": data}


def lookup_customer(query: dict[str, Any]) -> dict[str, Any]:
    email = str(query.get("email", "")).strip().lower()
    region = str(query.get("region", "")).upper()
    if region not in REGION:
        return fail(f"invalid region: {region}", False)
    row = CUSTOMERS.get((email, region))
    if row is None:
        return fail("customer_not_found", False)
    return ok(row)


def list_open_orders(customer_id: str, status: str) -> dict[str, Any]:
    if status not in ORDER_STATUS:
        return fail(f"invalid status: {status}", False)
    time.sleep(0.05)  # 模擬 I/O；可改 sleep(3) 測 timeout
    rows = [o for o in ORDERS.get(customer_id, []) if o["status"] == status]
    return ok({"orders": rows})


def create_ticket(customer_id: str, order_id: str, ticket: dict[str, Any]) -> dict[str, Any]:
    pr = ticket.get("priority")
    cat = ticket.get("category")
    if pr not in PRIORITY:
        return fail(f"invalid priority: {pr}", False)
    if cat not in CATEGORY:
        return fail(f"invalid category: {cat}", False)
    if not ticket.get("subject") or not ticket.get("body"):
        return fail("subject and body are required", False)
    owned = {o["order_id"] for o in ORDERS.get(customer_id, [])}
    if order_id not in owned:
        return fail("order_not_owned_by_customer", False)
    ticket_id = f"TCK-{len(TICKETS) + 1:04d}"
    TICKETS.append({"ticket_id": ticket_id, "customer_id": customer_id, "order_id": order_id, **ticket})
    return ok({"ticket_id": ticket_id})


DISPATCH: dict[str, Callable[..., dict[str, Any]]] = {
    "lookup_customer": lambda **kw: lookup_customer(kw["query"]),
    "list_open_orders": lambda **kw: list_open_orders(kw["customer_id"], kw["status"]),
    "create_ticket": lambda **kw: create_ticket(kw["customer_id"], kw["order_id"], kw["ticket"]),
}


class CircuitBreaker:
    def __init__(self, threshold: int) -> None:
        self.threshold = threshold
        self.consecutive_failures = 0
        self.open = False

    def record(self, payload: dict[str, Any]) -> None:
        if payload.get("ok"):
            self.consecutive_failures = 0
            return
        self.consecutive_failures += 1
        if self.consecutive_failures >= self.threshold:
            self.open = True


def run_one(name: str, raw_args: str, breaker: CircuitBreaker) -> str:
    if breaker.open:
        return json.dumps(fail("circuit_open: too many consecutive tool errors", False), ensure_ascii=False)
    try:
        args = json.loads(raw_args or "{}")
        if not isinstance(args, dict):
            raise ValueError("arguments must be object")
    except (json.JSONDecodeError, ValueError) as exc:
        payload = fail(f"invalid_json_arguments: {exc}", False)
        breaker.record(payload)
        return json.dumps(payload, ensure_ascii=False)

    fn = DISPATCH.get(name)
    if fn is None:
        payload = fail(f"unknown_tool: {name}", False)
        breaker.record(payload)
        return json.dumps(payload, ensure_ascii=False)

    try:
        with ThreadPoolExecutor(max_workers=1) as pool:
            payload = pool.submit(fn, **args).result(timeout=TOOL_TIMEOUT_SEC)
    except FuturesTimeout:
        payload = fail(f"timeout after {TOOL_TIMEOUT_SEC}s calling {name}", True)
    except TypeError as exc:
        payload = fail(f"argument_mismatch: {exc}", False)
    except Exception as exc:  # 下游未預期錯誤：可重試但不要漏給模型 traceback 細節
        payload = fail(f"internal_error in {name}: {type(exc).__name__}", True)

    breaker.record(payload)
    return json.dumps(payload, ensure_ascii=False)


def chain(user_text: str) -> str:
    client = OpenAI()
    breaker = CircuitBreaker(BREAKER_THRESHOLD)
    hops = 0
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": (
                "你是客服編排器。必須依序：lookup_customer → list_open_orders → create_ticket。"
                "每步只信 tool JSON。ok=false 且 retryable=true 時可改參數重試；"
                "retryable=false 或 circuit_open 時停止並向使用者說明。"
                "禁止在沒有 tool 成功結果時捏造 customer_id / ticket_id。"
            ),
        },
        {"role": "user", "content": user_text},
    ]

    while True:
        resp = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content or ""

        remaining = MAX_HOPS - hops
        if remaining <= 0:
            return f"stopped: exceeded MAX_HOPS={MAX_HOPS}"

        batch = msg.tool_calls[:remaining]
        hops += len(batch)

        messages.append(
            {
                "role": "assistant",
                "content": msg.content,
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                    }
                    for tc in batch
                ],
            }
        )
        for tc in batch:
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": run_one(tc.function.name, tc.function.arguments, breaker),
                }
            )
        if breaker.open:
            messages.append(
                {
                    "role": "system",
                    "content": "circuit breaker is open; do not call more tools; summarize failure to the user.",
                }
            )


def main() -> None:
    print(
        chain(
            "顧客 ada@example.com（台灣）抱怨 ORD-1001 還沒出貨，"
            "請查未結訂單並開一張 high / shipping 的 ticket。"
        )
    )


if __name__ == "__main__":
    main()

Chaining 規則寫在 system Prompt 不夠，要在 runtime 強制：MAX_HOPS = 5 計算的是 tool 執行次數，不是 LLM round。同一則 assistant 若一次發 3 個 tool_calls，hop += 3。超限就停，把已執行結果交給模型收尾，或直接回 stopped: exceeded MAX_HOPS。不要把 hop 設成「看起來很大的 20」——每 hop 都是一次下游 I/O 加一次後續 completion，成本與延遲是乘積。

Circuit breaker 與 hop 上限正交：hop 防「模型愛呼叫」；breaker 防「下游持續失敗還一直打」。BREAKER_THRESHOLD = 3 表示連續 3 次 ok:false 就 circuit_open。成功必須歸零；不要用「歷史累計失敗」當熔斷，否則偶發 timeout 會永久鎖死整條客服鏈。熔斷開啟後，下一則 tool 結果應是 retryable: false 的 circuit_open，並加一則 system 指示「不要再呼叫工具、向使用者說明失敗」。若只熔斷卻繼續 tool_choice=auto，模型仍可能空轉耗 hop。

錯誤處理不可省略的三層：（1）JSON 解析與 enum / 必填，失敗 retryable: false；（2）timeout 與未預期 exception，失敗 retryable: true，訊息只含短碼與型別名，不含 traceback；（3）業務不變式，例如 order_id 必須屬於該 customer_id，失敗同樣不可重試。漏任何一層，Chaining 都會在 production 變成「亂開 ticket」或「把內部 host 餵給模型」。

測 timeout：把 list_open_orders 的 sleep(0.05) 改成 sleep(3)，應得到 retryable: true。測 enum：手動呼叫 create_ticket(..., ticket={"priority":"asap",...}) 應 retryable: false。測熔斷：連續丟三次非法 JSON arguments。這三條要寫單元測試，不能只靠模型「看起來有處理」。

注意事項與常見失敗 (Pitfalls)

• 把 exception 字串當 tool 結果： traceback 含路徑、SQL、內部 host，會進模型 Context 也可能進日誌。一律收斂成 {ok,error,retryable}；error 給短碼 + 安全說明。正式環境還要把同一份 JSON 寫進內部 log（可含 request id），與給模型的字串分開，方便對帳。
• 相信 Schema 就跳過 enum 檢核： 模型仍可能輸出 "priority": "asap"。非法值必須 retryable: false，否則模型會用同一壞值重試直到 hop 用盡。單元測試應直接呼叫 run_one("create_ticket", '{"ticket":{"priority":"asap"}}', ...)，不要只測「模型高興時」。
• nested 參數直接 fn(**args) 卻未驗證 query / ticket 是 dict： TypeError 要轉成 argument_mismatch + retryable: false。缺 query 時不要用空 dict 蒙混，否則會查到錯誤顧客。
• timeout 包在 HTTP client 卻沒包純 Python 工具： CPU 死迴圈或漏設的 sleep 仍會卡住 worker。用 ThreadPoolExecutor.result(timeout=...)（或同等機制）當統一上限。逾時後不要再等原執行緒結束才回模型，先回 retryable: true。
• 忽略 retryable： 對 customer_not_found 重試是浪費 Token 與 hop。system Prompt 要明講，runtime 也可在連續相同 error 時提早停。可重試錯誤建議帶「下一試可改什麼」（例如換 region），否則模型只會再送同一包參數。
• create_ticket 不檢查訂單歸屬： 模型可能把 A 顧客的 order_id 寫進 B 的 ticket。這是授權 bug，不是 Prompt 問題；函式內必須驗證 order_id in owned。客服系統的寫入 tool 都要做這層，不能只靠「Prompt 叫它先 lookup」。
• breaker 在部分成功後未歸零： 第一步偶發 timeout、第二步成功，應允許後續 create_ticket。只對連續失敗計數。熔斷開啟後要有明確的使用者文案（「下游暫時不可用，請稍後再試」），不要回半成品 ticket id。

本日總結 (Takeaways)

• Tool Chaining 是「多 hop 編排 + 每 hop 可失敗」，不是把三支 API 寫進同一個 function。
• 錯誤一律 {ok:false,error,retryable}；timeout / 5xx 可重試，enum 與授權失敗不可重試。
• Nested 參數與 enum 要 Schema 與 runtime 雙檢；I/O 必須有 timeout。
• MAX_HOPS=5 限制模型呼叫次數；circuit breaker 限制連續失敗。兩者都要實作，不能只寫在 Prompt。
• 業務不變式（訂單屬於該顧客）放在 tool 實作，不放在模型「善意」。

明日預告 (Next)

明日進入 多模態工作流 (Multimodal Workflow)：從 UI/UX 設計圖自動生成前端程式碼，用 vision 把設計圖拆成元件清單，再產出單一卡片的 HTML/CSS。