← 返回博客列表
OpenAI

OpenAI VO 面試真題 #2:信用系統設計 - 亂序請求處理專家級攻略

2025-12-25

OAVOService 獨家解析:為什麼這題掛掉90%的候選人?

OpenAI 這道信用系統設計題看似簡單,實際上是分散式系統、時序處理和狀態管理的綜合考察。許多候選人在「亂序請求」這個關鍵點上栽跟頭,導致整個面試失利。

OAVOService 專業提醒:這類系統設計題不僅考察程式設計能力,更看重對現實業務場景的理解。我們已幫助300+學員成功攻克 OpenAI VO 面試,其中這道題的通過率達到95%。

題目完整描述

你需要為 OpenAI API 信用購買系統構建一個餘額追蹤系統。

核心功能需求

  1. 信用授予(Grant):使用者可以購買信用額度,每筆額度有ID、啟用時間和過期時間
  2. 信用扣減(Subtract):API呼叫時扣減信用,優先扣減即將過期的額度
  3. 餘額查詢(Balance):獲取指定時間點的可用餘額

關鍵挑戰:亂序請求處理

核心難點:由於網路不穩定,請求可能亂序到達。例如:

介面定義

class Credits:
    def create_grant(
        self, 
        timestamp: int, 
        grant_id: str, 
        amount: int, 
        expiration_timestamp: int
    ) -> None:
        """建立信用授予"""
        pass
    
    def subtract(self, timestamp: int, amount: int) -> None:
        """扣減信用(優先扣減即將過期的)"""
        pass
    
    def get_balance(self, timestamp: int) -> int | None:
        """獲取指定時間點餘額"""
        pass

OAVOService 專家級解決方案

核心設計思路

  1. 事件溯源(Event Sourcing):記錄所有操作,按時間戳重排執行
  2. 延遲計算:查詢時才重新計算狀態,確保時序正確
  3. 優先級佇列:自動處理過期時間優先級

完整實現方案

from heapq import heappush, heappop
from collections import defaultdict
import bisect

class Credits:
    def __init__(self):
        # 儲存所有事件,按時間戳排序
        self.events = []  # [(timestamp, event_type, data), ...]
        
        # 快取已計算的狀態
        self.state_cache = {}  # {timestamp: (grants, balance)}
        self.sorted_timestamps = []
    
    def create_grant(self, timestamp, grant_id, amount, expiration_timestamp):
        """建立信用授予事件"""
        event = (timestamp, 'grant', {
            'grant_id': grant_id,
            'amount': amount,
            'expiration_timestamp': expiration_timestamp
        })
        
        # 插入事件並保持時間序
        bisect.insort(self.events, event)
        
        # 清除受影響的快取
        self._invalidate_cache_from(timestamp)
    
    def subtract(self, timestamp, amount):
        """建立扣減事件"""
        event = (timestamp, 'subtract', {'amount': amount})
        bisect.insort(self.events, event)
        self._invalidate_cache_from(timestamp)
    
    def get_balance(self, timestamp):
        """獲取指定時間點的餘額"""
        if timestamp in self.state_cache:
            return self.state_cache[timestamp][1]
        
        # 重新計算到目標時間點的狀態
        balance = self._compute_balance_at(timestamp)
        return balance
    
    def _compute_balance_at(self, target_timestamp):
        """計算指定時間點的餘額"""
        # 獲取所有在目標時間之前或等於的事件
        relevant_events = [
            event for event in self.events 
            if event[0] <= target_timestamp
        ]
        
        # 維護活躍的授予(按過期時間排序的最小堆積)
        active_grants = []  # [(expiration_time, grant_id, remaining_amount)]
        
        for timestamp, event_type, data in relevant_events:
            if event_type == 'grant':
                # 只有在授予時間 <= target_timestamp 的才啟用
                if timestamp <= target_timestamp:
                    heappush(active_grants, (
                        data['expiration_timestamp'],
                        data['grant_id'], 
                        data['amount']
                    ))
            
            elif event_type == 'subtract':
                # 執行扣減操作
                remaining_to_subtract = data['amount']
                temp_grants = []
                
                # 從最先過期的開始扣減
                while active_grants and remaining_to_subtract > 0:
                    exp_time, grant_id, remaining = heappop(active_grants)
                    
                    # 檢查是否已過期
                    if exp_time <= target_timestamp:
                        continue  # 跳過已過期的
                    
                    if remaining <= remaining_to_subtract:
                        # 這個grant被完全消耗
                        remaining_to_subtract -= remaining
                    else:
                        # 部分消耗
                        temp_grants.append((exp_time, grant_id, remaining - remaining_to_subtract))
                        remaining_to_subtract = 0
                
                # 將未完全消耗的grants放回
                for grant in temp_grants:
                    heappush(active_grants, grant)
                
                if remaining_to_subtract > 0:
                    # 餘額不足,但不拋異常(根據題目要求)
                    return 0
        
        # 計算最終餘額(過濾掉已過期的)
        total_balance = 0
        for exp_time, grant_id, remaining in active_grants:
            if exp_time > target_timestamp:  # 未過期
                total_balance += remaining
        
        # 快取結果
        self.state_cache[target_timestamp] = (active_grants.copy(), total_balance)
        bisect.insort(self.sorted_timestamps, target_timestamp)
        
        return total_balance
    
    def _invalidate_cache_from(self, timestamp):
        """清除從指定時間點開始的所有快取"""
        # 找到需要清除的快取
        idx = bisect.bisect_left(self.sorted_timestamps, timestamp)
        
        # 清除快取
        for ts in self.sorted_timestamps[idx:]:
            self.state_cache.pop(ts, None)
        
        self.sorted_timestamps = self.sorted_timestamps[:idx]

優化版本:增量更新

class OptimizedCredits:
    def __init__(self):
        self.grants = {}  # grant_id -> Grant物件
        self.operations = []  # 按時間排序的操作記錄
        self.snapshots = {}  # 時間點快照快取
    
    def create_grant(self, timestamp, grant_id, amount, expiration_timestamp):
        self.grants[grant_id] = {
            'amount': amount,
            'expiration_timestamp': expiration_timestamp,
            'created_at': timestamp
        }
        
        operation = (timestamp, 'grant', grant_id, amount, expiration_timestamp)
        bisect.insort(self.operations, operation)
        
        # 清理受影響的快照
        self._cleanup_snapshots_from(timestamp)
    
    def subtract(self, timestamp, amount):
        operation = (timestamp, 'subtract', amount)
        bisect.insort(self.operations, operation)
        self._cleanup_snapshots_from(timestamp)
    
    def get_balance(self, timestamp):
        # 檢查快取
        if timestamp in self.snapshots:
            return self.snapshots[timestamp]
        
        # 找到最近的快照作為起點
        start_snapshot = self._find_nearest_snapshot(timestamp)
        
        # 從快照開始計算
        balance = self._compute_incremental_balance(start_snapshot, timestamp)
        
        # 儲存快照
        self.snapshots[timestamp] = balance
        
        return balance
    
    def _find_nearest_snapshot(self, timestamp):
        """找到時間戳之前最近的快照"""
        best_ts = 0
        for ts in self.snapshots:
            if ts <= timestamp and ts > best_ts:
                best_ts = ts
        return best_ts
    
    def _compute_incremental_balance(self, from_timestamp, to_timestamp):
        """從某個時間點增量計算到目標時間點"""
        if from_timestamp in self.snapshots:
            # 從已有快照開始
            current_grants = self.snapshots[from_timestamp]['grants'].copy()
        else:
            # 從零開始
            current_grants = []
        
        # 處理時間區間內的所有操作
        relevant_ops = [
            op for op in self.operations 
            if from_timestamp < op[0] <= to_timestamp
        ]
        
        for operation in relevant_ops:
            timestamp, op_type = operation[0], operation[1]
            
            if op_type == 'grant':
                _, _, grant_id, amount, exp_time = operation
                current_grants.append({
                    'grant_id': grant_id,
                    'amount': amount,
                    'expiration_timestamp': exp_time,
                    'remaining': amount
                })
            
            elif op_type == 'subtract':
                _, _, amount = operation
                self._process_subtraction(current_grants, amount, timestamp)
        
        # 計算餘額(排除過期的)
        total = sum(
            grant['remaining'] for grant in current_grants
            if grant['expiration_timestamp'] > to_timestamp
        )
        
        return total
    
    def _process_subtraction(self, grants, amount, current_time):
        """處理扣減操作"""
        # 按過期時間排序(最先過期的優先)
        grants.sort(key=lambda x: x['expiration_timestamp'])
        
        remaining = amount
        for grant in grants:
            if grant['expiration_timestamp'] <= current_time:
                continue  # 跳過已過期
            
            if grant['remaining'] <= 0:
                continue  # 跳過已用完
            
            if remaining <= 0:
                break  # 扣減完成
            
            deduct = min(grant['remaining'], remaining)
            grant['remaining'] -= deduct
            remaining -= deduct

面試官常見追問與 OAVOService 標準應答

Q1: 如何處理並行存取?

專家回答

Q2: 如何優化大量歷史資料查詢?

技術方案

Q3: 系統容災如何設計?

架構考慮

OAVOService 獨家面試技巧

回答框架 STAR-T

  1. Situation:明確業務場景和技術挑戰
  2. Task:拆解具體的技術任務
  3. Action:展示解決方案的設計思路
  4. Result:說明方案的優勢和權衡
  5. Technical Deep Dive:主動深入技術細節

程式碼實現要點

  1. 邊界處理:空餘額、負數、重複ID等
  2. 效能考慮:時間複雜度分析和優化思路
  3. 擴展性:如何支援更多功能(批次操作、交易等)

相關延伸題目

總結與面試成功策略

信用系統設計考察的是候選人對複雜業務場景的抽象能力和系統性思維。成功的關鍵在於:

  1. 正確理解亂序請求的本質問題
  2. 合理選擇資料結構和演算法
  3. 主動考慮效能、並行、容災等工程問題
  4. 清晰表達設計思路和技術權衡

OAVOService 專業面試輔助優勢

想要在 OpenAI VO 面試中一次通過?立即聯繫 OAVOService 專業團隊:

聯繫方式

服務承諾: ✓ 程式碼獨家定製,絕不重複使用 ✓ 嚴格保密協定,資訊安全無憂
✓ 品質層層把關,滿意度保證


SEO標籤:OpenAI面試、信用系統設計、亂序請求處理、VO面試輔助、面試作弊、分散式系統、時序資料、SDE面試、面試代做、一畝三分地、系統設計面試、OAVOService