Pine/DPM
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(s2s): 后端嵌入实时语音栈（云 ASR/LLM/TTS + 本地 VAD）

Browse Source 
- s2s_bridge：同进程构建 s2s 管线池（silero VAD + dashscope ASR + qwen-plus + qwen3-tts）并起 RealtimeServer(:8765)
- ThreadManager 启动管线 handler 线程；启动重试与完整配置日志
- config：DPM_S2S_* 系列配置（开关/端口/模型/重试/日志级别）
- main lifespan 按 DPM_S2S_ENABLED 启停；pyproject 增加 s2s 依赖（httpx/nltk/scipy/torch 等）
This commit is contained in:
Pine
2026-08-18 01:37:48 +08:00
parent 4e2224e9dc
commit 9f64b4273b
5 changed files with 1559 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/config.py
+14
View File
@@ -70,5 +70,19 @@ class Settings:
# ---- 管理后台 ----
ADMIN_SECRET = _env("DPM_ADMIN_SECRET", "dpm-admin-secret-change-me")
# ---- s2s 实时语音栈（VAD本地 + 云ASR/LLM/TTS） ----
S2S_ENABLED = _env("DPM_S2S_ENABLED", "0") == "1"
S2S_HOST = _env("DPM_S2S_HOST", "0.0.0.0")
S2S_PORT = int(_env("DPM_S2S_PORT", "8765"))
S2S_NUM_PIPELINES = int(_env("DPM_S2S_NUM_PIPELINES", "1"))
S2S_STT_MODEL = _env("DPM_S2S_STT_MODEL", "qwen3-asr-flash-realtime")
S2S_LLM_MODEL = _env("DPM_S2S_LLM_MODEL", "qwen-plus")
S2S_TTS_MODEL = _env("DPM_S2S_TTS_MODEL", "qwen3-tts-flash-realtime")
S2S_TTS_VOICE = _env("DPM_S2S_TTS_VOICE", "Cherry")
S2S_BUILD_RETRIES = int(_env("DPM_S2S_BUILD_RETRIES", "4"))
S2S_BUILD_RETRY_DELAY = float(_env("DPM_S2S_BUILD_RETRY_DELAY", "10"))
S2S_LOG_LEVEL = _env("DPM_S2S_LOG_LEVEL", "INFO").upper()
settings = Settings()
backend/app/main.py
+5
View File
@@ -22,6 +22,7 @@ from .event_bus import bus
from .mqtt import hub
from .routers import router
from .sim_engine import sim_engine
from .s2s_bridge import start_s2s_backend
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
log = logging.getLogger("dpm.main")
@@ -38,6 +39,10 @@ async def lifespan(app: FastAPI):
stop = threading.Event()
# s2s 实时语音栈（可选）：后台线程构建管线池并起 RealtimeServer(:8765)
if settings.S2S_ENABLED:
start_s2s_backend(stop)
def tick_loop():
"""数据引擎每 2.2s 推进一次；MQTT 连接时同步推送快照"""
while not stop.is_set():
backend/app/s2s_bridge.py
+143
View File
@@ -0,0 +1,143 @@
# -*- coding: utf-8 -*-
"""s2s 实时语音栈（VAD 本地 + 云 ASR/LLM/TTS）桥接
在 DPM 进程内以后台线程运行 s2s 的 RealtimeServer（默认 :8765 /v1/realtime）。
- 复用 live-avatar 云化栈（speech-to-speech v0.2.11 + cloud-migration 补丁）。
- 依赖已并入 backend/pyproject.toml；本模块负责按 DPM Settings 组装并启动。
- 前端语音对话页通过 WS 直连 ws://<host>:<S2S_PORT>/v1/realtime。
"""
import logging
import threading
import time
from .config import settings
log = logging.getLogger("dpm.s2s")
def _build_pool(stop_event: threading.Event):
"""按 DPM 配置构建 s2s 管线池（含 VAD/silero 加载 + LLM warmup，耗时约 30~60s）。"""
import sys
from speech_to_speech.api.openai_realtime.websocket_router import create_app
from speech_to_speech.s2s_pipeline import (
_build_realtime_pipeline_unit,
parse_arguments,
prepare_all_args,
)
# 与 start_backend.sh 等价的参数（api_key 留空 → handler 从 env DASHSCOPE_API_KEY 读取）
sys.argv = [
"dpm-s2s",
"--mode", "realtime",
"--stt", "dashscope-asr",
"--llm_backend", "chat-completions",
"--tts", "qwen3-cloud",
"--model_name", settings.S2S_LLM_MODEL,
"--responses_api_base_url", settings.LLM_BASE_URL,
"--responses_api_api_key", settings.DASHSCOPE_API_KEY,
"--dashscope_asr_model", settings.S2S_STT_MODEL,
"--qwen3_cloud_model", settings.S2S_TTS_MODEL,
"--qwen3_cloud_voice", settings.S2S_TTS_VOICE,
"--enable_live_transcription", "false",
"--thresh", "0.8",
"--min_silence_ms", "600",
"--min_speech_ms", "500",
"--log_level", settings.S2S_LOG_LEVEL,
]
args = parse_arguments()
prepare_all_args(
args.module_kwargs,
args.whisper_stt_handler_kwargs,
args.paraformer_stt_handler_kwargs,
args.faster_whisper_stt_handler_kwargs,
args.mlx_audio_whisper_stt_handler_kwargs,
args.parakeet_tdt_stt_handler_kwargs,
args.dashscope_asr_stt_handler_kwargs,
args.language_model_handler_kwargs,
args.responses_api_language_model_handler_kwargs,
args.chat_tts_handler_kwargs,
args.facebook_mms_tts_handler_kwargs,
args.pocket_tts_handler_kwargs,
args.kokoro_tts_handler_kwargs,
args.qwen3_tts_handler_kwargs,
args.qwen3_cloud_tts_handler_kwargs,
)
pool = [
_build_realtime_pipeline_unit(
index=i,
stop_event=stop_event,
module_kwargs=args.module_kwargs,
vad_handler_kwargs=args.vad_handler_kwargs,
whisper_stt_handler_kwargs=args.whisper_stt_handler_kwargs,
faster_whisper_stt_handler_kwargs=args.faster_whisper_stt_handler_kwargs,
paraformer_stt_handler_kwargs=args.paraformer_stt_handler_kwargs,
mlx_audio_whisper_stt_handler_kwargs=args.mlx_audio_whisper_stt_handler_kwargs,
parakeet_tdt_stt_handler_kwargs=args.parakeet_tdt_stt_handler_kwargs,
dashscope_asr_stt_handler_kwargs=args.dashscope_asr_stt_handler_kwargs,
language_model_handler_kwargs=args.language_model_handler_kwargs,
responses_api_language_model_handler_kwargs=args.responses_api_language_model_handler_kwargs,
chat_tts_handler_kwargs=args.chat_tts_handler_kwargs,
facebook_mms_tts_handler_kwargs=args.facebook_mms_tts_handler_kwargs,
pocket_tts_handler_kwargs=args.pocket_tts_handler_kwargs,
kokoro_tts_handler_kwargs=args.kokoro_tts_handler_kwargs,
qwen3_tts_handler_kwargs=args.qwen3_tts_handler_kwargs,
qwen3_cloud_tts_handler_kwargs=args.qwen3_cloud_tts_handler_kwargs,
)
for i in range(settings.S2S_NUM_PIPELINES)
]
app = create_app(pool=pool, stop_event=stop_event)
return app, pool
def start_s2s_backend(stop_event: threading.Event) -> None:
"""后台线程构建管线并启动 RealtimeServer（阻塞调用，放线程里跑）。"""
def _run() -> None:
from speech_to_speech.api.openai_realtime.server import RealtimeServer
# LLM warmup / 云 ASR/TTS 连接在弱网下会瞬断：整个池构建重试几次
last_exc: Exception | None = None
log.info(
"s2s: 启动配置 STT=%s LLM=%s TTS=%s(%s) pipelines=%d ws://%s:%d/v1/realtime retries=%d",
settings.S2S_STT_MODEL, settings.S2S_LLM_MODEL, settings.S2S_TTS_MODEL,
settings.S2S_TTS_VOICE, settings.S2S_NUM_PIPELINES,
settings.S2S_HOST, settings.S2S_PORT, settings.S2S_BUILD_RETRIES,
)
for attempt in range(settings.S2S_BUILD_RETRIES):
try:
log.info(
"s2s: 构建语音管线池（%d 路，尝试 %d/%d，首次约 30~60s）…",
settings.S2S_NUM_PIPELINES,
attempt + 1,
settings.S2S_BUILD_RETRIES,
)
app, pool = _build_pool(stop_event)
server = RealtimeServer(
stop_event=stop_event,
pool=pool,
host=settings.S2S_HOST,
port=settings.S2S_PORT,
)
# 关键：与 s2s_pipeline.main() 的 realtime 分支一致——用 ThreadManager
# 启动 RealtimeServer + 全部 handler 线程（VAD/STT/LLM/TTS）。
# 只跑 server 不启动 handler 线程会导致管线不处理任何音频。
from speech_to_speech.utils.thread_manager import ThreadManager
all_handlers = [server] + [h for unit in pool for h in unit.handlers]
thread_manager = ThreadManager(all_handlers)
thread_manager.start()
log.info("s2s: 实时语音服务 ws://%s:%d/v1/realtime", settings.S2S_HOST, settings.S2S_PORT)
thread_manager.wait()
return
except Exception as e: # noqa: BLE001
last_exc = e
log.warning("s2s: 构建尝试 %d/%d 失败：%s", attempt + 1, settings.S2S_BUILD_RETRIES, e)
if attempt + 1 < settings.S2S_BUILD_RETRIES:
time.sleep(settings.S2S_BUILD_RETRY_DELAY)
log.error("s2s: 启动失败（DPM_S2S_ENABLED=0 可关闭）：%s", last_exc)
threading.Thread(target=_run, daemon=True).start()
backend/pyproject.toml
+19
View File
@@ -12,4 +12,23 @@ dependencies = [
"uvicorn[standard]>=0.29",
"dashscope>=1.26",
"jinja2>=3.1",
# ---- s2s 实时语音栈（VAD本地 + 云ASR/LLM/TTS）----
# speech-to-speech 本体以 --no-deps 安装（路径依赖见下方 [tool.uv.sources]），
# 这里只声明其运行所需的最小依赖，避免 base 依赖把 mlx/transformers 等拖进来。
"httpx>=0.28",
"nltk==3.9.4",
"numpy>=1.26",
"openai==2.28.0",
"scipy>=1.10",
"soundfile>=0.13",
"torch>=2.4",
"transformers>=4.40",
"websockets>=12",
"ultralytics>=8.4.121",
]
[tool.uv.sources]
# 补丁后的 s2s 云化栈，已 vendor 进项目（backend/vendor/s2s-cloud），相对路径，无跨项目依赖
# 说明：speech-to-speech 以 `uv pip install --no-deps --no-build-isolation -e vendor/s2s-cloud`
# 手动安装（避免其 base 依赖把 mlx/transformers 等拖入）；此处仅作路径声明。
speech-to-speech = { path = "vendor/s2s-cloud" }
backend/uv.lock
Generated
+1378
View File
File diff suppressed because it is too large Load Diff