From 72856bf3a7a9c6de3bbf0d275886acda9246d3a6 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Tue, 14 Jul 2026 09:36:28 +0800 Subject: [PATCH] Enhance workflow routing and agent configuration management - Introduce WorkflowLLMRouter for pre-response LLM routing, allowing agents to determine the appropriate function to call based on user input. - Implement UserTurnRoutingProcessor to manage user turns before reaching the LLM, ensuring proper routing and handling of user messages. - Refactor WorkflowBrain to integrate new routing logic and enhance agent stage configuration, including entry modes and resource management. - Update service factory to support dynamic LLM resource configuration based on workflow settings. - Add tests for new routing functionality and ensure proper handling of user messages in various scenarios. --- backend/routes/assistants.py | 11 +- backend/services/brains/base.py | 15 +- backend/services/brains/workflow_brain.py | 233 ++- backend/services/node_specs.py | 131 +- backend/services/pipecat/pipeline.py | 114 +- backend/services/pipecat/service_factory.py | 13 +- backend/services/workflow_engine.py | 74 +- backend/services/workflow_router.py | 129 ++ backend/tests/test_brains.py | 206 ++- backend/tests/test_pipeline_knowledge.py | 36 + backend/tests/test_workflow_router.py | 75 + backend/tests/test_workflow_v3.py | 212 ++- .../knowledge-retrieval-config-dialog.tsx | 168 ++ .../src/components/editor/section-card.tsx | 92 ++ .../src/components/pages/AssistantPage.tsx | 209 ++- .../src/components/turn-config-editor.tsx | 35 +- .../src/components/workflow/GenericNode.tsx | 24 +- .../components/workflow/WorkflowEditor.tsx | 1362 +++++++++++++---- frontend/src/components/workflow/specs.ts | 24 +- 19 files changed, 2611 insertions(+), 552 deletions(-) create mode 100644 backend/services/workflow_router.py create mode 100644 backend/tests/test_workflow_router.py create mode 100644 frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx create mode 100644 frontend/src/components/editor/section-card.tsx diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index 1f3c9ca..abc07c2 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -48,14 +48,23 @@ async def _validate_workflow_references( settings = graph.get("settings") or {} resource_expectations: dict[str, str] = {} for key, capability in ( + ("defaultLlmResourceId", "LLM"), ("defaultAsrResourceId", "ASR"), ("defaultTtsResourceId", "TTS"), ): if settings.get(key): resource_expectations[str(settings[key])] = capability - knowledge_ids: set[str] = set() + knowledge_ids: set[str] = ( + {str(settings["knowledgeBaseId"])} + if settings.get("knowledgeBaseId") + else set() + ) for node in graph.get("nodes") or []: data = node.get("data") or {} + if node.get("type") == "agent" and data.get("inheritGlobalConfig", True): + continue + if data.get("llmResourceId"): + resource_expectations[str(data["llmResourceId"])] = "LLM" if data.get("asrResourceId"): resource_expectations[str(data["asrResourceId"])] = "ASR" if data.get("ttsResourceId"): diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index ee07a90..bd738ff 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -55,7 +55,9 @@ class BrainRuntime: worker: Any = None context_aggregator: Any = None transport: Any = None - switch_services: Callable[[str | None, str | None], Awaitable[None]] | None = None + switch_services: ( + Callable[[str | None, str | None, str | None], Awaitable[None]] | None + ) = None set_knowledge_scope: Callable[[dict[str, Any]], None] | None = None set_input_enabled: Callable[[bool], None] | None = None flow_global_functions: list[Any] = field(default_factory=list) @@ -84,6 +86,15 @@ class BaseBrain: def record_user_message(self, content: str) -> None: """Observe a committed user message for brain-owned routing state.""" + async def on_user_turn_end(self, content: str) -> bool: + """Handle a complete user turn before the conversational LLM runs. + + Return True when the brain scheduled the next action itself and the + in-flight context frame must not reach the previous Agent's LLM. + """ + self.record_user_message(content) + return False + async def on_assistant_text_start(self, turn_id: str) -> None: """Observe the start of a generated assistant turn.""" @@ -114,6 +125,8 @@ class Brain(Protocol): def record_user_message(self, content: str) -> None: ... + async def on_user_turn_end(self, content: str) -> bool: ... + async def on_assistant_text_start(self, turn_id: str) -> None: ... async def on_assistant_text_end( diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 2aedef5..5676a7a 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -15,6 +15,7 @@ from pipecat.flows import ( NodeConfig, ) from pipecat.frames.frames import ( + LLMRunFrame, LLMUpdateSettingsFrame, OutputTransportMessageUrgentFrame, TTSSpeakFrame, @@ -22,18 +23,20 @@ from pipecat.frames.frames import ( from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor from pipecat.services.settings import LLMSettings +from pipecat.utils.time import time_now_iso8601 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.workflow_engine import WorkflowEngine +from services.workflow_router import STAY_ON_CURRENT_AGENT, WorkflowLLMRouter MAX_AUTOMATIC_HOPS = 50 AGENT_STAGE_INSTRUCTION = ( - "完成当前阶段任务。需要流转时必须调用对应的转移函数;" - "不要在调用转移函数后继续生成口头回复。" + "工作流路由已在用户一轮输入结束时完成。只执行当前阶段任务," + "不要自行解释、模拟或宣布节点切换。" ) @@ -58,6 +61,7 @@ class WorkflowBrain(BaseBrain): } self._runtime: BrainRuntime | None = None self._manager: FlowManager | None = None + self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow")) self._ended = False async def greeting(self, cfg: AssistantConfig) -> str: @@ -79,6 +83,7 @@ class WorkflowBrain(BaseBrain): self._store = DynamicVariableStore.from_config(cfg) self._tools = ToolExecutor(self._store) self._tool_by_id = {tool.id: tool for tool in cfg.tools} + self._router = WorkflowLLMRouter(cfg) self._manager = FlowManager( worker=runtime.worker, llm=runtime.llm, @@ -95,9 +100,13 @@ class WorkflowBrain(BaseBrain): self._store, include_default=True, ) - if not edge: + if not edge and self._engine.has_outgoing(self._engine.start_id): raise RuntimeError("Start 初始化后没有命中的表达式边或默认边") - node_config = await self._follow_edge(edge) + node_config = ( + await self._follow_edge(edge) + if edge + else self._passive_node_config(self._engine.start_id) + ) if self._manager is None: raise RuntimeError("Workflow FlowManager 尚未初始化") await self._manager.initialize(node_config) @@ -107,6 +116,76 @@ class WorkflowBrain(BaseBrain): if content and not self._ended: self._store.record("user", content) + async def on_user_turn_end(self, content: str) -> bool: + """Route a complete user turn before any Agent is allowed to reply.""" + if not content or self._ended: + return True + self.record_user_message(content) + manager = self._require_manager() + current = manager.current_node + if not current or self._engine.node_type(current) != "agent": + return True + + edge = self._engine.deterministic_edge( + current, + self._store, + include_default=False, + ) + outgoing = self._engine.outgoing(current) + llm_edges = [ + candidate + for candidate in outgoing + if self._engine.edge_mode(candidate) == "llm" + ] + default_edge = next( + ( + candidate + for candidate in outgoing + if self._engine.edge_mode(candidate) == "always" + ), + None, + ) + + if edge is None and llm_edges: + selected = await self._router_for_node(current).select_edge( + node_name=self._engine.name(current), + node_prompt=self._engine.prompt_for(current, self._store), + edges=llm_edges, + history=self._store.history, + variables={ + key: value + for key, value in self._store.values.items() + if not key.startswith("system__") + }, + edge_name=self._engine.edge_fn_name, + edge_description=self._engine.edge_description, + ) + if selected and selected != STAY_ON_CURRENT_AGENT: + edge = next( + ( + candidate + for candidate in llm_edges + if self._engine.edge_fn_name(candidate) == selected + ), + None, + ) + elif selected == STAY_ON_CURRENT_AGENT: + edge = default_edge + elif edge is None and not llm_edges: + edge = default_edge + + if edge and manager.current_node == current: + next_config = await self._follow_edge(edge) + await manager.set_node_from_config(next_config) + return True + + # The incoming LLMContextFrame is intentionally suppressed by the + # pipeline router. Queue prompt refresh + inference in this order so + # this user turn is answered with the current Agent's latest variables. + await self._refresh_agent_prompt(current) + await self._require_runtime().queue_frame(LLMRunFrame()) + return True + async def on_assistant_text_end( self, _turn_id: str, @@ -116,19 +195,6 @@ class WorkflowBrain(BaseBrain): if not content or interrupted or self._ended: return self._store.record("agent", content, completed_agent_turn=True) - manager = self._require_manager() - current = manager.current_node - if not current or self._engine.node_type(current) != "agent": - return - await self._refresh_agent_prompt(current) - edge = self._engine.deterministic_edge( - current, - self._store, - include_default=False, - ) - if edge and manager.current_node == current: - next_config = await self._follow_edge(edge) - await manager.set_node_from_config(next_config) async def _refresh_agent_prompt(self, node_id: str) -> None: runtime = self._require_runtime() @@ -145,61 +211,104 @@ class WorkflowBrain(BaseBrain): stage_prompt = self._engine.prompt_for(node_id, self._store) return f"{stage_prompt}\n\n[工作流执行规则]\n{AGENT_STAGE_INSTRUCTION}" + def _router_for_node(self, node_id: str) -> WorkflowLLMRouter: + stage = self._engine.agent_stage_config(node_id) + resource_id = stage.llm_resource_id + cfg = self._cfg + resource = cfg.workflow_model_resources.get(resource_id) if cfg else None + if not cfg or not resource: + return self._router + from services.pipecat.service_factory import config_with_resource + + return WorkflowLLMRouter(config_with_resource(cfg, resource)) + async def _apply_agent_stage(self, node_id: str) -> None: - data = self._engine.data(node_id) + stage = self._engine.agent_stage_config(node_id) await self._emit_node_active(node_id) if self._runtime and self._runtime.set_input_enabled: self._runtime.set_input_enabled(True) - asr_id = str( - data.get("asrResourceId") - or self._engine.settings.get("defaultAsrResourceId") - or "" - ) - tts_id = str( - data.get("ttsResourceId") - or self._engine.settings.get("defaultTtsResourceId") - or "" - ) runtime = self._require_runtime() if runtime.switch_services: - await runtime.switch_services(asr_id or None, tts_id or None) + await runtime.switch_services( + stage.llm_resource_id or None, + stage.asr_resource_id or None, + stage.tts_resource_id or None, + ) if runtime.set_knowledge_scope: runtime.set_knowledge_scope( { - "knowledge_base_id": data.get("knowledgeBaseId"), - "mode": data.get("knowledgeMode", "disabled"), - "top_n": int(data.get("knowledgeTopN") or 5), - "score_threshold": float(data.get("knowledgeScoreThreshold") or 0.0), + "knowledge_base_id": stage.knowledge_base_id, + "mode": stage.knowledge_mode, + "top_n": stage.knowledge_top_n, + "score_threshold": stage.knowledge_score_threshold, } ) def _agent_config(self, node_id: str) -> NodeConfig: data = self._engine.data(node_id) + entry_mode = str(data.get("entryMode") or "wait_user") + entry_speech = self._store.render(str(data.get("entrySpeech") or "")) strategy = ( ContextStrategy.RESET if data.get("contextPolicy") == "fresh" else ContextStrategy.APPEND ) + stage = self._engine.agent_stage_config(node_id) functions: list[FlowsFunctionSchema] = [] - for tool_id in data.get("toolIds") or []: + for tool_id in stage.tool_ids: tool = self._tool_by_id.get(str(tool_id)) if tool and tool.type == "http": functions.append(self._flow_tool(tool, node_id)) knowledge_function = self._knowledge_function(node_id) if knowledge_function: functions.append(knowledge_function) - for edge in self._engine.llm_edges(node_id): - functions.append(self._flow_edge(edge)) - return { + config: NodeConfig = { "name": node_id, "role_message": self._agent_role_message(node_id), - "task_messages": [], + "task_messages": ( + [{"role": "assistant", "content": entry_speech}] + if entry_mode == "fixed_speech" + else [] + ), "functions": functions, "context_strategy": ContextStrategyConfig(strategy=strategy), - "respond_immediately": True, + "respond_immediately": entry_mode == "generate", } + if entry_mode == "fixed_speech": + config["pre_actions"] = [ + { + "type": "workflow_fixed_speech", + "text": entry_speech, + "handler": self._play_fixed_speech, + } + ] + return config - def _terminal_config(self, node_id: str) -> NodeConfig: + async def _play_fixed_speech(self, action: dict, _flow_manager: FlowManager) -> None: + """Play and persist Agent entry speech without creating an LLM turn.""" + await self._queue_visible_speech(str(action.get("text") or "")) + + async def _queue_visible_speech(self, text: str) -> None: + """Show and persist fixed workflow speech before sending it to TTS.""" + content = text.strip() + if not content: + return + self._store.record("agent", content) + runtime = self._require_runtime() + await runtime.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "transcript", + "role": "assistant", + "content": content, + "timestamp": time_now_iso8601(), + } + ) + ) + await runtime.queue_frame(TTSSpeakFrame(content, append_to_context=False)) + + def _passive_node_config(self, node_id: str) -> NodeConfig: + """Keep a non-conversational terminal node active without ending the call.""" return { "name": node_id, "role_message": self._store.render(self._engine.global_prompt()), @@ -235,24 +344,13 @@ class WorkflowBrain(BaseBrain): properties=properties, required=required, handler=handler, - ) - - def _flow_edge(self, edge: dict) -> FlowsFunctionSchema: - async def handler(_args, _flow_manager): - return None, await self._follow_edge(edge) - - return FlowsFunctionSchema( - name=self._engine.edge_fn_name(edge), - description=self._engine.edge_description(edge), - properties={}, - required=[], - handler=handler, + cancel_on_interruption=True, ) def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None: - data = self._engine.data(node_id) - knowledge_id = str(data.get("knowledgeBaseId") or "") - if not knowledge_id or data.get("knowledgeMode") != "on_demand": + stage = self._engine.agent_stage_config(node_id) + knowledge_id = str(stage.knowledge_base_id or "") + if not knowledge_id or stage.knowledge_mode != "on_demand": return None cfg = self._cfg or AssistantConfig(type="workflow") knowledge = cfg.workflow_knowledge_bases.get(knowledge_id) @@ -270,8 +368,8 @@ class WorkflowBrain(BaseBrain): session, knowledge_id, query, - top_k=int(data.get("knowledgeTopN") or 5), - score_threshold=float(data.get("knowledgeScoreThreshold") or 0.0), + top_k=stage.knowledge_top_n, + score_threshold=stage.knowledge_score_threshold, ) return {"status": "ok", "results": results} except Exception as exc: # noqa: BLE001 - tool errors are returned to the LLM @@ -286,14 +384,13 @@ class WorkflowBrain(BaseBrain): }, required=["query"], handler=handler, + cancel_on_interruption=True, ) async def _follow_edge(self, edge: dict) -> NodeConfig: speech = self._engine.edge_transition_speech(edge) if speech: - await self._require_runtime().queue_frame( - TTSSpeakFrame(self._store.render(speech), append_to_context=False) - ) + await self._queue_visible_speech(self._store.render(speech)) return await self._resolve_path(str(edge.get("target") or "")) async def _resolve_path(self, node_id: str) -> NodeConfig: @@ -304,7 +401,7 @@ class WorkflowBrain(BaseBrain): return self._agent_config(node_id) if node_type == "end": await self._enter_end(node_id) - return self._terminal_config(node_id) + return self._passive_node_config(node_id) if node_type == "action": await self._enter_action(node_id) elif node_type == "handoff": @@ -313,6 +410,8 @@ class WorkflowBrain(BaseBrain): await self._emit_node_active(node_id) else: raise RuntimeError(f"工作流指向未知节点:{node_id}") + if not self._engine.has_outgoing(node_id): + return self._passive_node_config(node_id) edge = self._engine.deterministic_edge( node_id, self._store, @@ -322,9 +421,7 @@ class WorkflowBrain(BaseBrain): raise RuntimeError(f"自动节点 {node_id} 没有命中的表达式边或默认边") speech = self._engine.edge_transition_speech(edge) if speech: - await self._require_runtime().queue_frame( - TTSSpeakFrame(self._store.render(speech), append_to_context=False) - ) + await self._queue_visible_speech(self._store.render(speech)) node_id = str(edge.get("target") or "") raise RuntimeError("工作流连续自动跳转超过安全上限") @@ -366,9 +463,7 @@ class WorkflowBrain(BaseBrain): ) ) if message: - await self._require_runtime().queue_frame( - TTSSpeakFrame(message, append_to_context=False) - ) + await self._queue_visible_speech(message) self._store.values["system__handoff_status"] = "requested" async def _enter_end(self, node_id: str) -> None: @@ -389,12 +484,12 @@ class WorkflowBrain(BaseBrain): ) ) if message: - await runtime.queue_frame(TTSSpeakFrame(message, append_to_context=False)) + await self._queue_visible_speech(message) return runtime.call_end.begin("workflow_completed") if message: runtime.call_end.arm_after_speech() - await runtime.queue_frame(TTSSpeakFrame(message, append_to_context=False)) + await self._queue_visible_speech(message) else: await runtime.call_end.finish() diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 7c4b915..02ebace 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -10,6 +10,8 @@ from typing import Any SPEC_VERSION = "3" NODE_TYPES = {"start", "agent", "action", "handoff", "end"} EDGE_MODES = {"llm", "expression", "always"} +AGENT_ENTRY_MODES = {"wait_user", "generate", "fixed_speech"} +AUTOMATIC_NODE_TYPES = {"start", "action", "handoff"} EXPRESSION_OPERATORS = { "eq", "neq", @@ -34,7 +36,7 @@ NODE_SPECS: list[dict[str, Any]] = [ "constraints": { "minIncoming": 0, "maxIncoming": 0, - "minOutgoing": 1, + "minOutgoing": 0, "minInstances": 1, "maxInstances": 1, }, @@ -51,7 +53,7 @@ NODE_SPECS: list[dict[str, Any]] = [ "icon": "Bot", "accent": "sky", "addable": True, - "constraints": {"minIncoming": 1, "minOutgoing": 1}, + "constraints": {"minIncoming": 1, "minOutgoing": 0}, "fields": [ {"key": "name", "label": "节点名称", "type": "text", "default": "Agent"}, { @@ -84,7 +86,7 @@ NODE_SPECS: list[dict[str, Any]] = [ "icon": "PhoneForwarded", "accent": "lavender", "addable": True, - "constraints": {"minIncoming": 1, "minOutgoing": 1}, + "constraints": {"minIncoming": 1, "minOutgoing": 0}, "fields": [ {"key": "name", "label": "节点名称", "type": "text", "default": "Handoff"}, {"key": "target", "label": "转交目标", "type": "text", "default": ""}, @@ -132,13 +134,49 @@ def _edge_data_v3(edge: dict, source_type: str) -> dict: return data +def _normalize_agent_data(data: dict[str, Any]) -> None: + """Add v3 Agent defaults without changing existing node-level behavior.""" + data.setdefault("contextPolicy", "inherit") + data.setdefault("entryMode", "wait_user") + data.setdefault("entrySpeech", "") + if "inheritGlobalConfig" not in data: + has_node_overrides = any( + ( + data.get("llmResourceId"), + data.get("asrResourceId"), + data.get("ttsResourceId"), + data.get("knowledgeBaseId"), + data.get("toolIds"), + ) + ) + data["inheritGlobalConfig"] = not has_node_overrides + + +def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None: + settings.setdefault("globalPrompt", global_prompt) + settings.setdefault("defaultLlmResourceId", "") + settings.setdefault("defaultAsrResourceId", "") + settings.setdefault("defaultTtsResourceId", "") + settings.setdefault("toolIds", []) + settings.setdefault("knowledgeBaseId", "") + settings.setdefault("knowledgeMode", "automatic") + settings.setdefault("knowledgeTopN", 5) + settings.setdefault("knowledgeScoreThreshold", 0.0) + + def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: """Return a deep-copied v3 graph; preserve v3 IDs and migrate v2 semantics.""" source = deepcopy(graph or {}) if str(source.get("specVersion") or "") == SPEC_VERSION: - source.setdefault("settings", {}) + settings = source.setdefault("settings", {}) + _normalize_settings(settings) source.setdefault("nodes", []) source.setdefault("edges", []) + for node in source["nodes"]: + if node.get("type") != "agent": + continue + data = node.setdefault("data", {}) + _normalize_agent_data(data) return source nodes = source.get("nodes") or [] @@ -171,9 +209,7 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: data["message"] = data.pop("message", data.pop("prompt", "")) data.setdefault("scope", "session") elif new_type == "agent": - data.setdefault("contextPolicy", "inherit") - data.setdefault("toolIds", []) - data.setdefault("knowledgeMode", "disabled") + _normalize_agent_data(data) elif new_type == "start": prompt = str(data.pop("prompt", "") or "").strip() if prompt: @@ -207,8 +243,9 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: "name": "迁移的开场 Agent", "prompt": prompt, "contextPolicy": "inherit", - "toolIds": [], - "knowledgeMode": "disabled", + "inheritGlobalConfig": True, + "entryMode": "wait_user", + "entrySpeech": "", }, } ) @@ -228,9 +265,7 @@ def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: ) settings = deepcopy(source.get("settings") or {}) - settings.setdefault("globalPrompt", global_prompt) - settings.setdefault("defaultAsrResourceId", "") - settings.setdefault("defaultTtsResourceId", "") + _normalize_settings(settings, global_prompt=global_prompt) return { "specVersion": 3, "settings": settings, @@ -281,11 +316,18 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: node_by_id[node_id] = node counts[node_type] += 1 + if node_type == "agent": + data = node.get("data") or {} + entry_mode = data.get("entryMode", "wait_user") + if entry_mode not in AGENT_ENTRY_MODES: + errors.append(f"Agent 节点 {node_id} 的进入模式无效:{entry_mode}") + elif entry_mode == "fixed_speech" and not str( + data.get("entrySpeech") or "" + ).strip(): + errors.append(f"Agent 节点 {node_id} 的固定进入语不能为空") + if counts["start"] != 1: errors.append("工作流必须有且仅有一个 Start 节点") - if counts["end"] < 1: - errors.append("工作流至少需要一个 End 节点") - incoming: dict[str, int] = defaultdict(int) outgoing: dict[str, int] = defaultdict(int) adj: dict[str, list[str]] = defaultdict(list) @@ -313,7 +355,8 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: if mode == "llm" and not str(data.get("condition") or "").strip(): errors.append(f"LLM 判断边缺少自然语言条件:{edge_id}") if mode == "expression": - errors.extend(f"边 {edge_id}:{item}" for item in _validate_expression(data.get("expression"))) + expression_errors = _validate_expression(data.get("expression")) + errors.extend(f"边 {edge_id}:{item}" for item in expression_errors) try: priority = int(data.get("priority", 10)) except (TypeError, ValueError): @@ -329,7 +372,9 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: incoming[target_id] += 1 outgoing[source_id] += 1 adj[source_id].append(target_id) - if node_by_id[source_id].get("type") != "agent" and node_by_id[target_id].get("type") != "agent": + source_is_automatic = node_by_id[source_id].get("type") != "agent" + target_is_automatic = node_by_id[target_id].get("type") != "agent" + if source_is_automatic and target_is_automatic: auto_adj[source_id].append(target_id) for node_id, node in node_by_id.items(): @@ -347,10 +392,18 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: errors.append(f"节点 {node_id} 的{label}不能少于 {lo}") if hi is not None and actual > hi: errors.append(f"节点 {node_id} 的{label}不能多于 {hi}") - if node.get("type") in {"start", "action", "handoff"} and always_counts[node_id] != 1: - errors.append(f"自动节点 {node_id} 必须有且仅有一条默认边") + node_type = node.get("type") + if ( + node_type in AUTOMATIC_NODE_TYPES + and outgoing[node_id] > 0 + and always_counts[node_id] != 1 + ): + errors.append(f"自动节点 {node_id} 存在出边时必须有且仅有一条默认边") - start_id = next((nid for nid, n in node_by_id.items() if n.get("type") == "start"), None) + start_id = next( + (node_id for node_id, node in node_by_id.items() if node.get("type") == "start"), + None, + ) if start_id: reached = {start_id} queue = deque([start_id]) @@ -380,7 +433,12 @@ def validate_graph(graph: dict[str, Any]) -> list[str]: visited.add(node_id) return False - if any(visit(node_id) for node_id, node in node_by_id.items() if node.get("type") != "agent"): + automatic_node_ids = ( + node_id + for node_id, node in node_by_id.items() + if node.get("type") != "agent" + ) + if any(visit(node_id) for node_id in automatic_node_ids): errors.append("Start/Action/Handoff/End 之间不能形成无等待循环") return list(dict.fromkeys(errors)) @@ -392,22 +450,35 @@ def graph_references(graph: dict[str, Any]) -> dict[str, set[str]]: resources = { str(value) for value in ( + settings.get("defaultLlmResourceId"), settings.get("defaultAsrResourceId"), settings.get("defaultTtsResourceId"), ) if value } - tools: set[str] = set() - knowledge: set[str] = set() + tools: set[str] = {str(tool_id) for tool_id in settings.get("toolIds") or []} + knowledge: set[str] = ( + {str(settings["knowledgeBaseId"])} + if settings.get("knowledgeBaseId") + else set() + ) for node in normalized.get("nodes") or []: data = node.get("data") or {} - for resource_id in (data.get("asrResourceId"), data.get("ttsResourceId")): - if resource_id: - resources.add(str(resource_id)) - for tool_id in data.get("toolIds") or []: - tools.add(str(tool_id)) + inherits_global = ( + node.get("type") == "agent" and data.get("inheritGlobalConfig", True) + ) + if not inherits_global: + for resource_id in ( + data.get("llmResourceId"), + data.get("asrResourceId"), + data.get("ttsResourceId"), + ): + if resource_id: + resources.add(str(resource_id)) + for tool_id in data.get("toolIds") or []: + tools.add(str(tool_id)) + if data.get("knowledgeBaseId"): + knowledge.add(str(data["knowledgeBaseId"])) if data.get("toolId"): tools.add(str(data["toolId"])) - if data.get("knowledgeBaseId"): - knowledge.add(str(data["knowledgeBaseId"])) return {"model_resources": resources, "tools": tools, "knowledge_bases": knowledge} diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index b62eeca..8896465 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -24,6 +24,7 @@ from services.pipecat.call_lifecycle import ( ) from services.pipecat.service_factory import ( config_with_resource, + create_llm, create_realtime_service, create_stt, create_tts, @@ -51,6 +52,7 @@ from pipecat.frames.frames import ( UserImageRequestFrame, ) from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.llm_switcher import LLMSwitcher from pipecat.pipeline.service_switcher import ServiceSwitcher from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext @@ -487,6 +489,49 @@ class KnowledgeRetrievalProcessor(FrameProcessor): await self.push_frame(frame, direction) +class UserTurnRoutingProcessor(FrameProcessor): + """Give a brain first right of refusal before a new user turn reaches the LLM.""" + + def __init__(self, brain: Brain): + super().__init__() + self._brain = brain + self._last_user_message: dict | None = None + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction != FrameDirection.DOWNSTREAM or not isinstance( + frame, LLMContextFrame + ): + await self.push_frame(frame, direction) + return + + user_message = next( + ( + 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() + ), + None, + ) + if user_message is None: + await self.push_frame(frame, direction) + return + + if user_message is self._last_user_message: + # Programmatic LLMRunFrame after a node transition reuses the same + # user message. It is a response run, not another routing event. + await self.push_frame(frame, direction) + return + self._last_user_message = user_message + + content = str(user_message.get("content") or "").strip() + handled = await self._brain.on_user_turn_end(content) + if not handled: + await self.push_frame(frame, direction) + + class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): """聚合 LLM 回复进上下文,同时继续把回复帧交给下游 TTS。""" @@ -589,6 +634,29 @@ def _workflow_service_switcher( return ServiceSwitcher(services=services), services_by_id, primary +def _workflow_llm_switcher(cfg: AssistantConfig, base_service): + """Build an LLM switcher for the global model and Agent overrides.""" + settings = cfg.graph.get("settings") or {} + default_id = str(settings.get("defaultLlmResourceId") or "") + services_by_id = {} + for resource_id, resource in cfg.workflow_model_resources.items(): + if resource.capability != "LLM": + continue + services_by_id[resource_id] = ( + base_service + if resource_id == default_id + else create_llm(config_with_resource(cfg, resource)) + ) + primary = services_by_id.get(default_id, base_service) + services = [primary] + services.extend( + service for service in services_by_id.values() if service is not primary + ) + if base_service is not primary: + services.append(base_service) + return LLMSwitcher(llms=services), services_by_id, primary + + async def run_pipeline( transport, cfg: AssistantConfig, @@ -630,6 +698,9 @@ async def run_pipeline( return graph_settings = cfg.graph.get("settings") or {} + default_llm_resource = cfg.workflow_model_resources.get( + str(graph_settings.get("defaultLlmResourceId") or "") + ) default_asr_resource = cfg.workflow_model_resources.get( str(graph_settings.get("defaultAsrResourceId") or "") ) @@ -713,7 +784,16 @@ async def run_pipeline( ) input_state = {"enabled": True} # LLM 槽由大脑提供:本地模型或 Dify/FastGPT 外部托管适配器。 - llm = brain.build_llm(cfg, context) + llm = brain.build_llm( + config_with_resource(cfg, default_llm_resource) + if cfg.type == "workflow" and default_llm_resource + else cfg, + context, + ) + llm_services: dict[str, FrameProcessor] = {} + current_llm_service = llm + if cfg.type == "workflow": + llm, llm_services, current_llm_service = _workflow_llm_switcher(cfg, llm) user_aggregator = LLMUserAggregator( context, params=LLMUserAggregatorParams( @@ -730,6 +810,7 @@ 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"] @@ -880,6 +961,7 @@ async def run_pipeline( properties=vision_schema.properties, required=vision_schema.required, handler=flow_fetch_user_image, + cancel_on_interruption=True, ) ) @@ -913,6 +995,7 @@ async def run_pipeline( text_input, stt_processor, user_aggregator, + user_turn_router, knowledge_retrieval, llm, # Aggregate the streamed LLM text before TTS. On interruption, @@ -934,24 +1017,42 @@ async def run_pipeline( enable_rtvi=False, ) worker_holder["worker"] = worker - default_voice_services = dict(current_voice_services) + default_workflow_services = { + "llm": current_llm_service, + **current_voice_services, + } async def switch_workflow_services( + llm_resource_id: str | None, asr_resource_id: str | None, tts_resource_id: str | None, ) -> None: + nonlocal current_llm_service requested = ( + ("llm", llm_services, llm_resource_id), ("asr", stt_services, asr_resource_id), ("tts", tts_services, tts_resource_id), ) for kind, services, resource_id in requested: - target = services.get(resource_id) if resource_id else default_voice_services[kind] + target = ( + services.get(resource_id) + if resource_id + else default_workflow_services[kind] + ) if target is None: raise ValueError(f"Workflow {kind.upper()} 资源未加载:{resource_id}") - if current_voice_services[kind] is target: + current = ( + current_llm_service + if kind == "llm" + else current_voice_services[kind] + ) + if current is target: continue await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) - current_voice_services[kind] = target + if kind == "llm": + current_llm_service = target + else: + current_voice_services[kind] = target await worker.queue_frame( OutputTransportMessageUrgentFrame( message={ @@ -1020,8 +1121,6 @@ async def run_pipeline( @user_aggregator.event_handler("on_user_turn_stopped") async def on_user_turn_stopped(_aggregator, _strategy, message): - if message.content: - brain.record_user_message(message.content) await queue_transcript("user", message.content, message.timestamp) @assistant_aggregator.event_handler("on_assistant_text_start") @@ -1066,7 +1165,6 @@ async def run_pipeline( @text_input.event_handler("on_text_input") async def on_text_input(_processor, text): pending_text_inputs.append(text) - brain.record_user_message(text) # 前端显示不依赖 interruption 后续事件,必须在打断前先排入发送队列。 await queue_transcript("user", text, time_now_iso8601()) diff --git a/backend/services/pipecat/service_factory.py b/backend/services/pipecat/service_factory.py index 907bc68..f575be7 100644 --- a/backend/services/pipecat/service_factory.py +++ b/backend/services/pipecat/service_factory.py @@ -29,11 +29,18 @@ TTS_STOP_FRAME_TIMEOUT_S = 1.0 def config_with_resource( cfg: AssistantConfig, resource: RuntimeModelResource ) -> AssistantConfig: - """Return a call-local config view for one workflow ASR/TTS resource.""" + """Return a call-local config view for one workflow model resource.""" result = cfg.model_copy(deep=True) values = resource.values or {} secrets = resource.secrets or {} - if resource.capability == "ASR": + if resource.capability == "LLM": + result.model = str(values.get("modelId") or "") + result.llm_interface_type = resource.interface_type + result.llm_values = values + result.llm_secrets = secrets + result.llm_api_key = str(secrets.get("apiKey") or "") + result.llm_base_url = str(values.get("apiUrl") or "") + elif resource.capability == "ASR": result.asr = str(values.get("modelId") or "") result.stt_language = str(values.get("language") or "") result.stt_interface_type = resource.interface_type @@ -51,7 +58,7 @@ def config_with_resource( result.tts_api_key = str(secrets.get("apiKey") or "") result.tts_base_url = str(values.get("apiUrl") or "") else: - raise ValueError(f"工作流语音资源能力无效:{resource.capability}") + raise ValueError(f"工作流模型资源能力无效:{resource.capability}") return result diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 94b94f4..98078e9 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -3,12 +3,28 @@ from __future__ import annotations import re +from dataclasses import dataclass from typing import Any from services.node_specs import normalize_graph from services.runtime_variables import DynamicVariableStore +@dataclass(frozen=True) +class AgentStageConfig: + """The complete assistant configuration active inside one Agent node.""" + + inherits_global: bool + llm_resource_id: str + asr_resource_id: str + tts_resource_id: str + tool_ids: tuple[str, ...] + knowledge_base_id: str | None + knowledge_mode: str + knowledge_top_n: int + knowledge_score_threshold: float + + class WorkflowEngine: def __init__(self, graph: dict[str, Any]): self.graph = normalize_graph(graph) @@ -20,7 +36,11 @@ class WorkflowEngine: } self.edges: list[dict] = list(self.graph.get("edges") or []) self.start_id = next( - (node_id for node_id, node in self.nodes.items() if node.get("type") == "start"), + ( + node_id + for node_id, node in self.nodes.items() + if node.get("type") == "start" + ), None, ) @@ -38,7 +58,13 @@ class WorkflowEngine: def outgoing(self, node_id: str | None) -> list[dict]: result = [edge for edge in self.edges if edge.get("source") == node_id] - return sorted(result, key=lambda edge: int((edge.get("data") or {}).get("priority", 10))) + return sorted( + result, + key=lambda edge: int((edge.get("data") or {}).get("priority", 10)), + ) + + def has_outgoing(self, node_id: str | None) -> bool: + return any(edge.get("source") == node_id for edge in self.edges) def edge_mode(self, edge: dict) -> str: return str((edge.get("data") or {}).get("mode") or "always") @@ -60,15 +86,49 @@ class WorkflowEngine: if not edge: return "" data = edge.get("data") or {} - return str(data.get("transitionSpeech") or data.get("transition_speech") or "") + return str( + data.get("transitionSpeech") or data.get("transition_speech") or "" + ) def global_prompt(self) -> str: return str(self.settings.get("globalPrompt") or "").strip() + def inherits_global_config(self, node_id: str) -> bool: + """Return the Agent's explicit configuration scope, defaulting to global.""" + return bool(self.data(node_id).get("inheritGlobalConfig", True)) + + def agent_stage_config(self, node_id: str) -> AgentStageConfig: + """Resolve either Workflow defaults or one Agent's complete override.""" + data = self.data(node_id) + inherits_global = self.inherits_global_config(node_id) + source = self.settings if inherits_global else data + llm_key = "defaultLlmResourceId" if inherits_global else "llmResourceId" + asr_key = "defaultAsrResourceId" if inherits_global else "asrResourceId" + tts_key = "defaultTtsResourceId" if inherits_global else "ttsResourceId" + knowledge_base_id = str(source.get("knowledgeBaseId") or "") + return AgentStageConfig( + inherits_global=inherits_global, + llm_resource_id=str(source.get(llm_key) or ""), + asr_resource_id=str(source.get(asr_key) or ""), + tts_resource_id=str(source.get(tts_key) or ""), + tool_ids=tuple(str(tool_id) for tool_id in source.get("toolIds") or []), + knowledge_base_id=knowledge_base_id or None, + knowledge_mode=( + str(source.get("knowledgeMode") or "automatic") + if knowledge_base_id + else "disabled" + ), + knowledge_top_n=int(source.get("knowledgeTopN") or 5), + knowledge_score_threshold=float( + source.get("knowledgeScoreThreshold") or 0.0 + ), + ) + def prompt_for(self, node_id: str, store: DynamicVariableStore) -> str: + """Build the Agent system prompt according to its inheritance setting.""" prompt = store.render(str(self.data(node_id).get("prompt") or "").strip()) sections = [f"[当前阶段:{self.name(node_id)}]"] - if self.global_prompt(): + if self.inherits_global_config(node_id) and self.global_prompt(): sections.append(f"[全局规则]\n{store.render(self.global_prompt())}") if prompt: sections.append(f"[当前阶段任务]\n{prompt}") @@ -111,7 +171,11 @@ class WorkflowEngine: results.append(matched) if not results: return False - return all(results) if expression.get("combinator", "and") == "and" else any(results) + return ( + all(results) + if expression.get("combinator", "and") == "and" + else any(results) + ) def deterministic_edge( self, diff --git a/backend/services/workflow_router.py b/backend/services/workflow_router.py new file mode 100644 index 0000000..8170a3f --- /dev/null +++ b/backend/services/workflow_router.py @@ -0,0 +1,129 @@ +"""Pre-response LLM routing for Workflow Agent edges. + +The router deliberately uses a separate, short completion. Its only output is +a required function choice, so the current Agent cannot speak before the graph +has decided whether the user turn belongs to another node. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import Any + +from loguru import logger +from models import AssistantConfig +from openai import AsyncOpenAI + + +STAY_ON_CURRENT_AGENT = "workflow_stay_on_current_agent" +MAX_ROUTING_HISTORY_ENTRIES = 20 + + +class WorkflowLLMRouter: + """Select one LLM edge before the conversational LLM is allowed to reply.""" + + def __init__(self, cfg: AssistantConfig): + self._cfg = cfg + + async def select_edge( + self, + *, + node_name: str, + node_prompt: str, + edges: list[dict[str, Any]], + history: list[dict[str, str]], + variables: dict[str, Any], + edge_name: Callable[[dict[str, Any]], str], + edge_description: Callable[[dict[str, Any]], str], + ) -> str | None: + """Return an edge function name, STAY, or None when routing failed.""" + if not edges: + return STAY_ON_CURRENT_AGENT + + names = {edge_name(edge) for edge in edges} + stay_name = STAY_ON_CURRENT_AGENT + while stay_name in names: + stay_name = f"_{stay_name}" + + tools = [ + { + "type": "function", + "function": { + "name": edge_name(edge), + "description": edge_description(edge), + "parameters": {"type": "object", "properties": {}}, + }, + } + for edge in edges + ] + tools.append( + { + "type": "function", + "function": { + "name": stay_name, + "description": "所有转移条件都不满足,继续由当前 Agent 处理用户消息。", + "parameters": {"type": "object", "properties": {}}, + }, + } + ) + + ordered_conditions = "\n".join( + f"{index + 1}. {edge_description(edge)}" + for index, edge in enumerate(edges) + ) + router_prompt = ( + "你是工作流路由器,不是对话助手。收到一轮完整用户输入后," + "必须且只能调用一个提供的函数,禁止输出任何口头回复。\n" + "按给出的顺序判断转移条件;选择第一个明确满足的转移函数。" + "如果没有条件满足,调用留在当前 Agent 的函数。\n\n" + f"当前节点:{node_name}\n" + f"当前节点任务:{node_prompt or '未配置'}\n" + f"转移条件:\n{ordered_conditions}" + ) + recent_history = history[-MAX_ROUTING_HISTORY_ENTRIES:] + routing_input = json.dumps( + { + "conversation": recent_history, + "session_variables": variables, + }, + ensure_ascii=False, + separators=(",", ":"), + ) + extra_body = self._cfg.llm_values.get("extraBody") + request_extra = ( + {"extra_body": extra_body} if isinstance(extra_body, dict) else {} + ) + client = AsyncOpenAI( + api_key=self._cfg.llm_api_key, + base_url=self._cfg.llm_base_url, + timeout=15.0, + ) + try: + response = await client.chat.completions.create( + model=self._cfg.model, + messages=[ + {"role": "system", "content": router_prompt}, + {"role": "user", "content": routing_input}, + ], + tools=tools, + tool_choice="required", + temperature=0, + **request_extra, + ) + tool_calls = response.choices[0].message.tool_calls or [] + if not tool_calls: + logger.warning("Workflow 路由 LLM 未返回函数调用,留在当前 Agent") + return STAY_ON_CURRENT_AGENT + selected = str(tool_calls[0].function.name or "") + if selected == stay_name: + return STAY_ON_CURRENT_AGENT + if selected not in names: + logger.warning(f"Workflow 路由 LLM 返回未知函数:{selected}") + return STAY_ON_CURRENT_AGENT + return selected + except Exception as exc: # noqa: BLE001 - routing failure must not end the call + logger.warning(f"Workflow LLM 边判断失败,留在当前 Agent:{exc}") + return None + finally: + await client.close() diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 3ac44fd..b2e2e43 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -9,7 +9,10 @@ from pipecat.frames.frames import ( LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMRunFrame, LLMTextFrame, + OutputTransportMessageUrgentFrame, + TTSSpeakFrame, ) from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection @@ -388,10 +391,108 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_nodes_without_outgoing_edges_remain_active(self): + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + runtime = BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ) + + class FakeManager: + def __init__(self, current_node=None): + self.current_node = current_node + + async def initialize(self, config): + self.current_node = config["name"] + + start_brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [{"id": "start", "type": "start", "data": {}}], + "edges": [], + } + ) + start_brain._runtime = runtime + start_brain._manager = FakeManager() + await start_brain.on_connected() + self.assertEqual(start_brain._manager.current_node, "start") + + agent_brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {"globalPrompt": "全局规则"}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "agent", + "type": "agent", + "data": {"prompt": "持续回答"}, + }, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + } + ], + } + ) + agent_brain._runtime = runtime + agent_brain._manager = FakeManager("agent") + queued.clear() + handled = await agent_brain.on_user_turn_end("请继续回答") + self.assertTrue(handled) + self.assertEqual(agent_brain._manager.current_node, "agent") + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued)) + + handoff_brain = WorkflowBrain( + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "handoff", + "type": "handoff", + "data": {"targetType": "human"}, + }, + ], + "edges": [], + } + ) + handoff_brain._runtime = runtime + handoff_config = await handoff_brain._resolve_path("handoff") + self.assertEqual(handoff_config["name"], "handoff") + self.assertTrue( + any( + isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "handoff-requested" + for frame in queued + ) + ) + async def test_transition_and_end_are_owned_by_workflow_brain(self): graph = { "specVersion": 3, - "settings": {"globalPrompt": "全局规则"}, + "settings": { + "globalPrompt": "全局规则", + "defaultLlmResourceId": "llm_global", + "defaultAsrResourceId": "asr_global", + "defaultTtsResourceId": "tts_global", + "knowledgeBaseId": "kb_global", + "knowledgeMode": "automatic", + }, "nodes": [ { "id": "start", @@ -428,6 +529,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): "mode": "llm", "priority": 10, "condition": "需求已收集", + "transitionSpeech": "正在为你结束流程", }, } ], @@ -447,6 +549,8 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): llm = FakeLLM() context = LLMContext(messages=[]) queued = [] + service_switches = [] + knowledge_scopes = [] call_end = FakeCallEnd() class FakeWorker: @@ -478,6 +582,9 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def queue_frame(frame): queued.append(frame) + async def switch_services(llm_id, asr_id, tts_id): + service_switches.append((llm_id, asr_id, tts_id)) + runtime = BrainRuntime( context=context, llm=llm, @@ -487,29 +594,112 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): call_end=call_end, worker=worker, context_aggregator=pair, + switch_services=switch_services, + set_knowledge_scope=knowledge_scopes.append, ) await brain.setup(cfg, runtime) await brain.on_connected() self.assertEqual(brain._manager.current_node, "agent") + self.assertEqual( + service_switches, + [("llm_global", "asr_global", "tts_global")], + ) + self.assertEqual(knowledge_scopes[-1]["knowledge_base_id"], "kb_global") + + brain._engine.data("agent").update( + { + "inheritGlobalConfig": False, + "llmResourceId": "llm_agent", + "asrResourceId": "asr_agent", + "ttsResourceId": "tts_agent", + "knowledgeBaseId": "kb_agent", + "knowledgeMode": "on_demand", + } + ) + await brain._apply_agent_stage("agent") + self.assertEqual( + service_switches[-1], + ("llm_agent", "asr_agent", "tts_agent"), + ) + self.assertEqual(knowledge_scopes[-1]["knowledge_base_id"], "kb_agent") agent_config = brain._agent_config("agent") self.assertIn("王先生", agent_config["role_message"]) - self.assertIn("完成当前阶段任务", agent_config["role_message"]) + self.assertIn("工作流路由已在用户一轮输入结束时完成", agent_config["role_message"]) self.assertEqual(agent_config["task_messages"], []) + self.assertFalse(agent_config["respond_immediately"]) + self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) self.assertEqual( agent_config["context_strategy"].strategy.value, "reset", ) - edge_function = next( - function - for function in brain._agent_config("agent")["functions"] - if function.name == "goto_finish" + brain._engine.data("agent")["entryMode"] = "generate" + generate_config = brain._agent_config("agent") + self.assertTrue(generate_config["respond_immediately"]) + worker.frames.clear() + await brain._manager.set_node_from_config(generate_config) + self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + + brain._engine.data("agent").update( + {"entryMode": "fixed_speech", "entrySpeech": "您好,{{user_name}}"} ) - _, terminal = await edge_function.handler({}, brain._manager) - self.assertEqual(terminal["name"], "end") + fixed_config = brain._agent_config("agent") + self.assertFalse(fixed_config["respond_immediately"]) + self.assertEqual( + fixed_config["pre_actions"][0]["type"], + "workflow_fixed_speech", + ) + self.assertEqual(fixed_config["pre_actions"][0]["text"], "您好,王先生") + self.assertEqual( + fixed_config["task_messages"], + [{"role": "assistant", "content": "您好,王先生"}], + ) + worker.frames.clear() + queued.clear() + await brain._manager.set_node_from_config(fixed_config) + self.assertTrue(any(isinstance(frame, TTSSpeakFrame) for frame in queued)) + self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) + + self.assertFalse( + any( + function.name == "goto_finish" + for function in brain._agent_config("agent")["functions"] + ) + ) + await brain.on_assistant_text_end("old-turn", "需求已收集", False) + self.assertEqual(brain._manager.current_node, "agent") + + class FakeRouter: + async def select_edge(self, **_kwargs): + return "goto_finish" + + brain._router = FakeRouter() + handled = await brain.on_user_turn_end("我的需求已经说完了") + self.assertTrue(handled) + self.assertEqual(brain._manager.current_node, "end") + self.assertIn("我的需求已经说完了", brain._store.values["system__conversation_history"]) self.assertTrue(call_end.ending) self.assertTrue(call_end.armed) self.assertTrue(any(getattr(frame, "text", "") == "感谢来电" for frame in queued)) + assistant_transcripts = [ + frame.message.get("content") + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "transcript" + and frame.message.get("role") == "assistant" + ] + self.assertEqual( + assistant_transcripts, + ["您好,王先生", "正在为你结束流程", "感谢来电"], + ) + self.assertIn( + "正在为你结束流程", + brain._store.values["system__conversation_history"], + ) + self.assertIn( + "感谢来电", + brain._store.values["system__conversation_history"], + ) if __name__ == "__main__": diff --git a/backend/tests/test_pipeline_knowledge.py b/backend/tests/test_pipeline_knowledge.py index 9b8f328..0e80113 100644 --- a/backend/tests/test_pipeline_knowledge.py +++ b/backend/tests/test_pipeline_knowledge.py @@ -1,9 +1,13 @@ import unittest from models import AssistantConfig +from pipecat.frames.frames import LLMContextFrame +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameDirection from services.pipecat.pipeline import ( KNOWLEDGE_CONTEXT_MARKER, KnowledgeRetrievalProcessor, + UserTurnRoutingProcessor, _knowledge_tool_description, ) @@ -57,5 +61,37 @@ class KnowledgeToolDescriptionTest(unittest.TestCase): self.assertFalse(any(message["role"] == "developer" for message in messages)) +class UserTurnRoutingProcessorTest(unittest.IsolatedAsyncioTestCase): + async def test_routes_each_user_message_once_before_response_run(self): + class FakeBrain: + def __init__(self): + self.turns = [] + + async def on_user_turn_end(self, content): + self.turns.append(content) + return True + + brain = FakeBrain() + processor = UserTurnRoutingProcessor(brain) + forwarded = [] + + async def push_frame(frame, direction): + forwarded.append((frame, direction)) + + processor.push_frame = push_frame + context = LLMContext(messages=[{"role": "user", "content": "我叫李白"}]) + frame = LLMContextFrame(context) + + await processor.process_frame(frame, FrameDirection.DOWNSTREAM) + self.assertEqual(brain.turns, ["我叫李白"]) + self.assertEqual(forwarded, []) + + # A queued LLMRunFrame after the transition uses the same context. It + # must reach the target Agent without invoking routing a second time. + await processor.process_frame(frame, FrameDirection.DOWNSTREAM) + self.assertEqual(brain.turns, ["我叫李白"]) + self.assertEqual(forwarded, [(frame, FrameDirection.DOWNSTREAM)]) + + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_workflow_router.py b/backend/tests/test_workflow_router.py new file mode 100644 index 0000000..67d7766 --- /dev/null +++ b/backend/tests/test_workflow_router.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from models import AssistantConfig +from services.workflow_router import WorkflowLLMRouter + + +class WorkflowLLMRouterTest(unittest.IsolatedAsyncioTestCase): + async def test_uses_required_tool_choice_without_developer_messages(self): + requests = [] + + class FakeCompletions: + async def create(self, **kwargs): + requests.append(kwargs) + return SimpleNamespace( + choices=[ + SimpleNamespace( + message=SimpleNamespace( + tool_calls=[ + SimpleNamespace( + function=SimpleNamespace(name="goto_age", arguments="{}") + ) + ] + ) + ) + ] + ) + + class FakeClient: + def __init__(self, **_kwargs): + self.chat = SimpleNamespace(completions=FakeCompletions()) + self.closed = False + + async def close(self): + self.closed = True + + cfg = AssistantConfig( + type="workflow", + model="deepseek-chat", + llm_api_key="secret", + llm_base_url="https://llm.test/v1", + ) + router = WorkflowLLMRouter(cfg) + edges = [ + { + "id": "age", + "data": {"condition": "用户已经回答姓名", "priority": 10}, + } + ] + + with patch("services.workflow_router.AsyncOpenAI", FakeClient): + selected = await router.select_edge( + node_name="询问姓名", + node_prompt="询问用户姓名", + edges=edges, + history=[{"role": "user", "message": "我叫李白"}], + variables={"customer_type": "new"}, + edge_name=lambda _edge: "goto_age", + edge_description=lambda _edge: "用户已经回答姓名", + ) + + self.assertEqual(selected, "goto_age") + self.assertEqual(requests[0]["tool_choice"], "required") + self.assertEqual( + [message["role"] for message in requests[0]["messages"]], + ["system", "user"], + ) + self.assertNotIn("developer", str(requests[0]["messages"])) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index cb6dfac..1cfa553 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -4,7 +4,7 @@ import unittest from models import AssistantConfig, RuntimeModelResource from services.pipecat.service_factory import config_with_resource -from services.node_specs import normalize_graph, validate_graph +from services.node_specs import graph_references, normalize_graph, validate_graph from services.runtime_variables import DynamicVariableStore, prepare_dynamic_config from services.workflow_engine import WorkflowEngine @@ -49,6 +49,22 @@ def valid_graph(): class WorkflowGraphTests(unittest.TestCase): + def test_agent_entry_mode_defaults_and_validation(self): + graph = valid_graph() + normalized = normalize_graph(graph) + agent = next(node for node in normalized["nodes"] if node["type"] == "agent") + self.assertEqual(agent["data"]["entryMode"], "wait_user") + self.assertEqual(agent["data"]["entrySpeech"], "") + self.assertTrue(agent["data"]["inheritGlobalConfig"]) + self.assertEqual(agent["data"]["contextPolicy"], "fresh") + + agent["data"]["entryMode"] = "fixed_speech" + self.assertTrue( + any("固定进入语不能为空" in error for error in validate_graph(normalized)) + ) + agent["data"]["entrySpeech"] = "您好,{{customer}}" + self.assertEqual(validate_graph(normalized), []) + def test_voice_resource_creates_isolated_runtime_config(self): base = AssistantConfig(type="workflow", asr="default", voice="default") asr = RuntimeModelResource( @@ -63,6 +79,191 @@ class WorkflowGraphTests(unittest.TestCase): self.assertEqual(resolved.stt_api_key, "secret") self.assertEqual(base.asr, "default") + llm = RuntimeModelResource( + id="llm_1", + capability="LLM", + interface_type="openai-llm", + values={"modelId": "deepseek-chat", "apiUrl": "https://llm.test/v1"}, + secrets={"apiKey": "llm-secret"}, + ) + llm_resolved = config_with_resource(base, llm) + self.assertEqual(llm_resolved.model, "deepseek-chat") + self.assertEqual(llm_resolved.llm_api_key, "llm-secret") + + def test_global_and_custom_agent_references_are_preserved(self): + graph = valid_graph() + graph["settings"].update( + { + "defaultLlmResourceId": "llm_global", + "defaultAsrResourceId": "asr_global", + "defaultTtsResourceId": "tts_global", + "toolIds": ["tool_global"], + "knowledgeBaseId": "kb_global", + } + ) + agent = next(node for node in graph["nodes"] if node["type"] == "agent") + agent["data"].update( + { + "inheritGlobalConfig": False, + "llmResourceId": "llm_agent", + "asrResourceId": "asr_agent", + "ttsResourceId": "tts_agent", + "toolIds": ["tool_agent"], + "knowledgeBaseId": "kb_agent", + } + ) + + refs = graph_references(graph) + self.assertEqual( + refs["model_resources"], + { + "llm_global", + "asr_global", + "tts_global", + "llm_agent", + "asr_agent", + "tts_agent", + }, + ) + self.assertEqual(refs["tools"], {"tool_global", "tool_agent"}) + self.assertEqual(refs["knowledge_bases"], {"kb_global", "kb_agent"}) + + def test_existing_agent_override_disables_implicit_inheritance(self): + graph = valid_graph() + agent = next(node for node in graph["nodes"] if node["type"] == "agent") + agent["data"]["toolIds"] = ["legacy_tool"] + normalized = normalize_graph(graph) + normalized_agent = next( + node for node in normalized["nodes"] if node["type"] == "agent" + ) + self.assertFalse(normalized_agent["data"]["inheritGlobalConfig"]) + + def test_inherited_agent_ignores_stale_custom_references(self): + graph = valid_graph() + agent = next(node for node in graph["nodes"] if node["type"] == "agent") + agent["data"].update( + { + "inheritGlobalConfig": True, + "llmResourceId": "stale_llm", + "asrResourceId": "stale_asr", + "ttsResourceId": "stale_tts", + "toolIds": ["stale_tool"], + "knowledgeBaseId": "stale_kb", + } + ) + + refs = graph_references(graph) + + self.assertNotIn("stale_llm", refs["model_resources"]) + self.assertNotIn("stale_tool", refs["tools"]) + self.assertNotIn("stale_kb", refs["knowledge_bases"]) + + def test_agent_effective_config_inherits_then_switches_to_override(self): + graph = valid_graph() + graph["settings"].update( + { + "defaultLlmResourceId": "llm_global", + "defaultAsrResourceId": "asr_global", + "defaultTtsResourceId": "tts_global", + "toolIds": ["tool_global"], + "knowledgeBaseId": "kb_global", + "knowledgeMode": "on_demand", + "knowledgeTopN": 8, + "knowledgeScoreThreshold": 0.4, + } + ) + engine = WorkflowEngine(graph) + inherited = engine.agent_stage_config("agent") + self.assertEqual(inherited.llm_resource_id, "llm_global") + self.assertEqual(inherited.tool_ids, ("tool_global",)) + self.assertEqual(inherited.knowledge_mode, "on_demand") + + engine.data("agent").update( + { + "inheritGlobalConfig": False, + "llmResourceId": "llm_agent", + "toolIds": ["tool_agent"], + "knowledgeBaseId": "", + } + ) + custom = engine.agent_stage_config("agent") + self.assertEqual(custom.llm_resource_id, "llm_agent") + self.assertEqual(custom.tool_ids, ("tool_agent",)) + self.assertEqual(custom.knowledge_mode, "disabled") + + def test_start_agent_and_handoff_may_have_no_outgoing_edge(self): + terminal_graphs = [ + { + "specVersion": 3, + "settings": {}, + "nodes": [{"id": "start", "type": "start", "data": {}}], + "edges": [], + }, + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "agent", + "type": "agent", + "data": {"prompt": "持续处理用户问题"}, + }, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "agent", + "data": {"mode": "always", "priority": 0}, + } + ], + }, + { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "handoff", "type": "handoff", "data": {}}, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "handoff", + "data": {"mode": "always", "priority": 0}, + } + ], + }, + ] + + for graph in terminal_graphs: + with self.subTest(node=graph["nodes"][-1]["type"]): + self.assertEqual(validate_graph(graph), []) + + action_without_exit = { + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + {"id": "action", "type": "action", "data": {}}, + ], + "edges": [ + { + "id": "begin", + "source": "start", + "target": "action", + "data": {"mode": "always", "priority": 0}, + } + ], + } + self.assertTrue( + any( + "action 的出边不能少于 1" in error + for error in validate_graph(action_without_exit) + ) + ) + def test_v2_start_prompt_is_preserved_in_synthetic_agent(self): graph = normalize_graph( { @@ -113,6 +314,15 @@ class WorkflowGraphTests(unittest.TestCase): ) self.assertIn("王先生", engine.prompt_for("agent", store)) + inherited_prompt = engine.prompt_for("agent", store) + self.assertIn("服务 王先生", inherited_prompt) + self.assertIn("处理订单", inherited_prompt) + + engine.data("agent")["inheritGlobalConfig"] = False + custom_prompt = engine.prompt_for("agent", store) + self.assertNotIn("服务 王先生", custom_prompt) + self.assertIn("处理订单", custom_prompt) + if __name__ == "__main__": unittest.main() diff --git a/frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx b/frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx new file mode 100644 index 0000000..b41412f --- /dev/null +++ b/frontend/src/components/editor/knowledge-retrieval-config-dialog.tsx @@ -0,0 +1,168 @@ +"use client"; + +import { Settings2 } from "lucide-react"; +import { useState } from "react"; + +import { Button } from "@/components/ui/button"; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { Input } from "@/components/ui/input"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import type { KnowledgeRetrievalConfig } from "@/lib/api"; + +export const DEFAULT_KNOWLEDGE_RETRIEVAL_CONFIG: KnowledgeRetrievalConfig = { + mode: "automatic", + topN: 5, + scoreThreshold: 0, +}; + +export function KnowledgeRetrievalConfigDialog({ + disabled, + value, + onChange, +}: { + disabled: boolean; + value: KnowledgeRetrievalConfig; + onChange: (config: KnowledgeRetrievalConfig) => void; +}) { + const [open, setOpen] = useState(false); + const [draft, setDraft] = useState(value); + const [error, setError] = useState(null); + + function openDialog() { + setDraft(value); + setError(null); + setOpen(true); + } + + function saveDraft() { + if (draft.topN === 0 || draft.topN < -1 || !Number.isInteger(draft.topN)) { + setError("Top N 必须为 -1 或大于 0 的整数"); + return; + } + if (draft.scoreThreshold < 0 || draft.scoreThreshold > 1) { + setError("最低相关度必须在 0 到 1 之间"); + return; + } + onChange(draft); + setOpen(false); + } + + return ( + <> + + + + + + 知识库高级配置 + + 设置检索触发方式、返回数量和相关度过滤条件。 + + + +
+
+
检索方式
+ +

+ {draft.mode === "automatic" + ? "每轮用户提问后自动检索,响应行为更稳定。" + : "由大模型判断是否调用知识库,依赖模型的工具调用能力。"} +

+
+ + + + + + {error &&

{error}

} +
+ + + + + +
+
+ + ); +} diff --git a/frontend/src/components/editor/section-card.tsx b/frontend/src/components/editor/section-card.tsx new file mode 100644 index 0000000..8f0954e --- /dev/null +++ b/frontend/src/components/editor/section-card.tsx @@ -0,0 +1,92 @@ +"use client"; + +/** + * Compact section chrome shared by assistant editors and workflow node panels. + * Density matches the debug preview drawer (text-sm titles, tight padding). + */ + +import { HelpCircle } from "lucide-react"; +import type { ReactNode } from "react"; + +import { + Card, + CardContent, + CardHeader, + CardTitle, +} from "@/components/ui/card"; +import { + Popover, + PopoverContent, + PopoverTrigger, +} from "@/components/ui/popover"; +import { cn } from "@/lib/utils"; + +export function SectionCard({ + icon, + title, + description, + children, + className, +}: { + icon?: ReactNode; + title?: string; + description?: string; + children: ReactNode; + className?: string; +}) { + const hasHeader = Boolean(title); + + return ( + + {hasHeader && ( + +
+ {icon && ( +
+ {icon} +
+ )} +
+ + {title} + + {description && } +
+
+
+ )} + + {children} + +
+ ); +} + +export function HelpHint({ text }: { text: string }) { + return ( + + + + + + {text} + + + ); +} diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 1e647ba..a171382 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -21,7 +21,6 @@ import { Save, Mic, Send, - HelpCircle, Waypoints, AudioLines, Terminal, @@ -87,12 +86,6 @@ import { PageHeader } from "@/components/ui/page-header"; import { FilterPills } from "@/components/ui/filter-pills"; import { SearchInput } from "@/components/ui/search-input"; import { ListToolbar } from "@/components/ui/list-toolbar"; -import { - Card, - CardContent, - CardHeader, - CardTitle, -} from "@/components/ui/card"; import { useCallback, useEffect, useRef, useState } from "react"; import { useRouter } from "next/navigation"; import { @@ -124,6 +117,7 @@ import { WorkflowEditor, type WorkflowSettings, } from "@/components/workflow/WorkflowEditor"; +import { HelpHint, SectionCard } from "@/components/editor/section-card"; import { defaultGraph, type WorkflowGraph, @@ -362,7 +356,7 @@ type AssistantTypeOption = { label: string; description: string; icon: React.ReactNode; - /** 提示词、Dify、FastGPT 类型已落地,工作流暂时显示占位页 */ + /** 提示词、工作流、Dify、FastGPT 已落地;OpenCode 暂时显示即将上线 */ available: boolean; }; @@ -379,7 +373,7 @@ const assistantTypeOptions: AssistantTypeOption[] = [ label: "使用工作流构建", description: "用可视化编排串联多个节点,适合多步骤、带分支的复杂流程。", icon: , - available: false, + available: true, }, { type: "Dify", @@ -400,7 +394,7 @@ const assistantTypeOptions: AssistantTypeOption[] = [ label: "使用 OpenCode 构建", description: "对接 OpenCode 服务,通过提示词驱动代码助手并支持实时语音对话。", icon: , - available: true, + available: false, }, ]; @@ -472,12 +466,23 @@ export function AssistantPage(props: AssistantPageProps) { ); const [workflowSettings, setWorkflowSettings] = useState({ globalPrompt: defaultGraph().settings.globalPrompt, + llm: defaultGraph().settings.defaultLlmResourceId, + asr: defaultGraph().settings.defaultAsrResourceId, + tts: defaultGraph().settings.defaultTtsResourceId, + toolIds: defaultGraph().settings.toolIds, + knowledgeBaseId: defaultGraph().settings.knowledgeBaseId, + knowledgeRetrievalConfig: { + mode: defaultGraph().settings.knowledgeMode, + topN: defaultGraph().settings.knowledgeTopN, + scoreThreshold: defaultGraph().settings.knowledgeScoreThreshold, + }, allowInterrupt: true, turnConfig: defaultTurnConfig(), }); const [workflowDynamicVariableDefinitions, setWorkflowDynamicVariableDefinitions] = useState>({}); const [workflowDebugOpen, setWorkflowDebugOpen] = useState(false); + const [workflowSettingsOpen, setWorkflowSettingsOpen] = useState(false); const [workflowEditingNodeId, setWorkflowEditingNodeId] = useState(null); const [workflowEditingEdgeId, setWorkflowEditingEdgeId] = useState(null); const [activeNodeId, setActiveNodeId] = useState(null); @@ -847,9 +852,25 @@ export function AssistantPage(props: AssistantPageProps) { ? (assistant.graph as WorkflowGraph) : defaultGraph(); const wfSettings: WorkflowSettings = { - llm: assistant.modelResourceIds.LLM, + llm: + graph.settings?.defaultLlmResourceId || + assistant.modelResourceIds.LLM, asr: graph.settings?.defaultAsrResourceId || assistant.modelResourceIds.ASR, tts: graph.settings?.defaultTtsResourceId || assistant.modelResourceIds.TTS, + toolIds: graph.settings?.toolIds ?? [], + knowledgeBaseId: + graph.settings?.knowledgeBaseId || assistant.knowledgeBaseId || "", + knowledgeRetrievalConfig: { + mode: + graph.settings?.knowledgeMode || + assistant.knowledgeRetrievalConfig.mode, + topN: + graph.settings?.knowledgeTopN ?? + assistant.knowledgeRetrievalConfig.topN, + scoreThreshold: + graph.settings?.knowledgeScoreThreshold ?? + assistant.knowledgeRetrievalConfig.scoreThreshold, + }, globalPrompt: graph.settings?.globalPrompt ?? "", allowInterrupt: assistant.enableInterrupt, turnConfig: assistant.turnConfig, @@ -916,6 +937,9 @@ export function AssistantPage(props: AssistantPageProps) { ...(workflowSettings.asr ? { ASR: workflowSettings.asr } : {}), ...(workflowSettings.tts ? { TTS: workflowSettings.tts } : {}), }, + knowledgeBaseId: workflowSettings.knowledgeBaseId || null, + knowledgeRetrievalConfig: workflowSettings.knowledgeRetrievalConfig, + toolIds: workflowSettings.toolIds, graph: workflowGraph as unknown as Record, dynamicVariableDefinitions: workflowDynamicVariableDefinitions, }), @@ -1386,7 +1410,7 @@ export function AssistantPage(props: AssistantPageProps) { {saveError} @@ -1396,6 +1420,7 @@ export function AssistantPage(props: AssistantPageProps) { className="gap-2 border-hairline-strong text-foreground hover:bg-surface-strong" disabled={!editingId} onClick={() => { + setWorkflowSettingsOpen(false); setWorkflowEditingNodeId(null); setWorkflowEditingEdgeId(null); setWorkflowDebugOpen(true); @@ -1430,6 +1455,8 @@ export function AssistantPage(props: AssistantPageProps) { onEditingNodeIdChange={setWorkflowEditingNodeId} editingEdgeId={workflowEditingEdgeId} onEditingEdgeIdChange={setWorkflowEditingEdgeId} + settingsOpen={workflowSettingsOpen} + onSettingsOpenChange={setWorkflowSettingsOpen} debugOpen={workflowDebugOpen} onDebugOpenChange={(open) => { setWorkflowDebugOpen(open); @@ -1508,10 +1535,10 @@ export function AssistantPage(props: AssistantPageProps) { -
-
+
+
} + icon={} title="Dify 应用配置" description="从「模型资源」中选择 Dify 应用。开场白、知识库、提示词等对话编排请在 Dify 平台配置,本页不重复设置。" > @@ -1525,7 +1552,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="语音配置" description="从「模型资源」中选择语音识别与语音合成。大模型、知识库与开场白由 Dify 应用提供,请前往 Dify 平台配置。" > @@ -1546,7 +1573,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -1604,10 +1631,10 @@ export function AssistantPage(props: AssistantPageProps) {
-
-
+
+
} + icon={} title="FastGPT 应用配置" description="从「模型资源」中选择 FastGPT 应用。开场白、知识库、提示词等对话编排请在 FastGPT 平台配置,本页不重复设置。" > @@ -1621,7 +1648,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="语音配置" description="从「模型资源」中选择语音识别与语音合成。大模型、知识库与开场白由 FastGPT 应用提供,请前往 FastGPT 平台配置。" > @@ -1642,7 +1669,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -1704,10 +1731,10 @@ export function AssistantPage(props: AssistantPageProps) {
-
-
+
+
} + icon={} title="OpenCode 服务配置" description="从「模型资源」中选择 OpenCode 服务资源。" > @@ -1721,7 +1748,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="提示词" description="描述助手的角色、能力和回答要求" > @@ -1734,7 +1761,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="模型与语音配置" description="配置 OpenCode 使用的大语言模型、语音识别与语音合成资源。" > @@ -1779,7 +1806,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -1843,10 +1870,10 @@ export function AssistantPage(props: AssistantPageProps) { } /> -
-
+
+
-
+
-
-
- +
+
+
- Pipeline 模式 + Pipeline 模式
{form.runtimeMode === "pipeline" && ( - - + + )}
@@ -1893,25 +1920,25 @@ export function AssistantPage(props: AssistantPageProps) { } }} className={[ - "cursor-pointer rounded-2xl border p-5 text-left transition-colors", + "cursor-pointer rounded-xl border p-3.5 text-left transition-colors", form.runtimeMode === "realtime" ? "border-primary bg-primary/5 ring-1 ring-primary" : "border-hairline bg-canvas-soft hover:border-hairline-strong", ].join(" ")} >
-
-
- +
+
+
- Realtime 模式 + Realtime 模式
{form.runtimeMode === "realtime" && ( - - + + )}
@@ -1920,7 +1947,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="提示词" description="描述助手的角色、能力和回答要求" > @@ -1938,7 +1965,7 @@ export function AssistantPage(props: AssistantPageProps) { {form.runtimeMode === "pipeline" ? ( } + icon={} title="模型配置" description="从「模型资源」中选择大语言模型、语音识别与语音合成" > @@ -1983,7 +2010,7 @@ export function AssistantPage(props: AssistantPageProps) { ) : ( } + icon={} title="模型配置" description="当前模式下 ASR 与 TTS 由 Realtime 模型内置完成" > @@ -1998,7 +2025,7 @@ export function AssistantPage(props: AssistantPageProps) { )} } + icon={} title="开场白" description="助手与用户首次对话时的开场语" > @@ -2015,7 +2042,7 @@ export function AssistantPage(props: AssistantPageProps) { {form.runtimeMode === "pipeline" && ( } + icon={} title="知识库配置" description="选择助手回答时可检索的业务知识来源" > @@ -2041,7 +2068,7 @@ export function AssistantPage(props: AssistantPageProps) { )} } + icon={} title="工具" description="配置该提示词助手可以调用的工具" > @@ -2053,7 +2080,7 @@ export function AssistantPage(props: AssistantPageProps) { } + icon={} title="交互策略" description="设置实时视频对话时的交互体验" > @@ -3377,29 +3404,6 @@ function EditableTitle({ ); } -function HelpHint({ text }: { text: string }) { - return ( - - - - - - {text} - - - ); -} - function DynamicVariableEditorHint({ count, onOpen, @@ -3657,45 +3661,6 @@ function DynamicVariablesDialog({ ); } -function SectionCard({ - icon, - title, - description, - children, -}: { - icon?: React.ReactNode; - title?: string; - description?: string; - children: React.ReactNode; -}) { - const hasHeader = Boolean(title); - - return ( - - {hasHeader && ( - -
- {icon && ( -
- {icon} -
- )} - -
- {title} - {description && } -
-
-
- )} - - - {children} - -
- ); -} - function TextAreaField({ label, value, @@ -3712,7 +3677,7 @@ function TextAreaField({ return (