[面经] OpenAI Coding：内存滑动窗口会话分析（3 Parts）

[面经] OpenAI Coding：内存滑动窗口会话分析（3 Parts）

前几天参加了 OpenAI 的技术面试，coding 轮遇到这道多 part 的题目，分享一下题面和解法。

## Part 1：最近交互计数

实现一个内存、单线程的聊天会话分析服务。输入是实时、全局 ts 单调不减的交互日志流。每条日志：`user: str`、`chat: int`（每个用户内唯一）、`ts: int`（分钟）。

经常要查某个 (user, chat) 在过去 15 分钟内的交互数，定义为闭区间 [T-14, T]，其中 T 取流中出现过的最大时间戳。

实现：
- `ingest(user, chat, ts)`
- `count_recent(user, chat) -> int`

要求：ts 单调不减、无重复；日志量和查询量都大，两个操作都要高效；内存有界——正比于过去 15 分钟内活跃的 chat 数，而不是历史总量。

### 思路

容易掉的坑是全存下来、查询时过滤——内存直接爆炸。利用“单调不减”这个条件：

- 每个 (user, chat) 维护一个 ts deque（到达顺序即时间顺序）。
- 再维护一个全局 deque，按到达顺序存 (ts, key)。每次 `ingest` 时从头部踢掉 `ts < T-14` 的，同时把对应 key 的 deque 头部也踢掉；deque 空了就删 key。
- 查询直接返回 `len(deque)`，O(1)，读时零工作。

所有存着的事件都保证落在 [T-14, T] 内，内存天然有界。均摊每次操作 O(1)。

关键细节：窗口是**闭区间**。比如 T=20 时窗口是 [6, 20]，t=6 的事件还算数。

## Part 2：活跃会话数

事件多了一个字段 `kind`：`touch`（用户发消息并收到回复）或 `close`（用户显式结束会话；可能在任意久之后到达，也可能永不到达）。

还要实时维护每个用户的**当前活跃会话数**。会话活跃当且仅当：过去 15 分钟内有 `touch` 事件，且在**最新一次 touch 之后**没有收到 `close` 事件。

- `ingest(user, chat, ts, kind="touch")`
- `count_live(user) -> int`

### 思路

每个 (user, chat) 存 (ts, kind) 的 deque。会话状态只取决于两件事：最新 touch 的时间，以及有没有 ts **严格大于**它的 close。

- 收到 `touch`：它成为最新，清除 closed 标记（新的 touch 会重新激活会话，覆盖之前的 close）。
- 收到 `close`：只有 `close_ts > latest_touch_ts` 才算结束。同一 ts 的 close 不算（“之后”是严格大于）。
- 活跃 = 最新 touch 落在 [T-14, T] 内且未被 close。

过期用 lazy 方式处理：`count_live` 时 sweep 该用户的 chat，窗口内已无事件的直接删（内存保持有界），数活跃的个数。写路径不需要全局扫描。

举例：t=2 touch → 1；t=8 close → 0。

## Part 3：乱序到达

第三个 part 放宽一个假设：事件现在可能乱序到达（ts 可能比已处理的更小）。查询看到的 T 依然单调不减，清理可以放到查询时做。

### 思路

核心洞察：**T 永远不回退**，所以查询时踢掉的东西以后永远进不了窗口——lazy cleanup 是安全的。

- `ingest` 只记录：每个 key 一个有序的 touch 时间列表（bisect 插入），每个 key 一个最大 close 时间戳。到达时已经 `< T-14` 的直接丢弃（它永远进不了未来的窗口）。
- 查询时：二分踢掉过期的 touch，删掉已死的 chat，然后计数。最新 touch 就是列表尾（保持有序）；closed = max_close > latest。

一个迟到的 `close`（ts 大于最新 touch）会正确地把会话置为不活跃；一个迟到的 `touch`（ts 大于已记录的 close）会重新激活它。不需要像乱序 ledger 那样重放——比较两个数（最大 touch ts vs 最大 close ts）就够了。

Part 3 用随机生成的乱序流和暴力实现对拍验证（计数 + 每用户活跃数），全部一致。

## 易错点

1. 闭区间 [T-14, T]，不是 [T-15, T]。差一分钟样例就过不了。
2. “最新 touch 之后的 close”是**严格大于**：同一 ts 的 close 不结束会话。
3. 新的 touch 会清除之前的 close——会话可以重新激活。
4. 内存界：Part 1 用全局到达序队列驱动过期；Part 3 靠单调的 T 在查询时清理。说清楚用哪种、为什么。
5. T 是全局的（全流最大值），不是 per user——别的用户的事件会把你的窗口往前推。

## 代码

```python
from bisect import bisect_left, insort
from co

您好！
本帖隐藏的内容需要积分高于 188 才可浏览
您当前积分为 0。
使用VIP即刻解锁阅读权限或查看其他获取积分的方式

游客，您好！
本帖隐藏的内容需要积分高于 188 才可浏览
您当前积分为 0。
VIP即刻解锁阅读权限 或 查看其他获取积分的方式
Unlock interview details and practice with AI
Curated Interview Questions from Top Companies

else:
            self._last_close[key] = max(ts, self._last_close.get(key, float("-inf")))
        self._chats_of[user].add(key)

    def _prune(self, key) -> bool:
        """Evict expired touches; drop the chat if nothing in-window remains."""
        lst = self._touches.get(key)
        if lst:
            del lst[:bisect_left(lst, self._start())]
            if not lst:
                del self._touches[key]
        if key not in self._touches:
            self._last_close.pop(key, None)
            return False
        return True

    def count_recent(self, user: str, chat: int) -> int:
        key = (user, chat)
        if not self._prune(key):
            self._chats_of[user].discard(key)
            return 0
        return len(self._touches[key])

    def count_live(self, user: str) -> int:
        n = 0
        for key in list(self._chats_of[user]):
            if not self._prune(key):
                self._chats_of[user].discard(key)
                continue
            latest = self._touches[key][-1]  # list kept sorted
            if not self._last_close.get(key, float("-inf")) > latest:
                n += 1
        return n
```

有问题欢迎讨论。