"""面向前端的请求/响应 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 datetime import datetime from typing import Annotated, Any, Literal, Union from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from pydantic.alias_generators import to_camel RuntimeMode = Literal["pipeline", "realtime"] ModelType = Literal["LLM", "ASR", "TTS", "Realtime", "Embedding", "Agent"] AssistantType = Literal["prompt", "workflow", "dify", "fastgpt", "opencode"] ToolType = Literal["end_call", "http"] ToolStatus = Literal["active", "archived", "draft"] ToolParameterType = Literal["string", "number", "integer", "boolean", "object", "array"] ToolParameterLocation = Literal["path", "query", "body", "header"] # 外部应用类型:其 config.apiKey 是该助手私有密钥,读时打码 / 写时哨兵 EXTERNAL_TYPES = {"dify", "fastgpt", "opencode"} # 支持 realtime(语音到语音)的类型;外部托管大脑只能走 cascade。 # 与 services.brains 各 BrainSpec.supported_runtime_modes 对齐(此处独立声明, # 避免 HTTP schema 层为做校验而引入 pipecat 重依赖)。 REALTIME_CAPABLE_TYPES = {"prompt", "workflow"} 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=(), ) # 各 type 允许的瘦字段(其余字段写入时清零,防止跨类型脏数据) ALLOWED_FIELDS: dict[str, set[str]] = { "prompt": {"prompt"}, "workflow": {"graph"}, "dify": {"api_url", "api_key"}, "fastgpt": {"app_id", "api_url", "api_key"}, "opencode": {"prompt", "api_url", "api_key"}, } # ---------- 助手(单表 STI:瘦类型真列 + workflow 图 JSON 列) ---------- class AssistantUpsert(CamelModel): name: str type: AssistantType = "prompt" runtime_mode: RuntimeMode = "pipeline" greeting: str = "" enable_interrupt: bool = True vision_enabled: bool = False vision_model_resource_id: str | None = None model_resource_ids: dict[ModelType, str] = Field(default_factory=dict) knowledge_base_id: str | None = None tool_ids: list[str] = Field(default_factory=list) # 瘦类型专属(真列);按 type 取用,无关字段写入时清零 prompt: str = "" api_url: str = "" api_key: str = "" # 写时:占位符/空 → 保留旧(哨兵) app_id: str = "" # workflow 专属:图 graph: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def _strip_irrelevant_fields(self): allowed = ALLOWED_FIELDS[self.type] for field in ("prompt", "api_url", "api_key", "app_id"): if field not in allowed: setattr(self, field, "") if "graph" not in allowed: self.graph = {} if self.type != "prompt": self.tool_ids = [] # 外部托管大脑只能 cascade,拦住不兼容的 realtime if self.runtime_mode == "realtime" and self.type not in REALTIME_CAPABLE_TYPES: raise ValueError(f"类型 {self.type} 不支持 realtime 运行模式") return self class AssistantOut(AssistantUpsert): id: str updated_at: str | None = None # ---------- 可复用工具 ---------- class ToolParameter(CamelModel): name: str type: ToolParameterType = "string" location: ToolParameterLocation = "body" description: str = "" required: bool = True class EndCallToolConfig(CamelModel): message_type: Literal["none", "custom"] = "none" custom_message: str = "" capture_reason: bool = True class HttpToolConfig(CamelModel): method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "GET" url: str timeout_seconds: int = Field(default=15, ge=1, le=120) headers: dict[str, str] = Field(default_factory=dict) parameters: list[ToolParameter] = Field(default_factory=list) body: dict[str, Any] = Field(default_factory=dict) @field_validator("url") @classmethod def validate_url(cls, value: str) -> str: if not value.startswith(("http://", "https://")): raise ValueError("HTTP 工具 URL 必须使用 http:// 或 https://") return value class EndCallToolDefinition(CamelModel): schema_version: int = 1 type: Literal["end_call"] = "end_call" config: EndCallToolConfig = Field(default_factory=EndCallToolConfig) class HttpToolDefinition(CamelModel): schema_version: int = 1 type: Literal["http"] = "http" config: HttpToolConfig ToolDefinition = Annotated[ Union[EndCallToolDefinition, HttpToolDefinition], Field(discriminator="type") ] class ToolUpsert(CamelModel): name: str = Field(min_length=1, max_length=128) function_name: str = Field(pattern=r"^[a-z][a-z0-9_]{0,63}$") description: str = Field(default="", max_length=2048) definition: ToolDefinition secrets: dict[str, Any] = Field(default_factory=dict) status: ToolStatus = "active" class ToolOut(ToolUpsert): id: str type: ToolType updated_at: str | None = None # ---------- 知识库 ---------- class KnowledgeBaseUpsert(CamelModel): name: str description: str = "" embedding_model_resource_id: str | None = None class KnowledgeBaseOut(KnowledgeBaseUpsert): id: str status: str = "active" updated_at: str | None = None # ---------- 接口定义驱动的统一模型资源 ---------- class InterfaceDefinitionOut(CamelModel): interface_type: str name: str capability: ModelType field_schema: dict[str, Any] enabled: bool version: int class ModelResourceUpsert(CamelModel): name: str interface_type: str values: dict[str, Any] = Field(default_factory=dict) secrets: dict[str, Any] = Field(default_factory=dict) support_image_input: bool = False enabled: bool = True is_default: bool = False class ModelResourceOut(ModelResourceUpsert): id: str capability: ModelType updated_at: str | None = None class ModelResourceTestResult(CamelModel): ok: bool latency_ms: int | None = None message: str detail: str = "" # ---------- 对话历史 ---------- class ConversationMessageOut(CamelModel): id: str sequence: int role: Literal["user", "assistant"] content_type: str content: str occurred_at: datetime extra: dict[str, Any] = Field(default_factory=dict) class ConversationOut(CamelModel): id: str assistant_id: str | None assistant_name: str channel: str runtime_mode: str status: str message_count: int started_at: datetime ended_at: datetime | None class ConversationDetailOut(ConversationOut): messages: list[ConversationMessageOut] = Field(default_factory=list) class ConversationListOut(CamelModel): items: list[ConversationOut] total: int page: int page_size: int