Day 06: Function Calling 實戰 (上)：讓 ChatGPT 自動鏈接外部 API 與資料庫 (Function Calling Part 1)

本日核心價值 (Core Focus): 用 Chat Completions 的 tools= 把模型從「猜答案」改成「選工具 → 你執行 → 再回答」。本日實作兩支 tool：get_order 打假 REST、query_inventory 查 SQLite，並跑完整 tool loop。

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

客服問「ORD-1001 能不能出貨」，模型不該憑訓練資料發明庫存。Function Calling 的契約很窄：模型只輸出 tool_call（函式名 + JSON 參數），真正的 HTTP / SQL 由你的程式執行。執行結果以 role: "tool" 塞回 messages，模型再生成最終答案。本日用 gpt-4o（可換成團隊現用模型）示範最小可跑 loop：訂單 REST + 庫存 SQLite。這是 Day 03 JSON Schema 的執行面——Schema 不再只是輸出格式，而是工具介面。

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

Loop 固定四步，不要讓模型「直接回文字當查詢結果」：

1. chat.completions.create(..., tools=TOOLS)。
2. 若 finish_reason == "tool_calls"，逐筆解析 tool_call.function.name / arguments。
3. 在本機執行對應函式（HTTP 或 SQL），把回傳值 json.dumps。
4. messages append assistant（含 tool_calls）與各 tool 結果，再打一次 API；直到沒有新的 tool_call。

這四步是控制流，不是 Prompt 修辭。漏第 4 步等於把 tool 結果丟進垃圾桶，模型下一輪只會再猜。tool_choice="auto" 讓模型決定要不要呼叫；若你確定「這個 user 問題一定要查訂單」，可在第一輪用 tool_choice 指定 get_order，查完後改回 auto 或 none 讓它寫自然語言。不要全程 required，否則永遠無法結束。

為何拆成兩支 tool，而不是 get_order_and_inventory(order_id)？訂單主檔與庫存的來源、權限、逾時與失敗模式不同。REST 可能 404，SQLite 可能缺 SKU；拆開後模型可以只重試失敗的那一支，也方便你對 get_order 設 HTTP timeout、對 query_inventory 做 SQL 參數化。正式環境把假 REST 換成訂單服務、把 :memory: 換成唯讀 replica，loop 本身不必改。

工具 JSON Schema 必須讓模型「只填你願意執行的欄位」。additionalProperties: false 可減少亂加參數。description 寫使用時機（何時該呼叫、參數格式），不要寫業務散文；模型是靠 Schema + system 指令選工具，不是靠你在聊天裡「拜託查一下」。

"""function_calling_loop.py
依賴: pip install openai
環境: OPENAI_API_KEY 必填；OPENAI_MODEL 可選，預設 gpt-4o
執行: python function_calling_loop.py
"""
from __future__ import annotations

import json
import os
import sqlite3
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from openai import OpenAI

MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o")  # 可換 gpt-4o-mini / 團隊現用模型
FAKE_REST_PORT = 8765

ORDERS = {
    "ORD-1001": {
        "order_id": "ORD-1001",
        "status": "paid",
        "sku": "SKU-WIDGET-M",
        "qty": 2,
        "customer_id": "C-001",
    },
    "ORD-1002": {
        "order_id": "ORD-1002",
        "status": "shipped",
        "sku": "SKU-GADGET-L",
        "qty": 1,
        "customer_id": "C-002",
    },
}

TOOLS: list[dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "get_order",
            "description": "依訂單編號向訂單 REST API 查詢主檔（狀態、SKU、數量）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "訂單編號，例如 ORD-1001",
                    }
                },
                "required": ["order_id"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "query_inventory",
            "description": "依 SKU 查詢庫存數量與倉別（SQLite）。",
            "parameters": {
                "type": "object",
                "properties": {
                    "sku": {
                        "type": "string",
                        "description": "料號，例如 SKU-WIDGET-M",
                    }
                },
                "required": ["sku"],
                "additionalProperties": False,
            },
        },
    },
]


class OrderHandler(BaseHTTPRequestHandler):
    def log_message(self, fmt: str, *args: Any) -> None:
        return

    def do_GET(self) -> None:
        prefix = "/orders/"
        if not self.path.startswith(prefix):
            self.send_error(404, "not found")
            return
        order_id = self.path[len(prefix) :].strip("/")
        row = ORDERS.get(order_id)
        if row is None:
            body = json.dumps({"error": "order_not_found", "order_id": order_id}).encode()
            self.send_response(404)
        else:
            body = json.dumps(row).encode()
            self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def start_fake_rest() -> ThreadingHTTPServer:
    httpd = ThreadingHTTPServer(("127.0.0.1", FAKE_REST_PORT), OrderHandler)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd


def init_inventory_db() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:", check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute(
        """
        CREATE TABLE inventory (
            sku TEXT PRIMARY KEY,
            on_hand INTEGER NOT NULL,
            warehouse TEXT NOT NULL
        )
        """
    )
    conn.executemany(
        "INSERT INTO inventory (sku, on_hand, warehouse) VALUES (?, ?, ?)",
        [
            ("SKU-WIDGET-M", 8, "TW-TPE-01"),
            ("SKU-GADGET-L", 0, "TW-KHH-02"),
        ],
    )
    conn.commit()
    return conn


def get_order(order_id: str) -> dict[str, Any]:
    url = f"http://127.0.0.1:{FAKE_REST_PORT}/orders/{order_id}"
    req = Request(url, headers={"Accept": "application/json"})
    try:
        with urlopen(req, timeout=3) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except HTTPError as exc:
        payload = exc.read().decode("utf-8")
        return {"http_status": exc.code, "body": json.loads(payload)}
    except URLError as exc:
        return {"error": "rest_unreachable", "detail": str(exc.reason)}


def query_inventory(conn: sqlite3.Connection, sku: str) -> dict[str, Any]:
    row = conn.execute(
        "SELECT sku, on_hand, warehouse FROM inventory WHERE sku = ?",
        (sku,),
    ).fetchone()
    if row is None:
        return {"error": "sku_not_found", "sku": sku}
    return dict(row)


def run_tool(name: str, raw_args: str, conn: sqlite3.Connection) -> str:
    args = json.loads(raw_args or "{}")
    if name == "get_order":
        result = get_order(str(args["order_id"]))
    elif name == "query_inventory":
        result = query_inventory(conn, str(args["sku"]))
    else:
        result = {"error": "unknown_tool", "name": name}
    return json.dumps(result, ensure_ascii=False)


def tool_loop(user_text: str, conn: sqlite3.Connection) -> str:
    client = OpenAI()
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": (
                "你是訂單助理。庫存與訂單狀態只能來自 tool 結果，"
                "禁止臆測。需要資料時呼叫 get_order 或 query_inventory。"
            ),
        },
        {"role": "user", "content": user_text},
    ]

    for _ in range(8):
        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 ""

        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 msg.tool_calls
                ],
            }
        )
        for tc in msg.tool_calls:
            output = run_tool(tc.function.name, tc.function.arguments, conn)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": output,
                }
            )
    return "tool loop exceeded max turns"


def main() -> None:
    httpd = start_fake_rest()
    conn = init_inventory_db()
    try:
        answer = tool_loop("ORD-1001 現在能不能出貨？請說明庫存與訂單狀態。", conn)
        print(answer)
    finally:
        httpd.shutdown()
        conn.close()


if __name__ == "__main__":
    main()

同一輪可能出現兩個 tool_calls（先拿訂單再查 SKU）。你的 loop 必須支援一次 message.tool_calls 多筆，且每筆都要帶正確的 tool_call_id。漏 append assistant 那則含 tool_calls 的訊息，下一輪 API 會回 400。

messages 正確順序是：system → user → assistant（含 tool_calls）→ 各 role: "tool" → 最終 assistant。兩個 tool 就兩則 tool 訊息。回寫給 API 時，arguments 保持 JSON 字串；若你先 json.loads 再把 dict 塞回去，部分請求會 400。Debug 只印 name、參數與回傳前 200 字元，不要印 Authorization。

執行前設定 OPENAI_API_KEY。假 REST 綁 127.0.0.1:8765，庫存在 process 內的 :memory: SQLite，不需 Docker。建議三組對照：ORD-1001（paid + 有貨）、ORD-1002（shipped + 零庫存）、ORD-9999（REST 404）。若 tool 回 on_hand: 0 但最終答案仍說「可出貨」，問題在 system 沒有寫死「只能引用 tool JSON」，不是 Schema。本日錯誤先原樣回傳；結構化 {ok,error,retryable} 與 hop 上限留給明天。

對照時除了看最終中文，還要看 tool 名稱順序。ORD-1001 的合理鏈是 get_order → query_inventory（SKU 來自訂單）。若模型沒拿訂單就直接查庫存，system 應補「SKU 必須來自 get_order 結果」。同一 order_id 被呼叫三次，就是無限 loop 雛形。本機用 ThreadingHTTPServer 是為了讓 urlopen 真的走 HTTP，暴露 404 JSON 與 timeout；不要讓 get_order 直接讀 ORDERS dict，否則「模型呼叫 ≠ 網路成功」這課會消失。

這條 loop 與 Day 03 的 Structured Output 互補：Schema 約束參數，tool 結果約束事實。不要把訂單 JSON 貼進 Prompt 讓模型「假裝查過」；那會讓後續 Tool Chaining 無法接 timeout 與授權檢查。

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

• 漏掉 assistant tool_calls 再只丟 tool 結果： Chat Completions 要求成對：assistant（含 tool_calls）→ 各 role: "tool"。只 append tool 會 400。修法是先把 SDK 的 message 原樣轉成 dict 再追加，不要自己重造一則「只有 content」的 assistant。
• arguments 當 Python dict 用： API 給的是 JSON 字串。先 json.loads；解析失敗應回傳錯誤 JSON，不要讓 exception 炸掉整個 process。壞 JSON 在本日可回 {"error":"invalid_json"}，明天再升級成 {ok,error,retryable}。
• SQL 字串拼接 SKU： query_inventory 必須用 ? placeholder。Function Calling 的參數來自模型，視同不可信輸入。即便 Schema 寫了 type: string，模型仍可能塞進 ' OR 1=1；資料庫層的參數化不能省。
• tool_choice="required" 用在最終回答輪： 模型無法收斂成自然語言。查詢階段可用 auto；若要強制先查再答，查完後應拿掉 tools 或改 tool_choice="none"。否則使用者會一直看到空白或重複的 tool 請求。
• 把假 REST 當正式服務： 本日 ThreadingHTTPServer 僅供本機示範。正式環境要真實 base URL、timeout、auth header，且 API key 只從環境變數讀，禁止寫進 Prompt。訂單服務的 token 也不要當 tool 參數讓模型填。
• 無限 loop： 模型可能反覆呼叫同一 tool。本日用 range(8) 硬上限；明天會改成 hop 計數 + circuit breaker。除錯時把每一 round 的 finish_reason 與 tool 名稱印出來，很快就能看出是「查不到就再查」還是「查完仍 required」。

本日總結 (Takeaways)

• Function Calling 的單位是 tool_call，不是聊天句子；HTTP / SQL 永遠在你的 runtime 執行。
• tools= 的 JSON Schema 就是 API 契約：name、description、required 寫清楚，模型才選得對。
• 標準 loop：create → 執行 tools → append assistant + tool → 再 create，直到沒有 tool_calls。
• 兩類資料源可以並存：REST（訂單）與 SQLite（庫存）；模型負責編排，程式負責 I/O。
• gpt-4o 可換成現用模型；換模型時先用同一組 Schema 跑回歸，確認仍會發 tool_calls。

明日預告 (Next)

明日進入 Function Calling 實戰 (下)：處理複雜參數、錯誤捕捉與 Tool Chaining，處理 nested 參數、enum 檢核、timeout，以及 lookup customer → list orders → create ticket 的多 hop 編排。