ai-video-fullstack/backend/services/config_resolver.py

"""assistant_id → 运行时配置(把真 key 在服务端组装好)。

浏览器只传 assistant_id;真 key 在这里从 model_resources 取出注入。
助手按 capability binding 引用资源;取不到则回退该能力默认资源,再回退 .env。
"""

import config
from db.models import Assistant, AssistantModelBinding, ModelResource
from models import AssistantConfig
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession


async def _resource_for(
    session: AsyncSession,
    assistant_id: str,
    capability: str,
) -> ModelResource | None:
    binding = await session.get(AssistantModelBinding, (assistant_id, capability))
    resource_id = binding.model_resource_id if binding else None
    resource = await session.get(ModelResource, resource_id) if resource_id else None
    if resource and resource.capability != capability:
        resource = None
    if resource is None:
        resource = (
            await session.execute(
                select(ModelResource)
                .where(ModelResource.capability == capability, ModelResource.enabled.is_(True))
                .order_by(ModelResource.is_default.desc(), ModelResource.id.asc())
                .limit(1)
            )
        ).scalar_one_or_none()
    return resource


def _value(resource: ModelResource | None, key: str, default):
    if not resource:
        return default
    value = (resource.values or {}).get(key, default)
    return default if value is None else value


def _secret(resource: ModelResource | None, key: str, default: str) -> str:
    if not resource:
        return default
    return str((resource.secrets or {}).get(key) or default)


async def resolve_runtime_config(
    session: AsyncSession, assistant_id: str
) -> AssistantConfig:
    """加载助手 + 解析模型资源,产出可直接交给管线的运行时配置(含真 key)。"""
    assistant = await session.get(Assistant, assistant_id)
    if assistant is None:
        raise ValueError(f"助手不存在: {assistant_id}")

    llm_resource = await _resource_for(session, assistant.id, "LLM")
    stt_resource = await _resource_for(session, assistant.id, "ASR")
    tts_resource = await _resource_for(session, assistant.id, "TTS")
    realtime_resource = await _resource_for(session, assistant.id, "Realtime")
    vision_resource = (
        await session.get(ModelResource, assistant.vision_model_resource_id)
        if assistant.vision_model_resource_id
        else llm_resource
    )

    return AssistantConfig(
        name=assistant.name,
        type=assistant.type,
        greeting=assistant.greeting,
        # prompt 现在是真列;外部类型由其平台编排,这里给个兜底
        prompt=assistant.prompt or "你是一个有帮助的助手。",
        runtimeMode=assistant.runtime_mode,  # type: ignore[arg-type]
        enableInterrupt=assistant.enable_interrupt,
        # workflow 图:仅 workflow 类型非空,引擎据此启用图驱动对话
        graph=(assistant.graph or {}) if assistant.type == "workflow" else {},
        # 外部托管类型连接信息(DB 存真 key,直接注入)
        fastgpt_api_url=assistant.api_url,
        fastgpt_api_key=assistant.api_key,
        fastgpt_app_id=assistant.app_id,
        # 模型/音色:模型资源中的配置优先
        model=str(_value(llm_resource, "modelId", "")),
        asr=str(_value(stt_resource, "modelId", "")),
        tts_model=str(_value(tts_resource, "modelId", "")),
        voice=str(_value(tts_resource, "voice", "")),
        stt_language=str(_value(stt_resource, "language", "")),
        tts_speed=float(_value(tts_resource, "speed", 1.0)),
        llm_interface_type=(llm_resource.interface_type if llm_resource else "openai-llm"),
        stt_interface_type=(stt_resource.interface_type if stt_resource else "openai-asr"),
        tts_interface_type=(tts_resource.interface_type if tts_resource else "openai-tts"),
        realtimeModel=str(_value(realtime_resource, "modelId", "")),
        llm_values=(llm_resource.values or {}) if llm_resource else {},
        llm_secrets=(llm_resource.secrets or {}) if llm_resource else {},
        llm_support_image_input=(
            bool(llm_resource.support_image_input) if llm_resource else False
        ),
        vision_enabled=assistant.vision_enabled,
        vision_model_resource_id=assistant.vision_model_resource_id,
        vision_model=str(_value(vision_resource, "modelId", "")),
        vision_llm_interface_type=(
            vision_resource.interface_type if vision_resource else "openai-llm"
        ),
        vision_llm_values=(vision_resource.values or {}) if vision_resource else {},
        vision_llm_secrets=(vision_resource.secrets or {}) if vision_resource else {},
        vision_llm_support_image_input=(
            bool(vision_resource.support_image_input) if vision_resource else False
        ),
        stt_values=(stt_resource.values or {}) if stt_resource else {},
        stt_secrets=(stt_resource.secrets or {}) if stt_resource else {},
        tts_values=(tts_resource.values or {}) if tts_resource else {},
        tts_secrets=(tts_resource.secrets or {}) if tts_resource else {},
        realtime_interface_type=(
            realtime_resource.interface_type if realtime_resource else ""
        ),
        realtime_values=(realtime_resource.values or {}) if realtime_resource else {},
        realtime_secrets=(realtime_resource.secrets or {}) if realtime_resource else {},
        # 运行时连接信息(真 key + url):模型资源优先,否则 .env 兜底
        llm_api_key=_secret(llm_resource, "apiKey", config.LLM_API_KEY),
        llm_base_url=str(_value(llm_resource, "apiUrl", config.LLM_BASE_URL)),
        vision_llm_api_key=_secret(
            vision_resource, "apiKey", config.LLM_API_KEY
        ),
        vision_llm_base_url=str(
            _value(vision_resource, "apiUrl", config.LLM_BASE_URL)
        ),
        stt_api_key=_secret(stt_resource, "apiKey", config.STT_API_KEY),
        stt_base_url=str(_value(stt_resource, "apiUrl", config.STT_BASE_URL)),
        tts_api_key=_secret(tts_resource, "apiKey", config.TTS_API_KEY),
        tts_base_url=str(_value(tts_resource, "apiUrl", config.TTS_BASE_URL)),
        realtime_api_key=_secret(realtime_resource, "apiKey", ""),
        realtime_base_url=str(_value(realtime_resource, "apiUrl", "")),
    )