綜合實戰:客服 API 服務上線

Step 1 — 建立專案結構

mkdir cs-api && cd cs-api
python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn anthropic pydantic python-dotenv

目錄結構:

cs-api/
├── main.py          # FastAPI 入口
├── config.py        # 金鑰與設定
├── prompts/
│   ├── system.txt   # System Prompt 主體
│   └── faq.txt      # FAQ 靜態內容
├── schemas.py       # 結構化輸出 Pydantic 模型
├── retry.py         # 重試邏輯
└── .env             # 金鑰(不進 git!)

.env 檔——這個檔案要加進 .gitignore,永遠不能推上 git:

ANTHROPIC_API_KEY=sk-ant-...
ANTHROPIC_API_KEY_BACKUP=sk-ant-...   # 備用金鑰
API_SECRET_TOKEN=your-own-service-token

requirements.txt(部署時用):

fastapi==0.115.0
uvicorn[standard]==0.32.0
anthropic>=0.42.0
pydantic>=2.7.0
python-dotenv>=1.0.0

Step 2 — System Prompt 注入 FAQ + Prompt Caching

prompts/system.txt(主體,幾乎不變,適合快取):

你是「Aria」,{company} 的官方客服助理。
- 回覆語言:繁體中文
- 語氣:專業但親切,每次回覆不超過 200 字
- 無法處理的問題(退款、帳號異常)請引導客戶填寫工單,連結:{ticket_url}
- 不要捏造資料,不確定一律說「我幫你確認後回覆」
- 不回答與產品無關的問題,禮貌拒絕即可

config.py — 組裝 System Prompt 並加 cache_control:

from pathlib import Path
from dotenv import load_dotenv
import os

load_dotenv()

SYSTEM_PROMPT_TEMPLATE = Path("prompts/system.txt").read_text(encoding="utf-8")
FAQ_CONTENT            = Path("prompts/faq.txt").read_text(encoding="utf-8")

def build_system_blocks(company: str, ticket_url: str) -> list[dict]:
    """
    把 system prompt 拆成兩個 block:
    - block[0]: 主體 prompt(長、幾乎不變) → cache_control ephemeral
    - block[1]: FAQ 內容(長、幾乎不變) → cache_control ephemeral

    兩個 block 各自獨立快取。快取命中後,input token 費用只剩
    原價的 10%(Sonnet 4.6 是 $0.30/MTok,而非 $3.00/MTok)。
    """
    system_text = SYSTEM_PROMPT_TEMPLATE.format(
        company=company, ticket_url=ticket_url
    )
    return [
        {
            "type": "text",
            "text": system_text,
            "cache_control": {"type": "ephemeral"},
        },
        {
            "type": "text",
            "text": f"## 常見問題知識庫\n\n{FAQ_CONTENT}",
            "cache_control": {"type": "ephemeral"},
        },
    ]

Prompt Caching 只要 system block 超過 1,024 tokens 就划算。第一次呼叫是 cache write(費用 1.25 倍),之後每次命中只付 cache read(費用 0.10 倍)。一個 3,000 token 的 System Prompt,每天 1,000 次呼叫,快取命中後每月節省約 $243(用 Sonnet 4.6 算)。

Step 3 — 定義結構化輸出 Schema

schemas.py:

from pydantic import BaseModel, Field
from enum import Enum

class IntentCategory(str, Enum):
    refund   = "refund"
    technical = "technical"
    billing  = "billing"
    general  = "general"

class ClassifyResult(BaseModel):
    intent:   IntentCategory
    urgency:  int  = Field(description="緊急程度 1–5,5 最緊急")
    summary:  str  = Field(description="一句話摘要,給客服人員看的內部備注")

class CustomerReply(BaseModel):
    classify:     ClassifyResult
    reply_text:   str  = Field(description="給客戶看的回覆正文,繁體中文,不超過 200 字")
    needs_human:  bool = Field(description="是否需要轉人工客服")

用 Pydantic .parse() 做結構化輸出呼叫(GA 版,不需要 beta header):

import os
from anthropic import Anthropic
from schemas import CustomerReply

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def classify_and_draft(user_msg: str, system_blocks: list) -> CustomerReply:
    """
    用結構化輸出拿到 JSON 分類結果。
    這個函式用同步 Anthropic client(不是 Async),因為它只在
    非 streaming 路徑呼叫,結果拿到後才決定要不要開 streaming。
    """
    resp = client.messages.parse(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system_blocks,
        messages=[{"role": "user", "content": user_msg}],
        output_format=CustomerReply,  # SDK 自動把 Pydantic 轉成 JSON Schema
    )

    # stop_reason 防禦:拒絕或截斷時都要攔截
    if resp.stop_reason == "refusal":
        raise ValueError("模型拒絕回應(可能含不當內容)")
    if resp.stop_reason == "max_tokens":
        raise ValueError("輸出被截斷,請增加 max_tokens")

    return resp.parsed_output   # 已驗證的 CustomerReply 物件

Step 4 — FastAPI Streaming Endpoint

main.py:

import os, json
from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from anthropic import AsyncAnthropic, APIStatusError
from dotenv import load_dotenv
from config import build_system_blocks
from schemas import classify_and_draft, CustomerReply

load_dotenv()

app = FastAPI()

# Streaming 一定要用 AsyncAnthropic,避免堵住 event loop
async_client = AsyncAnthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    timeout=30.0,   # 非 streaming 請求的最長等待秒數
)

SYSTEM_BLOCKS = build_system_blocks(
    company="Acme Inc.",
    ticket_url="https://support.acme.com/ticket",
)

class ChatRequest(BaseModel):
    message:    str
    session_id: str

def verify_token(x_api_token: str = Header(...)):
    """簡易 token 驗證:前端帶 X-Api-Token header。"""
    if x_api_token != os.environ["API_SECRET_TOKEN"]:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/chat")
async def chat(
    req: ChatRequest,
    _: None = Depends(verify_token),
):
    # 1. 先做結構化分類(同步,用一般 Anthropic client)
    try:
        classified: CustomerReply = classify_and_draft(req.message, SYSTEM_BLOCKS)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e))

    # 2. 需要轉人工時直接回 JSON
    if classified.needs_human:
        return {
            "type":       "redirect",
            "intent":     classified.classify.intent,
            "ticket_url": "https://support.acme.com/ticket",
        }

    # 3. 不需轉人工:用 Streaming 推回覆文字給前端
    async def event_stream():
        # 先推一筆 metadata(分類結果),前端可以同步顯示
        meta = {
            "type":    "meta",
            "intent":  classified.classify.intent,
            "urgency": classified.classify.urgency,
        }
        yield f"data: {json.dumps(meta, ensure_ascii=False)}\n\n"

        # 再 stream 正文
        import asyncio
        try:
            async with asyncio.timeout(90):   # stream 最多跑 90 秒
                # 注意:不要把 classified.reply_text 當成 assistant 訊息「prefill」
                # 塞進去——Sonnet 4.6 不支援最後一則 assistant 前綴,會回 400。
                # 分類草稿只在內部使用,streaming 這一趟直接從 user 訊息重新生成。
                async with async_client.messages.stream(
                    model="claude-sonnet-4-6",
                    max_tokens=512,
                    system=SYSTEM_BLOCKS,
                    messages=[
                        {"role": "user", "content": req.message},
                    ],
                ) as stream:
                    async for text in stream.text_stream:
                        chunk = {"type": "text", "delta": text}
                        yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
        except asyncio.TimeoutError:
            yield 'data: {"type":"error","msg":"stream timeout"}\n\n'

        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control":               "no-cache",
            "X-Accel-Buffering":           "no",   # 讓 Nginx 不 buffer SSE
        },
    )

一定要用 AsyncAnthropic 而不是同步的 Anthropic。同步客戶端會在 Streaming 期間堵住 uvicorn 的 event loop,造成其他請求全部排隊等待,高流量時直接服務崩潰。

Express(Node.js)版等效寫法:

import Anthropic from "@anthropic-ai/sdk";
import express from "express";

const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
const app = express();
app.use(express.json());

function verifyToken(req, res, next) {
  if (req.headers["x-api-token"] !== process.env.API_SECRET_TOKEN) {
    return res.status(401).json({ error: "Invalid token" });
  }
  next();
}

app.post("/chat", verifyToken, async (req, res) => {
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("X-Accel-Buffering", "no");

  try {
    const stream = await anthropic.messages.stream({
      model: "claude-sonnet-4-6",
      max_tokens: 512,
      messages: [{ role: "user", content: req.body.message }],
    });

    for await (const chunk of stream) {
      if (chunk.type === "content_block_delta" && chunk.delta.type === "text_delta") {
        const data = { type: "text", delta: chunk.delta.text };
        res.write(`data: ${JSON.stringify(data)}\n\n`);
      }
    }
    res.write("data: [DONE]\n\n");
  } catch (err) {
    res.write(`data: ${JSON.stringify({ type: "error", msg: err.message })}\n\n`);
  }
  res.end();
});

app.listen(3000);

Step 5 — 重試與超時處理

retry.py:

import asyncio
from anthropic import APIStatusError, APIConnectionError

# 值得重試的 HTTP 狀態碼(伺服器端暫時性錯誤)
RETRYABLE_STATUS = {500, 502, 503, 529}

async def with_retry(func, max_retries: int = 3, base_delay: float = 1.0):
    """
    指數退避重試。
    - 529 Overloaded / 5xx Server Error → 重試,等待時間 1s/2s/4s
    - 429 Rate Limit → 讀 Retry-After header 決定等待時間
    - 4xx(非 429)→ 立即拋出,不重試(這類錯誤重試沒有意義)
    - APIConnectionError(網路斷) → 重試
    """
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except APIStatusError as e:
            if e.status_code == 429:
                # 讀 Retry-After header,沒有就用 base_delay
                retry_after = float(
                    e.response.headers.get("retry-after", base_delay)
                )
                await asyncio.sleep(retry_after)
                continue
            if e.status_code in RETRYABLE_STATUS and attempt < max_retries:
                wait = base_delay * (2 ** attempt)
                await asyncio.sleep(wait)
                continue
            raise   # 4xx 或超過重試次數:往上拋
        except APIConnectionError:
            if attempt < max_retries:
                await asyncio.sleep(base_delay * (2 ** attempt))
                continue
            raise

# 用法範例:
# result = await with_retry(
#     lambda: async_client.messages.create(model=..., messages=...)
# )

超時設定的正確層次:

# 層次一:SDK 層設 HTTP request timeout(整個非 streaming 請求的上限)
client = AsyncAnthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    timeout=30.0,    # 非 streaming 請求 30 秒沒回應就拋 APITimeoutError
)

# 層次二:Streaming 層用 asyncio.timeout 包 stream 迴圈
# (SDK 的 request timeout 不管 streaming 期間的 idle,要自己設)
async def safe_stream(messages: list):
    async with asyncio.timeout(90):     # 整個 stream 最多 90 秒
        async with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=512,
            messages=messages,
        ) as stream:
            async for text in stream.text_stream:
                yield text

不同層次各司其職:SDK timeout 管「等第一個回應等多久」,asyncio.timeout 管「整段 stream 跑多久」。兩個都要設,缺一個都可能讓請求無限期掛著。

Step 6 — 金鑰輪替

import os, asyncio
from anthropic import AsyncAnthropic, APIStatusError

class KeyManager:
    """
    雙金鑰輪替:主金鑰失效時自動切到備用金鑰,
    同時觸發警報讓你去申請新金鑰替換 BACKUP。
    """
    def __init__(self):
        self._keys = [
            os.environ["ANTHROPIC_API_KEY"],
            os.environ.get("ANTHROPIC_API_KEY_BACKUP", ""),
        ]
        self._active = 0

    def current_key(self) -> str:
        return self._keys[self._active]

    def client(self) -> AsyncAnthropic:
        return AsyncAnthropic(api_key=self.current_key())

    def rotate(self):
        self._active = (self._active + 1) % len(self._keys)
        print(f"[ALERT] API key rotated → index {self._active}")
        # 接 Slack webhook / PagerDuty:
        # requests.post(SLACK_WEBHOOK, json={"text": "Claude API key rotated!"})

key_manager = KeyManager()

async def call_with_key_fallback(messages: list, **kwargs):
    """最多嘗試兩把金鑰,401 才輪替。"""
    for _ in range(2):
        try:
            return await key_manager.client().messages.create(
                messages=messages, **kwargs
            )
        except APIStatusError as e:
            if e.status_code == 401:
                key_manager.rotate()
                continue
            raise

緊急換鑰 SOP(零停機):

1. Anthropic Console → API Keys → 建立新金鑰,複製新 key
2. 更新部署環境(Railway/Fly.io/Cloud Run)的環境變數:
   ANTHROPIC_API_KEY        = <新金鑰>
   ANTHROPIC_API_KEY_BACKUP = <舊金鑰(暫時保留,還沒 revoke)>
3. 觀察 5 分鐘,確認新金鑰呼叫正常、日誌無錯誤
4. 回 Console revoke 舊金鑰
5. 不需要重啟 server:KeyManager.client() 每次呼叫都重新讀取環境變數

Step 7 — 上線前費用估算

先確認 2026 年 7 月的定價(每百萬 token,美元):

估算函式——上線前先跑這個,心裡有底:

def estimate_monthly_cost(
    daily_requests:    int,
    system_tokens:     int,    # system prompt + FAQ 的 token 數
    avg_user_tokens:   int,    # 平均每次使用者輸入的 token 數
    avg_output_tokens: int,    # 平均 Claude 回覆的 token 數
    cache_hit_rate:    float,  # 預估快取命中率,建議保守用 0.8
    model: str = "claude-sonnet-4-6",
) -> dict:
    pricing = {
        "claude-haiku-4-5":  {"in": 1.0,  "out": 5.0,  "cr": 0.10, "cw": 1.25},
        "claude-sonnet-4-6": {"in": 3.0,  "out": 15.0, "cr": 0.30, "cw": 3.75},
        "claude-opus-4-8":   {"in": 5.0,  "out": 25.0, "cr": 0.50, "cw": 6.25},
    }
    p = pricing[model]
    monthly = daily_requests * 30

    # system token 費用:每次都要 write 一次快取(1.25x),命中後讀取(0.10x)
    cache_write_cost = (system_tokens / 1e6) * p["cw"]
    cache_read_cost  = (system_tokens / 1e6) * p["cr"] * cache_hit_rate
    uncached_in_cost = (system_tokens / 1e6) * p["in"] * (1 - cache_hit_rate)
    user_in_cost     = (avg_user_tokens / 1e6) * p["in"]
    output_cost      = (avg_output_tokens / 1e6) * p["out"]

    per_request = (
        cache_write_cost + cache_read_cost +
        uncached_in_cost + user_in_cost + output_cost
    )
    monthly_total = per_request * monthly

    return {
        "per_request_usd": round(per_request, 6),
        "monthly_usd":     round(monthly_total, 2),
        "monthly_ntd":     round(monthly_total * 32, 0),  # 約 1 USD = 32 NTD
    }


# 情境一:每天 500 通客服對話,Sonnet 4.6
print(estimate_monthly_cost(
    daily_requests=500,
    system_tokens=3000,      # system.txt + faq.txt 合計約 3K tokens
    avg_user_tokens=80,
    avg_output_tokens=200,
    cache_hit_rate=0.85,
    model="claude-sonnet-4-6",
))
# → {"per_request_usd": 0.010827, "monthly_usd": 162.4, "monthly_ntd": 5197.0}

# 情境二:每天 1,000 通
print(estimate_monthly_cost(
    daily_requests=1000,
    system_tokens=3000,
    avg_user_tokens=80,
    avg_output_tokens=200,
    cache_hit_rate=0.85,
    model="claude-sonnet-4-6",
))
# → {"per_request_usd": 0.010827, "monthly_usd": 324.8, "monthly_ntd": 10394.0}

每通對話約 $0.011 USD = 0.35 元台幣,比人工客服的每小時人力成本便宜兩個數量級。這個數字拿去說服主管或客戶時很好用。

Step 8 — 部署選項

Railway 最快上線做法:

# 1. 在根目錄建 Procfile
echo "web: uvicorn main:app --host 0.0.0.0 --port \$PORT" > Procfile

# 2. 安裝 Railway CLI
npm install -g @railway/cli

# 3. 登入、建立專案、部署
railway login
railway init
railway up

# 4. 在 Railway 後台 → Variables 設定環境變數:
# ANTHROPIC_API_KEY / ANTHROPIC_API_KEY_BACKUP / API_SECRET_TOKEN

GCP Cloud Run(有 Docker 環境時):

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

gcloud run deploy cs-api \
  --source . \
  --region asia-east1 \
  --allow-unauthenticated \
  --min-instances 1 \
  --set-env-vars ANTHROPIC_API_KEY=sk-ant-...

注意 Cloud Run 一定要加 --min-instances 1,否則冷啟動會讓 SSE 連線在首次建立時出現 2–5 秒延遲,SSE 客戶端可能超時重連。