diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index bd738ff..b9f9036 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -60,6 +60,9 @@ class BrainRuntime: ) = None set_knowledge_scope: Callable[[dict[str, Any]], None] | None = None set_input_enabled: Callable[[bool], None] | None = None + apply_turn_config: ( + Callable[[bool, dict[str, Any]], Awaitable[None]] | None + ) = None flow_global_functions: list[Any] = field(default_factory=list) diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 5676a7a..128e890 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -2,6 +2,7 @@ from __future__ import annotations +from copy import deepcopy from typing import Any from loguru import logger @@ -49,7 +50,13 @@ class WorkflowBrain(BaseBrain): def __init__(self, cfg_or_graph: AssistantConfig | dict[str, Any]): cfg = cfg_or_graph if isinstance(cfg_or_graph, AssistantConfig) else None - graph = cfg.graph if cfg is not None else cfg_or_graph + graph = deepcopy(cfg.graph if cfg is not None else cfg_or_graph) + if cfg is not None: + # Graph v3 owns Workflow defaults. Keep older saved graphs compatible + # by filling the new interaction settings from the assistant row. + settings = graph.setdefault("settings", {}) + settings.setdefault("enableInterrupt", cfg.enableInterrupt) + settings.setdefault("turnConfig", deepcopy(cfg.turnConfig)) self._engine = WorkflowEngine(graph or {}) if not self._engine.has_graph() or not self._engine.start_id: raise ValueError("WorkflowBrain 缺少有效的 Start 节点") @@ -95,6 +102,10 @@ class WorkflowBrain(BaseBrain): async def on_connected(self) -> None: await self._emit_node_active(self._engine.start_id) + await self._emit_variables( + reason="initialized", + node_id=self._engine.start_id, + ) edge = self._engine.deterministic_edge( self._engine.start_id, self._store, @@ -228,6 +239,11 @@ class WorkflowBrain(BaseBrain): if self._runtime and self._runtime.set_input_enabled: self._runtime.set_input_enabled(True) runtime = self._require_runtime() + if runtime.apply_turn_config: + await runtime.apply_turn_config( + stage.enable_interrupt, + stage.turn_config, + ) if runtime.switch_services: await runtime.switch_services( stage.llm_resource_id or None, @@ -248,6 +264,11 @@ class WorkflowBrain(BaseBrain): 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 "")) + fixed_reply_messages = ( + [{"role": "assistant", "content": entry_speech}] + if entry_mode == "fixed_speech" and entry_speech + else [] + ) strategy = ( ContextStrategy.RESET if data.get("contextPolicy") == "fresh" @@ -265,11 +286,10 @@ class WorkflowBrain(BaseBrain): config: NodeConfig = { "name": node_id, "role_message": self._agent_role_message(node_id), - "task_messages": ( - [{"role": "assistant", "content": entry_speech}] - if entry_mode == "fixed_speech" - else [] - ), + # Flows writes task_messages into the Pipecat LLM context. The + # pre-action below is responsible only for display, persistence, + # dynamic conversation history, and TTS playback. + "task_messages": fixed_reply_messages, "functions": functions, "context_strategy": ContextStrategyConfig(strategy=strategy), "respond_immediately": entry_mode == "generate", @@ -279,6 +299,7 @@ class WorkflowBrain(BaseBrain): { "type": "workflow_fixed_speech", "text": entry_speech, + "node_id": node_id, "handler": self._play_fixed_speech, } ] @@ -286,9 +307,19 @@ class WorkflowBrain(BaseBrain): 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 "")) + await self._queue_visible_speech( + str(action.get("text") or ""), + source="workflow-fixed-reply", + node_id=str(action.get("node_id") or "") or None, + ) - async def _queue_visible_speech(self, text: str) -> None: + async def _queue_visible_speech( + self, + text: str, + *, + source: str = "workflow-speech", + node_id: str | None = None, + ) -> None: """Show and persist fixed workflow speech before sending it to TTS.""" content = text.strip() if not content: @@ -302,6 +333,8 @@ class WorkflowBrain(BaseBrain): "role": "assistant", "content": content, "timestamp": time_now_iso8601(), + "source": source, + **({"nodeId": node_id} if node_id else {}), } ) ) @@ -327,7 +360,13 @@ class WorkflowBrain(BaseBrain): result = await self._tools.execute(tool, dict(args or {})) except ToolExecutionError as exc: return {"status": "error", "message": str(exc)} - if result.get("updated_variables"): + updated_variables = list(result.get("updated_variables") or []) + if updated_variables: + await self._emit_variables( + reason="tool", + node_id=node_id, + changed=updated_variables, + ) await self._refresh_agent_prompt(node_id) edge = self._engine.deterministic_edge( node_id, @@ -436,11 +475,18 @@ class WorkflowBrain(BaseBrain): return try: arguments = self._store.render_data(data.get("arguments") or {}) - await self._tools.execute( + result = await self._tools.execute( tool, arguments, result_assignments=data.get("resultAssignments") or {}, ) + updated_variables = list(result.get("updated_variables") or []) + if updated_variables: + await self._emit_variables( + reason="action", + node_id=node_id, + changed=updated_variables, + ) self._store.values["system__last_action_status"] = "ok" self._store.values["system__last_action_error"] = "" except (ToolExecutionError, ValueError) as exc: @@ -501,6 +547,40 @@ class WorkflowBrain(BaseBrain): ) ) + def _public_variables(self) -> dict[str, str | int | float | bool]: + """Return the browser-safe part of this session's variable state.""" + return { + name: value + for name, value in self._store.values.items() + if not name.startswith(("system__", "secret__")) + and isinstance(value, (str, int, float, bool)) + } + + async def _emit_variables( + self, + *, + reason: str, + node_id: str | None, + changed: list[str] | None = None, + ) -> None: + """Publish a safe snapshot so Workflow debug mirrors runtime state.""" + message: dict[str, Any] = { + "type": "workflow-variables", + "reason": reason, + "variables": self._public_variables(), + } + if node_id: + message["nodeId"] = node_id + if changed: + message["changed"] = [ + name + for name in changed + if not name.startswith(("system__", "secret__")) + ] + await self._require_runtime().queue_frame( + OutputTransportMessageUrgentFrame(message=message) + ) + def _require_runtime(self) -> BrainRuntime: if self._runtime is None: raise RuntimeError("WorkflowBrain 尚未绑定 pipeline runtime") diff --git a/backend/services/conversation_history.py b/backend/services/conversation_history.py index 28a7f3c..9139274 100644 --- a/backend/services/conversation_history.py +++ b/backend/services/conversation_history.py @@ -77,6 +77,10 @@ class ConversationRecorder: role = str(message.get("role") or "") content = str(message.get("content") or "").strip() event_key = f"transcript:{role}:{timestamp}:{content}" + if message.get("source"): + extra["source"] = str(message["source"]) + if message.get("nodeId"): + extra["node_id"] = str(message["nodeId"]) elif event_type == "assistant-text-end": role = "assistant" content = str(message.get("content") or "").strip() diff --git a/backend/services/node_specs.py b/backend/services/node_specs.py index 02ebace..0291afb 100644 --- a/backend/services/node_specs.py +++ b/backend/services/node_specs.py @@ -162,6 +162,8 @@ def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> settings.setdefault("knowledgeMode", "automatic") settings.setdefault("knowledgeTopN", 5) settings.setdefault("knowledgeScoreThreshold", 0.0) + settings.setdefault("enableInterrupt", True) + settings.setdefault("turnConfig", {}) def normalize_graph(graph: dict[str, Any] | None) -> dict[str, Any]: diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 8896465..4593554 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -10,6 +10,7 @@ import asyncio import base64 from collections.abc import Callable from io import BytesIO +from typing import Any from uuid import uuid4 from loguru import logger @@ -50,6 +51,7 @@ from pipecat.frames.frames import ( TTSSpeakFrame, UserImageRawFrame, UserImageRequestFrame, + VADParamsUpdateFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.llm_switcher import LLMSwitcher @@ -58,7 +60,6 @@ from pipecat.pipeline.worker import PipelineParams, PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import ( LLMAssistantAggregator, - LLMUserAggregator, LLMUserAggregatorParams, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -72,8 +73,10 @@ from pipecat.turns.user_mute.function_call_user_mute_strategy import ( FunctionCallUserMuteStrategy, ) from services.pipecat.turn_config import ( + ConfigurableLLMUserAggregator, create_user_turn_strategies, create_vad_analyzer, + create_vad_params, ) from pipecat.utils.time import time_now_iso8601 from pipecat.workers.runner import WorkerRunner @@ -794,7 +797,7 @@ async def run_pipeline( current_llm_service = llm if cfg.type == "workflow": llm, llm_services, current_llm_service = _workflow_llm_switcher(cfg, llm) - user_aggregator = LLMUserAggregator( + user_aggregator = ConfigurableLLMUserAggregator( context, params=LLMUserAggregatorParams( vad_analyzer=create_vad_analyzer(cfg.turnConfig), @@ -1063,6 +1066,31 @@ async def run_pipeline( ) ) + current_enable_interrupt = cfg.enableInterrupt + current_turn_config = dict(cfg.turnConfig) + + async def apply_workflow_turn_config( + enable_interrupt: bool, + turn_config: dict[str, Any], + ) -> None: + """Apply one Agent's interaction policy before its next user turn.""" + nonlocal current_enable_interrupt, current_turn_config + normalized = dict(turn_config or {}) + if ( + current_enable_interrupt == enable_interrupt + and current_turn_config == normalized + ): + return + await user_aggregator.apply_turn_strategies( + normalized, + enable_interruptions=enable_interrupt, + ) + await worker.queue_frame( + VADParamsUpdateFrame(params=create_vad_params(normalized)) + ) + current_enable_interrupt = enable_interrupt + current_turn_config = normalized + async def queue_transcript(role: str, content: str, timestamp: str) -> None: if content: await worker.queue_frame( @@ -1107,6 +1135,7 @@ async def run_pipeline( switch_services=switch_workflow_services, set_knowledge_scope=knowledge_retrieval.set_scope, set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled), + apply_turn_config=apply_workflow_turn_config, flow_global_functions=flow_global_functions, ), ) diff --git a/backend/services/pipecat/turn_config.py b/backend/services/pipecat/turn_config.py index 9dc1d9d..42eed19 100644 --- a/backend/services/pipecat/turn_config.py +++ b/backend/services/pipecat/turn_config.py @@ -7,6 +7,11 @@ from typing import Any from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMUserAggregator, + LLMUserAggregatorParams, +) from pipecat.turns.user_start import ( TranscriptionUserTurnStartStrategy, VADUserTurnStartStrategy, @@ -39,18 +44,21 @@ def _value(config: dict[str, Any], snake: str, camel: str, default: Any) -> Any: return config.get(snake, config.get(camel, default)) -def create_vad_analyzer(turn_config: dict[str, Any]) -> SileroVADAnalyzer: +def create_vad_params(turn_config: dict[str, Any]) -> VADParams: + """Translate product settings into Pipecat's runtime VAD parameters.""" vad = _section(turn_config, "vad", "vad") - return SileroVADAnalyzer( - params=VADParams( - confidence=float(vad.get("confidence", DEFAULT_VAD["confidence"])), - start_secs=float(_value(vad, "start_secs", "startSecs", 0.2)), - stop_secs=float(_value(vad, "stop_secs", "stopSecs", 0.2)), - min_volume=float(_value(vad, "min_volume", "minVolume", 0.6)), - ) + return VADParams( + confidence=float(vad.get("confidence", DEFAULT_VAD["confidence"])), + start_secs=float(_value(vad, "start_secs", "startSecs", 0.2)), + stop_secs=float(_value(vad, "stop_secs", "stopSecs", 0.2)), + min_volume=float(_value(vad, "min_volume", "minVolume", 0.6)), ) +def create_vad_analyzer(turn_config: dict[str, Any]) -> SileroVADAnalyzer: + return SileroVADAnalyzer(params=create_vad_params(turn_config)) + + def create_user_turn_strategies( turn_config: dict[str, Any], *, enable_interruptions: bool ) -> UserTurnStrategies: @@ -87,3 +95,34 @@ def create_user_turn_strategies( ) ] return UserTurnStrategies(start=start, stop=stop) + + +class ConfigurableLLMUserAggregator(LLMUserAggregator): + """LLM user aggregator with one stable project-level runtime update API. + + Pipecat 1.5 exposes ``UserTurnController.update_strategies`` but does not + surface it on ``LLMUserAggregator``. Keeping that version-specific bridge + here prevents Workflow orchestration from depending on Pipecat internals. + VAD threshold updates still use Pipecat's public ``VADParamsUpdateFrame``. + """ + + def __init__( + self, + context: LLMContext, + *, + params: LLMUserAggregatorParams | None = None, + **kwargs: Any, + ) -> None: + super().__init__(context, params=params, **kwargs) + + async def apply_turn_strategies( + self, + turn_config: dict[str, Any], + *, + enable_interruptions: bool, + ) -> None: + strategies = create_user_turn_strategies( + turn_config, + enable_interruptions=enable_interruptions, + ) + await self._user_turn_controller.update_strategies(strategies) diff --git a/backend/services/runtime_variables.py b/backend/services/runtime_variables.py index ed1f14a..262c94c 100644 --- a/backend/services/runtime_variables.py +++ b/backend/services/runtime_variables.py @@ -1,4 +1,4 @@ -"""Conversation-scoped dynamic variables for prompt pipeline assistants. +"""Conversation-scoped dynamic variables shared by Prompt and Workflow. The renderer is deliberately small: it only understands ``{{ name }}`` placeholders and never evaluates expressions. A value is substituted once, @@ -21,6 +21,7 @@ from models import AssistantConfig Primitive = str | int | float | bool VARIABLE_NAME = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,63}$") PLACEHOLDER = re.compile(r"{{\s*([A-Za-z][A-Za-z0-9_]*)\s*}}") +FULL_PLACEHOLDER = re.compile(r"^{{\s*([A-Za-z][A-Za-z0-9_]*)\s*}}$") MAX_VARIABLES = 50 MAX_VALUE_LENGTH = 2048 MAX_HISTORY_ENTRIES = 50 @@ -91,37 +92,71 @@ class DynamicVariableStore: self, values: dict[str, Primitive], secrets: dict[str, str] | None = None, + *, + optional_names: set[str] | None = None, + variable_types: dict[str, str] | None = None, ): self.values = dict(values) self.secrets = dict(secrets or {}) + self.optional_names = set(optional_names or set()) + self.variable_types = dict(variable_types or {}) self.history: list[dict[str, str]] = [] @classmethod def from_config(cls, cfg: AssistantConfig) -> "DynamicVariableStore": - return cls(cfg.dynamic_variables, cfg.secret_dynamic_variables) + definitions = cfg.dynamic_variable_definitions or {} + optional_names = { + name + for name, definition in definitions.items() + if not definition.get("required", False) + and definition.get("default") is None + } + variable_types = { + name: str(definition.get("type") or "string") + for name, definition in definitions.items() + } + return cls( + cfg.dynamic_variables, + cfg.secret_dynamic_variables, + optional_names=optional_names, + variable_types=variable_types, + ) - def render(self, template: str, *, allow_secrets: bool = False) -> str: - if not template: - return template + def _refresh_time(self) -> None: timezone = str(self.values.get("system__timezone") or "Asia/Shanghai") try: now = datetime.now(ZoneInfo(timezone)) self.values["system__time"] = now.strftime("%A, %H:%M %d %B %Y") - self.values["system__time_utc"] = now.astimezone(ZoneInfo("UTC")).isoformat() + self.values["system__time_utc"] = now.astimezone( + ZoneInfo("UTC") + ).isoformat() except ZoneInfoNotFoundError: pass + def _resolve(self, name: str, *, allow_secrets: bool) -> Primitive: + if name.startswith("secret__"): + if not allow_secrets: + raise DynamicVariableError(f"密钥变量 {name} 只能用于 HTTP Header") + if name not in self.secrets: + raise DynamicVariableError(f"缺少密钥变量: {name}") + return self.secrets[name] + if name in self.values: + return self.values[name] + # Optional variables intentionally remain absent from ``values`` so an + # ``exists`` expression can distinguish unset from an explicit value. + # Text templates still render predictably instead of failing the call. + if name in self.optional_names: + return "" + raise DynamicVariableError(f"缺少动态变量: {name}") + + def render(self, template: str, *, allow_secrets: bool = False) -> str: + if not template: + return template + self._refresh_time() + def replace(match: re.Match[str]) -> str: name = match.group(1) - if name.startswith("secret__"): - if not allow_secrets: - raise DynamicVariableError(f"密钥变量 {name} 只能用于 HTTP Header") - if name not in self.secrets: - raise DynamicVariableError(f"缺少密钥变量: {name}") - return self.secrets[name] - if name not in self.values: - raise DynamicVariableError(f"缺少动态变量: {name}") - value = self.values[name] + value = self._resolve(name, allow_secrets=allow_secrets) if isinstance(value, bool): return "true" if value else "false" return str(value) @@ -130,6 +165,12 @@ class DynamicVariableStore: def render_data(self, value: Any, *, allow_secrets: bool = False) -> Any: if isinstance(value, str): + exact = FULL_PLACEHOLDER.fullmatch(value) + if exact: + self._refresh_time() + return deepcopy( + self._resolve(exact.group(1), allow_secrets=allow_secrets) + ) return self.render(value, allow_secrets=allow_secrets) if isinstance(value, list): return [self.render_data(item, allow_secrets=allow_secrets) for item in value] @@ -163,7 +204,11 @@ class DynamicVariableStore: def assign(self, name: str, value: Any) -> None: if name.startswith(("system__", "secret__")) or not VARIABLE_NAME.fullmatch(name): raise DynamicVariableError(f"工具不能更新保留变量: {name}") - self.values[name] = _primitive(value, name) + primitive = _primitive(value, name) + expected = self.variable_types.get(name) + if expected and not _type_matches(primitive, expected): + raise DynamicVariableError(f"动态变量 {name} 类型应为 {expected}") + self.values[name] = primitive def prepare_dynamic_config( diff --git a/backend/services/workflow_engine.py b/backend/services/workflow_engine.py index 98078e9..1114231 100644 --- a/backend/services/workflow_engine.py +++ b/backend/services/workflow_engine.py @@ -23,6 +23,8 @@ class AgentStageConfig: knowledge_mode: str knowledge_top_n: int knowledge_score_threshold: float + enable_interrupt: bool + turn_config: dict[str, Any] class WorkflowEngine: @@ -106,6 +108,12 @@ class WorkflowEngine: asr_key = "defaultAsrResourceId" if inherits_global else "asrResourceId" tts_key = "defaultTtsResourceId" if inherits_global else "ttsResourceId" knowledge_base_id = str(source.get("knowledgeBaseId") or "") + global_turn_config = self.settings.get("turnConfig") + if not isinstance(global_turn_config, dict): + global_turn_config = {} + turn_config = source.get("turnConfig", global_turn_config) + if not isinstance(turn_config, dict): + turn_config = global_turn_config return AgentStageConfig( inherits_global=inherits_global, llm_resource_id=str(source.get(llm_key) or ""), @@ -122,6 +130,13 @@ class WorkflowEngine: knowledge_score_threshold=float( source.get("knowledgeScoreThreshold") or 0.0 ), + enable_interrupt=bool( + source.get( + "enableInterrupt", + self.settings.get("enableInterrupt", True), + ) + ), + turn_config=dict(turn_config), ) def prompt_for(self, node_id: str, store: DynamicVariableStore) -> str: diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index b2e2e43..cd43868 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -9,6 +9,7 @@ from pipecat.frames.frames import ( LLMContextFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMMessagesUpdateFrame, LLMRunFrame, LLMTextFrame, OutputTransportMessageUrgentFrame, @@ -391,6 +392,78 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_action_publishes_updated_session_variables(self): + tool = RuntimeTool( + id="lookup", + name="查询订单", + function_name="lookup_order", + type="http", + ) + cfg = prepare_dynamic_config( + AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {}, + "nodes": [ + {"id": "start", "type": "start", "data": {}}, + { + "id": "lookup_action", + "type": "action", + "data": { + "toolId": "lookup", + "resultAssignments": { + "order_status": "order.status" + }, + }, + }, + ], + "edges": [], + }, + dynamic_variable_definitions={ + "order_status": {"type": "string", "default": "pending"} + }, + tools=[tool], + ), + {}, + assistant_id="asst_workflow_action", + ) + brain = WorkflowBrain(cfg) + queued = [] + + async def queue_frame(frame): + queued.append(frame) + + async def execute(_tool, _arguments, *, result_assignments=None): + self.assertEqual(result_assignments, {"order_status": "order.status"}) + brain._store.assign("order_status", "paid") + return { + "status": "ok", + "updated_variables": ["order_status"], + } + + brain._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(), + ) + brain._tools.execute = execute + + await brain._enter_action("lookup_action") + + variable_events = [ + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "workflow-variables" + ] + self.assertEqual(variable_events[-1]["reason"], "action") + self.assertEqual(variable_events[-1]["changed"], ["order_status"]) + self.assertEqual(variable_events[-1]["variables"], {"order_status": "paid"}) + async def test_nodes_without_outgoing_edges_remain_active(self): queued = [] @@ -492,6 +565,11 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): "defaultTtsResourceId": "tts_global", "knowledgeBaseId": "kb_global", "knowledgeMode": "automatic", + "enableInterrupt": False, + "turnConfig": { + "bargeIn": {"strategy": "transcription"}, + "vad": {"confidence": 0.55}, + }, }, "nodes": [ { @@ -551,6 +629,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): queued = [] service_switches = [] knowledge_scopes = [] + turn_configs = [] call_end = FakeCallEnd() class FakeWorker: @@ -585,6 +664,9 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): async def switch_services(llm_id, asr_id, tts_id): service_switches.append((llm_id, asr_id, tts_id)) + async def apply_turn_config(enable_interrupt, turn_config): + turn_configs.append((enable_interrupt, turn_config)) + runtime = BrainRuntime( context=context, llm=llm, @@ -596,15 +678,27 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): context_aggregator=pair, switch_services=switch_services, set_knowledge_scope=knowledge_scopes.append, + apply_turn_config=apply_turn_config, ) await brain.setup(cfg, runtime) await brain.on_connected() self.assertEqual(brain._manager.current_node, "agent") + variable_events = [ + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "workflow-variables" + ] + self.assertEqual(variable_events[0]["reason"], "initialized") + self.assertEqual(variable_events[0]["variables"], {"user_name": "王先生"}) + self.assertNotIn("system__conversation_id", variable_events[0]["variables"]) self.assertEqual( service_switches, [("llm_global", "asr_global", "tts_global")], ) self.assertEqual(knowledge_scopes[-1]["knowledge_base_id"], "kb_global") + self.assertEqual(turn_configs[-1][0], False) + self.assertEqual(turn_configs[-1][1]["vad"]["confidence"], 0.55) brain._engine.data("agent").update( { @@ -614,6 +708,11 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): "ttsResourceId": "tts_agent", "knowledgeBaseId": "kb_agent", "knowledgeMode": "on_demand", + "enableInterrupt": True, + "turnConfig": { + "bargeIn": {"strategy": "vad"}, + "turnDetection": {"strategy": "smart_turn"}, + }, } ) await brain._apply_agent_stage("agent") @@ -622,6 +721,11 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ("llm_agent", "asr_agent", "tts_agent"), ) self.assertEqual(knowledge_scopes[-1]["knowledge_base_id"], "kb_agent") + self.assertEqual(turn_configs[-1][0], True) + self.assertEqual( + turn_configs[-1][1]["turnDetection"]["strategy"], + "smart_turn", + ) agent_config = brain._agent_config("agent") self.assertIn("王先生", agent_config["role_message"]) self.assertIn("工作流路由已在用户一轮输入结束时完成", agent_config["role_message"]) @@ -654,11 +758,30 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): fixed_config["task_messages"], [{"role": "assistant", "content": "您好，王先生"}], ) + self.assertEqual(fixed_config["pre_actions"][0]["node_id"], "agent") 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)) + context_updates = [ + frame + for frame in worker.frames + if isinstance(frame, LLMMessagesUpdateFrame) + ] + self.assertEqual( + context_updates[-1].messages, + [{"role": "assistant", "content": "您好，王先生"}], + ) + fixed_reply_events = [ + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("source") == "workflow-fixed-reply" + ] + self.assertEqual(fixed_reply_events[0]["content"], "您好，王先生") + self.assertEqual(fixed_reply_events[0]["nodeId"], "agent") + self.assertIn("您好，王先生", brain._store.values["system__conversation_history"]) self.assertFalse( any( diff --git a/backend/tests/test_conversation_history.py b/backend/tests/test_conversation_history.py new file mode 100644 index 0000000..fcba94c --- /dev/null +++ b/backend/tests/test_conversation_history.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import unittest +from unittest.mock import AsyncMock + +from services.conversation_history import ConversationRecorder + + +class ConversationRecorderTest(unittest.IsolatedAsyncioTestCase): + async def test_fixed_reply_transcript_keeps_workflow_metadata(self): + recorder = ConversationRecorder("conv_test") + recorder._append = AsyncMock() + + await recorder.record_transport_message( + { + "type": "transcript", + "role": "assistant", + "content": "请稍等，我正在处理。", + "timestamp": "2026-07-14T10:00:00+08:00", + "source": "workflow-fixed-reply", + "nodeId": "agent_service", + } + ) + + recorder._append.assert_awaited_once_with( + "assistant", + "请稍等，我正在处理。", + "2026-07-14T10:00:00+08:00", + { + "source": "workflow-fixed-reply", + "node_id": "agent_service", + }, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_runtime_variables.py b/backend/tests/test_runtime_variables.py index 6a3dad4..570b870 100644 --- a/backend/tests/test_runtime_variables.py +++ b/backend/tests/test_runtime_variables.py @@ -79,6 +79,58 @@ class DynamicVariableTests(unittest.TestCase): "Bearer private", ) + def test_optional_workflow_variable_renders_empty_but_remains_unset(self): + cfg = AssistantConfig( + type="workflow", + graph={ + "specVersion": 3, + "settings": {"globalPrompt": "称呼 {{nickname}}"}, + "nodes": [ + { + "id": "start", + "type": "start", + "data": {"greeting": "您好 {{nickname}}"}, + } + ], + "edges": [], + }, + greeting="", + dynamic_variable_definitions={ + "nickname": { + "type": "string", + "required": False, + "default": None, + } + }, + ) + prepared = prepare_dynamic_config(cfg, {}, assistant_id="asst_workflow") + store = DynamicVariableStore.from_config(prepared) + + self.assertEqual(store.render("您好 {{nickname}}"), "您好 ") + self.assertNotIn("nickname", store.values) + with self.assertRaisesRegex(DynamicVariableError, "缺少动态变量"): + store.render("{{not_defined}}") + + def test_exact_placeholder_keeps_action_argument_type(self): + store = DynamicVariableStore( + {"count": 3, "confirmed": True}, + variable_types={"count": "number", "confirmed": "boolean"}, + ) + + rendered = store.render_data( + { + "count": "{{count}}", + "confirmed": "{{confirmed}}", + "label": "数量 {{count}}", + } + ) + self.assertEqual(rendered["count"], 3) + self.assertIs(rendered["confirmed"], True) + self.assertEqual(rendered["label"], "数量 3") + + with self.assertRaisesRegex(DynamicVariableError, "类型应为 number"): + store.assign("count", "four") + def test_tool_assignment_and_system_history(self): store = DynamicVariableStore( { diff --git a/backend/tests/test_workflow_v3.py b/backend/tests/test_workflow_v3.py index 1cfa553..9fde4bc 100644 --- a/backend/tests/test_workflow_v3.py +++ b/backend/tests/test_workflow_v3.py @@ -170,6 +170,11 @@ class WorkflowGraphTests(unittest.TestCase): "knowledgeMode": "on_demand", "knowledgeTopN": 8, "knowledgeScoreThreshold": 0.4, + "enableInterrupt": False, + "turnConfig": { + "bargeIn": {"strategy": "transcription"}, + "turnDetection": {"strategy": "silence", "silenceTimeoutSecs": 1.2}, + }, } ) engine = WorkflowEngine(graph) @@ -177,6 +182,11 @@ class WorkflowGraphTests(unittest.TestCase): self.assertEqual(inherited.llm_resource_id, "llm_global") self.assertEqual(inherited.tool_ids, ("tool_global",)) self.assertEqual(inherited.knowledge_mode, "on_demand") + self.assertFalse(inherited.enable_interrupt) + self.assertEqual( + inherited.turn_config["bargeIn"]["strategy"], + "transcription", + ) engine.data("agent").update( { @@ -184,12 +194,22 @@ class WorkflowGraphTests(unittest.TestCase): "llmResourceId": "llm_agent", "toolIds": ["tool_agent"], "knowledgeBaseId": "", + "enableInterrupt": True, + "turnConfig": { + "bargeIn": {"strategy": "vad"}, + "turnDetection": {"strategy": "smart_turn"}, + }, } ) 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") + self.assertTrue(custom.enable_interrupt) + self.assertEqual( + custom.turn_config["turnDetection"]["strategy"], + "smart_turn", + ) def test_start_agent_and_handoff_may_have_no_outgoing_edge(self): terminal_graphs = [ diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index a171382..43d5b5c 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -122,6 +122,10 @@ import { defaultGraph, type WorkflowGraph, } from "@/components/workflow/specs"; +import { + defaultTurnConfig, + normalizeTurnConfig, +} from "@/lib/turn-config"; type RuntimeMode = "pipeline" | "realtime"; @@ -185,24 +189,6 @@ const assistantTypes: AssistantType[] = [ "OpenCode", ]; -function defaultTurnConfig(): TurnConfig { - return { - bargeIn: { - strategy: "vad", - }, - vad: { - confidence: 0.7, - startSecs: 0.2, - stopSecs: 0.2, - minVolume: 0.6, - }, - turnDetection: { - strategy: "silence", - silenceTimeoutSecs: 0.6, - }, - }; -} - function defaultKnowledgeRetrievalConfig(): KnowledgeRetrievalConfig { return { mode: "automatic", @@ -305,6 +291,45 @@ function activeDynamicVariableDefinitions( ); } +function activeWorkflowDynamicVariableDefinitions( + graph: WorkflowGraph, + saved: Record, +): Record { + const names = new Set(extractDynamicVariableNames(JSON.stringify(graph))); + + // Expression operands and Action assignment destinations are variable + // references even though they do not use the {{placeholder}} syntax. + for (const edge of graph.edges) { + for (const rule of edge.data.expression?.rules ?? []) { + if (isPublicDynamicVariableName(rule.variable)) names.add(rule.variable); + } + } + for (const node of graph.nodes) { + for (const name of Object.keys(node.data.resultAssignments ?? {})) { + if (isPublicDynamicVariableName(name)) names.add(name); + } + } + + return Object.fromEntries( + [...names].sort().map((name) => [ + name, + saved[name] ?? { + type: "string", + required: false, + default: null, + }, + ]), + ); +} + +function isPublicDynamicVariableName(name: string): boolean { + return ( + /^[A-Za-z][A-Za-z0-9_]{0,63}$/.test(name) && + !name.startsWith("system__") && + !name.startsWith("secret__") + ); +} + function blankFastGptForm(name: string): FastGptForm { return { name, @@ -492,6 +517,11 @@ export function AssistantPage(props: AssistantPageProps) { [form.prompt, form.greeting], form.dynamicVariableDefinitions, ); + const effectiveWorkflowDynamicVariableDefinitions = + activeWorkflowDynamicVariableDefinitions( + workflowGraph, + workflowDynamicVariableDefinitions, + ); const loadAssistants = useCallback(async () => { setListLoading(true); @@ -603,7 +633,7 @@ export function AssistantPage(props: AssistantPageProps) { knowledgeRetrievalConfig: a.knowledgeRetrievalConfig ?? defaultKnowledgeRetrievalConfig(), enableInterrupt: a.enableInterrupt, - turnConfig: a.turnConfig, + turnConfig: normalizeTurnConfig(a.turnConfig), visionEnabled: a.visionEnabled, visionModelResourceId: a.visionModelResourceId ?? "", toolIds: a.toolIds ?? [], @@ -711,7 +741,7 @@ export function AssistantPage(props: AssistantPageProps) { name: workflowName, graph: workflowGraph, settings: workflowSettings, - dynamicVariableDefinitions: workflowDynamicVariableDefinitions, + dynamicVariableDefinitions: effectiveWorkflowDynamicVariableDefinitions, }), ); } @@ -756,7 +786,7 @@ export function AssistantPage(props: AssistantPageProps) { asr: a.modelResourceIds.ASR ?? "", voice: a.modelResourceIds.TTS ?? "", enableInterrupt: a.enableInterrupt, - turnConfig: a.turnConfig, + turnConfig: normalizeTurnConfig(a.turnConfig), }; setDifyForm(next); return next; @@ -786,7 +816,7 @@ export function AssistantPage(props: AssistantPageProps) { asr: a.modelResourceIds.ASR ?? "", voice: a.modelResourceIds.TTS ?? "", enableInterrupt: a.enableInterrupt, - turnConfig: a.turnConfig, + turnConfig: normalizeTurnConfig(a.turnConfig), }; setFastGptForm(next); return next; @@ -818,7 +848,7 @@ export function AssistantPage(props: AssistantPageProps) { asr: a.modelResourceIds.ASR ?? "", voice: a.modelResourceIds.TTS ?? "", enableInterrupt: a.enableInterrupt, - turnConfig: a.turnConfig, + turnConfig: normalizeTurnConfig(a.turnConfig), visionEnabled: a.visionEnabled, visionModelResourceId: a.visionModelResourceId ?? "", }; @@ -872,21 +902,29 @@ export function AssistantPage(props: AssistantPageProps) { assistant.knowledgeRetrievalConfig.scoreThreshold, }, globalPrompt: graph.settings?.globalPrompt ?? "", - allowInterrupt: assistant.enableInterrupt, - turnConfig: assistant.turnConfig, + allowInterrupt: + graph.settings?.enableInterrupt ?? assistant.enableInterrupt, + turnConfig: normalizeTurnConfig( + graph.settings?.turnConfig ?? assistant.turnConfig, + ), }; setWorkflowName(assistant.name); setWorkflowGraph(graph); setWorkflowSettings(wfSettings); + const dynamicVariableDefinitions = + activeWorkflowDynamicVariableDefinitions( + graph, + assistant.dynamicVariableDefinitions ?? {}, + ); setWorkflowDynamicVariableDefinitions( - assistant.dynamicVariableDefinitions ?? {}, + dynamicVariableDefinitions, ); setSavedSnapshot( JSON.stringify({ name: assistant.name, graph, settings: wfSettings, - dynamicVariableDefinitions: assistant.dynamicVariableDefinitions ?? {}, + dynamicVariableDefinitions, }), ); } @@ -941,7 +979,7 @@ export function AssistantPage(props: AssistantPageProps) { knowledgeRetrievalConfig: workflowSettings.knowledgeRetrievalConfig, toolIds: workflowSettings.toolIds, graph: workflowGraph as unknown as Record, - dynamicVariableDefinitions: workflowDynamicVariableDefinitions, + dynamicVariableDefinitions: effectiveWorkflowDynamicVariableDefinitions, }), ); } @@ -961,7 +999,8 @@ export function AssistantPage(props: AssistantPageProps) { name: workflowName, graph: workflowGraph, settings: workflowSettings, - dynamicVariableDefinitions: workflowDynamicVariableDefinitions, + dynamicVariableDefinitions: + effectiveWorkflowDynamicVariableDefinitions, }) : null; const dirty = @@ -1473,7 +1512,9 @@ export function AssistantPage(props: AssistantPageProps) { hasUnsavedChanges={dirty} onNodeActive={setActiveNodeId} dynamicVariablesEnabled - dynamicVariableDefinitions={workflowDynamicVariableDefinitions} + dynamicVariableDefinitions={ + effectiveWorkflowDynamicVariableDefinitions + } /> } activeNodeId={activeNodeId} @@ -1492,7 +1533,7 @@ export function AssistantPage(props: AssistantPageProps) { @@ -1872,80 +1913,6 @@ export function AssistantPage(props: AssistantPageProps) {
- -
-
updateForm("runtimeMode", "pipeline")} - onKeyDown={(event) => { - if (event.key === "Enter" || event.key === " ") { - event.preventDefault(); - updateForm("runtimeMode", "pipeline"); - } - }} - className={[ - "cursor-pointer rounded-xl border p-3.5 text-left transition-colors", - form.runtimeMode === "pipeline" - ? "border-primary bg-primary/5 ring-1 ring-primary" - : "border-hairline bg-canvas-soft hover:border-hairline-strong", - ].join(" ")} - > -
-
-
- -
-
- Pipeline 模式 - -
-
- {form.runtimeMode === "pipeline" && ( - - - - )} -
-
- -
updateForm("runtimeMode", "realtime")} - onKeyDown={(event) => { - if (event.key === "Enter" || event.key === " ") { - event.preventDefault(); - updateForm("runtimeMode", "realtime"); - } - }} - className={[ - "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 模式 - -
-
- {form.runtimeMode === "realtime" && ( - - - - )} -
-
-
- - } title="提示词" @@ -1963,12 +1930,38 @@ export function AssistantPage(props: AssistantPageProps) { /> - {form.runtimeMode === "pipeline" ? ( - } - title="模型配置" - description="从「模型资源」中选择大语言模型、语音识别与语音合成" - > + } + title="开场白" + description="助手与用户首次对话时的开场语" + > + updateForm("greeting", value)} + placeholder="请输入助手开场白" + /> + setDynamicVariablesOpen(true)} + /> + + + } + title="模型配置" + description={ + form.runtimeMode === "pipeline" + ? "选择运行方式，以及大语言模型、语音识别与语音合成资源" + : "选择运行方式；Realtime 模型内置语音识别与语音合成" + } + > + updateForm("runtimeMode", runtimeMode)} + /> + + {form.runtimeMode === "pipeline" ? ( + <> - - ) : ( - } - title="模型配置" - description="当前模式下 ASR 与 TTS 由 Realtime 模型内置完成" - > + + ) : ( - - )} - - } - title="开场白" - description="助手与用户首次对话时的开场语" - > - updateForm("greeting", value)} - placeholder="请输入助手开场白" - /> - setDynamicVariablesOpen(true)} - /> + )} {form.runtimeMode === "pipeline" && ( @@ -2206,10 +2178,23 @@ function DebugDrawer({ >({}); const recording = preview.status === "connecting" || preview.status === "connected"; - const dynamicVariableEntries = Object.entries(dynamicVariableDefinitions); + const displayedDefinitions = { ...dynamicVariableDefinitions }; + for (const [name, value] of Object.entries(preview.sessionVariables)) { + displayedDefinitions[name] ??= { + type: + typeof value === "number" + ? "number" + : typeof value === "boolean" + ? "boolean" + : "string", + required: false, + default: null, + }; + } + const dynamicVariableEntries = Object.entries(displayedDefinitions); const resolvedDynamicVariables: Record = {}; let dynamicVariablesError = ""; - for (const [name, definition] of dynamicVariableEntries) { + for (const [name, definition] of Object.entries(dynamicVariableDefinitions)) { const value = dynamicVariableValues[name] ?? definition.default; if (value === null || value === undefined || value === "") { if (definition.required && !dynamicVariablesError) { @@ -2265,7 +2250,8 @@ function DebugDrawer({ )} @@ -2330,12 +2316,14 @@ function DebugDrawer({ function DynamicVariableValuesPopover({ entries, values, - disabled, + sessionValues, + readOnly, onChange, }: { entries: [string, DynamicVariableDefinition][]; values: Record; - disabled: boolean; + sessionValues: Record; + readOnly: boolean; onChange: React.Dispatch< React.SetStateAction> >; @@ -2354,10 +2342,9 @@ function DynamicVariableValuesPopover({

- 这些值只用于下一次调试会话，不会修改助手配置。 + {readOnly + ? "当前值会在 Action 或工具更新变量后实时刷新。" + : "这些值只用于下一次调试会话，不会修改助手配置。"}

@@ -2386,11 +2375,14 @@ function DynamicVariableValuesPopover({ 当前没有会话变量

- 在提示词或开场白中添加 {"{{variable_name}}"} 后，可在这里设置调试值。 + 在工作流提示词、节点话术、边条件或 Action 中引用变量后， + 可在这里设置调试值。

) : entries.map(([name, definition]) => { - const value = values[name] ?? definition.default ?? ""; + const value = readOnly + ? sessionValues[name] ?? definition.default ?? "" + : values[name] ?? definition.default ?? ""; return (