42cab2a6ef
Add pipecat-based backend with WebRTC/WS voice routes, Next.js frontend, and Docker Compose orchestration. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
"""WebRTC 输出:SmallWebRTC 信令握手。
|
|
|
|
参考 dograh 的 webrtc_signaling.py,砍掉鉴权/配额/DB/org/ICE 过滤策略/TURN。
|
|
握手消息:
|
|
client → {type:"offer", payload:{pc_id, sdp, type, config}}
|
|
server → {type:"answer", payload:{pc_id, sdp, type}}
|
|
both → {type:"ice-candidate", payload:{pc_id, candidate:{...}}}
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from db.session import SessionLocal
|
|
from fastapi import APIRouter, WebSocket
|
|
from loguru import logger
|
|
from models import AssistantConfig, SignalingOffer
|
|
from services.config_resolver import resolve_runtime_config
|
|
from starlette.websockets import WebSocketDisconnect, WebSocketState
|
|
|
|
# 注意:pipecat / aiortc 都是重依赖(语音才用),改成函数内"惰性导入",
|
|
# 这样不装 pipecat 也能启动后端、验证 CRUD。语音真正用到时才加载。
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _ice_servers():
|
|
from aiortc import RTCIceServer
|
|
|
|
# 本地只用 STUN;公网部署在此追加 TURN(参考 dograh get_ice_servers)
|
|
return [RTCIceServer(urls="stun:stun.l.google.com:19302")]
|
|
|
|
|
|
@router.websocket("/ws/voice")
|
|
async def voice_signaling(websocket: WebSocket):
|
|
await websocket.accept()
|
|
peers: dict = {}
|
|
try:
|
|
while True:
|
|
message = await websocket.receive_json()
|
|
if message.get("type") == "offer":
|
|
await _handle_offer(websocket, message.get("payload", {}), peers)
|
|
elif message.get("type") == "ice-candidate":
|
|
await _handle_ice(message.get("payload", {}), peers)
|
|
except WebSocketDisconnect:
|
|
logger.info("WebRTC 信令断开")
|
|
except Exception as e:
|
|
logger.error(f"WebRTC 信令出错: {e}")
|
|
finally:
|
|
for pc in peers.values():
|
|
await pc.disconnect()
|
|
|
|
|
|
async def _resolve_config(offer: SignalingOffer) -> AssistantConfig:
|
|
"""优先用 assistant_id 从 DB 解析(含真 key);否则用调试内联配置。"""
|
|
if offer.assistant_id:
|
|
async with SessionLocal() as session:
|
|
return await resolve_runtime_config(session, offer.assistant_id)
|
|
if offer.inline_config:
|
|
return offer.inline_config
|
|
raise ValueError("offer 缺少 assistant_id 或 inline_config")
|
|
|
|
|
|
async def _handle_offer(websocket, payload, peers):
|
|
from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
|
|
from services.pipecat.pipeline import run_pipeline
|
|
from services.pipecat.transports import build_webrtc_transport
|
|
|
|
offer = SignalingOffer(**payload)
|
|
pc_id = offer.pc_id
|
|
|
|
if pc_id and pc_id in peers:
|
|
pc = peers[pc_id]
|
|
await pc.renegotiate(sdp=offer.sdp, type=offer.type, restart_pc=False)
|
|
else:
|
|
cfg = await _resolve_config(offer) # 解析放在建连前,配置错就别建连
|
|
pc = SmallWebRTCConnection(ice_servers=_ice_servers())
|
|
if pc_id:
|
|
pc._pc_id = pc_id
|
|
await pc.initialize(sdp=offer.sdp, type=offer.type)
|
|
peers[pc.pc_id] = pc
|
|
|
|
@pc.event_handler("closed")
|
|
async def _on_closed(conn: SmallWebRTCConnection):
|
|
peers.pop(conn.pc_id, None)
|
|
|
|
# 后台跑管线:WebRTC transport + 解析出的运行时配置
|
|
transport = build_webrtc_transport(pc)
|
|
asyncio.create_task(run_pipeline(transport, cfg))
|
|
|
|
answer = pc.get_answer()
|
|
if websocket.application_state == WebSocketState.CONNECTED:
|
|
await websocket.send_json(
|
|
{
|
|
"type": "answer",
|
|
"payload": {
|
|
"pc_id": answer["pc_id"],
|
|
"sdp": answer["sdp"],
|
|
"type": answer["type"],
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
async def _handle_ice(payload, peers):
|
|
from aiortc.sdp import candidate_from_sdp
|
|
|
|
pc_id = payload.get("pc_id")
|
|
candidate_data = payload.get("candidate")
|
|
pc = peers.get(pc_id) if pc_id else None
|
|
if not pc or not candidate_data:
|
|
return
|
|
try:
|
|
candidate = candidate_from_sdp(candidate_data.get("candidate", ""))
|
|
candidate.sdpMid = candidate_data.get("sdpMid")
|
|
candidate.sdpMLineIndex = candidate_data.get("sdpMLineIndex")
|
|
await pc.add_ice_candidate(candidate)
|
|
except Exception as e:
|
|
logger.error(f"添加 ICE candidate 失败: {e}")
|