From 86639692ba88907e8fe087767fb3e030e2b0a90c Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Tue, 11 Aug 2026 10:05:55 +0800 Subject: [PATCH] feat: implement OpenAI-compatible Realtime API with authentication and management features - Added support for public Realtime API, including new routes for managing API keys and handling WebRTC connections. - Introduced RealtimeApiKey model and associated CRUD operations for admin management of API keys. - Implemented authentication mechanisms for API keys and client secrets. - Enhanced environment configuration with new secrets for Realtime API. - Created OpenAIRealtime session management and event processing for real-time interactions. - Updated schemas and settings to accommodate new features and ensure compatibility with existing systems. --- backend/.env.example | 5 + backend/app.py | 4 + backend/db/models.py | 26 + .../20260810_0018_add_realtime_api_keys.py | 58 ++ backend/routes/openai_realtime.py | 341 +++++++ backend/routes/realtime_api_keys.py | 85 ++ backend/routes/voice_webrtc.py | 140 +-- backend/schemas.py | 38 +- backend/services/openai_realtime/__init__.py | 1 + backend/services/openai_realtime/auth.py | 181 ++++ backend/services/openai_realtime/bridge.py | 916 ++++++++++++++++++ backend/services/openai_realtime/events.py | 115 +++ backend/services/openai_realtime/session.py | 141 +++ backend/services/openai_realtime/webrtc.py | 26 + backend/services/openai_realtime/websocket.py | 51 + backend/services/pipecat/pipeline.py | 115 +++ .../services/pipecat/qwen_audio_realtime.py | 30 +- backend/services/pipecat/stepfun_realtime.py | 45 +- backend/services/pipecat/transports.py | 26 +- backend/services/pipecat/turn_config.py | 15 +- backend/services/realtime/__init__.py | 2 + backend/services/realtime/launcher.py | 195 ++++ backend/services/realtime/lifecycle.py | 109 +++ backend/services/realtime/protocol.py | 42 + backend/settings.py | 8 + docs-developer/docs.json | 2 +- docs-developer/index.mdx | 4 + docs-developer/realtime-protocol.mdx | 693 +++++++++++++ 28 files changed, 3286 insertions(+), 128 deletions(-) create mode 100644 backend/migrations/versions/20260810_0018_add_realtime_api_keys.py create mode 100644 backend/routes/openai_realtime.py create mode 100644 backend/routes/realtime_api_keys.py create mode 100644 backend/services/openai_realtime/__init__.py create mode 100644 backend/services/openai_realtime/auth.py create mode 100644 backend/services/openai_realtime/bridge.py create mode 100644 backend/services/openai_realtime/events.py create mode 100644 backend/services/openai_realtime/session.py create mode 100644 backend/services/openai_realtime/webrtc.py create mode 100644 backend/services/openai_realtime/websocket.py create mode 100644 backend/services/realtime/__init__.py create mode 100644 backend/services/realtime/launcher.py create mode 100644 backend/services/realtime/lifecycle.py create mode 100644 backend/services/realtime/protocol.py create mode 100644 docs-developer/realtime-protocol.mdx diff --git a/backend/.env.example b/backend/.env.example index 1910aff..000cc95 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -14,6 +14,11 @@ PORT=8000 # 前端开发地址,允许跨域(公网部署时加上实际前端 origin) CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000 +# ---- OpenAI-compatible Realtime API ---- +# 用独立随机值签名长期 API Key 摘要和短期 Client Secret；生产环境必须修改。 +REALTIME_TOKEN_SECRET=replace-with-a-long-random-secret +REALTIME_CLIENT_SECRET_TTL_SECONDS=60 + # ---- RustFS / S3-compatible storage ---- # Use 127.0.0.1 (not localhost) if HTTP_PROXY is set — proxies often return 502 for localhost. S3_ENDPOINT_URL=http://127.0.0.1:9000 diff --git a/backend/app.py b/backend/app.py index 3d795b5..50789f0 100644 --- a/backend/app.py +++ b/backend/app.py @@ -44,6 +44,8 @@ from routes import ( mcp_servers, model_registry, node_types, + openai_realtime, + realtime_api_keys, site_settings, test_cases, test_runs, @@ -98,6 +100,8 @@ app.include_router(knowledge_bases.router) app.include_router(mcp_servers.router) app.include_router(model_registry.router) app.include_router(node_types.router) +app.include_router(openai_realtime.router) +app.include_router(realtime_api_keys.router) app.include_router(site_settings.router) app.include_router(test_cases.router) app.include_router(test_runs.router) diff --git a/backend/db/models.py b/backend/db/models.py index 4230dc0..d832003 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -389,6 +389,32 @@ class SiteSetting(Base): ) +class RealtimeApiKey(Base): + """Hashed deployment-level credential for the public Realtime API.""" + + __tablename__ = "realtime_api_keys" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + name: Mapped[str] = mapped_column(String(128)) + key_prefix: Mapped[str] = mapped_column(String(32), unique=True, index=True) + key_hash: Mapped[str] = mapped_column(String(64)) + expires_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + last_used_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + revoked_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + class WebhookDelivery(Base): """Immutable event snapshot plus its asynchronous delivery state.""" diff --git a/backend/migrations/versions/20260810_0018_add_realtime_api_keys.py b/backend/migrations/versions/20260810_0018_add_realtime_api_keys.py new file mode 100644 index 0000000..e57420b --- /dev/null +++ b/backend/migrations/versions/20260810_0018_add_realtime_api_keys.py @@ -0,0 +1,58 @@ +"""add public Realtime API keys + +Revision ID: 20260810_0018 +Revises: 20260810_0017 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260810_0018" +down_revision: str | Sequence[str] | None = "20260810_0017" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "realtime_api_keys", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("name", sa.String(length=128), nullable=False), + sa.Column("key_prefix", sa.String(length=32), nullable=False), + sa.Column("key_hash", sa.String(length=64), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ix_realtime_api_keys_key_prefix", + "realtime_api_keys", + ["key_prefix"], + unique=True, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_realtime_api_keys_key_prefix", + table_name="realtime_api_keys", + ) + op.drop_table("realtime_api_keys") diff --git a/backend/routes/openai_realtime.py b/backend/routes/openai_realtime.py new file mode 100644 index 0000000..382b2b9 --- /dev/null +++ b/backend/routes/openai_realtime.py @@ -0,0 +1,341 @@ +"""Public OpenAI-compatible Realtime HTTP, WebRTC, and WebSocket entries.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from typing import Any +from uuid import uuid4 + +from db.session import SessionLocal +from fastapi import APIRouter, Request, WebSocket +from fastapi.responses import JSONResponse, Response +from loguru import logger +from starlette.websockets import WebSocketState + +from services.openai_realtime.auth import ( + RealtimeAuthError, + RealtimeCredential, + authenticate_bearer, + bearer_from_authorization, + create_client_secret, + hash_safety_identifier, +) +from services.openai_realtime.bridge import OpenAIRealtimeBridge +from services.openai_realtime.events import ( + RealtimeEventError, + assistant_id_from_model, +) +from services.openai_realtime.session import OpenAIRealtimeSession +from services.openai_realtime.webrtc import OpenAIRealtimeWebRTCConnection +from services.openai_realtime.websocket import build_openai_websocket_transport +from services.pipecat.pipeline import run_pipeline +from services.pipecat.transports import build_webrtc_transport +from services.realtime import lifecycle +from services.realtime.launcher import ( + resolve_assistant_config, + validate_runtime_requirements, + validate_visual_runtime, +) +from services.webrtc_ice import aiortc_ice_servers + + +router = APIRouter(prefix="/v1/realtime", tags=["openai-realtime"]) +_webrtc_peers: dict[str, OpenAIRealtimeWebRTCConnection] = {} + + +def _api_error(message: str, *, code: str, status_code: int) -> JSONResponse: + return JSONResponse( + status_code=status_code, + content={ + "error": { + "type": "invalid_request_error", + "code": code, + "message": message, + "param": None, + } + }, + ) + + +async def _credential(request: Request) -> RealtimeCredential: + token = bearer_from_authorization(request.headers.get("authorization")) + async with SessionLocal() as db: + return await authenticate_bearer(db, token) + + +def _session_payload(value: object) -> dict[str, Any]: + if not isinstance(value, dict): + raise RealtimeEventError("session must be an object", param="session") + session = value.get("session", value) + if not isinstance(session, dict): + raise RealtimeEventError("session must be an object", param="session") + return session + + +def _reject_locked_creation_fields(session: dict[str, Any]) -> None: + locked = {"instructions", "voice", "tools", "tool_choice"}.intersection(session) + audio = session.get("audio") + if isinstance(audio, dict) and isinstance(audio.get("output"), dict): + if "voice" in audio["output"]: + locked.add("audio.output.voice") + if locked: + field = sorted(locked)[0] + raise RealtimeEventError( + f"{field} is owned by the selected assistant", + code="immutable_session_field", + param=f"session.{field}", + ) + modalities = session.get("output_modalities") + if modalities is not None and modalities not in (["audio"], ["text"]): + raise RealtimeEventError( + 'output_modalities must be ["audio"] or ["text"]', + param="session.output_modalities", + ) + + +async def _build_session( + credential: RealtimeCredential, + requested: dict[str, Any] | None, + *, + safety_identifier_hash: str | None, +) -> OpenAIRealtimeSession: + session_value = dict(credential.session or requested or {}) + _reject_locked_creation_fields(session_value) + requested_assistant_id = assistant_id_from_model(session_value.get("model")) + if credential.assistant_id and requested_assistant_id != credential.assistant_id: + raise RealtimeEventError( + "Client secret is bound to another assistant", + code="invalid_model", + param="session.model", + ) + config = await resolve_assistant_config(requested_assistant_id) + validate_runtime_requirements(config) + vision_enabled = validate_visual_runtime(config) + state = OpenAIRealtimeSession( + assistant_id=requested_assistant_id, + config=config, + vision_enabled=vision_enabled, + safety_identifier_hash=( + credential.safety_identifier_hash or safety_identifier_hash + ), + ) + state.apply_initial_options(session_value) + return state + + +@router.post("/client_secrets") +async def create_realtime_client_secret(request: Request): + try: + credential = await _credential(request) + if credential.ephemeral: + raise RealtimeAuthError("A long-lived API key is required") + body = await request.json() + session = _session_payload(body) + _reject_locked_creation_fields(session) + assistant_id = assistant_id_from_model(session.get("model")) + # Resolve now so invalid models/configurations fail before a browser gets a token. + config = await resolve_assistant_config(assistant_id) + validate_runtime_requirements(config) + validate_visual_runtime(config) + safety_hash = hash_safety_identifier( + request.headers.get("openai-safety-identifier") + ) + value, expires_at = create_client_secret( + api_key_id=credential.api_key_id, + assistant_id=assistant_id, + session=session, + safety_identifier_hash=safety_hash, + ) + return { + "value": value, + "expires_at": expires_at, + "session": session, + } + except RealtimeAuthError as exc: + return _api_error(str(exc), code="invalid_api_key", status_code=401) + except RealtimeEventError as exc: + return _api_error(str(exc), code=exc.code, status_code=400) + except ValueError as exc: + code = "invalid_model" if "助手不存在" in str(exc) else "invalid_session" + return _api_error(str(exc), code=code, status_code=400) + + +@router.post("/calls") +async def create_realtime_call(request: Request): + try: + credential = await _credential(request) + content_type = request.headers.get("content-type", "").lower() + if content_type.startswith("multipart/form-data"): + if credential.ephemeral: + raise RealtimeAuthError( + "Client secrets must send an application/sdp offer" + ) + form = await request.form() + sdp = str(form.get("sdp") or "") + raw_session = str(form.get("session") or "{}") + requested = _session_payload(json.loads(raw_session)) + elif content_type.startswith("application/sdp"): + if not credential.ephemeral: + raise RealtimeAuthError( + "application/sdp requires a short-lived client secret" + ) + sdp = (await request.body()).decode("utf-8") + requested = None + else: + raise RealtimeEventError( + "Use multipart/form-data or application/sdp", + code="unsupported_content_type", + ) + if not sdp.strip(): + raise RealtimeEventError("SDP offer is empty", param="sdp") + state = await _build_session( + credential, + requested, + safety_identifier_hash=hash_safety_identifier( + request.headers.get("openai-safety-identifier") + ), + ) + answer = await _start_webrtc(sdp, state) + return Response( + content=answer, + media_type="application/sdp", + headers={"Location": f"/v1/realtime/calls/{state.id}"}, + ) + except RealtimeAuthError as exc: + return _api_error(str(exc), code="invalid_api_key", status_code=401) + except (RealtimeEventError, json.JSONDecodeError, UnicodeDecodeError) as exc: + code = exc.code if isinstance(exc, RealtimeEventError) else "invalid_request_error" + return _api_error(str(exc), code=code, status_code=400) + except ValueError as exc: + code = "invalid_model" if "助手不存在" in str(exc) else "invalid_session" + return _api_error(str(exc), code=code, status_code=400) + except Exception as exc: # noqa: BLE001 - no internal details in public response + logger.exception(f"OpenAI Realtime WebRTC 启动失败: {exc}") + return _api_error( + "Realtime connection could not be established", + code="connection_error", + status_code=500, + ) + + +async def _start_webrtc(sdp: str, state: OpenAIRealtimeSession) -> str: + connection = OpenAIRealtimeWebRTCConnection( + ice_servers=aiortc_ice_servers() + ) + await connection.initialize(sdp=sdp, type="offer") + _webrtc_peers[connection.pc_id] = connection + bridge = OpenAIRealtimeBridge(state, channel="webrtc") + transport = build_webrtc_transport( + connection, + video_in_enabled=state.vision_enabled, + ) + task = lifecycle.start_pipeline_task( + connection, + run_pipeline( + transport, + state.config, + vision_enabled=state.vision_enabled, + assistant_id=state.assistant_id, + channel="openai-webrtc", + protocol_adapter=bridge, + ), + protocol="openai-webrtc", + ) + + @connection.event_handler("closed") + async def on_closed(conn: OpenAIRealtimeWebRTCConnection): + _webrtc_peers.pop(conn.pc_id, None) + lifecycle.active_connections.discard(conn) + await lifecycle.wait_for_pipeline_close( + task, + connection_id=conn.pc_id, + ) + + answer = connection.get_answer() + if not answer: + raise RuntimeError("WebRTC answer was not created") + return str(answer["sdp"]) + + +def _websocket_token(websocket: WebSocket) -> tuple[str, str | None]: + authorization = websocket.headers.get("authorization") + if authorization: + return bearer_from_authorization(authorization), None + protocols = [ + item.strip() + for item in websocket.headers.get("sec-websocket-protocol", "").split(",") + if item.strip() + ] + for protocol in protocols: + prefix = "openai-insecure-api-key." + if protocol.startswith(prefix): + return protocol.removeprefix(prefix), "realtime" if "realtime" in protocols else None + raise RealtimeAuthError("Missing Bearer credential") + + +@dataclass(eq=False) +class _ManagedWebSocket: + websocket: WebSocket + pc_id: str = field(default_factory=lambda: f"ws_{uuid4().hex}") + + async def disconnect(self) -> None: + if self.websocket.application_state == WebSocketState.CONNECTED: + await self.websocket.close(code=1001) + + +@router.websocket("") +async def realtime_websocket(websocket: WebSocket): + managed: _ManagedWebSocket | None = None + task = None + try: + token, accepted_subprotocol = _websocket_token(websocket) + async with SessionLocal() as db: + credential = await authenticate_bearer(db, token) + requested = ( + credential.session + if credential.ephemeral + else {"model": websocket.query_params.get("model")} + ) + state = await _build_session( + credential, + requested, + safety_identifier_hash=hash_safety_identifier( + websocket.headers.get("openai-safety-identifier") + ), + ) + await websocket.accept(subprotocol=accepted_subprotocol) + managed = _ManagedWebSocket(websocket) + transport = build_openai_websocket_transport(websocket) + bridge = OpenAIRealtimeBridge(state, channel="websocket") + task = lifecycle.start_pipeline_task( + managed, + run_pipeline( + transport, + state.config, + vision_enabled=False, + assistant_id=state.assistant_id, + channel="openai-websocket", + protocol_adapter=bridge, + ), + protocol="openai-websocket", + ) + await task + except (RealtimeAuthError, RealtimeEventError, ValueError) as exc: + logger.warning(f"拒绝 OpenAI Realtime WebSocket: {exc}") + if websocket.application_state == WebSocketState.CONNECTED: + await websocket.close(code=1008, reason=str(exc)[:120]) + else: + await websocket.close(code=1008) + except Exception as exc: # noqa: BLE001 - pipeline callback logs full exception + logger.warning(f"OpenAI Realtime WebSocket 已关闭: {type(exc).__name__}") + if websocket.application_state == WebSocketState.CONNECTED: + await websocket.close(code=1011) + finally: + if managed: + lifecycle.active_connections.discard(managed) + if task and not task.done(): + await lifecycle.wait_for_pipeline_close( + task, + connection_id=managed.pc_id if managed else "websocket", + ) diff --git a/backend/routes/realtime_api_keys.py b/backend/routes/realtime_api_keys.py new file mode 100644 index 0000000..776ba55 --- /dev/null +++ b/backend/routes/realtime_api_keys.py @@ -0,0 +1,85 @@ +"""Admin-only management of public Realtime API credentials.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from db.models import RealtimeApiKey +from db.session import get_session +from fastapi import APIRouter, Depends, HTTPException, Response, status +from schemas import RealtimeApiKeyCreate, RealtimeApiKeyCreated, RealtimeApiKeyOut +from services.auth import require_admin +from services.openai_realtime.auth import create_api_key_value +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + + +router = APIRouter( + prefix="/api/realtime/api-keys", + tags=["realtime-api-keys"], + dependencies=[Depends(require_admin)], +) + + +def _status(row: RealtimeApiKey) -> str: + if row.revoked_at is not None: + return "revoked" + if row.expires_at is not None and row.expires_at <= datetime.now(UTC): + return "expired" + return "active" + + +def _out(row: RealtimeApiKey) -> RealtimeApiKeyOut: + return RealtimeApiKeyOut( + id=row.id, + name=row.name, + key_prefix=row.key_prefix, + status=_status(row), + expires_at=row.expires_at, + last_used_at=row.last_used_at, + revoked_at=row.revoked_at, + created_at=row.created_at, + ) + + +@router.post("", response_model=RealtimeApiKeyCreated, status_code=status.HTTP_201_CREATED) +async def create_realtime_api_key( + body: RealtimeApiKeyCreate, + session: AsyncSession = Depends(get_session), +): + key_id, key, prefix, digest = create_api_key_value() + row = RealtimeApiKey( + id=key_id, + name=body.name, + key_prefix=prefix, + key_hash=digest, + expires_at=body.expires_at, + ) + session.add(row) + await session.commit() + await session.refresh(row) + return RealtimeApiKeyCreated(**_out(row).model_dump(), key=key) + + +@router.get("", response_model=list[RealtimeApiKeyOut]) +async def list_realtime_api_keys(session: AsyncSession = Depends(get_session)): + rows = ( + await session.execute( + select(RealtimeApiKey).order_by(RealtimeApiKey.created_at.desc()) + ) + ).scalars().all() + return [_out(row) for row in rows] + + +@router.delete("/{key_id}", status_code=status.HTTP_204_NO_CONTENT) +async def revoke_realtime_api_key( + key_id: str, + session: AsyncSession = Depends(get_session), +): + row = await session.get(RealtimeApiKey, key_id) + if row is None: + raise HTTPException(404, "Realtime API Key 不存在") + if row.revoked_at is None: + row.revoked_at = datetime.now(UTC) + await session.commit() + return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/backend/routes/voice_webrtc.py b/backend/routes/voice_webrtc.py index 729353c..b9a5b3c 100644 --- a/backend/routes/voice_webrtc.py +++ b/backend/routes/voice_webrtc.py @@ -8,18 +8,20 @@ server → {type:"error", payload:{message}} """ -import asyncio import base64 import json from collections.abc import Coroutine from typing import Any -from db.session import SessionLocal from fastapi import APIRouter, Body, Depends, Request, WebSocket from loguru import logger from models import AssistantConfig, SignalingOffer from services.auth import require_admin, require_admin_websocket -from services.config_resolver import resolve_runtime_config +from services.realtime import lifecycle as realtime_lifecycle +from services.realtime.launcher import ( + resolve_assistant_config, + validate_visual_runtime, +) from services.runtime_variables import DynamicVariableError, prepare_dynamic_config from starlette.websockets import WebSocketDisconnect, WebSocketState @@ -29,116 +31,42 @@ from services.webrtc_ice import aiortc_ice_servers, client_ice_servers router = APIRouter(tags=["voice"]) _http_peers: dict[str, object] = {} -_active_connections: set[object] = set() -_pipeline_tasks: set[asyncio.Task[None]] = set() -_connection_tasks: dict[object, asyncio.Task[None]] = {} +_active_connections = realtime_lifecycle.active_connections +_pipeline_tasks = realtime_lifecycle.pipeline_tasks +_connection_tasks = realtime_lifecycle.connection_tasks PIPELINE_CLOSE_GRACE_SECONDS = 10.0 -def _consume_pipeline_result(task: asyncio.Task[None], connection: object) -> None: - """Retrieve background exceptions and release the task's strong reference.""" - _pipeline_tasks.discard(task) - if _connection_tasks.get(connection) is task: - _connection_tasks.pop(connection, None) - try: - error = task.exception() - except asyncio.CancelledError: - logger.info(f"WebRTC pipeline 已取消: task={task.get_name()}") - return - if error is not None: - logger.opt(exception=error).error( - f"WebRTC pipeline 异常结束: task={task.get_name()}" - ) - - def _start_pipeline_task( connection: object, coroutine: Coroutine[Any, Any, None], -) -> asyncio.Task[None]: - """Start one strongly referenced pipeline task for a WebRTC connection.""" - connection_id = str(getattr(connection, "pc_id", "unknown")) - task = asyncio.create_task( +) -> object: + """Compatibility proxy for existing RTVI callers and tests.""" + return realtime_lifecycle.start_pipeline_task( + connection, coroutine, - name=f"webrtc-pipeline:{connection_id}", + protocol="webrtc", ) - _active_connections.add(connection) - _pipeline_tasks.add(task) - _connection_tasks[connection] = task - task.add_done_callback( - lambda completed, connection=connection: _consume_pipeline_result( - completed, - connection, - ) - ) - return task async def _wait_for_pipeline_close( - task: asyncio.Task[None] | None, + task, *, connection_id: str, ) -> None: - """Let transport disconnect finish normally, then cancel a stuck pipeline.""" - if task is None: - return - try: - await asyncio.wait_for( - asyncio.shield(task), - timeout=PIPELINE_CLOSE_GRACE_SECONDS, - ) - return - except TimeoutError: - logger.warning( - f"WebRTC pipeline 关闭超过 {PIPELINE_CLOSE_GRACE_SECONDS:g} 秒，" - f"执行取消: pc_id={connection_id}" - ) - except asyncio.CancelledError: - raise - except Exception: - # The done callback owns exception reporting and retrieval. - return - - task.cancel() - done, _pending = await asyncio.wait( - {task}, + await realtime_lifecycle.wait_for_pipeline_close( + task, + connection_id=connection_id, timeout=PIPELINE_CLOSE_GRACE_SECONDS, ) - if not done: - logger.error(f"WebRTC pipeline 取消后仍未退出: pc_id={connection_id}") async def shutdown_active_sessions() -> None: - """Close active peers and drain every managed pipeline during app shutdown.""" - connections = list(_active_connections) - if connections: - await asyncio.gather( - *(connection.disconnect() for connection in connections), - return_exceptions=True, - ) - _active_connections.difference_update(connections) - _http_peers.clear() - - tasks = list(_pipeline_tasks) - if not tasks: - return - done, pending = await asyncio.wait( - tasks, + """Compatibility proxy; now drains RTVI and OpenAI Realtime sessions.""" + _http_peers.clear() + await realtime_lifecycle.shutdown_active_sessions( timeout=PIPELINE_CLOSE_GRACE_SECONDS, ) - for task in pending: - task.cancel() - if pending: - cancelled, stuck = await asyncio.wait( - pending, - timeout=PIPELINE_CLOSE_GRACE_SECONDS, - ) - if stuck: - logger.error(f"应用关闭时仍有 {len(stuck)} 个 WebRTC pipeline 未退出") - else: - cancelled = set() - logger.info( - f"WebRTC 会话清理完成: normal={len(done)} cancelled={len(cancelled)}" - ) @router.get("/api/webrtc/ice-servers", dependencies=[Depends(require_admin)]) @@ -229,12 +157,9 @@ async def voice_signaling(websocket: WebSocket): async def _resolve_config(offer: SignalingOffer) -> AssistantConfig: """优先用 assistant_id 从 DB 解析(含真 key);否则用调试内联配置。""" if offer.assistant_id: - async with SessionLocal() as session: - cfg = await resolve_runtime_config(session, offer.assistant_id) - return prepare_dynamic_config( - cfg, - offer.dynamic_variables, - assistant_id=offer.assistant_id, + return await resolve_assistant_config( + offer.assistant_id, + dynamic_variables=offer.dynamic_variables, ) if offer.inline_config: return prepare_dynamic_config( @@ -280,24 +205,7 @@ async def _handle_offer_payload(payload, peers): else: cfg = await _resolve_config(offer) # 解析放在建连前,配置错就别建连 # 服务端助手配置是视觉理解的唯一授权来源；客户端 offer 只负责携带媒体轨。 - if cfg.type == "workflow": - from services.workflow_engine import WorkflowEngine - - vision_enabled = WorkflowEngine(cfg.graph).uses_vision() - else: - vision_enabled = cfg.vision_enabled - if vision_enabled and cfg.type != "workflow": - has_native_vision = ( - not cfg.vision_model_resource_id and cfg.llm_support_image_input - ) - has_aux_vision_model = ( - bool(cfg.vision_model_resource_id) - and cfg.vision_llm_support_image_input - ) - if not (has_native_vision or has_aux_vision_model): - raise ValueError( - "当前模型不支持图片输入,请在模型资源中选择支持图片输入的视觉模型" - ) + vision_enabled = validate_visual_runtime(cfg) pc = SmallWebRTCConnection(ice_servers=aiortc_ice_servers()) if pc_id: pc._pc_id = pc_id diff --git a/backend/schemas.py b/backend/schemas.py index 96e43ee..1aca5cd 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -7,7 +7,7 @@ JSON 用 camelCase(modelId/interfaceType/apiUrl/apiKey),Python 内部用 snake_c from __future__ import annotations -from datetime import datetime +from datetime import UTC, datetime import re from typing import Annotated, Any, Literal, Union @@ -601,3 +601,39 @@ class AnalysisWebhookTestOut(CamelModel): latency_ms: int | None = None message: str payload: dict[str, Any] + + +# ---------- Realtime API keys ---------- +class RealtimeApiKeyCreate(CamelModel): + name: str = Field(min_length=1, max_length=128) + expires_at: datetime | None = None + + @field_validator("name", mode="before") + @classmethod + def normalize_name(cls, value: object) -> object: + return value.strip() if isinstance(value, str) else value + + @field_validator("expires_at") + @classmethod + def validate_expiry(cls, value: datetime | None) -> datetime | None: + if value is None: + return None + normalized = value if value.tzinfo else value.replace(tzinfo=UTC) + if normalized <= datetime.now(UTC): + raise ValueError("expires_at 必须晚于当前时间") + return normalized + + +class RealtimeApiKeyOut(CamelModel): + id: str + name: str + key_prefix: str + status: Literal["active", "expired", "revoked"] + expires_at: datetime | None = None + last_used_at: datetime | None = None + revoked_at: datetime | None = None + created_at: datetime + + +class RealtimeApiKeyCreated(RealtimeApiKeyOut): + key: str diff --git a/backend/services/openai_realtime/__init__.py b/backend/services/openai_realtime/__init__.py new file mode 100644 index 0000000..c70d952 --- /dev/null +++ b/backend/services/openai_realtime/__init__.py @@ -0,0 +1 @@ +"""OpenAI Realtime-compatible northbound protocol adapter.""" diff --git a/backend/services/openai_realtime/auth.py b/backend/services/openai_realtime/auth.py new file mode 100644 index 0000000..ad3d337 --- /dev/null +++ b/backend/services/openai_realtime/auth.py @@ -0,0 +1,181 @@ +"""Authentication for the public OpenAI-compatible Realtime API.""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import secrets +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any +from uuid import uuid4 + +import settings +from db.models import RealtimeApiKey +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + + +API_KEY_PREFIX = "sk-rt-" +CLIENT_SECRET_PREFIX = "ek-rt-" + + +class RealtimeAuthError(ValueError): + """Raised when a public Realtime credential cannot be accepted.""" + + +@dataclass(frozen=True) +class RealtimeCredential: + api_key_id: str + assistant_id: str | None = None + session: dict[str, Any] | None = None + safety_identifier_hash: str | None = None + ephemeral: bool = False + + +def _b64encode(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64decode(value: str) -> bytes: + padding = "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(f"{value}{padding}".encode("ascii")) + + +def _peppered_digest(value: str) -> str: + return hmac.new( + settings.REALTIME_TOKEN_SECRET.encode("utf-8"), + value.encode("utf-8"), + hashlib.sha256, + ).hexdigest() + + +def _token_signature(encoded_payload: str) -> str: + digest = hmac.new( + settings.REALTIME_TOKEN_SECRET.encode("utf-8"), + encoded_payload.encode("ascii"), + hashlib.sha256, + ).digest() + return _b64encode(digest) + + +def hash_safety_identifier(value: str | None) -> str | None: + normalized = (value or "").strip() + if not normalized: + return None + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def create_api_key_value() -> tuple[str, str, str, str]: + key_id = f"rtkey_{uuid4().hex}" + random_secret = secrets.token_urlsafe(32) + value = f"{API_KEY_PREFIX}{key_id[6:18]}.{random_secret}" + return key_id, value, value[:24], _peppered_digest(value) + + +def create_client_secret( + *, + api_key_id: str, + assistant_id: str, + session: dict[str, Any], + safety_identifier_hash: str | None, +) -> tuple[str, int]: + now = int(time.time()) + expires_at = now + settings.REALTIME_CLIENT_SECRET_TTL_SECONDS + payload = { + "sub": api_key_id, + "assistant_id": assistant_id, + "session": session, + "safety_identifier_hash": safety_identifier_hash, + "iat": now, + "exp": expires_at, + "jti": f"rtcs_{uuid4().hex}", + } + encoded = _b64encode( + json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode( + "utf-8" + ) + ) + return f"{CLIENT_SECRET_PREFIX}{encoded}.{_token_signature(encoded)}", expires_at + + +async def _active_api_key( + session: AsyncSession, + *, + key_id: str | None = None, + key_prefix: str | None = None, +) -> RealtimeApiKey | None: + if key_id: + row = await session.get(RealtimeApiKey, key_id) + elif key_prefix: + row = ( + await session.execute( + select(RealtimeApiKey).where( + RealtimeApiKey.key_prefix == key_prefix + ) + ) + ).scalar_one_or_none() + else: + return None + now = datetime.now(UTC) + if row is None or row.revoked_at is not None: + return None + if row.expires_at is not None and row.expires_at <= now: + return None + return row + + +async def authenticate_bearer( + session: AsyncSession, + token: str, +) -> RealtimeCredential: + if token.startswith(API_KEY_PREFIX): + prefix = token[:24] + row = await _active_api_key(session, key_prefix=prefix) + if row is None or not hmac.compare_digest( + row.key_hash, + _peppered_digest(token), + ): + raise RealtimeAuthError("Invalid or expired API key") + row.last_used_at = datetime.now(UTC) + await session.commit() + return RealtimeCredential(api_key_id=row.id) + + if not token.startswith(CLIENT_SECRET_PREFIX): + raise RealtimeAuthError("Unsupported Realtime credential") + compact = token[len(CLIENT_SECRET_PREFIX) :] + try: + encoded, signature = compact.rsplit(".", 1) + except ValueError as exc: + raise RealtimeAuthError("Invalid client secret") from exc + if not hmac.compare_digest(_token_signature(encoded), signature): + raise RealtimeAuthError("Invalid client secret") + try: + payload = json.loads(_b64decode(encoded)) + except (ValueError, json.JSONDecodeError) as exc: + raise RealtimeAuthError("Invalid client secret") from exc + if int(payload.get("exp", 0)) < int(time.time()): + raise RealtimeAuthError("Client secret expired") + api_key_id = str(payload.get("sub") or "") + row = await _active_api_key(session, key_id=api_key_id) + if row is None: + raise RealtimeAuthError("Parent API key is no longer active") + row.last_used_at = datetime.now(UTC) + await session.commit() + return RealtimeCredential( + api_key_id=api_key_id, + assistant_id=str(payload.get("assistant_id") or "") or None, + session=(payload.get("session") if isinstance(payload.get("session"), dict) else None), + safety_identifier_hash=str(payload.get("safety_identifier_hash") or "") or None, + ephemeral=True, + ) + + +def bearer_from_authorization(value: str | None) -> str: + scheme, _, token = (value or "").partition(" ") + if scheme.lower() != "bearer" or not token.strip(): + raise RealtimeAuthError("Missing Bearer credential") + return token.strip() diff --git a/backend/services/openai_realtime/bridge.py b/backend/services/openai_realtime/bridge.py new file mode 100644 index 0000000..d9e6aac --- /dev/null +++ b/backend/services/openai_realtime/bridge.py @@ -0,0 +1,916 @@ +"""Translate OpenAI Realtime events to the project's neutral pipeline messages.""" + +from __future__ import annotations + +import asyncio +import base64 +import binascii +import json +from typing import Any +from uuid import uuid4 + +from pipecat.audio.utils import create_stream_resampler +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + InputAudioRawFrame, + InputTransportMessageFrame, + LLMContextFrame, + OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, + StartFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + VADParamsUpdateFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +from services.input_assets import store_input_image +from services.openai_realtime.events import ( + SUPPORTED_CAPABILITIES, + RealtimeEventError, + error_event, + normalize_turn_detection, + require_event, +) +from services.openai_realtime.session import OpenAIRealtimeSession +from services.pipecat.turn_config import create_vad_params +from services.realtime.protocol import PipelineProtocolRuntime + + +MAX_BUFFERED_AUDIO_BYTES = 24000 * 2 * 120 + + +def _event_id() -> str: + return f"event_{uuid4().hex}" + + +def _server_event(event_type: str, **payload: Any) -> dict[str, Any]: + return {"type": event_type, "event_id": _event_id(), **payload} + + +def _decode_base64(value: object, *, param: str) -> bytes: + if not isinstance(value, str) or not value: + raise RealtimeEventError(f"{param} must be non-empty Base64", param=param) + try: + return base64.b64decode(value, validate=True) + except (ValueError, binascii.Error) as exc: + raise RealtimeEventError(f"{param} is not valid Base64", param=param) from exc + + +def _decode_data_url(value: object) -> bytes: + if not isinstance(value, str) or not value.startswith("data:image/"): + raise RealtimeEventError( + "input_image only supports image Data URLs", + code="unsupported_content_type", + param="item.content", + ) + header, separator, encoded = value.partition(",") + if not separator or ";base64" not in header: + raise RealtimeEventError( + "input_image Data URL must use Base64 encoding", + param="item.content", + ) + return _decode_base64(encoded, param="item.content") + + +class OpenAIRealtimeInputProcessor(FrameProcessor): + def __init__(self, bridge: "OpenAIRealtimeBridge") -> None: + super().__init__() + self._bridge = bridge + self._started = False + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StartFrame) and not self._started: + self._started = True + # Initialize every downstream processor/serializer before the first + # business event is sent to the client. + await self.push_frame(frame, direction) + await self._bridge.emit( + _server_event( + "session.created", + session=self._bridge.session.public_value(), + ) + ) + return + + if isinstance(frame, InputAudioRawFrame): + if ( + self._bridge.session.external_turn_control + and frame.transport_source != "openai-committed" + ): + try: + self._bridge.buffer_audio( + frame.audio, + sample_rate=frame.sample_rate, + num_channels=frame.num_channels, + ) + except RealtimeEventError as exc: + await self._bridge.emit(error_event(exc)) + return + await self.push_frame(frame, direction) + return + + if not isinstance(frame, InputTransportMessageFrame): + await self.push_frame(frame, direction) + return + if isinstance(frame.message, dict) and frame.message.pop( + "_pipeline_internal", False + ): + await self.push_frame(frame, direction) + return + if not self._bridge.client_ready_sent: + self._bridge.client_ready_sent = True + await self.push_frame( + InputTransportMessageFrame( + message={"type": "client-ready"} + ), + direction, + ) + try: + await self._bridge.handle_client_event(require_event(frame.message)) + except RealtimeEventError as exc: + await self._bridge.emit(error_event(exc)) + except Exception as exc: # noqa: BLE001 - event errors must stay connection-local + client_event_id = ( + str(frame.message.get("event_id") or "") or None + if isinstance(frame.message, dict) + else None + ) + await self._bridge.emit( + error_event( + RealtimeEventError( + f"Event processing failed: {exc}", + code="event_processing_error", + event_id=client_event_id, + ) + ) + ) + + +class OpenAIRealtimeOutputProcessor(FrameProcessor): + def __init__(self, bridge: "OpenAIRealtimeBridge") -> None: + super().__init__() + self._bridge = bridge + self._audio_resampler = create_stream_resampler() + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, BotStartedSpeakingFrame): + if not self._bridge.session.active_response_id: + for event in self._bridge._start_assistant_output(): + await self.push_frame( + OutputTransportMessageUrgentFrame(message=event) + ) + await self.push_frame(frame, direction) + return + if isinstance(frame, BotStoppedSpeakingFrame): + if self._bridge.session.active_response_id: + for event in self._bridge._end_assistant_output(False): + await self.push_frame( + OutputTransportMessageUrgentFrame(message=event) + ) + await self.push_frame(frame, direction) + return + if isinstance(frame, UserStartedSpeakingFrame): + item_id = f"item_{uuid4().hex}" + self._bridge.session.active_input_item_id = item_id + await self.push_frame( + OutputTransportMessageUrgentFrame( + message=_server_event( + "input_audio_buffer.speech_started", + audio_start_ms=0, + item_id=item_id, + ) + ) + ) + await self.push_frame(frame, direction) + return + if isinstance(frame, UserStoppedSpeakingFrame): + item_id = ( + self._bridge.session.active_input_item_id + or f"item_{uuid4().hex}" + ) + self._bridge.session.active_input_item_id = None + await self.push_frame( + OutputTransportMessageUrgentFrame( + message=_server_event( + "input_audio_buffer.speech_stopped", + audio_end_ms=0, + item_id=item_id, + ) + ) + ) + await self.push_frame(frame, direction) + return + if isinstance(frame, OutputAudioRawFrame): + if not self._bridge.session.output_is_audio: + return + if self._bridge.channel == "websocket": + if not self._bridge.session.active_response_id: + for event in self._bridge._start_assistant_output(): + await self.push_frame( + OutputTransportMessageUrgentFrame(message=event) + ) + response_id, item_id = self._bridge.session.begin_response() + audio = frame.audio + if frame.sample_rate != 24000: + audio = await self._audio_resampler.resample( + audio, + frame.sample_rate, + 24000, + ) + await self.push_frame( + OutputTransportMessageUrgentFrame( + message=_server_event( + "response.output_audio.delta", + response_id=response_id, + item_id=item_id, + output_index=0, + content_index=0, + delta=base64.b64encode(audio).decode("ascii"), + ) + ) + ) + return + await self.push_frame(frame, direction) + return + + if not isinstance( + frame, + (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame), + ): + await self.push_frame(frame, direction) + return + message = frame.message + if not isinstance(message, dict): + return + translated = self._bridge.translate_server_message(message) + for event in translated: + await self.push_frame( + OutputTransportMessageUrgentFrame(message=event), + direction, + ) + + +class OpenAIResponseGateProcessor(FrameProcessor): + """Hold automatic inference when create_response is disabled or PTT is active.""" + + def __init__(self, bridge: "OpenAIRealtimeBridge") -> None: + super().__init__() + self._bridge = bridge + self._held: list[tuple[Any, FrameDirection]] = [] + self._allow_next = False + + 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 + turn = self._bridge.session.turn_detection + auto_response = bool(turn and turn.get("create_response", True)) + if auto_response or self._allow_next: + self._allow_next = False + await self.push_frame(frame, direction) + return + self._held.append((frame, direction)) + + async def allow_one_response(self) -> None: + if self._held: + held, self._held = self._held, [] + for frame, direction in held: + await self.push_frame(frame, direction) + return + self._allow_next = True + + +class OpenAIRealtimeBridge: + """One adapter instance is owned by exactly one public Realtime session.""" + + def __init__(self, session: OpenAIRealtimeSession, *, channel: str) -> None: + self.session = session + self.channel = channel + self._runtime: PipelineProtocolRuntime | None = None + self.client_ready_sent = False + self._input = OpenAIRealtimeInputProcessor(self) + self._inference = OpenAIResponseGateProcessor(self) + self._output = OpenAIRealtimeOutputProcessor(self) + + def input_processors(self) -> list[FrameProcessor]: + return [self._input] + + def output_processors(self) -> list[FrameProcessor]: + return [self._output] + + def inference_processors(self) -> list[FrameProcessor]: + return [self._inference] + + async def bind(self, runtime: PipelineProtocolRuntime) -> None: + self._runtime = runtime + await runtime.set_external_turn_control(self.session.external_turn_control) + turn = self.session.turn_detection + if turn: + await runtime.set_response_interruption( + bool(turn.get("interrupt_response", True)) + ) + + @property + def runtime(self) -> PipelineProtocolRuntime: + if self._runtime is None: + raise RuntimeError("OpenAI Realtime bridge is not bound to a pipeline") + return self._runtime + + async def emit(self, event: dict[str, Any]) -> None: + await self._input.push_frame( + OutputTransportMessageUrgentFrame(message=event) + ) + + async def queue_internal_message(self, message: dict[str, Any]) -> None: + await self.runtime.queue_frame( + InputTransportMessageFrame( + message={**message, "_pipeline_internal": True} + ) + ) + + def buffer_audio( + self, + audio: bytes, + *, + sample_rate: int, + num_channels: int, + ) -> None: + if self.session.buffered_audio_bytes + len(audio) > MAX_BUFFERED_AUDIO_BYTES: + self.clear_buffered_audio() + raise RealtimeEventError( + "input_audio_buffer exceeds the 120 second MVP limit", + code="input_audio_buffer_too_large", + ) + self.session.audio_chunks.append((audio, sample_rate, num_channels)) + self.session.buffered_audio_bytes += len(audio) + + def clear_buffered_audio(self) -> None: + self.session.audio_chunks.clear() + self.session.buffered_audio_bytes = 0 + self.session.audio_committed = False + + async def handle_client_event(self, event: dict[str, Any]) -> None: + event_type = str(event["type"]) + if event_type == "session.update": + await self._update_session(event) + elif event_type == "conversation.item.create": + await self._create_item(event) + elif event_type == "conversation.item.truncate": + await self.runtime.cancel_response() + await self.emit( + _server_event( + "conversation.item.truncated", + item_id=event.get("item_id"), + content_index=int(event.get("content_index") or 0), + audio_end_ms=int(event.get("audio_end_ms") or 0), + ) + ) + elif event_type == "input_audio_buffer.append": + await self._append_audio(event) + elif event_type == "input_audio_buffer.commit": + await self._commit_audio() + elif event_type == "input_audio_buffer.clear": + self.clear_buffered_audio() + await self.runtime.clear_audio() + await self.emit(_server_event("input_audio_buffer.cleared")) + elif event_type == "output_audio_buffer.clear": + await self.runtime.cancel_response() + await self._finish_response(status="cancelled") + await self.emit(_server_event("output_audio_buffer.cleared")) + elif event_type == "response.create": + await self._create_response(event) + elif event_type == "response.cancel": + await self.runtime.cancel_response() + await self._finish_response(status="cancelled") + elif event_type == "x.interactive_media.capabilities.update": + await self._update_capabilities(event) + elif event_type == "x.interactive_media.session.variables.update": + await self._update_variables(event) + + async def _update_session(self, event: dict[str, Any]) -> None: + update = event.get("session") + if not isinstance(update, dict): + raise RealtimeEventError("session.update requires session", param="session") + locked = {"model", "instructions", "voice", "tools", "tool_choice"} + changed_locked = sorted(locked.intersection(update)) + audio = update.get("audio") + if isinstance(audio, dict): + output = audio.get("output") + if isinstance(output, dict) and "voice" in output: + changed_locked.append("audio.output.voice") + if changed_locked: + raise RealtimeEventError( + f"Assistant-owned session fields cannot be changed: {', '.join(changed_locked)}", + code="immutable_session_field", + param=changed_locked[0], + event_id=str(event.get("event_id") or "") or None, + ) + + modalities = update.get("output_modalities") + if modalities is not None: + if modalities not in (["audio"], ["text"]): + raise RealtimeEventError( + 'output_modalities must be ["audio"] or ["text"]', + param="session.output_modalities", + ) + self.session.output_modalities = list(modalities) + + marker = object() + turn_detection: object = marker + if isinstance(audio, dict) and isinstance(audio.get("input"), dict): + turn_detection = audio["input"].get("turn_detection", marker) + if turn_detection is marker: + turn_detection = update.get("turn_detection", marker) + if turn_detection is not marker: + self.session.turn_detection = self._validate_turn_detection(turn_detection) + await self.runtime.set_external_turn_control( + self.session.external_turn_control + ) + if self.session.turn_detection: + await self.runtime.set_response_interruption( + bool(self.session.turn_detection.get("interrupt_response", True)) + ) + if self.session.turn_detection: + turn = self.session.turn_detection + config = { + "vad": { + "confidence": turn["threshold"], + "start_secs": max(0.05, turn["prefix_padding_ms"] / 1000), + "stop_secs": 0.2, + }, + "turn_detection": { + "strategy": "silence", + "silence_timeout_secs": turn["silence_duration_ms"] / 1000, + }, + } + await self.runtime.queue_frame( + VADParamsUpdateFrame(params=create_vad_params(config)) + ) + await self.emit( + _server_event("session.updated", session=self.session.public_value()) + ) + + def _validate_turn_detection(self, value: object) -> dict[str, Any] | None: + return normalize_turn_detection(value) + + async def _create_item(self, event: dict[str, Any]) -> None: + item = event.get("item") + if not isinstance(item, dict): + raise RealtimeEventError("conversation.item.create requires item", param="item") + item_type = item.get("type") + if item_type == "function_call_output": + call_id = str(item.get("call_id") or "") + if not call_id: + raise RealtimeEventError("function_call_output requires call_id", param="item.call_id") + raw_output = item.get("output") + try: + data = json.loads(raw_output) if isinstance(raw_output, str) else raw_output + except json.JSONDecodeError: + data = raw_output + await self.queue_internal_message( + { + "type": "client-tool-result", + "tool_call_id": call_id, + "status": "ok", + "data": data, + } + ) + await self._emit_item_ack(item) + return + if item_type != "message" or item.get("role") != "user": + raise RealtimeEventError( + "Only user messages and function_call_output items are accepted", + code="unsupported_item_type", + param="item.type", + ) + + wire_parts: list[dict[str, Any]] = [] + for part in item.get("content") or []: + if not isinstance(part, dict): + continue + if part.get("type") == "input_text": + text = str(part.get("text") or "").strip() + if text: + wire_parts.append({"type": "input_text", "text": text}) + elif part.get("type") == "input_image": + if self.session.config.runtimeMode == "realtime": + raise RealtimeEventError( + "input_image is not supported by the assistant's realtime runtime", + code="unsupported_content_type", + ) + if not self.session.vision_enabled: + raise RealtimeEventError( + "This assistant has not enabled image input", + code="unsupported_content_type", + ) + data = _decode_data_url(part.get("image_url")) + stored = await asyncio.to_thread(store_input_image, data) + wire_parts.append( + { + "type": "input_image", + "source": { + "type": "uploaded_asset", + "asset_token": stored.token, + }, + } + ) + else: + raise RealtimeEventError( + f"Unsupported content part: {part.get('type')}", + code="unsupported_content_type", + param="item.content", + ) + if not wire_parts: + raise RealtimeEventError("User item has no supported content", param="item.content") + item_id = str(item.get("id") or f"item_{uuid4().hex}") + public_item = {**item, "id": item_id, "status": "completed"} + self.session.pending_item = { + "type": "user-input", + "schema_version": 1, + "input_id": item_id, + "parts": wire_parts, + "options": { + "run_immediately": self.session.config.runtimeMode != "realtime", + "interrupt": True, + }, + } + await self._emit_item_ack(public_item) + + async def _emit_item_ack(self, item: dict[str, Any]) -> None: + await self.emit( + _server_event( + "conversation.item.added", + previous_item_id=None, + item=item, + ) + ) + await self.emit(_server_event("conversation.item.done", item=item)) + + async def _append_audio(self, event: dict[str, Any]) -> None: + audio = _decode_base64(event.get("audio"), param="audio") + if len(audio) % 2: + raise RealtimeEventError( + "WebSocket audio must contain complete PCM16 samples", + code="invalid_audio_format", + param="audio", + ) + if self.channel != "websocket": + raise RealtimeEventError( + "input_audio_buffer.append is only used by WebSocket audio", + code="invalid_event", + ) + if self.session.external_turn_control: + self.buffer_audio(audio, sample_rate=24000, num_channels=1) + return + await self.runtime.queue_frame( + InputAudioRawFrame(audio=audio, sample_rate=24000, num_channels=1) + ) + + async def _commit_audio(self) -> None: + if not self.session.audio_chunks and self.session.external_turn_control: + raise RealtimeEventError( + "input_audio_buffer is empty", + code="input_audio_buffer_commit_empty", + ) + self.session.audio_committed = True + item_id = f"item_{uuid4().hex}" + await self.emit( + _server_event("input_audio_buffer.committed", item_id=item_id) + ) + + async def _create_response(self, event: dict[str, Any]) -> None: + response = event.get("response") + if isinstance(response, dict): + forbidden = {"instructions", "voice", "tools", "tool_choice", "model"} + changed = forbidden.intersection(response) + if changed: + raise RealtimeEventError( + "Per-response assistant configuration is locked", + code="immutable_session_field", + param=f"response.{sorted(changed)[0]}", + ) + if ( + self.session.external_turn_control + and self.session.audio_chunks + and not self.session.audio_committed + ): + raise RealtimeEventError( + "Commit input_audio_buffer before response.create", + code="input_audio_buffer_not_committed", + ) + await self._inference.allow_one_response() + submitted_item = bool(self.session.pending_item) + if self.session.pending_item: + pending = self.session.pending_item + self.session.pending_item = None + await self.queue_internal_message(pending) + if self.session.external_turn_control and self.session.audio_chunks: + await self.runtime.queue_frame(UserStartedSpeakingFrame()) + for audio, sample_rate, num_channels in self.session.audio_chunks: + await self.runtime.queue_frame( + InputAudioRawFrame( + audio=audio, + sample_rate=sample_rate, + num_channels=num_channels, + transport_source="openai-committed", + ) + ) + await self.runtime.queue_frame(UserStoppedSpeakingFrame()) + await self.runtime.commit_audio() + self.clear_buffered_audio() + if not submitted_item or self.session.config.runtimeMode == "realtime": + await self.runtime.request_response() + + async def _finish_response(self, *, status: str) -> None: + response_id, _item_id, text = self.session.finish_response() + if not response_id: + return + await self.emit( + _server_event( + "response.done", + response={ + "id": response_id, + "object": "realtime.response", + "status": status, + "output": [], + "output_text": text, + }, + ) + ) + + async def _update_capabilities(self, event: dict[str, Any]) -> None: + requested = event.get("capabilities") or [] + if not isinstance(requested, list) or not all(isinstance(x, str) for x in requested): + raise RealtimeEventError("capabilities must be a string array", param="capabilities") + enabled = set(requested).intersection(SUPPORTED_CAPABILITIES) + if "video_track" in enabled and not self.session.vision_enabled: + enabled.remove("video_track") + self.session.capabilities = enabled + await self.emit( + _server_event( + "x.interactive_media.capabilities.updated", + capabilities=sorted(enabled), + rejected=sorted(set(requested) - enabled), + ) + ) + + async def _update_variables(self, event: dict[str, Any]) -> None: + if "dynamic_variables" not in self.session.capabilities: + raise RealtimeEventError( + "dynamic_variables capability was not negotiated", + code="extension_not_negotiated", + ) + variables = event.get("variables") + if not isinstance(variables, dict) or not variables: + raise RealtimeEventError("variables must be a non-empty object", param="variables") + await self.queue_internal_message( + { + "type": "session-update", + "schema_version": 1, + "update_id": str(event.get("event_id") or f"update_{uuid4().hex}"), + "dynamic_variables": variables, + } + ) + + def translate_server_message(self, message: dict[str, Any]) -> list[dict[str, Any]]: + message_type = str(message.get("type") or "") + if message_type == "error": + return [message] + if "." in message_type: + return [message] + if message_type == "transcript": + return self._translate_transcript(message) + if message_type == "assistant-text-start": + return self._start_assistant_output() + if message_type == "assistant-text-delta": + return self._assistant_delta(str(message.get("delta") or "")) + if message_type == "assistant-text-end": + return self._end_assistant_output(bool(message.get("interrupted"))) + if message_type == "client-tool-call": + return self._client_tool_call(message) + if message_type == "user-input-result" and message.get("status") == "error": + return [ + error_event( + RealtimeEventError( + str(message.get("message") or "User input failed"), + code="input_error", + ) + ) + ] + extension = self._extension_event(message) + return [extension] if extension else [] + + def _translate_transcript(self, message: dict[str, Any]) -> list[dict[str, Any]]: + role = message.get("role") + text = str(message.get("content") or "") + item_id = f"item_{uuid4().hex}" + if role == "user": + return [ + _server_event( + "conversation.item.input_audio_transcription.completed", + item_id=item_id, + content_index=0, + transcript=text, + ) + ] + response_id, output_item_id = self.session.begin_response() + item = { + "id": output_item_id, + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": text}], + } + self.session.response_text += text + response = { + "id": response_id, + "object": "realtime.response", + "status": "completed", + "output": [item], + } + self.session.finish_response() + return [ + _server_event("response.created", response={**response, "status": "in_progress", "output": []}), + _server_event("response.output_item.added", response_id=response_id, output_index=0, item=item), + _server_event("response.output_item.done", response_id=response_id, output_index=0, item=item), + _server_event("response.done", response=response), + ] + + def _start_assistant_output(self) -> list[dict[str, Any]]: + if self.session.active_response_id: + return [] + response_id, item_id = self.session.begin_response() + content_type = "audio" if self.session.output_is_audio else "text" + item = { + "id": item_id, + "type": "message", + "role": "assistant", + "status": "in_progress", + "content": [{"type": content_type}], + } + return [ + _server_event( + "response.created", + response={ + "id": response_id, + "object": "realtime.response", + "status": "in_progress", + "output": [], + }, + ), + _server_event( + "response.output_item.added", + response_id=response_id, + output_index=0, + item=item, + ), + _server_event( + "response.content_part.added", + response_id=response_id, + item_id=item_id, + output_index=0, + content_index=0, + part={"type": content_type}, + ), + ] + + def _assistant_delta(self, delta: str) -> list[dict[str, Any]]: + events = self._start_assistant_output() + response_id, item_id = self.session.begin_response() + self.session.response_text += delta + event_type = ( + "response.output_audio_transcript.delta" + if self.session.output_is_audio + else "response.output_text.delta" + ) + events.append( + _server_event( + event_type, + response_id=response_id, + item_id=item_id, + output_index=0, + content_index=0, + delta=delta, + ) + ) + return events + + def _end_assistant_output(self, interrupted: bool) -> list[dict[str, Any]]: + response_id, item_id, text = self.session.finish_response() + if not response_id or not item_id: + return [] + content_type = "audio" if self.session.output_is_audio else "text" + done_type = ( + "response.output_audio_transcript.done" + if self.session.output_is_audio + else "response.output_text.done" + ) + item = { + "id": item_id, + "type": "message", + "role": "assistant", + "status": "incomplete" if interrupted else "completed", + "content": [{"type": content_type, "transcript" if self.session.output_is_audio else "text": text}], + } + status = "cancelled" if interrupted else "completed" + events = [ + _server_event(done_type, response_id=response_id, item_id=item_id, output_index=0, content_index=0, transcript=text, text=text), + _server_event("response.content_part.done", response_id=response_id, item_id=item_id, output_index=0, content_index=0, part=item["content"][0]), + _server_event("response.output_item.done", response_id=response_id, output_index=0, item=item), + _server_event( + "response.done", + response={ + "id": response_id, + "object": "realtime.response", + "status": status, + "output": [item], + }, + ), + ] + if self.session.output_is_audio: + events.insert( + 1, + _server_event( + "response.output_audio.done", + response_id=response_id, + item_id=item_id, + output_index=0, + content_index=0, + ), + ) + return events + + def _client_tool_call(self, message: dict[str, Any]) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + if not self.session.active_response_id: + response_id, _unused_item_id = self.session.begin_response() + events.append( + _server_event( + "response.created", + response={ + "id": response_id, + "object": "realtime.response", + "status": "in_progress", + "output": [], + }, + ) + ) + else: + response_id = self.session.active_response_id + item_id = f"item_{uuid4().hex}" + call_id = str(message.get("tool_call_id") or "") + arguments = json.dumps(message.get("arguments") or {}, ensure_ascii=False) + item = { + "id": item_id, + "type": "function_call", + "status": "completed", + "name": str(message.get("function_name") or ""), + "call_id": call_id, + "arguments": arguments, + } + events.extend([ + _server_event("response.output_item.added", response_id=response_id, output_index=0, item=item), + _server_event("response.function_call_arguments.delta", response_id=response_id, item_id=item_id, output_index=0, call_id=call_id, delta=arguments), + _server_event("response.function_call_arguments.done", response_id=response_id, item_id=item_id, output_index=0, call_id=call_id, name=item["name"], arguments=arguments), + _server_event("response.output_item.done", response_id=response_id, output_index=0, item=item), + ]) + self.session.active_output_item_id = None + return events + + def _extension_event(self, message: dict[str, Any]) -> dict[str, Any] | None: + kind = str(message.get("type") or "") + if kind == "session-update-result" and "dynamic_variables" in self.session.capabilities: + return _server_event( + "x.interactive_media.session.variables.updated", + update_id=message.get("update_id"), + status=message.get("status"), + message=message.get("message"), + ) + if kind in {"node-active", "workflow-event", "workflow-variables", "workflow-error"}: + if "workflow_events" not in self.session.capabilities: + return None + names = { + "node-active": "x.interactive_media.workflow.node_active", + "workflow-event": "x.interactive_media.workflow.event", + "workflow-variables": "x.interactive_media.workflow.variables.updated", + "workflow-error": "x.interactive_media.workflow.error", + } + return _server_event(names[kind], **{k: v for k, v in message.items() if k != "type"}) + if kind in {"handoff-requested", "call-ended"}: + if "handoff" not in self.session.capabilities: + return None + event_type = ( + "x.interactive_media.call.handoff_requested" + if kind == "handoff-requested" + else "x.interactive_media.call.ended" + ) + return _server_event(event_type, **{k: v for k, v in message.items() if k != "type"}) + return None diff --git a/backend/services/openai_realtime/events.py b/backend/services/openai_realtime/events.py new file mode 100644 index 0000000..50e5a44 --- /dev/null +++ b/backend/services/openai_realtime/events.py @@ -0,0 +1,115 @@ +"""Small, explicit validation helpers for the supported Realtime event subset.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any +from uuid import uuid4 + + +SUPPORTED_CLIENT_EVENTS = { + "session.update", + "conversation.item.create", + "conversation.item.truncate", + "input_audio_buffer.append", + "input_audio_buffer.commit", + "input_audio_buffer.clear", + "output_audio_buffer.clear", + "response.create", + "response.cancel", + "x.interactive_media.capabilities.update", + "x.interactive_media.session.variables.update", +} + +SUPPORTED_CAPABILITIES = { + "dynamic_variables", + "workflow_events", + "handoff", + "video_track", +} + + +@dataclass +class RealtimeEventError(ValueError): + message: str + code: str = "invalid_request_error" + param: str | None = None + event_id: str | None = None + + def __str__(self) -> str: + return self.message + + +def require_event(message: object) -> dict[str, Any]: + if not isinstance(message, dict): + raise RealtimeEventError("Event must be a JSON object") + event_type = str(message.get("type") or "") + if event_type not in SUPPORTED_CLIENT_EVENTS: + raise RealtimeEventError( + f"Unsupported client event: {event_type or ''}", + code="invalid_event", + param="type", + event_id=str(message.get("event_id") or "") or None, + ) + return message + + +def assistant_id_from_model(model: object) -> str: + value = str(model or "").strip() + if not value.startswith("assistant:") or not value.removeprefix("assistant:"): + raise RealtimeEventError( + "model must use assistant:asst_xxx", + code="invalid_model", + param="session.model", + ) + return value.removeprefix("assistant:") + + +def normalize_turn_detection(value: object) -> dict[str, Any] | None: + if value is None: + return None + if not isinstance(value, dict) or value.get("type") != "server_vad": + raise RealtimeEventError( + "turn_detection must be server_vad or null", + param="session.audio.input.turn_detection", + ) + try: + threshold = float(value.get("threshold", 0.7)) + prefix_ms = int(value.get("prefix_padding_ms", 200)) + silence_ms = int(value.get("silence_duration_ms", 600)) + except (TypeError, ValueError) as exc: + raise RealtimeEventError( + "Invalid server_vad threshold, padding, or silence duration", + param="session.audio.input.turn_detection", + ) from exc + if ( + not 0 <= threshold <= 1 + or not 0 <= prefix_ms <= 5000 + or not 100 <= silence_ms <= 10000 + ): + raise RealtimeEventError( + "Invalid server_vad threshold, padding, or silence duration", + param="session.audio.input.turn_detection", + ) + return { + "type": "server_vad", + "threshold": threshold, + "prefix_padding_ms": prefix_ms, + "silence_duration_ms": silence_ms, + "create_response": bool(value.get("create_response", True)), + "interrupt_response": bool(value.get("interrupt_response", True)), + } + + +def error_event(error: RealtimeEventError) -> dict[str, Any]: + return { + "type": "error", + "event_id": f"event_{uuid4().hex}", + "error": { + "type": "invalid_request_error", + "code": error.code, + "message": error.message, + "param": error.param, + "event_id": error.event_id, + }, + } diff --git a/backend/services/openai_realtime/session.py b/backend/services/openai_realtime/session.py new file mode 100644 index 0000000..ea59b24 --- /dev/null +++ b/backend/services/openai_realtime/session.py @@ -0,0 +1,141 @@ +"""Connection-local state for the OpenAI-compatible Realtime wire protocol.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any +from uuid import uuid4 + +from models import AssistantConfig +from services.openai_realtime.events import normalize_turn_detection + + +def _id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex}" + + +@dataclass +class OpenAIRealtimeSession: + assistant_id: str + config: AssistantConfig + vision_enabled: bool + safety_identifier_hash: str | None = None + id: str = field(default_factory=lambda: _id("sess")) + output_modalities: list[str] = field(default_factory=lambda: ["audio"]) + turn_detection: dict[str, Any] | None = field( + default_factory=lambda: { + "type": "server_vad", + "threshold": 0.7, + "prefix_padding_ms": 200, + "silence_duration_ms": 600, + "create_response": True, + "interrupt_response": True, + } + ) + capabilities: set[str] = field(default_factory=set) + pending_item: dict[str, Any] | None = None + audio_chunks: list[tuple[bytes, int, int]] = field(default_factory=list) + buffered_audio_bytes: int = 0 + audio_committed: bool = False + active_response_id: str | None = None + active_output_item_id: str | None = None + active_input_item_id: str | None = None + response_text: str = "" + + @property + def model(self) -> str: + return f"assistant:{self.assistant_id}" + + @property + def external_turn_control(self) -> bool: + return self.turn_detection is None + + @property + def output_is_audio(self) -> bool: + return self.output_modalities == ["audio"] + + def client_tools(self) -> list[dict[str, Any]]: + result = [] + for tool in self.config.tools: + if tool.type != "client": + continue + definition = tool.definition or {} + parameters = (definition.get("config") or {}).get("parameters") or [] + properties: dict[str, Any] = {} + required: list[str] = [] + for parameter in parameters: + name = str(parameter.get("name") or "") + if not name: + continue + properties[name] = { + "type": parameter.get("type") or "string", + **( + {"description": parameter["description"]} + if parameter.get("description") + else {} + ), + } + if parameter.get("required"): + required.append(name) + result.append( + { + "type": "function", + "name": tool.function_name, + "description": tool.description, + "parameters": { + "type": "object", + "properties": properties, + "required": required, + }, + } + ) + return result + + def public_value(self) -> dict[str, Any]: + return { + "id": self.id, + "type": "realtime", + "model": self.model, + "output_modalities": self.output_modalities, + "audio": { + "input": { + "format": {"type": "audio/pcm", "rate": 24000}, + "turn_detection": self.turn_detection, + }, + "output": { + "format": {"type": "audio/pcm", "rate": 24000}, + }, + }, + "tools": self.client_tools(), + "tool_choice": "auto", + } + + def apply_initial_options(self, value: dict[str, Any]) -> None: + modalities = value.get("output_modalities") + if modalities in (["audio"], ["text"]): + self.output_modalities = list(modalities) + audio = value.get("audio") + if isinstance(audio, dict) and isinstance(audio.get("input"), dict): + turn_detection = audio["input"].get("turn_detection", self.turn_detection) + else: + turn_detection = value.get("turn_detection", self.turn_detection) + self.turn_detection = normalize_turn_detection(turn_detection) + + def begin_response(self) -> tuple[str, str]: + if not self.active_response_id: + self.active_response_id = _id("resp") + if not self.active_output_item_id: + self.active_output_item_id = _id("item") + self.response_text = "" + return self.active_response_id, str(self.active_output_item_id) + + def finish_response(self) -> tuple[str | None, str | None, str]: + result = ( + self.active_response_id, + self.active_output_item_id, + self.response_text, + ) + self.active_response_id = None + self.active_output_item_id = None + self.response_text = "" + return result diff --git a/backend/services/openai_realtime/webrtc.py b/backend/services/openai_realtime/webrtc.py new file mode 100644 index 0000000..92692ef --- /dev/null +++ b/backend/services/openai_realtime/webrtc.py @@ -0,0 +1,26 @@ +"""Project-local SmallWebRTC specialization for the OpenAI event channel.""" + +from __future__ import annotations + +from typing import Any + +from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + + +class OpenAIRealtimeWebRTCConnection(SmallWebRTCConnection): + """Prevent Pipecat's private signalling envelope from leaking to clients.""" + + data_channel_label = "oai-events" + + def _setup_listeners(self): + super()._setup_listeners() + + @self._pc.on("datachannel") + def require_openai_event_channel(channel): + if channel.label != self.data_channel_label: + channel.close() + + def send_app_message(self, message: Any): + if isinstance(message, dict) and message.get("type") == "signalling": + return + super().send_app_message(message) diff --git a/backend/services/openai_realtime/websocket.py b/backend/services/openai_realtime/websocket.py new file mode 100644 index 0000000..89b5d00 --- /dev/null +++ b/backend/services/openai_realtime/websocket.py @@ -0,0 +1,51 @@ +"""UTF-8 JSON serializer and transport builder for public Realtime WebSockets.""" + +from __future__ import annotations + +import json +from typing import Any + +from fastapi import WebSocket +from pipecat.frames.frames import ( + Frame, + InputTransportMessageFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, +) +from pipecat.serializers.base_serializer import FrameSerializer + +from services.pipecat.transports import build_serialized_ws_transport + + +class OpenAIRealtimeJSONSerializer(FrameSerializer): + async def serialize(self, frame: Frame) -> str | bytes | None: + if self.should_ignore_frame(frame): + return None + if isinstance( + frame, + (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame), + ) and isinstance(frame.message, dict): + return json.dumps(frame.message, ensure_ascii=False, separators=(",", ":")) + return None + + async def deserialize(self, data: str | bytes) -> Frame | None: + if isinstance(data, bytes): + message: Any = { + "type": "_invalid_binary_frame", + "event_id": None, + } + else: + try: + message = json.loads(data) + except json.JSONDecodeError: + message = {"type": "_invalid_json", "event_id": None} + return InputTransportMessageFrame(message=message) + + +def build_openai_websocket_transport(websocket: WebSocket): + return build_serialized_ws_transport( + websocket, + serializer=OpenAIRealtimeJSONSerializer(), + sample_rate=24000, + ) + diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 1cd9e41..71f4d41 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -47,6 +47,7 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.flows import FlowsFunctionSchema from pipecat.frames.frames import ( EndFrame, + InterruptionFrame, LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, UserImageRawFrame, @@ -94,6 +95,11 @@ from services.pipecat.pipeline_events import ( bind_cascade_pipeline_events, bind_realtime_pipeline_events, ) +from services.realtime.protocol import ( + PipelineProtocolAdapter, + PipelineProtocolRuntime, + RealtimeProviderControlFrame, +) from pipecat.workers.runner import WorkerRunner @@ -201,6 +207,7 @@ async def run_pipeline( vision_enabled: bool = False, assistant_id: str | None = None, channel: str = "webrtc", + protocol_adapter: PipelineProtocolAdapter | None = None, ) -> None: """在给定 transport 上构建并运行管线,直到连接结束。 @@ -234,6 +241,7 @@ async def run_pipeline( vision_enabled=vision_enabled, assistant_id=assistant_id, channel=channel, + protocol_adapter=protocol_adapter, ) return @@ -626,15 +634,22 @@ async def run_pipeline( analysis_plan=cfg.analysis_config, extra=(workflow_engine.session_metadata() if workflow_engine else None), ) + protocol_inputs = protocol_adapter.input_processors() if protocol_adapter else [] + protocol_inference = ( + protocol_adapter.inference_processors() if protocol_adapter else [] + ) + protocol_outputs = protocol_adapter.output_processors() if protocol_adapter else [] pipeline = Pipeline( [ transport.input(), + *protocol_inputs, client_tools, session_update, vision_capture, user_input, stt_processor, user_aggregator, + *protocol_inference, user_turn_router, knowledge_retrieval, llm, @@ -645,6 +660,7 @@ async def run_pipeline( tts_processor, EndCallAfterSpeechProcessor(call_end), ConversationHistoryProcessor(recorder), + *protocol_outputs, transport.output(), ] ) @@ -657,6 +673,47 @@ async def run_pipeline( enable_rtvi=False, ) worker_holder["worker"] = worker + protocol_turn_state: dict[str, bool | None] = { + "external": False, + "interrupt_response": None, + } + if protocol_adapter: + + async def set_external_turn_control(enabled: bool) -> None: + protocol_turn_state["external"] = enabled + await user_aggregator.apply_external_turn_control(enabled) + + async def request_response() -> None: + # Cascade inference is triggered when the adapter submits user-input. + return None + + async def commit_audio() -> None: + return None + + async def clear_audio() -> None: + return None + + async def set_response_interruption(enabled: bool) -> None: + protocol_turn_state["interrupt_response"] = enabled + await user_aggregator.apply_turn_strategies( + cfg.turnConfig, + enable_interruptions=enabled, + ) + + async def cancel_response() -> None: + await worker.queue_frame(InterruptionFrame()) + + await protocol_adapter.bind( + PipelineProtocolRuntime( + queue_frame=worker.queue_frame, + set_external_turn_control=set_external_turn_control, + set_response_interruption=set_response_interruption, + commit_audio=commit_audio, + clear_audio=clear_audio, + request_response=request_response, + cancel_response=cancel_response, + ) + ) service_controller = WorkflowServiceController( worker=worker, llm_services=llm_services, @@ -685,6 +742,15 @@ async def run_pipeline( normalized, enable_interruptions=enable_interrupt, ) + if protocol_turn_state["external"]: + await user_aggregator.apply_external_turn_control(True) + elif protocol_turn_state["interrupt_response"] is not None: + await user_aggregator.apply_turn_strategies( + normalized, + enable_interruptions=bool( + protocol_turn_state["interrupt_response"] + ), + ) await worker.queue_frame( VADParamsUpdateFrame(params=create_vad_params(normalized)) ) @@ -872,6 +938,7 @@ async def run_realtime_pipeline( vision_enabled: bool = False, assistant_id: str | None = None, channel: str = "webrtc", + protocol_adapter: PipelineProtocolAdapter | None = None, ) -> None: """Run a speech-to-speech model that owns ASR, reasoning, and synthesis.""" realtime = create_realtime_service( @@ -960,18 +1027,26 @@ async def run_realtime_pipeline( else None ), ) + protocol_inputs = protocol_adapter.input_processors() if protocol_adapter else [] + protocol_inference = ( + protocol_adapter.inference_processors() if protocol_adapter else [] + ) + protocol_outputs = protocol_adapter.output_processors() if protocol_adapter else [] pipeline = Pipeline( [ transport.input(), + *protocol_inputs, vision_capture, client_tools, session_update, user_input, input_gate, + *protocol_inference, realtime, dynamic_variables, EndCallAfterSpeechProcessor(call_end), ConversationHistoryProcessor(recorder), + *protocol_outputs, transport.output(), ] ) @@ -985,6 +1060,46 @@ async def run_realtime_pipeline( enable_rtvi=False, ) worker_holder["worker"] = worker + if protocol_adapter: + + async def set_external_turn_control(enabled: bool) -> None: + update = getattr(realtime, "update_turn_detection", None) + if callable(update): + await update(None if enabled else cfg.turnConfig) + + async def request_response() -> None: + await worker.queue_frame( + RealtimeProviderControlFrame(action="request_response") + ) + + async def commit_audio() -> None: + await worker.queue_frame( + RealtimeProviderControlFrame(action="commit_audio") + ) + + async def clear_audio() -> None: + await worker.queue_frame( + RealtimeProviderControlFrame(action="clear_audio") + ) + + async def cancel_response() -> None: + await realtime.interrupt() + + async def set_response_interruption(_enabled: bool) -> None: + # Provider-side settings vary; explicit response.cancel remains portable. + return None + + await protocol_adapter.bind( + PipelineProtocolRuntime( + queue_frame=worker.queue_frame, + set_external_turn_control=set_external_turn_control, + set_response_interruption=set_response_interruption, + commit_audio=commit_audio, + clear_audio=clear_audio, + request_response=request_response, + cancel_response=cancel_response, + ) + ) def set_input_enabled(enabled: bool) -> None: input_state["enabled"] = enabled diff --git a/backend/services/pipecat/qwen_audio_realtime.py b/backend/services/pipecat/qwen_audio_realtime.py index 88cdb1c..a082550 100644 --- a/backend/services/pipecat/qwen_audio_realtime.py +++ b/backend/services/pipecat/qwen_audio_realtime.py @@ -41,13 +41,14 @@ from services.pipecat.realtime_tools import ( RealtimeToolDispatcher, RealtimeToolSession, ) +from services.realtime.protocol import RealtimeProviderControlFrame DEFAULT_QWEN_AUDIO_REALTIME_MODEL = "qwen-audio-3.0-realtime-flash" DEFAULT_QWEN_AUDIO_REALTIME_VOICE = "longanqian" QWEN_INPUT_SAMPLE_RATE = 16_000 QWEN_OUTPUT_SAMPLE_RATE = 24_000 -SUPPORTED_TURN_DETECTION_MODES = frozenset({"server_vad", "smart_turn"}) +SUPPORTED_TURN_DETECTION_MODES = frozenset({"none", "server_vad", "smart_turn"}) ExtraEventHandler = Callable[[dict[str, Any]], Awaitable[None] | None] SpeechStartedHandler = Callable[[], Awaitable[None]] @@ -161,6 +162,15 @@ class QwenAudioRealtimeService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await super().process_frame(frame, direction) + if isinstance(frame, RealtimeProviderControlFrame): + if frame.action == "commit_audio": + await self.commit_audio_buffer() + elif frame.action == "clear_audio": + await self.clear_audio_buffer() + else: + await self.request_response() + return + if isinstance(frame, InputAudioRawFrame): if ( frame.sample_rate != self._input_sample_rate @@ -234,6 +244,20 @@ class QwenAudioRealtimeService(AIService): ) await self._send_event({"type": "response.create"}) + async def commit_audio_buffer(self) -> None: + await self._send_event({"type": "input_audio_buffer.commit"}) + + async def clear_audio_buffer(self) -> None: + await self._send_event({"type": "input_audio_buffer.clear"}) + + async def update_turn_detection(self, config: dict[str, Any] | None) -> None: + """Apply OpenAI PTT before provider startup; reject unsafe live changes.""" + + mode = "none" if config is None else "server_vad" + if self._session_ready.is_set(): + raise ValueError("Qwen Realtime 不支持会话建立后切换轮次检测") + self._turn_detection_mode = mode + async def wait_for_response_boundary(self) -> None: """Wait until Qwen has finished or cancelled the current response.""" await self._response_done.wait() @@ -355,7 +379,9 @@ class QwenAudioRealtimeService(AIService): "tool_choice": "auto", } - def _turn_detection_config(self) -> dict[str, Any]: + def _turn_detection_config(self) -> dict[str, Any] | None: + if self._turn_detection_mode == "none": + return None if self._turn_detection_mode == "smart_turn": return {"type": "smart_turn"} return { diff --git a/backend/services/pipecat/stepfun_realtime.py b/backend/services/pipecat/stepfun_realtime.py index d65c58b..83431f7 100644 --- a/backend/services/pipecat/stepfun_realtime.py +++ b/backend/services/pipecat/stepfun_realtime.py @@ -35,6 +35,7 @@ from services.pipecat.realtime_tools import ( RealtimeToolDispatcher, RealtimeToolSession, ) +from services.realtime.protocol import RealtimeProviderControlFrame DEFAULT_STEPFUN_REALTIME_URL = "wss://api.stepfun.com/v1/realtime" SpeechStartedHandler = Callable[[], Awaitable[None]] @@ -69,6 +70,7 @@ class StepFunRealtimeService(AIService): self._prefix_padding_ms = prefix_padding_ms self._silence_duration_ms = silence_duration_ms self._energy_awakeness_threshold = energy_awakeness_threshold + self._turn_detection_enabled = True self._warned_input_sample_rate = False self._websocket = None self._receive_task: asyncio.Task | None = None @@ -115,6 +117,15 @@ class StepFunRealtimeService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await super().process_frame(frame, direction) + if isinstance(frame, RealtimeProviderControlFrame): + if frame.action == "commit_audio": + await self.commit_audio_buffer() + elif frame.action == "clear_audio": + await self.clear_audio_buffer() + else: + await self.request_response() + return + if isinstance(frame, InputAudioRawFrame): if ( frame.sample_rate != self._input_sample_rate @@ -183,6 +194,23 @@ class StepFunRealtimeService(AIService): ) await self._send_event({"type": "response.create"}) + async def commit_audio_buffer(self) -> None: + await self._send_event({"type": "input_audio_buffer.commit"}) + + async def clear_audio_buffer(self) -> None: + await self._send_event({"type": "input_audio_buffer.clear"}) + + async def update_turn_detection(self, config: dict[str, Any] | None) -> None: + self._turn_detection_enabled = config is not None + if self._session_ready.is_set(): + await self._send_event( + { + "type": "session.update", + "session": {"turn_detection": self._turn_detection_config()}, + }, + wait_until_ready=False, + ) + async def wait_for_response_boundary(self) -> None: """Wait until StepFun has finished or cancelled the current response.""" await self._response_done.wait() @@ -386,12 +414,7 @@ class StepFunRealtimeService(AIService): "voice": self._voice, "input_audio_format": "pcm16", "output_audio_format": "pcm16", - "turn_detection": { - "type": "server_vad", - "prefix_padding_ms": self._prefix_padding_ms, - "silence_duration_ms": self._silence_duration_ms, - "energy_awakeness_threshold": self._energy_awakeness_threshold, - }, + "turn_detection": self._turn_detection_config(), "tools": [tool.provider_schema() for tool in self._tools], "tool_choice": "auto", }, @@ -399,6 +422,16 @@ class StepFunRealtimeService(AIService): wait_until_ready=False, ) + def _turn_detection_config(self) -> dict[str, Any] | None: + if not self._turn_detection_enabled: + return None + return { + "type": "server_vad", + "prefix_padding_ms": self._prefix_padding_ms, + "silence_duration_ms": self._silence_duration_ms, + "energy_awakeness_threshold": self._energy_awakeness_threshold, + } + async def update_instructions(self, instructions: str) -> None: """Refresh model instructions without rebuilding the realtime session.""" self._instructions = instructions diff --git a/backend/services/pipecat/transports.py b/backend/services/pipecat/transports.py index 0edbbc3..3e6d768 100644 --- a/backend/services/pipecat/transports.py +++ b/backend/services/pipecat/transports.py @@ -25,6 +25,7 @@ from pipecat.transports.websocket.fastapi import ( FastAPIWebsocketTransport, FastAPIWebsocketParams, ) +from pipecat.serializers.base_serializer import FrameSerializer from pipecat.serializers.protobuf import ProtobufFrameSerializer from services.pipecat.call_lifecycle import FixedSpeechPlaybackMarkerFrame @@ -103,10 +104,33 @@ def build_ws_transport(websocket: WebSocket) -> FastAPIWebsocketTransport: """裸 WS 输出。序列化用 protobuf(自定义客户端用同款解码); 若对接电话商,把 serializer 换成对应的 TwilioFrameSerializer 等即可。 """ + return build_serialized_ws_transport( + websocket, + serializer=ProtobufFrameSerializer(), + ) + + +def build_serialized_ws_transport( + websocket: WebSocket, + *, + serializer: FrameSerializer, + sample_rate: int | None = None, +) -> FastAPIWebsocketTransport: + """Build a text/binary WS transport without coupling it to one protocol.""" + + sample_rates = ( + { + "audio_in_sample_rate": sample_rate, + "audio_out_sample_rate": sample_rate, + } + if sample_rate + else {} + ) return _WebsocketTransport( websocket=websocket, params=FastAPIWebsocketParams( - serializer=ProtobufFrameSerializer(), + serializer=serializer, **_base_params(), + **sample_rates, ), ) diff --git a/backend/services/pipecat/turn_config.py b/backend/services/pipecat/turn_config.py index 42eed19..0d97a1a 100644 --- a/backend/services/pipecat/turn_config.py +++ b/backend/services/pipecat/turn_config.py @@ -20,7 +20,10 @@ from pipecat.turns.user_stop import ( SpeechTimeoutUserTurnStopStrategy, TurnAnalyzerUserTurnStopStrategy, ) -from pipecat.turns.user_turn_strategies import UserTurnStrategies +from pipecat.turns.user_turn_strategies import ( + ExternalUserTurnStrategies, + UserTurnStrategies, +) DEFAULT_VAD = { @@ -126,3 +129,13 @@ class ConfigurableLLMUserAggregator(LLMUserAggregator): enable_interruptions=enable_interruptions, ) await self._user_turn_controller.update_strategies(strategies) + + async def apply_external_turn_control(self, enabled: bool) -> None: + """Switch between client-controlled PTT and the configured VAD.""" + + strategies = ( + ExternalUserTurnStrategies() + if enabled + else self._params.user_turn_strategies + ) + await self._user_turn_controller.update_strategies(strategies) diff --git a/backend/services/realtime/__init__.py b/backend/services/realtime/__init__.py new file mode 100644 index 0000000..914bbc0 --- /dev/null +++ b/backend/services/realtime/__init__.py @@ -0,0 +1,2 @@ +"""Protocol-neutral helpers for realtime transports.""" + diff --git a/backend/services/realtime/launcher.py b/backend/services/realtime/launcher.py new file mode 100644 index 0000000..04f93f4 --- /dev/null +++ b/backend/services/realtime/launcher.py @@ -0,0 +1,195 @@ +"""Resolve and validate an assistant before creating a media connection.""" + +from __future__ import annotations + +from typing import Any + +from db.session import SessionLocal +from models import AssistantConfig +from services.config_resolver import resolve_runtime_config +from services.node_specs import graph_references +from services.runtime_variables import prepare_dynamic_config +from services.workflow_engine import WorkflowEngine + + +async def resolve_assistant_config( + assistant_id: str, + *, + dynamic_variables: dict[str, Any] | None = None, +) -> AssistantConfig: + async with SessionLocal() as session: + config = await resolve_runtime_config(session, assistant_id) + return prepare_dynamic_config( + config, + dynamic_variables or {}, + assistant_id=assistant_id, + ) + + +def validate_visual_runtime(config: AssistantConfig) -> bool: + """Return the authoritative video-input permission or fail before connect.""" + + if config.type == "workflow": + return WorkflowEngine(config.graph).uses_vision() + + vision_enabled = config.vision_enabled + if not vision_enabled: + return False + has_native_vision = ( + not config.vision_model_resource_id and config.llm_support_image_input + ) + has_aux_vision_model = ( + bool(config.vision_model_resource_id) + and config.vision_llm_support_image_input + ) + if not (has_native_vision or has_aux_vision_model): + raise ValueError( + "当前模型不支持图片输入,请在模型资源中选择支持图片输入的视觉模型" + ) + return True + + +def _require_values(labels: list[tuple[str, Any]]) -> None: + missing = [label for label, value in labels if not value] + if missing: + raise ValueError(f"助手运行配置不完整: {', '.join(missing)}") + + +def _validate_voice_resource( + capability: str, + *, + interface_type: str, + values: dict[str, Any], + secrets: dict[str, Any], +) -> None: + labels: list[tuple[str, Any]] = [] + if capability == "ASR": + if interface_type not in {"openai-asr", "dashscope-asr", "xfyun-asr"}: + raise ValueError(f"不支持的 ASR 接口类型: {interface_type}") + if interface_type == "xfyun-asr": + labels.extend( + (f"ASR {key}", secrets.get(key)) + for key in ("appId", "apiKey", "apiSecret") + ) + else: + labels.extend( + [ + ("ASR modelId", values.get("modelId")), + ("ASR apiUrl", values.get("apiUrl")), + ("ASR apiKey", secrets.get("apiKey")), + ] + ) + elif capability == "TTS": + if interface_type not in { + "openai-tts", + "dashscope-tts", + "xfyun-tts", + "xfyun-super-tts", + }: + raise ValueError(f"不支持的 TTS 接口类型: {interface_type}") + labels.append(("TTS voice", values.get("voice"))) + if interface_type in {"xfyun-tts", "xfyun-super-tts"}: + labels.extend( + (f"TTS {key}", secrets.get(key)) + for key in ("appId", "apiKey", "apiSecret") + ) + else: + labels.extend( + [ + ("TTS modelId", values.get("modelId")), + ("TTS apiUrl", values.get("apiUrl")), + ("TTS apiKey", secrets.get("apiKey")), + ] + ) + elif capability == "LLM": + if interface_type not in {"openai-llm", "dashscope-llm"}: + raise ValueError(f"不支持的 LLM 接口类型: {interface_type}") + labels.extend( + [ + ("LLM modelId", values.get("modelId")), + ("LLM apiUrl", values.get("apiUrl")), + ("LLM apiKey", secrets.get("apiKey")), + ] + ) + _require_values(labels) + + +def validate_runtime_requirements(config: AssistantConfig) -> None: + """Reject structurally incomplete assistants before media negotiation.""" + + if config.type not in {"prompt", "workflow", "dify", "fastgpt"}: + raise ValueError(f"当前助手类型不支持 Realtime API: {config.type}") + if config.runtimeMode == "realtime": + if config.type not in {"prompt", "workflow"}: + raise ValueError(f"助手类型 {config.type} 不支持 realtime 运行模式") + if config.realtime_interface_type not in { + "qwen-audio-realtime", + "stepfun-realtime", + }: + raise ValueError( + f"不支持的 Realtime 接口类型: {config.realtime_interface_type}" + ) + _require_values( + [ + ("Realtime interfaceType", config.realtime_interface_type), + ("Realtime modelId", config.realtimeModel), + ("Realtime apiUrl", config.realtime_base_url), + ("Realtime apiKey", config.realtime_api_key), + ], + ) + return + + _validate_voice_resource( + "ASR", + interface_type=config.stt_interface_type, + values=config.stt_values, + secrets=config.stt_secrets, + ) + _validate_voice_resource( + "TTS", + interface_type=config.tts_interface_type, + values=config.tts_values, + secrets=config.tts_secrets, + ) + if config.type == "prompt": + _validate_voice_resource( + "LLM", + interface_type=config.llm_interface_type, + values=config.llm_values, + secrets=config.llm_secrets, + ) + elif config.type == "dify": + _require_values( + [ + ("Dify apiUrl", config.dify_api_url), + ("Dify apiKey", config.dify_api_key), + ], + ) + elif config.type == "fastgpt": + _require_values( + [ + ("FastGPT apiUrl", config.fastgpt_api_url), + ("FastGPT apiKey", config.fastgpt_api_key), + ], + ) + + if config.type != "workflow": + return + references = graph_references(config.graph) + missing_models = references["model_resources"] - set( + config.workflow_model_resources + ) + missing_knowledge = references["knowledge_bases"] - set( + config.workflow_knowledge_bases + ) + if missing_models or missing_knowledge: + missing = sorted([*missing_models, *missing_knowledge]) + raise ValueError(f"Workflow 引用了不可用资源: {', '.join(missing)}") + for resource in config.workflow_model_resources.values(): + if resource.capability in {"ASR", "TTS", "LLM"}: + _validate_voice_resource( + resource.capability, + interface_type=resource.interface_type, + values=resource.values, + secrets=resource.secrets, + ) diff --git a/backend/services/realtime/lifecycle.py b/backend/services/realtime/lifecycle.py new file mode 100644 index 0000000..91c773d --- /dev/null +++ b/backend/services/realtime/lifecycle.py @@ -0,0 +1,109 @@ +"""Shared ownership of peer connections and their pipeline tasks.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Coroutine +from typing import Any + +from loguru import logger + + +active_connections: set[object] = set() +pipeline_tasks: set[asyncio.Task[None]] = set() +connection_tasks: dict[object, asyncio.Task[None]] = {} +DEFAULT_CLOSE_GRACE_SECONDS = 10.0 + + +def _consume_pipeline_result(task: asyncio.Task[None], connection: object) -> None: + pipeline_tasks.discard(task) + if connection_tasks.get(connection) is task: + connection_tasks.pop(connection, None) + try: + error = task.exception() + except asyncio.CancelledError: + logger.info(f"Realtime pipeline 已取消: task={task.get_name()}") + return + if error is not None: + logger.opt(exception=error).error( + f"Realtime pipeline 异常结束: task={task.get_name()}" + ) + + +def start_pipeline_task( + connection: object, + coroutine: Coroutine[Any, Any, None], + *, + protocol: str, +) -> asyncio.Task[None]: + connection_id = str(getattr(connection, "pc_id", id(connection))) + task = asyncio.create_task( + coroutine, + name=f"{protocol}-pipeline:{connection_id}", + ) + active_connections.add(connection) + pipeline_tasks.add(task) + connection_tasks[connection] = task + task.add_done_callback( + lambda completed, connection=connection: _consume_pipeline_result( + completed, + connection, + ) + ) + return task + + +async def wait_for_pipeline_close( + task: asyncio.Task[None] | None, + *, + connection_id: str, + timeout: float = DEFAULT_CLOSE_GRACE_SECONDS, +) -> None: + if task is None: + return + try: + await asyncio.wait_for(asyncio.shield(task), timeout=timeout) + return + except TimeoutError: + logger.warning( + f"Realtime pipeline 关闭超过 {timeout:g} 秒，执行取消: " + f"connection_id={connection_id}" + ) + except asyncio.CancelledError: + raise + except Exception: + return + + task.cancel() + done, _pending = await asyncio.wait({task}, timeout=timeout) + if not done: + logger.error(f"Realtime pipeline 取消后仍未退出: connection_id={connection_id}") + + +async def shutdown_active_sessions( + *, + timeout: float = DEFAULT_CLOSE_GRACE_SECONDS, +) -> None: + connections = list(active_connections) + if connections: + await asyncio.gather( + *(connection.disconnect() for connection in connections), + return_exceptions=True, + ) + active_connections.difference_update(connections) + + tasks = list(pipeline_tasks) + if not tasks: + return + done, pending = await asyncio.wait(tasks, timeout=timeout) + for task in pending: + task.cancel() + cancelled: set[asyncio.Task[None]] = set() + if pending: + cancelled, stuck = await asyncio.wait(pending, timeout=timeout) + if stuck: + logger.error(f"应用关闭时仍有 {len(stuck)} 个 Realtime pipeline 未退出") + logger.info( + f"Realtime 会话清理完成: normal={len(done)} cancelled={len(cancelled)}" + ) + diff --git a/backend/services/realtime/protocol.py b/backend/services/realtime/protocol.py new file mode 100644 index 0000000..6268db8 --- /dev/null +++ b/backend/services/realtime/protocol.py @@ -0,0 +1,42 @@ +"""Neutral extension point between a transport protocol and the voice pipeline.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Literal, Protocol + +from pipecat.frames.frames import Frame, SystemFrame +from pipecat.processors.frame_processor import FrameProcessor + + +@dataclass(frozen=True) +class PipelineProtocolRuntime: + """Operations a wire-protocol adapter may request from a running pipeline.""" + + queue_frame: Callable[[Frame], Awaitable[None]] + set_external_turn_control: Callable[[bool], Awaitable[None]] + set_response_interruption: Callable[[bool], Awaitable[None]] + commit_audio: Callable[[], Awaitable[None]] + clear_audio: Callable[[], Awaitable[None]] + request_response: Callable[[], Awaitable[None]] + cancel_response: Callable[[], Awaitable[None]] + + +@dataclass +class RealtimeProviderControlFrame(SystemFrame): + """Ordered control sent to a speech-to-speech provider after prior input.""" + + action: Literal["commit_audio", "clear_audio", "request_response"] + + +class PipelineProtocolAdapter(Protocol): + """A protocol adapter only translates frames; it does not own the assistant.""" + + def input_processors(self) -> list[FrameProcessor]: ... + + def inference_processors(self) -> list[FrameProcessor]: ... + + def output_processors(self) -> list[FrameProcessor]: ... + + async def bind(self, runtime: PipelineProtocolRuntime) -> None: ... diff --git a/backend/settings.py b/backend/settings.py index 31807fd..668f1a2 100644 --- a/backend/settings.py +++ b/backend/settings.py @@ -38,6 +38,14 @@ AUTH_TOKEN_EXPIRE_MINUTES = int(os.getenv("AUTH_TOKEN_EXPIRE_MINUTES", "1440")) AUTH_COOKIE_SECURE = os.getenv("AUTH_COOKIE_SECURE", "false").lower() == "true" AUTH_COOKIE_SAMESITE = os.getenv("AUTH_COOKIE_SAMESITE", "lax") +# ---- Public Realtime API auth ---- +# Keep public integration credentials separate from the admin login cookie. +# Production deployments should set a dedicated random secret. +REALTIME_TOKEN_SECRET = os.getenv("REALTIME_TOKEN_SECRET", AUTH_SECRET_KEY) +REALTIME_CLIENT_SECRET_TTL_SECONDS = int( + os.getenv("REALTIME_CLIENT_SECRET_TTL_SECONDS", "60") +) + # ---- WebRTC STUN / TURN ---- # Override STUN_URL in remote deployments to use the colocated coturn server # instead of waiting for a public STUN service that may be unreachable. diff --git a/docs-developer/docs.json b/docs-developer/docs.json index 4448d16..5476154 100644 --- a/docs-developer/docs.json +++ b/docs-developer/docs.json @@ -46,7 +46,7 @@ "groups": [ { "group": "开始与对接", - "pages": ["index", "integrations"] + "pages": ["index", "integrations", "realtime-protocol"] } ] } diff --git a/docs-developer/index.mdx b/docs-developer/index.mdx index 71dc492..3595105 100644 --- a/docs-developer/index.mdx +++ b/docs-developer/index.mdx @@ -40,6 +40,10 @@ make db-seed 保存后，从助手详情页打开语音预览即可测试。 +## 4. 对接 Realtime 协议 + +- [Realtime 兼容协议设计](/realtime-protocol)：了解后端提供的 OpenAI Realtime 兼容 WebRTC、WebSocket、鉴权与 Interactive Media 扩展协议。 + 浏览器麦克风仅在 localhost 或 HTTPS 下可用。局域网、远程环境请按仓库中的 `deploy/README.md` 配置 HTTPS 与 TURN。 diff --git a/docs-developer/realtime-protocol.mdx b/docs-developer/realtime-protocol.mdx new file mode 100644 index 0000000..fde3be4 --- /dev/null +++ b/docs-developer/realtime-protocol.mdx @@ -0,0 +1,693 @@ +--- +title: Realtime 兼容协议设计 +description: 面向集成方的 OpenAI Realtime 兼容 WebRTC、WebSocket 与扩展事件协议。 +icon: radio +--- + +# Realtime 兼容协议设计 + +> 当前实现版本为 `Interactive Media Realtime Extensions v1`。OpenAI 兼容接口与现有 `/api/webrtc/offer`、`/ws/voice`、`/ws/stream`、RTVI 消息并行提供；RTVI 不是待删除的迁移接口。 + +## 设计目标 + +AI 视频助手以 OpenAI Realtime 协议作为对外接口，同时保留助手、知识库、工具、工作流、动态变量和视频输入等现有能力。 + +协议适配层只负责连接、事件和 Pipecat Frame 之间的转换，不引入第二套语音运行时： + +```text +OpenAI Realtime 客户端 + │ + │ WebRTC / WebSocket + ▼ +Realtime 协议适配层 + │ + │ Pipecat transport / frame + ▼ +现有 run_pipeline() + │ + ├── Prompt / Workflow / External Agent + ├── ASR / LLM / TTS / Realtime Model + ├── Knowledge / System、HTTP、Client、MCP Tools + └── History / Handoff / End Call +``` + +协议分为两层： + +| 层级 | 约束 | 面向对象 | +| --- | --- | --- | +| OpenAI Realtime Core | 保持标准事件名称、字段含义和生命周期，不改变标准事件语义 | 任意 OpenAI Realtime 兼容客户端 | +| `x.interactive_media.*` Extensions | 所有项目扩展均有命名空间和能力协商 | 了解 Interactive Media 扩展的客户端 | + +普通兼容客户端不声明扩展能力时，服务端不会发送 `x.interactive_media.*` 事件。它仍可使用音频、文本、图片、转写、回复流、打断和函数调用，但不会获得工作流节点、动态变量、连续视频等项目专属状态。 + +## 接口概览 + +假设服务地址为 `https://api.example.com`： + +| 接口 | 用途 | 推荐客户端 | +| --- | --- | --- | +| `POST /v1/realtime/client_secrets` | 创建短期、绑定助手的临时令牌 | 浏览器或移动端 | +| `POST /v1/realtime/calls` | 用 SDP Offer 创建 WebRTC 会话并返回 SDP Answer | 浏览器、移动端 | +| `wss://api.example.com/v1/realtime?model=assistant:` | 创建 WebSocket Realtime 会话 | 服务端、话务网关、自定义客户端 | + +`model` 使用 `assistant:`，其中 `id` 是当前数据库中的助手 ID，例如： + +```text +assistant:asst_xxx +``` + +这里的 `model` 代表一个完整助手，而不是直接选择底层模型。助手绑定的系统提示词、模型凭证、知识库、服务端工具、工作流和运行模式仍由服务端控制。 + +### 鉴权 + +- 长期 API Key 仅供可信服务端使用，通过 `Authorization: Bearer ` 发送。 +- 浏览器先由自己的业务后端创建临时令牌，再用临时令牌连接 Realtime 接口。 +- 临时令牌绑定助手和会话配置，默认有效期为 60 秒；首版不保证单次使用，令牌只用于建连，建连后到期不会中止会话。 +- 父长期 Key 被撤销或过期后，尚未使用的临时令牌立即失效。 +- 管理后台 Cookie 不属于公开兼容协议，也不应作为第三方客户端的鉴权方式。 +- 会话和事件不会返回底层模型密钥、助手系统提示词或工具密钥。 + +### 长期 API Key 管理 + +Realtime Key 存在数据库中，每个 Key 默认可访问当前及未来创建的全部助手。管理接口仅允许后台管理员 Cookie 或 Basic Auth 调用： + +| 接口 | 行为 | +| --- | --- | +| `POST /api/realtime/api-keys` | 创建 Key；完整 `sk-rt-...` 只在本次响应显示一次 | +| `GET /api/realtime/api-keys` | 查看名称、前缀、有效期、最后使用时间和状态，不返回密钥 | +| `DELETE /api/realtime/api-keys/{id}` | 逻辑撤销并保留审计记录 | + +轮换方式为先创建新 Key，再撤销旧 Key。服务端只保存使用部署级 pepper 计算的 HMAC-SHA256，不保存可恢复的 Key 明文。 + +## 会话配置 + +首次建连配置以及后续 `session.update` 使用标准 Realtime session 结构： + +```json +{ + "type": "realtime", + "model": "assistant:asst_xxx", + "output_modalities": ["audio"], + "audio": { + "input": { + "format": { + "type": "audio/pcm", + "rate": 24000 + }, + "turn_detection": { + "type": "server_vad" + } + }, + "output": { + "format": { + "type": "audio/pcm" + } + } + } +} +``` + +客户端可修改输出模态，以及输入端 Server VAD / Push-to-Talk 参数。服务端在 `session.updated` 中返回**实际生效**的配置。 + +以下内容始终以服务端助手配置为准： + +- 助手身份和基础系统提示词； +- Pipeline 或 Realtime 运行模式； +- 底层模型资源及凭证； +- 知识库和服务端 System、HTTP、MCP 工具； +- 工作流图、结束条件和接管策略。 +- voice、工具定义、系统提示词和每次回复的临时指令。 + +服务端不能静默接受一个实际未生效的字段。字段不支持、被锁定或没有权限时，应返回标准 `error` 事件。 + +## WebRTC + +WebRTC 是浏览器和移动端实时音视频交互的首选通道。音频和可选视频走媒体轨道，JSON 事件走 DataChannel。 + +### 建连方式一：统一接口 + +可信业务后端向 `/v1/realtime/calls` 发送 `multipart/form-data`： + +| 表单字段 | 类型 | 说明 | +| --- | --- | --- | +| `sdp` | string | 浏览器生成的 SDP Offer | +| `session` | JSON string | Realtime session 配置，必须选择一个已授权助手 | + +请求使用长期 API Key 鉴权。成功响应为 `Content-Type: application/sdp` 的 SDP Answer。 + +```bash +curl -X POST "https://api.example.com/v1/realtime/calls" \ + -H "Authorization: Bearer $REALTIME_API_KEY" \ + -F 'sdp= { + if (event.track.kind === "audio") { + remoteAudio.srcObject = event.streams[0]; + } +}; + +const localStream = await navigator.mediaDevices.getUserMedia({ audio: true }); +pc.addTrack(localStream.getAudioTracks()[0], localStream); + +const events = pc.createDataChannel("oai-events"); +events.onmessage = (event) => { + const serverEvent = JSON.parse(event.data); + console.log(serverEvent); +}; + +const offer = await pc.createOffer(); +await pc.setLocalDescription(offer); + +const answerSdp = await fetch("https://api.example.com/v1/realtime/calls", { + method: "POST", + headers: { + Authorization: `Bearer ${ephemeralToken}`, + "Content-Type": "application/sdp" + }, + body: offer.sdp +}).then((response) => response.text()); + +await pc.setRemoteDescription({ type: "answer", sdp: answerSdp }); +``` + +### WebRTC 数据分工 + +| 数据 | 通道 | 说明 | +| --- | --- | --- | +| 用户麦克风 | WebRTC audio track | 不再通过 JSON 重复发送音频块 | +| 助手语音 | WebRTC remote audio track | 浏览器直接播放，不依赖 `response.output_audio.delta` 拼接 | +| 标准和扩展事件 | `oai-events` DataChannel | 使用 UTF-8 JSON 文本，保持可靠、有序传输 | +| 用户摄像头 | 可选 WebRTC video track | 通过 `video_track` capability 协商，非 OpenAI 标准能力 | + +摄像头轨道只在助手启用视觉、令牌允许且 SDP 协商成功时接收。普通 OpenAI 兼容客户端不添加视频轨道，仍可正常完成语音和文本会话。 + +切换摄像头、静音、音量控制和 `RTCPeerConnection.getStats()` 属于客户端 WebRTC 能力，不需要新增协议事件。切换设备时可使用 `RTCRtpSender.replaceTrack()`，无需重建会话。 + +## WebSocket + +WebSocket 适合服务到服务、话务网关和希望自行处理音频缓冲的客户端。连接时在 query 中指定助手： + +```text +wss://api.example.com/v1/realtime?model=assistant%3Aasst_xxx +``` + +可信服务端在握手请求中发送： + +```http +Authorization: Bearer +``` + +浏览器无法设置 `Authorization` 握手头时，可同时发送 `realtime` 与 `openai-insecure-api-key.` WebSocket 子协议；服务端选择 `realtime`。这只适用于短期令牌，不应把长期 Key 放入浏览器。 + +连接成功后，客户端和服务端都只发送 UTF-8 JSON 文本帧。浏览器无法安全保存长期 API Key，浏览器实时通话仍应优先使用 WebRTC。 + +### 文本输入 + +先创建用户消息，再请求生成回复： + +```json +{ + "event_id": "evt_text_001", + "type": "conversation.item.create", + "item": { + "type": "message", + "role": "user", + "content": [ + { + "type": "input_text", + "text": "请介绍办理流程" + } + ] + } +} +``` + +```json +{ + "event_id": "evt_response_001", + "type": "response.create" +} +``` + +### 音频输入和输出 + +WebSocket 音频以 Base64 编码放入标准事件： + +```json +{ + "event_id": "evt_audio_001", + "type": "input_audio_buffer.append", + "audio": "" +} +``` + +- 开启 Server VAD 时，服务端自动产生 `speech_started`、`speech_stopped` 并触发回复。 +- 关闭 VAD 时，客户端发送 `input_audio_buffer.commit`，然后发送 `response.create`。 +- 服务端通过 `response.output_audio.delta` 返回 Base64 音频块，并以 `response.output_audio.done` 结束。 +- 助手文本或音频转写分别通过 `response.output_text.*` 和 `response.output_audio_transcript.*` 返回。 + +WebSocket MVP 不传输连续原始视频。视觉输入使用标准 `input_image`；需要连续摄像头的客户端使用 WebRTC 视频轨道。 + +### 图片输入 + +图片和文本可放在同一个用户消息中： + +```json +{ + "type": "conversation.item.create", + "item": { + "type": "message", + "role": "user", + "content": [ + { + "type": "input_text", + "text": "画面里有什么？" + }, + { + "type": "input_image", + "image_url": "data:image/jpeg;base64," + } + ] + } +} +``` + +## Push-to-Talk + +Push-to-Talk（按住说话）属于 OpenAI Realtime Core 兼容能力，不需要新增 `x.interactive_media.push_to_talk.*` 扩展事件。按钮是否按下、快捷键和麦克风 UI 状态由客户端本地管理，服务端只处理标准音频缓冲与回复事件。 + +### 关闭自动 VAD + +客户端建立会话后先通过 `session.update` 关闭自动轮次检测： + +```json +{ + "event_id": "evt_ptt_config_001", + "type": "session.update", + "session": { + "type": "realtime", + "audio": { + "input": { + "turn_detection": null + } + } + } +} +``` + +关闭 VAD 后，服务端不会根据静音自动提交用户语音或创建回复。客户端必须在用户松开按钮时发送 `input_audio_buffer.commit`，然后发送 `response.create`。 + +### WebSocket 时序 + +WebSocket 的音频和控制事件在同一条有序连接上传输： + +| 阶段 | 客户端行为 | +| --- | --- | +| 按下 | 开始在客户端录音；如果已有活动回复，发送 `response.cancel` | +| 打断播放 | 立即停止本地音频播放，并发送 `conversation.item.truncate`，用 `audio_end_ms` 删除用户未听到的内容 | +| 松开 | 通过一个或多个 `input_audio_buffer.append` 发送本次录音 | +| 提交 | 依次发送 `input_audio_buffer.commit` 和 `response.create` | + +完整的松开事件序列如下： + +```json +{ + "event_id": "evt_ptt_audio_001", + "type": "input_audio_buffer.append", + "audio": "" +} +``` + +```json +{ + "event_id": "evt_ptt_commit_001", + "type": "input_audio_buffer.commit" +} +``` + +```json +{ + "event_id": "evt_ptt_response_001", + "type": "response.create" +} +``` + +如果用户在录音完成前取消本次输入，客户端应丢弃本地录音；已经发送到服务端的未提交音频则使用 `input_audio_buffer.clear` 清除。 + +### WebRTC 时序 + +WebRTC 的音频媒体轨道和 `oai-events` 控制事件属于不同通道，因此开始新一轮输入前必须显式清理旧缓冲： + +| 阶段 | 客户端行为 | +| --- | --- | +| 按下 | 先发送 `input_audio_buffer.clear`，再开始或放开本地麦克风输入门控 | +| 打断生成 | 如果已有活动回复，发送 `response.cancel` | +| 打断播放 | 如果助手音频仍在播放，发送 `output_audio_buffer.clear`；服务端同时截断未播放的对话内容 | +| 松开 | 关闭本地麦克风输入门控，依次发送 `input_audio_buffer.commit` 和 `response.create` | + +```js +function sendEvent(event) { + dataChannel.send(JSON.stringify(event)); +} + +function onPushDown() { + sendEvent({ type: "input_audio_buffer.clear" }); + + if (hasActiveResponse) { + sendEvent({ type: "response.cancel" }); + } + + if (isAssistantAudioPlaying) { + sendEvent({ type: "output_audio_buffer.clear" }); + } + + microphoneTrack.enabled = true; +} + +function onPushUp() { + microphoneTrack.enabled = false; + sendEvent({ type: "input_audio_buffer.commit" }); + sendEvent({ type: "response.create" }); +} +``` + +客户端应处理指针移出、窗口失焦和权限撤销等情况，确保一次按下只产生一次提交。没有有效音频时不要发送 `input_audio_buffer.commit`。 + +### 保留 VAD、手动触发回复 + +如果只想由客户端决定何时生成回复，但仍希望服务端判断说话开始和结束，可以保留 VAD，并关闭自动回复及自动打断： + +```json +{ + "type": "session.update", + "session": { + "type": "realtime", + "audio": { + "input": { + "turn_detection": { + "type": "server_vad", + "interrupt_response": false, + "create_response": false + } + } + } + } +} +``` + +这种模式适合在提交前进行审核、输入校验或知识检索，不等同于严格的按住说话。 + +## 标准事件范围 + +首版兼容层至少支持以下 OpenAI Realtime 事件。 + +### 客户端发送 + +| 事件 | 用途 | +| --- | --- | +| `session.update` | 更新允许覆盖的会话配置 | +| `conversation.item.create` | 发送文本、整段音频、图片或函数结果 | +| `conversation.item.truncate` | 打断后删除未播放的助手内容 | +| `input_audio_buffer.append` | WebSocket 追加音频 | +| `input_audio_buffer.commit` | 无 VAD 时提交音频 | +| `input_audio_buffer.clear` | 清除未提交音频 | +| `output_audio_buffer.clear` | WebRTC/SIP 清除未播放音频并截断上下文 | +| `response.create` | 请求助手生成回复 | +| `response.cancel` | 取消正在生成的回复 | + +### 服务端发送 + +| 事件组 | 事件 | +| --- | --- | +| 会话 | `session.created`、`session.updated`、`error` | +| 用户语音 | `input_audio_buffer.speech_started`、`input_audio_buffer.speech_stopped`、`input_audio_buffer.committed` | +| 用户转写 | `conversation.item.input_audio_transcription.completed` | +| 对话项 | `conversation.item.added`、`conversation.item.done` | +| 回复生命周期 | `response.created`、`response.output_item.added`、`response.output_item.done`、`response.done` | +| 文本 | `response.output_text.delta`、`response.output_text.done` | +| 音频 | `response.output_audio.delta`、`response.output_audio.done` | +| 音频转写 | `response.output_audio_transcript.delta`、`response.output_audio_transcript.done` | +| 函数调用 | `response.function_call_arguments.delta`、`response.function_call_arguments.done` | + +WebRTC 中的助手音频本体走媒体轨道，因此客户端不应依赖 `response.output_audio.delta` 播放声音；音频生命周期和转写事件仍通过 `oai-events` 发送。 + +## 首版兼容矩阵 + +| 分类 | 状态 | 说明 | +| --- | --- | --- | +| WebRTC 音频、`oai-events` DataChannel | 支持 | 复用 SmallWebRTC 和现有 Pipeline | +| WebSocket 文本与 24 kHz PCM16 mono Base64 音频 | 支持 | 只接受 UTF-8 JSON 文本帧 | +| 文本轮次、Server VAD、PTT、取消和清空输出 | 支持 | `turn_detection: null` 时由 `commit` + `response.create` 结束轮次 | +| `input_image` Data URL | 条件支持 | 仅 Pipeline 模式且助手已授权视觉输入；Realtime 模式明确报错 | +| Client Tool 标准 function call | 支持 | System、HTTP、MCP 工具仍在服务端内部执行 | +| `x.interactive_media.*` | 协商后支持 | 未协商的客户端不会收到扩展事件 | +| `model`、instructions、voice、工具和系统提示词 | 锁定 | 由助手配置决定，覆盖请求返回标准 `error` | +| WebSocket 其他采样率、压缩音频、连续视频 | 暂不支持 | WebRTC 音频由 SDP 协商；连续视频仅走 WebRTC | +| 公开助手别名、断线续传、并发回复、SIP | 暂不支持 | `model` 必须使用实际的 `assistant:asst_xxx` | + +Qwen Realtime provider 只允许在 provider 会话首次配置前选择 PTT 或自动轮次检测；建立连接后再次切换会返回明确错误。Pipeline 与支持动态更新的 Realtime provider 可在会话内切换。 + +## 工具调用 + +服务端 System、HTTP 和 MCP 工具在现有运行时内部执行。Client 工具使用标准函数调用协议，不再建立一套平行的 `client-tool-call` 公开协议。 + +服务端完成函数参数后发送标准函数调用事件；客户端执行函数并回传： + +```json +{ + "type": "conversation.item.create", + "item": { + "type": "function_call_output", + "call_id": "call_abc123", + "output": "{\"ok\":true,\"result\":{\"ticket_id\":\"T-1001\"}}" + } +} +``` + +随后发送 `response.create` 让助手基于工具结果继续当前轮次。`call_id` 在整个会话内唯一，客户端应原样返回。 + +## Interactive Media 扩展 + +### 能力协商 + +客户端收到 `session.created` 后主动声明所需扩展： + +```json +{ + "event_id": "evt_capabilities_001", + "type": "x.interactive_media.capabilities.update", + "capabilities": [ + "workflow_events", + "dynamic_variables", + "handoff", + "video_track" + ] +} +``` + +服务端返回实际接受的能力： + +```json +{ + "event_id": "evt_server_001", + "type": "x.interactive_media.capabilities.updated", + "capabilities": [ + "workflow_events", + "dynamic_variables", + "handoff", + "video_track" + ], + "rejected": [] +} +``` + +服务端只发送已接受能力对应的扩展事件。扩展版本与 OpenAI Realtime Core 独立演进。 + +### 扩展事件目录 + +| 方向 | 事件 | 用途 | +| --- | --- | --- | +| Client → Server | `x.interactive_media.capabilities.update` | 请求扩展能力 | +| Server → Client | `x.interactive_media.capabilities.updated` | 返回允许的能力 | +| Client → Server | `x.interactive_media.session.variables.update` | 静默更新动态变量，不创建用户消息 | +| Server → Client | `x.interactive_media.session.variables.updated` | 确认变量更新状态 | +| Server → Client | `x.interactive_media.workflow.node_active` | 当前工作流节点变化 | +| Server → Client | `x.interactive_media.workflow.variables.updated` | 工作流变量变化 | +| Server → Client | `x.interactive_media.workflow.event` | 工作流业务事件 | +| Server → Client | `x.interactive_media.workflow.error` | 工作流执行失败 | +| Server → Client | `x.interactive_media.call.handoff_requested` | 助手请求人工接管 | +| Server → Client | `x.interactive_media.call.ended` | 会话结束及原因 | + +动态变量更新示例： + +```json +{ + "event_id": "evt_variables_001", + "type": "x.interactive_media.session.variables.update", + "update_id": "update_001", + "variables": { + "user_name": "王先生", + "region": "上海" + } +} +``` + +```json +{ + "event_id": "evt_server_002", + "type": "x.interactive_media.session.variables.updated", + "update_id": "update_001", + "status": "accepted" +} +``` + +工作流节点事件示例： + +```json +{ + "event_id": "evt_server_003", + "type": "x.interactive_media.workflow.node_active", + "nodeId": "collect_materials" +} +``` + +会话结束示例： + +```json +{ + "event_id": "evt_server_004", + "type": "x.interactive_media.call.ended", + "reason": "assistant_completed" +} +``` + +`x.interactive_media.call.ended` 发出后，服务端停止产生新回复并正常关闭媒体与信令通道。未启用扩展的客户端只观察到标准回复结束和连接关闭。 + +## 打断和回复并发 + +- 默认同一会话只允许一个活动回复。 +- 用户开始说话且助手允许打断时，服务端取消当前生成并停止后续音频输出。 +- 客户端可显式发送 `response.cancel`。 +- WebSocket 客户端自行管理音频播放，因此停止播放后应发送 `conversation.item.truncate`，用 `audio_end_ms` 告诉服务端用户实际听到的位置。 +- WebRTC/SIP 的输出音频由服务端缓冲；VAD 打断时由服务端自动截断，显式 Push-to-Talk 打断使用 `output_audio_buffer.clear`。 +- 助手配置禁止打断时，服务端继续当前回复，并对不允许的显式取消返回 `error`。 + +## 错误格式 + +协议错误使用标准 `error` 事件，不使用只存在于某个传输的自定义错误结构： + +```json +{ + "event_id": "evt_server_error_001", + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "session_field_locked", + "message": "The assistant does not allow overriding instructions.", + "param": "session.instructions", + "event_id": "evt_update_001" + } +} +``` + +常见错误码包括： + +| code | 含义 | +| --- | --- | +| `authentication_failed` | API Key 或临时令牌无效 | +| `assistant_not_found` | 助手不存在、未发布或不可访问 | +| `session_field_locked` | 客户端尝试覆盖服务端锁定配置 | +| `unsupported_event` | 当前协议版本不支持该事件 | +| `capability_not_enabled` | 未协商或无权使用某项扩展 | +| `response_in_progress` | 已有活动回复且当前请求不能并发执行 | +| `tool_result_timeout` | Client 工具未在规定时间内返回 | +| `internal_error` | 运行时出现不可恢复错误 | + +错误只终止相关事件；只有鉴权失败、协议严重错误或运行时不可恢复时才关闭整个连接。 + +## 事件顺序和重连 + +- WebSocket 文本帧和 `oai-events` DataChannel 都按连接内的发送顺序处理。 +- 客户端事件建议携带唯一 `event_id`；服务端错误通过 `error.event_id` 指回原事件。 +- `session.created` 是连接建立后的第一个业务事件。 +- 每个回复由 `response.created` 开始，以 `response.done` 结束；Delta 必须携带所属的 response、item 和 content 索引。 +- `Interactive Media Realtime Extensions v1` 不支持会话断线续传。重连会创建新会话；服务端历史记录仍按平台策略持久化。 +- 客户端不得把收到的旧连接事件写入新连接。 + +## 现有能力映射 + +| 现有能力或旧事件 | 新协议 | +| --- | --- | +| WebRTC 麦克风与助手音频 | WebRTC audio track | +| 摄像头连续输入 | 可选 WebRTC video track + `video_track` capability | +| `user-input` 文本或图片 | `conversation.item.create` + `response.create` | +| `session-update` | `x.interactive_media.session.variables.update` | +| `transcript` 用户转写 | `conversation.item.input_audio_transcription.*` | +| `assistant-text-start/delta/end` | `response.output_text.*` 或 `response.output_audio_transcript.*` | +| `client-tool-call/result` | 标准 function call + `function_call_output` | +| `node-active` | `x.interactive_media.workflow.node_active` | +| `workflow-variables` | `x.interactive_media.workflow.variables.updated` | +| `workflow-event` | `x.interactive_media.workflow.event` | +| `call-ended` | `x.interactive_media.call.ended` | +| 知识库、服务端工具、工作流执行 | 继续在现有运行时内部完成 | + +Pipeline 与 Realtime 两种助手运行模式对客户端使用同一套北向事件。协议适配层负责把不同内部模型事件归一化，客户端不需要根据运行模式切换协议。 + +原有 RTVI 入口和消息格式继续并行保留，当前前端不需要迁移；新兼容接口面向新的第三方集成。 + +## 传输选择 + +| 场景 | 推荐传输 | 原因 | +| --- | --- | --- | +| 浏览器实时语音、打断和摄像头 | WebRTC | 媒体自适应、播放简单、支持视频轨道 | +| 移动端实时语音 | WebRTC | 更适合不稳定网络和双向媒体 | +| 服务端 Agent 或批处理式实时流 | WebSocket | JSON 与音频缓冲可完全由调用方控制 | +| 电话网关 | WebSocket | 易于接入现有 PCM/PCMU 音频桥接 | +| 纯文本 Realtime | WebSocket | 无需建立媒体轨道 | + +## 兼容性原则 + +1. 不修改标准事件的名称和含义。 +2. 所有项目扩展使用 `x.interactive_media.*` 命名空间。 +3. 扩展必须先协商再发送，普通客户端只看到标准事件。 +4. 未支持字段明确报错，不静默伪装为成功。 +5. OpenAI Realtime Core 和 Interactive Media 扩展分别进行版本管理。 +6. 新协议只替换北向连接层，不改变现有 Pipecat 运行时和助手配置模型。 + +## 参考 + +- [OpenAI Realtime API with WebRTC](https://developers.openai.com/api/docs/guides/realtime-webrtc) +- [OpenAI Realtime API with WebSocket](https://developers.openai.com/api/docs/guides/realtime-websocket) +- [OpenAI Realtime conversations](https://developers.openai.com/api/docs/guides/realtime-conversations) +- [OpenAI Realtime Push-to-Talk](https://developers.openai.com/api/docs/guides/realtime-conversations#push-to-talk)