diff --git a/backend/routes/mcp_servers.py b/backend/routes/mcp_servers.py index c9a7379..eb9c4ad 100644 --- a/backend/routes/mcp_servers.py +++ b/backend/routes/mcp_servers.py @@ -237,6 +237,12 @@ async def sync_mcp_tools( "remote_tool_name": remote_name, "input_schema": remote.get("input_schema") or {}, "schema_hash": remote.get("schema_hash") or "", + "allow_interruptions": previous_config.get( + "allow_interruptions", True + ), + "execution_mode": previous_config.get( + "execution_mode", "immediate" + ), "dynamic_variable_assignments": previous_config.get( "dynamic_variable_assignments" ) diff --git a/backend/schemas.py b/backend/schemas.py index b01b118..d9aaddb 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -19,12 +19,13 @@ ModelType = Literal["LLM", "ASR", "TTS", "Realtime", "Embedding", "Agent"] AssistantType = Literal["prompt", "workflow", "dify", "fastgpt", "opencode"] TurnEndStrategy = Literal["silence", "smart_turn"] KnowledgeRetrievalMode = Literal["automatic", "on_demand"] -ToolType = Literal["end_call", "http", "mcp"] +ToolType = Literal["end_call", "http", "mcp", "client"] ToolStatus = Literal["active", "archived", "draft"] McpServerStatus = Literal["active", "archived", "draft"] McpTransport = Literal["streamable_http", "sse"] ToolParameterType = Literal["string", "number", "integer", "boolean", "object", "array"] ToolParameterLocation = Literal["path", "query", "body", "header"] +ToolExecutionMode = Literal["immediate", "async"] DynamicVariableType = Literal["string", "number", "boolean"] # 外部应用类型:其 config.apiKey 是该助手私有密钥,读时打码 / 写时哨兵 @@ -193,6 +194,8 @@ class EndCallToolConfig(CamelModel): class HttpToolConfig(CamelModel): + allow_interruptions: bool = True + execution_mode: ToolExecutionMode = "immediate" method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "GET" url: str timeout_seconds: int = Field(default=15, ge=1, le=120) @@ -221,7 +224,30 @@ class HttpToolDefinition(CamelModel): config: HttpToolConfig +class ClientToolConfig(CamelModel): + allow_interruptions: bool = True + execution_mode: ToolExecutionMode = "async" + wait_for_response: bool = True + parameters: list[ToolParameter] = Field(default_factory=list) + timeout_seconds: int = Field(default=3, ge=1, le=30) + dynamic_variable_assignments: dict[str, str] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_result_assignments(self): + if not self.wait_for_response and self.dynamic_variable_assignments: + raise ValueError("Client Tool 不等待响应时不能配置结果变量赋值") + return self + + +class ClientToolDefinition(CamelModel): + schema_version: int = 1 + type: Literal["client"] = "client" + config: ClientToolConfig = Field(default_factory=ClientToolConfig) + + class McpToolConfig(CamelModel): + allow_interruptions: bool = True + execution_mode: ToolExecutionMode = "immediate" remote_tool_name: str = Field(min_length=1, max_length=255) input_schema: dict[str, Any] = Field(default_factory=dict) schema_hash: str = "" @@ -235,7 +261,12 @@ class McpToolDefinition(CamelModel): ToolDefinition = Annotated[ - Union[EndCallToolDefinition, HttpToolDefinition, McpToolDefinition], + Union[ + EndCallToolDefinition, + HttpToolDefinition, + ClientToolDefinition, + McpToolDefinition, + ], Field(discriminator="type"), ] diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 28d8070..1ff43cd 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -17,6 +17,7 @@ from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.frames.frames import Frame from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor +from services.client_tools import ClientToolPort GREETING_CONTEXT_MARKER = "[会话事实：助手开场白已播放]" @@ -74,6 +75,7 @@ class BrainRuntime: set_system_prompt: Callable[[str], None] set_tools: Callable[[list[FunctionSchema] | None], None] call_end: CallEndPort + client_tools: ClientToolPort | None = None worker: Any = None context_aggregator: Any = None transport: Any = None diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index ec1d123..664ea97 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -20,6 +20,7 @@ from services.runtime_variables import ( DynamicVariableStore, ) from services.tool_executor import ToolExecutionError, ToolExecutor +from services.tool_policy import policy_for_tool class PromptBrain(BaseBrain): @@ -50,17 +51,23 @@ class PromptBrain(BaseBrain): async def setup(self, cfg: AssistantConfig, runtime: BrainRuntime) -> None: self._runtime = runtime + self._tools.set_client_tools(runtime.client_tools) self._waiting_for_generated_end_speech = False schemas: list[FunctionSchema] = [] for tool in cfg.tools: if tool.type == "end_call": schema, handler = self._make_end_call_tool(tool, runtime) - elif tool.type in {"http", "mcp"}: + elif tool.type in {"http", "mcp", "client"}: schema, handler = self._make_remote_tool(tool, runtime) else: continue schemas.append(schema) - runtime.llm.register_function(tool.function_name, handler) + policy = policy_for_tool(tool) + runtime.llm.register_function( + tool.function_name, + handler, + cancel_on_interruption=policy.cancel_on_interruption, + ) runtime.set_tools(schemas) def record_user_message(self, content: str) -> None: @@ -99,16 +106,27 @@ class PromptBrain(BaseBrain): def _make_remote_tool(self, tool, runtime: BrainRuntime): properties, required = self._tools.schema_parts(tool) self._tools.register_secrets(tool) + policy = policy_for_tool(tool) - async def call_http(params: FunctionCallParams) -> None: + async def return_result(params: FunctionCallParams, result: dict) -> None: + if not policy.runs_llm_after_result: + await params.result_callback( + result, + properties=FunctionCallResultProperties(run_llm=False), + ) + else: + await params.result_callback(result) + + async def call_tool(params: FunctionCallParams) -> None: try: result = await self._tools.execute(tool, dict(params.arguments or {})) if result["updated_variables"]: self._refresh_prompt() - await params.result_callback(result) + await return_result(params, result) except (ToolExecutionError, ValueError) as exc: - await params.result_callback( - {"status": "error", "message": f"工具调用失败: {exc}"} + await return_result( + params, + {"status": "error", "message": f"工具调用失败: {exc}"}, ) schema = FunctionSchema( @@ -117,7 +135,7 @@ class PromptBrain(BaseBrain): properties=properties, required=required, ) - return schema, call_http + return schema, call_tool def _make_end_call_tool(self, tool, runtime: BrainRuntime): config = (tool.definition or {}).get("config") or {} diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 255e5cb..1a7c045 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio from copy import deepcopy +from dataclasses import replace from typing import Any from loguru import logger @@ -23,10 +24,15 @@ from pipecat.frames.frames import ( ) from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor +from pipecat.services.llm_service import ( + FunctionCallParams, + FunctionCallResultProperties, +) from services.brains.base import BaseBrain, BrainRuntime, BrainSpec 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 ( RouteStatus, @@ -42,6 +48,38 @@ from services.workflow_router import WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 +class ConfiguredFlowManager(FlowManager): + """Preserve Flow transitions while suppressing late async-tool replies.""" + + async def _create_transition_func(self, name, handler): + transition = await super()._create_transition_func(name, handler) + if not getattr(handler, "_suppress_followup_llm", False): + return transition + + async def configured_transition(params: FunctionCallParams) -> None: + original_callback = params.result_callback + + async def result_callback(result, *, properties=None): + if properties and properties.on_context_updated: + # Deterministic Workflow transitions already use run_llm=False + # and must retain their context-updated callback. + configured_properties = properties + elif properties: + configured_properties = replace(properties, run_llm=False) + else: + configured_properties = FunctionCallResultProperties( + run_llm=False + ) + await original_callback( + result, + properties=configured_properties, + ) + + await transition(replace(params, result_callback=result_callback)) + + return configured_transition + + class WorkflowBrain(BaseBrain): spec = BrainSpec( type="workflow", @@ -100,7 +138,7 @@ class WorkflowBrain(BaseBrain): self._cfg = cfg self._runtime = runtime self._store = DynamicVariableStore.from_config(cfg) - self._tools = ToolExecutor(self._store) + self._tools = ToolExecutor(self._store, client_tools=runtime.client_tools) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) self._edge_evaluator = WorkflowEdgeEvaluator( @@ -120,7 +158,7 @@ class WorkflowBrain(BaseBrain): self._ended = False self._greeting_context_message = None self._startup_waiting_for_greeting = False - self._manager = FlowManager( + self._manager = ConfiguredFlowManager( worker=runtime.worker, llm=runtime.llm, context_aggregator=runtime.context_aggregator, @@ -328,7 +366,7 @@ class WorkflowBrain(BaseBrain): functions: list[FlowsFunctionSchema] = [] for tool_id in stage.tool_ids: tool = self._tool_by_id.get(str(tool_id)) - if tool and tool.type in {"http", "mcp"}: + if tool and tool.type in {"http", "mcp", "client"}: functions.append(self._flow_tool(tool, node_id)) knowledge_function = self._knowledge_function(node_id) if knowledge_function: @@ -409,6 +447,7 @@ class WorkflowBrain(BaseBrain): def _flow_tool(self, tool: RuntimeTool, node_id: str) -> FlowsFunctionSchema: properties, required = self._tools.schema_parts(tool) self._tools.register_secrets(tool) + policy = policy_for_tool(tool) async def handler(args, _flow_manager): transition_id = self._state.transition_id @@ -456,13 +495,21 @@ class WorkflowBrain(BaseBrain): ) return result + if not policy.runs_llm_after_result: + setattr(handler, "_suppress_followup_llm", True) + return FlowsFunctionSchema( name=tool.function_name, description=tool.description or f"调用 {tool.name}", properties=properties, required=required, handler=handler, - cancel_on_interruption=True, + cancel_on_interruption=policy.cancel_on_interruption, + timeout_secs=( + float(((tool.definition or {}).get("config") or {}).get("timeout_seconds") or 3) + if tool.type == "client" + else None + ), ) def _flow_managed_transition_config( diff --git a/backend/services/client_tools.py b/backend/services/client_tools.py new file mode 100644 index 0000000..f2f72ea --- /dev/null +++ b/backend/services/client_tools.py @@ -0,0 +1,136 @@ +"""Conversation-scoped bridge for tools implemented by the connected client.""" + +from __future__ import annotations + +import asyncio +from typing import Any, Protocol + +from loguru import logger +from pipecat.frames.frames import ( + EndFrame, + InputTransportMessageFrame, + OutputTransportMessageUrgentFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class ClientToolError(RuntimeError): + """Raised when a client tool cannot be delivered or completed.""" + + +class ClientToolPort(Protocol): + async def call( + self, + function_name: str, + arguments: dict[str, Any], + *, + timeout_seconds: float, + wait_for_response: bool = True, + ) -> dict[str, Any]: ... + + +class ClientToolBroker(FrameProcessor): + """Send client tool calls and correlate their app-message results.""" + + def __init__(self) -> None: + super().__init__() + self._pending: dict[str, asyncio.Future[dict[str, Any]]] = {} + + async def call( + self, + function_name: str, + arguments: dict[str, Any], + *, + timeout_seconds: float, + wait_for_response: bool = True, + ) -> dict[str, Any]: + from uuid import uuid4 + + tool_call_id = f"client_{uuid4().hex}" + message = { + "type": "client-tool-call", + "tool_call_id": tool_call_id, + "function_name": function_name, + "arguments": dict(arguments), + "wait_for_response": wait_for_response, + } + if not wait_for_response: + try: + await self.push_frame( + OutputTransportMessageUrgentFrame(message=message) + ) + except Exception as exc: + raise ClientToolError( + f"客户端工具调用发送失败: {function_name}" + ) from exc + return { + "status": "ok", + "data": {"dispatched": True}, + } + + loop = asyncio.get_running_loop() + future: asyncio.Future[dict[str, Any]] = loop.create_future() + self._pending[tool_call_id] = future + try: + await self.push_frame( + OutputTransportMessageUrgentFrame(message=message) + ) + return await asyncio.wait_for(future, timeout=timeout_seconds) + except TimeoutError as exc: + raise ClientToolError(f"客户端工具调用超时: {function_name}") from exc + except ClientToolError: + raise + except Exception as exc: + raise ClientToolError(f"客户端工具调用失败: {function_name}") from exc + finally: + self._pending.pop(tool_call_id, None) + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, EndFrame): + self._fail_pending("会话已结束") + await self.push_frame(frame, direction) + return + + if not isinstance(frame, InputTransportMessageFrame): + await self.push_frame(frame, direction) + return + + message = frame.message + if not isinstance(message, dict) or message.get("type") != "client-tool-result": + await self.push_frame(frame, direction) + return + + tool_call_id = str(message.get("tool_call_id") or "") + future = self._pending.get(tool_call_id) + if future is None or future.done(): + logger.debug(f"忽略未知或过期的客户端工具结果: {tool_call_id}") + return + + status = str(message.get("status") or "error") + if status == "ok": + future.set_result( + { + "status": "ok", + "data": message.get("data"), + } + ) + else: + future.set_result( + { + "status": "error", + "message": str(message.get("message") or "客户端工具执行失败"), + "data": message.get("data"), + } + ) + + def _fail_pending(self, message: str) -> None: + for future in self._pending.values(): + if not future.done(): + future.set_exception(ClientToolError(message)) + self._pending.clear() + + async def cleanup(self) -> None: + self._fail_pending("客户端工具通道已关闭") + await super().cleanup() diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 5a397f3..ca342d9 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -30,6 +30,8 @@ from services.pipecat.service_factory import ( ) from db.session import SessionLocal from services.knowledge import search as search_knowledge +from services.client_tools import ClientToolBroker +from services.tool_policy import policy_for_tool from services.vision import ( VISION_ANALYSIS_SYSTEM_PROMPT, VISION_SYSTEM_HINT, @@ -44,6 +46,7 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.flows import FlowsFunctionSchema from pipecat.frames.frames import ( EndFrame, + LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, UserImageRawFrame, UserImageRequestFrame, @@ -57,9 +60,6 @@ from pipecat.processors.aggregators.llm_response_universal import ( ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.llm_service import FunctionCallParams -from pipecat.turns.user_mute.function_call_user_mute_strategy import ( - FunctionCallUserMuteStrategy, -) from services.pipecat.turn_config import ( ConfigurableLLMUserAggregator, create_user_turn_strategies, @@ -73,11 +73,13 @@ from services.pipecat.processors import ( KnowledgeRetrievalProcessor, PassthroughLLMAssistantAggregator, RealtimeDynamicVariableProcessor, - RealtimeTextInputProcessor, - TextInputProcessor, + RealtimeUserInputProcessor, + UserInput, + UserInputProcessor, UserTurnRoutingProcessor, VisionCaptureProcessor, WorkflowAggregatorPair, + ToolInterruptionUserMuteStrategy, ) from services.pipecat.workflow_services import ( WorkflowServiceController, @@ -319,6 +321,14 @@ async def run_pipeline( ) ) input_state = {"enabled": True} + tool_interruption_strategy = ToolInterruptionUserMuteStrategy( + { + tool.function_name: policy_for_tool(tool).execution_mode + for tool in cfg.tools + if tool.type in {"http", "mcp", "client"} + and not policy_for_tool(tool).allow_interruptions + } + ) # LLM 槽由大脑提供:本地模型或 Dify/FastGPT 外部托管适配器。 llm = brain.build_llm( config_with_resource(cfg, default_llm_resource) @@ -335,10 +345,10 @@ async def run_pipeline( params=LLMUserAggregatorParams( vad_analyzer=create_vad_analyzer(cfg.turnConfig), user_mute_strategies=[ - FunctionCallUserMuteStrategy(), CallEndingUserMuteStrategy( lambda: call_end.ending or not input_state["enabled"] ), + tool_interruption_strategy, ], user_turn_strategies=create_user_turn_strategies( cfg.turnConfig, @@ -348,9 +358,14 @@ async def run_pipeline( ) user_turn_router = UserTurnRoutingProcessor(brain) assistant_aggregator = PassthroughLLMAssistantAggregator(context) - text_input = TextInputProcessor( - should_ignore_input=lambda: call_end.ending or not input_state["enabled"] + user_input = UserInputProcessor( + should_ignore_input=lambda: ( + call_end.ending + or not input_state["enabled"] + or tool_interruption_strategy.is_muted + ) ) + client_tools = ClientToolBroker() vision_capture = VisionCaptureProcessor() knowledge_retrieval = KnowledgeRetrievalProcessor( automatic_knowledge_id, @@ -471,6 +486,24 @@ async def run_pipeline( flow_global_functions = [] workflow_vision_function = None + + def active_vision_config() -> AssistantConfig: + if cfg.type != "workflow": + return cfg + if not workflow_vision_scope.get("enabled"): + raise ValueError("当前 Workflow Agent 节点未启用视觉能力") + vision_resource_id = str( + workflow_vision_scope.get("vision_model_resource_id") or "" + ) + llm_resource_id = str(workflow_vision_scope.get("llm_resource_id") or "") + resource_id = vision_resource_id or llm_resource_id + resource = cfg.workflow_model_resources.get(resource_id) + if resource: + return config_with_vision_resource(cfg, resource) + if not vision_resource_id: + return config_with_main_llm_as_vision(cfg) + raise ValueError(f"视觉模型资源未加载:{vision_resource_id}") + if cfg.type == "workflow" and vision_enabled: async def flow_fetch_user_image(args, _flow_manager): if not workflow_vision_scope.get("enabled"): @@ -492,23 +525,9 @@ async def run_pipeline( function_name=VISION_TOOL_NAME, ) try: - vision_resource_id = str( - workflow_vision_scope.get("vision_model_resource_id") or "" - ) - llm_resource_id = str( - workflow_vision_scope.get("llm_resource_id") or "" - ) - resource_id = vision_resource_id or llm_resource_id - resource = cfg.workflow_model_resources.get(resource_id) - if resource: - vision_cfg = config_with_vision_resource(cfg, resource) - elif not vision_resource_id: - vision_cfg = config_with_main_llm_as_vision(cfg) - else: - raise ValueError(f"视觉模型资源未加载:{vision_resource_id}") frame = await vision_capture.request_image(llm, request) observation = await _analyze_image_with_vision_model( - vision_cfg, + active_vision_config(), frame, question, ) @@ -558,8 +577,9 @@ async def run_pipeline( pipeline = Pipeline( [ transport.input(), + client_tools, vision_capture, - text_input, + user_input, stt_processor, user_aggregator, user_turn_router, @@ -638,6 +658,7 @@ async def run_pipeline( set_system_prompt=set_system_prompt, set_tools=set_visible_tools, call_end=call_end, + client_tools=client_tools, worker=worker, context_aggregator=WorkflowAggregatorPair( user_aggregator, @@ -654,17 +675,78 @@ async def run_pipeline( ), ) + async def submit_user_input(value: UserInput) -> None: + if not value.has_camera_frame: + if not value.run_immediately: + brain.record_user_message(value.text) + await worker.queue_frame( + LLMMessagesAppendFrame( + messages=[{"role": "user", "content": value.text}], + run_llm=value.run_immediately, + ) + ) + return + + if not value.run_immediately: + raise ValueError("P0 图片输入必须立即触发回复") + if not vision_enabled: + raise ValueError("当前助手未启用视觉能力") + user_id = vision_state.get("client_id") + if not user_id: + raise ValueError("当前没有可用的摄像头视频流") + analysis_cfg = None if vision_native_mode else active_vision_config() + + request = UserImageRequestFrame( + user_id=user_id, + text=value.prompt_text, + append_to_context=False, + ) + try: + image_frame = await vision_capture.request_image(llm, request) + except asyncio.TimeoutError as exc: + raise ValueError("等待摄像头视频帧超时") from exc + + if vision_native_mode: + image_frame.text = value.prompt_text + image_frame.append_to_context = True + image_frame.request = None + await worker.queue_frame(image_frame) + return + + try: + assert analysis_cfg is not None + observation = await _analyze_image_with_vision_model( + analysis_cfg, + image_frame, + value.prompt_text, + ) + except Exception as exc: + logger.exception(f"用户图片视觉分析失败: {exc}") + raise ValueError("视觉理解暂时不可用") from exc + content = ( + f"{value.prompt_text}\n\n" + "[视觉模型对用户刚提交图片的观察]\n" + f"{observation or '视觉模型没有返回有效观察结果。'}" + ) + await worker.queue_frame( + LLMMessagesAppendFrame( + messages=[{"role": "user", "content": content}], + run_llm=True, + ) + ) + bind_cascade_pipeline_events( transport=transport, worker=worker, brain=brain, context=context, - text_input=text_input, + text_input=user_input, user_aggregator=user_aggregator, assistant_aggregator=assistant_aggregator, greeting=greeting, vision_enabled=vision_enabled, vision_state=vision_state, + submit_user_input=submit_user_input, ) runner = WorkerRunner(handle_sigint=False) run_status = "completed" @@ -694,7 +776,7 @@ async def run_realtime_pipeline( instructions=brain.system_prompt(cfg), ) input_sample_rate, output_sample_rate = realtime_audio_sample_rates(cfg) - text_input = RealtimeTextInputProcessor() + user_input = RealtimeUserInputProcessor() dynamic_variables = RealtimeDynamicVariableProcessor(brain, cfg, realtime) greeting = await brain.greeting(cfg) @@ -708,7 +790,7 @@ async def run_realtime_pipeline( pipeline = Pipeline( [ transport.input(), - text_input, + user_input, realtime, dynamic_variables, ConversationHistoryProcessor(recorder), @@ -729,7 +811,7 @@ async def run_realtime_pipeline( transport=transport, worker=worker, realtime=realtime, - text_input=text_input, + text_input=user_input, greeting=greeting, ) runner = WorkerRunner(handle_sigint=False) diff --git a/backend/services/pipecat/pipeline_events.py b/backend/services/pipecat/pipeline_events.py index ff8e979..5785440 100644 --- a/backend/services/pipecat/pipeline_events.py +++ b/backend/services/pipecat/pipeline_events.py @@ -1,12 +1,13 @@ """Event registration for cascade and realtime conversation pipelines.""" +from collections.abc import Awaitable, Callable + from loguru import logger from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, EndFrame, - LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, TTSSpeakFrame, ) @@ -15,6 +16,7 @@ from pipecat.runner.utils import ( maybe_capture_participant_camera, ) from pipecat.utils.time import time_now_iso8601 +from services.pipecat.processors import UserInput def bind_cascade_pipeline_events( @@ -29,10 +31,11 @@ def bind_cascade_pipeline_events( greeting: str, vision_enabled: bool, vision_state: dict[str, str | None], + submit_user_input: Callable[[UserInput], Awaitable[None]] | None = None, ) -> None: """Connect processors to transport events without owning pipeline assembly.""" - pending_text_inputs: list[str] = [] + pending_user_inputs: list[UserInput] = [] greeting_transcript_sent = False greeting_timestamp = "" greeting_playback_pending = False @@ -72,14 +75,33 @@ def bind_cascade_pipeline_events( ) ) - async def append_user_text_to_context(text: str, *, run_llm: bool) -> None: + async def queue_input_result( + user_input: UserInput, + status: str, + message: str = "", + ) -> None: await worker.queue_frame( - LLMMessagesAppendFrame( - messages=[{"role": "user", "content": text}], - run_llm=run_llm, + OutputTransportMessageUrgentFrame( + message={ + "type": "user-input-result", + "input_id": user_input.input_id, + "status": status, + **({"message": message} if message else {}), + } ) ) + async def finish_user_input(user_input: UserInput) -> None: + try: + if submit_user_input is None: + raise RuntimeError("用户输入提交器尚未配置") + await submit_user_input(user_input) + except Exception as exc: # noqa: BLE001 - input errors must reach the client + logger.warning(f"用户输入处理失败: {exc}") + await queue_input_result(user_input, "error", str(exc)) + return + await queue_input_result(user_input, "accepted") + @user_aggregator.event_handler("on_user_turn_stopped") async def on_user_turn_stopped(_aggregator, _strategy, message): await queue_transcript("user", message.content, message.timestamp) @@ -123,24 +145,23 @@ def bind_cascade_pipeline_events( ) await brain.on_assistant_text_end(turn_id, content, interrupted) - @text_input.event_handler("on_text_input") - async def on_text_input(_processor, text): - pending_text_inputs.append(text) - # The transcript must be queued before the interruption is broadcast. - await queue_transcript("user", text, time_now_iso8601()) + @text_input.event_handler("on_user_input") + async def on_user_input(_processor, user_input: UserInput): + await queue_transcript( + "user", + user_input.transcript_text, + time_now_iso8601(), + ) + if user_input.run_immediately and user_input.interrupt: + pending_user_inputs.append(user_input) + return + await finish_user_input(user_input) @assistant_aggregator.event_handler("on_interruption_processed") async def on_interruption_processed(_aggregator): - if not pending_text_inputs: + if not pending_user_inputs: return - text = pending_text_inputs.pop(0) - await append_user_text_to_context(text, run_llm=True) - - @text_input.event_handler("on_text_append") - async def on_text_append(_processor, text): - brain.record_user_message(text) - await queue_transcript("user", text, time_now_iso8601()) - await append_user_text_to_context(text, run_llm=False) + await finish_user_input(pending_user_inputs.pop(0)) @text_input.event_handler("on_client_ready") async def on_client_ready(_processor): @@ -218,16 +239,24 @@ def bind_realtime_pipeline_events( ) ) - @text_input.event_handler("on_text_input") - async def on_text_input(_processor, text): - await queue_transcript("user", text) - await realtime.interrupt() - await realtime.send_text(text, run_immediately=True) - - @text_input.event_handler("on_text_append") - async def on_text_append(_processor, text): - await queue_transcript("user", text) - await realtime.send_text(text, run_immediately=False) + @text_input.event_handler("on_user_input") + async def on_user_input(_processor, user_input: UserInput): + await queue_transcript("user", user_input.text) + if user_input.run_immediately and user_input.interrupt: + await realtime.interrupt() + await realtime.send_text( + user_input.text, + run_immediately=user_input.run_immediately, + ) + await worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "user-input-result", + "input_id": user_input.input_id, + "status": "accepted", + } + ) + ) @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py index 7452b03..180f0d2 100644 --- a/backend/services/pipecat/processors.py +++ b/backend/services/pipecat/processors.py @@ -2,6 +2,7 @@ import asyncio from collections.abc import Callable +from dataclasses import dataclass from uuid import uuid4 from loguru import logger @@ -12,6 +13,11 @@ from services.knowledge import search as search_knowledge from db.session import SessionLocal from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + FunctionCallCancelFrame, + FunctionCallResultFrame, + FunctionCallsStartedFrame, InputTransportMessageFrame, InterruptionFrame, LLMContextFrame, @@ -34,42 +40,120 @@ from pipecat.utils.time import time_now_iso8601 KNOWLEDGE_CONTEXT_MARKER = "" -def _text_input(message) -> tuple[str, bool] | None: - """解析现有 user-text 与 RTVI send-text 两种前端文字消息。""" +@dataclass(frozen=True) +class UserInput: + """Validated application input submitted as one user turn.""" + + input_id: str + text: str + has_camera_frame: bool + run_immediately: bool + interrupt: bool + + @property + def prompt_text(self) -> str: + if self.text: + return self.text + return "请根据用户刚提交的图片进行回复。" + + @property + def transcript_text(self) -> str: + if not self.has_camera_frame: + return self.text + return f"{self.text}\n已发送一张图片".strip() + + +class UserInputError(ValueError): + def __init__(self, message: str, *, input_id: str = "") -> None: + super().__init__(message) + self.input_id = input_id + + +def parse_user_input(message) -> UserInput | None: + """Parse the sole public user-input wire format.""" if not isinstance(message, dict): return None - if message.get("type") == "user-text": - text = str(message.get("text") or "").strip() - return (text, True) if text else None - if message.get("type") == "send-text": - data = message.get("data") - if not isinstance(data, dict): - return None - text = str(data.get("content") or "").strip() - options = data.get("options") - run_immediately = not isinstance(options, dict) or options.get( - "run_immediately", True - ) - return (text, bool(run_immediately)) if text else None - return None + if message.get("type") != "user-input": + return None + + input_id = str(message.get("input_id") or "").strip() + if not input_id or len(input_id) > 128: + raise UserInputError("user-input 缺少有效的 input_id", input_id=input_id) + if message.get("schema_version") != 1: + raise UserInputError("不支持的 user-input schema_version", input_id=input_id) + + parts = message.get("parts") + if not isinstance(parts, list) or not parts: + raise UserInputError("user-input parts 不能为空", input_id=input_id) + + text = "" + has_camera_frame = False + for part in parts: + if not isinstance(part, dict): + raise UserInputError("user-input part 格式不正确", input_id=input_id) + part_type = part.get("type") + if part_type == "input_text": + if text: + raise UserInputError("P0 只支持一个 input_text", input_id=input_id) + text = str(part.get("text") or "").strip() + if not text: + raise UserInputError("input_text 不能为空", input_id=input_id) + elif part_type == "input_image": + if has_camera_frame: + raise UserInputError("P0 只支持一张图片", input_id=input_id) + source = part.get("source") + if ( + not isinstance(source, dict) + or source.get("type") != "camera_frame" + or source.get("frame") != "current" + ): + raise UserInputError( + "P0 只支持当前摄像头画面", + input_id=input_id, + ) + has_camera_frame = True + else: + raise UserInputError(f"不支持的输入类型: {part_type}", input_id=input_id) + + if not text and not has_camera_frame: + raise UserInputError("user-input 没有有效内容", input_id=input_id) + + options = message.get("options") + options = options if isinstance(options, dict) else {} + run_immediately = bool(options.get("run_immediately", True)) + interrupt = bool(options.get("interrupt", True)) + return UserInput( + input_id=input_id, + text=text, + has_camera_frame=has_camera_frame, + run_immediately=run_immediately, + interrupt=interrupt, + ) -class TextInputProcessor(FrameProcessor): - """把 transport 文字消息转换成 LLM 可消费的帧。 +def message_text(message: dict) -> str: + """Extract textual content from plain or multimodal LLM messages.""" + content = message.get("content") + if isinstance(content, str): + return content.strip() + if not isinstance(content, list): + return "" + return "\n".join( + str(part.get("text") or "").strip() + for part in content + if isinstance(part, dict) + and part.get("type") in {"text", "input_text"} + and str(part.get("text") or "").strip() + ).strip() - run_immediately(默认/打断):先通过 on_text_input 事件把用户文字交给 - run_pipeline 登记,再用 broadcast_interruption() 打断当前播报。新的 LLM - 回复由 assistant aggregator 确认处理完 interruption 后触发。 - run_immediately=False(RTVI send-text 静默追加):仅把文字写进上下文, - 不打断、不触发推理。 - """ + +class UserInputProcessor(FrameProcessor): + """Validate app inputs and coordinate their interruption boundary.""" def __init__(self, should_ignore_input: Callable[[], bool] | None = None): super().__init__() self._should_ignore_input = should_ignore_input or (lambda: False) - # 立即触发的文字(含打断语义)走 on_text_input;静默追加另走一条事件 - self._register_event_handler("on_text_input") - self._register_event_handler("on_text_append") + self._register_event_handler("on_user_input") self._register_event_handler("on_client_ready") async def process_frame(self, frame, direction: FrameDirection): @@ -83,23 +167,35 @@ class TextInputProcessor(FrameProcessor): await self._call_event_handler("on_client_ready") return - parsed = _text_input(frame.message) - if not parsed: + try: + user_input = parse_user_input(frame.message) + except UserInputError as exc: + await self._emit_result(exc.input_id, "error", str(exc)) + return + if user_input is None: await self.push_frame(frame, direction) return if self._should_ignore_input(): logger.debug("通话正在结束,忽略后续文字输入") + await self._emit_result(user_input.input_id, "error", "当前不能接收新的用户输入") return - text, run_immediately = parsed - if run_immediately: - # 先登记文字再打断。下一轮 LLM 由 assistant aggregator 在真正处理完 - # InterruptionFrame 后触发,避免新回复被这次 interruption 一起取消。 - await self._call_event_handler("on_text_input", text) + await self._call_event_handler("on_user_input", user_input) + if user_input.run_immediately and user_input.interrupt: await self.broadcast_interruption() - else: - await self._call_event_handler("on_text_append", text) + + async def _emit_result(self, input_id: str, status: str, message: str = "") -> None: + await self.push_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "user-input-result", + "input_id": input_id, + "status": status, + **({"message": message} if message else {}), + } + ) + ) class CallEndingUserMuteStrategy(BaseUserMuteStrategy): @@ -114,6 +210,84 @@ class CallEndingUserMuteStrategy(BaseUserMuteStrategy): return self._is_call_ending() +@dataclass +class _BlockedToolCall: + tool_finished: bool = False + response_started: bool = False + response_finished: bool = False + + +class ToolInterruptionUserMuteStrategy(BaseUserMuteStrategy): + """Mute user media for selected tools and their associated bot response. + + Pipecat's ``cancel_on_interruption`` also controls whether a tool is + synchronous or asynchronous. This strategy keeps the independent product + setting ("allow user interruptions") at the user-input boundary. + """ + + def __init__( + self, + blocked_tools: dict[str, str] | set[str], + ): + super().__init__() + self._blocked_tools = ( + dict(blocked_tools) + if isinstance(blocked_tools, dict) + else {name: "immediate" for name in blocked_tools} + ) + self._calls: dict[str, _BlockedToolCall] = {} + self._bot_speaking = False + + @property + def is_muted(self) -> bool: + return bool(self._calls) + + async def process_frame(self, frame) -> bool: + await super().process_frame(frame) + + if isinstance(frame, BotStartedSpeakingFrame): + self._bot_speaking = True + for call in self._calls.values(): + call.response_started = True + elif isinstance(frame, BotStoppedSpeakingFrame): + self._bot_speaking = False + for call in self._calls.values(): + if call.response_started: + call.response_finished = True + self._release_completed() + elif isinstance(frame, FunctionCallsStartedFrame): + for call in frame.function_calls: + execution_mode = self._blocked_tools.get(call.function_name) + if execution_mode: + self._calls[call.tool_call_id] = _BlockedToolCall( + # An async tool may continue within the current model + # turn. An immediate tool's current speech is a preamble; + # its protected follow-up starts only after the result. + response_started=( + self._bot_speaking and execution_mode == "async" + ), + ) + elif isinstance(frame, FunctionCallResultFrame): + call = self._calls.get(frame.tool_call_id) + if call is not None: + call.tool_finished = True + self._release_completed() + elif isinstance(frame, FunctionCallCancelFrame): + # A canceled call cannot reliably produce a follow-up response. + self._calls.pop(frame.tool_call_id, None) + + return self.is_muted + + def _release_completed(self) -> None: + completed = [ + call_id + for call_id, state in self._calls.items() + if state.tool_finished and state.response_finished + ] + for call_id in completed: + self._calls.pop(call_id, None) + + class VisionCaptureProcessor(FrameProcessor): """Capture one requested video frame for auxiliary vision-model analysis.""" @@ -193,13 +367,12 @@ class RealtimeDynamicVariableProcessor(FrameProcessor): await self.push_frame(frame, direction) -class RealtimeTextInputProcessor(FrameProcessor): - """Route text input directly to a realtime service without cascade semantics.""" +class RealtimeUserInputProcessor(FrameProcessor): + """Route text-only user-input messages to a realtime service.""" def __init__(self): super().__init__() - self._register_event_handler("on_text_input") - self._register_event_handler("on_text_append") + self._register_event_handler("on_user_input") async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -208,15 +381,32 @@ class RealtimeTextInputProcessor(FrameProcessor): await self.push_frame(frame, direction) return - parsed = _text_input(frame.message) - if not parsed: + try: + user_input = parse_user_input(frame.message) + except UserInputError as exc: + await self._emit_error(exc.input_id, str(exc)) + return + if user_input is None: await self.push_frame(frame, direction) return + if user_input.has_camera_frame: + await self._emit_error( + user_input.input_id, + "Realtime 模式暂不支持图片输入", + ) + return + await self._call_event_handler("on_user_input", user_input) - text, run_immediately = parsed - await self._call_event_handler( - "on_text_input" if run_immediately else "on_text_append", - text, + async def _emit_error(self, input_id: str, message: str) -> None: + await self.push_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "user-input-result", + "input_id": input_id, + "status": "error", + "message": message, + } + ) ) @@ -304,7 +494,7 @@ class KnowledgeRetrievalProcessor(FrameProcessor): if not user_messages: await self.push_frame(frame, direction) return - query = str(user_messages[-1].get("content") or "").strip() + query = message_text(user_messages[-1]) signature = f"{len(user_messages)}:{query}" if not query or signature == self._last_signature: await self.push_frame(frame, direction) @@ -354,8 +544,7 @@ class UserTurnRoutingProcessor(FrameProcessor): message for message in reversed(frame.context.get_messages()) if message.get("role") == "user" - and isinstance(message.get("content"), str) - and str(message.get("content") or "").strip() + and message_text(message) ), None, ) @@ -370,7 +559,7 @@ class UserTurnRoutingProcessor(FrameProcessor): return self._last_user_message = user_message - content = str(user_message.get("content") or "").strip() + content = message_text(user_message) handled = await self._brain.on_user_turn_end(content) if not handled: await self.push_frame(frame, direction) @@ -447,6 +636,3 @@ class WorkflowAggregatorPair: def assistant(self): return self._assistant - - - diff --git a/backend/services/tool_executor.py b/backend/services/tool_executor.py index 98fcd82..c0858dc 100644 --- a/backend/services/tool_executor.py +++ b/backend/services/tool_executor.py @@ -14,6 +14,8 @@ from services.runtime_variables import ( DynamicVariableStore, value_at_path, ) +from services.client_tools import ClientToolError, ClientToolPort +from services.tool_policy import policy_for_tool from services.tools import McpClientError, McpToolClient @@ -29,9 +31,14 @@ class ToolExecutor: store: DynamicVariableStore, *, mcp_client: McpToolClient | None = None, + client_tools: ClientToolPort | None = None, ): self.store = store self._mcp_client = mcp_client or McpToolClient() + self._client_tools = client_tools + + def set_client_tools(self, client_tools: ClientToolPort | None) -> None: + self._client_tools = client_tools def register_secrets(self, tool: RuntimeTool) -> None: dynamic = (tool.secrets or {}).get("dynamic_variables") or {} @@ -78,8 +85,12 @@ class ToolExecutor: result = await self._execute_http(tool, normalized_arguments) elif tool.type == "mcp": result = await self._execute_mcp(tool, normalized_arguments) + elif tool.type == "client": + result = await self._execute_client(tool, normalized_arguments) else: raise ToolExecutionError(f"不支持工具类型: {tool.type}") + if result.get("status") != "ok": + return {**result, "updated_variables": []} return self._apply_result_assignments( tool, result, @@ -187,6 +198,25 @@ class ToolExecutor: raise ToolExecutionError(str(exc)) from exc return {"status": "ok", "data": payload} + async def _execute_client( + self, + tool: RuntimeTool, + arguments: dict[str, Any], + ) -> dict[str, Any]: + if self._client_tools is None: + raise ToolExecutionError("当前运行模式不支持客户端工具") + config = (tool.definition or {}).get("config") or {} + policy = policy_for_tool(tool) + try: + return await self._client_tools.call( + tool.function_name, + arguments, + timeout_seconds=float(config.get("timeout_seconds") or 3), + wait_for_response=policy.wait_for_response, + ) + except ClientToolError as exc: + raise ToolExecutionError(str(exc)) from exc + def _apply_result_assignments( self, tool: RuntimeTool, diff --git a/backend/services/tool_policy.py b/backend/services/tool_policy.py new file mode 100644 index 0000000..f7171d3 --- /dev/null +++ b/backend/services/tool_policy.py @@ -0,0 +1,45 @@ +"""Runtime behavior shared by HTTP, MCP, and client tools. + +The stored definitions are intentionally plain dictionaries. Keeping policy +normalization here gives old definitions stable defaults without spreading +fallback rules across PromptBrain, WorkflowBrain, and the pipeline. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from models import RuntimeTool + + +@dataclass(frozen=True) +class ToolRuntimePolicy: + """Normalized interaction policy for one reusable tool.""" + + allow_interruptions: bool + execution_mode: str + wait_for_response: bool + + @property + def cancel_on_interruption(self) -> bool: + """Pipecat uses this flag to distinguish waiting from async tools.""" + return self.execution_mode == "immediate" + + @property + def runs_llm_after_result(self) -> bool: + """Late async results update context without starting another reply.""" + return self.execution_mode == "immediate" + + +def policy_for_tool(tool: RuntimeTool) -> ToolRuntimePolicy: + """Read policy with backwards-compatible defaults per tool kind.""" + config = (tool.definition or {}).get("config") or {} + default_mode = "async" if tool.type == "client" else "immediate" + mode = str(config.get("execution_mode") or default_mode) + if mode not in {"immediate", "async"}: + mode = default_mode + return ToolRuntimePolicy( + allow_interruptions=bool(config.get("allow_interruptions", True)), + execution_mode=mode, + wait_for_response=bool(config.get("wait_for_response", True)), + ) diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 210422e..463680f 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -34,9 +34,11 @@ from services.workflow.models import LLMRouteResult, RouteStatus class FakeLLM: def __init__(self): self.functions = {} + self.function_options = {} - def register_function(self, name, handler): + def register_function(self, name, handler, **options): self.functions[name] = handler + self.function_options[name] = options class FakeCallEnd: diff --git a/backend/tests/test_client_tools.py b/backend/tests/test_client_tools.py new file mode 100644 index 0000000..728d57c --- /dev/null +++ b/backend/tests/test_client_tools.py @@ -0,0 +1,157 @@ +import asyncio +import unittest + +from models import RuntimeTool +from pipecat.frames.frames import ( + InputTransportMessageFrame, + OutputTransportMessageUrgentFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from services.client_tools import ClientToolBroker, ClientToolError +from services.runtime_variables import DynamicVariableStore +from services.tool_executor import ToolExecutor +from schemas import ClientToolConfig + + +def client_tool() -> RuntimeTool: + return RuntimeTool( + id="tool_photo_button", + name="显示拍照按钮", + function_name="set_photo_button_visible", + type="client", + definition={ + "schema_version": 1, + "type": "client", + "config": { + "parameters": [ + { + "name": "visible", + "type": "boolean", + "required": True, + } + ], + "timeout_seconds": 3, + "dynamic_variable_assignments": { + "photo_button_visible": "visible" + }, + }, + }, + ) + + +class FakeClientTools: + def __init__(self, result): + self.result = result + self.calls = [] + + async def call( + self, + function_name, + arguments, + *, + timeout_seconds, + wait_for_response=True, + ): + self.calls.append( + (function_name, arguments, timeout_seconds, wait_for_response) + ) + return self.result + + +class ClientToolExecutorTests(unittest.IsolatedAsyncioTestCase): + def test_fire_and_forget_rejects_result_assignments(self): + with self.assertRaisesRegex(ValueError, "不能配置结果变量赋值"): + ClientToolConfig( + wait_for_response=False, + dynamic_variable_assignments={"visible": "visible"}, + ) + + async def test_success_updates_configured_variable(self): + store = DynamicVariableStore( + {"photo_button_visible": False}, + variable_types={"photo_button_visible": "boolean"}, + ) + port = FakeClientTools({"status": "ok", "data": {"visible": True}}) + executor = ToolExecutor(store, client_tools=port) + + result = await executor.execute(client_tool(), {"visible": True}) + + self.assertEqual(result["updated_variables"], ["photo_button_visible"]) + self.assertIs(store.values["photo_button_visible"], True) + self.assertEqual( + port.calls, + [("set_photo_button_visible", {"visible": True}, 3.0, True)], + ) + + async def test_failure_does_not_update_variable(self): + store = DynamicVariableStore({"photo_button_visible": False}) + executor = ToolExecutor( + store, + client_tools=FakeClientTools( + {"status": "error", "message": "unsupported"} + ), + ) + + result = await executor.execute(client_tool(), {"visible": True}) + + self.assertEqual(result["updated_variables"], []) + self.assertIs(store.values["photo_button_visible"], False) + + +class ClientToolBrokerTests(unittest.IsolatedAsyncioTestCase): + async def test_correlates_result_and_times_out(self): + broker = ClientToolBroker() + outbound = [] + + async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): + outbound.append((frame, direction)) + + broker.push_frame = push_frame + call = asyncio.create_task( + broker.call("set_photo_button_visible", {"visible": True}, timeout_seconds=1) + ) + await asyncio.sleep(0) + message = outbound[0][0].message + self.assertIsInstance(outbound[0][0], OutputTransportMessageUrgentFrame) + + await broker.process_frame( + InputTransportMessageFrame( + message={ + "type": "client-tool-result", + "tool_call_id": message["tool_call_id"], + "status": "ok", + "data": {"visible": True}, + } + ), + FrameDirection.DOWNSTREAM, + ) + self.assertEqual( + await call, + {"status": "ok", "data": {"visible": True}}, + ) + + with self.assertRaisesRegex(ClientToolError, "超时"): + await broker.call("never_returns", {}, timeout_seconds=0.001) + + async def test_fire_and_forget_does_not_create_pending_call(self): + broker = ClientToolBroker() + outbound = [] + + async def push_frame(frame, direction=FrameDirection.DOWNSTREAM): + outbound.append((frame, direction)) + + broker.push_frame = push_frame + result = await broker.call( + "open_panel", + {"visible": True}, + timeout_seconds=1, + wait_for_response=False, + ) + + self.assertEqual(result, {"status": "ok", "data": {"dispatched": True}}) + self.assertEqual(broker._pending, {}) + self.assertFalse(outbound[0][0].message["wait_for_response"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_pipeline_knowledge.py b/backend/tests/test_pipeline_knowledge.py index 0e80113..3eeadad 100644 --- a/backend/tests/test_pipeline_knowledge.py +++ b/backend/tests/test_pipeline_knowledge.py @@ -92,6 +92,44 @@ class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(brain.turns, ["我叫李白"]) self.assertEqual(forwarded, [(frame, FrameDirection.DOWNSTREAM)]) + async def test_routes_multimodal_user_message_by_its_text_part(self): + class FakeBrain: + def __init__(self): + self.turns = [] + + async def on_user_turn_end(self, content): + self.turns.append(content) + return False + + brain = FakeBrain() + processor = UserTurnRoutingProcessor(brain) + processor.push_frame = lambda *_args, **_kwargs: _async_none() + context = LLMContext( + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "看看这张照片"}, + { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,AA=="}, + }, + ], + } + ] + ) + + await processor.process_frame( + LLMContextFrame(context), + FrameDirection.DOWNSTREAM, + ) + + self.assertEqual(brain.turns, ["看看这张照片"]) + + +async def _async_none(): + return None + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_tool_policy.py b/backend/tests/test_tool_policy.py new file mode 100644 index 0000000..7d0acbf --- /dev/null +++ b/backend/tests/test_tool_policy.py @@ -0,0 +1,142 @@ +import unittest +from types import SimpleNamespace + +from models import RuntimeTool +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + FunctionCallResultFrame, + FunctionCallsStartedFrame, +) +from services.pipecat.processors import ToolInterruptionUserMuteStrategy +from services.tool_policy import policy_for_tool + + +def runtime_tool(tool_type: str, config: dict | None = None) -> RuntimeTool: + return RuntimeTool( + id=f"tool_{tool_type}", + name=tool_type, + function_name=f"run_{tool_type}", + type=tool_type, + definition={"type": tool_type, "config": config or {}}, + ) + + +class ToolPolicyTests(unittest.TestCase): + def test_backwards_compatible_execution_defaults(self): + self.assertEqual( + policy_for_tool(runtime_tool("http")).execution_mode, + "immediate", + ) + self.assertEqual( + policy_for_tool(runtime_tool("client")).execution_mode, + "async", + ) + + def test_explicit_policy_is_normalized(self): + policy = policy_for_tool( + runtime_tool( + "client", + { + "allow_interruptions": False, + "execution_mode": "immediate", + "wait_for_response": False, + }, + ) + ) + self.assertFalse(policy.allow_interruptions) + self.assertTrue(policy.cancel_on_interruption) + self.assertFalse(policy.wait_for_response) + + +class ToolInterruptionStrategyTests(unittest.IsolatedAsyncioTestCase): + async def test_releases_only_after_tool_and_following_speech_finish(self): + strategy = ToolInterruptionUserMuteStrategy({"lookup_order"}) + started = FunctionCallsStartedFrame( + function_calls=[ + SimpleNamespace( + function_name="lookup_order", + tool_call_id="call_1", + ) + ] + ) + + self.assertTrue(await strategy.process_frame(started)) + self.assertTrue( + await strategy.process_frame(BotStartedSpeakingFrame()) + ) + self.assertTrue( + await strategy.process_frame(BotStoppedSpeakingFrame()) + ) + self.assertFalse( + await strategy.process_frame( + FunctionCallResultFrame( + function_name="lookup_order", + tool_call_id="call_1", + arguments={}, + result={"status": "ok"}, + ) + ) + ) + + async def test_waits_for_response_when_tool_finishes_first(self): + strategy = ToolInterruptionUserMuteStrategy({"lookup_order"}) + await strategy.process_frame( + FunctionCallsStartedFrame( + function_calls=[ + SimpleNamespace( + function_name="lookup_order", + tool_call_id="call_2", + ) + ] + ) + ) + self.assertTrue( + await strategy.process_frame( + FunctionCallResultFrame( + function_name="lookup_order", + tool_call_id="call_2", + arguments={}, + result={"status": "ok"}, + ) + ) + ) + await strategy.process_frame(BotStartedSpeakingFrame()) + self.assertFalse( + await strategy.process_frame(BotStoppedSpeakingFrame()) + ) + + async def test_immediate_tool_does_not_count_existing_preamble(self): + strategy = ToolInterruptionUserMuteStrategy( + {"lookup_order": "immediate"} + ) + await strategy.process_frame(BotStartedSpeakingFrame()) + await strategy.process_frame( + FunctionCallsStartedFrame( + function_calls=[ + SimpleNamespace( + function_name="lookup_order", + tool_call_id="call_3", + ) + ] + ) + ) + await strategy.process_frame(BotStoppedSpeakingFrame()) + self.assertTrue( + await strategy.process_frame( + FunctionCallResultFrame( + function_name="lookup_order", + tool_call_id="call_3", + arguments={}, + result={"status": "ok"}, + ) + ) + ) + await strategy.process_frame(BotStartedSpeakingFrame()) + self.assertFalse( + await strategy.process_frame(BotStoppedSpeakingFrame()) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_user_input.py b/backend/tests/test_user_input.py new file mode 100644 index 0000000..e13a634 --- /dev/null +++ b/backend/tests/test_user_input.py @@ -0,0 +1,57 @@ +import unittest + +from services.pipecat.processors import UserInputError, parse_user_input + + +class UserInputParserTests(unittest.TestCase): + def test_parses_text_and_current_camera_frame(self): + value = parse_user_input( + { + "type": "user-input", + "schema_version": 1, + "input_id": "input_1", + "parts": [ + {"type": "input_text", "text": "帮我看看"}, + { + "type": "input_image", + "source": { + "type": "camera_frame", + "frame": "current", + }, + }, + ], + "options": { + "run_immediately": True, + "interrupt": True, + }, + } + ) + + self.assertIsNotNone(value) + self.assertEqual(value.text, "帮我看看") + self.assertTrue(value.has_camera_frame) + self.assertEqual(value.transcript_text, "帮我看看\n已发送一张图片") + + def test_rejects_legacy_and_unsupported_image_sources(self): + self.assertIsNone(parse_user_input({"type": "user-text", "text": "旧协议"})) + with self.assertRaisesRegex(UserInputError, "当前摄像头"): + parse_user_input( + { + "type": "user-input", + "schema_version": 1, + "input_id": "input_2", + "parts": [ + { + "type": "input_image", + "source": { + "type": "uploaded_asset", + "asset_id": "asset_1", + }, + } + ], + } + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/assistant-editor/debug-preview.tsx b/frontend/src/components/assistant-editor/debug-preview.tsx index 7bb326a..0da9b00 100644 --- a/frontend/src/components/assistant-editor/debug-preview.tsx +++ b/frontend/src/components/assistant-editor/debug-preview.tsx @@ -12,6 +12,7 @@ import { Mic, Orbit, PhoneOff, + ScanLine, Send, Smartphone, Sparkles, @@ -43,6 +44,7 @@ import { Textarea } from "@/components/ui/textarea"; import { WaveVisualizer } from "@/components/ui/wave-visualizer"; import { WaveformTimelinePanel } from "@/components/ui/waveform-timeline"; import { useCameraPreview, type CameraPreview } from "@/hooks/use-camera-preview"; +import { usePhotoCaptureTool } from "@/hooks/use-photo-capture-tool"; import { useVoicePreview, type ChatMessage, @@ -626,6 +628,7 @@ function DebugVoicePanel({ dynamicVariables: Record; dynamicVariablesError: string; }) { + const photoCapture = usePhotoCaptureTool(preview); const { status, error, @@ -689,6 +692,10 @@ function DebugVoicePanel({ recording={recording} camera={camera} videoStream={preview.videoStream} + photoCaptureVisible={photoCapture.visible} + photoCapturing={photoCapture.capturing} + photoError={photoCapture.error} + onPhotoCapture={photoCapture.capture} /> ) : view === "video" ? ( showIdleHub ? ( @@ -966,6 +973,10 @@ function DebugVisionWorkspace({ recording, camera, videoStream, + photoCaptureVisible, + photoCapturing, + photoError, + onPhotoCapture, }: { view: DebugView; onViewChange: (view: DebugView) => void; @@ -973,6 +984,10 @@ function DebugVisionWorkspace({ recording: boolean; camera: CameraPreview; videoStream: MediaStream | null; + photoCaptureVisible: boolean; + photoCapturing: boolean; + photoError: string | null; + onPhotoCapture: () => Promise; }) { const containerRef = useRef(null); const [position, setPosition] = useState({ x: 16, y: 16 }); @@ -1038,6 +1053,14 @@ function DebugVisionWorkspace({ + {photoCaptureVisible && ( + + )} ); } @@ -1074,6 +1097,51 @@ function DebugVisionWorkspace({ > + {photoCaptureVisible && ( + + )} + + ); +} + +function DebugPhotoCaptureButton({ + capturing, + enabled, + error, + onCapture, +}: { + capturing: boolean; + enabled: boolean; + error: string | null; + onCapture: () => Promise; +}) { + return ( +
+ {error && ( +
+ {error} +
+ )} +
); } @@ -1250,4 +1318,3 @@ function DebugTranscriptPanel({ ); } - diff --git a/frontend/src/components/assistant-editor/editor-controls.tsx b/frontend/src/components/assistant-editor/editor-controls.tsx index 6b85fb6..f2fc8e9 100644 --- a/frontend/src/components/assistant-editor/editor-controls.tsx +++ b/frontend/src/components/assistant-editor/editor-controls.tsx @@ -546,7 +546,9 @@ export function ToolPicker({ ? "End Call" : tool.type === "mcp" ? "MCP" - : "HTTP"} + : tool.type === "client" + ? "Client" + : "HTTP"}
diff --git a/frontend/src/components/pages/ComponentsToolsPage.tsx b/frontend/src/components/pages/ComponentsToolsPage.tsx index 6db51bb..30408a1 100644 --- a/frontend/src/components/pages/ComponentsToolsPage.tsx +++ b/frontend/src/components/pages/ComponentsToolsPage.tsx @@ -58,19 +58,26 @@ import { type McpServer, type Tool, type ToolParameter, + type ToolExecutionMode, type ToolStatus, type ToolUpsert, } from "@/lib/api"; -type ToolKind = "end_call" | "http"; +type ToolKind = "end_call" | "http" | "client"; type HttpMethod = HttpToolDefinition["config"]["method"]; -type ToolFilter = "全部" | "End Call" | "HTTP" | "MCP"; +type ToolFilter = "全部" | "End Call" | "HTTP" | "Client" | "MCP"; type SortOrder = "newest" | "oldest"; type ToolResource = | { kind: "tool"; id: string; tool: Tool } | { kind: "mcp"; id: string; server: McpServer }; -const toolFilters: readonly ToolFilter[] = ["全部", "End Call", "HTTP", "MCP"]; +const toolFilters: readonly ToolFilter[] = [ + "全部", + "End Call", + "HTTP", + "Client", + "MCP", +]; type ToolForm = { name: string; @@ -84,10 +91,14 @@ type ToolForm = { method: HttpMethod; url: string; timeoutSeconds: string; + allowInterruptions: boolean; + executionMode: ToolExecutionMode; + waitForResponse: boolean; headers: string; secretHeaders: string; parameters: string; body: string; + updateDynamicVariables: boolean; dynamicVariableAssignments: string; secretDynamicVariables: string; }; @@ -108,10 +119,14 @@ function blankForm(): ToolForm { method: "GET", url: "", timeoutSeconds: "15", + allowInterruptions: true, + executionMode: "immediate", + waitForResponse: true, headers: EMPTY_OBJECT, secretHeaders: EMPTY_OBJECT, parameters: EMPTY_ARRAY, body: EMPTY_OBJECT, + updateDynamicVariables: false, dynamicVariableAssignments: EMPTY_OBJECT, secretDynamicVariables: EMPTY_OBJECT, }; @@ -136,7 +151,25 @@ function formFromTool(tool: Tool): ToolForm { return base; } if (tool.definition.type === "mcp") return base; + if (tool.definition.type === "client") { + base.allowInterruptions = + tool.definition.config.allowInterruptions ?? true; + base.executionMode = tool.definition.config.executionMode ?? "async"; + base.waitForResponse = tool.definition.config.waitForResponse ?? true; + base.timeoutSeconds = String(tool.definition.config.timeoutSeconds); + base.parameters = pretty(tool.definition.config.parameters, EMPTY_ARRAY); + base.dynamicVariableAssignments = pretty( + tool.definition.config.dynamicVariableAssignments ?? {}, + EMPTY_OBJECT, + ); + base.updateDynamicVariables = + Object.keys(tool.definition.config.dynamicVariableAssignments ?? {}).length > 0; + return base; + } base.method = tool.definition.config.method; + base.allowInterruptions = + tool.definition.config.allowInterruptions ?? true; + base.executionMode = tool.definition.config.executionMode ?? "immediate"; base.url = tool.definition.config.url; base.timeoutSeconds = String(tool.definition.config.timeoutSeconds); base.headers = pretty(tool.definition.config.headers, EMPTY_OBJECT); @@ -215,6 +248,39 @@ function payloadFromForm(form: ToolForm): ToolUpsert { }, }; } + if (form.type === "client") { + const timeoutSeconds = Number(form.timeoutSeconds); + if ( + !Number.isInteger(timeoutSeconds) || + timeoutSeconds < 1 || + timeoutSeconds > 30 + ) { + throw new Error("Client Tool 超时时间必须是 1 到 30 秒之间的整数"); + } + const dynamicVariableAssignments = form.updateDynamicVariables + ? parseObject(form.dynamicVariableAssignments, "变量赋值") + : {}; + return { + name: form.name.trim(), + functionName: form.functionName, + description: form.description.trim(), + status: form.status, + definition: { + schemaVersion: 1, + type: "client", + config: { + allowInterruptions: form.allowInterruptions, + executionMode: form.executionMode, + waitForResponse: form.waitForResponse, + parameters: parseParameters(form.parameters), + timeoutSeconds, + dynamicVariableAssignments: + dynamicVariableAssignments as Record, + }, + }, + secrets: {}, + }; + } const timeoutSeconds = Number(form.timeoutSeconds); if (!Number.isInteger(timeoutSeconds) || timeoutSeconds < 1 || timeoutSeconds > 120) { throw new Error("超时时间必须是 1 到 120 秒之间的整数"); @@ -240,6 +306,8 @@ function payloadFromForm(form: ToolForm): ToolUpsert { schemaVersion: 1, type: "http", config: { + allowInterruptions: form.allowInterruptions, + executionMode: form.executionMode, method: form.method, url: form.url.trim(), timeoutSeconds, @@ -340,7 +408,8 @@ export function ComponentsToolsPage() { return ( (filter === "全部" || (filter === "End Call" && tool.type === "end_call") || - (filter === "HTTP" && tool.type === "http")) && + (filter === "HTTP" && tool.type === "http") || + (filter === "Client" && tool.type === "client")) && (!query || [tool.name, tool.functionName, tool.description].some((value) => value.toLowerCase().includes(query), @@ -491,7 +560,9 @@ export function ComponentsToolsPage() { ? "MCP" : resource.tool.type === "end_call" ? "End Call" - : "HTTP"} + : resource.tool.type === "client" + ? "Client" + : "HTTP"} ), }, @@ -684,13 +755,22 @@ export function ComponentsToolsPage() { value={form.type} disabled={Boolean(editing)} onValueChange={(type: ToolKind) => - setForm((current) => ({ ...current, type })) + setForm((current) => ({ + ...current, + type, + ...(type === "client" + ? { timeoutSeconds: "3", executionMode: "async" as const } + : type === "http" + ? { timeoutSeconds: "15", executionMode: "immediate" as const } + : {}), + })) } > End Call HTTP + Client Tool @@ -733,6 +813,8 @@ export function ComponentsToolsPage() { {form.type === "end_call" ? ( + ) : form.type === "client" ? ( + ) : ( )} @@ -813,6 +895,97 @@ function EndCallFields({ ); } +function ClientToolFields({ + form, + setForm, +}: { + form: ToolForm; + setForm: React.Dispatch>; +}) { + return ( +
+ +
+
+
等待客户端响应
+
+ 关闭后只下发调用，不等待结果、错误或超时 +
+
+ + setForm((current) => ({ + ...current, + waitForResponse, + updateDynamicVariables: waitForResponse + ? current.updateDynamicVariables + : false, + })) + } + /> +
+ {form.waitForResponse && ( + + + setForm((current) => ({ + ...current, + timeoutSeconds: event.target.value, + })) + } + /> + + )} + + setForm((current) => ({ ...current, parameters })) + } + rows={8} + /> +
+
+
+ 成功后更新动态变量 +
+
+ 只在客户端返回 status=ok 时应用结果映射 +
+
+ + setForm((current) => ({ ...current, updateDynamicVariables })) + } + /> +
+ {form.updateDynamicVariables && ( + + setForm((current) => ({ + ...current, + dynamicVariableAssignments, + })) + } + /> + )} +

+ 函数名必须与客户端注册的 handler 名称一致，例如 + set_photo_button_visible。 +

+
+ ); +} + function HttpFields({ form, setForm, @@ -822,6 +995,7 @@ function HttpFields({ }) { return (
+
+ setForm((current) => ({ ...current, executionMode })) + } + > + + + + + 立即执行 · 等待结果后继续 + 异步执行 · 工具与对话并行 + + + +
+
+
允许用户打断
+
+ 关闭后，工具运行和紧随其后的 Agent 播报期间会忽略用户输入 +
+
+ + setForm((current) => ({ ...current, allowInterruptions })) + } + /> +
+
+ ); +} diff --git a/frontend/src/components/pages/MobileCallPage.tsx b/frontend/src/components/pages/MobileCallPage.tsx index c995299..7aa3e8a 100644 --- a/frontend/src/components/pages/MobileCallPage.tsx +++ b/frontend/src/components/pages/MobileCallPage.tsx @@ -9,6 +9,7 @@ import { Mic, Phone, PhoneOff, + ScanLine, Video, } from "lucide-react"; import { Button } from "@/components/ui/button"; @@ -30,6 +31,7 @@ import { SelectValue, } from "@/components/ui/select"; import { useCameraPreview } from "@/hooks/use-camera-preview"; +import { usePhotoCaptureTool } from "@/hooks/use-photo-capture-tool"; import { useVoicePreview, type ChatMessage, @@ -420,6 +422,7 @@ function CallCameraDeviceSelect({ export function MobileCallPage({ assistantId }: { assistantId: string }) { const preview = useVoicePreview(assistantId); + const photoCapture = usePhotoCaptureTool(preview); const camera = useCameraPreview(); const { status, @@ -475,7 +478,7 @@ export function MobileCallPage({ assistantId }: { assistantId: string }) { // 首次打开页面时自动发起通话；hooks 会在卸载时各自释放媒体资源。 }, [startCall]); - const error = previewError || cameraError; + const error = previewError || cameraError || photoCapture.error; return (
@@ -562,6 +565,22 @@ export function MobileCallPage({ assistantId }: { assistantId: string }) { > + {photoCapture.visible && ( + + )} toggleTool(tool.id)} + onUpdated={(updated) => { + setServerTools((current) => + current.map((item) => + item.id === updated.id ? updated : item, + ), + ); + void onChanged(); + }} /> ))}
@@ -390,12 +401,22 @@ function McpToolRow({ tool, expanded, onToggle, + onUpdated, }: { tool: Tool; expanded: boolean; onToggle: () => void; + onUpdated: (tool: Tool) => void; }) { const definition = tool.definition as McpToolDefinition; + const [allowInterruptions, setAllowInterruptions] = useState( + definition.config.allowInterruptions ?? true, + ); + const [executionMode, setExecutionMode] = useState( + definition.config.executionMode ?? "immediate", + ); + const [savingPolicy, setSavingPolicy] = useState(false); + const [policyError, setPolicyError] = useState(null); const schema = definition.config.inputSchema ?? {}; const properties = useMemo( () => Object.entries((schema.properties as Record | undefined) ?? {}), @@ -403,6 +424,37 @@ function McpToolRow({ ); const required = new Set(Array.isArray(schema.required) ? schema.required.map(String) : []); + async function savePolicy() { + if (savingPolicy) return; + setSavingPolicy(true); + setPolicyError(null); + try { + const updated = await toolsApi.update(tool.id, { + name: tool.name, + functionName: tool.functionName, + description: tool.description, + definition: { + ...definition, + config: { + ...definition.config, + allowInterruptions, + executionMode, + }, + }, + secrets: tool.secrets, + status: tool.status, + mcpServerId: tool.mcpServerId, + }); + onUpdated(updated); + } catch (error) { + setPolicyError( + error instanceof Error ? error.message : "保存工具运行策略失败", + ); + } finally { + setSavingPolicy(false); + } + } + return (
+ {policyError && ( + {policyError} + )} +
+
{properties.length === 0 ? (
此工具不需要参数。
) : ( diff --git a/frontend/src/hooks/use-photo-capture-tool.ts b/frontend/src/hooks/use-photo-capture-tool.ts new file mode 100644 index 0000000..1e0db9f --- /dev/null +++ b/frontend/src/hooks/use-photo-capture-tool.ts @@ -0,0 +1,66 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; + +import type { VoicePreview } from "@/hooks/use-voice-preview"; + +const PHOTO_BUTTON_TOOL = "set_photo_button_visible"; + +export function usePhotoCaptureTool(preview: VoicePreview) { + const { registerClientTool, sendUserInput, status } = preview; + const [visible, setVisible] = useState(false); + const [capturing, setCapturing] = useState(false); + const [error, setError] = useState(null); + + useEffect( + () => + registerClientTool(PHOTO_BUTTON_TOOL, ({ visible: nextValue }) => { + if (typeof nextValue !== "boolean") { + throw new Error("visible 参数必须是布尔值"); + } + setVisible(nextValue); + setError(null); + return { visible: nextValue }; + }), + [registerClientTool], + ); + + useEffect(() => { + if (status === "connected") return; + // eslint-disable-next-line react-hooks/set-state-in-effect + setVisible(false); + setCapturing(false); + }, [status]); + + const capture = useCallback(async () => { + if (capturing || status !== "connected") return; + setCapturing(true); + setError(null); + try { + await sendUserInput( + [ + { + type: "input_image", + source: { type: "camera_frame", frame: "current" }, + }, + ], + { runImmediately: true, interrupt: true }, + ); + } catch (captureError) { + setError( + captureError instanceof Error ? captureError.message : "拍照提交失败", + ); + } finally { + setCapturing(false); + } + }, [capturing, sendUserInput, status]); + + return { + visible, + capturing, + error, + capture, + }; +} + +export type PhotoCaptureTool = ReturnType; diff --git a/frontend/src/hooks/use-voice-preview.ts b/frontend/src/hooks/use-voice-preview.ts index 4fddee1..ba591aa 100644 --- a/frontend/src/hooks/use-voice-preview.ts +++ b/frontend/src/hooks/use-voice-preview.ts @@ -47,6 +47,25 @@ export type ChatMessage = { type AppMessage = Record & { type?: string }; type DynamicVariableValue = string | number | boolean; +export type UserInputPart = + | { type: "input_text"; text: string } + | { + type: "input_image"; + source: { type: "camera_frame"; frame: "current" }; + }; +export type UserInputResult = { + inputId: string; + status: "accepted"; +}; +export type ClientToolHandler = ( + argumentsValue: Record, +) => unknown | Promise; + +type PendingUserInput = { + resolve: (result: UserInputResult) => void; + reject: (error: Error) => void; + timeout: number; +}; function publicVariableSnapshot( value: unknown, @@ -116,6 +135,13 @@ function encodeHeaderJson(value: unknown): string { return btoa(binary).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, ""); } +function newInputId(): string { + if (typeof crypto !== "undefined" && "randomUUID" in crypto) { + return `input_${crypto.randomUUID()}`; + } + return `input_${Date.now()}_${Math.random().toString(16).slice(2)}`; +} + function readNetworkMetrics(report: RTCStatsReport): NetworkMetrics | null { const outbound: Partial> = {}; const lost: Partial> = {}; @@ -219,6 +245,8 @@ export function useVoicePreview( const pendingAssistantTurnsRef = useRef( new Map(), ); + const pendingUserInputsRef = useRef(new Map()); + const clientToolHandlersRef = useRef(new Map()); const endedByServerRef = useRef(false); const selectedDeviceIdRef = useRef(""); const selectedOutputDeviceIdRef = useRef(""); @@ -273,6 +301,11 @@ export function useVoicePreview( if (audioRef.current) audioRef.current.srcObject = null; startingRef.current = false; pendingAssistantTurnsRef.current.clear(); + pendingUserInputsRef.current.forEach(({ reject, timeout }) => { + window.clearTimeout(timeout); + reject(new Error("连接已关闭，用户输入未完成")); + }); + pendingUserInputsRef.current.clear(); networkStatsRef.current = null; onNodeActiveRef.current?.(null); }, []); @@ -301,8 +334,78 @@ export function useVoicePreview( setStatus("failed"); }, [releaseResources]); + const dispatchClientTool = useCallback(async (msg: AppMessage) => { + const toolCallId = + typeof msg.tool_call_id === "string" ? msg.tool_call_id : ""; + const functionName = + typeof msg.function_name === "string" ? msg.function_name : ""; + const waitForResponse = msg.wait_for_response !== false; + if (!toolCallId || !functionName) return; + + const handler = clientToolHandlersRef.current.get(functionName); + const transport = transportRef.current; + if (!transport || transport.state !== "connected") return; + if (!handler) { + if (waitForResponse) { + transport.sendAppMessage({ + type: "client-tool-result", + tool_call_id: toolCallId, + status: "error", + message: `客户端未注册工具: ${functionName}`, + }); + } + return; + } + + try { + const argumentsValue = + msg.arguments && + typeof msg.arguments === "object" && + !Array.isArray(msg.arguments) + ? (msg.arguments as Record) + : {}; + const data = await handler(argumentsValue); + if (transportRef.current !== transport) return; + if (!waitForResponse) return; + transport.sendAppMessage({ + type: "client-tool-result", + tool_call_id: toolCallId, + status: "ok", + data, + }); + } catch (toolError) { + if (transportRef.current !== transport) return; + if (!waitForResponse) return; + transport.sendAppMessage({ + type: "client-tool-result", + tool_call_id: toolCallId, + status: "error", + message: errorMessage(toolError, "客户端工具执行失败"), + }); + } + }, []); + const handleAppMessage = useCallback((msg: AppMessage) => { - if ( + if (msg.type === "client-tool-call") { + void dispatchClientTool(msg); + } else if ( + msg.type === "user-input-result" && + typeof msg.input_id === "string" + ) { + const pending = pendingUserInputsRef.current.get(msg.input_id); + if (!pending) return; + pendingUserInputsRef.current.delete(msg.input_id); + window.clearTimeout(pending.timeout); + if (msg.status === "accepted") { + pending.resolve({ inputId: msg.input_id, status: "accepted" }); + } else { + pending.reject( + new Error( + typeof msg.message === "string" ? msg.message : "用户输入处理失败", + ), + ); + } + } else if ( msg.type === "assistant-text-start" && typeof msg.turn_id === "string" ) { @@ -413,7 +516,7 @@ export function useVoicePreview( setCallEnded(true); disconnect(); } - }, [disconnect]); + }, [disconnect, dispatchClientTool]); const connect = useCallback(async (options: ConnectOptions = {}) => { if (startingRef.current || transportRef.current) return; @@ -601,10 +704,68 @@ export function useVoicePreview( const trimmed = text.trim(); const transport = transportRef.current; if (!trimmed || !transport || transport.state !== "connected") return false; - transport.sendAppMessage({ type: "user-text", text: trimmed }); + const inputId = newInputId(); + transport.sendAppMessage({ + type: "user-input", + schema_version: 1, + input_id: inputId, + parts: [{ type: "input_text", text: trimmed }], + options: { run_immediately: true, interrupt: true }, + }); return true; }, []); + const sendUserInput = useCallback( + ( + parts: UserInputPart[], + options: { runImmediately?: boolean; interrupt?: boolean } = {}, + ): Promise => { + const transport = transportRef.current; + if (!transport || transport.state !== "connected") { + return Promise.reject(new Error("当前未连接语音服务")); + } + const inputId = newInputId(); + return new Promise((resolve, reject) => { + const timeout = window.setTimeout(() => { + pendingUserInputsRef.current.delete(inputId); + reject(new Error("等待用户输入处理结果超时")); + }, 30_000); + pendingUserInputsRef.current.set(inputId, { resolve, reject, timeout }); + try { + transport.sendAppMessage({ + type: "user-input", + schema_version: 1, + input_id: inputId, + parts, + options: { + run_immediately: options.runImmediately ?? true, + interrupt: options.interrupt ?? true, + }, + }); + } catch (sendError) { + window.clearTimeout(timeout); + pendingUserInputsRef.current.delete(inputId); + reject( + new Error(errorMessage(sendError, "发送用户输入失败")), + ); + } + }); + }, + [], + ); + + const registerClientTool = useCallback( + (functionName: string, handler: ClientToolHandler): (() => void) => { + clientToolHandlersRef.current.set(functionName, handler); + return () => { + if (clientToolHandlersRef.current.get(functionName) === handler) { + clientToolHandlersRef.current.delete(functionName); + } + }; + }, + [], + ); + useEffect(() => releaseResources, [releaseResources]); return { @@ -627,6 +788,8 @@ export function useVoicePreview( selectOutputDevice, supportsOutputSelection, sendText, + sendUserInput, + registerClientTool, connect, replaceVideoStream, selectCamera, diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index fc307c3..eaee13f 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -315,6 +315,7 @@ export const conversationsApi = { // ---------- 工具 ---------- export type ToolStatus = "active" | "archived" | "draft"; +export type ToolExecutionMode = "immediate" | "async"; export type ToolParameter = { name: string; type: "string" | "number" | "integer" | "boolean" | "object" | "array"; @@ -337,6 +338,8 @@ export type HttpToolDefinition = { schemaVersion: number; type: "http"; config: { + allowInterruptions: boolean; + executionMode: ToolExecutionMode; method: "GET" | "POST" | "PUT" | "PATCH" | "DELETE"; url: string; timeoutSeconds: number; @@ -347,10 +350,25 @@ export type HttpToolDefinition = { }; }; +export type ClientToolDefinition = { + schemaVersion: number; + type: "client"; + config: { + allowInterruptions: boolean; + executionMode: ToolExecutionMode; + waitForResponse: boolean; + parameters: ToolParameter[]; + timeoutSeconds: number; + dynamicVariableAssignments: Record; + }; +}; + export type McpToolDefinition = { schemaVersion: number; type: "mcp"; config: { + allowInterruptions: boolean; + executionMode: ToolExecutionMode; remoteToolName: string; inputSchema: Record; schemaHash: string; @@ -362,9 +380,13 @@ export type Tool = { id: string; name: string; functionName: string; - type: "end_call" | "http" | "mcp"; + type: "end_call" | "http" | "mcp" | "client"; description: string; - definition: EndCallToolDefinition | HttpToolDefinition | McpToolDefinition; + definition: + | EndCallToolDefinition + | HttpToolDefinition + | ClientToolDefinition + | McpToolDefinition; secrets: Record; status: ToolStatus; mcpServerId?: string | null;