用户对着麦克风说了一句话："明天早上八点提醒我开会。"
然后 Gemini 说"好的"，对话继续。
第二天早上八点整，扬声器里真的有声音响起来，说"该开会了"。
这中间到底发生了什么？
我在做自己的私人 AI 助手时，实现了这套提醒机制。整条链路从麦克风一路跑到扬声器，中间经过了语音识别、工具调用、数据库、Redis 事件、WebSocket、再回到 Gemini Live 的语音合成——但没有一行轮询代码，也没有引入任何任务队列框架。
把这条路从头跟到尾，每一步为什么这么做，有什么坑，说清楚。
背景：这套系统的基本架构
先说清楚大背景，不然后面很多东西没有上下文。
这个 AI 助手用的是 Gemini Live API，一个端到端的全双工实时语音接口。它不是"STT → LLM → TTS"三段式拼接，而是一个原生的语音模型：用户直接把 PCM 音频流送进去，它直接返回 PCM 音频流，中间没有独立的语音识别和语音合成步骤。
前端用 Web Audio API 采集麦克风，采样率 16kHz，16-bit PCM，通过 WebSocket 持续发给后端。后端转手就发给 Gemini Live 的长连接。Gemini 说话时返回 24kHz 的 PCM 流，后端再通过 WebSocket 推给前端播放。
整个系统大概长这样：
前端麦克风 (16kHz PCM)
↕ WebSocket
Go 后端
↕ HTTP/2 长连接
Gemini Live API在这个架构里，Gemini 不仅能说话，还能调用工具——后端预先注册了一批函数（查时间、存记忆、搜网页、设提醒等），Gemini 在理解用户意图后，会主动发起函数调用请求，后端执行完把结果塞回去，Gemini 再用语音告诉用户结果。
提醒功能就是通过这套工具调用机制接进来的。
第一跳：声音变成工具调用
用户说"明天早上八点提醒我开会"，这句话以 16kHz PCM 的格式通过 WebSocket 持续发往后端，后端直接转给 Gemini Live 的实时连接。
Gemini Live 边听边理解，不需要等用户说完再处理。当它从语音里识别出"提醒"这个意图，会在内部决策：这个需求对应我注册的 setReminder 工具，然后发起一次函数调用请求，结构大概长这样：
json
{
"name": "setReminder",
"args": {
"content": "开会",
"remind_at": "2026-10-07 08:00:00"
}
}后端的 receive loop 监听 Gemini Live 的返回流，收到工具调用请求后，查找注册的工具函数，直接执行，执行完把结果塞回 Gemini 的会话上下文。这一套流程在一个独立的 goroutine 里跑，不阻塞主音频流的接收。
Gemini 收到工具执行结果（"提醒设置成功"）之后，用语音告诉用户"好的，明天八点我会提醒你"。用户听到了回复，整件事在用户感知里就这么简单。
但后端这边，工具函数里真正的工作才刚开始。
第二跳：写两个地方
setReminder 工具函数被调用时，需要把这条提醒记下来，并且安排好到时候怎么触发。
最直接的做法是写数据库，然后起一个定时任务每分钟扫一次，看有没有到期的提醒。很多项目就是这么做的。缺点是精度最多到分钟级，而且扫表的频率和表的大小成反比——表越大越慢，即使加了索引，几十万条数据每分钟扫一遍也是无谓的 IO。
我用的是另一个方案：Redis 的键过期事件。
思路很简单：在 Redis 里写一个带 TTL 的 key，TTL 就是距离提醒时间的秒数。key 一过期，Redis 会自动往 Pub/Sub 频道推一条消息，后端订阅这个频道，收到消息就触发提醒逻辑。全程没有轮询，Redis 主动推过来才动。
但这里有一个细节要处理：key 过期之后，这个 key 的数据也跟着消失了。如果把提醒内容直接存在那个会过期的 key 上，等到过期事件触发时，数据已经没了，什么都拿不到。
所以实际上写了两个地方，分工明确：
go
func AddReminder(ctx context.Context, userID int, content string, remindAt time.Time) (*Reminder, error) {
now := time.Now()
ttl := remindAt.Sub(now)
if ttl <= 0 {
return nil, fmt.Errorf("提醒时间不能早于当前时间")
}
// 第一笔：写数据库，持久化，这是最终的真实数据
rem, err := db.Reminder.Create().
SetUserID(userID).
SetContent(content).
SetRemindAt(remindAt).
SetIsSent(false).
Save(ctx)
if err != nil {
return nil, fmt.Errorf("写入数据库失败: %w", err)
}
// 第二笔：把核心字段快照存进 Redis Hash（不过期，纯内存）
// key: "reminder:data", field: "提醒ID", value: JSON
payload, _ := json.Marshal(ReminderPayload{
ID: rem.ID,
UserID: userID,
Content: content,
RemindAt: remindAt.Format(time.DateTime),
})
redisClient.HSet(ctx, "reminder:data", strconv.Itoa(rem.ID), payload)
// 第三笔：写一个纯计时用的空 key，TTL 到期即触发过期事件
// key: "reminder:timer:42"，value 是空字符串，我们只关心它什么时候消失
timerKey := fmt.Sprintf("reminder:timer:%d", rem.ID)
redisClient.Set(ctx, timerKey, "", ttl)
return rem, nil
}reminder:data 这个 Hash 不过期，存真正的提醒内容；reminder:timer:{id} 这个 String key 是倒计时炸弹，TTL 耗尽自动消失，消失的瞬间触发过期事件。
两张牌各司其职，不互相干扰。
第三跳：Redis 的过期事件
Redis 有一个叫 Keyspace Notification 的功能，默认是关闭的，需要手动开启。开启之后，每当任意 key 过期，Redis 会往 __keyevent@{db}__:expired 这个 Pub/Sub 频道推一条消息，内容就是那个 key 的名字。
后端在启动时订阅这个频道，然后一直等着：
go
func StartReminderListener(ctx context.Context) {
// 自动开启键空间过期通知，省去手动配置 redis.conf 的麻烦
redisClient.ConfigSet(ctx, "notify-keyspace-events", "Ex")
// 用模式订阅，匹配所有数据库的过期事件
pubsub := redisClient.PSubscribe(ctx, "__keyevent@*__:expired")
defer pubsub.Close()
ch := pubsub.Channel()
log.Info("Redis 过期事件监听器已就绪")
for {
select {
case <-ctx.Done():
return
case msg, ok := <-ch:
if !ok {
return
}
// Redis 里所有 key 过期都会推过来，过滤出我们自己的
if !strings.HasPrefix(msg.Payload, "reminder:timer:") {
continue
}
// 从 key 名里解析出提醒 ID
idStr := strings.TrimPrefix(msg.Payload, "reminder:timer:")
id, err := strconv.Atoi(idStr)
if err != nil {
continue
}
// 异步处理，不阻塞监听循环
go handleExpired(id)
}
}
}这个监听器跑在一个独立的 goroutine 里，和业务逻辑完全隔离。它不知道提醒的内容是什么，也不关心怎么通知用户，它只负责一件事：告诉后端"第 42 号提醒到时间了"。
第四跳：拿数据，两级降级
知道提醒 ID 之后，需要找到对应的内容，才能告诉用户该做什么。
最快的路是 Redis Hash——它就在内存里，一次 HGET 几乎没有延迟。但 Redis 不是永久存储，可能重启，可能内存不足触发 LRU 驱逐。如果只靠 Redis，一旦数据不在了，提醒就会静悄悄地失火，用户什么都收不到。
所以有两级：先查 Redis，没有就查数据库。
go
func handleExpired(reminderID int) {
ctx := context.Background()
idStr := strconv.Itoa(reminderID)
var payload ReminderPayload
// 第一优先：查 Redis Hash，内存里直接拿，毫秒级
raw, err := redisClient.HGet(ctx, "reminder:data", idStr).Result()
if err == nil && raw != "" {
json.Unmarshal([]byte(raw), &payload)
}
// 兜底：Redis 没有数据，降级查数据库
// 这是防御性设计，正常情况下 Redis 里应该有
if payload.UserID == 0 {
rem, err := db.Reminder.Query().
Where(
reminder.IDEQ(reminderID),
reminder.IsSentEQ(false), // 幂等保护：已发送的不重复处理
).
WithUser().
Only(ctx)
if err != nil {
log.Warn("提醒触发但找不到数据", "id", reminderID, "err", err)
return
}
payload = ReminderPayload{
ID: rem.ID,
UserID: rem.Edges.User.ID,
Content: rem.Content,
}
}
log.Info("提醒到期触发", "id", payload.ID, "content", payload.Content)
// 触发外部通知（具体怎么通知，不在这个包里决定）
if Notifier != nil {
Notifier(uint(payload.UserID), payload.ID, payload.Content)
}
// 收尾：标记数据库状态为已发送，清理 Redis Hash
db.Reminder.UpdateOneID(payload.ID).
SetIsSent(true).
SetSentAt(time.Now()).
Exec(ctx)
redisClient.HDel(ctx, "reminder:data", idStr)
}数据库查询加了 IsSentEQ(false) 的条件，这是一道幂等保护——在极端情况下（比如重启时有积压的过期事件），同一条提醒可能被触发两次，这个条件确保已经发送过的不会重复处理。
第五跳：函数指针注入，切断循环依赖
注意上面代码里 Notifier 这个变量，它是一个函数指针，定义在 reminder 包里：
go
// reminder 包
type NotifierFunc func(userID uint, reminderID int, content string)
var Notifier NotifierFunc
func SetNotifier(fn NotifierFunc) {
Notifier = fn
}为什么要绕这么一圈，不直接在 handleExpired 里调用 WebSocket 的推送函数？
因为包依赖关系不允许。
负责推送 WebSocket 消息的是 ws 包。ws 包要管理 Gemini Live 会话，所以它引用了 genkit 包。genkit 包注册工具函数时，需要用 reminder 包的 AddReminder。
画出来就是：
ws → genkit → reminder如果再让 reminder → ws，就形成了一个环：reminder → ws → genkit → reminder。Go 编译器对循环依赖零容忍，直接报错，没有商量余地。
解法是把"怎么通知"这件事的决策权从 reminder 包里抽走。reminder 包只定义一个函数指针类型，声明"我需要有人来处理通知"，但不关心是谁、怎么处理。
然后在 ws 包启动时，把具体的推送逻辑注入进去：
go
// ws 包，Hub 启动时执行
func (m *Manager) bindReminderNotifier() {
reminder.SetNotifier(func(userID uint, reminderID int, content string) {
// 第一路：推 WebSocket 消息给前端界面
wsMsg := WSMessage{
Type: "reminder_notify",
Data: map[string]any{
"id": reminderID,
"content": content,
"ts": time.Now().UnixMilli(),
},
}
if sent := m.SendToUser(userID, wsMsg); !sent {
log.Warn("用户不在线，WebSocket 通知未送达", "uid", userID)
}
// 第二路：如果用户正在语音对话，让 Gemini 亲口说出来
go m.notifyLiveSession(userID, content)
})
}依赖方向保持单向：ws 知道 reminder，reminder 不知道任何人。函数指针在这里比定义一个 interface 再去实现它要省事得多——没有额外的类型定义，注入一行代码就完事。
第六跳：WebSocket 推送前端
SendToUser 找到这个用户当前所有在线的 WebSocket 连接，逐个塞进发送 channel。每个 Client 有一个带缓冲的 Send chan []byte，写 goroutine 会把 channel 里的数据真正写出去。
go
func (m *Manager) SendToUser(userID uint, msg any) bool {
// 先做一次快照，避免持锁时间过长
m.Lock.RLock()
devices := m.Clients[userID]
snapshot := make([]*Client, 0, len(devices))
for _, c := range devices {
snapshot = append(snapshot, c)
}
m.Lock.RUnlock()
if len(snapshot) == 0 {
return false
}
data, _ := json.Marshal(msg)
sent := false
for _, client := range snapshot {
select {
case client.Send <- data:
sent = true
default:
// 发送缓冲区满了，这条消息只能丢掉
// 对于个人系统，这种情况极少发生，记个日志就行
log.Warn("发送缓冲区已满，丢弃消息", "uid", userID, "device", client.DeviceID)
}
}
return sent
}前端收到 reminder_notify 消息类型，弹出提醒通知。
但这里有一个场景是前端通知覆盖不到的：用户戴着耳机，正在和 AI 语音对话，界面是最小化的，屏幕可能都锁着。这时候一个悄无声息的 WebSocket 消息，根本引不起任何注意。
最后一跳：让 Gemini 亲口说出来
如果用户当前有一个活跃的 Gemini Live 实时语音会话，最自然的通知方式不是弹个消息框，而是让 Gemini 直接开口说。
go
func (m *Manager) notifyLiveSession(userID uint, content string) {
m.Lock.RLock()
devices := m.Clients[userID]
var sessionIDs []int64
for _, client := range devices {
if client.SessionID > 0 {
sessionIDs = append(sessionIDs, client.SessionID)
}
}
m.Lock.RUnlock()
for _, sessID := range sessionIDs {
ls, ok := liveSessionStore.Get(sessID)
if !ok {
continue
}
// 往 Gemini Live 的会话上下文里注入一条"系统消息"
// Gemini 收到后会自己决定怎么开口，用它的语音风格说出来
err := ls.SendClientContent(LiveClientContentInput{
Turns: []Content{{
Role: "user",
Parts: []Part{{Text: fmt.Sprintf("提醒：%s", content)}},
}},
TurnComplete: true, // 告诉 Gemini 这条消息说完了，可以回答了
})
if err != nil {
log.Warn("触发 Gemini Live 语音播报失败", "session", sessID, "err", err)
}
break // 找到一个活跃会话就够了
}
}TurnComplete: true 是这里的关键参数。它的意思是：这条消息到这里就结束了，不会有更多内容追加，Gemini 你现在可以开口回答了。如果不加这个，Gemini 会以为用户还在说话，继续等待，什么都不会说。
Gemini 收到这条注入消息后，会用它自己的风格组织语言，然后用语音说出来。整个系统里没有独立的 TTS 模块——AI 的语音输出就是 Gemini Live 直接返回的 24kHz PCM 流，和正常对话时的声音是同一个声音、同一个质感。
完整链路，串起来看
把六跳放在一起：
[用户说话]
麦克风 16kHz PCM
↓ WebSocket binary frame
Go 后端转发给 Gemini Live 长连接
↓ Gemini 识别意图，发起工具调用
setReminder 工具函数执行
├─→ 写 PostgreSQL（持久化，真实数据）
├─→ 写 Redis Hash（元数据快照，内存级速度）
└─→ 写 Redis SETEX（倒计时炸弹，TTL 结束即触发）
[N 分钟后，TTL 耗尽]
Redis key 过期
↓ __keyevent expired Pub/Sub 推出
Go 监听 goroutine 收到事件
↓
handleExpired(id)
├─ 查 Redis Hash → 拿到提醒内容（正常路径）
└─ 降级查 PostgreSQL → 兜底（Redis 数据丢失时）
↓
调用 reminder.Notifier（函数指针回调，解耦循环依赖）
├─→ WebSocket push → 前端界面弹通知
└─→ 检查是否有活跃 Gemini Live 会话
↓ 有
SendClientContent → Gemini 判断如何回答
↓
Gemini 开口，24kHz PCM 流 → WebSocket → 前端播放三个设计决定
为什么不用 cron 定时扫表？
每分钟扫一次，精度最多 ±60 秒。表一大，扫的代价就上去了。Redis 过期事件的触发精度通常在 1 秒以内，对个人助手来说完全够用，而且没有任何额外的 IO 负担。
为什么 Hash 和 SETEX 要分成两个 key？
因为 key 过期时，key 上的数据也跟着没了。如果把提醒内容直接存在那个计时 key 上，触发时数据已经消失，什么都拿不到。把"倒计时"和"数据"拆成两个独立的 key，职责清晰，互不影响。
为什么不引入任务队列框架（比如 Asynq、Machinery）？
这是个人系统，单机单实例，没有高并发场景。引入任务队列意味着更多的部署依赖、更复杂的失败重试逻辑、以及更长的排查链路。Redis 键空间通知这套方案，核心代码不到 150 行，能覆盖的场景已经够用，超出这个范围再说。
这条链路里让我觉得最有意思的，是最后一跳：把一条文字消息注入 Gemini Live 的会话上下文，然后 Gemini 自己开口把它说出来。没有 TTS，没有语音模板，声音的质感和平时对话完全一样，因为本来就是同一个模型在说话。
AI
评论交流