AI 语音聊天 Pipecat 源码阅读笔记

# frames.py:116-154
@dataclass
class Frame:
    id: int = field(init=False)       # globally unique
    name: str = field(init=False)     # e.g. "TextFrame#42"
    pts: Optional[int] = field(init=False)  # presentation timestamp (nanoseconds)
    metadata: Dict[str, Any] = field(init=False)

# frames.py:198-207
@dataclass
class UninterruptibleFrame:
    """A marker for data or control frames that must not be interrupted."""
    pass

class FrameProcessorQueue(asyncio.PriorityQueue):
    HIGH_PRIORITY = 1
    LOW_PRIORITY = 2

    async def put(self, item):
        frame, _, _ = item
        if isinstance(frame, SystemFrame):
            await super().put((self.HIGH_PRIORITY, self.__high_counter, item))
        else:
            await super().put((self.LOW_PRIORITY, self.__low_counter, item))

async def __input_frame_task_handler(self):
    while True:
        (frame, direction, callback) = await self.__input_queue.get()
        if isinstance(frame, SystemFrame):
            await self.__process_frame(frame, direction, callback)  # immediate
        else:
            await self.__process_queue.put((frame, direction, callback))  # deferred

# frame_processor.py:612-632
async def queue_frame(self, frame, direction, callback=None):
    if self._enable_direct_mode:
        await self.__process_frame(frame, direction, callback)  # no queue
    else:
        await self.__input_queue.put((frame, direction, callback))

# frame_processor.py:581-589
def link(self, processor: "FrameProcessor"):
    self._next = processor
    processor._prev = self

# frame_processor.py:909-944
async def __internal_push_frame(self, frame, direction):
    if direction == FrameDirection.DOWNSTREAM and self._next:
        await self._next.queue_frame(frame, direction)
    elif direction == FrameDirection.UPSTREAM and self._prev:
        await self._prev.queue_frame(frame, direction)

# pipeline.py:99-121
class Pipeline(BasePipeline):
    def __init__(self, processors, source=None, sink=None):
        self._source = source or PipelineSource(self.push_frame)
        self._sink = sink or PipelineSink(self.push_frame)
        self._processors = [self._source] + processors + [self._sink]
        self._link_processors()

    def _link_processors(self):
        prev = self._processors[0]
        for curr in self._processors[1:]:
            prev.link(curr)
            prev = curr

# pipeline.py:183-195
async def process_frame(self, frame, direction):
    if direction == FrameDirection.DOWNSTREAM:
        await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
    elif direction == FrameDirection.UPSTREAM:
        await self._sink.queue_frame(frame, FrameDirection.UPSTREAM)

# pipeline.py:21-52 (PipelineSource)
async def process_frame(self, frame, direction):
    match direction:
        case FrameDirection.UPSTREAM:
            await self._upstream_push_frame(frame, direction)  # escape to parent
        case FrameDirection.DOWNSTREAM:
            await self.push_frame(frame, direction)  # continue in sub-pipeline

# frame_processor.py:892-907
async def _start_interruption(self):
    if isinstance(self.__process_current_frame, UninterruptibleFrame):
        self.__reset_process_queue()      # drain queue, keep current frame
    else:
        await self.__cancel_process_task()  # cancel current processing
        self.__create_process_task()        # start fresh

# frame_processor.py:995-1012
def __reset_process_queue(self):
    new_queue = asyncio.Queue()
    while not self.__process_queue.empty():
        item = self.__process_queue.get_nowait()
        if isinstance(item[0], UninterruptibleFrame):
            new_queue.put_nowait(item)
    # swap queues...

# task.py:391-397
source = PipelineSource(self._source_push_frame)
self._sink = PipelineSink(self._sink_push_frame)
processors = [pipeline]  # user's pipeline
self._pipeline = Pipeline(processors, source=source, sink=self._sink)

# task.py:823-864
async def _process_push_queue(self):
    self._clock.start()

    # Step 1: Create and push StartFrame through the entire pipeline
    start_frame = StartFrame(
        allow_interruptions=...,
        audio_in_sample_rate=...,
        enable_metrics=...,
        ...
    )
    await self._pipeline.queue_frame(start_frame)

    # Step 2: Wait for StartFrame to traverse all processors
    await self._wait_for_pipeline_start(start_frame)

    # Step 3: Main loop — pull user-queued frames and feed them
    while running:
        frame = await self._push_queue.get()
        await self._pipeline.queue_frame(frame)
        if isinstance(frame, (CancelFrame, EndFrame, StopFrame)):
            await self._wait_for_pipeline_end(frame)
        running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame))

async def _source_push_frame(self, frame, direction):
    if isinstance(frame, EndTaskFrame):
        await self.queue_frame(EndFrame())       # graceful stop
    elif isinstance(frame, CancelTaskFrame):
        await self.queue_frame(CancelFrame())    # immediate cancel
    elif isinstance(frame, ErrorFrame):
        if frame.fatal:
            await self.queue_frame(CancelFrame())  # fatal → cancel everything

async def _sink_push_frame(self, frame, direction):
    if isinstance(frame, StartFrame):
        self._pipeline_start_event.set()    # signal: pipeline is ready
    elif isinstance(frame, EndFrame):
        self._pipeline_end_event.set()      # signal: pipeline is done
    elif isinstance(frame, HeartbeatFrame):
        await self._heartbeat_queue.put(frame)

@task.event_handler("on_pipeline_started")
async def on_pipeline_started(task, frame):
    # Pipeline is ready, StartFrame has traversed all processors
    ...

@task.event_handler("on_pipeline_finished")
async def on_pipeline_finished(task, frame):
    # Pipeline has terminated (EndFrame, StopFrame, or CancelFrame)
    ...

@task.event_handler("on_pipeline_error")
async def on_pipeline_error(task, frame):
    # An ErrorFrame reached the source
    ...

@task.event_handler("on_idle_timeout")
async def on_idle_timeout(task):
    # No activity for idle_timeout_secs
    ...

# runner.py:65-95
class PipelineRunner:
    async def run(self, task: PipelineTask):
        try:
            params = PipelineTaskParams(loop=self._loop)
            await task.run(params)
        except asyncio.CancelledError:
            pass

# run.py:114-153
def _get_bot_module():
    # 1. Check __main__ module (the file that was executed)
    main_module = sys.modules["__main__"]
    if hasattr(main_module, "bot"):
        return main_module

    # 2. Try importing 'bot' module from current directory
    # 3. Scan .py files in current directory for a 'bot' function

@dataclass
class RunnerArguments:                    # Base: body, cli_args
class DailyRunnerArguments(RunnerArguments):    # + room_url, token
class WebSocketRunnerArguments(RunnerArguments): # + websocket
class SmallWebRTCRunnerArguments(RunnerArguments): # + webrtc_connection

async def bot(runner_args: RunnerArguments):
    if isinstance(runner_args, DailyRunnerArguments):
        transport = DailyTransport(runner_args.room_url, runner_args.token, ...)
    elif isinstance(runner_args, SmallWebRTCRunnerArguments):
        transport = SmallWebRTCTransport(runner_args.webrtc_connection, ...)

@app.api_route(
    "/sessions/{session_id}/{path:path}",
    methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
)
async def proxy_request(session_id, path, request, background_tasks):
    # 1. Validate session exists
    active_session = active_sessions.get(session_id)
    if active_session is None:
        return Response(content="Invalid or not-yet-ready session_id", status_code=404)

    # 2. Route WebRTC signaling to the offer/ice handlers
    if path.endswith("api/offer"):
        if request.method == "POST":
            # Parse SDP offer, delegate to offer() handler
            webrtc_request = SmallWebRTCRequest(sdp=..., type=..., ...)
            return await offer(webrtc_request, background_tasks)
        elif request.method == "PATCH":
            # Parse ICE candidate, delegate to ice_candidate() handler
            return await ice_candidate(patch_request)

    # 3. Other paths: just acknowledge
    return Response(status_code=200)

# frame_processor.py:660-675
async def process_frame(self, frame: Frame, direction: FrameDirection):
    if self._observer:
        timestamp = self._clock.get_time() if self._clock else 0
        data = FrameProcessed(
            processor=self,      # 谁在处理
            frame=frame,         # 处理什么帧
            direction=direction, # 哪个方向
            timestamp=timestamp, # 精确时间戳
        )
        await self._observer.on_process_frame(data)
    # ... 然后才执行真正的处理逻辑

# frame_processor.py:910-942
async def __internal_push_frame(self, frame, direction):
    timestamp = self._clock.get_time() if self._clock else 0
    if direction == FrameDirection.DOWNSTREAM and self._next:
        if self._observer:
            data = FramePushed(
                source=self,           # 谁推送的
                destination=self._next, # 推给谁
                frame=frame,
                direction=direction,
                timestamp=timestamp,
            )
            await self._observer.on_push_frame(data)
        await self._next.queue_frame(frame, direction)
    elif direction == FrameDirection.UPSTREAM and self._prev:
        if self._observer:
            data = FramePushed(
                source=self,
                destination=self._prev,
                frame=frame,
                direction=direction,
                timestamp=timestamp,
            )
            await self._observer.on_push_frame(data)
        await self._prev.queue_frame(frame, direction)

# base_observer.py
class BaseObserver(BaseObject):
    async def on_process_frame(self, data: FrameProcessed):
        """帧到达处理器时调用"""
        pass

    async def on_push_frame(self, data: FramePushed):
        """帧从处理器推出时调用"""
        pass

    async def on_pipeline_started(self):
        """StartFrame 遍历完所有处理器后调用"""
        pass

# task_observer.py
class TaskObserver(BaseObserver):
    def _create_proxy(self, observer):
        queue = asyncio.Queue()            # 每个 Observer 独立队列
        task = self._task_manager.create_task(
            self._proxy_task_handler(queue, observer),  # 独立消费协程
            f"TaskObserver::{observer}::_proxy_task_handler",
        )
        return Proxy(queue=queue, task=task, observer=observer)

    async def _send_to_proxy(self, data):
        for proxy in self._proxies.values():
            await proxy.queue.put(data)     # 非阻塞地分发到所有队列

    async def _proxy_task_handler(self, queue, observer):
        while True:
            data = await queue.get()
            if isinstance(data, _PipelineStartedSignal):
                await observer.on_pipeline_started()
            elif isinstance(data, FramePushed):
                await observer.on_push_frame(data)
            elif isinstance(data, FrameProcessed):
                await observer.on_process_frame(data)
            queue.task_done()

# task.py:799-804 — Observer 注入到管道的每个处理器
setup = FrameProcessorSetup(
    clock=self._clock,
    task_manager=self._task_manager,
    observer=self._observer,     # TaskObserver 实例
)
await self._pipeline.setup(setup)  # 递归传播到所有处理器

# task.py:402 — TaskObserver 创建
self._observer = TaskObserver(observers=observers, task_manager=self._task_manager)

# startup_timing_observer.py:53-59
@dataclass
class _ArrivalInfo:
    """内部记录：StartFrame 到达某处理器的时间"""
    processor: FrameProcessor
    arrival_ts_ns: int          # 纳秒级时间戳

# 核心状态
self._arrivals: Dict[int, _ArrivalInfo] = {}   # processor.id → 到达信息
self._timings: List[ProcessorStartupTiming] = []  # 按管道顺序收集的耗时
self._start_frame_id: Optional[str] = None     # 锁定第一个 StartFrame（防重复）
self._start_frame_arrival_ns: Optional[int] = None  # 全局起始时间（算 offset 用）

# startup_timing_observer.py:224-247
async def on_process_frame(self, data: FrameProcessed):
    if self._startup_timing_reported:
        return                           # 报告已发，不再追踪

    if not isinstance(data.frame, StartFrame):
        return                           # 只关心 StartFrame

    # 锁定第一个 StartFrame（ParallelPipeline 中可能有多个 StartFrame 副本）
    if self._start_frame_id is None:
        self._start_frame_id = data.frame.id
        self._start_frame_arrival_ns = data.timestamp  # 全局起点
        self._start_wall_clock = time.time()
    elif data.frame.id != self._start_frame_id:
        return                           # 忽略其他 StartFrame

    if self._should_track(data.processor):
        self._arrivals[data.processor.id] = _ArrivalInfo(
            processor=data.processor,
            arrival_ts_ns=data.timestamp, # 记录到达时间
        )

# startup_timing_observer.py:200-212
_INTERNAL_TYPES = (PipelineSource, BasePipeline)  # 内部管道组件

def _should_track(self, processor):
    if self._processor_types is not None:
        return isinstance(processor, self._processor_types)  # 用户指定类型
    return not isinstance(processor, _INTERNAL_TYPES)        # 默认排除内部组件

# startup_timing_observer.py:249-289
async def on_push_frame(self, data: FramePushed):
    # ---- 传输层建连度量（与启动度量独立）----
    if isinstance(data.frame, BotConnectedFrame):
        self._handle_bot_connected(data)
        return
    if isinstance(data.frame, ClientConnectedFrame):
        await self._handle_client_connected(data)
        return

    # ---- StartFrame 度量 ----
    if self._startup_timing_reported:
        return
    if not isinstance(data.frame, StartFrame):
        return
    if self._start_frame_id is not None and data.frame.id != self._start_frame_id:
        return

    arrival = self._arrivals.pop(data.source.id, None)   # 取出到达记录
    if arrival is None:
        return

    duration_ns = data.timestamp - arrival.arrival_ts_ns  # 核心公式：推出时间 - 到达时间
    duration_secs = duration_ns / 1e9
    # offset = 该处理器开始的时间点 - 全局起始时间
    start_offset_secs = (arrival.arrival_ts_ns - self._start_frame_arrival_ns) / 1e9

    self._timings.append(
        ProcessorStartupTiming(
            processor_name=arrival.processor.name,
            start_offset_secs=start_offset_secs,    # 可以画 Gantt 图
            duration_secs=duration_secs,
        )
    )

# startup_timing_observer.py:214-222
async def on_pipeline_started(self):
    """StartFrame 遍历完所有处理器后由 PipelineTask 调用"""
    if self._timings:
        await self._emit_report()

async def _emit_report(self):
    self._startup_timing_reported = True   # 防重复
    total = sum(t.duration_secs for t in self._timings)
    report = StartupTimingReport(
        start_time=self._start_wall_clock or 0.0,
        total_duration_secs=total,
        processor_timings=self._timings,
    )
    await self._call_event_handler("on_startup_timing_report", report)

# startup_timing_observer.py:291-312
def _handle_bot_connected(self, data: FramePushed):
    """第一个 BotConnectedFrame → bot 加入房间的延迟"""
    if self._bot_connected_secs is not None:
        return  # 只记第一次
    delta_ns = data.timestamp - self._start_frame_arrival_ns
    self._bot_connected_secs = delta_ns / 1e9

async def _handle_client_connected(self, data: FramePushed):
    """第一个 ClientConnectedFrame → 客户端连入的延迟，并触发传输报告"""
    if self._transport_timing_reported:
        return
    self._transport_timing_reported = True
    delta_ns = data.timestamp - self._start_frame_arrival_ns
    report = TransportTimingReport(
        start_time=self._start_wall_clock or 0.0,
        bot_connected_secs=self._bot_connected_secs,     # SFU 传输才有
        client_connected_secs=delta_ns / 1e9,
    )
    await self._call_event_handler("on_transport_timing_report", report)

# turn_tracking_observer.py:46-67
def __init__(self, max_frames=100, turn_end_timeout_secs=2.5, **kwargs):
    self._turn_count = 0            # 总轮次计数
    self._is_turn_active = False    # 当前是否有活跃轮次
    self._is_bot_speaking = False   # Bot 正在说话？
    self._has_bot_spoken = False    # 当前轮次中 Bot 是否说过话？
    self._turn_start_time = 0       # 当前轮次开始时间（纳秒）
    self._turn_end_timeout_secs = turn_end_timeout_secs  # 默认 2.5 秒

    # 帧去重机制（同一帧可能被多个处理器推送，只处理一次）
    self._processed_frames = set()
    self._frame_history = deque(maxlen=max_frames)  # 有界，防内存泄漏

# turn_tracking_observer.py:78-89
async def on_push_frame(self, data: FramePushed):
    if data.frame.id in self._processed_frames:
        return  # 已处理过，跳过

    self._processed_frames.add(data.frame.id)
    self._frame_history.append(data.frame.id)

    # 有界 deque 满了后会自动丢弃最老的，但 set 不会
    # 所以需要定期从 deque 重建 set，防止 set 无限增长
    if len(self._processed_frames) > len(self._frame_history):
        self._processed_frames = set(self._frame_history)

# turn_tracking_observer.py:91-104 — 主分发逻辑
if isinstance(data.frame, StartFrame):
    if self._turn_count == 0:
        await self._start_turn(data)         # 管道启动 → 第一轮开始
elif isinstance(data.frame, UserStartedSpeakingFrame):
    await self._handle_user_started_speaking(data)
elif isinstance(data.frame, BotStartedSpeakingFrame):
    await self._handle_bot_started_speaking(data)
elif isinstance(data.frame, BotStoppedSpeakingFrame) and self._is_bot_speaking:
    await self._handle_bot_stopped_speaking(data)
elif isinstance(data.frame, (EndFrame, CancelFrame)):
    await self._handle_pipeline_end(data)

# turn_tracking_observer.py:131-149
async def _handle_user_started_speaking(self, data: FramePushed):
    if self._is_bot_speaking:
        # 场景 1: Bot 正在说话时用户开口 → 打断
        self._cancel_turn_end_timer()
        await self._end_turn(data, was_interrupted=True)   # 结束旧轮
        self._is_bot_speaking = False
        await self._start_turn(data)                        # 开始新轮

    elif self._is_turn_active and self._has_bot_spoken:
        # 场景 2: Bot 停了但还没超时（2.5s 内用户又说话了）
        # 这不算打断，但算新轮次的自然开始
        self._cancel_turn_end_timer()
        await self._end_turn(data, was_interrupted=False)
        await self._start_turn(data)

    elif not self._is_turn_active:
        # 场景 3: 前一轮已结束，用户重新开口
        await self._start_turn(data)

    else:
        # 场景 4: 用户在 Bot 回复前继续说话（比如补充了一句）
        logger.trace(f"User is already speaking in Turn {self._turn_count}")

# turn_tracking_observer.py:106-116
def _schedule_turn_end(self, data: FramePushed):
    self._cancel_turn_end_timer()
    loop = asyncio.get_event_loop()
    self._end_turn_timer = loop.call_later(
        self._turn_end_timeout_secs,   # 默认 2.5 秒
        lambda: asyncio.create_task(self._end_turn_after_timeout(data)),
    )

# turn_tracking_observer.py:151-156
async def _handle_bot_started_speaking(self, data: FramePushed):
    self._is_bot_speaking = True
    self._has_bot_spoken = True
    self._cancel_turn_end_timer()  # Bot 又开始说话了，取消超时

# turn_tracking_observer.py:174-193
async def _start_turn(self, data: FramePushed):
    self._is_turn_active = True
    self._has_bot_spoken = False
    self._turn_count += 1
    self._turn_start_time = data.timestamp
    await self._call_event_handler("on_turn_started", self._turn_count)

async def _end_turn(self, data: FramePushed, was_interrupted: bool):
    if not self._is_turn_active:
        return
    duration = (data.timestamp - self._turn_start_time) / 1_000_000_000
    self._is_turn_active = False
    await self._call_event_handler("on_turn_ended", self._turn_count, duration, was_interrupted)

# user_bot_latency_observer.py:172-200
def __init__(self, *, max_frames=100, **kwargs):
    # 核心时间戳
    self._user_stopped_time: Optional[float] = None     # VAD 确认用户停止说话的时间
    self._user_turn_start_time: Optional[float] = None  # 同上（用于 breakdown）
    self._user_turn: Optional[float] = None             # user turn 持续时间

    # 首次 Bot 说话追踪
    self._client_connected_time: Optional[float] = None
    self._first_bot_speech_measured: bool = False

    # 帧去重（同 TurnTrackingObserver）
    self._processed_frames: set = set()
    self._frame_history: deque = deque(maxlen=max_frames)

    # 每轮周期的指标累加器
    self._ttfb: List[TTFBBreakdownMetrics] = []
    self._text_aggregation: Optional[TextAggregationBreakdownMetrics] = None
    self._function_call_starts: Dict[str, tuple[str, float]] = {}
    self._function_call_metrics: List[FunctionCallMetrics] = []

# user_bot_latency_observer.py:206-280
async def on_push_frame(self, data: FramePushed):
    if data.direction != FrameDirection.DOWNSTREAM:
        return  # 只关心下游帧（数据流方向）

    # 帧去重（同 TurnTrackingObserver 的 deque+set 方案）
    if data.frame.id in self._processed_frames:
        return
    self._processed_frames.add(data.frame.id)
    self._frame_history.append(data.frame.id)
    ...

if isinstance(data.frame, VADUserStartedSpeakingFrame):
    self._user_stopped_time = None
    self._user_turn_start_time = None
    self._user_turn = None
    self._reset_accumulators()          # 清空 TTFB、文本聚合、工具调用指标
    self._first_bot_speech_measured = True  # 放弃首次 Bot 说话测量

elif isinstance(data.frame, VADUserStoppedSpeakingFrame):
    # 关键：减去 stop_secs（VAD 需要的静默检测时间）
    # 这样得到的是用户"真正"停止说话的时间
    self._user_stopped_time = data.frame.timestamp - data.frame.stop_secs
    self._user_turn_start_time = self._user_stopped_time

elif isinstance(data.frame, UserStoppedSpeakingFrame):
    if self._user_stopped_time is not None:
        self._user_turn = time.time() - self._user_stopped_time

elif isinstance(data.frame, InterruptionFrame):
    self._reset_accumulators()  # LLM/TTS 被取消了，之前收集的指标无效

elif isinstance(data.frame, FunctionCallInProgressFrame):
    # 工具开始执行 → 记录 tool_call_id 和开始时间
    self._function_call_starts[data.frame.tool_call_id] = (
        data.frame.function_name,
        time.time(),
    )
elif isinstance(data.frame, FunctionCallResultFrame):
    # 工具返回结果 → 计算耗时
    start = self._function_call_starts.pop(data.frame.tool_call_id, None)
    if start is not None:
        function_name, start_time = start
        self._function_call_metrics.append(
            FunctionCallMetrics(
                function_name=function_name,
                start_time=start_time,
                duration_secs=time.time() - start_time,
            )
        )

elif isinstance(data.frame, MetricsFrame):
    self._handle_metrics_frame(data.frame)

# user_bot_latency_observer.py:310-342
def _handle_metrics_frame(self, frame: MetricsFrame):
    # 只在有效测量窗口内收集（用户停了或等首次 Bot 说话）
    waiting_for_first_speech = (
        self._client_connected_time is not None and not self._first_bot_speech_measured
    )
    if self._user_stopped_time is None and not waiting_for_first_speech:
        return

    now = time.time()
    for metrics_data in frame.data:
        if isinstance(metrics_data, TTFBMetricsData) and metrics_data.value > 0:
            self._ttfb.append(
                TTFBBreakdownMetrics(
                    processor=metrics_data.processor,   # "DeepgramSTT", "OpenAILLM" 等
                    model=metrics_data.model,            # "gpt-4o", "nova-2" 等
                    start_time=now - metrics_data.value,  # 反算开始时间
                    duration_secs=metrics_data.value,
                )
            )
        elif isinstance(metrics_data, TextAggregationMetricsData):
            if self._text_aggregation is None:  # 只取第一个（影响首段语音的那个）
                self._text_aggregation = TextAggregationBreakdownMetrics(
                    processor=metrics_data.processor,
                    start_time=now - metrics_data.value,
                    duration_secs=metrics_data.value,
                )

# user_bot_latency_observer.py:282-308
async def _handle_bot_started_speaking(self):
    emit_breakdown = False

    # 一次性测量：客户端连入 → 首次 Bot 说话（greeting 延迟）
    if self._client_connected_time is not None and not self._first_bot_speech_measured:
        self._first_bot_speech_measured = True
        latency = time.time() - self._client_connected_time
        await self._call_event_handler("on_first_bot_speech_latency", latency)
        emit_breakdown = True

    # 常规测量：用户停止说话 → Bot 开始说话
    if self._user_stopped_time is not None:
        latency = time.time() - self._user_stopped_time
        self._user_stopped_time = None
        await self._call_event_handler("on_latency_measured", latency)
        emit_breakdown = True

    if emit_breakdown:
        breakdown = LatencyBreakdown(
            ttfb=list(self._ttfb),
            text_aggregation=self._text_aggregation,
            user_turn_start_time=self._user_turn_start_time,
            user_turn_secs=self._user_turn,
            function_calls=list(self._function_call_metrics),
        )
        await self._call_event_handler("on_latency_breakdown", breakdown)
        self._reset_accumulators()

# user_bot_latency_observer.py:114-141
class LatencyBreakdown(BaseModel):
    ttfb: List[TTFBBreakdownMetrics]                      # LLM TTFB, TTS TTFB
    text_aggregation: Optional[TextAggregationBreakdownMetrics]  # 句子聚合延迟
    user_turn_start_time: Optional[float]                  # 用户停止说话的时间
    user_turn_secs: Optional[float]                        # VAD+STT+分析器延迟
    function_calls: List[FunctionCallMetrics]               # 工具调用延迟

    def chronological_events(self) -> List[str]:
        """按时间排序输出人类可读的事件链"""
        events = []
        if self.user_turn_start_time and self.user_turn_secs:
            events.append((self.user_turn_start_time, f"User turn: {self.user_turn_secs:.3f}s"))
        for t in self.ttfb:
            events.append((t.start_time, f"{t.processor}: TTFB {t.duration_secs:.3f}s"))
        for fc in self.function_calls:
            events.append((fc.start_time, f"{fc.function_name}: {fc.duration_secs:.3f}s"))
        if self.text_aggregation:
            ta = self.text_aggregation
            events.append((ta.start_time, f"{ta.processor}: text aggregation {ta.duration_secs:.3f}s"))
        events.sort(key=lambda e: e[0])
        return [label for _, label in events]

class MySTTService(STTService):
    async def process_frame(self, frame, direction):
        # ❌ 业务逻辑和打点代码交织在一起
        start = time.time()
        logger.info(f"Processing {frame}")
        metrics.record("stt_process_start", start)

        result = await self._recognize(frame)   # 真正的业务逻辑

        elapsed = time.time() - start
        metrics.record("stt_latency", elapsed)
        logger.info(f"STT took {elapsed:.3f}s")
        await self.push_frame(result)

# STT 服务只关心业务
class MySTTService(STTService):
    async def process_frame(self, frame, direction):
        result = await self._recognize(frame)
        await self.push_frame(result)

# 性能监控独立为 Observer，零侵入
observer = StartupTimingObserver()

@observer.event_handler("on_startup_timing_report")
async def on_report(observer, report):
    for t in report.processor_timings:
        print(f"{t.processor_name}: {t.duration_secs:.3f}s")

task = PipelineTask(pipeline, observers=[observer])

# vad_analyzer.py:31-44
class VADState(Enum):
    QUIET = 1      # 安静
    STARTING = 2   # 可能开始说话（需要确认）
    SPEAKING = 3   # 确认正在说话
    STOPPING = 4   # 可能停止说话（需要确认）

# vad_analyzer.py:207-244
speaking = confidence >= self._params.confidence and volume >= self._params.min_volume

if speaking:
    match self._vad_state:
        case VADState.QUIET:
            self._vad_state = VADState.STARTING
            self._vad_starting_count = 1          # 开始计数
        case VADState.STARTING:
            self._vad_starting_count += 1         # 继续计数
        case VADState.STOPPING:
            self._vad_state = VADState.SPEAKING   # 取消停止，回到说话
            self._vad_stopping_count = 0

# 只有连续达到阈值帧数才确认状态转换
if self._vad_state == VADState.STARTING and self._vad_starting_count >= self._vad_start_frames:
    self._vad_state = VADState.SPEAKING   # 确认开始说话

# vad_analyzer.py:161-164
vad_frames_per_sec = self._vad_frames / self.sample_rate   # 一帧的时长
self._vad_start_frames = round(start_secs / vad_frames_per_sec)  # 需要多少帧
self._vad_stop_frames = round(stop_secs / vad_frames_per_sec)

# vad_analyzer.py
self._vad_buffer += buffer

num_required_bytes = self._vad_frames_num_bytes
if len(self._vad_buffer) < num_required_bytes:
    return self._vad_state

# vad_analyzer.py
confidence = self.voice_confidence(audio_frames)

volume = self._get_smoothed_volume(audio_frames)
self._prev_volume = volume

speaking = confidence >= self._params.confidence and volume >= self._params.min_volume

# vad_analyzer.py
if speaking:
    match self._vad_state:
        case VADState.QUIET:
            self._vad_state = VADState.STARTING
            self._vad_starting_count = 1
        case VADState.STARTING:
            self._vad_starting_count += 1
        case VADState.STOPPING:
            self._vad_state = VADState.SPEAKING
            self._vad_stopping_count = 0
else:
    match self._vad_state:
        case VADState.STARTING:
            self._vad_state = VADState.QUIET
            self._vad_starting_count = 0
        case VADState.SPEAKING:
            self._vad_state = VADState.STOPPING
            self._vad_stopping_count = 1
        case VADState.STOPPING:
            self._vad_stopping_count += 1

# vad_analyzer.py
def _get_smoothed_volume(self, audio: bytes) -> float:
    volume = calculate_audio_volume(audio, self.sample_rate)
    return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)

# vad_analyzer.py
async def analyze_audio(self, buffer: bytes) -> VADState:
    loop = asyncio.get_running_loop()
    state = await loop.run_in_executor(self._executor, self._run_analyzer, buffer)
    return state

# vad_controller.py:126-140
async def _handle_vad(self, audio: bytes, vad_state: VADState) -> VADState:
    new_vad_state = await self._vad_analyzer.analyze_audio(audio)
    # 只关心 SPEAKING 和 QUIET 的变化（忽略中间态 STARTING/STOPPING）
    if new_vad_state != vad_state \
       and new_vad_state != VADState.STARTING \
       and new_vad_state != VADState.STOPPING:
        if new_vad_state == VADState.SPEAKING:
            await self._call_event_handler("on_speech_started")
        elif new_vad_state == VADState.QUIET:
            await self._call_event_handler("on_speech_stopped")
    return new_vad_state

# vad_controller.py:84-88
self._register_event_handler("on_speech_started", sync=True)
self._register_event_handler("on_speech_stopped", sync=True)
self._register_event_handler("on_speech_activity", sync=True)

# vad_controller.py
if isinstance(frame, StartFrame):
    await self._start(frame)
elif isinstance(frame, InputAudioRawFrame):
    await self._handle_audio(frame)
elif isinstance(frame, VADParamsUpdateFrame):
    self._vad_analyzer.set_params(frame.params)
    await self.broadcast_frame(SpeechControlParamsFrame, vad_params=frame.params)

# vad_controller.py
if (
    new_vad_state != vad_state
    and new_vad_state != VADState.STARTING
    and new_vad_state != VADState.STOPPING
):
    if new_vad_state == VADState.SPEAKING:
        await self._call_event_handler("on_speech_started")
    elif new_vad_state == VADState.QUIET:
        await self._call_event_handler("on_speech_stopped")

# vad_controller.py
self._vad_state = await self._handle_vad(frame.audio, self._vad_state)

if self._vad_state == VADState.SPEAKING:
    await self._call_event_handler("on_speech_activity")

# vad_controller.py
async def _start(self, frame: StartFrame):
    self._vad_analyzer.set_sample_rate(frame.audio_in_sample_rate)
    await self.broadcast_frame(SpeechControlParamsFrame, vad_params=self._vad_analyzer.params)

# vad_analyzer.py:91,186-188
self._executor = ThreadPoolExecutor(max_workers=1)

async def analyze_audio(self, buffer: bytes) -> VADState:
    loop = asyncio.get_running_loop()
    state = await loop.run_in_executor(self._executor, self._run_analyzer, buffer)
    return state

# silero.py
def voice_confidence(self, buffer) -> float:
    audio_int16 = np.frombuffer(buffer, np.int16)
    audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0
    new_confidence = self._model(audio_float32, self.sample_rate)[0]
    ...
    return new_confidence

# silero.py
def num_frames_required(self) -> int:
    return 512 if self.sample_rate == 16000 else 256

# silero.py
if diff_time >= _MODEL_RESET_STATES_TIME:
    self._model.reset_states()
    self._last_reset_time = curr_time

# llm_response_universal.py
if self._vad_controller:
    await self._vad_controller.process_frame(frame)

# llm_response_universal.py
async def _on_vad_speech_started(self, controller):
    await self._queued_broadcast_frame(
        VADUserStartedSpeakingFrame,
        start_secs=controller._vad_analyzer.params.start_secs,
    )

async def _on_vad_speech_stopped(self, controller):
    await self._queued_broadcast_frame(
        VADUserStoppedSpeakingFrame,
        stop_secs=controller._vad_analyzer.params.stop_secs,
    )

async def _on_vad_speech_activity(self, controller):
    await self._queued_broadcast_frame(UserSpeakingFrame)

# llm_response_universal.py
async def _queued_broadcast_frame(self, frame_cls: Type[Frame], **kwargs):
    await self._internal_queue_frame(frame_cls(**kwargs))
    await self.push_frame(frame_cls(**kwargs), FrameDirection.UPSTREAM)

InputAudioRawFrame
    -> VADAnalyzer (confidence + volume + state machine)
    -> VADController (started/stopped/activity events)
    -> LLMUserAggregator (wrap into VAD frames)
    -> UserTurnController / strategies
    -> semantic user turn frames

# base_user_turn_start_strategy.py
class BaseUserTurnStartStrategy(BaseObject):
    def __init__(self, *, enable_interruptions=True, enable_user_speaking_frames=True):
        # enable_interruptions: 触发时是否中断当前 bot 输出
        # enable_user_speaking_frames: 是否广播 UserStartedSpeakingFrame

    async def process_frame(self, frame: Frame):
        """子类重写，判断是否触发 turn start"""
        pass

    async def trigger_user_turn_started(self):
        """触发 on_user_turn_started 事件，携带参数"""
        await self._call_event_handler("on_user_turn_started",
            UserTurnStartedParams(
                enable_interruptions=self._enable_interruptions,
                enable_user_speaking_frames=self._enable_user_speaking_frames,
            ))

# vad_user_turn_start_strategy.py — 完整实现仅 ~10 行
class VADUserTurnStartStrategy(BaseUserTurnStartStrategy):
    async def process_frame(self, frame: Frame):
        await super().process_frame(frame)
        if isinstance(frame, VADUserStartedSpeakingFrame):
            await self.trigger_user_turn_started()

# user_turn_strategies.py:46-50
class UserTurnStrategies:
    def __post_init__(self):
        if not self.start:
            self.start = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
        if not self.stop:
            self.stop = [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]

# turn_analyzer_user_turn_stop_strategy.py
self._turn_analyzer = turn_analyzer  # ML 模型
self._stt_timeout: float = 0.0       # STT P99 延迟（从 STTMetadataFrame 获取）
self._stop_secs: float = 0.0         # VAD stop_secs

self._text = ""                      # 累积的转录文本
self._turn_complete = False           # 模型是否判断 turn 完成
self._vad_user_speaking = False       # VAD 是否检测到语音
self._transcript_finalized = False    # 是否收到 finalized transcript

# turn_analyzer_user_turn_stop_strategy.py:205-227
async def _maybe_trigger_user_turn_stopped(self):
    # 必须有转录文本 AND 模型判断 turn 完成
    if not self._text or not self._turn_complete:
        return

    # finalized transcript → 立即触发
    if self._transcript_finalized:
        await self.trigger_user_turn_stopped()
        return

    # 非 finalized → 只在超时任务完成后触发
    if self._timeout_task is None:
        await self.trigger_user_turn_stopped()

# turn_analyzer_user_turn_stop_strategy.py:155-159
# VAD 的 stop_secs 已经过去了（那段时间是 VAD 在等待确认静音）
# 所以 STT 超时需要减去这段时间
timeout = max(0, self._stt_timeout - self._stop_secs)

# speech_timeout_user_turn_stop_strategy.py:145-158
def _calculate_timeout(self) -> float:
    effective_stt_wait = max(0, self._stt_timeout - self._stop_secs)
    if self._transcript_finalized:
        return self._user_speech_timeout          # 已有最终转录，只等用户恢复说话
    return max(effective_stt_wait, self._user_speech_timeout)  # 取大值

# audio/turn/base_turn_analyzer.py
class BaseTurnAnalyzer(ABC):
    @abstractmethod
    def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
        """逐帧添加音频。流式分析器可能在此返回 COMPLETE。"""

    @abstractmethod
    async def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
        """批量分析是否 turn 完成。在 VAD 停止时调用。"""

    def update_vad_start_secs(self, vad_start_secs: float):
        """同步 VAD 的 start 延迟，调整预缓冲大小。"""

# user_turn_controller.py
class UserTurnController(BaseObject):
    def __init__(self, *, user_turn_strategies, user_turn_stop_timeout=5.0):
        self._user_turn = False        # 当前是否在一个轮次中
        self._user_speaking = False    # 用户是否正在说话

# user_turn_controller.py:141-167
async def process_frame(self, frame: Frame):
    # 1. 先处理 VAD/User speaking 状态（重置超时）
    if isinstance(frame, UserStartedSpeakingFrame):
        self._user_speaking = True
        self._user_turn_stop_timeout_event.set()  # 重置超时
    elif isinstance(frame, VADUserStartedSpeakingFrame):
        self._user_speaking = True
        self._user_turn_stop_timeout_event.set()
    # ... 类似处理 stopped 和 transcription

    # 2. 分发给所有 start 策略
    for strategy in self._user_turn_strategies.start or []:
        await strategy.process_frame(frame)

    # 3. 分发给所有 stop 策略
    for strategy in self._user_turn_strategies.stop or []:
        await strategy.process_frame(frame)

# user_turn_controller.py:245-259
async def _trigger_user_turn_start(self, strategy, params):
    if self._user_turn:     # 已经在轮次中，忽略
        return
    self._user_turn = True
    # 重置所有 start 策略
    for s in self._user_turn_strategies.start or []:
        await s.reset()
    await self._call_event_handler("on_user_turn_started", strategy, params)

async def _trigger_user_turn_stop(self, strategy, params):
    if not self._user_turn:  # 不在轮次中，忽略
        return
    self._user_turn = False
    # 重置所有 stop 策略
    for s in self._user_turn_strategies.stop or []:
        await s.reset()
    await self._call_event_handler("on_user_turn_stopped", strategy, params)

# user_turn_controller.py:277-290
async def _user_turn_stop_timeout_task_handler(self):
    while True:
        try:
            await asyncio.wait_for(
                self._user_turn_stop_timeout_event.wait(),
                timeout=self._user_turn_stop_timeout,  # 默认 5.0s
            )
            self._user_turn_stop_timeout_event.clear()
        except asyncio.TimeoutError:
            # 5秒内没有任何活动 → 强制结束轮次
            if self._user_turn and not self._user_speaking:
                await self._trigger_user_turn_stop(None, UserTurnStoppedParams(...))

# user_turn_processor.py:115-147
async def process_frame(self, frame: Frame, direction: FrameDirection):
    await super().process_frame(frame, direction)

    # 1. 先推送帧给下游（StartFrame/EndFrame 必须先到达所有处理器）
    if isinstance(frame, StartFrame):
        await self.push_frame(frame, direction)
        await self._start(frame)       # 初始化 controller
    elif isinstance(frame, EndFrame):
        await self.push_frame(frame, direction)
        await self._stop(frame)
    else:
        await self.push_frame(frame, direction)

    # 2. 交给 controller 处理
    await self._user_turn_controller.process_frame(frame)
    # 3. 交给 idle controller 处理
    await self._user_idle_controller.process_frame(frame)

# user_turn_processor.py:171-187
async def _on_user_turn_started(self, controller, strategy, params):
    # 1. 广播 UserStartedSpeakingFrame（如果策略允许）
    if params.enable_user_speaking_frames:
        await self.broadcast_frame(UserStartedSpeakingFrame)

    # 2. 中断当前 bot 输出（如果策略允许且处理器允许）
    if params.enable_interruptions and self._allow_interruptions:
        await self.broadcast_interruption()

    # 3. 触发外部事件
    await self._call_event_handler("on_user_turn_started", strategy)

# user_idle_controller.py:95-136
async def process_frame(self, frame: Frame):
    if isinstance(frame, BotStoppedSpeakingFrame):
        # Bot 说完了 → 启动闲置计时器
        # 但要排除：用户正在说话（中断场景）、function call 进行中
        if not self._user_turn_in_progress and self._function_calls_in_progress == 0:
            await self._start_idle_timer()
    elif isinstance(frame, BotStartedSpeakingFrame):
        await self._cancel_idle_timer()        # Bot 又开始说了
    elif isinstance(frame, UserStartedSpeakingFrame):
        self._user_turn_in_progress = True
        await self._cancel_idle_timer()        # 用户开始说了
    elif isinstance(frame, FunctionCallsStartedFrame):
        self._function_calls_in_progress += len(frame.function_calls)
        await self._cancel_idle_timer()        # Function call 开始了

# silero.py:34-127 — SileroOnnxModel
class SileroOnnxModel:
    def __init__(self, path, force_onnx_cpu=True):
        opts = onnxruntime.SessionOptions()
        opts.inter_op_num_threads = 1    # 单线程推理
        opts.intra_op_num_threads = 1
        self.session = onnxruntime.InferenceSession(path, providers=["CPUExecutionProvider"])
        self.reset_states()
        self.sample_rates = [8000, 16000]  # 只支持这两种采样率

    def __call__(self, x, sr: int):
        """每次处理固定大小的帧"""
        num_samples = 512 if sr == 16000 else 256  # 32ms @ 16kHz 或 32ms @ 8kHz

        # 关键：模型维护内部状态（RNN 隐状态 + 上下文缓冲）
        context_size = 64 if sr == 16000 else 32
        x = np.concatenate((self._context, x), axis=1)  # 拼接上下文

        ort_inputs = {"input": x, "state": self._state, "sr": np.array(sr, dtype="int64")}
        ort_outs = self.session.run(None, ort_inputs)
        out, state = ort_outs
        self._state = state                             # 更新 RNN 隐状态
        self._context = x[..., -context_size:]          # 保留上下文尾部
        return out  # 置信度值 [0, 1]

# silero.py:216-219
_MODEL_RESET_STATES_TIME = 5.0

curr_time = time.time()
if diff_time >= _MODEL_RESET_STATES_TIME:
    self._model.reset_states()

# vad_analyzer.py:190-200
def _run_analyzer(self, buffer: bytes) -> VADState:
    self._vad_buffer += buffer   # 追加到内部缓冲

    num_required_bytes = self._vad_frames_num_bytes  # 512*2=1024 bytes @ 16kHz
    if len(self._vad_buffer) < num_required_bytes:
        return self._vad_state   # 不够一帧，返回当前状态

    # 可能积累了多帧（如果上游给了大 chunk），逐帧处理
    while len(self._vad_buffer) >= num_required_bytes:
        audio_frames = self._vad_buffer[:num_required_bytes]
        self._vad_buffer = self._vad_buffer[num_required_bytes:]
        confidence = self.voice_confidence(audio_frames)  # 调用模型
        volume = self._get_smoothed_volume(audio_frames)
        # ... 状态机转换逻辑 ...

# vad_analyzer.py:207
speaking = confidence >= self._params.confidence and volume >= self._params.min_volume

# stt_service.py:330-383 — process_frame()
async def process_frame(self, frame: Frame, direction: FrameDirection):
    if isinstance(frame, StartFrame):
        await self.push_frame(frame, direction)
        await self._push_stt_metadata()          # ★ 广播 STTMetadataFrame
    elif isinstance(frame, AudioRawFrame):
        await self.process_audio_frame(frame, direction)  # 送给 run_stt()
        if self._audio_passthrough:
            await self.push_frame(frame, direction)       # 透传音频给下游
    elif isinstance(frame, VADUserStartedSpeakingFrame):
        await self._handle_vad_user_started_speaking(frame)
        await self.push_frame(frame, direction)
    elif isinstance(frame, VADUserStoppedSpeakingFrame):
        await self._handle_vad_user_stopped_speaking(frame)
        await self.push_frame(frame, direction)

# frames.py
@dataclass
class InterimTranscriptionFrame(TextFrame):
    """部分/中间转录结果，随着用户说话持续更新"""
    text: str
    user_id: str
    timestamp: str

@dataclass
class TranscriptionFrame(TextFrame):
    """最终转录结果"""
    user_id: str
    timestamp: str
    finalized: bool = False  # ★ 核心字段

# stt_service.py:158-181
def request_finalize(self):
    """标记已发送 finalize 请求"""
    self._finalize_requested = True

def confirm_finalize(self):
    """服务端确认 finalize"""
    if self._finalize_requested:
        self._finalize_pending = True
        self._finalize_requested = False

# 在 push_frame 中自动标记
async def push_frame(self, frame: Frame, direction=...):
    if isinstance(frame, TranscriptionFrame):
        if self._finalize_pending:
            frame.finalized = True       # ★ 自动标记
            self._finalize_pending = False

# stt_service.py:597-609 — SegmentedSTTService.push_frame()
async def push_frame(self, frame, direction=...):
    if isinstance(frame, TranscriptionFrame):
        frame.finalized = True  # ★ 分段 STT 的每个转录都是 finalized
    await super().push_frame(frame, direction)

# stt_service.py:414-420
async def _push_stt_metadata(self):
    ttfs = self._ttfs_p99_latency
    if ttfs is None:
        ttfs = DEFAULT_TTFS_P99      # 1.0 秒（保守默认值）
    await self.broadcast_frame(STTMetadataFrame, service_name=self.name, ttfs_p99_latency=ttfs)

# turn_analyzer_user_turn_stop_strategy.py:99
if isinstance(frame, STTMetadataFrame):
    self._stt_timeout = frame.ttfs_p99_latency

# stt_service.py:99-104 — 注释说明
# STT "TTFB" differs from traditional TTFB (which measures from a discrete
# request to first response byte). Since STT receives continuous audio, we measure
# from when the user stops speaking to when the final transcript arrives—capturing
# the latency that matters for voice AI applications.

# base_smart_turn.py:101-147
def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
    # 转换为 float32 并追加到带时间戳的缓冲区
    audio_float32 = np.frombuffer(buffer, dtype=np.int16).astype(np.float32) / 32768.0
    self._audio_buffer.append((time.time(), audio_float32))

    if is_speech:
        self._silence_ms = 0
        self._speech_triggered = True
        if self._speech_start_time == 0:
            self._speech_start_time = time.time()
    else:
        if self._speech_triggered:
            chunk_duration_ms = len(audio_int16) / (self._sample_rate / 1000)
            self._silence_ms += chunk_duration_ms
            # ★ 静音超过阈值（默认 3 秒）→ 直接判定为 COMPLETE
            if self._silence_ms >= self._stop_ms:
                state = EndOfTurnState.COMPLETE
                self._clear(state)
        else:
            # 语音开始前，限制缓冲区大小防止无限增长
            max_buffer_time = pre_speech_ms/1000 + stop_secs + max_duration_secs
            while self._audio_buffer[0][0] < time.time() - max_buffer_time:
                self._audio_buffer.pop(0)

# base_smart_turn.py:181-248
def _process_speech_segment(self, audio_buffer):
    # 1. 计算有效的预语音缓冲时间（加上 VAD 启动延迟）
    effective_pre_speech_ms = self._params.pre_speech_ms + (self._vad_start_secs * 1000)
    #                          500ms（默认）        +    200ms（VAD start_secs）= 700ms

    # 2. 从 speech_start_time 往前取 effective_pre_speech_ms 的音频
    start_time = self._speech_start_time - (effective_pre_speech_ms / 1000)
    # 在缓冲区中找到对应位置...

    # 3. 拼接音频段
    segment_audio = np.concatenate(segment_audio_chunks)

    # 4. 限制最大时长（默认 8 秒，取尾部）
    max_samples = int(self._params.max_duration_secs * self.sample_rate)
    if len(segment_audio) > max_samples:
        segment_audio = segment_audio[-max_samples:]  # ★ 保留最后 8 秒

    # 5. 调用子类的 ML 模型
    result = self._predict_endpoint(segment_audio)
    state = EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE

# local_smart_turn_v3.py:28-81
class LocalSmartTurnAnalyzerV3(BaseSmartTurn):
    def __init__(self, *, smart_turn_model_path=None, cpu_count=1, **kwargs):
        # 内置模型文件
        model_name = "smart-turn-v3.2-cpu.onnx"

        # ONNX 推理配置
        so = ort.SessionOptions()
        so.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL
        so.inter_op_num_threads = 1
        so.intra_op_num_threads = cpu_count      # 可配置 CPU 核数
        so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL

        # ★ Whisper 特征提取器（8 秒窗口）
        self._feature_extractor = WhisperFeatureExtractor(chunk_length=8)
        self._session = ort.InferenceSession(smart_turn_model_path, sess_options=so)

# local_smart_turn_v3.py:141-197
def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
    # 1. 重采样到 16kHz（模型固定输入）
    audio_array = self._resample_to_model_rate(audio_array)  # 使用 soxr VHQ

    # 2. 截断或填充到 8 秒
    audio_array = truncate_audio_to_last_n_seconds(audio_array, n_seconds=8)
    #   > 8s → 取最后 8 秒
    #   < 8s → 在开头补零

    # 3. Whisper 特征提取（Log-Mel Spectrogram）
    inputs = self._feature_extractor(
        audio_array,
        sampling_rate=16000,
        return_tensors="np",
        padding="max_length",
        max_length=8 * 16000,     # 128000 个采样点
        truncation=True,
        do_normalize=True,
    )

    # 4. 准备 ONNX 输入 (Log-Mel 特征图)
    input_features = inputs.input_features.squeeze(0).astype(np.float32)
    input_features = np.expand_dims(input_features, axis=0)  # [1, n_mels, time_steps]

    # 5. ONNX 推理
    outputs = self._session.run(None, {"input_features": input_features})

    # 6. 解读结果（sigmoid 概率）
    probability = outputs[0][0].item()       # [0, 1]
    prediction = 1 if probability > 0.5 else 0  # 二分类阈值

    return {"prediction": prediction, "probability": probability}

# base_smart_turn.py:156-158
loop = asyncio.get_running_loop()
state, result = await loop.run_in_executor(
    self._executor, self._process_speech_segment, self._audio_buffer
)

# krisp_viva_turn.py:243-343
def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
    # 将音频分成固定帧（默认 20ms）
    frames = audio_float32.reshape(-1, self._samples_per_frame)

    for frame in frames:
        if is_speech:
            self._speech_triggered = True
            self._speech_stopped_time = None
        else:
            if self._speech_triggered and self._speech_stopped_time is None:
                self._speech_stopped_time = time.perf_counter()

        # ★ 逐帧调用 Krisp SDK 处理
        prob = self._tt_session.process(frame.tolist())

        if prob < 0:
            continue  # 模型还不够数据（需要约 100ms 预热）

        self._last_probability = prob

        # ★ 实时判断：语音已触发 && 概率超阈值 → COMPLETE
        if self._speech_triggered and prob >= self._params.threshold:
            # 计算 e2e 处理时间
            if self._speech_stopped_time is not None:
                self._e2e_processing_time_ms = (time.perf_counter() - self._speech_stopped_time) * 1000
            state = EndOfTurnState.COMPLETE
            self.clear()
            break

    return state

# krisp_viva_turn.py:351-362
async def analyze_end_of_turn(self):
    # 流式分析器已经在 append_audio 中做了判断
    # 这里只是返回最后状态和消费 metrics
    metrics = self._last_metrics
    self._last_metrics = None
    return self._last_state, metrics

# turn_analyzer_user_turn_stop_strategy.py:114-127
async def _handle_input_audio(self, frame: InputAudioRawFrame):
    state = self._turn_analyzer.append_audio(frame.audio, self._vad_user_speaking)

    # ★ 流式分析器（Krisp）在这里返回 COMPLETE
    if state == EndOfTurnState.COMPLETE:
        _, prediction = await self._turn_analyzer.analyze_end_of_turn()
        await self._handle_prediction_result(prediction)
        self._turn_complete = True
        await self._maybe_trigger_user_turn_stopped()

# ...

async def _handle_vad_user_stopped_speaking(self, frame):
    self._vad_user_speaking = False
    # ★ 批量分析器（SmartTurnV3）在这里才做实际分析
    state, prediction = await self._turn_analyzer.analyze_end_of_turn()
    self._turn_complete = state == EndOfTurnState.COMPLETE
    # 启动 STT 超时等待转录...

# http_smart_turn.py:107-123
def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
    # ★ 这个方法在 ThreadPoolExecutor 中运行（从 BaseSmartTurn.analyze_end_of_turn 调用）
    serialized_array = self._serialize_array(audio_array)  # np.save 到 bytes

    # 跨线程调用异步 HTTP 请求
    loop = asyncio.get_running_loop()
    future = asyncio.run_coroutine_threadsafe(
        self._send_raw_request(serialized_array), loop
    )
    return future.result()  # 阻塞等待结果

# http_smart_turn.py:67
timeout = aiohttp.ClientTimeout(total=self._params.stop_secs)  # 默认 3 秒

# user_turn_strategies.py:50
self.stop = [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]

user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        vad_analyzer=SileroVADAnalyzer(),
        user_turn_strategies=UserTurnStrategies(...),
    ),
)

# src/pipecat/turns/user_turn_strategies.py
start = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
stop = [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]