Initial commit: AI Video Assistant fullstack platform.

Add pipecat-based backend with WebRTC/WS voice routes, Next.js frontend, and Docker Compose orchestration.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/.env.example

@@ -0,0 +1,35 @@
+# 复制为 .env 并填入真实值。
+# 国产栈走 OpenAI 兼容协议:每类服务给一个 base_url + api_key + 模型名即可。
+
+# ---- LLM:DeepSeek(OpenAI 兼容,直连云端,只需 key) ----
+LLM_BASE_URL=https://api.deepseek.com/v1
+LLM_API_KEY=sk-your-deepseek-key
+LLM_MODEL=deepseek-chat
+
+# ---- STT:SenseVoice / FunASR ----
+# 需要本地起一个 OpenAI 兼容的语音转写服务(/v1/audio/transcriptions),
+# 例如用 funasr / sherpa-onnx / speaches 包一层。下面填那个服务地址。
+STT_BASE_URL=http://localhost:8001/v1
+STT_API_KEY=local
+STT_MODEL=sensevoice
+
+# ---- TTS:CosyVoice ----
+# 同样需要本地起一个 OpenAI 兼容的 TTS 服务(/v1/audio/speech)。
+TTS_BASE_URL=http://localhost:8002/v1
+TTS_API_KEY=local
+TTS_MODEL=cosyvoice
+TTS_VOICE=中文女
+
+# ---- Realtime 模式(可选,先不接也行) ----
+REALTIME_API_KEY=
+REALTIME_MODEL=gpt-realtime
+
+# ---- 数据库(Postgres) ----
+# 本地直连;docker compose 里则用 postgres:5432
+DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/postgres
+
+# ---- 服务监听 & 跨域 ----
+HOST=0.0.0.0
+PORT=8000
+# 前端开发地址,允许跨域
+CORS_ORIGINS=http://localhost:3000

backend/.gitignore

@@ -0,0 +1,4 @@
+.venv/
+__pycache__/
+*.pyc
+.env

backend/.python-version

@@ -0,0 +1 @@
+3.12

backend/Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.11-slim
+
+# aiortc(WebRTC)和 onnxruntime(Silero VAD)需要的系统库
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    ffmpeg \
+    libopus0 \
+    libvpx-dev \
+    libsrtp2-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

backend/README.md

@@ -0,0 +1,104 @@
+# AI Video Assistant — 后端引擎
+
+参考 [dograh](../dograh),用 **pipecat** 作语音引擎的自建后端。目标是逐步长成
+类 dograh 平台,但**同时支持 WebRTC 和 WS 两种音频输出**。
+
+## 双输出架构(核心)
+
+pipecat 把"管线"和"输出方式"解耦:同一条 `STT→LLM→TTS` 管线可挂不同 transport。
+
+```
+                                ┌─────────────────────────────────┐
+  浏览器 ──WebRTC──► /ws/voice ──┤                                 │
+                                │  run_pipeline(transport, cfg):  │
+  自定义/话务 ─WS──► /ws/stream ─┤  input→STT→LLM→TTS→output       │
+                                └─────────────────────────────────┘
+```
+
+- **WebRTC**(`/ws/voice`):浏览器,低延迟,带 NAT 穿透。`SmallWebRTCTransport`
+- **WS**(`/ws/stream`):裸音频流,服务端/话务/自定义客户端,无 ICE/TURN。`FastAPIWebsocketTransport`
+
+加第三种输出(如 Twilio 电话)= 在 `services/pipecat/transports.py` 再加一个 `build_xxx_transport` + serializer,管线一行不用改。
+
+## 目录结构(对齐 dograh 的 `api/`,便于生长)
+
+```
+ai-video-backend/
+├── app.py                       # FastAPI 入口,挂路由 + CORS
+├── config.py                    # 读 .env,所有 provider 接入点
+├── models.py                    # AssistantConfig(对齐前端 AssistantForm)
+├── routes/                      # 一个文件一组端点(对齐 dograh routes/)
+│   ├── health.py
+│   ├── voice_webrtc.py          # WebRTC 信令
+│   └── voice_ws.py              # WS 裸音频流
+├── services/
+│   └── pipecat/                 # 引擎(对齐 dograh services/pipecat/)
+│       ├── service_factory.py   # 建 STT/LLM/TTS(加 provider 在此)
+│       ├── transports.py        # transport 工厂(加输出方式在此)
+│       └── pipeline.py          # 管线拼装与运行(transport 无关)
+├── Dockerfile
+├── requirements.txt
+└── .env.example
+
+# 平台长大后会再加(对齐 dograh):
+#   db/        SQLAlchemy 模型 + 会话(助手/对话/向量)
+#   schemas/   pydantic 请求响应
+#   tasks/     后台任务(转写、报表)—— 配 redis
+#   services/storage.py  录音存储 —— 配 minio
+```
+
+## 国产栈(全走 OpenAI 兼容,换栈只改 .env)
+
+| 类型 | 默认 | 接入 |
+|------|------|------|
+| LLM  | DeepSeek | 云端直连,只需 key |
+| STT  | SenseVoice / FunASR | 本地 OpenAI 兼容转写服务 |
+| TTS  | CosyVoice | 本地 OpenAI 兼容 TTS 服务 |
+
+## 本地运行(用 uv,Python 3.12)
+
+```bash
+cd ai-video/backend
+uv venv                       # 按 .python-version 用 3.12
+
+# 阶段 A:只验证存储/CRUD(不装 pipecat,秒级)
+uv pip install fastapi "uvicorn[standard]" sqlalchemy asyncpg greenlet python-dotenv pydantic loguru
+# 阶段 B:做语音时再装全量(含 pipecat,需 3.10+)
+# uv pip install -r requirements.txt
+
+cp .env.example .env          # CRUD 阶段只需 DATABASE_URL;语音再填模型 key
+# 起 Postgres:在 ai-video/ 下 docker compose up -d postgres
+.venv/bin/uvicorn app:app --reload --port 8000
+```
+
+> pipecat 相关代码用**惰性导入**,所以阶段 A 不装 pipecat 也能启动并跑 `/api/*` 与 `/health`;
+> 只有真正连 `/ws/voice`、`/ws/stream` 时才需要全量依赖。
+>
+> 交互式 API 文档:启动后访问 http://localhost:8000/docs(手动戳 CRUD、定 schema 用)。
+
+## Docker(ai-video/docker-compose.yaml)—— 调试主路径
+
+api 服务挂了源码 + `--reload`,前端用 npm dev + HMR,改代码都即时生效。
+
+```bash
+cd ai-video
+docker compose up                 # 前台起 pg + api(:8000)+ ui(:3000),日志直出
+docker compose up -d              # 后台起;看日志 docker compose logs -f api
+docker compose down               # 停止全部
+
+# 可选:对象存储 / 后台任务
+docker compose --profile data up  # + rustfs(S3) / redis
+# 可选:公网部署(WebRTC 需 TURN)
+docker compose --profile remote up -d
+```
+
+> 首次 `up` 会构建 api 镜像(装全量 `requirements.txt`,含 pipecat,较慢)。
+> 之后改 Python 代码靠 `--reload` 热更新,不用重建;只有改 `requirements.txt` 才 `docker compose build api`。
+
+## 待联调 / TODO
+
+- [ ] `pip install` 后跑通,核对 pipecat 版本的服务/transport 构造参数(代码内有注释)
+- [ ] 起本地 SenseVoice / CosyVoice 的 OpenAI 兼容服务
+- [ ] `realtime` 模式(目前只 `pipeline` 级联)
+- [ ] 前端 `DebugVoicePanel` 接 `/ws/voice`(抄 dograh `useWebSocketRTC.tsx`)
+- [ ] 加 DB 后:助手配置入库(目前随请求内联)

backend/app.py

@@ -0,0 +1,48 @@
+"""FastAPI 入口。挂载路由,放行前端跨域,启动时建表。
+
+启动:  uvicorn app:app --reload --port 8000
+
+路由分组(对齐 dograh 的 routes/ 结构):
+  /health              健康检查
+  /api/assistants      助手 CRUD
+  /api/credentials     模型凭证 CRUD(key 打码)
+  /ws/voice            WebRTC 输出(浏览器)
+  /ws/stream           WS 输出(裸音频流)
+"""
+
+from contextlib import asynccontextmanager
+
+import config
+import uvicorn
+from db.session import init_db
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from routes import assistants, credentials, health, voice_webrtc, voice_ws
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    await init_db()  # MVP:启动建表;表稳定后切 alembic
+    yield
+
+
+app = FastAPI(title="AI Video Assistant 平台 - 后端", lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=config.CORS_ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(health.router)
+app.include_router(assistants.router)
+app.include_router(credentials.router)
+app.include_router(voice_webrtc.router)
+app.include_router(voice_ws.router)
+
+
+if __name__ == "__main__":
+    uvicorn.run("app:app", host=config.HOST, port=config.PORT, reload=True)

backend/config.py

@@ -0,0 +1,43 @@
+"""集中读取环境变量。所有 provider 的接入点都在这里,改栈只改 .env。"""
+
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+def _split(value: str) -> list[str]:
+    return [item.strip() for item in value.split(",") if item.strip()]
+
+
+# ---- LLM(DeepSeek 等,OpenAI 兼容) ----
+LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://api.deepseek.com/v1")
+LLM_API_KEY = os.getenv("LLM_API_KEY", "")
+LLM_MODEL = os.getenv("LLM_MODEL", "deepseek-chat")
+
+# ---- STT(SenseVoice / FunASR,OpenAI 兼容) ----
+STT_BASE_URL = os.getenv("STT_BASE_URL", "http://localhost:8001/v1")
+STT_API_KEY = os.getenv("STT_API_KEY", "local")
+STT_MODEL = os.getenv("STT_MODEL", "sensevoice")
+
+# ---- TTS(CosyVoice,OpenAI 兼容) ----
+TTS_BASE_URL = os.getenv("TTS_BASE_URL", "http://localhost:8002/v1")
+TTS_API_KEY = os.getenv("TTS_API_KEY", "local")
+TTS_MODEL = os.getenv("TTS_MODEL", "cosyvoice")
+TTS_VOICE = os.getenv("TTS_VOICE", "中文女")
+
+# ---- Realtime(可选) ----
+REALTIME_API_KEY = os.getenv("REALTIME_API_KEY", "")
+REALTIME_MODEL = os.getenv("REALTIME_MODEL", "gpt-realtime")
+
+# ---- 数据库(Postgres) ----
+DATABASE_URL = os.getenv(
+    "DATABASE_URL",
+    "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres",
+)
+
+# ---- 服务 ----
+HOST = os.getenv("HOST", "0.0.0.0")
+PORT = int(os.getenv("PORT", "8000"))
+CORS_ORIGINS = _split(os.getenv("CORS_ORIGINS", "http://localhost:3000"))

backend/db/models.py

@@ -0,0 +1,56 @@
+"""数据表定义(SQLAlchemy 2.0)。
+
+两张表,职责分离(见设计):
+  - ProviderCredential:模型凭证(key 明文存,同 dograh,靠 DB 访问控制兜底;读时打码)
+  - Assistant:助手配置,**只存模型/音色的"选项名",不嵌 key**
+
+助手运行时再用 kind 去 ProviderCredential 取真 key(services/config_resolver.py)。
+"""
+
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, String, func
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class ProviderCredential(Base):
+    """模型资源凭证。字段对齐前端 ComponentsModelsPage 的 ModelResource。"""
+
+    __tablename__ = "provider_credentials"
+
+    id: Mapped[str] = mapped_column(String(40), primary_key=True)   # model_xxx
+    name: Mapped[str] = mapped_column(String(128), default="")      # 资源名称,如 "DeepSeek-V3"
+    model_id: Mapped[str] = mapped_column(String(128), default="")  # 模型ID,如 "deepseek-chat"
+    type: Mapped[str] = mapped_column(String(16), index=True)       # LLM|ASR|TTS|Realtime|Embedding
+    interface_type: Mapped[str] = mapped_column(String(32), default="openai")  # openai|xfyun|dashscope|gemini
+    api_url: Mapped[str] = mapped_column(String(512), default="")
+    api_key: Mapped[str] = mapped_column(String(512), default="")   # 明文
+    # 同一 type 下的默认凭证(后端解析用;前端 ModelResource 无此字段,留作可选)
+    is_default: Mapped[bool] = mapped_column(Boolean, default=False)
+    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 Assistant(Base):
+    __tablename__ = "assistants"
+
+    id: Mapped[str] = mapped_column(String(40), primary_key=True)  # asst_xxx
+    name: Mapped[str] = mapped_column(String(128))
+    greeting: Mapped[str] = mapped_column(String(2048), default="")
+    prompt: Mapped[str] = mapped_column(String(8192), default="")
+    runtime_mode: Mapped[str] = mapped_column(String(16), default="pipeline")
+    # 模型/音色的"选项名",不是 key
+    model: Mapped[str] = mapped_column(String(128), default="")
+    asr: Mapped[str] = mapped_column(String(128), default="")
+    voice: Mapped[str] = mapped_column(String(128), default="")
+    enable_interrupt: Mapped[bool] = mapped_column(Boolean, default=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()
+    )

backend/db/session.py

@@ -0,0 +1,29 @@
+"""异步数据库引擎 + 会话。
+
+- engine / SessionLocal:全局单例
+- get_session:FastAPI 依赖,按请求注入一个会话
+- init_db:启动时建表(MVP 用 create_all;表结构稳定后切 alembic 迁移,对齐 dograh)
+"""
+
+from collections.abc import AsyncGenerator
+
+import config
+from db.models import Base
+from sqlalchemy.ext.asyncio import (
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+engine = create_async_engine(config.DATABASE_URL, echo=False, pool_pre_ping=True)
+SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+
+async def get_session() -> AsyncGenerator[AsyncSession, None]:
+    async with SessionLocal() as session:
+        yield session
+
+
+async def init_db() -> None:
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)

backend/models.py

@@ -0,0 +1,55 @@
+"""请求/配置数据模型。
+
+分两层(重要):
+  - AssistantConfig:**运行时**配置,含真 key,只在后端内部流转,绝不返回前端。
+                     由 config_resolver 从 DB 组装(或信令内联传入)。
+  - schemas.py 里的 *Request/*Response:面向前端的 DTO(key 打码)。
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+RuntimeMode = Literal["pipeline", "realtime"]
+
+
+class AssistantConfig(BaseModel):
+    """运行时配置:前端可见部分(name/prompt/...) + 服务端注入部分(*_api_key/*_base_url)。"""
+
+    name: str = "未命名助手"
+    greeting: str = "您好,我是 AI 视频助手,请问有什么可以帮您?"
+    prompt: str = "你是一个有帮助的助手。"
+    runtimeMode: RuntimeMode = "pipeline"
+
+    # 模型/音色选项
+    model: str = ""  # LLM
+    asr: str = ""    # STT
+    voice: str = ""  # TTS 音色
+    realtimeModel: str = ""
+
+    enableInterrupt: bool = True
+
+    # ---- 运行时连接信息(服务端注入,不来自浏览器) ----
+    # 为空时,service_factory 会回退到 config.py 的 .env 默认值。
+    llm_api_key: str = ""
+    llm_base_url: str = ""
+    stt_api_key: str = ""
+    stt_base_url: str = ""
+    tts_api_key: str = ""
+    tts_base_url: str = ""
+
+
+class SignalingOffer(BaseModel):
+    """WS 信令里 offer 消息的 payload。
+
+    推荐用 assistant_id(浏览器只传 id,key 在服务端解析);
+    inline_config 仅用于调试/无库场景。
+    """
+
+    pc_id: str | None = None
+    sdp: str
+    type: str
+    assistant_id: str | None = None
+    inline_config: AssistantConfig | None = None

backend/requirements.txt

@@ -0,0 +1,16 @@
+# 薄后端依赖。语音引擎用 pipecat 官方库。
+# webrtc  -> SmallWebRTCTransport / SmallWebRTCConnection + aiortc
+# silero  -> 本地 VAD(判断用户说话起止),语音必备
+# openai  -> OpenAI 兼容的 LLM/STT/TTS 客户端(DeepSeek、SenseVoice、CosyVoice 都走它)
+pipecat-ai[webrtc,silero,openai]~=0.0.60
+
+fastapi
+uvicorn[standard]
+python-dotenv
+pydantic
+loguru
+
+# 存储:Postgres(SQLAlchemy 2.0 异步 + asyncpg 驱动)
+sqlalchemy[asyncio]>=2.0
+asyncpg
+greenlet  # SQLAlchemy 异步运行时必需(部分平台不会自动带上)

backend/routes/assistants.py

@@ -0,0 +1,87 @@
+"""助手 CRUD。前端「助手列表 / 创建 / 编辑」对接这里。
+
+助手配置不含 key,所以无需打码。
+"""
+
+import uuid
+
+from db.models import Assistant
+from db.session import get_session
+from fastapi import APIRouter, Depends, HTTPException
+from schemas import AssistantOut, AssistantUpsert
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+router = APIRouter(prefix="/api/assistants", tags=["assistants"])
+
+
+def _to_out(a: Assistant) -> AssistantOut:
+    return AssistantOut(
+        id=a.id,
+        name=a.name,
+        greeting=a.greeting,
+        prompt=a.prompt,
+        runtime_mode=a.runtime_mode,  # type: ignore[arg-type]
+        model=a.model,
+        asr=a.asr,
+        voice=a.voice,
+        enable_interrupt=a.enable_interrupt,
+        updated_at=a.updated_at.isoformat() if a.updated_at else None,
+    )
+
+
+@router.get("", response_model=list[AssistantOut])
+async def list_assistants(session: AsyncSession = Depends(get_session)):
+    rows = (
+        await session.execute(select(Assistant).order_by(Assistant.updated_at.desc()))
+    ).scalars().all()
+    return [_to_out(a) for a in rows]
+
+
+@router.post("", response_model=AssistantOut)
+async def create_assistant(
+    body: AssistantUpsert, session: AsyncSession = Depends(get_session)
+):
+    a = Assistant(id=f"asst_{uuid.uuid4().hex[:12]}", **body.model_dump())
+    session.add(a)
+    await session.commit()
+    await session.refresh(a)
+    return _to_out(a)
+
+
+@router.get("/{assistant_id}", response_model=AssistantOut)
+async def get_assistant(
+    assistant_id: str, session: AsyncSession = Depends(get_session)
+):
+    a = await session.get(Assistant, assistant_id)
+    if not a:
+        raise HTTPException(404, "助手不存在")
+    return _to_out(a)
+
+
+@router.put("/{assistant_id}", response_model=AssistantOut)
+async def update_assistant(
+    assistant_id: str,
+    body: AssistantUpsert,
+    session: AsyncSession = Depends(get_session),
+):
+    a = await session.get(Assistant, assistant_id)
+    if not a:
+        raise HTTPException(404, "助手不存在")
+    for k, v in body.model_dump().items():
+        setattr(a, k, v)
+    await session.commit()
+    await session.refresh(a)
+    return _to_out(a)
+
+
+@router.delete("/{assistant_id}")
+async def delete_assistant(
+    assistant_id: str, session: AsyncSession = Depends(get_session)
+):
+    a = await session.get(Assistant, assistant_id)
+    if not a:
+        raise HTTPException(404, "助手不存在")
+    await session.delete(a)
+    await session.commit()
+    return {"ok": True}

backend/routes/credentials.py

@@ -0,0 +1,106 @@
+"""模型资源凭证 CRUD。前端 ComponentsModelsPage 对接这里。
+
+字段对齐前端 ModelResource。读:api_key 打码;写:占位符表示不改(写时哨兵)。
+设默认时清掉同 type 其它默认。
+"""
+
+import uuid
+
+from db.models import ProviderCredential
+from db.session import get_session
+from fastapi import APIRouter, Depends, HTTPException
+from schemas import CredentialOut, CredentialUpsert
+from services.masking import mask, resolve_incoming_key
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+router = APIRouter(prefix="/api/credentials", tags=["credentials"])
+
+
+def _to_out(c: ProviderCredential) -> CredentialOut:
+    return CredentialOut(
+        id=c.id,
+        name=c.name,
+        model_id=c.model_id,
+        type=c.type,
+        interface_type=c.interface_type,
+        api_url=c.api_url,
+        api_key=mask(c.api_key),  # 永远打码
+        is_default=c.is_default,
+    )
+
+
+async def _clear_other_defaults(session: AsyncSession, type_: str, keep_id: str):
+    await session.execute(
+        update(ProviderCredential)
+        .where(ProviderCredential.type == type_, ProviderCredential.id != keep_id)
+        .values(is_default=False)
+    )
+
+
+@router.get("", response_model=list[CredentialOut])
+async def list_credentials(session: AsyncSession = Depends(get_session)):
+    rows = (
+        await session.execute(
+            select(ProviderCredential).order_by(ProviderCredential.type)
+        )
+    ).scalars().all()
+    return [_to_out(c) for c in rows]
+
+
+@router.post("", response_model=CredentialOut)
+async def create_credential(
+    body: CredentialUpsert, session: AsyncSession = Depends(get_session)
+):
+    c = ProviderCredential(
+        id=f"model_{uuid.uuid4().hex[:12]}",
+        name=body.name,
+        model_id=body.model_id,
+        type=body.type,
+        interface_type=body.interface_type,
+        api_url=body.api_url,
+        api_key=resolve_incoming_key(body.api_key, ""),
+        is_default=body.is_default,
+    )
+    session.add(c)
+    if c.is_default:
+        await _clear_other_defaults(session, c.type, c.id)
+    await session.commit()
+    await session.refresh(c)
+    return _to_out(c)
+
+
+@router.put("/{cred_id}", response_model=CredentialOut)
+async def update_credential(
+    cred_id: str,
+    body: CredentialUpsert,
+    session: AsyncSession = Depends(get_session),
+):
+    c = await session.get(ProviderCredential, cred_id)
+    if not c:
+        raise HTTPException(404, "凭证不存在")
+    c.name = body.name
+    c.model_id = body.model_id
+    c.type = body.type
+    c.interface_type = body.interface_type
+    c.api_url = body.api_url
+    c.is_default = body.is_default
+    # 写时哨兵:打码占位符 → 保留旧 key
+    c.api_key = resolve_incoming_key(body.api_key, c.api_key)
+    if c.is_default:
+        await _clear_other_defaults(session, c.type, c.id)
+    await session.commit()
+    await session.refresh(c)
+    return _to_out(c)
+
+
+@router.delete("/{cred_id}")
+async def delete_credential(
+    cred_id: str, session: AsyncSession = Depends(get_session)
+):
+    c = await session.get(ProviderCredential, cred_id)
+    if not c:
+        raise HTTPException(404, "凭证不存在")
+    await session.delete(c)
+    await session.commit()
+    return {"ok": True}

backend/routes/health.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/health")
+async def health():
+    return {"status": "ok"}

backend/routes/voice_webrtc.py

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

backend/routes/voice_ws.py

@@ -0,0 +1,50 @@
+"""WS 输出:裸 WebSocket 音频流(第二种输出方式)。
+
+比 WebRTC 简单——没有 SDP/ICE/STUN/TURN,一条 WS 直接收发音频帧。
+适合:服务端对接、话务网关、自定义客户端、调试。
+
+约定:连接建立后,**第一条文本消息**发 JSON 启动参数:
+    {"assistant_id": "asst_xxx"}            # 推荐:key 服务端解析
+    {"inline_config": {...AssistantConfig}} # 调试:内联
+之后的二进制消息即音频帧(protobuf,与 transports.py serializer 对应)。
+"""
+
+import json
+
+from db.session import SessionLocal
+from fastapi import APIRouter, WebSocket
+from loguru import logger
+from models import AssistantConfig
+from services.config_resolver import resolve_runtime_config
+from starlette.websockets import WebSocketDisconnect
+
+# pipecat 重依赖,惰性导入(见 voice_webrtc.py 说明)
+
+router = APIRouter()
+
+
+async def _resolve_start_config(raw: str) -> AssistantConfig:
+    data = json.loads(raw)
+    if data.get("assistant_id"):
+        async with SessionLocal() as session:
+            return await resolve_runtime_config(session, data["assistant_id"])
+    if data.get("inline_config"):
+        return AssistantConfig(**data["inline_config"])
+    raise ValueError("启动参数缺少 assistant_id 或 inline_config")
+
+
+@router.websocket("/ws/stream")
+async def voice_stream(websocket: WebSocket):
+    from services.pipecat.pipeline import run_pipeline
+    from services.pipecat.transports import build_ws_transport
+
+    await websocket.accept()
+    try:
+        cfg = await _resolve_start_config(await websocket.receive_text())
+        transport = build_ws_transport(websocket)
+        # 直接 await:管线持续读这条 WS 的音频帧,直到对端断开
+        await run_pipeline(transport, cfg)
+    except WebSocketDisconnect:
+        logger.info("WS 音频流断开")
+    except Exception as e:
+        logger.error(f"WS 音频流出错: {e}")

backend/schemas.py

@@ -0,0 +1,66 @@
+"""面向前端的请求/响应 DTO。与 DB 模型解耦,**响应里的 key 一律打码**。
+
+凭证 DTO 字段对齐前端 ComponentsModelsPage 的 ModelResource:
+JSON 用 camelCase(modelId/interfaceType/apiUrl/apiKey),Python 内部用 snake_case,
+靠 Pydantic alias 自动互转。FastAPI 响应默认 by_alias=True,所以出参也是 camelCase。
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+RuntimeMode = Literal["pipeline", "realtime"]
+ModelType = Literal["LLM", "ASR", "TTS", "Realtime", "Embedding"]
+InterfaceType = Literal["openai", "xfyun", "dashscope", "gemini"]
+
+
+class CamelModel(BaseModel):
+    """JSON camelCase ↔ Python snake_case。protected_namespaces 关掉以允许 model_id。"""
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+        protected_namespaces=(),
+    )
+
+
+# ---------- 助手 ----------
+class AssistantUpsert(CamelModel):
+    name: str
+    greeting: str = ""
+    prompt: str = ""
+    runtime_mode: RuntimeMode = "pipeline"
+    model: str = ""
+    asr: str = ""
+    voice: str = ""
+    enable_interrupt: bool = True
+
+
+class AssistantOut(AssistantUpsert):
+    id: str
+    updated_at: str | None = None
+
+
+# ---------- 模型凭证(对齐前端 ModelResource) ----------
+class CredentialUpsert(CamelModel):
+    name: str = ""                              # 资源名称
+    model_id: str = ""                          # 模型ID
+    type: ModelType                             # LLM/ASR/TTS/Realtime/Embedding
+    interface_type: InterfaceType = "openai"    # openai/xfyun/dashscope/gemini
+    api_url: str = ""
+    api_key: str = ""                           # 写时:占位符/空表示不改
+    is_default: bool = False
+
+
+class CredentialOut(CamelModel):
+    id: str
+    name: str
+    model_id: str
+    type: str
+    interface_type: str
+    api_url: str
+    api_key: str                                # 读时:打码后的值
+    is_default: bool

backend/services/config_resolver.py

@@ -0,0 +1,64 @@
+"""assistant_id → 运行时配置(把真 key 在服务端组装好)。
+
+浏览器只传 assistant_id;真 key 在这里从 provider_credentials 取出注入。
+取不到凭证记录时,降级用 .env 默认值(开发期零配置仍能跑)。
+"""
+
+import config
+from db.models import Assistant, ProviderCredential
+from models import AssistantConfig
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _get_credential(
+    session: AsyncSession, type_: str, name: str = ""
+) -> ProviderCredential | None:
+    """取某类(LLM/ASR/TTS)凭证:优先按资源名匹配,否则取该类默认。"""
+    stmt = select(ProviderCredential).where(ProviderCredential.type == type_)
+    if name:
+        # 助手按资源名引用(如 model="DeepSeek-V3");命中则用它
+        named = (
+            await session.execute(stmt.where(ProviderCredential.name == name).limit(1))
+        ).scalar_one_or_none()
+        if named:
+            return named
+    stmt = stmt.order_by(
+        ProviderCredential.is_default.desc(), ProviderCredential.id.asc()
+    ).limit(1)
+    return (await session.execute(stmt)).scalar_one_or_none()
+
+
+async def resolve_runtime_config(
+    session: AsyncSession, assistant_id: str
+) -> AssistantConfig:
+    """加载助手 + 解析凭证,产出可直接交给管线的运行时配置(含真 key)。
+
+    type 映射:LLM→大模型, ASR→语音识别, TTS→语音合成。
+    """
+    assistant = await session.get(Assistant, assistant_id)
+    if assistant is None:
+        raise ValueError(f"助手不存在: {assistant_id}")
+
+    llm = await _get_credential(session, "LLM", assistant.model)
+    stt = await _get_credential(session, "ASR", assistant.asr)
+    tts = await _get_credential(session, "TTS")
+
+    return AssistantConfig(
+        name=assistant.name,
+        greeting=assistant.greeting,
+        prompt=assistant.prompt,
+        runtimeMode=assistant.runtime_mode,  # type: ignore[arg-type]
+        enableInterrupt=assistant.enable_interrupt,
+        # 模型/音色:凭证的模型ID优先,否则助手里填的
+        model=(llm.model_id if llm else assistant.model),
+        asr=(stt.model_id if stt else assistant.asr),
+        voice=assistant.voice,
+        # 运行时连接信息(真 key + url):凭证优先,否则 .env 兜底
+        llm_api_key=(llm.api_key if llm else config.LLM_API_KEY),
+        llm_base_url=(llm.api_url if llm else config.LLM_BASE_URL),
+        stt_api_key=(stt.api_key if stt else config.STT_API_KEY),
+        stt_base_url=(stt.api_url if stt else config.STT_BASE_URL),
+        tts_api_key=(tts.api_key if tts else config.TTS_API_KEY),
+        tts_base_url=(tts.api_url if tts else config.TTS_BASE_URL),
+    )

backend/services/masking.py

@@ -0,0 +1,27 @@
+"""API Key 打码 / 写时哨兵(抄 dograh masking.py + merge.py 思路)。
+
+- mask:返回前端时把真 key 变成 sk-****1234,真 key 永不出后端
+- is_masked:判断前端回传的是不是打码占位符
+- resolve_incoming_key:前端回传若是占位符 → 保留旧值;否则用新值
+"""
+
+MASK_VISIBLE_TAIL = 4
+
+
+def mask(api_key: str) -> str:
+    if not api_key:
+        return ""
+    if len(api_key) <= MASK_VISIBLE_TAIL:
+        return "****"
+    return f"{api_key[:2]}****{api_key[-MASK_VISIBLE_TAIL:]}"
+
+
+def is_masked(value: str) -> bool:
+    return "****" in (value or "")
+
+
+def resolve_incoming_key(incoming: str | None, stored: str) -> str:
+    """写入时决定最终 key:占位符/空 → 保留旧;否则用新。"""
+    if incoming is None or incoming == "" or is_masked(incoming):
+        return stored
+    return incoming

backend/services/pipecat/pipeline.py

@@ -0,0 +1,67 @@
+"""管线核心:给定一个 transport + 配置,跑完整的语音闭环。
+
+关键设计:**transport 由调用方传入**,管线本身不关心是 WebRTC 还是 WS。
+这就是"同时支持多种输出"的落点——加输出方式不用动这里。
+
+对应 dograh 的 pipeline_builder.py + run_pipeline.py(已砍掉 workflow 引擎/DB/录音/指标)。
+"""
+
+from loguru import logger
+from models import AssistantConfig
+from services.pipecat.service_factory import create_services
+
+from pipecat.frames.frames import EndFrame, TTSSpeakFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+
+
+async def run_pipeline(transport, cfg: AssistantConfig) -> None:
+    """在给定 transport 上构建并运行管线,直到连接结束。
+
+    Args:
+        transport: 任意 pipecat transport(WebRTC / WS / 电话…),
+                   只要有 .input() / .output() / event_handler 即可。
+        cfg:       助手配置(随请求内联传入)。
+    """
+    logger.info(f"启动管线: assistant={cfg.name} mode={cfg.runtimeMode}")
+
+    stt, llm, tts = create_services(cfg)
+
+    context = OpenAILLMContext(messages=[{"role": "system", "content": cfg.prompt}])
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            context_aggregator.user(),
+            llm,
+            tts,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=cfg.enableInterrupt,
+            enable_metrics=False,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(_transport, _client):
+        if cfg.greeting:
+            await task.queue_frame(TTSSpeakFrame(cfg.greeting))
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(_transport, _client):
+        logger.info("对端断开,结束管线")
+        await task.queue_frame(EndFrame())
+
+    runner = PipelineRunner(handle_sigint=False)
+    await runner.run(task)
+    logger.info("管线已结束")

backend/services/pipecat/service_factory.py

@@ -0,0 +1,53 @@
+"""创建 STT / LLM / TTS 服务。
+
+对应 dograh 的 service_factory.py,但只留一套国产栈(OpenAI 兼容),
+按 provider 扩展时在这里加分支即可——这是未来接更多模型的唯一入口。
+"""
+
+import config
+from loguru import logger
+from models import AssistantConfig
+
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.openai.stt import OpenAISTTService
+from pipecat.services.openai.tts import OpenAITTSService
+
+
+def create_stt(cfg: AssistantConfig):
+    """SenseVoice / FunASR 等,走 OpenAI 兼容的 /v1/audio/transcriptions。
+
+    连接信息优先用 cfg(由 config_resolver 从 DB 注入),为空回退 .env 默认。
+    """
+    return OpenAISTTService(
+        api_key=cfg.stt_api_key or config.STT_API_KEY,
+        base_url=cfg.stt_base_url or config.STT_BASE_URL,
+        model=cfg.asr or config.STT_MODEL,
+    )
+
+
+def create_llm(cfg: AssistantConfig):
+    """DeepSeek 等,走 OpenAI 兼容的 /v1/chat/completions。"""
+    return OpenAILLMService(
+        api_key=cfg.llm_api_key or config.LLM_API_KEY,
+        base_url=cfg.llm_base_url or config.LLM_BASE_URL,
+        model=cfg.model or config.LLM_MODEL,
+    )
+
+
+def create_tts(cfg: AssistantConfig):
+    """CosyVoice 等,走 OpenAI 兼容的 /v1/audio/speech。"""
+    return OpenAITTSService(
+        api_key=cfg.tts_api_key or config.TTS_API_KEY,
+        base_url=cfg.tts_base_url or config.TTS_BASE_URL,
+        model=config.TTS_MODEL,
+        voice=cfg.voice or config.TTS_VOICE,
+    )
+
+
+def create_services(cfg: AssistantConfig):
+    logger.info(
+        f"创建服务: stt={cfg.asr or config.STT_MODEL} "
+        f"llm={cfg.model or config.LLM_MODEL} "
+        f"tts={cfg.voice or config.TTS_VOICE}"
+    )
+    return create_stt(cfg), create_llm(cfg), create_tts(cfg)

backend/services/pipecat/transports.py

@@ -0,0 +1,54 @@
+"""Transport 工厂——管线与"输出方式"解耦的关键。
+
+同一条 STT→LLM→TTS 管线,可以挂在不同 transport 上:
+  - WebRTC:浏览器,低延迟,带 NAT 穿透           -> build_webrtc_transport
+  - WS:    裸音频流,服务端/话务/自定义客户端,简单 -> build_ws_transport
+
+未来加电话(Twilio/Vonage)只是再加一个 build_xxx_transport + 对应 serializer。
+对应 dograh 的 transport_setup.py(WebRTC)+ 各 telephony provider 的 transport.py(WS)。
+"""
+
+from fastapi import WebSocket
+
+from pipecat.transports.base_transport import TransportParams
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+
+# WebRTC
+from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
+from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport
+
+# 裸 WS 音频流
+from pipecat.transports.network.fastapi_websocket import (
+    FastAPIWebsocketTransport,
+    FastAPIWebsocketParams,
+)
+from pipecat.serializers.protobuf import ProtobufFrameSerializer
+
+
+def _base_params() -> dict:
+    """两种 transport 共享的音频参数。"""
+    return dict(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),  # 本地 VAD,打断功能依赖它
+    )
+
+
+def build_webrtc_transport(connection: SmallWebRTCConnection) -> SmallWebRTCTransport:
+    return SmallWebRTCTransport(
+        webrtc_connection=connection,
+        params=TransportParams(**_base_params()),
+    )
+
+
+def build_ws_transport(websocket: WebSocket) -> FastAPIWebsocketTransport:
+    """裸 WS 输出。序列化用 protobuf(自定义客户端用同款解码);
+    若对接电话商,把 serializer 换成对应的 TwilioFrameSerializer 等即可。
+    """
+    return FastAPIWebsocketTransport(
+        websocket=websocket,
+        params=FastAPIWebsocketParams(
+            serializer=ProtobufFrameSerializer(),
+            **_base_params(),
+        ),
+    )