宜博 / pixelphone

Commit 03db2ccf1c2dad70b47d929c53bc3cba445e48b9

Authored by yibo
1 parent a764c53c

feat(lifelog/diarize): v1.2 长会议分块 diarize + 断点续跑 + 阈值 0.35/0.40 

整场会议不再截断在前 ~140 段：按 3min speech 分块独立 diarize（块内说话人
index 不跨块，身份交全局 reclusterAll），断点 cursor 持久化到 prefs，下轮从
上次中断的段续跑而非重头。

修卡死真 bug：WorkManager 对每个 worker 有 10min 硬上限，原来 8min 大 chunk
一次 diarize 逼近上限被系统 kill、cursor 写不回去 → 永远卡在同一段。改成
3min chunk + 5min 预算，一轮跑完多块、多次写 cursor，绝不撞红线。实测 473 段
真实会议 4 轮自动跑完（142→232→336→424→done）。

SettingsScreen 阈值下拉补 0.35/0.40（0.4 过合并、0.55 过分裂，甜点 ~0.45）。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Inline Side-by-side
Showing 2 changed files with 83 additions and 56 deletions
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/SettingsScreen.kt
  View file @03db2cc
... ... @@ -88,7 +88,7 @@ fun SettingsScreen(activity: Context) {
88 88 HorizontalDivider()
89 89 Text("声纹", style = MaterialTheme.typography.titleMedium)
90 90 DropdownSetting("聚类阈值（越大分越细）", "%.2f".format(threshold),
91   - listOf(0.45f, 0.50f, 0.55f, 0.60f, 0.65f).map { "%.2f".format(it) to it }) {
  91 + listOf(0.35f, 0.40f, 0.45f, 0.50f, 0.55f, 0.60f, 0.65f).map { "%.2f".format(it) to it }) {
92 92 threshold = it; prefs.edit().putFloat(SpeakerClusterer.KEY_THRESHOLD, it).apply()
93 93 }
94 94 Text("改完到「声纹」tab 点「重新聚类」对历史段生效；新录音自动用新值。",
... ...
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt
  View file @03db2cc
... ... @@ -34,10 +34,15 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
34 34 companion object {
35 35 const val TAG = "DiarizationWorker"
36 36 private val isRunning = AtomicBoolean(false)
37   - const val RUN_BUDGET_MS = 7 * 60 * 1000L
  37 + // WorkManager 给每个 worker 硬上限 10min，超时直接 kill。预算压到 5min，
  38 + // 留足余量让最后一块 diarize 收尾 + 全局 recluster + 写 cursor，绝不撞红线。
  39 + const val RUN_BUDGET_MS = 5 * 60 * 1000L
38 40 const val MIN_SESSION_SEGMENTS = 2 // 单段会话无需 diarize
39   - const val MAX_CONCAT_SAMPLES = 10 * 60 * 16000 // ≤10min speech，防 OOM（与 photolog 共进程）/超时
  41 + // 单块 ≤3min speech：一次 diarize ~3min，远小于 10min 红线，一轮能跑完多块、
  42 + // 多次写 cursor 推进；块内说话人 index 不跨块——身份交全局 reclusterAll。
  43 + const val CHUNK_SAMPLES = 3 * 60 * 16000
40 44 private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs
  45 + private const val KEY_CURSOR = "diar_cursor_" // + key → 已处理段数（断点续跑长会议）
41 46 }
42 47  
43 48 private val sr = 16000
... ... @@ -79,8 +84,10 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
79 84 var engine: DiarizationEngine? = null
80 85 var embedder: SpeakerEmbedder? = null
81 86 var processed = 0
  87 + var attempted = false
82 88 try {
83   - for (session in sessions) {
  89 + // 最新会话优先（用户更关心最近的；旧会话后续轮次补）。
  90 + for (session in sessions.reversed()) {
84 91 val start = session.first().startTs
85 92 val end = session.last().endTs
86 93 val key = start.toString()
... ... @@ -98,8 +105,17 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
98 105 embedder = SpeakerEmbedder.load(speakerFile.path, numThreads = 2)
99 106 }
100 107  
101   - val ok = diarizeSession(repo, session, engine!!, embedder!!, threshold)
102   - if (ok) { done.add(key); processed++ }
  108 + attempted = true
  109 + val startIdx = prefs.getInt(KEY_CURSOR + key, 0)
  110 + val reached = diarizeSession(repo, session, engine!!, embedder!!, runStart, startIdx)
  111 + if (reached >= session.size) {
  112 + done.add(key); processed++
  113 + prefs.edit().remove(KEY_CURSOR + key).apply()
  114 + } else {
  115 + // 预算中途用完：记录断点，下轮从此处续跑（不重头），然后退出本轮。
  116 + prefs.edit().putInt(KEY_CURSOR + key, reached).apply()
  117 + Log.i(TAG, "session $key partial: cursor=$reached/${session.size}, resume next round"); break
  118 + }
103 119 }
104 120 } finally {
105 121 engine?.release()
... ... @@ -107,8 +123,8 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
107 123 }
108 124  
109 125 // 全局重聚类：用 diarization 产出的"干净"per-turn embedding 统一聚类
110   - //（reclusterAll 本就全局 + 保留人工命名锚点）。只在有新会话被处理时跑，处理完即稳定。
111   - if (processed > 0) {
  126 + //（reclusterAll 本就全局 + 保留人工命名锚点）。只要这轮碰过会话就跑。
  127 + if (attempted) {
112 128 val n = SpeakerClusterer.reclusterAll(repo, matchThreshold = threshold)
113 129 Log.i(TAG, "global recluster → $n speakers")
114 130 }
... ... @@ -117,63 +133,74 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
117 133 return Result.success()
118 134 }
119 135  
120   - /** 对单个会话做 diarization + 段重标。返回是否成功处理。 */
  136 + /**
  137 + * 分块 diarize 整个会话（长会议不再只取前 N min），从 [startIdx] 段断点续跑。
  138 + * 返回**已处理到的段下标**（== session.size 表示整场跑完；< 表示预算中途用完，
  139 + * 外层据此存断点、下轮从这里续跑——块内说话人 index 不跨块，身份交全局 reclusterAll）。
  140 + */
121 141 private suspend fun diarizeSession(
122 142 repo: Repository,
123 143 session: List<Segment>,
124 144 engine: DiarizationEngine,
125 145 embedder: SpeakerEmbedder,
126   - threshold: Float,
127   - ): Boolean = withContext(Dispatchers.Default) {
128   - // 1. 解码 + 拼接，记录每段在拼接波形里的 [offset, offset+len]（采样）
129   - data class SegSpan(val seg: Segment, val from: Int, val to: Int)
130   - val spans = ArrayList<SegSpan>(session.size)
131   - val chunks = ArrayList<FloatArray>(session.size) // 不装箱：存 FloatArray 块，最后一次性拼
132   - var total = 0
133   - for (seg in session) {
134   - val pcm = AudioDecoder.decodeToFloat(seg.audioPath)
135   - if (pcm.isEmpty()) continue
136   - val from = total
137   - chunks.add(pcm); total += pcm.size
138   - spans.add(SegSpan(seg, from, total))
139   - if (total >= MAX_CONCAT_SAMPLES) break
140   - }
141   - if (spans.size < MIN_SESSION_SEGMENTS || total == 0) return@withContext false
142   - val samples = FloatArray(total)
143   - var off = 0
144   - for (c in chunks) { c.copyInto(samples, off); off += c.size }
145   - chunks.clear()
146   -
147   - // 2. diarize
148   - val turns = engine.diarize(samples)
149   - if (turns.isEmpty()) return@withContext false
150   -
151   - // 3. 每段 → 重叠最大的 turn → 取该 turn 的"干净"embedding（单说话人）覆盖段 embedding。
152   - // 不在这里建全局说话人——交给最后的全局 reclusterAll 统一聚类（避免每会话贪心爆炸）。
153   - val turnEmb = HashMap<Int, FloatArray?>() // turn 下标 → embedding（缓存，多段共用一个 turn）
  146 + runStart: Long,
  147 + startIdx: Int,
  148 + ): Int = withContext(Dispatchers.Default) {
  149 + var idx = startIdx.coerceIn(0, session.size)
  150 + var nChunks = 0
  151 + var totalTurns = 0
154 152 var embedded = 0
155   - for (span in spans) {
156   - val s0 = span.from.toFloat() / sr
157   - val s1 = span.to.toFloat() / sr
158   - var bestIdx = -1
159   - var bestOv = 0f
160   - turns.forEachIndexed { i, t ->
161   - val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec)
162   - if (ov > bestOv) { bestOv = ov; bestIdx = i }
  153 + while (idx < session.size) {
  154 + if (System.currentTimeMillis() - runStart > RUN_BUDGET_MS) {
  155 + Log.i(TAG, "session ${session.first().startTs}: budget hit at seg $idx/${session.size}")
  156 + return@withContext idx // 断点，外层续跑
163 157 }
164   - if (bestIdx < 0) continue
165   - val emb = turnEmb.getOrPut(bestIdx) {
166   - val t = turns[bestIdx]
167   - val a = (t.startSec * sr).toInt().coerceIn(0, samples.size)
168   - val b = (t.endSec * sr).toInt().coerceIn(a, samples.size)
169   - if (b - a < sr / 4) null else embedder.embed(samples.copyOfRange(a, b), sr)
  158 + // 攒一块（≤CHUNK_SAMPLES），解码即拼，记录每段在块内的 [from,to]
  159 + val segs = ArrayList<Segment>()
  160 + val froms = ArrayList<Int>()
  161 + val tos = ArrayList<Int>()
  162 + val pcms = ArrayList<FloatArray>()
  163 + var len = 0
  164 + while (idx < session.size && len < CHUNK_SAMPLES) {
  165 + val seg = session[idx]; idx++
  166 + val pcm = AudioDecoder.decodeToFloat(seg.audioPath)
  167 + if (pcm.isEmpty()) continue
  168 + segs.add(seg); froms.add(len); pcms.add(pcm); len += pcm.size; tos.add(len)
170 169 }
171   - if (emb != null) {
172   - withContext(Dispatchers.IO) { repo.segments.updateEmbedding(span.seg.id, emb.toLeBytes()) }
173   - embedded++
  170 + if (segs.isEmpty()) continue
  171 + val samples = FloatArray(len)
  172 + var off = 0
  173 + for (p in pcms) { p.copyInto(samples, off); off += p.size }
  174 + pcms.clear()
  175 +
  176 + val turns = engine.diarize(samples)
  177 + if (turns.isEmpty()) continue
  178 + nChunks++; totalTurns += turns.size
  179 +
  180 + val turnEmb = HashMap<Int, FloatArray?>() // turn 下标 → embedding（多段共用同一 turn 缓存）
  181 + for (k in segs.indices) {
  182 + val s0 = froms[k].toFloat() / sr
  183 + val s1 = tos[k].toFloat() / sr
  184 + var bestIdx = -1
  185 + var bestOv = 0f
  186 + turns.forEachIndexed { i, t ->
  187 + val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec)
  188 + if (ov > bestOv) { bestOv = ov; bestIdx = i }
  189 + }
  190 + if (bestIdx < 0) continue
  191 + val emb = turnEmb.getOrPut(bestIdx) {
  192 + val t = turns[bestIdx]
  193 + val a = (t.startSec * sr).toInt().coerceIn(0, samples.size)
  194 + val b = (t.endSec * sr).toInt().coerceIn(a, samples.size)
  195 + if (b - a < sr / 4) null else embedder.embed(samples.copyOfRange(a, b), sr)
  196 + }
  197 + if (emb != null) {
  198 + withContext(Dispatchers.IO) { repo.segments.updateEmbedding(segs[k].id, emb.toLeBytes()) }
  199 + embedded++
  200 + }
174 201 }
175 202 }
176   - Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, $embedded clean-embedded")
177   - embedded > 0
  203 + Log.i(TAG, "session ${session.first().startTs}: $startIdx→${session.size} segs, $nChunks chunks, $totalTurns turns, $embedded clean-embedded")
  204 + idx // == session.size，整场跑完
178 205 }
179 206 }
... ...