From 4a57b290d3c86b6fbf131bec296391eacb616bad Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 10 Jul 2026 14:32:10 +0800 Subject: [PATCH] Add RuntimeTool model and enhance AssistantConfig for tool management - Introduce a new `RuntimeTool` model to encapsulate tool data for runtime sessions, including attributes like `id`, `name`, `function_name`, `type`, and `description`. - Update the `AssistantConfig` model to include a list of reusable tools, allowing for better management of tools within assistant configurations. - Modify the `config_resolver` service to fetch and resolve tools associated with assistants, ensuring they are available during runtime. - Refactor tool-related CRUD operations in the `tools` route to support the new runtime execution model, enhancing the overall tool management system. - Update documentation and comments to reflect changes in tool execution and configuration handling, improving clarity for future development. --- backend/db/models.py | 2 +- backend/models.py | 14 +++ backend/routes/tools.py | 2 +- backend/services/config_resolver.py | 38 +++++- backend/services/pipecat/pipeline.py | 171 +++++++++++++++++++++++---- 5 files changed, 202 insertions(+), 25 deletions(-) diff --git a/backend/db/models.py b/backend/db/models.py index a5242da..6b48714 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -148,7 +148,7 @@ class AssistantModelBinding(Base): class Tool(Base): - """Reusable LLM tool definition. Runtime execution is added separately.""" + """Reusable LLM tool definition; supported types are executed at runtime.""" __tablename__ = "tools" diff --git a/backend/models.py b/backend/models.py index eede063..5209de6 100644 --- a/backend/models.py +++ b/backend/models.py @@ -15,6 +15,17 @@ from pydantic import BaseModel, Field RuntimeMode = Literal["pipeline", "realtime"] +class RuntimeTool(BaseModel): + """Tool data resolved from an assistant binding for one runtime session.""" + + id: str + name: str + function_name: str + type: str + description: str = "" + definition: dict = Field(default_factory=dict) + + class AssistantConfig(BaseModel): """运行时配置:前端可见部分(name/prompt/...) + 服务端注入部分(*_api_key/*_base_url)。""" @@ -61,6 +72,9 @@ class AssistantConfig(BaseModel): enableInterrupt: bool = True + # Prompt assistant reusable tools. Execution remains type-specific in the pipeline. + tools: list[RuntimeTool] = Field(default_factory=list) + # workflow 类型:节点图(nodes/edges)。非 workflow 为空,引擎据此决定是否启用。 graph: dict = {} diff --git a/backend/routes/tools.py b/backend/routes/tools.py index 29baf59..6e90585 100644 --- a/backend/routes/tools.py +++ b/backend/routes/tools.py @@ -1,4 +1,4 @@ -"""Reusable tool CRUD. Tool execution is intentionally not wired to pipelines yet.""" +"""Reusable tool CRUD. Runtime execution is implemented per supported tool type.""" import uuid diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index d9b96ea..7aff562 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -4,8 +4,14 @@ 助手按 capability binding 引用资源;取不到则回退该能力默认资源。 """ -from db.models import Assistant, AssistantModelBinding, ModelResource -from models import AssistantConfig +from db.models import ( + Assistant, + AssistantModelBinding, + AssistantToolBinding, + ModelResource, + Tool, +) +from models import AssistantConfig, RuntimeTool from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -75,6 +81,33 @@ def _secret(resource: ModelResource | None, key: str, default: str = "") -> str: return str((resource.secrets or {}).get(key) or default) +async def _tools_for(session: AsyncSession, assistant: Assistant) -> list[RuntimeTool]: + if assistant.type != "prompt": + return [] + tools = ( + await session.execute( + select(Tool) + .join(AssistantToolBinding, AssistantToolBinding.tool_id == Tool.id) + .where( + AssistantToolBinding.assistant_id == assistant.id, + Tool.status == "active", + ) + .order_by(AssistantToolBinding.created_at, Tool.id) + ) + ).scalars().all() + return [ + RuntimeTool( + id=tool.id, + name=tool.name, + function_name=tool.function_name, + type=tool.type, + description=tool.description, + definition=tool.definition or {}, + ) + for tool in tools + ] + + async def resolve_runtime_config( session: AsyncSession, assistant_id: str ) -> AssistantConfig: @@ -102,6 +135,7 @@ async def resolve_runtime_config( prompt=assistant.prompt or "你是一个有帮助的助手。", runtimeMode=assistant.runtime_mode, # type: ignore[arg-type] enableInterrupt=assistant.enable_interrupt, + tools=await _tools_for(session, assistant), # workflow 图:仅 workflow 类型非空,引擎据此启用图驱动对话 graph=(assistant.graph or {}) if assistant.type == "workflow" else {}, # 外部托管类型连接信息(DB 存真 key,直接注入) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index b2cf73f..923ede3 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -8,6 +8,7 @@ import asyncio import base64 +from collections.abc import Callable from io import BytesIO from uuid import uuid4 @@ -55,11 +56,18 @@ from pipecat.runner.utils import ( get_transport_client_id, maybe_capture_participant_camera, ) -from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.llm_service import ( + FunctionCallParams, + FunctionCallResultProperties, +) from pipecat.turns.user_start import ( TranscriptionUserTurnStartStrategy, VADUserTurnStartStrategy, ) +from pipecat.turns.user_mute.base_user_mute_strategy import BaseUserMuteStrategy +from pipecat.turns.user_mute.function_call_user_mute_strategy import ( + FunctionCallUserMuteStrategy, +) from pipecat.turns.user_turn_strategies import UserTurnStrategies from pipecat.utils.time import time_now_iso8601 from pipecat.workers.runner import WorkerRunner @@ -169,8 +177,9 @@ class TextInputProcessor(FrameProcessor): 不打断、不触发推理。 """ - def __init__(self): + 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") @@ -192,6 +201,10 @@ class TextInputProcessor(FrameProcessor): await self.push_frame(frame, direction) return + if self._should_ignore_input(): + logger.debug("通话正在结束,忽略后续文字输入") + return + text, run_immediately = parsed if run_immediately: # 先登记文字再打断。下一轮 LLM 由 assistant aggregator 在真正处理完 @@ -202,6 +215,18 @@ class TextInputProcessor(FrameProcessor): await self._call_event_handler("on_text_append", text) +class CallEndingUserMuteStrategy(BaseUserMuteStrategy): + """Keep user media muted after an end-call tool starts terminating a call.""" + + def __init__(self, is_call_ending: Callable[[], bool]): + super().__init__() + self._is_call_ending = is_call_ending + + async def process_frame(self, frame) -> bool: + await super().process_frame(frame) + return self._is_call_ending() + + class VisionCaptureProcessor(FrameProcessor): """Capture one requested video frame for auxiliary vision-model analysis.""" @@ -384,6 +409,13 @@ async def run_pipeline( "end_speaking": False, # 结束语音频已开始播报 "end_frame_queued": False, } + call_end_state = { + "ending": False, + "armed": False, + "speaking": False, + "frame_queued": False, + "reason": "completed", + } history: list[dict] = [] # 当前节点没有可调用转移工具(全是空条件)时,才启用文本兜底路由 FALLBACK_AFTER_TURNS = 2 @@ -418,6 +450,12 @@ async def run_pipeline( context, params=LLMUserAggregatorParams( vad_analyzer=SileroVADAnalyzer(), + user_mute_strategies=[ + FunctionCallUserMuteStrategy(), + CallEndingUserMuteStrategy( + lambda: bool(call_end_state["ending"]) + ), + ], user_turn_strategies=UserTurnStrategies( start=[ VADUserTurnStartStrategy(enable_interruptions=cfg.enableInterrupt), @@ -429,7 +467,9 @@ async def run_pipeline( ), ) assistant_aggregator = PassthroughLLMAssistantAggregator(context) - text_input = TextInputProcessor() + text_input = TextInputProcessor( + should_ignore_input=lambda: bool(call_end_state["ending"]) + ) vision_capture = VisionCaptureProcessor() vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg) vision_state: dict[str, str | None] = {"client_id": None} @@ -510,6 +550,99 @@ async def run_pipeline( if vision_enabled: llm.register_function(VISION_TOOL_NAME, fetch_user_image) + end_call_tools = [ + tool + for tool in cfg.tools + if cfg.type == "prompt" and tool.type == "end_call" + ] + end_call_schemas: list[FunctionSchema] = [] + worker_holder: dict = {} + + async def queue_call_end(reason: str) -> None: + if call_end_state["frame_queued"] or worker_holder.get("worker") is None: + return + call_end_state["frame_queued"] = True + logger.info(f"结束通话: reason={reason}") + await worker_holder["worker"].queue_frame( + OutputTransportMessageUrgentFrame( + message={"type": "call-ended", "reason": reason} + ) + ) + await worker_holder["worker"].queue_frame(EndFrame()) + + def make_end_call_handler(tool): + config = (tool.definition or {}).get("config") or {} + message_type = str(config.get("message_type") or "none") + custom_message = str(config.get("custom_message") or "").strip() + capture_reason = bool(config.get("capture_reason", True)) + + async def end_call(params: FunctionCallParams) -> None: + reason = str(params.arguments.get("reason") or "end_call_tool").strip() + call_end_state["ending"] = True + logger.info( + f"End Call Tool EXECUTED: {tool.function_name}, reason={reason}" + ) + await params.result_callback( + {"status": "success", "action": "ending_call"}, + properties=FunctionCallResultProperties(run_llm=False), + ) + + if message_type != "custom" or not custom_message: + await queue_call_end(reason) + return + + call_end_state["reason"] = reason + call_end_state["armed"] = True + turn_id = uuid4().hex + timestamp = time_now_iso8601() + for message in ( + { + "type": "assistant-text-start", + "turn_id": turn_id, + "timestamp": timestamp, + }, + { + "type": "assistant-text-delta", + "turn_id": turn_id, + "delta": custom_message, + }, + { + "type": "assistant-text-end", + "turn_id": turn_id, + "content": custom_message, + "interrupted": False, + }, + ): + await worker_holder["worker"].queue_frame( + OutputTransportMessageUrgentFrame(message=message) + ) + await worker_holder["worker"].queue_frame( + TTSSpeakFrame(custom_message, append_to_context=False) + ) + + properties = ( + { + "reason": { + "type": "string", + "description": "结束本次通话的简短原因。", + } + } + if capture_reason + else {} + ) + schema = FunctionSchema( + name=tool.function_name, + description=tool.description or "结束当前通话。", + properties=properties, + required=["reason"] if capture_reason else [], + ) + return schema, end_call + + for end_call_tool in end_call_tools: + schema, handler = make_end_call_handler(end_call_tool) + end_call_schemas.append(schema) + llm.register_function(end_call_tool.function_name, handler) + def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None: tools = list(schemas or []) if vision_enabled: @@ -519,34 +652,30 @@ async def run_pipeline( else: context.set_tools() - # 结束节点:等结束语「说完」(BotStoppedSpeakingFrame)再挂断,确保结束语的 - # 文字(走 data channel)与音频都已下发,避免前端只听到声音、看不到文字。 - worker_holder: dict = {} - + # Workflow 结束节点和 end_call 固定结束语都等到 BotStoppedSpeakingFrame + # 再挂断,确保文字(data channel)与音频完整送达。 class EndCallAfterSpeech(FrameProcessor): async def process_frame(self, frame, direction: FrameDirection): await super().process_frame(frame, direction) await self.push_frame(frame, direction) + armed = wf_state["end_armed"] or call_end_state["armed"] # 结束语文本生成完(end_armed)→ 其音频开始(end_speaking)→ 音频说完才挂断。 # 配对 started/stopped,避免被结束节点之前的话(如先答一句再转移)的 # stopped 事件提前触发,导致结束语被截断。 - if isinstance(frame, BotStartedSpeakingFrame) and wf_state["end_armed"]: - wf_state["end_speaking"] = True + if isinstance(frame, BotStartedSpeakingFrame) and armed: + if wf_state["end_armed"]: + wf_state["end_speaking"] = True + if call_end_state["armed"]: + call_end_state["speaking"] = True elif ( isinstance(frame, BotStoppedSpeakingFrame) - and wf_state["end_speaking"] - and not wf_state["end_frame_queued"] + and (wf_state["end_speaking"] or call_end_state["speaking"]) and worker_holder.get("worker") is not None ): - wf_state["end_frame_queued"] = True logger.info("结束语播报完毕,挂断通话") - # 先告知前端这是正常结束(而非连接异常),再优雅挂断 - await worker_holder["worker"].queue_frame( - OutputTransportMessageUrgentFrame( - message={"type": "call-ended", "reason": "completed"} - ) - ) - await worker_holder["worker"].queue_frame(EndFrame()) + wf_state["end_frame_queued"] = True + reason = str(call_end_state["reason"] or "completed") + await queue_call_end(reason) pipeline = Pipeline( [ @@ -690,8 +819,8 @@ async def run_pipeline( engine.edge_fn_name(edge), make_transition_handler(edge) ) apply_node(wf_state["current"]) # 设初始节点的提示与工具 - elif vision_enabled: - set_visible_tools([]) + else: + set_visible_tools(end_call_schemas) async def append_user_text_to_context(text: str, *, run_llm: bool) -> None: await worker.queue_frame(