"""管线核心:给定一个 transport + 配置,跑完整的语音闭环。 关键设计:**transport 由调用方传入**,管线本身不关心是 WebRTC 还是 WS。 这就是"同时支持多种输出"的落点——加输出方式不用动这里。 对应 dograh 的 pipeline_builder.py + run_pipeline.py(已砍掉 workflow 引擎/DB/录音/指标)。 """ from uuid import uuid4 from loguru import logger from models import AssistantConfig from services.pipecat.service_factory import create_realtime_service, create_services from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( EndFrame, InputTransportMessageFrame, InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMTextFrame, LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, TextFrame, TTSSpeakFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( LLMAssistantAggregator, LLMUserAggregator, LLMUserAggregatorParams, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.turns.user_start import ( TranscriptionUserTurnStartStrategy, VADUserTurnStartStrategy, ) from pipecat.turns.user_turn_strategies import UserTurnStrategies from pipecat.utils.time import time_now_iso8601 from pipecat.workers.runner import WorkerRunner def _text_input(message) -> tuple[str, bool] | None: """解析现有 user-text 与 RTVI send-text 两种前端文字消息。""" if not isinstance(message, dict): return None if message.get("type") == "user-text": text = str(message.get("text") or "").strip() return (text, True) if text else None if message.get("type") == "send-text": data = message.get("data") if not isinstance(data, dict): return None text = str(data.get("content") or "").strip() options = data.get("options") run_immediately = not isinstance(options, dict) or options.get( "run_immediately", True ) return (text, bool(run_immediately)) if text else None return None class TextInputProcessor(FrameProcessor): """把 transport 文字消息转换成 LLM 可消费的帧。 run_immediately(默认/打断):先通过 on_text_input 事件把用户文字交给 run_pipeline 登记,再用 broadcast_interruption() 打断当前播报。新的 LLM 回复由 assistant aggregator 确认处理完 interruption 后触发。 run_immediately=False(RTVI send-text 静默追加):仅把文字写进上下文, 不打断、不触发推理。 """ def __init__(self): super().__init__() # 立即触发的文字(含打断语义)走 on_text_input;静默追加另走一条事件 self._register_event_handler("on_text_input") self._register_event_handler("on_text_append") self._register_event_handler("on_client_ready") async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if not isinstance(frame, InputTransportMessageFrame): await self.push_frame(frame, direction) return if isinstance(frame.message, dict) and frame.message.get("type") == "client-ready": await self._call_event_handler("on_client_ready") return parsed = _text_input(frame.message) if not parsed: await self.push_frame(frame, direction) return text, run_immediately = parsed if run_immediately: # 先登记文字再打断。下一轮 LLM 由 assistant aggregator 在真正处理完 # InterruptionFrame 后触发,避免新回复被这次 interruption 一起取消。 await self._call_event_handler("on_text_input", text) await self.broadcast_interruption() else: await self._call_event_handler("on_text_append", text) class RealtimeTextInputProcessor(FrameProcessor): """Route text input directly to a realtime service without cascade semantics.""" def __init__(self): super().__init__() self._register_event_handler("on_text_input") self._register_event_handler("on_text_append") async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if not isinstance(frame, InputTransportMessageFrame): await self.push_frame(frame, direction) return parsed = _text_input(frame.message) if not parsed: await self.push_frame(frame, direction) return text, run_immediately = parsed await self._call_event_handler( "on_text_input" if run_immediately else "on_text_append", text, ) class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): """聚合 LLM 回复进上下文,同时继续把回复帧交给下游 TTS。""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._register_event_handler("on_interruption_processed") self._register_event_handler("on_assistant_text_start") self._register_event_handler("on_assistant_text_delta") self._register_event_handler("on_assistant_text_end") self._stream_turn_id: str | None = None self._stream_timestamp = "" self._stream_text = "" async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, LLMFullResponseStartFrame): self._stream_turn_id = uuid4().hex self._stream_timestamp = time_now_iso8601() self._stream_text = "" await self._call_event_handler( "on_assistant_text_start", self._stream_turn_id, self._stream_timestamp, ) elif isinstance(frame, LLMTextFrame) and self._stream_turn_id: self._stream_text += frame.text await self._call_event_handler( "on_assistant_text_delta", self._stream_turn_id, frame.text, ) elif isinstance(frame, LLMFullResponseEndFrame): await self._finish_text_stream(interrupted=False) # LLMAssistantAggregator 默认会消费这些帧。放在 TTS 前用于中断时保存 # 已生成前缀时,必须显式透传,否则 TTS 收不到任何 LLM 回复。 if isinstance( frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame, TextFrame), ): await self.push_frame(frame, direction) elif isinstance(frame, InterruptionFrame): await self._finish_text_stream(interrupted=True) await self._call_event_handler("on_interruption_processed") async def _finish_text_stream(self, *, interrupted: bool): if not self._stream_turn_id: return await self._call_event_handler( "on_assistant_text_end", self._stream_turn_id, self._stream_text, interrupted, ) self._stream_turn_id = None self._stream_timestamp = "" self._stream_text = "" async def run_pipeline(transport, cfg: AssistantConfig) -> None: """在给定 transport 上构建并运行管线,直到连接结束。 Args: transport: 任意 pipecat transport(WebRTC / WS / 电话…), 只要有 .input() / .output() / event_handler 即可。 cfg: 助手配置(随请求内联传入)。 """ logger.info(f"启动管线: assistant={cfg.name} mode={cfg.runtimeMode}") if cfg.runtimeMode == "realtime": await run_realtime_pipeline(transport, cfg) return stt, llm, tts = create_services(cfg) context = LLMContext(messages=[{"role": "system", "content": cfg.prompt}]) user_aggregator = LLMUserAggregator( context, params=LLMUserAggregatorParams( vad_analyzer=SileroVADAnalyzer(), user_turn_strategies=UserTurnStrategies( start=[ VADUserTurnStartStrategy(enable_interruptions=cfg.enableInterrupt), TranscriptionUserTurnStartStrategy( enable_interruptions=cfg.enableInterrupt ), ] ), ), ) assistant_aggregator = PassthroughLLMAssistantAggregator(context) text_input = TextInputProcessor() pipeline = Pipeline( [ transport.input(), text_input, stt, user_aggregator, llm, # Aggregate the streamed LLM text before TTS. On interruption, # Pipecat commits the generated prefix immediately instead of # waiting for a TTS provider to emit spoken-text/timestamp frames. assistant_aggregator, tts, transport.output(), ] ) worker = PipelineWorker( pipeline, params=PipelineParams( enable_metrics=False, ), enable_rtvi=False, ) async def queue_transcript(role: str, content: str, timestamp: str) -> None: if content: await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "transcript", "role": role, "content": content, "timestamp": timestamp, }, ) ) greeting_transcript_sent = False pending_text_inputs: list[str] = [] async def append_user_text_to_context(text: str, *, run_llm: bool) -> None: await worker.queue_frame( LLMMessagesAppendFrame( messages=[{"role": "user", "content": text}], run_llm=run_llm, ) ) @user_aggregator.event_handler("on_user_turn_stopped") async def on_user_turn_stopped(_aggregator, _strategy, message): await queue_transcript("user", message.content, message.timestamp) @assistant_aggregator.event_handler("on_assistant_text_start") async def on_assistant_text_start(_aggregator, turn_id, timestamp): await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "assistant-text-start", "turn_id": turn_id, "timestamp": timestamp, } ) ) @assistant_aggregator.event_handler("on_assistant_text_delta") async def on_assistant_text_delta(_aggregator, turn_id, delta): await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "assistant-text-delta", "turn_id": turn_id, "delta": delta, } ) ) @assistant_aggregator.event_handler("on_assistant_text_end") async def on_assistant_text_end(_aggregator, turn_id, content, interrupted): await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "assistant-text-end", "turn_id": turn_id, "content": content, "interrupted": interrupted, } ) ) @text_input.event_handler("on_text_input") async def on_text_input(_processor, text): pending_text_inputs.append(text) # 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。 await queue_transcript("user", text, time_now_iso8601()) @assistant_aggregator.event_handler("on_interruption_processed") async def on_interruption_processed(_aggregator): if not pending_text_inputs: return text = pending_text_inputs.pop(0) # assistant aggregator 已处理完 interruption,现在再启动下一轮 LLM。 await append_user_text_to_context(text, run_llm=True) @text_input.event_handler("on_text_append") async def on_text_append(_processor, text): # 静默追加:写进上下文但不打断、不触发推理;transcript 照常上报 await queue_transcript("user", text, time_now_iso8601()) await append_user_text_to_context(text, run_llm=False) @text_input.event_handler("on_client_ready") async def on_client_ready(_processor): nonlocal greeting_transcript_sent if cfg.greeting and not greeting_transcript_sent: greeting_transcript_sent = True await queue_transcript("assistant", cfg.greeting, time_now_iso8601()) @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): if cfg.greeting: context.add_message({"role": "assistant", "content": cfg.greeting}) await worker.queue_frame(TTSSpeakFrame(cfg.greeting, append_to_context=False)) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(_transport, _client): logger.info("对端断开,结束管线") await worker.queue_frame(EndFrame()) runner = WorkerRunner(handle_sigint=False) await runner.add_workers(worker) await runner.run() logger.info("管线已结束") async def run_realtime_pipeline(transport, cfg: AssistantConfig) -> None: """Run a speech-to-speech model that owns ASR, reasoning, and synthesis.""" realtime = create_realtime_service(cfg) text_input = RealtimeTextInputProcessor() pipeline = Pipeline( [ transport.input(), text_input, realtime, transport.output(), ] ) worker = PipelineWorker( pipeline, params=PipelineParams( enable_metrics=False, audio_in_sample_rate=int( cfg.realtime_values.get("inputSampleRate") or 24000 ), audio_out_sample_rate=int( cfg.realtime_values.get("outputSampleRate") or 24000 ), ), enable_rtvi=False, ) async def queue_transcript(role: str, content: str) -> None: if content: await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ "type": "transcript", "role": role, "content": content, "timestamp": time_now_iso8601(), }, ) ) @text_input.event_handler("on_text_input") async def on_text_input(_processor, text): await queue_transcript("user", text) await realtime.interrupt() await realtime.send_text(text, run_immediately=True) @text_input.event_handler("on_text_append") async def on_text_append(_processor, text): await queue_transcript("user", text) await realtime.send_text(text, run_immediately=False) @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): if cfg.greeting: await realtime.speak(cfg.greeting) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(_transport, _client): logger.info("Realtime 对端断开,结束管线") await worker.queue_frame(EndFrame()) runner = WorkerRunner(handle_sigint=False) await runner.add_workers(worker) await runner.run() logger.info("Realtime 管线已结束")