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

fix: honor strict workflow agent entry mode

Browse Source
This commit is contained in:
Xin Wang
2026-08-03 16:34:26 +08:00
parent 06719e4801
commit 67389fc5a1
5 changed files with 103 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
backend/services/brains/workflow_brain.py
View File

@@ -562,16 +562,24 @@ class WorkflowBrain(BaseBrain):
return
await self._emit_node_active(node_id)
data = self._engine.data(node_id)
entry_mode = str(data.get("entryMode") or "wait_user")
should_run = entry_mode == "generate" or bool(triggering_user_text)
if should_run:
if self._agent_runs_on_entry(node_id):
self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT)
await self._require_runtime().queue_frame(LLMRunFrame())
return
# The turn that selected this node belongs to the previous stage.
# A strict wait must not leave it pending until the next user message.
if triggering_user_text:
self._state.consume_user_turn()
self._state.enter(node_id, WorkflowStatus.WAITING_USER)
def _agent_runs_on_entry(self, node_id: str) -> bool:
"""Only explicit generate mode may start a reply on node entry."""
entry_mode = str(
self._engine.data(node_id).get("entryMode") or "wait_user"
)
return entry_mode == "generate"
async def _activate_node_config(
self,
node_config: NodeConfig,
@@ -708,10 +716,7 @@ class WorkflowBrain(BaseBrain):
return configured
if node_type != "agent":
return node_config
entry_mode = str(
self._engine.data(node_id).get("entryMode") or "wait_user"
)
should_run = entry_mode == "generate" or bool(triggering_user_text)
should_run = self._agent_runs_on_entry(node_id)
configured = dict(node_config)
configured["respond_immediately"] = should_run
configured["pre_actions"] = [
@@ -719,6 +724,8 @@ class WorkflowBrain(BaseBrain):
"type": ConfiguredFlowManager.ENTRY_ACTION_TYPE,
"node_id": node_id,
"should_run": should_run,
"consume_triggering_turn": bool(triggering_user_text)
and not should_run,
"handler": self._activate_from_flow_transition,
}
]
@@ -735,6 +742,8 @@ class WorkflowBrain(BaseBrain):
if action.get("should_run"):
self._state.enter(node_id, WorkflowStatus.RUNNING_AGENT)
else:
if action.get("consume_triggering_turn"):
self._state.consume_user_turn()
self._state.enter(node_id, WorkflowStatus.WAITING_USER)
def _knowledge_function(self, node_id: str) -> FlowsFunctionSchema | None:

83
backend/tests/test_brains.py
View File

@@ -867,6 +867,66 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual(brain._flow_tool(timeout_tool, "start").timeout_secs, 7.0)
self.assertIsNone(brain._flow_tool(session_tool, "start").timeout_secs)
async def test_flow_transition_honors_strict_agent_entry_mode(self):
brain = WorkflowBrain(
{
"specVersion": 3,
"settings": {},
"nodes": [
{"id": "start", "type": "start", "data": {}},
{
"id": "waiting",
"type": "agent",
"data": {"entryMode": "wait_user"},
},
{
"id": "immediate",
"type": "agent",
"data": {"entryMode": "generate"},
},
],
"edges": [],
}
)
async def queue_frame(_frame):
pass
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._state.begin_user_turn("完成上一阶段")
waiting_config = brain._flow_managed_transition_config(
{"name": "waiting"},
triggering_user_text="完成上一阶段",
)
waiting_action = waiting_config["pre_actions"][0]
self.assertFalse(waiting_config["respond_immediately"])
self.assertTrue(waiting_action["consume_triggering_turn"])
await waiting_action["handler"](waiting_action, SimpleNamespace())
self.assertEqual(brain._state.status, WorkflowStatus.WAITING_USER)
self.assertIsNone(brain._state.pending_user_turn)
brain._state.begin_user_turn("请立即处理")
immediate_config = brain._flow_managed_transition_config(
{"name": "immediate"},
triggering_user_text="请立即处理",
)
immediate_action = immediate_config["pre_actions"][0]
self.assertTrue(immediate_config["respond_immediately"])
self.assertFalse(immediate_action["consume_triggering_turn"])
await immediate_action["handler"](immediate_action, SimpleNamespace())
self.assertEqual(brain._state.status, WorkflowStatus.RUNNING_AGENT)
self.assertIsNotNone(brain._state.pending_user_turn)
async def test_session_update_refreshes_current_agent_without_routing(self):
cfg = prepare_dynamic_config(
AssistantConfig(
@@ -1614,6 +1674,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
"data": {
"prompt": "处理用户输入",
"contextPolicy": "fresh",
"entryMode": "generate",
},
},
],
@@ -1874,6 +1935,9 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
for message in manager.configs[-1]["task_messages"]
)
)
runs_before_second_transition = sum(
isinstance(frame, LLMRunFrame) for frame in queued
)
# The user-turn processor must return while the second Message is
# still waiting for its transport playback boundary.
@@ -1892,7 +1956,12 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
if manager.current_node == "agent2":
break
self.assertEqual(manager.current_node, "agent2")
self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued))
self.assertEqual(
sum(isinstance(frame, LLMRunFrame) for frame in queued),
runs_before_second_transition,
)
self.assertEqual(brain._state.status, WorkflowStatus.WAITING_USER)
self.assertIsNone(brain._state.pending_user_turn)
self.assertEqual(
manager.configs[-1]["task_messages"],
[
@@ -1906,7 +1975,6 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
],
)
await brain.on_assistant_text_end("agent2-turn", "信息确认完成", False)
await brain.on_user_turn_end("结束通话")
self.assertEqual(manager.current_node, "end")
self.assertTrue(call_end.finished)
@@ -2031,7 +2099,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
)
)
async def test_start_llm_conditions_wait_for_and_route_first_user_turn(self):
async def test_waiting_agent_does_not_reply_to_transitioning_user_turn(self):
brain = WorkflowBrain(
{
"specVersion": 3,
@@ -2161,9 +2229,16 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
image_message,
manager.config["task_messages"],
)
self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued))
self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in queued))
self.assertEqual(brain._state.status, WorkflowStatus.WAITING_USER)
self.assertIsNone(brain._state.pending_user_turn)
self.assertIn("我想吃饭", brain._store.values["system__conversation_history"])
await brain.on_user_turn_end("我要一份米饭")
self.assertTrue(any(isinstance(frame, LLMRunFrame) for frame in queued))
self.assertEqual(brain._state.status, WorkflowStatus.RUNNING_AGENT)
async def test_start_expression_condition_also_waits_for_user_turn(self):
brain = WorkflowBrain(
{

2
frontend/src/components/workflow/GenericNode.tsx
View File

@@ -42,7 +42,7 @@ export function GenericNode({ id, type, data, selected }: NodeProps) {
.toString()
.trim();
const entryModeLabel = {
wait_user: "等待用户",
wait_user: "等待下一轮",
generate: "立即回复",
}[nodeData.entryMode ?? "wait_user"];
const inheritsGlobal = nodeData.inheritGlobalConfig !== false;

6
frontend/src/components/workflow/panels/AgentNodePanel.tsx
View File

@@ -151,14 +151,14 @@ export function AgentNodePanel({
label="进入节点时"
value={(draft.entryMode as string) || "wait_user"}
options={[
{ value: "wait_user", label: "等待用户说话（默认）" },
{ value: "generate", label: "立即让 LLM 回复" },
{ value: "wait_user", label: "等待下一轮用户输入（默认）" },
{ value: "generate", label: "进入后立即回复" },
]}
onChange={(value) => set("entryMode", value || "wait_user")}
allowNone={false}
/>
<p className="text-xs leading-5 text-muted-foreground">
固定播报、客户端弹窗和确认门禁请使用独立的 Message 节点。
等待模式不会回复触发跳转的当前输入；立即回复会处理该输入。固定播报、客户端弹窗和确认门禁请使用独立的 Message 节点。
</p>
</SectionCard>

6
frontend/src/components/workflow/panels/NodeSettingsPanel.tsx
View File

@@ -169,14 +169,14 @@ export function NodeSettingsPanel({
label="进入节点时"
value={(draft.entryMode as string) || "wait_user"}
options={[
{ value: "wait_user", label: "等待用户说话（默认）" },
{ value: "generate", label: "立即让 LLM 回复" },
{ value: "wait_user", label: "等待下一轮用户输入（默认）" },
{ value: "generate", label: "进入后立即回复" },
]}
onChange={(value) => set("entryMode", value || "wait_user")}
allowNone={false}
/>
<p className="text-xs leading-5 text-muted-soft">
固定播报、客户端弹窗和确认门禁请使用独立的 Message 节点。
等待模式不会回复触发跳转的当前输入；立即回复会处理该输入。固定播报、客户端弹窗和确认门禁请使用独立的 Message 节点。
</p>
<div className="flex flex-col gap-2">
<label className="text-sm font-medium text-foreground">可用工具</label>