wx44wx/ai-video-fullstack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add RuntimeTool model and enhance AssistantConfig for tool management

Browse Source 
- 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.
This commit is contained in:
Xin Wang
2026-07-10 14:32:10 +08:00
parent 5fca14865f
commit 4a57b290d3
5 changed files with 202 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/db/models.py
View File

@@ -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"

14
backend/models.py
View File

@@ -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 = {}

2
backend/routes/tools.py
View File

@@ -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

38
backend/services/config_resolver.py
View File

@@ -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,直接注入)

171
backend/services/pipecat/pipeline.py
View File

@@ -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(