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

feat(workflow): enhance message stages and image routing

Browse Source
This commit is contained in:
Xin Wang
2026-08-03 12:38:02 +08:00
parent b0991f239e
commit 4c43e167db
13 changed files with 593 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/services/brains/prompt_brain.py
View File

@@ -33,6 +33,7 @@ from services.action_runtime import (
)
from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction
from services.fixed_speech import FixedSpeechOutput
from services.message_policy import MESSAGE_CONFIRMATION
from services.message_stage import (
MessageDisplaySpec,
MessageStageRunner,
@@ -271,7 +272,7 @@ class PromptBrain(BaseBrain):
)
).strip(),
),
require_confirmation=True,
completion_policy=MESSAGE_CONFIRMATION,
)
async def _publish_opening_outcome(

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

@@ -43,6 +43,12 @@ from services.action_runtime import (
)
from services.action_stage import ActionStageRunner, ActionStageSpec, StageAction
from services.knowledge import search as search_knowledge
from services.message_policy import (
MESSAGE_COMPLETION_POLICIES,
MESSAGE_CONFIRMATION,
MESSAGE_INTERRUPTIBLE,
MESSAGE_PLAYBACK,
)
from services.message_stage import (
MessageDisplaySpec,
MessageStageResult,
@@ -332,13 +338,33 @@ class WorkflowBrain(BaseBrain):
user_message: dict[str, Any] | None = None,
) -> bool:
"""Serialized implementation so one user turn cannot transition twice."""
self.record_user_message(content)
self._state.begin_user_turn(content)
manager = self._require_manager()
current = self._state.current_node_id
if not current:
return True
continuation = self._pending_message
if (
self._engine.node_type(current) == "message"
and continuation is not None
and continuation.node_id == current
):
if self._message_completion_policy(current) != MESSAGE_INTERRUPTIBLE:
# Protected Message stages keep their playback/confirmation gate.
# Normal transports reject this input before it reaches the brain;
# this guard also covers programmatic context injections.
return True
self.record_user_message(content)
self._state.begin_user_turn(content)
return await self._interrupt_message_continuation(
continuation,
content=content,
user_message=user_message,
)
self.record_user_message(content)
self._state.begin_user_turn(content)
self._state.status = WorkflowStatus.ROUTING
decision = await self._edge_evaluator.evaluate(
current,
@@ -366,6 +392,79 @@ class WorkflowBrain(BaseBrain):
return await self._continue_current_node_after_no_transition(current)
async def _interrupt_message_continuation(
self,
continuation: _MessageContinuation,
*,
content: str,
user_message: dict[str, Any] | None,
) -> bool:
"""Finish an interruptible Message once and forward the user turn."""
if self._pending_message is not continuation:
return True
self._pending_message = None
task = continuation.task
if task is not None and task is not asyncio.current_task():
task.cancel()
runtime = self._require_runtime()
if runtime.set_input_enabled is not None:
runtime.set_input_enabled(True)
await self._emit_trace(
"message_interrupted",
nodeId=continuation.node_id,
reason="user_input",
)
context_messages = [
dict(message) for message in continuation.context_messages
]
if not self._engine.has_outgoing(continuation.node_id):
self._state.enter(
continuation.node_id,
WorkflowStatus.WAITING_USER,
)
return True
self._state.status = WorkflowStatus.ROUTING
decision = await self._edge_evaluator.evaluate(
continuation.node_id,
current_user_message=user_message,
)
if decision.status == RouteStatus.ERROR:
await self._require_output().emit_error(
decision.error or "工作流路由失败",
node_id=continuation.node_id,
code="workflow_routing_error",
)
self._state.enter(
continuation.node_id,
WorkflowStatus.WAITING_USER,
)
return True
manager = self._require_manager()
if not decision.edge or manager.current_node != continuation.node_id:
self._state.enter(
continuation.node_id,
WorkflowStatus.WAITING_USER,
)
return True
next_config = await self._follow_edge(
decision.edge,
leading_messages=context_messages,
triggering_user_text=content,
triggering_user_message=user_message,
)
await self._activate_node_config(
next_config,
triggering_user_text=content,
)
return True
async def _continue_current_node_after_no_transition(
self,
node_id: str,
@@ -811,7 +910,9 @@ class WorkflowBrain(BaseBrain):
await self._emit_node_active(node_id)
runtime = self._require_runtime()
if runtime.set_input_enabled is not None:
runtime.set_input_enabled(False)
runtime.set_input_enabled(
self._message_completion_policy(node_id) == MESSAGE_INTERRUPTIBLE
)
continuation.task = asyncio.create_task(
self._complete_message_continuation(continuation),
name=f"workflow-message-{node_id}-{continuation.token}",
@@ -967,8 +1068,8 @@ class WorkflowBrain(BaseBrain):
data = self._engine.data(node_id)
runtime = self._require_runtime()
speech = self._store.render(str(data.get("speech") or "")).strip()
show_message = bool(data.get("showMessage", False))
require_confirmation = bool(data.get("requireConfirmation", False))
completion_policy = self._message_completion_policy(node_id)
require_confirmation = completion_policy == MESSAGE_CONFIRMATION
display = (
MessageDisplaySpec(
title=self._store.render(
@@ -981,14 +1082,14 @@ class WorkflowBrain(BaseBrain):
str(data.get("confirmLabel") or "确认")
).strip(),
)
if show_message
if require_confirmation
else None
)
result = await self._message_stages.run(
MessageStageSpec(
speech=speech,
display=display,
require_confirmation=require_confirmation,
completion_policy=completion_policy,
),
speak=lambda content: self._queue_visible_speech(
content,
@@ -1001,7 +1102,8 @@ class WorkflowBrain(BaseBrain):
"message_started",
nodeId=node_id,
hasSpeech=bool(speech),
showsMessage=show_message,
showsMessage=require_confirmation,
completionPolicy=completion_policy,
requiresConfirmation=require_confirmation,
),
)
@@ -1026,6 +1128,13 @@ class WorkflowBrain(BaseBrain):
)
return result
def _message_completion_policy(self, node_id: str) -> str:
value = str(
self._engine.data(node_id).get("completionPolicy")
or MESSAGE_PLAYBACK
)
return value if value in MESSAGE_COMPLETION_POLICIES else MESSAGE_PLAYBACK
def _set_last_action(self, outcome: ActionOutcome) -> None:
legacy_status = {
ActionStatus.SUCCESS: "ok",

20
backend/services/message_policy.py Normal file
View File

@@ -0,0 +1,20 @@
"""Shared completion policies for deterministic Message stages."""
from typing import Literal
MESSAGE_INTERRUPTIBLE = "interruptible"
MESSAGE_PLAYBACK = "playback"
MESSAGE_CONFIRMATION = "confirmation"
MESSAGE_COMPLETION_POLICIES = frozenset(
{
MESSAGE_INTERRUPTIBLE,
MESSAGE_PLAYBACK,
MESSAGE_CONFIRMATION,
}
)
MessageCompletionPolicy = Literal[
"interruptible",
"playback",
"confirmation",
]

32
backend/services/message_stage.py
View File

@@ -7,6 +7,13 @@ from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from services.client_tools import ClientToolError, ClientToolPort
from services.message_policy import (
MESSAGE_COMPLETION_POLICIES,
MESSAGE_CONFIRMATION,
MESSAGE_INTERRUPTIBLE,
MESSAGE_PLAYBACK,
MessageCompletionPolicy,
)
BUILTIN_SHOW_MESSAGE = "show_message"
@@ -30,7 +37,7 @@ class MessageStageSpec:
speech: str = ""
display: MessageDisplaySpec | None = None
require_confirmation: bool = False
completion_policy: MessageCompletionPolicy = MESSAGE_PLAYBACK
@dataclass(frozen=True)
@@ -69,7 +76,15 @@ class MessageStageRunner:
on_started: StartedHook | None = None,
) -> MessageStageResult:
input_setter = set_input_enabled
if input_setter is not None and not input_already_blocked:
if spec.completion_policy not in MESSAGE_COMPLETION_POLICIES:
return MessageStageResult(
succeeded=False,
speech=spec.speech.strip(),
error=f"未知的 Message 完成策略:{spec.completion_policy}",
)
block_input = spec.completion_policy != MESSAGE_INTERRUPTIBLE
require_confirmation = spec.completion_policy == MESSAGE_CONFIRMATION
if block_input and input_setter is not None and not input_already_blocked:
input_setter(False)
result: MessageStageResult | None = None
@@ -78,7 +93,7 @@ class MessageStageRunner:
await on_started()
speech = spec.speech.strip()
if spec.require_confirmation and spec.display is None:
if require_confirmation and spec.display is None:
result = MessageStageResult(
succeeded=False,
speech=speech,
@@ -106,7 +121,7 @@ class MessageStageRunner:
# audio completion future, so the user can continue immediately.
if (
playback_completion is not None
and not spec.require_confirmation
and not require_confirmation
):
await playback_completion
@@ -133,7 +148,7 @@ class MessageStageRunner:
or (not result.succeeded and release_input_on_failure)
)
)
if input_setter is not None and should_release:
if block_input and input_setter is not None and should_release:
input_setter(True)
async def _show_message(
@@ -152,6 +167,7 @@ class MessageStageRunner:
error="当前运行模式不支持客户端消息",
)
try:
require_confirmation = spec.completion_policy == MESSAGE_CONFIRMATION
response = await self._client_tools.call(
BUILTIN_SHOW_MESSAGE,
{
@@ -164,12 +180,12 @@ class MessageStageRunner:
"style": "primary",
}
],
"dismissible": not spec.require_confirmation,
"dismissible": not require_confirmation,
},
timeout_seconds=3,
wait_for_response=spec.require_confirmation,
wait_for_response=require_confirmation,
response_wait_mode=(
"session" if spec.require_confirmation else "timeout"
"session" if require_confirmation else "timeout"
),
)
except ClientToolError as exc:

38
backend/services/node_specs.py
View File

@@ -6,6 +6,12 @@ from collections import defaultdict, deque
from copy import deepcopy
from typing import Any
from services.message_policy import (
MESSAGE_COMPLETION_POLICIES,
MESSAGE_CONFIRMATION,
MESSAGE_PLAYBACK,
)
SPEC_VERSION = "3"
NODE_TYPES = {"start", "agent", "message", "action", "handoff", "end"}
@@ -191,11 +197,17 @@ def _normalize_action_data(data: dict[str, Any]) -> None:
def _normalize_message_data(data: dict[str, Any]) -> None:
"""Fill the small built-in Message contract used by runtime and editor."""
data.setdefault("speech", "")
data.setdefault("showMessage", False)
data.setdefault("title", "重要提示")
data.setdefault("message", "")
data.setdefault("confirmLabel", "确认")
data.setdefault("requireConfirmation", False)
if "completionPolicy" not in data:
data["completionPolicy"] = (
MESSAGE_CONFIRMATION
if data.get("requireConfirmation") is True
else MESSAGE_PLAYBACK
)
data.pop("requireConfirmation", None)
data.pop("showMessage", None)
def _normalize_settings(settings: dict[str, Any], *, global_prompt: str = "") -> None:
@@ -379,19 +391,19 @@ def validate_graph(graph: dict[str, Any]) -> list[str]:
elif node_type == "message":
data = node.get("data") or {}
speech = data.get("speech")
show_message = data.get("showMessage")
require_confirmation = data.get("requireConfirmation")
completion_policy = data.get("completionPolicy")
if not isinstance(speech, str):
errors.append(f"Message 节点 {node_id} 的播报内容必须是文本")
if not isinstance(show_message, bool):
errors.append(f"Message 节点 {node_id} 的弹窗开关必须是布尔值")
if not isinstance(require_confirmation, bool):
errors.append(f"Message 节点 {node_id} 的确认开关必须是布尔值")
if require_confirmation and show_message is not True:
errors.append(f"Message 节点 {node_id} 等待确认时必须显示弹窗")
if not str(speech or "").strip() and show_message is not True:
errors.append(f"Message 节点 {node_id} 至少需要播报或显示弹窗")
if show_message is True:
if (
not isinstance(completion_policy, str)
or completion_policy not in MESSAGE_COMPLETION_POLICIES
):
errors.append(
f"Message 节点 {node_id} 的完成策略无效:{completion_policy}"
)
if not str(speech or "").strip():
errors.append(f"Message 节点 {node_id} 必须配置播报内容")
if completion_policy == MESSAGE_CONFIRMATION:
title = data.get("title")
message = data.get("message")
confirm_label = data.get("confirmLabel")

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

@@ -173,6 +173,35 @@ def _image_data_uri(frame: UserImageRawFrame) -> str:
return f"data:image/jpeg;base64,{encoded}"
def _multimodal_user_input_frame(
image_frame: UserImageRawFrame,
prompt_text: str,
) -> LLMMessagesAppendFrame:
"""Submit an explicit camera capture through the normal user-turn path.
``UserImageRawFrame`` is appended by Pipecat's assistant-side aggregator,
which pushes context upstream directly into the LLM. Workflow routing sits
on the downstream user-turn path, so queuing the raw frame would let the
Agent see the image while skipping edge evaluation. A standard multimodal
user message keeps text and image turns on the same routing path.
"""
return LLMMessagesAppendFrame(
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt_text},
{
"type": "image_url",
"image_url": {"url": _image_data_uri(image_frame)},
},
],
}
],
run_llm=True,
)
async def _analyze_image_with_vision_model(
cfg: AssistantConfig,
frame: UserImageRawFrame,
@@ -783,10 +812,12 @@ async def run_pipeline(
raise ValueError("等待摄像头视频帧超时") from exc
if native_vision:
image_frame.text = value.prompt_text
image_frame.append_to_context = True
image_frame.request = None
await worker.queue_frame(image_frame)
input_frame = await asyncio.to_thread(
_multimodal_user_input_frame,
image_frame,
value.prompt_text,
)
await worker.queue_frame(input_frame)
return
try:

130
backend/tests/test_brains.py
View File

@@ -1021,7 +1021,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
"type": "message",
"data": {
"speech": "请问您怎么称呼？",
"showMessage": False,
"completionPolicy": "playback",
},
},
{
@@ -1457,11 +1457,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
"type": "message",
"data": {
"speech": "请先确认 {{customer}} 的重要信息。",
"showMessage": True,
"title": "重要提示",
"message": "请核对客户信息。",
"confirmLabel": "确认",
"requireConfirmation": True,
"completionPolicy": "confirmation",
},
},
],
@@ -1585,6 +1584,131 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase):
self.assertTrue(result.succeeded)
self.assertEqual(input_states, [False, True])
async def test_interruptible_message_forwards_multimodal_input_once(self):
graph = {
"specVersion": 3,
"settings": {},
"nodes": [
{"id": "start", "type": "start", "data": {}},
{
"id": "message",
"type": "message",
"data": {
"speech": "请按提示操作，也可以直接告诉我需求。",
"completionPolicy": "interruptible",
},
},
{
"id": "agent",
"type": "agent",
"data": {
"prompt": "处理用户输入",
"contextPolicy": "fresh",
},
},
],
"edges": [
{
"id": "start-message",
"source": "start",
"target": "message",
"data": {"mode": "always"},
},
{
"id": "message-agent",
"source": "message",
"target": "agent",
"data": {"mode": "always"},
},
],
}
brain = WorkflowBrain(graph)
queued = []
input_states = []
class PlaybackCallEnd(FakeCallEnd):
def __init__(self):
super().__init__()
self.completion = None
def track_speech(self):
self.completion = asyncio.get_running_loop().create_future()
return self.completion
class FakeManager:
def __init__(self):
self.current_node = None
self.configs = []
async def initialize(self, config):
self.current_node = config["name"]
self.configs.append(config)
async def set_node_from_config(self, config):
self.current_node = config["name"]
self.configs.append(config)
async def queue_frame(frame):
queued.append(frame)
call_end = PlaybackCallEnd()
manager = FakeManager()
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=call_end,
set_input_enabled=input_states.append,
)
brain._manager = manager
await brain.on_connected()
await asyncio.sleep(0)
self.assertEqual(manager.current_node, "message")
self.assertIsNotNone(call_end.completion)
self.assertTrue(all(input_states))
image_message = {
"role": "user",
"content": [
{"type": "text", "text": "已发送一张图片"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,AA=="},
},
],
}
await brain.on_user_turn_end(
"已发送一张图片",
user_message=image_message,
)
await asyncio.sleep(0)
self.assertEqual(manager.current_node, "agent")
self.assertEqual(
[config["name"] for config in manager.configs].count("agent"),
1,
)
self.assertTrue(call_end.completion.cancelled())
self.assertTrue(all(input_states))
self.assertEqual(
manager.configs[-1]["task_messages"],
[image_message],
)
self.assertEqual(
sum(isinstance(frame, LLMRunFrame) for frame in queued),
1,
)
self.assertTrue(
any(
isinstance(frame, OutputTransportMessageUrgentFrame)
and frame.message.get("event") == "message_interrupted"
for frame in queued
)
)
async def test_message_between_agents_resumes_after_playback(self):
graph = {
"specVersion": 3,

29
backend/tests/test_user_input.py
View File

@@ -1,5 +1,7 @@
import unittest
from pipecat.frames.frames import LLMMessagesAppendFrame, UserImageRawFrame
from services.pipecat.pipeline import _multimodal_user_input_frame
from services.pipecat.processors import UserInputError, parse_user_input
@@ -52,6 +54,33 @@ class UserInputParserTests(unittest.TestCase):
}
)
def test_native_image_uses_the_standard_multimodal_user_turn_path(self):
image = UserImageRawFrame(
image=bytes([220, 40, 40] * 16 * 16),
size=(16, 16),
format="RGB",
)
frame = _multimodal_user_input_frame(
image,
"请根据用户刚提交的图片进行回复。",
)
self.assertIsInstance(frame, LLMMessagesAppendFrame)
self.assertTrue(frame.run_llm)
self.assertEqual(frame.messages[0]["role"], "user")
content = frame.messages[0]["content"]
self.assertEqual(
content[0],
{"type": "text", "text": "请根据用户刚提交的图片进行回复。"},
)
self.assertEqual(content[1]["type"], "image_url")
self.assertTrue(
content[1]["image_url"]["url"].startswith(
"data:image/jpeg;base64,"
)
)
if __name__ == "__main__":
unittest.main()

41
backend/tests/test_workflow_v3.py
View File

@@ -200,32 +200,53 @@ class WorkflowGraphTests(unittest.TestCase):
message = next(
node for node in normalized["nodes"] if node["type"] == "message"
)
self.assertFalse(message["data"]["showMessage"])
self.assertFalse(message["data"]["requireConfirmation"])
self.assertEqual(message["data"]["completionPolicy"], "playback")
self.assertNotIn("requireConfirmation", message["data"])
self.assertNotIn("showMessage", message["data"])
self.assertEqual(message["data"]["confirmLabel"], "确认")
message["data"].update(
{
"speech": "",
"showMessage": True,
"title": "重要提示",
"message": "",
"confirmLabel": "确认",
"requireConfirmation": True,
"completionPolicy": "confirmation",
}
)
errors = validate_graph(normalized)
self.assertTrue(any("弹窗消息必须为" in error for error in errors))
self.assertTrue(any("必须配置播报内容" in error for error in errors))
message["data"].update(
{
"speech": "请确认",
"showMessage": False,
"requireConfirmation": True,
}
{"speech": "", "completionPolicy": "playback"}
)
errors = validate_graph(normalized)
self.assertTrue(any("等待确认时必须显示弹窗" in error for error in errors))
self.assertTrue(any("必须配置播报内容" in error for error in errors))
message["data"]["completionPolicy"] = "unknown"
errors = validate_graph(normalized)
self.assertTrue(any("完成策略无效" in error for error in errors))
legacy = valid_graph()
legacy["nodes"].append(
{
"id": "legacy-message",
"type": "message",
"data": {
"speech": "请确认",
"showMessage": True,
"title": "提示",
"message": "请确认",
"confirmLabel": "确认",
"requireConfirmation": True,
},
}
)
legacy_message = normalize_graph(legacy)["nodes"][-1]["data"]
self.assertEqual(legacy_message["completionPolicy"], "confirmation")
self.assertNotIn("requireConfirmation", legacy_message)
self.assertNotIn("showMessage", legacy_message)
def test_voice_resource_creates_isolated_runtime_config(self):
base = AssistantConfig(type="workflow", asr="default", voice="default")