_{🧠 图解记忆：先按意图与风险分流，再用证据和工具完成任务，并以置信度门控人工接管；点击图片可查看原图。}

💡 答案要点

题目理解：

百万 DAU 只描述每天有多少用户，不能直接推出峰值 QPS。面试时应先向面试官确认或声明假设：
- 日活用户：100 万；
- 假设每人每天 5 次请求，则日请求量约 500 万；
- 平均到达率约 `5,000,000 / 86,400 ≈ 58 QPS`；
- 若峰值系数取 5～10，入口峰值约 300～600 QPS；
- 若平均流式会话占用 8 秒，峰值并发连接约为 `QPS × 8`。

这些只是容量估算示例，最终要用真实时段分布、会话长度和重试率校准。

整体架构：

用户请求
    ↓
┌─────────────────────────────────────────────┐
│              CDN / 边缘节点                  │
│         （静态资源 + 就近接入）               │
└─────────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────────┐
│           负载均衡（Nginx/LB）               │
│        健康检查 + 熔断 + SSL 终结            │
└─────────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────────┐
│           API 网关层                         │
│    认证鉴权 │ 限流 │ 路由 │ 日志             │
└─────────────────────────────────────────────┘
    ↓
┌─────────────────────────────────────────────┐
│           应用服务层（无状态）                │
│   ┌─────────┐  ┌─────────┐  ┌─────────┐   │
│   │ 服务实例1 │  │ 服务实例2 │  │ 服务实例N │   │
│   └─────────┘  └─────────┘  └─────────┘   │
└─────────────────────────────────────────────┘
    ↓
┌──────────┐  ┌──────────┐  ┌──────────┐
│ LLM 网关  │  │ RAG 服务  │  │ 会话存储  │
│(多模型路由)│  │(知识检索) │  │ (Redis)  │
└──────────┘  └──────────┘  └──────────┘
    ↓
┌──────────┐  ┌──────────┐
│ LLM API  │  │ 知识库   │
│(OpenAI等)│  │(向量数据库)│
└──────────┘  └──────────┘

核心组件设计：

1. API 网关（限流 + 鉴权）：

# 这是“固定窗口计数器”的简化示例，不是令牌桶。
# 生产系统还要处理 Redis 原子性、故障降级和多维配额。
@app.middleware
async def rate_limit_middleware(request: Request, call_next):
    user_id = get_user_id(request)
    
    # 固定窗口：每个用户每秒最多 10 个请求
    key = f"rate_limit:{user_id}"
    allow = await redis.incr(key)
    if allow == 1:
        await redis.expire(key, 1)
    
    if allow > 10:  # 超过每秒 10 请求
        return JSONResponse(
            status_code=429,
            content={"error": "rate limit exceeded"}
        )
    
    return await call_next(request)

2. LLM 网关（多模型路由）：

class ModelRouter:
    """智能路由：根据请求类型选择最优模型"""
    
    ROUTING_RULES = {
        # 简单问答 → 便宜模型
        "simple_qa": {"model": "gpt-4o-mini", "cost": 0.001},
        # 复杂推理 → 贵但准
        "complex_reasoning": {"model": "gpt-4o", "cost": 0.01},
        # 代码生成 → 代码专用模型
        "code_gen": {"model": "claude-3.5-sonnet", "cost": 0.008},
        # 国内用户 → 国产模型
        "china": {"model": "qwen-plus", "cost": 0.004},
    }
    
    def route(self, request: ChatRequest) -> str:
        # 根据特征路由
        if request.is_code_related:
            return self.ROUTING_RULES["code_gen"]["model"]
        if request.user_region == "china":
            return self.ROUTING_RULES["china"]["model"]
        if request.complexity == "high":
            return self.ROUTING_RULES["complex_reasoning"]["model"]
        return self.ROUTING_RULES["simple_qa"]["model"]

3. RAG 增强（知识库检索）：

用户问题
    ↓
Embedding 模型 → 向量检索（Pinecone/Milvus）
    ↓
Top-K 相关文档 chunks
    ↓
注入 Prompt：[Context] + [用户问题]
    ↓
LLM 生成答案

4. 会话管理（对话上下文）：

# 对话历史存储：Redis + 定期持久化
class SessionManager:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.MAX_TURNS = 20  # 限制上下文长度
    
    async def get_context(self, session_id: str) -> list[Message]:
        key = f"session:{session_id}"
        history = await self.redis.lrange(key, 0, -1)
        
        # 如果太长，做摘要压缩
        if len(history) > self.MAX_TURNS:
            older = await self.summarize(history[:-self.MAX_TURNS])
            return older + history[-self.MAX_TURNS:]
        
        return [json.loads(m) for m in history]
    
    async def add_message(self, session_id: str, role: str, content: str):
        key = f"session:{session_id}"
        await self.redis.rpush(key, json.dumps({"role": role, "content": content}))
        await self.redis.expire(key, 86400 * 30)  # 30天过期

成本估算方法：

不要记固定总价。先拆出可测变量：

月请求数 = DAU × 人均请求数 × 30
模型成本 = Σ(输入 token × 对应输入单价 + 输出 token × 对应输出单价)
检索成本 = 查询次数 × 单次检索成本 + 索引存储/构建成本
平台成本 = 网关/应用/队列/缓存/可观测性/带宽/人工审核

还要区分缓存读取、缓存写入、重试、工具调用和不同模型路由。模型单价属于时效性数据，应在面试前查询供应商价格页，并说明币种和汇率。

高可用设计：

多地域部署：
北京 Active ──── 上海 Active
      \           /
       全局流量调度

如果设计是 Active / Standby，应称为主备灾备，不是多活。多活还需要说明会话、配额、配置和反馈数据如何保持一致，以及区域故障时怎样避免双写冲突。

• 熔断：下游 LLM 响应慢 → 自动切换备选模型
• 降级：事实型客服不能简单关闭 RAG 后让通用模型自由回答。可降级到已审核 FAQ、缓存答案、关键词检索、排队或人工客服，并明确告知能力受限。
• 重试：返回 5xx → 指数退避重试，最多重试 3 次

面试话术：

“我不会从百万 DAU 直接猜 QPS，而是先给出人均请求、峰值系数和会话时长，再分别估算入口 QPS、流式连接数和下游 token 吞吐。架构上把网关、会话、RAG、模型路由和异步工具解耦；降级优先返回已审核内容或转人工，不能牺牲事实安全。最后用容量测试、故障演练和质量回归验证设计。”

📚 参考：OpenAI：Latency Optimization（大规模对话系统权衡）

---

版本: v1.1 | 更新: 2026-05-09 | by 二狗子 🐕

---