最新OpenAI 面试真题：Chat Bot System Refactoring（legacy chat service + slash-command bots 重构题）
这是道聊天机器人重构题，真正的难点是：如何在不意外改变旧系统行为的前提下，重新划分职责、管理状态，并引入事件驱动架构。
1. 题目：重构一个 legacy 聊天服务
假设有一个聊天服务，支持 AwayBot、MeetBot 和 TacoBot，所有功能都由斜杠命令触发。
老实现是一个庞大的 send_message 函数，依赖全局变量保存聊天记录、用户离开状态和 taco 总数。
三个 bot 的规则如下：
- AwayBot：
/away 原因，设置用户的离开状态；普通消息提到离开的人时，自动回复其状态。 - MeetBot：
/meet @某人，宣布双方开会，并将双方状态设为开会中。 - TacoBot：
/givetaco 🌮🌮 @某人，统计消息中的 taco emoji 数量，累加到接收者的总数。
任务通常分成几步：理解旧代码、重构为面向对象设计、改成带中央事件总线的事件驱动架构、编写测试，并讨论扩展方案。
这道题的核心不是把函数拆成三个类，而是回答三个问题：谁负责什么、状态归谁所有、不同组件如何协作。
2. 解法：五步，顺序有讲究
2.1. 第一步：冻结旧行为，先写黄金测试
不要一上来就改代码。
先用固定的输入序列运行旧实现，记录完整输出日志，并验证状态变化。
例如：
Alice: /away lunch
Bob: Hi @Alice
Alice is away: lunch
Bob: /meet @Alice
Bob and Alice are meeting
Carol: Hi @Alice and @Bob
上面仅是输入输出结构示意。
真正的黄金测试必须来自旧实现的实际运行结果，包括回复顺序、措辞和后续消息受到的状态影响，不能凭印象手写预期结果。
随后，每完成一轮重构，就运行同一批测试。
这里要区分两件事：黄金测试保护既有行为，但不代表既有行为一定正确。
如果原实现有 bug，黄金测试可能恰恰会把这个 bug 记录下来。
是否修复它，需要另行决策。
2.2. 第二步：定义接口，分离判断与执行
每个 bot 至少实现两个方法：
class BotService(ABC):
@abstractmethod
def should_activate(self, sender, text) -> bool:
"""判断是否需要处理这条消息，不产生副作用。"""
@abstractmethod
def execute(self, sender, text) -> list[str]:
"""执行处理逻辑，返回需要输出的消息。"""
should_activate() 只负责判断，不应修改状态。
execute() 才执行具体操作，包括更新状态、发布事件或生成回复。
这样的接口让 ChatRoom 不必知道每个 bot 的内部逻辑，只需要依次判断、调用，并将返回的消息写入日志。
这里也有一个值得注意的设计选择：AwayBot 不仅需要处理 /away 命令，还要扫描普通消息中的人名。
因此，它的激活条件不能简单地等同于某一条斜杠命令。
2.3. 第三步：给状态找主人
把全局状态迁移到负责管理它的组件中：
- AwayBot 保存用户的离开状态。
- TacoBot 保存用户的 taco 总数。
- ChatRoom 只管理消息分发和聊天日志。
难点在于 MeetBot。
会议开始时，它需要将发起者和参与者的状态设为 in a meeting，但这份状态由 AwayBot 管理。
这正是本题最值得讨论的依赖关系：业务操作发生在 MeetBot，状态却归 AwayBot 所有。
小型系统可以让 MeetBot 直接调用负责状态管理的服务。
随着系统扩大，也可以将状态管理独立成 PresenceService，让 AwayBot 和 MeetBot 都依赖它。
不必为了体现设计模式而把每个概念都拆成一个类。
关键是让状态的归属和修改入口足够清楚。
2.4. 第四步：决定组件如何协作
这里有两种常见方案。
方案 A：直接调用
MeetBot 通过注入的依赖调用状态管理服务，直接更新双方状态。
优点是调用链明确、容易调试，读代码就能知道谁在修改状态。
缺点是调用方需要了解被调用的服务。
如果未来出现更多跨组件协作，直接依赖可能逐渐增多。
不过，依赖数量取决于实际的协作关系，并不意味着每新增一个 bot，就必须修改所有其他 bot。
方案 B：事件总线
MeetBot 发布 MeetingStarted 事件，AwayBot 订阅该事件并更新双方状态。
这样 MeetBot 不必直接依赖 AwayBot。
未来增加会议通知、统计等订阅者时，通常不需要修改 MeetBot，只需增加订阅处理器并完成系统注册。
代价是依赖关系变得隐式。
看到 publish()，不一定能从当前代码直接判断有哪些组件会响应；
排查问题也需要追踪事件订阅关系。
因此，事件总线并非天然优于直接调用。
它更适合发布者与多个潜在订阅者之间确实需要解耦的场景。
本题的合理做法是：先完成直接依赖下的职责拆分，再引入事件总线。
这样每次变化的目的更明确，也更容易定位回归问题。
对于同步事件总线，只要事件处理时机、状态更新时机、bot 调用顺序和输出顺序都保持一致，原有黄金测试就应继续通过。
如果输出变化，应先确定变化来自架构迁移还是预期中的行为修复。
2.5. 第五步：最后修 bug，把行为变化单独提交
结构重构完成并通过回归测试后，再修复原代码中的缺陷。
例如：
/meet缺少参与者时，不应该因为访问split()[1]而崩溃。/givetaco缺少接收者时，不应该因为解包元素不足而抛出异常。- 人名匹配应该使用明确的 token 匹配规则，而不是未经约束的子串匹配。
- 大小写、@ 前缀、标点以及斜杠命令的处理规则，都应该经过测试确认。
这些问题不能不加区分地混在结构重构里修复。
否则，即使最终测试失败，也很难知道是设计调整引入了回归，还是 bug 修复改变了旧行为。
每项行为修复单独提交，配套测试明确说明新规则。
这样，重构与功能变化的边界就清楚了。
3. 核心代码：一个最小的事件驱动实现
下面展示的是经过输入校验后的最小实现。
它体现了前面的设计思路，但不等于历史重构的每个中间版本。
实际面试中，应该先实现行为保持不变的版本，再单独提交行为修复。
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass
def command_of(text: str) -> str:
parts = text.strip().split(maxsplit=1)
return parts[0] if parts else ""
class EventBus:
"""最小同步事件总线。"""
def __init__(self):
self._subs = defaultdict(list)
def subscribe(self, event_type, handler):
self._subs[event_type].append(handler)
def publish(self, event):
# 使用快照，避免遍历期间订阅列表变化。
for handler in tuple(self._subs[type(event)]):
handler(event)
@dataclass(frozen=True)
class MeetingStarted:
participants: tuple[str, ...]
class BotService(ABC):
@abstractmethod
def should_activate(self, sender: str, text: str) -> bool:
"""只判断，不修改状态。"""
raise NotImplementedError
@abstractmethod
def execute(self, sender: str, text: str) -> list[str]:
"""执行逻辑并返回待输出的消息。"""
raise NotImplementedError
class AwayBot(BotService):
def __init__(self, bus: EventBus):
self._away = {}
bus.subscribe(MeetingStarted, self._on_meeting)
def _on_meeting(self, event: MeetingStarted):
for user in event.participants:
self._away[user] = "in a meeting"
def should_activate(self, sender: str, text: str) -> bool:
command = command_of(text)
return (
command == "/away"
or (bool(text.strip()) and not command.startswith("/"))
)
def execute(self, sender: str, text: str) -> list[str]:
parts = text.strip().split(maxsplit=1)
if not parts:
return []
command = parts[0]
if command == "/away":
if len(parts) < 2 or not parts[1].strip():
return ["Usage: /away <reason>"]
self._away[sender] = parts[1].strip()
return []
# 不对其他斜杠命令执行人名匹配。
if command.startswith("/"):
return []
# 去除常见标点和可选的 @ 前缀。
# 此示例采用区分大小写的用户名匹配。
names = {
word.strip(".,!?;:()[]{}<>\"'").lstrip("@")
for word in text.split()
}
return [
f"{user} is away: {reason}"
for user, reason in self._away.items()
if user in names
]
class MeetBot(BotService):
def __init__(self, bus: EventBus):
self._bus = bus
def should_activate(self, sender: str, text: str) -> bool:
return command_of(text) == "/meet"
def execute(self, sender: str, text: str) -> list[str]:
parts = text.strip().split()
if len(parts) != 2:
return ["Usage: /meet @user"]
target = parts[1].lstrip("@")
if not target or target.startswith("/"):
return ["Usage: /meet @user"]
self._bus.publish(
MeetingStarted(participants=(sender, target))
)
return [f"{sender} and {target} are meeting"]
class TacoBot(BotService):
def __init__(self):
self._totals = defaultdict(int)
def should_activate(self, sender: str, text: str) -> bool:
return command_of(text) == "/givetaco"
def execute(self, sender: str, text: str) -> list[str]:
parts = text.strip().split()
if len(parts) != 3:
return ["Usage: /givetaco 🌮🌮 @user"]
tacos, recipient = parts[1], parts[2].lstrip("@")
if not recipient or recipient.startswith("/"):
return ["Usage: /givetaco 🌮🌮 @user"]
self._totals[recipient] += tacos.count("🌮")
return [
f"{recipient} now has {self._totals[recipient]} tacos"
]
class ChatRoom:
def __init__(self, bots):
self.log = []
self._bots = bots
def send_message(self, sender: str, text: str):
self.log.append(f"{sender}: {text}")
for bot in self._bots:
if bot.should_activate(sender, text):
self.log.extend(bot.execute(sender, text))
bus = EventBus()
away_bot = AwayBot(bus)
room = ChatRoom([
away_bot,
MeetBot(bus),
TacoBot(),
])
这里有几个值得注意的细节。
第一，MeetBot 发布的是一场会议事件。 MeetingStarted 携带双方参与者，一次事件就能表达完整的业务事实。
AwayBot 收到事件后负责更新状态。
第二，事件总线仍然是同步的。 publish() 会依次执行订阅处理器；
如果其中一个处理器抛出异常，异常会向上传播，后续处理器也不会再执行。
这个最小实现没有提供重试、异常隔离或持久化能力，不应把它误认为生产级事件基础设施。
第三，bot 注册顺序会影响输出顺序。 当前 ChatRoom 按列表顺序处理 bot，所以迁移时应保持原有顺序。
只有题目要求引入独立于注册顺序的优先级时，才需要另行设计显式排序规则。
第四，命令帮助可以通过元数据扩展。 可以为每个 bot 定义 name、usage 等属性，再由 ChatRoom 或帮助处理器遍历注册表生成 /help。
但这属于扩展功能，不是本次重构必须解决的问题。
4. 测试：不能只验证即时回复
除了从旧实现生成的黄金测试，还应该编写针对状态和协作关系的单元测试。
例如，验证会议事件确实改变双方状态：
def test_meeting_marks_both_users_away():
bus = EventBus()
away_bot = AwayBot(bus)
room = ChatRoom([
away_bot,
MeetBot(bus),
TacoBot(),
])
room.send_message("Bob", "/meet @Alice")
room.send_message("Carol", "Hi @Bob and @Alice")
assert room.log == [
"Bob: /meet @Alice",
"Bob and Alice are meeting",
"Carol: Hi @Bob and @Alice",
"Bob is away: in a meeting",
"Alice is away: in a meeting",
]
再验证 taco 数量能否正确累计：
def test_tacos_accumulate():
bus = EventBus()
room = ChatRoom([
AwayBot(bus),
MeetBot(bus),
TacoBot(),
])
room.send_message("Carol", "/givetaco 🌮🌮 @Alice")
room.send_message("Bob", "/givetaco 🌮 @Alice")
assert room.log[-1] == "Alice now has 3 tacos"
这些测试展示了最终设计应具备的基本行为。
正式重构时，原实现的黄金测试仍然需要保留，不能用新写的单元测试取代它。
5. 面试追问：从能实现到能设计
5.1. 新增一种 bot，需要改几处？
在当前设计中，通常需要实现新的 bot 类，并将其注册进 ChatRoom。
如果它需要响应某类事件，还需要订阅相应的事件处理器。
关键是避免为了新增一种功能，反复修改已经稳定的其他 bot。
5.2. 事件总线换成异步之后，问题是什么？
消息回复的先后顺序不再天然确定。
还需要考虑事件是否持久化、失败是否重试、如何避免重复处理，以及订阅者之间是否需要隔离。
异步化不是把 publish() 改成异步函数就算完成。
5.3. 如果多个 bot 同时命中一条消息怎么办？
当前实现依赖注册顺序。
如果业务要求固定的优先级，就应定义显式优先级，并测试排序规则；
如果仍然依赖注册顺序，就应该把它视为系统契约，而不是偶然的实现细节。
5.4. 如果 AwayBot 越来越复杂怎么办？
当它开始负责大量不同的状态规则时，可以把状态管理抽成 PresenceService。
但不要仅仅为了套用单一职责原则，就提前制造过多的抽象。
6. 面试现场，可以这样回答
先把旧系统的输出和状态变化固定成回归测试，确保重构不会意外改变已有行为。
然后列出各个 bot 负责的业务与状态，把状态迁移到明确的管理者，设计职责清晰的接口。
最后解释 MeetBot 如何让双方进入开会状态：直接依赖更显式、更容易追踪；
事件总线降低发布者与订阅者的直接耦合，但会增加间接调用和排障成本。
先完成纯重构并通过回归测试，再单独修复行为 bug。
这道题真正考察的是工程判断：知道什么应该改变，也知道什么暂时不能改变，并且能够用测试证明自己的判断。