fix(lifelog/diarize): v1.1 全局重聚类 + 修 OOM

解决 v1 的说话人爆炸（98）+ 内存崩溃： - 不再每会话贪心建全局说话人。改为：diarization 给每段存"干净"的 dominant-turn embedding（覆盖实时的混合 embedding），run 末尾跑一次 SpeakerClusterer.reclusterAll 做全局聚类（本就全局 + 保留人工命名锚点）。 - 修 OutOfMemoryError：拼接会话音频原用 ArrayList<Float>（每 float 装箱 16B， 6min 会话 ~92MB 垃圾）→ 改 List<FloatArray> 一次性 copy；AudioDecoder 的 ArrayList<Short> → ByteArrayOutputStream。MAX_CONCAT 18→10min 留余量。注：Daos.SegmentDao.updateEmbedding 这一行（diarization 需要）未在此 commit，随 photolog WIP 一起；提交 photolog 时带上即可。实测：无 OOM；但说话人仍 89——因单轮 7min 预算只 diarize 了 3/14 会话，其余仍是脏 embedding。periodic 多跑几轮全部 diarize 后才会真正收缩。有声书测试数据本就对声纹是地狱级，真实多人数据 + 阈值微调是正路。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

fix(lifelog/diarize): v1.1 全局重聚类 + 修 OOM
解决 v1 的说话人爆炸（98）+ 内存崩溃： - 不再每会话贪心建全局说话人。改为：diarization 给每段存"干净"的 dominant-turn embedding（覆盖实时的混合 embedding），run 末尾跑一次 SpeakerClusterer.reclusterAll 做全局聚类（本就全局 + 保留人工命名锚点）。 - 修 OutOfMemoryError：拼接会话音频原用 ArrayList<Float>（每 float 装箱 16B， 6min 会话 ~92MB 垃圾）→ 改 List<FloatArray> 一次性 copy；AudioDecoder 的 ArrayList<Short> → ByteArrayOutputStream。MAX_CONCAT 18→10min 留余量。注：Daos.SegmentDao.updateEmbedding 这一行（diarization 需要）未在此 commit，随 photolog WIP 一起；提交 photolog 时带上即可。实测：无 OOM；但说话人仍 89——因单轮 7min 预算只 diarize 了 3/14 会话，其余仍是脏 embedding。periodic 多跑几轮全部 diarize 后才会真正收缩。有声书测试数据本就对声纹是地狱级，真实多人数据 + 阈值微调是正路。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
yibo
1 parent 67e45743
Showing 2 changed files with 50 additions and 42 deletions
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt
@@ -41,7 +41,7 @@ object AudioDecoder {
             var srcRate = inFormat.getIntOr(MediaFormat.KEY_SAMPLE_RATE, TARGET_SR)
             var channels = inFormat.getIntOr(MediaFormat.KEY_CHANNEL_COUNT, 1)
  
-            val shorts = ArrayList<Short>(TARGET_SR * 30)
+            val pcm = java.io.ByteArrayOutputStream(TARGET_SR * 60)  // 累积 PCM16 字节，不装箱
             val info = MediaCodec.BufferInfo()
             var inEos = false
             var outEos = false
@@ -72,9 +72,10 @@ object AudioDecoder {
                     oi >= 0 -> {
                         if (info.size > 0) {
                             val ob = codec.getOutputBuffer(oi)!!
-                            ob.position(info.offset); ob.limit(info.offset + info.size)
-                            val sb = ob.order(ByteOrder.LITTLE_ENDIAN).asShortBuffer()
-                            while (sb.hasRemaining()) shorts.add(sb.get())
+                            val bytes = ByteArray(info.size)
+                            ob.position(info.offset)
+                            ob.get(bytes)
+                            pcm.write(bytes)
                         }
                         codec.releaseOutputBuffer(oi, false)
                         if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) outEos = true
@@ -83,15 +84,18 @@ object AudioDecoder {
             }
             codec.stop(); codec.release()
  
+            val raw = pcm.toByteArray()
+            val sbuf = java.nio.ByteBuffer.wrap(raw).order(ByteOrder.LITTLE_ENDIAN).asShortBuffer()
+            val nShorts = sbuf.remaining()
             // 下混 stereo → mono
             var mono = if (channels >= 2) {
-                FloatArray(shorts.size / channels) { i ->
+                FloatArray(nShorts / channels) { i ->
                     var s = 0f
-                    for (c in 0 until channels) s += shorts[i * channels + c] / 32768f
+                    for (c in 0 until channels) s += sbuf.get(i * channels + c) / 32768f
                     s / channels
                 }
             } else {
-                FloatArray(shorts.size) { shorts[it] / 32768f }
+                FloatArray(nShorts) { sbuf.get(it) / 32768f }
             }
             if (srcRate != TARGET_SR) mono = resample(mono, srcRate, TARGET_SR)
             mono
@@ -8,6 +8,7 @@ import com.lemon.lifelog.asr.DiarizationEngine
 import com.lemon.lifelog.asr.ModelDownloader
 import com.lemon.lifelog.asr.SpeakerClusterer
 import com.lemon.lifelog.asr.SpeakerEmbedder
+import com.lemon.lifelog.asr.toLeBytes
 import com.lemon.lifelog.audio.AudioDecoder
 import com.lemon.lifelog.data.Repository
 import com.lemon.lifelog.data.Segment
@@ -35,7 +36,7 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
         private val isRunning = AtomicBoolean(false)
         const val RUN_BUDGET_MS = 7 * 60 * 1000L
         const val MIN_SESSION_SEGMENTS = 2          // 单段会话无需 diarize
-        const val MAX_CONCAT_SAMPLES = 18 * 60 * 16000  // ≤18min speech，防内存/超时
+        const val MAX_CONCAT_SAMPLES = 10 * 60 * 16000  // ≤10min speech，防 OOM（与 photolog 共进程）/超时
         private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs
     }
  
@@ -105,6 +106,12 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
             embedder?.release()
         }
  
+        // 全局重聚类：用 diarization 产出的"干净"per-turn embedding 统一聚类
+        //（reclusterAll 本就全局 + 保留人工命名锚点）。只在有新会话被处理时跑，处理完即稳定。
+        if (processed > 0) {
+            val n = SpeakerClusterer.reclusterAll(repo, matchThreshold = threshold)
+            Log.i(TAG, "global recluster → $n speakers")
+        }
         prefs.edit().putStringSet(KEY_DONE, done).apply()
         Log.i(TAG, "done: ${sessions.size} sessions, $processed diarized")
         return Result.success()
@@ -121,55 +128,52 @@ class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorke
         // 1. 解码 + 拼接，记录每段在拼接波形里的 [offset, offset+len]（采样）
         data class SegSpan(val seg: Segment, val from: Int, val to: Int)
         val spans = ArrayList<SegSpan>(session.size)
-        val audio = ArrayList<Float>(sr * 60)
+        val chunks = ArrayList<FloatArray>(session.size)  // 不装箱：存 FloatArray 块，最后一次性拼
+        var total = 0
         for (seg in session) {
             val pcm = AudioDecoder.decodeToFloat(seg.audioPath)
             if (pcm.isEmpty()) continue
-            val from = audio.size
-            for (v in pcm) audio.add(v)
-            spans.add(SegSpan(seg, from, audio.size))
-            if (audio.size >= MAX_CONCAT_SAMPLES) break
+            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 || audio.isEmpty()) return@withContext false
-        val samples = FloatArray(audio.size) { audio[it] }
+        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 的 embedding → 桥接全局
-        val byLocal = turns.groupBy { it.speaker }
-        val clusterer = SpeakerClusterer(repo, matchThreshold = threshold).also { it.load() }
-        val localToGlobal = HashMap<Int, Pair<Long, Float>>() // local → (globalSpeakerId, sim)
-        for ((local, ts) in byLocal) {
-            // 局部说话人总时长 <2s 视为噪声，不桥接（避免一堆短 turn 各建一个全局说话人）。
-            val totalSec = ts.sumOf { (it.endSec - it.startSec).toDouble() }
-            if (totalSec < 2.0) continue
-            val longest = ts.maxByOrNull { it.endSec - it.startSec } ?: continue
-            val a = (longest.startSec * sr).toInt().coerceIn(0, samples.size)
-            val b = (longest.endSec * sr).toInt().coerceIn(a, samples.size)
-            if (b - a < sr / 2) continue  // <0.5s 太短，跳过该局部说话人
-            val emb = embedder.embed(samples.copyOfRange(a, b), sr) ?: continue
-            val durMs = ((longest.endSec - longest.startSec) * 1000).toLong()
-            val asg = clusterer.assign(emb, durMs)
-            if (asg.speakerId != null) localToGlobal[local] = asg.speakerId to (asg.sim ?: 1f)
-        }
-        if (localToGlobal.isEmpty()) return@withContext false
-
-        // 4. 每段 → 重叠最大的 turn → 全局 speakerId
+        // 3. 每段 → 重叠最大的 turn → 取该 turn 的"干净"embedding（单说话人）覆盖段 embedding。
+        //    不在这里建全局说话人——交给最后的全局 reclusterAll 统一聚类（避免每会话贪心爆炸）。
+        val turnEmb = HashMap<Int, FloatArray?>()  // turn 下标 → embedding（缓存，多段共用一个 turn）
+        var embedded = 0
         for (span in spans) {
             val s0 = span.from.toFloat() / sr
             val s1 = span.to.toFloat() / sr
-            var bestTurn: DiarizationEngine.Turn? = null
+            var bestIdx = -1
             var bestOv = 0f
-            for (t in turns) {
+            turns.forEachIndexed { i, t ->
                 val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec)
-                if (ov > bestOv) { bestOv = ov; bestTurn = t }
+                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(span.seg.id, emb.toLeBytes()) }
+                embedded++
             }
-            val g = bestTurn?.let { localToGlobal[it.speaker] } ?: continue
-            withContext(Dispatchers.IO) { repo.segments.updateSpeaker(span.seg.id, g.first, g.second) }
         }
-        Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, ${byLocal.size} local speakers")
-        true
+        Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, $embedded clean-embedded")
+        embedded > 0
     }
 }