From 03db2ccf1c2dad70b47d929c53bc3cba445e48b9 Mon Sep 17 00:00:00 2001 From: yibo Date: Fri, 5 Jun 2026 17:44:28 +0800 Subject: [PATCH] feat(lifelog/diarize): v1.2 长会议分块 diarize + 断点续跑 + 阈值 0.35/0.40 --- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/SettingsScreen.kt | 2 +- lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt | 137 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------- 2 files changed, 83 insertions(+), 56 deletions(-) diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/SettingsScreen.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/SettingsScreen.kt index aad8b51..ecd8cac 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/SettingsScreen.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/SettingsScreen.kt @@ -88,7 +88,7 @@ fun SettingsScreen(activity: Context) { HorizontalDivider() Text("声纹", style = MaterialTheme.typography.titleMedium) DropdownSetting("聚类阈值（越大分越细）", "%.2f".format(threshold), - listOf(0.45f, 0.50f, 0.55f, 0.60f, 0.65f).map { "%.2f".format(it) to it }) { + listOf(0.35f, 0.40f, 0.45f, 0.50f, 0.55f, 0.60f, 0.65f).map { "%.2f".format(it) to it }) { threshold = it; prefs.edit().putFloat(SpeakerClusterer.KEY_THRESHOLD, it).apply() } Text("改完到「声纹」tab 点「重新聚类」对历史段生效；新录音自动用新值。", diff --git a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt index 7e4a06d..dd97cce 100644 --- a/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt +++ b/lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt @@ -34,10 +34,15 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke companion object { const val TAG = "DiarizationWorker" private val isRunning = AtomicBoolean(false) - const val RUN_BUDGET_MS = 7 * 60 * 1000L + // WorkManager 给每个 worker 硬上限 10min，超时直接 kill。预算压到 5min， + // 留足余量让最后一块 diarize 收尾 + 全局 recluster + 写 cursor，绝不撞红线。 + const val RUN_BUDGET_MS = 5 * 60 * 1000L const val MIN_SESSION_SEGMENTS = 2 // 单段会话无需 diarize - const val MAX_CONCAT_SAMPLES = 10 * 60 * 16000 // ≤10min speech，防 OOM（与 photolog 共进程）/超时 + // 单块 ≤3min speech：一次 diarize ~3min，远小于 10min 红线，一轮能跑完多块、 + // 多次写 cursor 推进；块内说话人 index 不跨块——身份交全局 reclusterAll。 + const val CHUNK_SAMPLES = 3 * 60 * 16000 private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs + private const val KEY_CURSOR = "diar_cursor_" // + key → 已处理段数（断点续跑长会议） } private val sr = 16000 @@ -79,8 +84,10 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke var engine: DiarizationEngine? = null var embedder: SpeakerEmbedder? = null var processed = 0 + var attempted = false try { - for (session in sessions) { + // 最新会话优先（用户更关心最近的；旧会话后续轮次补）。 + for (session in sessions.reversed()) { val start = session.first().startTs val end = session.last().endTs val key = start.toString() @@ -98,8 +105,17 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke embedder = SpeakerEmbedder.load(speakerFile.path, numThreads = 2) } - val ok = diarizeSession(repo, session, engine!!, embedder!!, threshold) - if (ok) { done.add(key); processed++ } + attempted = true + val startIdx = prefs.getInt(KEY_CURSOR + key, 0) + val reached = diarizeSession(repo, session, engine!!, embedder!!, runStart, startIdx) + if (reached >= session.size) { + done.add(key); processed++ + prefs.edit().remove(KEY_CURSOR + key).apply() + } else { + // 预算中途用完：记录断点，下轮从此处续跑（不重头），然后退出本轮。 + prefs.edit().putInt(KEY_CURSOR + key, reached).apply() + Log.i(TAG, "session $key partial: cursor=$reached/${session.size}, resume next round"); break + } } } finally { engine?.release() @@ -107,8 +123,8 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke } // 全局重聚类：用 diarization 产出的"干净"per-turn embedding 统一聚类 - //（reclusterAll 本就全局 + 保留人工命名锚点）。只在有新会话被处理时跑，处理完即稳定。 - if (processed > 0) { + //（reclusterAll 本就全局 + 保留人工命名锚点）。只要这轮碰过会话就跑。 + if (attempted) { val n = SpeakerClusterer.reclusterAll(repo, matchThreshold = threshold) Log.i(TAG, "global recluster → $n speakers") } @@ -117,63 +133,74 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke return Result.success() } - /** 对单个会话做 diarization + 段重标。返回是否成功处理。 */ + /** + * 分块 diarize 整个会话（长会议不再只取前 N min），从 [startIdx] 段断点续跑。 + * 返回**已处理到的段下标**（== session.size 表示整场跑完；< 表示预算中途用完， + * 外层据此存断点、下轮从这里续跑——块内说话人 index 不跨块，身份交全局 reclusterAll）。 + */ private suspend fun diarizeSession( repo: Repository, session: List, engine: DiarizationEngine, embedder: SpeakerEmbedder, - threshold: Float, - ): Boolean = withContext(Dispatchers.Default) { - // 1. 解码 + 拼接，记录每段在拼接波形里的 [offset, offset+len]（采样） - data class SegSpan(val seg: Segment, val from: Int, val to: Int) - val spans = ArrayList(session.size) - val chunks = ArrayList(session.size) // 不装箱：存 FloatArray 块，最后一次性拼 - var total = 0 - for (seg in session) { - val pcm = AudioDecoder.decodeToFloat(seg.audioPath) - if (pcm.isEmpty()) continue - val from = total - chunks.add(pcm); total += pcm.size - spans.add(SegSpan(seg, from, total)) - if (total >= MAX_CONCAT_SAMPLES) break - } - if (spans.size < MIN_SESSION_SEGMENTS || total == 0) return@withContext false - val samples = FloatArray(total) - var off = 0 - for (c in chunks) { c.copyInto(samples, off); off += c.size } - chunks.clear() - - // 2. diarize - val turns = engine.diarize(samples) - if (turns.isEmpty()) return@withContext false - - // 3. 每段 → 重叠最大的 turn → 取该 turn 的"干净"embedding（单说话人）覆盖段 embedding。 - // 不在这里建全局说话人——交给最后的全局 reclusterAll 统一聚类（避免每会话贪心爆炸）。 - val turnEmb = HashMap() // turn 下标 → embedding（缓存，多段共用一个 turn） + runStart: Long, + startIdx: Int, + ): Int = withContext(Dispatchers.Default) { + var idx = startIdx.coerceIn(0, session.size) + var nChunks = 0 + var totalTurns = 0 var embedded = 0 - for (span in spans) { - val s0 = span.from.toFloat() / sr - val s1 = span.to.toFloat() / sr - var bestIdx = -1 - var bestOv = 0f - turns.forEachIndexed { i, t -> - val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec) - if (ov > bestOv) { bestOv = ov; bestIdx = i } + while (idx < session.size) { + if (System.currentTimeMillis() - runStart > RUN_BUDGET_MS) { + Log.i(TAG, "session ${session.first().startTs}: budget hit at seg $idx/${session.size}") + return@withContext idx // 断点，外层续跑 } - if (bestIdx < 0) continue - val emb = turnEmb.getOrPut(bestIdx) { - val t = turns[bestIdx] - val a = (t.startSec * sr).toInt().coerceIn(0, samples.size) - val b = (t.endSec * sr).toInt().coerceIn(a, samples.size) - if (b - a < sr / 4) null else embedder.embed(samples.copyOfRange(a, b), sr) + // 攒一块（≤CHUNK_SAMPLES），解码即拼，记录每段在块内的 [from,to] + val segs = ArrayList() + val froms = ArrayList() + val tos = ArrayList() + val pcms = ArrayList() + var len = 0 + while (idx < session.size && len < CHUNK_SAMPLES) { + val seg = session[idx]; idx++ + val pcm = AudioDecoder.decodeToFloat(seg.audioPath) + if (pcm.isEmpty()) continue + segs.add(seg); froms.add(len); pcms.add(pcm); len += pcm.size; tos.add(len) } - if (emb != null) { - withContext(Dispatchers.IO) { repo.segments.updateEmbedding(span.seg.id, emb.toLeBytes()) } - embedded++ + if (segs.isEmpty()) continue + val samples = FloatArray(len) + var off = 0 + for (p in pcms) { p.copyInto(samples, off); off += p.size } + pcms.clear() + + val turns = engine.diarize(samples) + if (turns.isEmpty()) continue + nChunks++; totalTurns += turns.size + + val turnEmb = HashMap() // turn 下标 → embedding（多段共用同一 turn 缓存） + for (k in segs.indices) { + val s0 = froms[k].toFloat() / sr + val s1 = tos[k].toFloat() / sr + var bestIdx = -1 + var bestOv = 0f + turns.forEachIndexed { i, t -> + val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec) + if (ov > bestOv) { bestOv = ov; bestIdx = i } + } + if (bestIdx < 0) continue + val emb = turnEmb.getOrPut(bestIdx) { + val t = turns[bestIdx] + val a = (t.startSec * sr).toInt().coerceIn(0, samples.size) + val b = (t.endSec * sr).toInt().coerceIn(a, samples.size) + if (b - a < sr / 4) null else embedder.embed(samples.copyOfRange(a, b), sr) + } + if (emb != null) { + withContext(Dispatchers.IO) { repo.segments.updateEmbedding(segs[k].id, emb.toLeBytes()) } + embedded++ + } } } - Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, $embedded clean-embedded") - embedded > 0 + Log.i(TAG, "session ${session.first().startTs}: $startIdx→${session.size} segs, $nChunks chunks, $totalTurns turns, $embedded clean-embedded") + idx // == session.size，整场跑完 } } -- libgit2 0.21.4