记录三模型话轮检测实验、44点评估及延迟对比

experiments/smart-turn-v3.2/test_replay.py

@@ -0,0 +1,79 @@
+"""Verify replay timing assumptions against Pipecat's actual buffer logic."""
+
+import asyncio
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import numpy as np
+from pipecat.audio.turn.base_turn_analyzer import EndOfTurnState
+from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn, SmartTurnParams
+
+import evaluate_work_data as evaluation
+
+
+class RecordingTurn(BaseSmartTurn):
+    prediction = 0
+
+    def _predict_endpoint(self, audio):
+        self.last_audio = audio.copy()
+        return {"prediction": self.prediction, "probability": 0.9 if self.prediction else 0.1}
+
+
+class ReplayTests(unittest.TestCase):
+    def setUp(self):
+        self.clock = evaluation.MediaClock()
+        self.time_patch = patch.object(evaluation.base_module, "time", SimpleNamespace(
+            monotonic=self.clock.monotonic, perf_counter=time.perf_counter))
+        self.time_patch.start()
+        self.turn = RecordingTurn(sample_rate=8000, params=SmartTurnParams())
+        self.turn.set_sample_rate(8000)
+        self.turn.update_vad_start_secs(0.2)
+        self.chunk = np.full(256, 1234, dtype=np.int16).tobytes()
+
+    def tearDown(self):
+        if self.turn._executor:
+            self.turn._executor.shutdown(wait=True)
+        self.time_patch.stop()
+
+    def feed(self, count, speaking):
+        for _ in range(count):
+            self.clock.seconds += 0.032
+            state = self.turn.append_audio(self.chunk, speaking)
+        return state
+
+    def test_incomplete_keeps_first_start_and_resumed_speech_resets_silence(self):
+        self.feed(10, False)
+        self.feed(20, True)
+        first_start = self.turn._speech_start_time
+        self.feed(60, False)  # 1.92 s pause, below the timeout.
+        state, _ = asyncio.run(self.turn.analyze_end_of_turn())
+        self.assertEqual(state, EndOfTurnState.INCOMPLETE)
+        preserved_count = len(self.turn._audio_buffer)
+        self.feed(1, True)
+        self.assertEqual(self.turn._speech_start_time, first_start)
+        self.assertEqual(len(self.turn._audio_buffer), preserved_count + 1)
+        self.assertEqual(self.turn._silence_ms, 0)
+
+    def test_model_only_gets_last_eight_seconds_and_probe_does_not_clear(self):
+        self.feed(320, True)  # 10.24 s accumulated speech.
+        before_start = self.turn._speech_start_time
+        before_count = len(self.turn._audio_buffer)
+        self.turn.prediction = 1
+        state, _ = self.turn._process_speech_segment(self.turn._audio_buffer)
+        self.assertEqual(state, EndOfTurnState.COMPLETE)
+        self.assertEqual(len(self.turn.last_audio), 8 * 8000)
+        self.assertEqual(self.turn._speech_start_time, before_start)
+        self.assertEqual(len(self.turn._audio_buffer), before_count)
+
+    def test_three_second_silence_timeout_clears_old_origin(self):
+        self.feed(20, True)
+        state = self.feed(94, False)  # 3.008 s at 32 ms frame granularity.
+        self.assertEqual(state, EndOfTurnState.COMPLETE)
+        self.assertFalse(self.turn.speech_triggered)
+        self.assertEqual(self.turn._speech_start_time, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()