這是系列的最後一篇。
System Design 是 FDE 面試最能展現工程深度的地方。
答得好，你就是那個「懂技術也懂業務」的人。
System Design 面試的本質
面試官在問系統設計題的時候，不是要你給出標準答案，而是想看：
- 你有沒有釐清問題的習慣（不假設，先問）
- 你的 trade-off 思維（不說「最好的方案」，說「在這個場景下我選這個，因為…」）
- 你有沒有考慮到生產環境的現實（Auth、Scale、Cost、Failure）
第一題：設計企業知識庫 Chatbot
這是最高頻的考題之一。
題目：設計一個供企業內部使用的 AI 知識庫問答系統，員工可以用自然語言查詢公司政策、產品說明和技術文件。
第一步：釐清需求（你要主動問）
你應該問的問題：
- 同時使用的用戶數量級？（100人 vs 10萬人，差很多）
- 文件量多大？（1GB vs 1TB）
- 需要支援多語言嗎？
- 回答需要引用文件來源嗎？
- 有合規要求嗎？（GDPR、SOC2）
- Latency 要求？（秒級 vs 毫秒級）
- 不同部門能看的文件不同嗎？
最後一個問題很關鍵，決定了你需不需要 RBAC。
第二步：高層架構
用戶
↓ HTTPS
API Gateway（Auth Token 驗證 + Rate Limiting）
↓
Chatbot Service
├── RBAC 權限過濾（這個用戶能看哪些文件？）
├── Query Processor（問題前處理）
│ ├── 意圖分類
│ └── 查詢改寫（Query Rewriting）
├── RAG Engine
│ ├── Embedding Service
│ ├── Vector DB（根據 RBAC 過濾結果）
│ └── Reranker
├── LLM（Claude / Gemini）
└── Response Generator（加入引用來源）
↓
Cache Layer（相同問題不重複查）
↓
Logging & Monitoring
↓
回應給用戶
Authentication：怎麼做
企業場景通常用 SSO + JWT：
1from fastapi import FastAPI, Depends, HTTPException
2from fastapi.security import HTTPBearer
3import jwt
4
5app = FastAPI()
6security = HTTPBearer()
7
8def verify_token(credentials = Depends(security)) -> dict:
9 token = credentials.credentials
10 try:
11 payload = jwt.decode(
12 token,
13 settings.JWT_SECRET,
14 algorithms=["HS256"]
15 )
16 return {
17 "user_id": payload["sub"],
18 "department": payload["dept"],
19 "roles": payload["roles"]
20 }
21 except jwt.ExpiredSignatureError:
22 raise HTTPException(status_code=401, detail="Token 已過期")
23 except jwt.InvalidTokenError:
24 raise HTTPException(status_code=401, detail="無效的 Token")
25
26@app.post("/chat")
27async def chat(
28 request: ChatRequest,
29 user: dict = Depends(verify_token)
30):
31 # user 已驗證，包含部門和角色資訊
32 response = await chatbot.answer(
33 question=request.question,
34 user_context=user
35 )
36 return response
RBAC：按角色控制文件存取
1ROLE_PERMISSIONS = {
2 "employee": ["public", "general"],
3 "manager": ["public", "general", "internal"],
4 "hr": ["public", "general", "internal", "hr_confidential"],
5 "finance": ["public", "general", "internal", "finance_confidential"],
6 "admin": ["public", "general", "internal", "hr_confidential",
7 "finance_confidential", "executive"]
8}
9
10class RBACFilter:
11 def get_allowed_categories(self, user_roles: list[str]) -> list[str]:
12 allowed = set()
13 for role in user_roles:
14 allowed.update(ROLE_PERMISSIONS.get(role, []))
15 return list(allowed)
16
17 def filter_vector_search(
18 self,
19 query_embedding: list[float],
20 user_roles: list[str],
21 n_results: int = 5
22 ) -> list[dict]:
23 allowed_categories = self.get_allowed_categories(user_roles)
24
25 # 在向量搜尋時加入過濾條件
26 results = vector_db.query(
27 query_embeddings=[query_embedding],
28 n_results=n_results,
29 where={
30 "category": {"$in": allowed_categories}
31 }
32 )
33 return results
重點：RBAC 要在 Vector DB 查詢時就過濾，不能查出來後再過濾。因為那樣你可能查到 [機密文件、公開文件、機密文件, ...]，但因為前幾名都被過濾掉，最後給 LLM 的 context 品質會很差。
RAG 的 Query Rewriting
直接用用戶的原始問題去查向量資料庫，效果有時候不好。
例如用戶問：「我剛入職，想知道年假怎麼算」
這個問題包含太多無關資訊。應該先改寫：
1async def rewrite_query(original_query: str) -> str:
2 prompt = f"""
3 把以下問題改寫成適合搜尋的關鍵詞查詢。
4 移除個人背景資訊，保留核心問題。
5
6 原始問題：{original_query}
7 改寫後的搜尋查詢：
8 """
9
10 rewritten = await llm.generate(prompt, temperature=0.1)
11 return rewritten
12
13# 「我剛入職，想知道年假怎麼算」
14# → 「年假計算方式 員工請假規定」
Cache Layer：哪些東西要 Cache
1import hashlib
2import json
3from redis import Redis
4
5redis_client = Redis(host="localhost", port=6379)
6
7class ChatCache:
8 def __init__(self, ttl_seconds: int = 3600):
9 self.ttl = ttl_seconds
10
11 def _make_key(self, question: str, user_roles: list[str]) -> str:
12 # 相同問題但不同角色，答案可能不同（因為 RBAC）
13 cache_input = json.dumps({
14 "question": question.strip().lower(),
15 "roles": sorted(user_roles)
16 })
17 return f"chat:{hashlib.md5(cache_input.encode()).hexdigest()}"
18
19 def get(self, question: str, user_roles: list[str]) -> str | None:
20 key = self._make_key(question, user_roles)
21 cached = redis_client.get(key)
22 return cached.decode() if cached else None
23
24 def set(self, question: str, user_roles: list[str], answer: str):
25 key = self._make_key(question, user_roles)
26 redis_client.setex(key, self.ttl, answer)
Cache 的邊界：
- ✅ 適合 Cache：「年假幾天？」「請假流程？」這類政策問題，答案穩定
- ❌ 不適合 Cache：「我的訂單狀態？」這類個人化問題，每個人答案不同
- ❌ 不適合 Cache：需要即時資料的問題（股價、庫存）
Logging：要記錄什麼
1import structlog
2from datetime import datetime
3
4logger = structlog.get_logger()
5
6async def log_chat_event(
7 user_id: str,
8 question: str,
9 answer: str,
10 retrieved_docs: list[str],
11 latency_ms: float,
12 cache_hit: bool,
13 model_name: str,
14 input_tokens: int,
15 output_tokens: int
16):
17 logger.info(
18 "chat_event",
19 timestamp=datetime.utcnow().isoformat(),
20 user_id=user_id, # 誰問的（審計用）
21 question_hash=hash(question), # 不記錄原文，保護隱私
22 answer_length=len(answer),
23 retrieved_doc_ids=retrieved_docs, # 查了哪些文件（排查幻覺用）
24 latency_ms=latency_ms, # 性能監控
25 cache_hit=cache_hit, # Cache 命中率
26 model_name=model_name,
27 cost_usd=(input_tokens * 0.000003 + output_tokens * 0.000015)
28 )
第二題：設計 Internal AI Copilot
題目：設計一個 AI 助理，員工可以用自然語言查詢公司內部數據，例如「今年 Q3 的營收比 Q2 成長了多少？」
這題比知識庫 Chatbot 更難，因為要即時查詢結構化資料，而不是搜尋文件。
為什麼難
知識庫 Chatbot：問題 → 查文件 → 回答
Internal Copilot：問題 → 理解問題的數據意圖 → 查 BigQuery / CRM / ERP → 組合多來源數據 → 回答
高層架構
員工：「今年營收比去年成長多少？」
↓
Intent Classifier（這是數據查詢，不是文件查詢）
↓
NL2SQL Agent（把問題轉成 SQL）
↓
Tool Router
├── BigQuery Tool（財務、銷售數據）
├── CRM Tool（客戶、訂單數據）
├── ERP Tool（庫存、供應鏈）
└── HR System Tool（員工數據）
↓
Data Aggregator（整合多來源結果）
↓
LLM（把數字轉成自然語言回答）
↓
Response（含數據來源標示）
NL2SQL：把問題轉成 SQL
1async def natural_language_to_sql(
2 question: str,
3 schema_context: str,
4 user_department: str
5) -> str:
6 prompt = f"""
7 你是一位資深的 SQL 分析師。
8 根據以下資料庫 Schema，將問題轉換為 SQL 查詢。
9
10 只能查詢用戶有權限的資料表（部門：{user_department}）。
11
12 Schema：
13 {schema_context}
14
15 注意事項：
16 - 只生成 SELECT 語句，絕對不能生成 INSERT / UPDATE / DELETE
17 - 如果問題超出你的 Schema 範圍，請說「無法查詢此資料」
18
19 問題：{question}
20
21 SQL：
22 """
23
24 sql = await llm.generate(prompt, temperature=0.0)
25
26 # 安全驗證：確保只有 SELECT
27 if not is_safe_sql(sql):
28 raise SecurityError("生成了不安全的 SQL")
29
30 return sql
31
32def is_safe_sql(sql: str) -> bool:
33 """確保 SQL 只包含讀取操作"""
34 dangerous_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "TRUNCATE", "ALTER"]
35 sql_upper = sql.upper()
36 return not any(keyword in sql_upper for keyword in dangerous_keywords)
BigQuery Tool 實作
1from google.cloud import bigquery
2from typing import Any
3
4class BigQueryTool:
5 def __init__(self, project_id: str):
6 self.client = bigquery.Client(project=project_id)
7
8 async def execute_query(
9 self,
10 sql: str,
11 max_rows: int = 1000,
12 timeout_seconds: int = 30
13 ) -> dict[str, Any]:
14
15 job_config = bigquery.QueryJobConfig(
16 maximum_bytes_billed=10 * 1024 * 1024 * 1024, # 10GB 上限
17 use_query_cache=True,
18 )
19
20 try:
21 query_job = self.client.query(
22 sql,
23 job_config=job_config
24 )
25
26 results = query_job.result(
27 max_results=max_rows,
28 timeout=timeout_seconds
29 )
30
31 rows = [dict(row) for row in results]
32
33 return {
34 "success": True,
35 "rows": rows,
36 "total_rows": results.total_rows,
37 "bytes_processed": query_job.total_bytes_processed,
38 "cache_hit": query_job.cache_hit
39 }
40
41 except Exception as e:
42 return {
43 "success": False,
44 "error": str(e),
45 "rows": []
46 }
整合多來源資料
員工可能問：「哪個銷售員今年業績最好，他的客戶滿意度怎麼樣？」
這需要同時查 BigQuery（銷售數據）和 CRM（滿意度評分）：
1import asyncio
2
3class DataAggregator:
4 async def query_multiple_sources(
5 self,
6 queries: list[dict],
7 user_context: dict
8 ) -> dict:
9
10 tasks = []
11 for query in queries:
12 source = query["source"]
13 sql = query["sql"]
14
15 if source == "bigquery":
16 task = self.bq_tool.execute_query(sql)
17 elif source == "crm":
18 task = self.crm_tool.execute_query(sql)
19 elif source == "erp":
20 task = self.erp_tool.execute_query(sql)
21
22 tasks.append(task)
23
24 # 並行查詢所有來源
25 results = await asyncio.gather(*tasks, return_exceptions=True)
26
27 return {
28 source["source"]: result
29 for source, result in zip(queries, results)
30 }
把數字轉成自然語言
1async def generate_data_response(
2 question: str,
3 data: dict,
4 source_labels: list[str]
5) -> str:
6
7 data_summary = format_data_for_llm(data)
8 sources_text = "、".join(source_labels)
9
10 prompt = f"""
11 根據以下查詢結果，用清楚易懂的中文回答問題。
12
13 數據來源：{sources_text}
14
15 查詢結果：
16 {data_summary}
17
18 問題：{question}
19
20 要求：
21 - 直接回答問題，不要重複問題本身
22 - 如果涉及百分比，請計算並說明
23 - 在回答末尾標注「數據來源：{sources_text}」
24 """
25
26 return await llm.generate(prompt, temperature=0.1)
成本控制：這題必須說
BigQuery 按資料掃描量計費（$5 / TB）。如果 AI 生成的 SQL 每次掃整張 table，成本會很高。
1class CostAwareBigQueryTool:
2 MAX_BYTES_PER_QUERY = 10 * 1024 ** 3 # 10 GB
3 BYTES_PER_TB = 1024 ** 4
4 PRICE_PER_TB = 5.0 # USD
5
6 async def execute_with_cost_estimate(self, sql: str) -> dict:
7 # 先用 Dry Run 估算成本
8 dry_run_config = bigquery.QueryJobConfig(dry_run=True)
9 dry_run_job = self.client.query(sql, job_config=dry_run_config)
10
11 estimated_bytes = dry_run_job.total_bytes_processed
12 estimated_cost = (estimated_bytes / self.BYTES_PER_TB) * self.PRICE_PER_TB
13
14 if estimated_bytes > self.MAX_BYTES_PER_QUERY:
15 return {
16 "success": False,
17 "error": f"查詢預估掃描 {estimated_bytes/1e9:.1f} GB，超出限制。請縮小查詢範圍。"
18 }
19
20 # 成本可接受，執行查詢
21 return await self.execute_query(sql)
兩題的對比分析
| 知識庫 Chatbot | Internal Copilot | |
|---|---|---|
| 資料來源 | 文件（非結構化） | 資料庫（結構化） |
| 核心技術 | RAG + Vector DB | NL2SQL + 多來源查詢 |
| 幻覺風險 | 中（有文件 context） | 低（直接查資料）但有 SQL 錯誤風險 |
| RBAC 難度 | 中（文件分類） | 高（資料行 / 欄 level 的控制） |
| Latency | 較低（向量搜尋快） | 較高（SQL 查詢可能慢） |
| Cache 效益 | 高（政策類問題重複率高） | 中（數據每天更新，TTL 要短） |
面試中容易被追問的細節
Q：Cache 過期了怎麼辦？
知識庫 Chatbot 的 Cache TTL 設多長？文件更新時要主動 invalidate cache。
1async def update_document(doc_id: str, new_content: str):
2 # 1. 更新向量資料庫
3 await vector_db.update(doc_id, new_content)
4
5 # 2. 清除相關 Cache（使用 tag-based invalidation）
6 await cache.delete_by_tag(f"doc:{doc_id}")
Q：LLM 生成了錯誤的 SQL 怎麼辦？
1async def safe_nl2sql(question: str, max_retries: int = 3) -> str:
2 for attempt in range(max_retries):
3 sql = await nl_to_sql(question)
4
5 # 語法驗證
6 is_valid, error = validate_sql(sql)
7 if not is_valid:
8 # 把錯誤告訴 LLM，讓它自我修正
9 question = f"""
10 之前生成的 SQL 有錯誤：{error}
11 請修正後重新生成。
12 原始問題：{question}
13 """
14 continue
15
16 return sql
17
18 return None # 三次都失敗，回報無法處理
Q：用戶問了 Schema 裡沒有的資料怎麼辦？
系統要能識別這種情況並優雅地告知，而不是生成一個查不到資料的 SQL 或直接幻覺。
完整的面試回答結構
系統設計題，我建議這樣走：
第一分鐘：
"在開始設計之前，我想先確認幾個需求..."
（問 1-2 個關鍵問題：用戶規模、RBAC 需求、Latency 要求）
第二到五分鐘：
"好，根據這些需求，我的高層架構是這樣..."
（畫圖，解釋每個元件的職責）
第六到十五分鐘：
"我想深入說明幾個關鍵設計決策..."
（選 2-3 個最重要的部分：RBAC 實作、Cache 策略、Failure 處理）
最後幾分鐘：
"這個設計的主要 trade-off 是..."
"最可能出問題的地方是..."
系列總結
四篇下來，我想說的核心只有一件事：
FDE 是橋接器，不是工具使用者。
能說出「我會用 LangGraph 做這個」，只是入門。
能說出「在這個場景下，Multi-Agent 帶來的複雜度不值得，我會選擇 Single Agent + 多個 Tools，因為…」，才是 FDE 的水準。
面試考的是判斷力，不是背誦力。
系列文章索引
- 第一篇：RAG 完全解析 — Embedding、Chunk 策略、幻覺改善
- 第二篇：Agent System Design — ReAct、Multi-Agent、MCP、失控防範
- 第三篇：ML 基礎知識 — Transformer、Embedding、Fine-tuning、評估指標
- 第四篇：System Design 實戰 — 知識庫 Chatbot、Internal Copilot
