diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 4fac104..c095b7a 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -8,10 +8,8 @@ import asyncio import base64 -from collections.abc import Callable from io import BytesIO from typing import Any -from uuid import uuid4 from loguru import logger from models import AssistantConfig @@ -25,7 +23,6 @@ from services.pipecat.call_lifecycle import ( ) from services.pipecat.service_factory import ( config_with_resource, - create_llm, create_realtime_service, create_stt, create_tts, @@ -38,37 +35,19 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.flows import FlowsFunctionSchema from pipecat.frames.frames import ( EndFrame, - InputTransportMessageFrame, - InterruptionFrame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMContextFrame, - LLMTextFrame, - ManuallySwitchServiceFrame, - LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, - TextFrame, - TTSSpeakFrame, UserImageRawFrame, UserImageRequestFrame, VADParamsUpdateFrame, ) from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.llm_switcher import LLMSwitcher -from pipecat.pipeline.service_switcher import ServiceSwitcher from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( - LLMAssistantAggregator, LLMUserAggregatorParams, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.runner.utils import ( - get_transport_client_id, - maybe_capture_participant_camera, -) from pipecat.services.llm_service import FunctionCallParams -from pipecat.turns.user_mute.base_user_mute_strategy import BaseUserMuteStrategy from pipecat.turns.user_mute.function_call_user_mute_strategy import ( FunctionCallUserMuteStrategy, ) @@ -78,7 +57,28 @@ from services.pipecat.turn_config import ( create_vad_analyzer, create_vad_params, ) -from pipecat.utils.time import time_now_iso8601 +from services.pipecat.processors import ( + KNOWLEDGE_CONTEXT_MARKER, + CallEndingUserMuteStrategy, + ConversationHistoryProcessor, + KnowledgeRetrievalProcessor, + PassthroughLLMAssistantAggregator, + RealtimeDynamicVariableProcessor, + RealtimeTextInputProcessor, + TextInputProcessor, + UserTurnRoutingProcessor, + VisionCaptureProcessor, + WorkflowAggregatorPair, +) +from services.pipecat.workflow_services import ( + WorkflowServiceController, + build_workflow_llm_switcher, + build_workflow_voice_switcher, +) +from services.pipecat.pipeline_events import ( + bind_cascade_pipeline_events, + bind_realtime_pipeline_events, +) from pipecat.workers.runner import WorkerRunner @@ -101,9 +101,6 @@ ON_DEMAND_KNOWLEDGE_SYSTEM_HINT = ( "先调用 search_knowledge_base 检索；回答资料事实时只根据检索内容，" "资料不足要明确说明。" ) -KNOWLEDGE_CONTEXT_MARKER = "" - - def _compact_knowledge_metadata(value: str, max_length: int) -> str: """Keep tool metadata useful without letting it dominate the model context.""" compact = " ".join(value.split()) @@ -193,473 +190,6 @@ async def _analyze_image_with_vision_model( return str(content or "").strip() -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, should_ignore_input: Callable[[], bool] | None = None): - super().__init__() - self._should_ignore_input = should_ignore_input or (lambda: False) - # 立即触发的文字(含打断语义)走 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 - - if self._should_ignore_input(): - logger.debug("通话正在结束,忽略后续文字输入") - 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 CallEndingUserMuteStrategy(BaseUserMuteStrategy): - """Keep user media muted after an end-call tool starts terminating a call.""" - - def __init__(self, is_call_ending: Callable[[], bool]): - super().__init__() - self._is_call_ending = is_call_ending - - async def process_frame(self, frame) -> bool: - await super().process_frame(frame) - return self._is_call_ending() - - -class VisionCaptureProcessor(FrameProcessor): - """Capture one requested video frame for auxiliary vision-model analysis.""" - - def __init__(self, timeout_s: float = 3.0): - super().__init__() - self._timeout_s = timeout_s - self._pending: dict[str, asyncio.Future[UserImageRawFrame]] = {} - - async def request_image( - self, - requester: FrameProcessor, - request: UserImageRequestFrame, - ) -> UserImageRawFrame: - key = request.tool_call_id or str(uuid4()) - request.tool_call_id = key - request.append_to_context = False - request.result_callback = None - - loop = asyncio.get_running_loop() - future: asyncio.Future[UserImageRawFrame] = loop.create_future() - self._pending[key] = future - await requester.push_frame(request, FrameDirection.UPSTREAM) - try: - return await asyncio.wait_for(future, timeout=self._timeout_s) - finally: - self._pending.pop(key, None) - - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if ( - isinstance(frame, UserImageRawFrame) - and frame.request - and frame.request.tool_call_id - and frame.request.tool_call_id in self._pending - ): - future = self._pending[frame.request.tool_call_id] - if not future.done(): - future.set_result(frame) - return - - await self.push_frame(frame, direction) - - -class RealtimeDynamicVariableProcessor(FrameProcessor): - """Keep realtime system turn/history variables current between responses.""" - - def __init__(self, brain: Brain, cfg: AssistantConfig, realtime): - super().__init__() - self._brain = brain - self._cfg = cfg - self._realtime = realtime - - async def _refresh_instructions(self) -> None: - update = getattr(self._realtime, "update_instructions", None) - if callable(update): - await update(self._brain.system_prompt(self._cfg)) - - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if isinstance(frame, OutputTransportMessageUrgentFrame): - message = frame.message - if isinstance(message, dict): - event_type = message.get("type") - if event_type == "transcript" and message.get("role") == "user": - content = str(message.get("content") or "").strip() - if content: - self._brain.record_user_message(content) - await self._refresh_instructions() - elif event_type == "assistant-text-end": - await self._brain.on_assistant_text_end( - str(message.get("turn_id") or ""), - str(message.get("content") or ""), - bool(message.get("interrupted", False)), - ) - await self._refresh_instructions() - await self.push_frame(frame, direction) - - -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 ConversationHistoryProcessor(FrameProcessor): - """从最终客户端事件旁路保存历史，不改变 Pipecat 的上下文与帧语义。""" - - def __init__(self, recorder: ConversationRecorder | None): - super().__init__() - self._recorder = recorder - - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - await self.push_frame(frame, direction) - if self._recorder and isinstance(frame, OutputTransportMessageUrgentFrame): - await self._recorder.record_transport_message(frame.message) - - -class KnowledgeRetrievalProcessor(FrameProcessor): - """Retrieve before local LLM inference without changing Pipecat internals.""" - - def __init__( - self, - knowledge_base_id: str | None, - top_n: int = 5, - score_threshold: float = 0.0, - ): - super().__init__() - self._knowledge_base_id = knowledge_base_id - self._top_n = top_n - self._score_threshold = score_threshold - self._mode = "automatic" if knowledge_base_id else "disabled" - self._last_signature = "" - - def set_scope(self, scope: dict) -> None: - self._knowledge_base_id = scope.get("knowledge_base_id") or None - self._mode = str(scope.get("mode") or "disabled") - self._top_n = int(scope.get("top_n") or 5) - self._score_threshold = float(scope.get("score_threshold") or 0.0) - self._last_signature = "" - - def _clear_context(self, messages: list[dict]) -> None: - # Remove the legacy Workflow knowledge message so an in-flight context - # created before this compatibility fix cannot keep sending that role. - messages[:] = [ - message - for message in messages - if not ( - message.get("role") == "developer" - and KNOWLEDGE_CONTEXT_MARKER in str(message.get("content") or "") - ) - ] - system_message = next( - (message for message in messages if message.get("role") == "system"), - None, - ) - if system_message is not None: - content = str(system_message.get("content") or "") - system_message["content"] = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip() - - def _set_context(self, messages: list[dict], block: str) -> None: - """Store retrieved knowledge in a provider-compatible system message.""" - self._clear_context(messages) - system_message = next( - (message for message in messages if message.get("role") == "system"), - None, - ) - if system_message is None: - messages.insert(0, {"role": "system", "content": block}) - return - content = str(system_message.get("content") or "").rstrip() - system_message["content"] = f"{content}\n\n{block}" if content else block - - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if not isinstance(frame, LLMContextFrame): - await self.push_frame(frame, direction) - return - - messages = frame.context.get_messages() - if self._mode != "automatic" or not self._knowledge_base_id: - self._clear_context(messages) - await self.push_frame(frame, direction) - return - user_messages = [message for message in messages if message.get("role") == "user"] - if not user_messages: - await self.push_frame(frame, direction) - return - query = str(user_messages[-1].get("content") or "").strip() - signature = f"{len(user_messages)}:{query}" - if not query or signature == self._last_signature: - await self.push_frame(frame, direction) - return - self._last_signature = signature - - try: - async with SessionLocal() as session: - results = await search_knowledge( - session, - self._knowledge_base_id, - query, - top_k=self._top_n, - score_threshold=self._score_threshold, - ) - except Exception as exc: - logger.warning(f"自动知识库检索失败: {exc}") - results = [] - - sources = "\n\n".join( - f"[{index + 1}] 来源：{item['document']}（相关度 {item['score']}）\n{item['content']}" - for index, item in enumerate(results) - ) or "未检索到相关资料。" - block = f"{KNOWLEDGE_CONTEXT_MARKER}\n当前问题的知识库检索结果：\n{sources}" - self._set_context(messages, block) - await self.push_frame(frame, direction) - - -class UserTurnRoutingProcessor(FrameProcessor): - """Give a brain first right of refusal before a new user turn reaches the LLM.""" - - def __init__(self, brain: Brain): - super().__init__() - self._brain = brain - self._last_user_message: dict | None = None - - async def process_frame(self, frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if direction != FrameDirection.DOWNSTREAM or not isinstance( - frame, LLMContextFrame - ): - await self.push_frame(frame, direction) - return - - user_message = next( - ( - message - for message in reversed(frame.context.get_messages()) - if message.get("role") == "user" - and isinstance(message.get("content"), str) - and str(message.get("content") or "").strip() - ), - None, - ) - if user_message is None: - await self.push_frame(frame, direction) - return - - if user_message is self._last_user_message: - # Programmatic LLMRunFrame after a node transition reuses the same - # user message. It is a response run, not another routing event. - await self.push_frame(frame, direction) - return - self._last_user_message = user_message - - content = str(user_message.get("content") or "").strip() - handled = await self._brain.on_user_turn_end(content) - if not handled: - await self.push_frame(frame, direction) - - -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 = "" - - -class WorkflowAggregatorPair: - """Small public-shape adapter required by Pipecat FlowManager.""" - - def __init__(self, user_aggregator, assistant_aggregator): - self._user = user_aggregator - self._assistant = assistant_aggregator - - def user(self): - return self._user - - def assistant(self): - return self._assistant - - -def _workflow_service_switcher( - cfg: AssistantConfig, capability: str, base_service: FrameProcessor -): - """Build one switcher and an ID lookup for every referenced voice resource.""" - create = create_stt if capability == "ASR" else create_tts - settings = cfg.graph.get("settings") or {} - default_key = ( - "defaultAsrResourceId" if capability == "ASR" else "defaultTtsResourceId" - ) - default_id = str(settings.get(default_key) or "") - services_by_id = {} - for resource_id, resource in cfg.workflow_model_resources.items(): - if resource.capability != capability: - continue - services_by_id[resource_id] = ( - base_service - if resource_id == default_id - else create(config_with_resource(cfg, resource)) - ) - primary = services_by_id.get(default_id, base_service) - services = [primary] - services.extend( - service for service in services_by_id.values() if service is not primary - ) - if base_service is not primary: - services.append(base_service) - return ServiceSwitcher(services=services), services_by_id, primary - - -def _workflow_llm_switcher(cfg: AssistantConfig, base_service): - """Build an LLM switcher for the global model and Agent overrides.""" - settings = cfg.graph.get("settings") or {} - default_id = str(settings.get("defaultLlmResourceId") or "") - services_by_id = {} - for resource_id, resource in cfg.workflow_model_resources.items(): - if resource.capability != "LLM": - continue - services_by_id[resource_id] = ( - base_service - if resource_id == default_id - else create_llm(config_with_resource(cfg, resource)) - ) - primary = services_by_id.get(default_id, base_service) - services = [primary] - services.extend( - service for service in services_by_id.values() if service is not primary - ) - if base_service is not primary: - services.append(base_service) - return LLMSwitcher(llms=services), services_by_id, primary - - async def run_pipeline( transport, cfg: AssistantConfig, @@ -727,10 +257,10 @@ async def run_pipeline( current_voice_services: dict[str, FrameProcessor] = {"asr": stt, "tts": tts} if cfg.type == "workflow": stt_processor, stt_services, current_voice_services["asr"] = ( - _workflow_service_switcher(cfg, "ASR", stt) + build_workflow_voice_switcher(cfg, "ASR", stt) ) tts_processor, tts_services, current_voice_services["tts"] = ( - _workflow_service_switcher(cfg, "TTS", tts) + build_workflow_voice_switcher(cfg, "TTS", tts) ) greeting = await brain.greeting(cfg) @@ -796,7 +326,7 @@ async def run_pipeline( llm_services: dict[str, FrameProcessor] = {} current_llm_service = llm if cfg.type == "workflow": - llm, llm_services, current_llm_service = _workflow_llm_switcher(cfg, llm) + llm, llm_services, current_llm_service = build_workflow_llm_switcher(cfg, llm) user_aggregator = ConfigurableLLMUserAggregator( context, params=LLMUserAggregatorParams( @@ -1020,52 +550,15 @@ async def run_pipeline( enable_rtvi=False, ) worker_holder["worker"] = worker - default_workflow_services = { - "llm": current_llm_service, - **current_voice_services, - } - - async def switch_workflow_services( - llm_resource_id: str | None, - asr_resource_id: str | None, - tts_resource_id: str | None, - ) -> None: - nonlocal current_llm_service - requested = ( - ("llm", llm_services, llm_resource_id), - ("asr", stt_services, asr_resource_id), - ("tts", tts_services, tts_resource_id), - ) - for kind, services, resource_id in requested: - target = ( - services.get(resource_id) - if resource_id - else default_workflow_services[kind] - ) - if target is None: - raise ValueError(f"Workflow {kind.upper()} 资源未加载:{resource_id}") - current = ( - current_llm_service - if kind == "llm" - else current_voice_services[kind] - ) - if current is target: - continue - await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) - if kind == "llm": - current_llm_service = target - else: - current_voice_services[kind] = target - await worker.queue_frame( - OutputTransportMessageUrgentFrame( - message={ - "type": "service-switched", - "capability": kind.upper(), - "resourceId": resource_id, - } - ) - ) - + service_controller = WorkflowServiceController( + worker=worker, + llm_services=llm_services, + voice_services={"asr": stt_services, "tts": tts_services}, + current_services={ + "llm": current_llm_service, + **current_voice_services, + }, + ) current_enable_interrupt = cfg.enableInterrupt current_turn_config = dict(cfg.turnConfig) @@ -1091,21 +584,6 @@ async def run_pipeline( current_enable_interrupt = enable_interrupt current_turn_config = normalized - 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] = [] def set_system_prompt(text: str) -> None: """替换上下文里的系统提示(节点切换时整体替换,而非追加)。""" @@ -1132,7 +610,7 @@ async def run_pipeline( assistant_aggregator, ), transport=transport, - switch_services=switch_workflow_services, + switch_services=service_controller.switch, set_knowledge_scope=knowledge_retrieval.set_scope, set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled), apply_turn_config=apply_workflow_turn_config, @@ -1140,110 +618,18 @@ async def run_pipeline( ), ) - 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 brain.on_assistant_text_start(turn_id) - 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, - } - ) - ) - await brain.on_assistant_text_end(turn_id, content, 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 照常上报 - brain.record_user_message(text) - 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 greeting and not greeting_transcript_sent: - greeting_transcript_sent = True - await queue_transcript("assistant", greeting, time_now_iso8601()) - await brain.on_client_ready() - - @transport.event_handler("on_client_connected") - async def on_client_connected(_transport, _client): - if vision_enabled: - try: - vision_state["client_id"] = get_transport_client_id( - _transport, - _client, - ) - await maybe_capture_participant_camera(_transport, _client) - logger.info(f"视觉理解已接入视频客户端: {vision_state['client_id']}") - except Exception as e: - logger.warning(f"视觉理解摄像头捕获初始化失败: {e}") - if greeting: - # 外部托管类型的上下文由对方服务端维护,开场白不写入本地 context - if brain.spec.owns_context: - context.add_message({"role": "assistant", "content": greeting}) - await worker.queue_frame(TTSSpeakFrame(greeting, append_to_context=False)) - await brain.on_connected() - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(_transport, _client): - logger.info("对端断开,结束管线") - await worker.queue_frame(EndFrame()) - + bind_cascade_pipeline_events( + transport=transport, + worker=worker, + brain=brain, + context=context, + text_input=text_input, + user_aggregator=user_aggregator, + assistant_aggregator=assistant_aggregator, + greeting=greeting, + vision_enabled=vision_enabled, + vision_state=vision_state, + ) runner = WorkerRunner(handle_sigint=False) run_status = "completed" try: @@ -1306,40 +692,13 @@ async def run_realtime_pipeline( 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 greeting: - await realtime.speak(greeting) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(_transport, _client): - logger.info("Realtime 对端断开,结束管线") - await worker.queue_frame(EndFrame()) - + bind_realtime_pipeline_events( + transport=transport, + worker=worker, + realtime=realtime, + text_input=text_input, + greeting=greeting, + ) runner = WorkerRunner(handle_sigint=False) run_status = "completed" try: diff --git a/backend/services/pipecat/pipeline_events.py b/backend/services/pipecat/pipeline_events.py new file mode 100644 index 0000000..ca7b7c1 --- /dev/null +++ b/backend/services/pipecat/pipeline_events.py @@ -0,0 +1,199 @@ +"""Event registration for cascade and realtime conversation pipelines.""" + +from loguru import logger + +from pipecat.frames.frames import ( + EndFrame, + LLMMessagesAppendFrame, + OutputTransportMessageUrgentFrame, + TTSSpeakFrame, +) +from pipecat.runner.utils import ( + get_transport_client_id, + maybe_capture_participant_camera, +) +from pipecat.utils.time import time_now_iso8601 + + +def bind_cascade_pipeline_events( + *, + transport, + worker, + brain, + context, + text_input, + user_aggregator, + assistant_aggregator, + greeting: str, + vision_enabled: bool, + vision_state: dict[str, str | None], +) -> None: + """Connect processors to transport events without owning pipeline assembly.""" + + pending_text_inputs: list[str] = [] + greeting_transcript_sent = False + + async def queue_transcript(role: str, content: str, timestamp: str) -> None: + if not content: + return + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "transcript", + "role": role, + "content": content, + "timestamp": timestamp, + } + ) + ) + + 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 brain.on_assistant_text_start(turn_id) + 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, + } + ) + ) + await brain.on_assistant_text_end(turn_id, content, interrupted) + + @text_input.event_handler("on_text_input") + async def on_text_input(_processor, text): + pending_text_inputs.append(text) + # The transcript must be queued before the interruption is broadcast. + 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) + await append_user_text_to_context(text, run_llm=True) + + @text_input.event_handler("on_text_append") + async def on_text_append(_processor, text): + brain.record_user_message(text) + 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 greeting and not greeting_transcript_sent: + greeting_transcript_sent = True + await queue_transcript("assistant", greeting, time_now_iso8601()) + await brain.on_client_ready() + + @transport.event_handler("on_client_connected") + async def on_client_connected(_transport, _client): + if vision_enabled: + try: + vision_state["client_id"] = get_transport_client_id( + _transport, + _client, + ) + await maybe_capture_participant_camera(_transport, _client) + logger.info( + f"视觉理解已接入视频客户端: {vision_state['client_id']}" + ) + except Exception as exc: # noqa: BLE001 - media availability is optional + logger.warning(f"视觉理解摄像头捕获初始化失败: {exc}") + if greeting: + if brain.spec.owns_context: + context.add_message({"role": "assistant", "content": greeting}) + await worker.queue_frame( + TTSSpeakFrame(greeting, append_to_context=False) + ) + await brain.on_connected() + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(_transport, _client): + logger.info("对端断开,结束管线") + await worker.queue_frame(EndFrame()) + + +def bind_realtime_pipeline_events( + *, + transport, + worker, + realtime, + text_input, + greeting: str, +) -> None: + """Connect text and lifecycle events for a realtime model pipeline.""" + + async def queue_transcript(role: str, content: str) -> None: + if not content: + return + 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 greeting: + await realtime.speak(greeting) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(_transport, _client): + logger.info("Realtime 对端断开,结束管线") + await worker.queue_frame(EndFrame()) diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py new file mode 100644 index 0000000..7452b03 --- /dev/null +++ b/backend/services/pipecat/processors.py @@ -0,0 +1,452 @@ +"""Reusable frame processors shared by cascade and realtime pipelines.""" + +import asyncio +from collections.abc import Callable +from uuid import uuid4 + +from loguru import logger +from models import AssistantConfig +from services.brains import Brain +from services.conversation_history import ConversationRecorder +from services.knowledge import search as search_knowledge +from db.session import SessionLocal + +from pipecat.frames.frames import ( + InputTransportMessageFrame, + InterruptionFrame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + OutputTransportMessageUrgentFrame, + TextFrame, + UserImageRawFrame, + UserImageRequestFrame, +) +from pipecat.processors.aggregators.llm_response_universal import ( + LLMAssistantAggregator, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.turns.user_mute.base_user_mute_strategy import BaseUserMuteStrategy +from pipecat.utils.time import time_now_iso8601 + + +KNOWLEDGE_CONTEXT_MARKER = "" + + +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, should_ignore_input: Callable[[], bool] | None = None): + super().__init__() + self._should_ignore_input = should_ignore_input or (lambda: False) + # 立即触发的文字(含打断语义)走 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 + + if self._should_ignore_input(): + logger.debug("通话正在结束,忽略后续文字输入") + 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 CallEndingUserMuteStrategy(BaseUserMuteStrategy): + """Keep user media muted after an end-call tool starts terminating a call.""" + + def __init__(self, is_call_ending: Callable[[], bool]): + super().__init__() + self._is_call_ending = is_call_ending + + async def process_frame(self, frame) -> bool: + await super().process_frame(frame) + return self._is_call_ending() + + +class VisionCaptureProcessor(FrameProcessor): + """Capture one requested video frame for auxiliary vision-model analysis.""" + + def __init__(self, timeout_s: float = 3.0): + super().__init__() + self._timeout_s = timeout_s + self._pending: dict[str, asyncio.Future[UserImageRawFrame]] = {} + + async def request_image( + self, + requester: FrameProcessor, + request: UserImageRequestFrame, + ) -> UserImageRawFrame: + key = request.tool_call_id or str(uuid4()) + request.tool_call_id = key + request.append_to_context = False + request.result_callback = None + + loop = asyncio.get_running_loop() + future: asyncio.Future[UserImageRawFrame] = loop.create_future() + self._pending[key] = future + await requester.push_frame(request, FrameDirection.UPSTREAM) + try: + return await asyncio.wait_for(future, timeout=self._timeout_s) + finally: + self._pending.pop(key, None) + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if ( + isinstance(frame, UserImageRawFrame) + and frame.request + and frame.request.tool_call_id + and frame.request.tool_call_id in self._pending + ): + future = self._pending[frame.request.tool_call_id] + if not future.done(): + future.set_result(frame) + return + + await self.push_frame(frame, direction) + + +class RealtimeDynamicVariableProcessor(FrameProcessor): + """Keep realtime system turn/history variables current between responses.""" + + def __init__(self, brain: Brain, cfg: AssistantConfig, realtime): + super().__init__() + self._brain = brain + self._cfg = cfg + self._realtime = realtime + + async def _refresh_instructions(self) -> None: + update = getattr(self._realtime, "update_instructions", None) + if callable(update): + await update(self._brain.system_prompt(self._cfg)) + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, OutputTransportMessageUrgentFrame): + message = frame.message + if isinstance(message, dict): + event_type = message.get("type") + if event_type == "transcript" and message.get("role") == "user": + content = str(message.get("content") or "").strip() + if content: + self._brain.record_user_message(content) + await self._refresh_instructions() + elif event_type == "assistant-text-end": + await self._brain.on_assistant_text_end( + str(message.get("turn_id") or ""), + str(message.get("content") or ""), + bool(message.get("interrupted", False)), + ) + await self._refresh_instructions() + await self.push_frame(frame, direction) + + +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 ConversationHistoryProcessor(FrameProcessor): + """从最终客户端事件旁路保存历史，不改变 Pipecat 的上下文与帧语义。""" + + def __init__(self, recorder: ConversationRecorder | None): + super().__init__() + self._recorder = recorder + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + if self._recorder and isinstance(frame, OutputTransportMessageUrgentFrame): + await self._recorder.record_transport_message(frame.message) + + +class KnowledgeRetrievalProcessor(FrameProcessor): + """Retrieve before local LLM inference without changing Pipecat internals.""" + + def __init__( + self, + knowledge_base_id: str | None, + top_n: int = 5, + score_threshold: float = 0.0, + ): + super().__init__() + self._knowledge_base_id = knowledge_base_id + self._top_n = top_n + self._score_threshold = score_threshold + self._mode = "automatic" if knowledge_base_id else "disabled" + self._last_signature = "" + + def set_scope(self, scope: dict) -> None: + self._knowledge_base_id = scope.get("knowledge_base_id") or None + self._mode = str(scope.get("mode") or "disabled") + self._top_n = int(scope.get("top_n") or 5) + self._score_threshold = float(scope.get("score_threshold") or 0.0) + self._last_signature = "" + + def _clear_context(self, messages: list[dict]) -> None: + # Remove the legacy Workflow knowledge message so an in-flight context + # created before this compatibility fix cannot keep sending that role. + messages[:] = [ + message + for message in messages + if not ( + message.get("role") == "developer" + and KNOWLEDGE_CONTEXT_MARKER in str(message.get("content") or "") + ) + ] + system_message = next( + (message for message in messages if message.get("role") == "system"), + None, + ) + if system_message is not None: + content = str(system_message.get("content") or "") + system_message["content"] = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip() + + def _set_context(self, messages: list[dict], block: str) -> None: + """Store retrieved knowledge in a provider-compatible system message.""" + self._clear_context(messages) + system_message = next( + (message for message in messages if message.get("role") == "system"), + None, + ) + if system_message is None: + messages.insert(0, {"role": "system", "content": block}) + return + content = str(system_message.get("content") or "").rstrip() + system_message["content"] = f"{content}\n\n{block}" if content else block + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, LLMContextFrame): + await self.push_frame(frame, direction) + return + + messages = frame.context.get_messages() + if self._mode != "automatic" or not self._knowledge_base_id: + self._clear_context(messages) + await self.push_frame(frame, direction) + return + user_messages = [message for message in messages if message.get("role") == "user"] + if not user_messages: + await self.push_frame(frame, direction) + return + query = str(user_messages[-1].get("content") or "").strip() + signature = f"{len(user_messages)}:{query}" + if not query or signature == self._last_signature: + await self.push_frame(frame, direction) + return + self._last_signature = signature + + try: + async with SessionLocal() as session: + results = await search_knowledge( + session, + self._knowledge_base_id, + query, + top_k=self._top_n, + score_threshold=self._score_threshold, + ) + except Exception as exc: + logger.warning(f"自动知识库检索失败: {exc}") + results = [] + + sources = "\n\n".join( + f"[{index + 1}] 来源：{item['document']}（相关度 {item['score']}）\n{item['content']}" + for index, item in enumerate(results) + ) or "未检索到相关资料。" + block = f"{KNOWLEDGE_CONTEXT_MARKER}\n当前问题的知识库检索结果：\n{sources}" + self._set_context(messages, block) + await self.push_frame(frame, direction) + + +class UserTurnRoutingProcessor(FrameProcessor): + """Give a brain first right of refusal before a new user turn reaches the LLM.""" + + def __init__(self, brain: Brain): + super().__init__() + self._brain = brain + self._last_user_message: dict | None = None + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction != FrameDirection.DOWNSTREAM or not isinstance( + frame, LLMContextFrame + ): + await self.push_frame(frame, direction) + return + + user_message = next( + ( + message + for message in reversed(frame.context.get_messages()) + if message.get("role") == "user" + and isinstance(message.get("content"), str) + and str(message.get("content") or "").strip() + ), + None, + ) + if user_message is None: + await self.push_frame(frame, direction) + return + + if user_message is self._last_user_message: + # Programmatic LLMRunFrame after a node transition reuses the same + # user message. It is a response run, not another routing event. + await self.push_frame(frame, direction) + return + self._last_user_message = user_message + + content = str(user_message.get("content") or "").strip() + handled = await self._brain.on_user_turn_end(content) + if not handled: + await self.push_frame(frame, direction) + + +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 = "" + + +class WorkflowAggregatorPair: + """Small public-shape adapter required by Pipecat FlowManager.""" + + def __init__(self, user_aggregator, assistant_aggregator): + self._user = user_aggregator + self._assistant = assistant_aggregator + + def user(self): + return self._user + + def assistant(self): + return self._assistant + + + diff --git a/backend/services/pipecat/workflow_services.py b/backend/services/pipecat/workflow_services.py new file mode 100644 index 0000000..c940fcc --- /dev/null +++ b/backend/services/pipecat/workflow_services.py @@ -0,0 +1,131 @@ +"""Workflow model resource loading and runtime service switching.""" + +from models import AssistantConfig +from services.pipecat.service_factory import ( + config_with_resource, + create_llm, + create_stt, + create_tts, +) + +from pipecat.frames.frames import ( + ManuallySwitchServiceFrame, + OutputTransportMessageUrgentFrame, +) +from pipecat.pipeline.llm_switcher import LLMSwitcher +from pipecat.pipeline.service_switcher import ServiceSwitcher +from pipecat.processors.frame_processor import FrameProcessor + + +def build_workflow_voice_switcher( + cfg: AssistantConfig, capability: str, base_service: FrameProcessor +): + """Build one switcher and an ID lookup for every referenced voice resource.""" + create = create_stt if capability == "ASR" else create_tts + settings = cfg.graph.get("settings") or {} + default_key = ( + "defaultAsrResourceId" if capability == "ASR" else "defaultTtsResourceId" + ) + default_id = str(settings.get(default_key) or "") + services_by_id = {} + for resource_id, resource in cfg.workflow_model_resources.items(): + if resource.capability != capability: + continue + services_by_id[resource_id] = ( + base_service + if resource_id == default_id + else create(config_with_resource(cfg, resource)) + ) + primary = services_by_id.get(default_id, base_service) + services = [primary] + services.extend( + service for service in services_by_id.values() if service is not primary + ) + if base_service is not primary: + services.append(base_service) + return ServiceSwitcher(services=services), services_by_id, primary + + +def build_workflow_llm_switcher(cfg: AssistantConfig, base_service): + """Build an LLM switcher for the global model and Agent overrides.""" + settings = cfg.graph.get("settings") or {} + default_id = str(settings.get("defaultLlmResourceId") or "") + services_by_id = {} + for resource_id, resource in cfg.workflow_model_resources.items(): + if resource.capability != "LLM": + continue + services_by_id[resource_id] = ( + base_service + if resource_id == default_id + else create_llm(config_with_resource(cfg, resource)) + ) + primary = services_by_id.get(default_id, base_service) + services = [primary] + services.extend( + service for service in services_by_id.values() if service is not primary + ) + if base_service is not primary: + services.append(base_service) + return LLMSwitcher(llms=services), services_by_id, primary + + + + +class WorkflowServiceController: + """Switch one Workflow stage's model resources without leaking state.""" + + def __init__( + self, + *, + worker, + llm_services: dict[str, FrameProcessor], + voice_services: dict[str, dict[str, FrameProcessor]], + current_services: dict[str, FrameProcessor], + ) -> None: + self._worker = worker + self._services = { + "llm": llm_services, + "asr": voice_services["asr"], + "tts": voice_services["tts"], + } + self._current = dict(current_services) + self._defaults = dict(current_services) + + async def switch( + self, + llm_resource_id: str | None, + asr_resource_id: str | None, + tts_resource_id: str | None, + ) -> None: + requested = ( + ("llm", llm_resource_id), + ("asr", asr_resource_id), + ("tts", tts_resource_id), + ) + for kind, resource_id in requested: + target = ( + self._services[kind].get(resource_id) + if resource_id + else self._defaults[kind] + ) + if target is None: + raise ValueError( + f"Workflow {kind.upper()} 资源未加载:{resource_id}" + ) + if self._current[kind] is target: + continue + + await self._worker.queue_frame( + ManuallySwitchServiceFrame(service=target) + ) + self._current[kind] = target + await self._worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "service-switched", + "capability": kind.upper(), + "resourceId": resource_id, + } + ) + ) + diff --git a/frontend/src/components/assistant-editor/debug-preview.tsx b/frontend/src/components/assistant-editor/debug-preview.tsx new file mode 100644 index 0000000..7bb326a --- /dev/null +++ b/frontend/src/components/assistant-editor/debug-preview.tsx @@ -0,0 +1,1253 @@ +"use client"; + +import type React from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; +import { + AudioLines, + Braces, + Check, + Copy, + Loader2, + MessageSquareText, + Mic, + Orbit, + PhoneOff, + Send, + Smartphone, + Sparkles, + Video, + Waves, + X, +} from "lucide-react"; + +import { NetworkQualityIndicator } from "@/components/network-quality-indicator"; +import { AuraVisualizer } from "@/components/ui/aura-visualizer"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { NebulaVisualizer } from "@/components/ui/nebula-visualizer"; +import { + Popover, + PopoverContent, + PopoverTrigger, +} from "@/components/ui/popover"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { SpectrumVisualizer } from "@/components/ui/spectrum-visualizer"; +import { Textarea } from "@/components/ui/textarea"; +import { WaveVisualizer } from "@/components/ui/wave-visualizer"; +import { WaveformTimelinePanel } from "@/components/ui/waveform-timeline"; +import { useCameraPreview, type CameraPreview } from "@/hooks/use-camera-preview"; +import { + useVoicePreview, + type ChatMessage, + type VoicePreview, + type VoicePreviewStatus, +} from "@/hooks/use-voice-preview"; +import type { DynamicVariableDefinition } from "@/lib/api"; + +type VizStyle = "aura" | "nebula" | "bars" | "wave"; + +// 调试面板顶部主视图:聊天记录 / 视频流 +type DebugView = "chat" | "video"; +type DebugInputMode = "mic" | "text"; + +const VIZ_OPTIONS: { style: VizStyle; label: string; icon: React.ReactNode }[] = + [ + { style: "aura", label: "光环", icon: }, + { style: "nebula", label: "星云", icon: }, + { style: "bars", label: "频谱", icon: }, + { style: "wave", label: "波形", icon: }, + ]; + +// 中央语音可视化(光环/星云/频谱/波形)暂时隐藏:调试面板固定为 +// 「上聊天记录 + 下波形监控」布局。置 true 可恢复可视化视图与样式切换。 +const SHOW_VOICE_VIZ = false; + +function SegmentedIconGroup({ + children, + label, +}: { + children: React.ReactNode; + label: string; +}) { + return ( +
+ {children} +
+ ); +} + +function SegmentedIconButton({ + selected, + label, + onClick, + children, +}: { + selected: boolean; + label: string; + onClick: () => void; + children: React.ReactNode; +}) { + return ( + + ); +} + +export function DebugDrawer({ + assistantId, + overlay = false, + onClose, + hasUnsavedChanges = false, + onNodeActive, + vision = false, + dynamicVariablesEnabled = false, + dynamicVariableDefinitions = {}, +}: { + assistantId: string | null; + overlay?: boolean; + onClose?: () => void; + hasUnsavedChanges?: boolean; + onNodeActive?: (nodeId: string | null) => void; + vision?: boolean; + dynamicVariablesEnabled?: boolean; + dynamicVariableDefinitions?: Record; +}) { + const preview = useVoicePreview(assistantId, onNodeActive); + const camera = useCameraPreview(); + const [showTranscript, setShowTranscript] = useState(false); + const [vizStyle, setVizStyle] = useState("aura"); + const [view, setView] = useState("chat"); + const [dynamicVariableValues, setDynamicVariableValues] = useState< + Record + >({}); + const recording = + preview.status === "connecting" || preview.status === "connected"; + const displayedDefinitions = { ...dynamicVariableDefinitions }; + for (const [name, value] of Object.entries(preview.sessionVariables)) { + displayedDefinitions[name] ??= { + type: + typeof value === "number" + ? "number" + : typeof value === "boolean" + ? "boolean" + : "string", + required: false, + default: null, + }; + } + const dynamicVariableEntries = Object.entries(displayedDefinitions); + const resolvedDynamicVariables: Record = {}; + let dynamicVariablesError = ""; + for (const [name, definition] of Object.entries(dynamicVariableDefinitions)) { + const value = dynamicVariableValues[name] ?? definition.default; + if (value === null || value === undefined || value === "") { + if (definition.required && !dynamicVariablesError) { + dynamicVariablesError = `请先填写必填变量 ${name}`; + } + continue; + } + resolvedDynamicVariables[name] = value; + } + + const selectCamera = useCallback( + async (deviceId: string) => { + await camera.selectCamera(deviceId); + preview.selectCamera(deviceId); + }, + [camera, preview], + ); + + return ( + + ); +} + +function DynamicVariableValuesPopover({ + entries, + values, + sessionValues, + readOnly, + onChange, +}: { + entries: [string, DynamicVariableDefinition][]; + values: Record; + sessionValues: Record; + readOnly: boolean; + onChange: React.Dispatch< + React.SetStateAction> + >; +}) { + function setValue(name: string, value: string | number | boolean | undefined) { + onChange((current) => { + const next = { ...current }; + if (value === undefined) delete next[name]; + else next[name] = value; + return next; + }); + } + + return ( + + + + + +
+
+ + 本次会话变量 +
+

+ {readOnly + ? "当前值会在 Action 或工具更新变量后实时刷新。" + : "这些值只用于下一次调试会话，不会修改助手配置。"} +

+
+
+ {entries.length === 0 ? ( +
+
+ 当前没有会话变量 +
+

+ 在工作流提示词、节点话术、边条件或 Action 中引用变量后， + 可在这里设置调试值。 +

+
+ ) : entries.map(([name, definition]) => { + const value = readOnly + ? sessionValues[name] ?? definition.default ?? "" + : values[name] ?? definition.default ?? ""; + return ( + + ); + })} +
+ + + ); +} + +function CallPreviewLink({ assistantId }: { assistantId: string | null }) { + const [callUrl, setCallUrl] = useState(""); + const [copied, setCopied] = useState(false); + + const copyLink = useCallback(async () => { + if (!callUrl) return; + await navigator.clipboard.writeText(callUrl); + setCopied(true); + window.setTimeout(() => setCopied(false), 1600); + }, [callUrl]); + + return ( + { + if (open && assistantId) { + setCallUrl( + `${window.location.origin}/call/${encodeURIComponent(assistantId)}`, + ); + setCopied(false); + } + }} + > + + + + +
+
手机通话预览
+

+ 复制链接并在浏览器中打开，即可进入全屏视频通话界面。 +

+
+
+ event.currentTarget.select()} + className="h-9 min-w-0 flex-1 text-xs" + /> + +
+ {copied &&

链接已复制

} + + + ); +} + +function DeviceSelectField({ + icon, + ariaLabel, + placeholder, + fallbackLabel, + value, + devices, + onSelect, +}: { + icon: React.ReactNode; + ariaLabel: string; + placeholder: string; + fallbackLabel: string; + value: string; + devices: MediaDeviceInfo[]; + onSelect: (deviceId: string) => void; +}) { + return ( + + ); +} + +function MicrophoneDeviceField({ preview }: { preview: VoicePreview }) { + return ( + } + ariaLabel="选择麦克风" + placeholder="默认麦克风" + fallbackLabel="麦克风" + value={preview.selectedDeviceId} + devices={preview.audioInputs} + onSelect={(deviceId) => preview.selectDevice(deviceId)} + /> + ); +} + +function CameraDeviceField({ + camera, + onSelect, +}: { + camera: CameraPreview; + onSelect?: (deviceId: string) => void | Promise; +}) { + return ( + } + ariaLabel="选择摄像头" + placeholder="默认摄像头" + fallbackLabel="摄像头" + value={camera.deviceId} + devices={camera.devices} + onSelect={(deviceId) => void (onSelect ?? camera.selectCamera)(deviceId)} + /> + ); +} + +function DebugInputModeButton({ + selected, + label, + onClick, + children, +}: { + selected: boolean; + label: string; + onClick: () => void; + children: React.ReactNode; +}) { + return ( + + ); +} + +function DebugVoicePanel({ + view, + onViewChange, + showTranscript, + vizStyle, + assistantId, + preview, + camera, + hasUnsavedChanges, + vision, + dynamicVariables, + dynamicVariablesError, +}: { + view: DebugView; + onViewChange: (view: DebugView) => void; + showTranscript: boolean; + vizStyle: VizStyle; + assistantId: string | null; + preview: VoicePreview; + camera: CameraPreview; + hasUnsavedChanges: boolean; + vision: boolean; + dynamicVariables: Record; + dynamicVariablesError: string; +}) { + const { + status, + error, + micWarning, + localStream, + remoteStream, + messages, + sendText, + connect, + disconnect, + audioRef, + } = preview; + const recording = status === "connecting" || status === "connected"; + const [textDraft, setTextDraft] = useState(""); + const [inputMode, setInputMode] = useState("mic"); + const inChatView = view === "chat" && (!SHOW_VOICE_VIZ || showTranscript); + const idleOrFailed = status === "idle" || status === "failed"; + const showIdleHub = + idleOrFailed && + (view === "video" || (messages.length === 0 && inChatView)); + const startBlockedMessage = !assistantId + ? "请先保存助手，再开始对话。" + : hasUnsavedChanges + ? "请先保存当前改动，再开始对话。" + : dynamicVariablesError + ? dynamicVariablesError + : ""; + const startDisabled = status === "connecting" || Boolean(startBlockedMessage); + + function handleSendText() { + if (sendText(textDraft)) { + setTextDraft(""); + } + } + + const startConversation = useCallback(async () => { + if (!assistantId || hasUnsavedChanges) return; + if (dynamicVariablesError) return; + await connect({ + visionEnabled: vision, + dynamicVariables, + }); + }, [ + assistantId, + connect, + dynamicVariables, + dynamicVariablesError, + hasUnsavedChanges, + vision, + ]); + + return ( +
+ {/* 后端 TTS 音频经 WebRTC 媒体流过来,挂这里播放 */} +