diff --git a/backend/db/models.py b/backend/db/models.py index f8fbb14..f12fb63 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -153,6 +153,7 @@ class Assistant(Base): greeting: Mapped[str] = mapped_column(String(2048), default="") enable_interrupt: Mapped[bool] = mapped_column(Boolean, default=True) turn_config: Mapped[dict] = mapped_column(JSON, default=dict) + startup: Mapped[dict] = mapped_column(JSON, default=dict) vision_enabled: Mapped[bool] = mapped_column(Boolean, default=False) vision_model_resource_id: Mapped[str | None] = mapped_column( String(40), diff --git a/backend/db/session.py b/backend/db/session.py index 000916c..58ff9eb 100644 --- a/backend/db/session.py +++ b/backend/db/session.py @@ -49,35 +49,89 @@ async def sync_interface_definitions() -> None: async def sync_default_tools() -> None: """Ensure system-provided reusable tools exist without overwriting edits.""" - async with engine.begin() as conn: - await conn.execute( - text( - "INSERT INTO tools " - "(id, name, function_name, type, description, definition, secrets, status) " - "VALUES (" - ":id, :name, :function_name, :type, :description, " - "CAST(:definition AS jsonb), CAST(:secrets AS jsonb), :status" - ") " - "ON CONFLICT (function_name) DO NOTHING" - ), - { - "id": "tool_end_call_default", - "name": "结束对话", - "function_name": "end_call", + default_tools = [ + { + "id": "tool_end_call_default", + "name": "结束对话", + "function_name": "end_call", + "type": "end_call", + "description": "当用户明确要求结束对话，或任务已完成时调用。", + "definition": { + "schema_version": 1, "type": "end_call", - "description": "当用户明确要求结束对话，或任务已完成时调用。", - "definition": json.dumps( - { - "schema_version": 1, - "type": "end_call", - "config": { - "message_type": "none", - "custom_message": "", - "capture_reason": True, - }, - } - ), - "secrets": "{}", - "status": "active", + "config": { + "message_type": "none", + "custom_message": "", + "capture_reason": True, + }, }, - ) + }, + { + "id": "tool_show_message_default", + "name": "显示确认消息", + "function_name": "show_message", + "type": "client", + "description": "向用户显示必须主动确认的重要消息弹窗。", + "definition": { + "schema_version": 1, + "type": "client", + "config": { + "allow_interruptions": False, + "execution_mode": "immediate", + "wait_for_response": True, + "response_wait_mode": "session", + "timeout_seconds": 3, + "parameters": [ + { + "name": "title", + "type": "string", + "location": "body", + "description": "弹窗标题", + "required": False, + }, + { + "name": "message", + "type": "string", + "location": "body", + "description": "需要用户确认的重要信息", + "required": True, + }, + { + "name": "actions", + "type": "array", + "location": "body", + "description": "可选操作按钮", + "required": False, + }, + { + "name": "dismissible", + "type": "boolean", + "location": "body", + "description": "是否允许不选择按钮直接关闭", + "required": False, + }, + ], + "dynamic_variable_assignments": {}, + }, + }, + }, + ] + async with engine.begin() as conn: + for tool in default_tools: + await conn.execute( + text( + "INSERT INTO tools " + "(id, name, function_name, type, description, definition, secrets, status) " + "VALUES (" + ":id, :name, :function_name, :type, :description, " + "CAST(:definition AS jsonb), CAST(:secrets AS jsonb), :status" + ") " + "ON CONFLICT (function_name) DO NOTHING" + ), + { + **tool, + "definition": json.dumps(tool["definition"]), + "secrets": "{}", + "status": "active", + }, + ) diff --git a/backend/migrations/versions/20260801_0009_add_assistant_startup.py b/backend/migrations/versions/20260801_0009_add_assistant_startup.py new file mode 100644 index 0000000..54326c2 --- /dev/null +++ b/backend/migrations/versions/20260801_0009_add_assistant_startup.py @@ -0,0 +1,32 @@ +"""add assistant startup actions + +Revision ID: 20260801_0009 +Revises: 20260717_0008 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260801_0009" +down_revision: str | Sequence[str] | None = "20260717_0008" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "assistants", + sa.Column( + "startup", + sa.JSON(), + server_default=sa.text("'{}'"), + nullable=False, + ), + ) + + +def downgrade() -> None: + op.drop_column("assistants", "startup") diff --git a/backend/models.py b/backend/models.py index 721905e..e6f7b8d 100644 --- a/backend/models.py +++ b/backend/models.py @@ -106,6 +106,7 @@ class AssistantConfig(BaseModel): enableInterrupt: bool = True turnConfig: dict = Field(default_factory=dict) + startup: dict = Field(default_factory=dict) # Prompt assistant reusable tools. Execution remains type-specific in the pipeline. tools: list[RuntimeTool] = Field(default_factory=list) diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index dc47357..b1428a2 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -2,6 +2,7 @@ import uuid +from models import RuntimeTool from db.models import ( Assistant, AssistantModelBinding, @@ -16,6 +17,7 @@ from schemas import AssistantOut, AssistantUpsert from services.auth import require_admin from services.masking import mask, resolve_incoming_key from services.node_specs import graph_references, normalize_graph, validate_graph +from services.tool_policy import policy_for_tool from services.workflow_engine import WorkflowEngine from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -110,6 +112,57 @@ async def _validate_workflow_references( raise HTTPException(400, f"Workflow 引用了无效知识库:{knowledge_id}") +async def _validate_startup_actions( + session: AsyncSession, + body: AssistantUpsert, +) -> None: + """Keep startup deterministic and reject unsupported lifecycle/tool pairs.""" + + actions = body.startup.actions + if not actions: + return + bound_tool_ids = set(body.tool_ids) + for action in actions: + if action.tool_id not in bound_tool_ids: + raise HTTPException(400, f"启动 Action 必须先绑定工具:{action.tool_id}") + tool = await session.get(Tool, action.tool_id) + if not tool or tool.status != "active": + raise HTTPException(400, f"启动 Action 引用了无效工具:{action.tool_id}") + if action.phase == "preflight" and tool.type not in {"http", "mcp"}: + raise HTTPException(400, "preflight Action 仅支持 HTTP 或 MCP 工具") + if action.phase == "opening" and tool.type not in {"http", "mcp", "client"}: + raise HTTPException(400, "opening Action 不支持该工具类型") + if tool.type != "client": + continue + runtime_tool = RuntimeTool( + id=tool.id, + name=tool.name, + function_name=tool.function_name, + type=tool.type, + definition=tool.definition or {}, + ) + policy = policy_for_tool(runtime_tool) + if action.required and not policy.wait_for_response: + raise HTTPException(400, "必需的 Client 启动 Action 必须等待客户端响应") + if tool.function_name != "show_message": + continue + if policy.response_wait_mode != "session": + raise HTTPException(400, "show_message 启动 Action 必须使用会话内等待") + message = str(action.arguments.get("message") or "").strip() + buttons = action.arguments.get("actions") + if not message or len(message) > 2000: + raise HTTPException(400, "show_message 重要信息必须为 1-2000 个字符") + if not isinstance(buttons, list) or not buttons: + raise HTTPException(400, "show_message 至少需要一个确认按钮") + first_button = buttons[0] if isinstance(buttons[0], dict) else {} + if not str(first_button.get("id") or "").strip() or not str( + first_button.get("label") or "" + ).strip(): + raise HTTPException(400, "show_message 确认按钮必须配置 id 和文字") + if action.required and action.arguments.get("dismissible") is not False: + raise HTTPException(400, "必需的 show_message 启动 Action 不允许跳过确认") + + async def _validate_vision_model( session: AsyncSession, body: AssistantUpsert ) -> None: @@ -228,6 +281,7 @@ async def _to_out(session: AsyncSession, assistant: Assistant) -> AssistantOut: greeting=assistant.greeting, enable_interrupt=assistant.enable_interrupt, turn_config=assistant.turn_config or {}, + startup=assistant.startup or {}, vision_enabled=assistant.vision_enabled, vision_model_resource_id=assistant.vision_model_resource_id, model_resource_ids=await _resource_ids(session, assistant.id), @@ -262,6 +316,7 @@ async def create_assistant( ): _validate_workflow(body) await _validate_workflow_references(session, body) + await _validate_startup_actions(session, body) await _validate_vision_model(session, body) await _validate_knowledge_base(session, body) data = body.model_dump() @@ -302,6 +357,7 @@ async def duplicate_assistant( greeting=source.greeting, enable_interrupt=source.enable_interrupt, turn_config=dict(source.turn_config or {}), + startup=dict(source.startup or {}), vision_enabled=source.vision_enabled, vision_model_resource_id=source.vision_model_resource_id, knowledge_base_id=source.knowledge_base_id, @@ -335,6 +391,7 @@ async def update_assistant( raise HTTPException(404, "助手不存在") _validate_workflow(body) await _validate_workflow_references(session, body) + await _validate_startup_actions(session, body) await _validate_vision_model(session, body) await _validate_knowledge_base(session, body) data = body.model_dump() diff --git a/backend/schemas.py b/backend/schemas.py index b94c0a1..60b445a 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -92,6 +92,26 @@ ALLOWED_FIELDS: dict[str, set[str]] = { } +class StartupAction(CamelModel): + id: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_-]*$") + phase: Literal["preflight", "opening"] = "opening" + tool_id: str = Field(min_length=1, max_length=40) + arguments: dict[str, Any] = Field(default_factory=dict) + required: bool = True + + +class StartupConfig(CamelModel): + execution_mode: Literal["sequential"] = "sequential" + actions: list[StartupAction] = Field(default_factory=list, max_length=5) + + @model_validator(mode="after") + def validate_unique_action_ids(self): + ids = [action.id for action in self.actions] + if len(ids) != len(set(ids)): + raise ValueError("启动 Action 的 id 不能重复") + return self + + # ---------- 助手(单表 STI:瘦类型真列 + workflow 图 JSON 列) ---------- class AssistantUpsert(CamelModel): name: str @@ -100,6 +120,7 @@ class AssistantUpsert(CamelModel): greeting: str = "" enable_interrupt: bool = True turn_config: TurnConfig = Field(default_factory=TurnConfig) + startup: StartupConfig = Field(default_factory=StartupConfig) vision_enabled: bool = False vision_model_resource_id: str | None = None @@ -145,6 +166,10 @@ class AssistantUpsert(CamelModel): if self.type not in {"prompt", "workflow"}: self.tool_ids = [] self.dynamic_variable_definitions = {} + if self.type != "prompt": + self.startup = StartupConfig() + if self.runtime_mode == "realtime" and self.startup.actions: + raise ValueError("Prompt Realtime 模式暂不支持启动 Action") # 外部托管大脑只能 cascade,拦住不兼容的 realtime if self.runtime_mode == "realtime" and self.type not in REALTIME_CAPABLE_TYPES: raise ValueError(f"类型 {self.type} 不支持 realtime 运行模式") diff --git a/backend/services/action_runtime.py b/backend/services/action_runtime.py new file mode 100644 index 0000000..872a1bc --- /dev/null +++ b/backend/services/action_runtime.py @@ -0,0 +1,222 @@ +"""Deterministic Action execution shared by Prompt startup and Workflow.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from dataclasses import dataclass +from enum import StrEnum +from time import monotonic +from typing import Any +from uuid import uuid4 + +from models import RuntimeTool +from services.tool_executor import ToolExecutionError, ToolExecutor + + +class ActionStatus(StrEnum): + SUCCESS = "success" + FAILURE = "failure" + CANCELLED = "cancelled" + + +@dataclass(frozen=True) +class ActionError: + code: str + message: str + retryable: bool = False + + +@dataclass(frozen=True) +class ActionOutcome: + """One completed Action invocation. + + Raw tool results stay in memory. Persisted or client-visible events should + use ``trace_payload`` so private business data is not copied accidentally. + """ + + invocation_id: str + status: ActionStatus + duration_ms: int + result: dict[str, Any] | None = None + updated_variables: tuple[str, ...] = () + error: ActionError | None = None + + @property + def should_route(self) -> bool: + return self.status != ActionStatus.CANCELLED + + def trace_payload(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "invocationId": self.invocation_id, + "status": self.status.value, + "durationMs": self.duration_ms, + "updatedVariables": list(self.updated_variables), + } + if self.result is not None: + payload["resultKeys"] = sorted(str(key) for key in self.result) + if self.error is not None: + payload["error"] = { + "code": self.error.code, + "message": self.error.message, + "retryable": self.error.retryable, + } + return payload + + +class ActionInvocationCancelled(asyncio.CancelledError): + """Carry a structured outcome while preserving task cancellation.""" + + def __init__(self, outcome: ActionOutcome) -> None: + super().__init__(outcome.error.message if outcome.error else "Action cancelled") + self.outcome = outcome + + +class ActionRunner: + """Normalize ToolExecutor's heterogeneous responses into ActionOutcome.""" + + _CANCELLED_MARKERS = ( + "会话已结束", + "会话已取消", + "管线已停止", + "通道已关闭", + "连接已断开", + ) + + def __init__( + self, + executor: ToolExecutor, + *, + is_session_ending: Callable[[], bool] | None = None, + ) -> None: + self._executor = executor + self._is_session_ending = is_session_ending or (lambda: False) + + @staticmethod + def new_invocation_id() -> str: + return f"act_{uuid4().hex[:20]}" + + async def execute( + self, + tool: RuntimeTool | None, + arguments: dict[str, Any] | None = None, + *, + result_assignments: dict[str, str] | None = None, + invocation_id: str | None = None, + ) -> ActionOutcome: + invocation_id = invocation_id or self.new_invocation_id() + started_at = monotonic() + result: dict[str, Any] | None = None + if tool is None: + return self._failure( + invocation_id, + started_at, + code="tool_not_found", + message="Action 引用的工具不存在", + ) + + try: + rendered_arguments = self._executor.store.render_data(arguments or {}) + result = await self._executor.execute( + tool, + rendered_arguments, + result_assignments=result_assignments, + ) + if result.get("status") != "ok": + returned_status = str(result.get("status") or "error") + message = str(result.get("message") or "工具返回执行失败状态") + if self._is_cancelled(message): + return ActionOutcome( + invocation_id=invocation_id, + status=ActionStatus.CANCELLED, + duration_ms=self._elapsed_ms(started_at), + result=result, + error=ActionError( + code="session_ended", + message=message[:2048], + ), + ) + return self._failure( + invocation_id, + started_at, + code=str(result.get("code") or f"tool_{returned_status}"), + message=message, + retryable=bool(result.get("retryable", False)), + result=result, + ) + return ActionOutcome( + invocation_id=invocation_id, + status=ActionStatus.SUCCESS, + duration_ms=self._elapsed_ms(started_at), + result=result, + updated_variables=tuple( + str(name) for name in result.get("updated_variables") or [] + ), + ) + except asyncio.CancelledError as exc: + outcome = ActionOutcome( + invocation_id=invocation_id, + status=ActionStatus.CANCELLED, + duration_ms=self._elapsed_ms(started_at), + result=result, + error=ActionError( + code="action_cancelled", + message="Action 随当前任务取消", + ), + ) + raise ActionInvocationCancelled(outcome) from exc + except (ToolExecutionError, ValueError) as exc: + cancelled = self._is_cancelled(str(exc)) + if cancelled: + return ActionOutcome( + invocation_id=invocation_id, + status=ActionStatus.CANCELLED, + duration_ms=self._elapsed_ms(started_at), + result=result, + error=ActionError( + code="session_ended", + message=str(exc)[:2048], + ), + ) + return self._failure( + invocation_id, + started_at, + code=( + "invalid_action_configuration" + if isinstance(exc, ValueError) + else "tool_execution_error" + ), + message=str(exc), + result=result, + ) + + def _is_cancelled(self, message: str) -> bool: + return self._is_session_ending() or any( + marker in message for marker in self._CANCELLED_MARKERS + ) + + def _failure( + self, + invocation_id: str, + started_at: float, + *, + code: str, + message: str, + retryable: bool = False, + result: dict[str, Any] | None = None, + ) -> ActionOutcome: + return ActionOutcome( + invocation_id=invocation_id, + status=ActionStatus.FAILURE, + duration_ms=self._elapsed_ms(started_at), + result=result, + error=ActionError( + code=code, + message=message[:2048], + retryable=retryable, + ), + ) + + @staticmethod + def _elapsed_ms(started_at: float) -> int: + return max(0, round((monotonic() - started_at) * 1000)) diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 702195c..13d0a56 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -118,6 +118,9 @@ class BaseBrain: async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: """Register tools and initialize per-call orchestration.""" + async def run_preflight(self) -> None: + """Run deterministic server-side startup work before media starts.""" + async def on_connected(self, *, greeting_pending: bool = False) -> None: """Handle a connected client before an optional greeting is played. @@ -198,6 +201,8 @@ class Brain(Protocol): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: ... + async def run_preflight(self) -> None: ... + async def on_connected(self, *, greeting_pending: bool = False) -> None: ... async def on_greeting_finished(self) -> None: ... diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index 4fccace..29fe938 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -2,9 +2,11 @@ from __future__ import annotations +import asyncio from typing import Any from uuid import uuid4 +from loguru import logger from models import AssistantConfig from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.frames.frames import OutputTransportMessageUrgentFrame, TTSSpeakFrame @@ -22,11 +24,19 @@ from services.brains.base import ( BrainSpec, SessionVariableUpdate, ) +from services.action_runtime import ( + ActionInvocationCancelled, + ActionRunner, + ActionStatus, +) from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool +PREFLIGHT_TIMEOUT_SECONDS = 30 + + class PromptBrain(BaseBrain): spec = BrainSpec( type="prompt", @@ -39,8 +49,15 @@ class PromptBrain(BaseBrain): self._dynamic_enabled = True self._store = DynamicVariableStore.from_config(cfg) self._tools = ToolExecutor(self._store) + self._actions = ActionRunner(self._tools) + self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._runtime: BrainRuntime | None = None self._waiting_for_generated_end_speech = False + self._greeting_finished = True + self._preflight_finished = False + self._opening_started = False + self._opening_finished = False + self._startup_failed = False async def greeting(self, cfg: AssistantConfig) -> str: return self._store.render(cfg.greeting) if self._dynamic_enabled else cfg.greeting @@ -56,7 +73,17 @@ class PromptBrain(BaseBrain): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: self._runtime = runtime self._tools.set_client_tools(runtime.client_tools) + self._actions = ActionRunner( + self._tools, + is_session_ending=lambda: runtime.call_end.ending, + ) + self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._waiting_for_generated_end_speech = False + self._greeting_finished = True + self._preflight_finished = False + self._opening_started = False + self._opening_finished = not bool(self._startup_actions("opening")) + self._startup_failed = False schemas: list[FunctionSchema] = [] for tool in cfg.tools: if tool.type == "end_call": @@ -74,6 +101,122 @@ class PromptBrain(BaseBrain): ) runtime.set_tools(schemas) + async def run_preflight(self) -> None: + if self._preflight_finished: + return + try: + async with asyncio.timeout(PREFLIGHT_TIMEOUT_SECONDS): + succeeded = await self._run_startup_actions("preflight") + except TimeoutError as exc: + raise RuntimeError("Prompt preflight 超过 30 秒安全上限") from exc + if not succeeded: + raise RuntimeError("必需的 Prompt preflight Action 执行失败") + self._preflight_finished = True + + async def on_connected(self, *, greeting_pending: bool = False) -> None: + self._greeting_finished = not greeting_pending + if ( + self._startup_actions("opening") + and self._runtime is not None + and self._runtime.set_input_enabled is not None + ): + self._runtime.set_input_enabled(False) + + async def on_client_ready(self) -> None: + if self._opening_started or self._opening_finished or self._startup_failed: + return + self._opening_started = True + try: + succeeded = await self._run_startup_actions("opening") + except ActionInvocationCancelled: + self._startup_failed = True + raise + if not succeeded: + await self._fail_opening("必需的开场 Action 执行失败") + return + self._opening_finished = True + self._release_startup_gate_if_ready() + + async def on_greeting_finished(self) -> None: + self._greeting_finished = True + self._release_startup_gate_if_ready() + + def _startup_actions(self, phase: str) -> list[dict[str, Any]]: + startup = self._cfg.startup if isinstance(self._cfg.startup, dict) else {} + return [ + action + for action in startup.get("actions") or [] + if isinstance(action, dict) and action.get("phase", "opening") == phase + ] + + async def _run_startup_actions(self, phase: str) -> bool: + for action in self._startup_actions(phase): + action_id = str(action.get("id") or "startup_action") + tool_id = str(action.get("tool_id") or action.get("toolId") or "") + tool = self._tool_by_id.get(tool_id) + invocation_id = self._actions.new_invocation_id() + logger.info( + f"执行 Prompt {phase} Action: action={action_id} tool={tool_id}" + ) + outcome = await self._actions.execute( + tool, + action.get("arguments") or {}, + invocation_id=invocation_id, + ) + if outcome.updated_variables: + self._refresh_prompt() + if phase == "opening" and self._runtime is not None: + await self._runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "startup-action-result", + "actionId": action_id, + "phase": phase, + "outcome": outcome.trace_payload(), + } + ) + ) + if outcome.status == ActionStatus.SUCCESS: + continue + if outcome.status == ActionStatus.CANCELLED: + return False + if bool(action.get("required", True)): + logger.warning( + f"必需的 Prompt {phase} Action 失败: " + f"action={action_id} error={outcome.error}" + ) + return False + logger.warning( + f"忽略可选 Prompt {phase} Action 失败: " + f"action={action_id} error={outcome.error}" + ) + return True + + def _release_startup_gate_if_ready(self) -> None: + runtime = self._runtime + if ( + runtime is not None + and runtime.set_input_enabled is not None + and self._greeting_finished + and self._opening_finished + and not self._startup_failed + and not runtime.call_end.ending + ): + runtime.set_input_enabled(True) + + async def _fail_opening(self, message: str) -> None: + self._startup_failed = True + runtime = self._runtime + if runtime is None or runtime.call_end.ending: + return + await runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={"type": "startup-action-error", "message": message} + ) + ) + runtime.call_end.begin("startup_action_failed") + await runtime.call_end.finish() + def record_user_message(self, content: str) -> None: if not self._dynamic_enabled: return diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 142312e..d59ee50 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -5,9 +5,7 @@ from __future__ import annotations import asyncio from copy import deepcopy from dataclasses import replace -from time import monotonic from typing import Any -from uuid import uuid4 from loguru import logger from models import AssistantConfig, RuntimeTool @@ -36,19 +34,18 @@ from services.brains.base import ( BrainSpec, SessionVariableUpdate, ) +from services.action_runtime import ( + ActionInvocationCancelled, + ActionOutcome, + ActionRunner, + ActionStatus, +) from services.knowledge import search as search_knowledge from services.runtime_variables import DynamicVariableStore from services.tool_executor import ToolExecutionError, ToolExecutor from services.tool_policy import policy_for_tool from services.workflow.agent import WorkflowAgentStage -from services.workflow.models import ( - ActionError, - ActionOutcome, - ActionStatus, - RouteStatus, - WorkflowRuntimeState, - WorkflowStatus, -) +from services.workflow.models import RouteStatus, WorkflowRuntimeState, WorkflowStatus from services.workflow.output import WorkflowOutput from services.workflow.routing import WorkflowEdgeEvaluator from services.workflow_engine import WorkflowEngine @@ -58,23 +55,6 @@ from services.workflow_router import WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 -class _ActionFailure(RuntimeError): - """Internal adapter from heterogeneous tool failures to ActionOutcome.""" - - def __init__( - self, - message: str, - *, - code: str, - retryable: bool = False, - result: dict[str, Any] | None = None, - ) -> None: - super().__init__(message) - self.code = code - self.retryable = retryable - self.result = result - - class ConfiguredFlowManager(FlowManager): """Preserve Flow transitions while suppressing late async-tool replies.""" @@ -129,6 +109,7 @@ class WorkflowBrain(BaseBrain): self._cfg = cfg self._store = DynamicVariableStore.from_config(cfg or AssistantConfig(type="workflow")) self._tools = ToolExecutor(self._store) + self._actions = ActionRunner(self._tools) self._tool_by_id: dict[str, RuntimeTool] = { tool.id: tool for tool in (cfg.tools if cfg else []) } @@ -166,6 +147,10 @@ class WorkflowBrain(BaseBrain): self._runtime = runtime self._store = DynamicVariableStore.from_config(cfg) self._tools = ToolExecutor(self._store, client_tools=runtime.client_tools) + self._actions = ActionRunner( + self._tools, + is_session_ending=lambda: runtime.call_end.ending, + ) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) self._edge_evaluator = WorkflowEdgeEvaluator( @@ -749,8 +734,7 @@ class WorkflowBrain(BaseBrain): await self._emit_node_active(node_id) data = self._engine.data(node_id) runtime = self._require_runtime() - invocation_id = f"act_{uuid4().hex[:20]}" - started_at = monotonic() + invocation_id = self._actions.new_invocation_id() block_user_input = data.get("userInputPolicy") == "block" if block_user_input and runtime.set_input_enabled: # Blocking only suppresses new audio/text input while the Action @@ -760,7 +744,6 @@ class WorkflowBrain(BaseBrain): runtime.set_input_enabled(False) tool_id = str(data.get("toolId") or "") tool = self._tool_by_id.get(tool_id) - result: dict[str, Any] | None = None try: await self._emit_trace( "action_started", @@ -769,82 +752,24 @@ class WorkflowBrain(BaseBrain): toolId=tool_id, toolType=tool.type if tool else None, ) - if not tool: - raise _ActionFailure( - f"工具不存在:{tool_id}", - code="tool_not_found", - ) - arguments = self._store.render_data(data.get("arguments") or {}) - result = await self._tools.execute( + outcome = await self._actions.execute( tool, - arguments, + data.get("arguments") or {}, result_assignments=self._action_result_assignments(data), + invocation_id=invocation_id, ) - if result.get("status") != "ok": - returned_status = str(result.get("status") or "error") - raise _ActionFailure( - str(result.get("message") or "工具返回执行失败状态"), - code=str(result.get("code") or f"tool_{returned_status}"), - retryable=bool(result.get("retryable", False)), - result=result, - ) - updated_variables = list(result.get("updated_variables") or []) + updated_variables = list(outcome.updated_variables) if updated_variables: await self._emit_variables( reason="action", node_id=node_id, changed=updated_variables, ) - outcome = ActionOutcome( - invocation_id=invocation_id, - status=ActionStatus.SUCCESS, - duration_ms=self._elapsed_ms(started_at), - result=result, - updated_variables=tuple(str(name) for name in updated_variables), - ) - except asyncio.CancelledError: - outcome = ActionOutcome( - invocation_id=invocation_id, - status=ActionStatus.CANCELLED, - duration_ms=self._elapsed_ms(started_at), - result=result, - error=ActionError( - code="action_cancelled", - message="Action 随当前任务取消", - ), - ) + except ActionInvocationCancelled as exc: + outcome = exc.outcome self._set_last_action(outcome) await self._emit_action_outcome(node_id, outcome) raise - except (_ActionFailure, ToolExecutionError, ValueError) as exc: - cancelled = self._action_was_cancelled(exc, runtime) - outcome = ActionOutcome( - invocation_id=invocation_id, - status=( - ActionStatus.CANCELLED if cancelled else ActionStatus.FAILURE - ), - duration_ms=self._elapsed_ms(started_at), - result=(exc.result if isinstance(exc, _ActionFailure) else result), - error=ActionError( - code=( - "session_ended" - if cancelled - else ( - exc.code - if isinstance(exc, _ActionFailure) - else ( - "invalid_action_configuration" - if isinstance(exc, ValueError) - else "tool_execution_error" - ) - ) - ), - message=str(exc)[:2048], - retryable=( - exc.retryable if isinstance(exc, _ActionFailure) else False - ), - ), - ) finally: if block_user_input and runtime.set_input_enabled: runtime.set_input_enabled(True) @@ -852,26 +777,6 @@ class WorkflowBrain(BaseBrain): await self._emit_action_outcome(node_id, outcome) return outcome - @staticmethod - def _elapsed_ms(started_at: float) -> int: - return max(0, round((monotonic() - started_at) * 1000)) - - @staticmethod - def _action_was_cancelled(exc: Exception, runtime: BrainRuntime) -> bool: - if getattr(runtime.call_end, "ending", False): - return True - message = str(exc) - return any( - marker in message - for marker in ( - "会话已结束", - "会话已取消", - "管线已停止", - "通道已关闭", - "连接已断开", - ) - ) - def _set_last_action(self, outcome: ActionOutcome) -> None: legacy_status = { ActionStatus.SUCCESS: "ok", diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index 08f6909..505e04a 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -227,6 +227,7 @@ async def resolve_runtime_config( runtimeMode=assistant.runtime_mode, # type: ignore[arg-type] enableInterrupt=assistant.enable_interrupt, turnConfig=assistant.turn_config or {}, + startup=assistant.startup or {}, tools=await _tools_for(session, assistant), knowledge_base_id=assistant.knowledge_base_id, knowledge_base_name=knowledge_base.name if knowledge_base else "", diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index a91e553..76d6058 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -270,7 +270,6 @@ async def run_pipeline( build_workflow_voice_switcher(cfg, "TTS", tts) ) - greeting = await brain.greeting(cfg) system_content = brain.system_prompt(cfg) worker_holder: dict = {} @@ -683,6 +682,14 @@ async def run_pipeline( flow_global_functions=flow_global_functions, ), ) + try: + await brain.run_preflight() + except Exception: + if recorder: + await recorder.finish(status="failed") + raise + # Preflight tools may assign variables used by the opening speech. + greeting = await brain.greeting(cfg) async def submit_user_input(value: UserInput) -> None: if not value.has_camera_frame: diff --git a/backend/services/workflow/models.py b/backend/services/workflow/models.py index 7661fa8..e8a0c17 100644 --- a/backend/services/workflow/models.py +++ b/backend/services/workflow/models.py @@ -27,65 +27,6 @@ class RouteStatus(StrEnum): ERROR = "error" -class ActionStatus(StrEnum): - """Stable Action outcomes used by routing and future debug tooling.""" - - SUCCESS = "success" - FAILURE = "failure" - CANCELLED = "cancelled" - - -@dataclass(frozen=True) -class ActionError: - """Machine-readable failure details without losing the operator message.""" - - code: str - message: str - retryable: bool = False - - -@dataclass(frozen=True) -class ActionOutcome: - """One completed Action invocation. - - ``result`` remains an in-memory value because a tool response may contain - private business data. Trace events publish only its shape and variable - names, never the raw response. - """ - - invocation_id: str - status: ActionStatus - duration_ms: int - result: dict[str, Any] | None = None - updated_variables: tuple[str, ...] = () - error: ActionError | None = None - - @property - def should_route(self) -> bool: - """Cancellation is a lifecycle outcome, not a failure branch.""" - - return self.status != ActionStatus.CANCELLED - - def trace_payload(self) -> dict[str, Any]: - """Return a persistence-safe summary of the execution result.""" - - payload: dict[str, Any] = { - "invocationId": self.invocation_id, - "status": self.status.value, - "durationMs": self.duration_ms, - "updatedVariables": list(self.updated_variables), - } - if self.result is not None: - payload["resultKeys"] = sorted(str(key) for key in self.result) - if self.error is not None: - payload["error"] = { - "code": self.error.code, - "message": self.error.message, - "retryable": self.error.retryable, - } - return payload - - @dataclass(frozen=True) class UserTurn: """One committed user turn that may cross automatic Workflow nodes.""" diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 432eef7..8b3daea 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -2,7 +2,7 @@ from __future__ import annotations import unittest from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import AsyncMock, patch from models import AssistantConfig, RuntimeTool from pipecat.frames.frames import ( @@ -29,8 +29,8 @@ from services.brains.dify_llm import ( ) from services.brains.workflow_brain import WorkflowBrain from services.runtime_variables import prepare_dynamic_config +from services.action_runtime import ActionError, ActionOutcome, ActionStatus from services.workflow.models import ( - ActionStatus, LLMRouteResult, RouteStatus, WorkflowStatus, @@ -246,6 +246,206 @@ class DifyLLMServiceTests(unittest.IsolatedAsyncioTestCase): class PromptBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_preflight_runs_multiple_server_tools_in_order(self): + tools = [ + RuntimeTool( + id=f"preflight_{index}", + name=f"预检动作 {index}", + function_name=f"preflight_{index}", + type="http", + ) + for index in (1, 2) + ] + cfg = AssistantConfig( + type="prompt", + tools=tools, + startup={ + "actions": [ + { + "id": tool.id, + "phase": "preflight", + "tool_id": tool.id, + "required": True, + } + for tool in tools + ] + }, + ) + brain = build_brain(cfg) + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=noop_queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ), + ) + brain._actions.execute = AsyncMock( + side_effect=[ + ActionOutcome( + invocation_id=f"act_{index}", + status=ActionStatus.SUCCESS, + duration_ms=index, + ) + for index in (1, 2) + ] + ) + + await brain.run_preflight() + + self.assertEqual( + [call.args[0].id for call in brain._actions.execute.await_args_list], + ["preflight_1", "preflight_2"], + ) + + async def test_opening_actions_wait_for_confirmation_and_greeting(self): + tools = [ + RuntimeTool( + id=f"opening_{index}", + name=f"开场动作 {index}", + function_name="show_message" if index == 1 else "load_opening_data", + type="client" if index == 1 else "http", + ) + for index in (1, 2) + ] + cfg = AssistantConfig( + type="prompt", + tools=tools, + startup={ + "execution_mode": "sequential", + "actions": [ + { + "id": f"opening_{index}", + "phase": "opening", + "tool_id": f"opening_{index}", + "arguments": {}, + "required": True, + } + for index in (1, 2) + ], + }, + ) + brain = build_brain(cfg) + input_states = [] + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + set_input_enabled=input_states.append, + ), + ) + brain._actions.execute = AsyncMock( + side_effect=[ + ActionOutcome( + invocation_id=f"act_{index}", + status=ActionStatus.SUCCESS, + duration_ms=index, + ) + for index in (1, 2) + ] + ) + + await brain.on_connected(greeting_pending=True) + await brain.on_client_ready() + + self.assertEqual(input_states, [False]) + called_tool_ids = [ + call.args[0].id for call in brain._actions.execute.await_args_list + ] + self.assertEqual(called_tool_ids, ["opening_1", "opening_2"]) + self.assertEqual( + len( + [ + frame + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "startup-action-result" + ] + ), + 2, + ) + + await brain.on_greeting_finished() + self.assertEqual(input_states, [False, True]) + + # Replayed client-ready must not execute startup actions twice. + await brain.on_client_ready() + self.assertEqual(brain._actions.execute.await_count, 2) + + async def test_required_opening_failure_keeps_input_blocked_and_ends_call(self): + tool = RuntimeTool( + id="opening_message", + name="开场确认", + function_name="show_message", + type="client", + ) + cfg = AssistantConfig( + type="prompt", + tools=[tool], + startup={ + "actions": [ + { + "id": "opening_message", + "phase": "opening", + "tool_id": tool.id, + "arguments": {}, + "required": True, + } + ] + }, + ) + brain = build_brain(cfg) + call_end = FakeCallEnd() + input_states = [] + + async def queue_frame(_frame): + pass + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=call_end, + set_input_enabled=input_states.append, + ), + ) + brain._actions.execute = AsyncMock( + return_value=ActionOutcome( + invocation_id="act_failed", + status=ActionStatus.FAILURE, + duration_ms=1, + error=ActionError( + code="tool_error", + message="用户未确认", + ), + ) + ) + + await brain.on_connected(greeting_pending=False) + await brain.on_client_ready() + + self.assertEqual(input_states, [False]) + self.assertTrue(call_end.ending) + self.assertTrue(call_end.finished) + self.assertEqual(call_end.reason, "startup_action_failed") + async def test_realtime_prompt_brain_renders_dynamic_variables(self): cfg = prepare_dynamic_config( AssistantConfig( diff --git a/backend/tests/test_startup_actions.py b/backend/tests/test_startup_actions.py new file mode 100644 index 0000000..0e7fe90 --- /dev/null +++ b/backend/tests/test_startup_actions.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from fastapi import HTTPException +from routes.assistants import _validate_startup_actions +from schemas import AssistantUpsert + + +def startup_body(*, phase: str = "opening") -> AssistantUpsert: + return AssistantUpsert( + name="启动动作测试", + type="prompt", + runtimeMode="pipeline", + toolIds=["tool_message"], + startup={ + "executionMode": "sequential", + "actions": [ + { + "id": "opening_message", + "phase": phase, + "toolId": "tool_message", + "arguments": { + "title": "重要提示", + "message": "请确认已阅读。", + "actions": [ + { + "id": "confirmed", + "label": "确认", + "style": "primary", + } + ], + "dismissible": False, + }, + "required": True, + } + ], + }, + ) + + +class FakeSession: + def __init__(self, tool): + self.tool = tool + + async def get(self, _model, tool_id): + return self.tool if tool_id == self.tool.id else None + + +class StartupActionValidationTests(unittest.IsolatedAsyncioTestCase): + def test_realtime_rejects_startup_actions(self): + with self.assertRaisesRegex(ValueError, "Realtime"): + AssistantUpsert( + name="Realtime 启动动作", + type="prompt", + runtimeMode="realtime", + toolIds=["tool_message"], + startup=startup_body().startup, + ) + + async def test_show_message_requires_session_wait(self): + tool = SimpleNamespace( + id="tool_message", + name="重要提示", + function_name="show_message", + type="client", + status="active", + definition={ + "config": { + "wait_for_response": True, + "response_wait_mode": "timeout", + } + }, + ) + + with self.assertRaisesRegex(HTTPException, "会话内等待"): + await _validate_startup_actions(FakeSession(tool), startup_body()) + + tool.definition["config"]["response_wait_mode"] = "session" + await _validate_startup_actions(FakeSession(tool), startup_body()) + + async def test_preflight_rejects_client_tools(self): + tool = SimpleNamespace( + id="tool_message", + name="重要提示", + function_name="show_message", + type="client", + status="active", + definition={ + "config": { + "wait_for_response": True, + "response_wait_mode": "session", + } + }, + ) + + with self.assertRaisesRegex(HTTPException, "preflight"): + await _validate_startup_actions( + FakeSession(tool), + startup_body(phase="preflight"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/assistant-editor/prompt-editor.tsx b/frontend/src/components/assistant-editor/prompt-editor.tsx index 5c1f8de..996cefd 100644 --- a/frontend/src/components/assistant-editor/prompt-editor.tsx +++ b/frontend/src/components/assistant-editor/prompt-editor.tsx @@ -29,8 +29,10 @@ import { } from "@/components/assistant-editor/editor-controls"; import type { AssistantForm } from "@/components/assistant-editor/types"; import { SectionCard } from "@/components/editor/section-card"; +import { VisionConfigSection } from "@/components/editor/vision-config-section"; import { TurnConfigEditor } from "@/components/turn-config-editor"; import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; import type { DynamicVariableDefinition, Tool } from "@/lib/api"; type ResourceOption = { value: string; label: string }; @@ -83,6 +85,81 @@ export function PromptEditor({ handlePromptVisionEnabledChange, handlePromptModelChange, }: PromptEditorProps) { + const openingMessage = form.startup.actions.find( + (action) => action.id === "opening_message" && action.phase === "opening", + ); + const openingArguments = openingMessage?.arguments ?? {}; + const openingButtons = Array.isArray(openingArguments.actions) + ? openingArguments.actions + : []; + const openingButton = + openingButtons[0] && typeof openingButtons[0] === "object" + ? (openingButtons[0] as Record) + : {}; + const showMessageTools = tools + .filter( + (tool) => + tool.status === "active" && + tool.type === "client" && + tool.functionName === "show_message" && + tool.definition.type === "client" && + tool.definition.config.waitForResponse && + tool.definition.config.responseWaitMode === "session", + ) + .map((tool) => ({ value: tool.id, label: tool.name })); + + function setOpeningMessage(enabled: boolean) { + const otherActions = form.startup.actions.filter( + (action) => action.id !== "opening_message", + ); + const defaultToolId = showMessageTools[0]?.value ?? ""; + updateForm("startup", { + executionMode: "sequential", + actions: enabled + ? [ + ...otherActions, + { + id: "opening_message", + phase: "opening", + toolId: defaultToolId, + required: true, + arguments: { + title: "重要提示", + message: "请确认已阅读以上信息。", + actions: [ + { id: "confirmed", label: "确认", style: "primary" }, + ], + dismissible: false, + }, + }, + ] + : otherActions, + }); + if ( + enabled && + defaultToolId && + !form.toolIds.includes(defaultToolId) + ) { + updateForm("toolIds", [...form.toolIds, defaultToolId]); + } + } + + function updateOpeningMessage( + patch: Partial>, + ) { + if (!openingMessage) return; + updateForm("startup", { + ...form.startup, + actions: form.startup.actions.map((action) => + action.id === openingMessage.id ? { ...action, ...patch } : action, + ), + }); + } + + function updateOpeningArguments(patch: Record) { + updateOpeningMessage({ arguments: { ...openingArguments, ...patch } }); + } + return (
@@ -161,11 +238,87 @@ export function PromptEditor({ count={Object.keys(dynamicVariableDefinitions).length} onOpen={() => setDynamicVariablesOpen(true)} /> + {form.runtimeMode === "pipeline" && ( +
+ + + {openingMessage && ( +
+ { + updateOpeningMessage({ toolId }); + if (toolId && !form.toolIds.includes(toolId)) { + updateForm("toolIds", [...form.toolIds, toolId]); + } + }} + /> + {showMessageTools.length === 0 && ( +

+ 请先在组件管理中创建 functionName 为 show_message、会话内等待的 Client Tool。 +

+ )} + + updateOpeningArguments({ message })} + placeholder="请输入需要用户确认的重要信息" + rows={4} + /> + +

+ 弹窗不可通过关闭按钮、Esc 或点击外部跳过；失败时将结束本次通话。 +

+
+ )} +
+ )} } - title="模型配置" + title="模型与语音" description={ form.runtimeMode === "pipeline" ? "选择运行方式，以及大语言模型、语音识别与语音合成资源" @@ -174,49 +327,40 @@ export function PromptEditor({ > updateForm("runtimeMode", runtimeMode)} + onChange={(runtimeMode) => { + updateForm("runtimeMode", runtimeMode); + if (runtimeMode === "realtime" && form.startup.actions.length) { + updateForm("startup", { + executionMode: "sequential", + actions: [], + }); + } + }} /> {form.runtimeMode === "pipeline" ? ( <> - - {form.visionEnabled && ( - updateForm("visionModelResourceId", value) - } - options={visionModelOptions} - noneLabel="模型自己" + label="大语言模型" + value={form.model} + onChange={handlePromptModelChange} + options={llmOptions} + noneLabel="无" + /> + updateForm("asr", value)} + options={asrOptions} + noneLabel="无" + /> + updateForm("voice", value)} + options={ttsOptions} + noneLabel="无" /> - )} - - updateForm("asr", value)} - options={asrOptions} - noneLabel="无" - /> - updateForm("voice", value)} - options={ttsOptions} - noneLabel="无" - /> ) : ( + {form.runtimeMode === "pipeline" && ( + + updateForm("visionModelResourceId", value) + } + /> + )} + {form.runtimeMode === "pipeline" && ( } @@ -299,4 +458,3 @@ export function PromptEditor({
); } - diff --git a/frontend/src/components/assistant-editor/types.ts b/frontend/src/components/assistant-editor/types.ts index e6e0bda..74db0e8 100644 --- a/frontend/src/components/assistant-editor/types.ts +++ b/frontend/src/components/assistant-editor/types.ts @@ -1,6 +1,7 @@ import type { DynamicVariableDefinition, KnowledgeRetrievalConfig, + StartupConfig, TurnConfig, } from "@/lib/api"; @@ -20,6 +21,7 @@ export type AssistantForm = { knowledgeRetrievalConfig: KnowledgeRetrievalConfig; enableInterrupt: boolean; turnConfig: TurnConfig; + startup: StartupConfig; visionEnabled: boolean; visionModelResourceId: string; toolIds: string[]; diff --git a/frontend/src/components/editor/vision-config-section.tsx b/frontend/src/components/editor/vision-config-section.tsx new file mode 100644 index 0000000..81007d5 --- /dev/null +++ b/frontend/src/components/editor/vision-config-section.tsx @@ -0,0 +1,72 @@ +"use client"; + +import { Camera } from "lucide-react"; + +import { + ResourceSelectField, + ToggleRow, +} from "@/components/assistant-editor/editor-controls"; +import { SectionCard } from "@/components/editor/section-card"; + +type VisionModelOption = { + value: string; + label: string; +}; + +export function VisionConfigSection({ + description, + hint, + enabled, + modelResourceId, + mainModelResourceId, + modelOptions, + onEnabledChange, + onModelResourceIdChange, +}: { + description: string; + hint: string; + enabled: boolean; + modelResourceId: string; + mainModelResourceId: string; + modelOptions: VisionModelOption[]; + onEnabledChange: (enabled: boolean) => void; + onModelResourceIdChange: (modelResourceId: string) => void; +}) { + const mainModelSupportsVision = modelOptions.some( + (option) => option.value === mainModelResourceId, + ); + const independentModelOptions = modelOptions.filter( + (option) => option.value !== mainModelResourceId, + ); + + return ( + } + title="视觉理解" + description={description} + > + + {enabled && ( + <> + + {!modelResourceId && !mainModelSupportsVision && ( +

+ 当前大语言模型未标记支持图片输入，请选择独立视觉模型。 +

+ )} + + )} + + ); +} diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 63b18ef..5600262 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -177,6 +177,7 @@ function blankPromptForm(name: string): AssistantForm { knowledgeRetrievalConfig: defaultKnowledgeRetrievalConfig(), enableInterrupt: true, turnConfig: defaultTurnConfig(), + startup: { executionMode: "sequential", actions: [] }, visionEnabled: false, visionModelResourceId: "", toolIds: [], @@ -409,7 +410,7 @@ export function AssistantPage(props: AssistantPageProps) { resource.interfaceType === interfaceType, ) .map((resource) => ({ value: resource.id, label: resource.name })); - const visionModelOptionsFor = (currentModelId: string) => modelResources + const visionModelOptionsFor = (currentModelId?: string) => modelResources .filter( (c) => c.capability === "LLM" && @@ -465,6 +466,7 @@ export function AssistantPage(props: AssistantPageProps) { a.knowledgeRetrievalConfig ?? defaultKnowledgeRetrievalConfig(), enableInterrupt: a.enableInterrupt, turnConfig: normalizeTurnConfig(a.turnConfig), + startup: a.startup ?? { executionMode: "sequential", actions: [] }, visionEnabled: a.visionEnabled, visionModelResourceId: a.visionModelResourceId ?? "", toolIds: a.toolIds ?? [], @@ -534,6 +536,7 @@ export function AssistantPage(props: AssistantPageProps) { greeting: "", enableInterrupt: true, turnConfig: defaultTurnConfig(), + startup: { executionMode: "sequential", actions: [] }, visionEnabled: false, visionModelResourceId: null, modelResourceIds: {}, @@ -592,6 +595,7 @@ export function AssistantPage(props: AssistantPageProps) { greeting: form.greeting, enableInterrupt: form.enableInterrupt, turnConfig: form.turnConfig, + startup: form.startup, visionEnabled: form.visionEnabled, visionModelResourceId: form.visionModelResourceId || null, modelResourceIds: { @@ -1558,7 +1562,7 @@ export function AssistantPage(props: AssistantPageProps) { asrOptions={credOptions("ASR")} ttsOptions={credOptions("TTS")} realtimeOptions={credOptions("Realtime")} - visionModelOptions={visionModelOptionsFor(form.model)} + visionModelOptions={visionModelOptionsFor()} knowledgeOptions={kbOptions} tools={tools} onBack={() => router.push("/assistants")} diff --git a/frontend/src/components/workflow/panels/AgentNodePanel.tsx b/frontend/src/components/workflow/panels/AgentNodePanel.tsx index d2ae045..8fa8754 100644 --- a/frontend/src/components/workflow/panels/AgentNodePanel.tsx +++ b/frontend/src/components/workflow/panels/AgentNodePanel.tsx @@ -1,7 +1,6 @@ "use client"; import { - Camera, Bot, Brain, Database, @@ -12,12 +11,9 @@ import { Wrench, } from "lucide-react"; -import { - ResourceSelectField, - ToggleRow, -} from "@/components/assistant-editor/editor-controls"; import { KnowledgeRetrievalConfigDialog } from "@/components/editor/knowledge-retrieval-config-dialog"; import { SectionCard } from "@/components/editor/section-card"; +import { VisionConfigSection } from "@/components/editor/vision-config-section"; import { TurnConfigEditor } from "@/components/turn-config-editor"; import { Input } from "@/components/ui/input"; import { Switch } from "@/components/ui/switch"; @@ -235,48 +231,23 @@ export function AgentNodePanel({ /> - } - title="视觉理解" + - - setPatch({ - visionEnabled, - ...(!visionEnabled - ? { visionModelResourceId: "" } - : {}), - }) - } - /> - {draft.visionEnabled && ( - <> - - set("visionModelResourceId", visionModelResourceId) - } - options={visionOptions.filter( - (option) => option.value !== draft.llmResourceId, - )} - noneLabel="模型自己" - /> - {!draft.visionModelResourceId && - !visionOptions.some( - (option) => option.value === draft.llmResourceId, - ) && ( -

- 当前大语言模型未标记支持图片输入，请选择独立视觉模型。 -

- )} - - )} - + hint="开启后，该 Agent 会获得读取当前视频画面的工具。选择「模型自己」时，当前大语言模型必须支持图片输入。" + enabled={Boolean(draft.visionEnabled)} + modelResourceId={draft.visionModelResourceId ?? ""} + mainModelResourceId={(draft.llmResourceId as string) || ""} + modelOptions={visionOptions} + onEnabledChange={(visionEnabled) => + setPatch({ + visionEnabled, + ...(!visionEnabled ? { visionModelResourceId: "" } : {}), + }) + } + onModelResourceIdChange={(visionModelResourceId) => + set("visionModelResourceId", visionModelResourceId) + } + /> } diff --git a/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx b/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx index 2707259..f4cc419 100644 --- a/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx +++ b/frontend/src/components/workflow/panels/GlobalSettingsPanel.tsx @@ -1,7 +1,6 @@ "use client"; import { - AudioLines, Brain, Database, MessageSquareText, @@ -9,12 +8,9 @@ import { Wrench, } from "lucide-react"; -import { - ResourceSelectField, - ToggleRow, -} from "@/components/assistant-editor/editor-controls"; import { KnowledgeRetrievalConfigDialog } from "@/components/editor/knowledge-retrieval-config-dialog"; import { SectionCard } from "@/components/editor/section-card"; +import { VisionConfigSection } from "@/components/editor/vision-config-section"; import { TurnConfigEditor } from "@/components/turn-config-editor"; import { Textarea } from "@/components/ui/textarea"; @@ -60,8 +56,8 @@ export function GlobalSettingsPanel({ } - title="模型配置" - description="工作流中所有 Agent 共用的大语言模型" + title="模型与语音" + description="继承全局配置的 Agent 共用的推理、语音识别和语音合成资源" > - - onSettingsChange({ - ...settings, - visionEnabled, - ...(!visionEnabled ? { visionModelResourceId: "" } : {}), - }) - } - /> - {settings.visionEnabled && ( - <> - - onSettingsChange({ ...settings, visionModelResourceId }) - } - options={modelOptions.vision.filter( - (option) => option.value !== settings.llm, - )} - noneLabel="模型自己" - /> - {!settings.visionModelResourceId && - !modelOptions.vision.some( - (option) => option.value === settings.llm, - ) && ( -

- 当前大语言模型未标记支持图片输入，请选择独立视觉模型。 -

- )} - - )} - - - } - title="语音配置" - description="Agent 节点未单独选择资源时继承这里的默认值" - > onSettingsChange({ ...settings, asr: v })} + onChange={(asr) => onSettingsChange({ ...settings, asr })} /> onSettingsChange({ ...settings, tts: v })} + onChange={(tts) => onSettingsChange({ ...settings, tts })} /> + + onSettingsChange({ + ...settings, + visionEnabled, + ...(!visionEnabled ? { visionModelResourceId: "" } : {}), + }) + } + onModelResourceIdChange={(visionModelResourceId) => + onSettingsChange({ ...settings, visionModelResourceId }) + } + /> + } title="知识库配置" diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index a480268..4627a44 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -196,6 +196,19 @@ export type TurnConfig = { }; }; +export type StartupAction = { + id: string; + phase: "preflight" | "opening"; + toolId: string; + arguments: Record; + required: boolean; +}; + +export type StartupConfig = { + executionMode: "sequential"; + actions: StartupAction[]; +}; + /** 后端 AssistantOut(宽表 STI:瘦字段平铺,workflow 用 graph)。apiKey 读时打码 */ export type Assistant = { id: string; @@ -205,6 +218,7 @@ export type Assistant = { greeting: string; enableInterrupt: boolean; turnConfig: TurnConfig; + startup: StartupConfig; visionEnabled: boolean; visionModelResourceId: string | null; modelResourceIds: Partial>;