From eb35caffcb3f70ce3215f305cc08efa824f0e685 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 30 Oct 2026 14:39:22 +0800 Subject: [PATCH] feat: add optional bot backchannels to prompt pipelines --- backend/contracts/openapi.json | 124 ++++++++++++++ backend/schemas.py | 6 + backend/services/pipecat/backchannel.py | 100 +++++++++++ backend/services/pipecat/pipeline.py | 10 ++ backend/tests/test_bot_backchannel.py | 156 ++++++++++++++++++ docs-developer/prompt-task-backends.mdx | 8 + .../assistant-editor/prompt-editor.tsx | 21 +++ frontend/src/lib/api.ts | 1 + frontend/src/lib/turn-config.ts | 3 + 9 files changed, 429 insertions(+) create mode 100644 backend/services/pipecat/backchannel.py create mode 100644 backend/tests/test_bot_backchannel.py diff --git a/backend/contracts/openapi.json b/backend/contracts/openapi.json index 921a309..ae8005f 100644 --- a/backend/contracts/openapi.json +++ b/backend/contracts/openapi.json @@ -1198,6 +1198,24 @@ "title": "Body_upload_test_image_api_test_assets_images_post", "type": "object" }, + "BotBackchannelConfig": { + "properties": { + "delaySecs": { + "default": 4.0, + "maximum": 15.0, + "minimum": 2.0, + "title": "Delaysecs", + "type": "number" + }, + "enabled": { + "default": false, + "title": "Enabled", + "type": "boolean" + } + }, + "title": "BotBackchannelConfig", + "type": "object" + }, "BuiltInToolBase": { "properties": { "description": { @@ -6406,6 +6424,9 @@ "bargeIn": { "$ref": "#/components/schemas/BargeInConfig" }, + "botBackchannel": { + "$ref": "#/components/schemas/BotBackchannelConfig" + }, "turnDetection": { "$ref": "#/components/schemas/TurnDetectionConfig" }, @@ -6612,6 +6633,69 @@ ], "title": "VariableValueSource", "type": "object" + }, + "VoiceRequest": { + "properties": { + "action": { + "enum": [ + "create_voice", + "query_voice" + ], + "title": "Action", + "type": "string" + }, + "apiKey": { + "default": "", + "maxLength": 4096, + "title": "Apikey", + "type": "string" + }, + "apiUrl": { + "maxLength": 2048, + "title": "Apiurl", + "type": "string" + }, + "audio": { + "default": "", + "maxLength": 14000000, + "title": "Audio", + "type": "string" + }, + "model": { + "title": "Model", + "type": "string" + }, + "prefix": { + "default": "assistant", + "pattern": "^[A-Za-z0-9]{1,10}$", + "title": "Prefix", + "type": "string" + }, + "resourceId": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Resourceid" + }, + "voiceId": { + "default": "", + "maxLength": 512, + "title": "Voiceid", + "type": "string" + } + }, + "required": [ + "apiUrl", + "model", + "action" + ], + "title": "VoiceRequest", + "type": "object" } }, "securitySchemes": { @@ -8483,6 +8567,46 @@ ] } }, + "/api/model-resources/dashscope/voice": { + "post": { + "description": "Create/query a voice using draft settings and server-held credentials.", + "operationId": "manage_dashscope_voice_api_model_resources_dashscope_voice_post", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/VoiceRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Manage Dashscope Voice", + "tags": [ + "model-registry" + ] + } + }, "/api/model-resources/test": { "post": { "operationId": "test_new_model_resource_api_model_resources_test_post", diff --git a/backend/schemas.py b/backend/schemas.py index 8ff44c2..cd4cbd3 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -68,7 +68,13 @@ class TurnDetectionConfig(CamelModel): silence_timeout_secs: float = Field(default=0.6, ge=0.2, le=5.0) +class BotBackchannelConfig(CamelModel): + enabled: bool = False + delay_secs: float = Field(default=4.0, ge=2.0, le=15.0) + + class TurnConfig(CamelModel): + bot_backchannel: BotBackchannelConfig = Field(default_factory=BotBackchannelConfig) barge_in: BargeInConfig = Field(default_factory=BargeInConfig) vad: VadConfig = Field(default_factory=VadConfig) turn_detection: TurnDetectionConfig = Field(default_factory=TurnDetectionConfig) diff --git a/backend/services/pipecat/backchannel.py b/backend/services/pipecat/backchannel.py new file mode 100644 index 0000000..89b2d49 --- /dev/null +++ b/backend/services/pipecat/backchannel.py @@ -0,0 +1,100 @@ +"""Optional bot acknowledgments; never invoke the LLM or change user turns. + +Placed immediately before TTS. A pending timer is cancelled on silence or a +reply. Once handed to TTS, the very short acknowledgment may finish normally. +""" +from __future__ import annotations + +import asyncio +import time +from collections.abc import Callable + +from pipecat.frames.frames import ( + BotBackchannelFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, + CancelFrame, EndFrame, Frame, InterruptionFrame, LLMFullResponseStartFrame, + TTSSpeakFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class BotBackchannelProcessor(FrameProcessor): + """One short acknowledgment per long user turn, with a global cooldown.""" + + def __init__(self, *, delay_secs: float = 4, can_speak: Callable[[], bool]): + super().__init__() + self._delay = delay_secs + self._can_speak = can_speak + self._user_turn = False + self._voice_active = False + self._used = False + self._bot_busy = False + self._closed = False + self._last_sent = float("-inf") + self._timer: asyncio.Task | None = None + + async def _cancel_pending(self): + timer, self._timer = self._timer, None + if timer and timer is not asyncio.current_task(): + await self.cancel_task(timer) + + def _schedule(self): + if (not self._timer and self._user_turn and self._voice_active + and not self._used and not self._bot_busy and not self._closed): + self._timer = self.create_task(self._acknowledge()) + + async def _acknowledge(self): + try: + await asyncio.sleep(self._delay) + if (self._user_turn and self._voice_active and not self._used + and not self._bot_busy and not self._closed and self._can_speak() + and time.monotonic() - self._last_sent >= 10): + self._used = True + self._last_sent = time.monotonic() + # Reserve output before synthesis starts; playback events follow later. + self._bot_busy = True + await self.push_frame(BotBackchannelFrame(text="嗯。")) + finally: + self._timer = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # Playback boundaries travel upstream; user/VAD frames can be broadcast + # both ways. Handle user signals only on the downstream path. + if isinstance(frame, BotStartedSpeakingFrame): + self._bot_busy = True + await self._cancel_pending() + elif isinstance(frame, BotStoppedSpeakingFrame): + self._bot_busy = False + self._schedule() + elif isinstance(frame, (EndFrame, CancelFrame)): + self._closed = True + await self._cancel_pending() + elif direction == FrameDirection.DOWNSTREAM: + if isinstance(frame, UserStartedSpeakingFrame): + self._user_turn, self._used = True, False + self._schedule() + elif isinstance(frame, VADUserStartedSpeakingFrame): + self._voice_active = True + self._schedule() + elif isinstance(frame, VADUserStoppedSpeakingFrame): + self._voice_active = False + await self._cancel_pending() + elif isinstance(frame, UserStoppedSpeakingFrame): + self._user_turn = False + await self._cancel_pending() + elif isinstance(frame, (LLMFullResponseStartFrame, TTSSpeakFrame)): + self._bot_busy = True + await self._cancel_pending() + elif isinstance(frame, InterruptionFrame): + # User-start strategies broadcast an interruption after the + # turn-start signal, even when no bot audio was playing. + await self._cancel_pending() + self._bot_busy = False + self._schedule() + await self.push_frame(frame, direction) + + async def cleanup(self): + self._closed = True + await self._cancel_pending() + await super().cleanup() diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 26edc26..ba31328 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -14,6 +14,7 @@ from models import AssistantConfig from services.brains import Brain, BrainRuntime, build_brain from services.brains.base import RealtimeBrainRuntime from services.conversation_history import ConversationRecorder +from services.pipecat.backchannel import BotBackchannelProcessor from services.pipecat.call_lifecycle import ( CallEndCoordinator, EndCallAfterSpeechProcessor, @@ -722,6 +723,14 @@ async def run_pipeline( protocol_adapter.inference_processors() if protocol_adapter else [] ) protocol_outputs = protocol_adapter.output_processors() if protocol_adapter else [] + backchannel_config = cfg.turnConfig.get("bot_backchannel", cfg.turnConfig.get("botBackchannel", {})) + backchannel_processors = [] + if cfg.type == "prompt" and backchannel_config.get("enabled", False): + backchannel_processors.append(BotBackchannelProcessor( + delay_secs=float(backchannel_config.get("delay_secs", backchannel_config.get("delaySecs", 4))), + can_speak=lambda: (call_end.accepts_input and not call_end.ending + and input_state["enabled"] and not tool_interruption_strategy.is_muted), + )) pipeline = Pipeline( [ transport.input(), @@ -740,6 +749,7 @@ async def run_pipeline( # Pipecat commits the generated prefix immediately instead of # waiting for a TTS provider to emit spoken-text/timestamp frames. assistant_aggregator, + *backchannel_processors, tts_processor, EndCallAfterSpeechProcessor(call_end), ConversationHistoryProcessor(recorder), diff --git a/backend/tests/test_bot_backchannel.py b/backend/tests/test_bot_backchannel.py new file mode 100644 index 0000000..3f9703a --- /dev/null +++ b/backend/tests/test_bot_backchannel.py @@ -0,0 +1,156 @@ +import asyncio +import unittest +from unittest.mock import AsyncMock, patch + +from pipecat.frames.frames import ( + BotBackchannelFrame, BotStoppedSpeakingFrame, CancelFrame, + InterruptionFrame, LLMFullResponseStartFrame, TTSSpeakFrame, + UserStartedSpeakingFrame, UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from schemas import TurnConfig +from services.pipecat.backchannel import BotBackchannelProcessor + + +class BotBackchannelTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.allowed = True + self.frames = [] + self.processor = BotBackchannelProcessor(delay_secs=0.01, can_speak=lambda: self.allowed) + self.processor.create_task = asyncio.create_task + async def cancel(task): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self.processor.cancel_task = cancel + async def push(frame, direction=FrameDirection.DOWNSTREAM): + self.frames.append(frame) + self.processor.push_frame = push + self.patch = patch.object(FrameProcessor, "process_frame", AsyncMock()) + self.patch.start() + + async def asyncTearDown(self): + await self.processor.cleanup() + self.patch.stop() + + async def send(self, frame): + await self.processor.process_frame(frame, FrameDirection.DOWNSTREAM) + + async def start_turn(self): + await self.send(VADUserStartedSpeakingFrame()) + await self.send(UserStartedSpeakingFrame()) + await self.send(InterruptionFrame()) + + def acknowledgments(self): + return [frame for frame in self.frames if isinstance(frame, BotBackchannelFrame)] + + async def test_long_turn_once_and_excludes_context(self): + await self.start_turn() + await asyncio.sleep(0.03) + self.assertEqual(len(self.acknowledgments()), 1) + self.assertFalse(self.acknowledgments()[0].append_to_context) + self.assertFalse(self.acknowledgments()[0].interruptible) + await self.send(BotStoppedSpeakingFrame()) + await self.send(VADUserStoppedSpeakingFrame()) + await self.send(VADUserStartedSpeakingFrame()) + await asyncio.sleep(0.03) + self.assertEqual(len(self.acknowledgments()), 1) + + async def test_short_turn_does_not_acknowledge(self): + await self.start_turn() + await self.send(UserStoppedSpeakingFrame()) + await asyncio.sleep(0.03) + self.assertFalse(self.acknowledgments()) + + async def test_silence_cancels_pending(self): + await self.start_turn() + await self.send(VADUserStoppedSpeakingFrame()) + await asyncio.sleep(0.03) + self.assertFalse(self.acknowledgments()) + + async def test_formal_reply_cancels_pending(self): + await self.start_turn() + await self.send(LLMFullResponseStartFrame()) + await asyncio.sleep(0.03) + self.assertFalse(self.acknowledgments()) + + async def test_greeting_reserves_output(self): + await self.send(TTSSpeakFrame("欢迎")) + await self.send(UserStartedSpeakingFrame()) + await self.send(VADUserStartedSpeakingFrame()) + await asyncio.sleep(0.03) + self.assertFalse(self.acknowledgments()) + + async def test_closing_or_muted_blocks_timer(self): + await self.start_turn() + self.allowed = False + await asyncio.sleep(0.03) + self.assertFalse(self.acknowledgments()) + + async def test_cancel_closes_controller(self): + await self.start_turn() + await self.send(CancelFrame()) + await asyncio.sleep(0.03) + self.assertFalse(self.acknowledgments()) + self.assertIsNone(self.processor._timer) + + async def test_cooldown_applies_between_turns(self): + await self.start_turn() + await asyncio.sleep(0.03) + await self.send(UserStoppedSpeakingFrame()) + await self.send(BotStoppedSpeakingFrame()) + await self.start_turn() + await asyncio.sleep(0.03) + self.assertEqual(len(self.acknowledgments()), 1) + + def test_old_configuration_defaults_off_and_alias_roundtrip(self): + self.assertFalse(TurnConfig().bot_backchannel.enabled) + config = TurnConfig.model_validate({"botBackchannel": {"enabled": True, "delaySecs": 5}}) + self.assertEqual(config.bot_backchannel.delay_secs, 5) + + +class BackchannelPipelineTest(unittest.IsolatedAsyncioTestCase): + async def test_native_pipeline_delivers_backchannel_without_assistant_context(self): + """Exercise real frame queues and the project's pre-TTS aggregator.""" + from pipecat.frames.frames import EndFrame + from pipecat.pipeline.pipeline import Pipeline + from pipecat.pipeline.worker import PipelineWorker + from pipecat.processors.aggregators.llm_context import LLMContext + from services.pipecat.processors import PassthroughLLMAssistantAggregator + from services.pipecat.pipeline_runner import run_media_worker + + received = asyncio.Event() + context = LLMContext([]) + class Sink(FrameProcessor): + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if isinstance(frame, BotBackchannelFrame): + received.set() + await self.push_frame(frame, direction) + + controller = BotBackchannelProcessor(delay_secs=0.01, can_speak=lambda: True) + worker = PipelineWorker(Pipeline([ + PassthroughLLMAssistantAggregator(context), controller, Sink(), + ]), enable_rtvi=False) + @worker.event_handler("on_pipeline_started") + async def start(_worker, _frame): + for frame in (VADUserStartedSpeakingFrame(), UserStartedSpeakingFrame(), InterruptionFrame()): + await worker.queue_frame(frame) + + run = asyncio.create_task(run_media_worker(worker, None)) + try: + await asyncio.wait_for(received.wait(), 3) + self.assertEqual(context.messages, []) + await worker.queue_frame(EndFrame()) + await asyncio.wait_for(run, 3) + self.assertIsNone(controller._timer) + finally: + if not run.done(): + run.cancel() + try: + await run + except asyncio.CancelledError: + pass diff --git a/docs-developer/prompt-task-backends.mdx b/docs-developer/prompt-task-backends.mdx index 19d67db..e912dec 100644 --- a/docs-developer/prompt-task-backends.mdx +++ b/docs-developer/prompt-task-backends.mdx @@ -138,3 +138,11 @@ cd backend - [OpenCode Server](https://opencode.ai/docs/server/) - [OpenCode 权限](https://opencode.ai/docs/permissions/) - [OpenCode 技能](https://opencode.ai/docs/skills/) + +## 机器人附和（Prompt Pipeline） + +在提示词助手的「交互策略」开启「机器人附和」。默认关闭；开启后默认等待用户连续讲话 4 秒（可设 2～15 秒），使用当前 TTS 音色说一次“嗯。”。每个用户轮次最多一次，两次发送之间至少间隔 10 秒。短暂停顿会取消当前等待，继续讲话后重新计时。 + +附和直接使用 Pipecat `BotBackchannelFrame` 进入 TTS，不调用 Speaker LLM、不写入主对话上下文，也不生成普通回答转写。它不会识别或过滤用户的附和，不改变既有打断策略。Workflow 与 Realtime 暂不启用该能力。 + +用户停说、正式回答开始、固定播报开始或连接结束时取消待发送附和；执行时也检查输入开关、工具静音与挂断状态。已经交给 TTS 的短附和会正常播完，不能保证用户结束讲话的瞬间停止，较慢的 TTS 仍可能影响接续延迟。配置修改后需重新连接会话生效。真实麦克风下的回声、音色自然度与接续延迟需按部署环境验收。 diff --git a/frontend/src/components/assistant-editor/prompt-editor.tsx b/frontend/src/components/assistant-editor/prompt-editor.tsx index d525c93..a43ea4b 100644 --- a/frontend/src/components/assistant-editor/prompt-editor.tsx +++ b/frontend/src/components/assistant-editor/prompt-editor.tsx @@ -935,6 +935,7 @@ export function PromptEditor({ description="设置实时视频对话时的交互体验" > {form.runtimeMode === "pipeline" ? ( + <> +
+ +

用户持续讲话时轻声回应“嗯”，每轮最多一次，间隔至少 10 秒。不会触发模型回答，也不会改变用户打断策略。

+ {form.turnConfig.botBackchannel?.enabled && } +
+ ) : (

高级抢话与轮次检测当前仅支持 Pipeline 模式。 diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index c255239..e210437 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -268,6 +268,7 @@ export type AssistantType = | "fastgpt"; export type RuntimeMode = "pipeline" | "realtime"; export type TurnConfig = { + botBackchannel?: { enabled: boolean; delaySecs: number }; bargeIn: { strategy: "vad" | "transcription"; }; diff --git a/frontend/src/lib/turn-config.ts b/frontend/src/lib/turn-config.ts index c142051..9ad6d28 100644 --- a/frontend/src/lib/turn-config.ts +++ b/frontend/src/lib/turn-config.ts @@ -2,6 +2,7 @@ import type { TurnConfig } from "@/lib/api"; export function defaultTurnConfig(): TurnConfig { return { + botBackchannel: { enabled: false, delaySecs: 4 }, bargeIn: { strategy: "vad" }, vad: { confidence: 0.7, @@ -32,11 +33,13 @@ function numberOr(value: unknown, fallback: number): number { export function normalizeTurnConfig(value: unknown): TurnConfig { const defaults = defaultTurnConfig(); const input = record(value); + const botBackchannel = record(input.botBackchannel); const bargeIn = record(input.bargeIn); const vad = record(input.vad); const turnDetection = record(input.turnDetection); return { + botBackchannel: { enabled: botBackchannel.enabled === true, delaySecs: numberOr(botBackchannel.delaySecs, 4) }, bargeIn: { strategy: bargeIn.strategy === "transcription" ? "transcription" : "vad",