ai-video-fullstack/backend/models.py

"""请求/配置数据模型。

分两层(重要):
  - 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 RuntimeMcpServer(BaseModel):
    """Resolved MCP connection containing secrets for one runtime session."""

    id: str
    name: str = ""
    transport: str = "streamable_http"
    url: str
    timeout_seconds: int = 30
    headers: dict[str, str] = Field(default_factory=dict)
    secret_headers: dict[str, str] = Field(default_factory=dict)


class RuntimeTool(BaseModel):
    """Tool data resolved from an assistant binding for one runtime session."""

    id: str
    name: str
    function_name: str
    type: str
    description: str = ""
    definition: dict = Field(default_factory=dict)
    secrets: dict = Field(default_factory=dict)
    mcp_server: RuntimeMcpServer | None = None


class RuntimeModelResource(BaseModel):
    id: str
    name: str = ""
    capability: str
    interface_type: str
    values: dict = Field(default_factory=dict)
    secrets: dict = Field(default_factory=dict)
    support_image_input: bool = False


class RuntimeKnowledgeBase(BaseModel):
    id: str
    name: str = ""
    description: str = ""


class AssistantConfig(BaseModel):
    """运行时配置:前端可见部分(name/prompt/...) + 服务端注入部分(*_api_key/*_base_url)。"""

    name: str = "未命名助手"
    # prompt|workflow|dify|fastgpt|opencode;决定由哪种「大脑」驱动对话
    type: str = "prompt"
    greeting: str = "您好,我是 AI 视频助手,请问有什么可以帮您?"
    prompt: str = "你是一个有帮助的助手。"
    dynamic_variable_definitions: dict = Field(default_factory=dict)
    dynamic_variables: dict = Field(default_factory=dict)
    secret_dynamic_variables: dict = Field(default_factory=dict)
    conversation_id: str = ""
    runtimeMode: RuntimeMode = "pipeline"

    # 模型/音色选项
    model: str = ""  # LLM
    asr: str = ""    # STT
    tts_model: str = ""
    voice: str = ""  # TTS 音色
    stt_language: str = ""
    tts_speed: float = 1.0
    realtimeModel: str = ""
    realtime_interface_type: str = ""
    realtime_values: dict = {}
    realtime_secrets: dict = {}
    agent_interface_type: str = ""
    agent_values: dict = {}
    agent_secrets: dict = {}
    llm_interface_type: str = "openai-llm"
    stt_interface_type: str = "openai-asr"
    tts_interface_type: str = "openai-tts"
    llm_values: dict = {}
    llm_secrets: dict = {}
    llm_support_image_input: bool = False
    vision_enabled: bool = False
    vision_model_resource_id: str | None = None
    vision_model: str = ""
    vision_llm_interface_type: str = "openai-llm"
    vision_llm_values: dict = {}
    vision_llm_secrets: dict = {}
    vision_llm_support_image_input: bool = False
    vision_llm_api_key: str = ""
    vision_llm_base_url: str = ""
    stt_values: dict = {}
    stt_secrets: dict = {}
    tts_values: dict = {}
    tts_secrets: dict = {}

    enableInterrupt: bool = True
    turnConfig: dict = Field(default_factory=dict)
    startup: dict = Field(default_factory=dict)

    # ``tools`` is the complete runtime pool (conversation + lifecycle actions).
    # ``llm_tool_ids`` limits which tools are advertised to a Prompt model. None
    # preserves compatibility for inline/test configs that historically exposed
    # every item in ``tools``.
    tools: list[RuntimeTool] = Field(default_factory=list)
    llm_tool_ids: list[str] | None = None
    # 助手级系统工具仅供 Prompt Pipeline；Workflow 在 Agent 节点中配置。
    system_tools: list[str] = Field(default_factory=list)
    knowledge_base_id: str | None = None
    knowledge_base_name: str = ""
    knowledge_base_description: str = ""
    knowledge_retrieval_config: dict = Field(
        default_factory=lambda: {
            "mode": "automatic",
            "top_n": 5,
            "score_threshold": 0.0,
        }
    )

    # workflow 类型:节点图(nodes/edges)。非 workflow 为空,引擎据此决定是否启用。
    graph: dict = {}
    workflow_model_resources: dict[str, RuntimeModelResource] = Field(default_factory=dict)
    workflow_knowledge_bases: dict[str, RuntimeKnowledgeBase] = Field(default_factory=dict)

    # 外部托管类型(fastgpt/dify/opencode)的连接信息:context/KB/tools 由对方服务端接管。
    dify_api_url: str = ""
    dify_api_key: str = ""
    fastgpt_api_url: str = ""
    fastgpt_api_key: str = ""
    fastgpt_app_id: str = ""

    # ---- 运行时连接信息(服务端注入,不来自浏览器) ----
    # 由模型资源注入;调试 inline_config 也必须显式提供。
    llm_api_key: str = ""
    llm_base_url: str = ""
    stt_api_key: str = ""
    stt_base_url: str = ""
    tts_api_key: str = ""
    tts_base_url: str = ""
    realtime_api_key: str = ""
    realtime_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
    vision_enabled: bool = False
    dynamic_variables: dict = Field(default_factory=dict)