[Solution] OpenAI高频题 Social Network / Follow Graph — Parts 1-4 详解高频题
冷目灼灼的扁豆
This is my writeup of the Social Network / Follow Graph problem, one of the current high-frequency OpenAI coding questions. It comes in 4 progressive parts plus a design follow-up. Each part builds on the previous one, and each has exactly one key insight — get that right and the code writes itself.
Part 1: immutable snapshots
SocialNetwork holds user -> set(followees). create_snapshot() must return a view that never changes afterward. The trap: a shallow dict copy still shares the inner sets, so later follow() calls would leak into old snapshots. Deep-copy the inner sets (and freeze them as frozensets — cheap insurance against accidental mutation).
import bisect
import heapq
from collections import defaultdict
class Snapshot:
def __init__(self, following):
self._following = {u: frozenset(s) for u, s in following.items()}
followers = {u: [] for u in self._following}
for u, followees in self._following.items():
for v in followees:
followers[v].append(u)
self._following_sorted = {u: tuple(sorted(s)) for u, s in self._following.items()}
self._followers_sorted = {u: tuple(sorted(l)) for u, l in followers.items()}
def is_following(self, follower, followee):
return followee in self._following.get(follower, ())
def get_following(self, user_id):
return list(self._following_sorted.get(user_id, ()))
def get_followers(self, user_id):
return list(self._followers_sorted.get(user_id, ()))
def recommend(self, user_id, k):
if k <= 0:
return []
direct = self._following.get(user_id, frozenset())
score = defaultdict(int)
for f in direct:
for g in self._following.get(f, ()):
if g != user_id and g not in direct:
score[g] += 1
return heapq.nsmallest(k, score, key=lambda g: (-score[g], g))
class SocialNetwork:
def __init__(self):
self._following = {}
def add_user(self, user_id):
self._following.setdefault(user_id, set())
def follow(self, follower, followee):
if follower not in self._following or followee not in self._following:
raise KeyError("unknown user")
if follower == followee:
return # self-follow is a no-op
self._following[follower].add(followee) # duplicate follow is a no-op (set)
def create_snapshot(self):
return Snapshot(self._following)
Part 2: followers
get_followers needs the reverse direction. Don't scan every user per call — build the reverse index once at snapshot construction, plus sorted tuples for both directions (already in the code above). This is safe precisely because the snapshot is immutable; on a mutable graph it would be a staleness bug. Queries become O(answer); construction is O(n + m).
Part 3: two-hop recommendations
Candidate g is reached through a followee f of the user (user -> f -> g), excluding the user and anyone already followed. Score = number of distinct followees reaching g. Since each followee's followee-set has no duplicates, incrementing once per (f, g) pair counts distinct followees correctly. Top-k by (score desc, id asc) with a heap — no need to sort all candidates. Cost: O(sum of followees' out-degrees + C log k).
Part 4: historical queries
New class, FollowTimeline: follow/unfollow(follower, followee, t) arrive in non-decreasing t, and is_following(follower, followee, t) asks about any t. Snapshots copy the whole graph; here we store, per pair, only the timestamps where the state actually flipped. A pair starts "not following", so the state at time t is following iff an odd number of flips happened at or before t: bisect_right(flips, t) odd. Calls only ever append (thanks to the non-decreasing guarantee), so queries are O(log e).
Two subtleties: same-t follow-then-unfollow records both flips, and bisect_right counts both, so the later call wins. Redundant calls (follow while already following) record nothing.
class FollowTimeline:
def __init__(self):
self._flips = defaultdict(list)
self._last_t = None
def _check_time(self, t):
if self._last_t is not None and t < self._last_t:
raise ValueError("timestamps must be non-decreasing")
self._last_t = t
def _currently_following(self, key):
return len(self._flips[key]) % 2 == 1
def follow(self, follower, followee, t):
self._check_time(t)
key = (follower, followee)
if follower != followee and not self._currently_following(key):
self._flips[key].append(t)
def unfollow(self, follower, followee, t):
self._check_time(t)
key = (follower, followee)
if self._currently_following(key):
self._flips[key].append(t)
def is_following(self, follower, followee, t):
flips = self._flips.get((follower, followee))
return bool(flips) and bisect.bisect_right(flips, t) % 2 == 1
Follow-up (discussion): copy-on-write snapshots
If snapshots are frequent and edits are few, a snapshot can be an O(1) reference to the current dict of frozensets; follow() replaces only the touched user's frozenset (plus the top-level dict — or use a persistent hash map to make that O(log n) too). Old snapshots keep pointing at old versions. The reverse index and sorted views become lazy, cached per snapshot.
Pitfalls that actually decide this question
1. Shallow-copying the snapshot (shared inner sets).
2. Rebuilding the reverse index per query instead of once at construction.
3. Counting non-distinct followees in recommendations.
4. bisect_left instead of bisect_right in Part 4 — same-timestamp ordering breaks.
5. Recording flips for redundant calls — corrupts the parity.
Happy to discuss alternatives.
Part 1: immutable snapshots
SocialNetwork holds user -> set(followees). create_snapshot() must return a view that never changes afterward. The trap: a shallow dict copy still shares the inner sets, so later follow() calls would leak into old snapshots. Deep-copy the inner sets (and freeze them as frozensets — cheap insurance against accidental mutation).
import bisect
import heapq
from collections import defaultdict
class Snapshot:
def __init__(self, following):
self._following = {u: frozenset(s) for u, s in following.items()}
followers = {u: [] for u in self._following}
for u, followees in self._following.items():
for v in followees:
followers[v].append(u)
self._following_sorted = {u: tuple(sorted(s)) for u, s in self._following.items()}
self._followers_sorted = {u: tuple(sorted(l)) for u, l in followers.items()}
def is_following(self, follower, followee):
return followee in self._following.get(follower, ())
def get_following(self, user_id):
return list(self._following_sorted.get(user_id, ()))
def get_followers(self, user_id):
return list(self._followers_sorted.get(user_id, ()))
def recommend(self, user_id, k):
if k <= 0:
return []
direct = self._following.get(user_id, frozenset())
score = defaultdict(int)
for f in direct:
for g in self._following.get(f, ()):
if g != user_id and g not in direct:
score[g] += 1
return heapq.nsmallest(k, score, key=lambda g: (-score[g], g))
class SocialNetwork:
def __init__(self):
self._following = {}
def add_user(self, user_id):
self._following.setdefault(user_id, set())
def follow(self, follower, followee):
if follower not in self._following or followee not in self._following:
raise KeyError("unknown user")
if follower == followee:
return # self-follow is a no-op
self._following[follower].add(followee) # duplicate follow is a no-op (set)
def create_snapshot(self):
return Snapshot(self._following)
Part 2: followers
get_followers needs the reverse direction. Don't scan every user per call — build the reverse index once at snapshot construction, plus sorted tuples for both directions (already in the code above). This is safe precisely because the snapshot is immutable; on a mutable graph it would be a staleness bug. Queries become O(answer); construction is O(n + m).
Part 3: two-hop recommendations
Candidate g is reached through a followee f of the user (user -> f -> g), excluding the user and anyone already followed. Score = number of distinct followees reaching g. Since each followee's followee-set has no duplicates, incrementing once per (f, g) pair counts distinct followees correctly. Top-k by (score desc, id asc) with a heap — no need to sort all candidates. Cost: O(sum of followees' out-degrees + C log k).
Part 4: historical queries
New class, FollowTimeline: follow/unfollow(follower, followee, t) arrive in non-decreasing t, and is_following(follower, followee, t) asks about any t. Snapshots copy the whole graph; here we store, per pair, only the timestamps where the state actually flipped. A pair starts "not following", so the state at time t is following iff an odd number of flips happened at or before t: bisect_right(flips, t) odd. Calls only ever append (thanks to the non-decreasing guarantee), so queries are O(log e).
Two subtleties: same-t follow-then-unfollow records both flips, and bisect_right counts both, so the later call wins. Redundant calls (follow while already following) record nothing.
class FollowTimeline:
def __init__(self):
self._flips = defaultdict(list)
self._last_t = None
def _check_time(self, t):
if self._last_t is not None and t < self._last_t:
raise ValueError("timestamps must be non-decreasing")
self._last_t = t
def _currently_following(self, key):
return len(self._flips[key]) % 2 == 1
def follow(self, follower, followee, t):
self._check_time(t)
key = (follower, followee)
if follower != followee and not self._currently_following(key):
self._flips[key].append(t)
def unfollow(self, follower, followee, t):
self._check_time(t)
key = (follower, followee)
if self._currently_following(key):
self._flips[key].append(t)
def is_following(self, follower, followee, t):
flips = self._flips.get((follower, followee))
return bool(flips) and bisect.bisect_right(flips, t) % 2 == 1
Follow-up (discussion): copy-on-write snapshots
If snapshots are frequent and edits are few, a snapshot can be an O(1) reference to the current dict of frozensets; follow() replaces only the touched user's frozenset (plus the top-level dict — or use a persistent hash map to make that O(log n) too). Old snapshots keep pointing at old versions. The reverse index and sorted views become lazy, cached per snapshot.
Pitfalls that actually decide this question
1. Shallow-copying the snapshot (shared inner sets).
2. Rebuilding the reverse index per query instead of once at construction.
3. Counting non-distinct followees in recommendations.
4. bisect_left instead of bisect_right in Part 4 — same-timestamp ordering breaks.
5. Recording flips for redundant calls — corrupts the parity.
Happy to discuss alternatives.
已获得 12 大米
共0条回复
✨ 您正在体验新版论坛UI
