diff --git a/AGENTS.md b/AGENTS.md index 287e0bd..a072005 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,6 @@ -你编写具有可维护性、高可读性、模块化的的代码，尽量不去对pipecat框架本身修改 +你编写具有可维护性、高可读性、可扩展性、模块化的的代码，尽量不去对pipecat框架本身修改 以MVP构建为目标 +适合CS的本科学生阅读修改 编写代码之前先用易于理解的语言说清楚思路 界面设计要参考 frontend/DESIGN.md \ No newline at end of file diff --git a/backend/services/brains/base.py b/backend/services/brains/base.py index 3f3e3d2..ae2db18 100644 --- a/backend/services/brains/base.py +++ b/backend/services/brains/base.py @@ -19,6 +19,20 @@ from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameProcessor +GREETING_CONTEXT_MARKER = "[会话事实：助手开场白已播放]" + + +def greeting_context_message(greeting: str) -> dict[str, str] | None: + """Represent spoken greeting without starting model history as assistant.""" + content = greeting.strip() + if not content: + return None + return { + "role": "system", + "content": f"{GREETING_CONTEXT_MARKER}\n{content}", + } + + @dataclass(frozen=True) class BrainSpec: """Static capabilities used by validation and runtime dispatch.""" @@ -86,6 +100,27 @@ class BaseBrain: async def on_connected(self) -> None: """Handle a connected client after the common greeting is queued.""" + def prepare_greeting_context( + self, + greeting: str, + context: LLMContext, + ) -> dict[str, str] | None: + """Add a provider-safe fact describing the greeting to local context.""" + if not self.spec.owns_context: + return None + message = greeting_context_message(greeting) + if message is None: + return None + messages = context.get_messages() + messages[:] = [ + item + for item in messages + if GREETING_CONTEXT_MARKER not in str(item.get("content") or "") + ] + insert_at = 1 if messages and messages[0].get("role") == "system" else 0 + messages.insert(insert_at, message) + return message + async def on_client_ready(self) -> None: """Replay client-visible state after its app message channel is ready.""" @@ -129,6 +164,12 @@ class Brain(Protocol): async def on_connected(self) -> None: ... + def prepare_greeting_context( + self, + greeting: str, + context: LLMContext, + ) -> dict[str, str] | None: ... + async def on_client_ready(self) -> None: ... def record_user_message(self, content: str) -> None: ... diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 50b78df..cabde71 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -70,6 +70,9 @@ class WorkflowBrain(BaseBrain): self._manager: FlowManager | None = None self._router = WorkflowLLMRouter(cfg or AssistantConfig(type="workflow")) self._ended = False + self._greeting_context_message: dict[str, str] | None = None + self._client_ready = False + self._pending_visible_speech_events: list[dict[str, Any]] = [] async def greeting(self, cfg: AssistantConfig) -> str: return self._engine.greeting(self._store) or cfg.greeting @@ -91,6 +94,9 @@ class WorkflowBrain(BaseBrain): self._tools = ToolExecutor(self._store) self._tool_by_id = {tool.id: tool for tool in cfg.tools} self._router = WorkflowLLMRouter(cfg) + self._greeting_context_message = None + self._client_ready = False + self._pending_visible_speech_events = [] self._manager = FlowManager( worker=runtime.worker, llm=runtime.llm, @@ -100,6 +106,15 @@ class WorkflowBrain(BaseBrain): ) self._manager.state["variables"] = self._store.values + def prepare_greeting_context( + self, + greeting: str, + context: LLMContext, + ) -> dict[str, str] | None: + message = super().prepare_greeting_context(greeting, context) + self._greeting_context_message = deepcopy(message) if message else None + return message + async def on_connected(self) -> None: await self._emit_node_active(self._engine.start_id) await self._emit_variables( @@ -125,6 +140,13 @@ class WorkflowBrain(BaseBrain): async def on_client_ready(self) -> None: """Replay state that may have been emitted before WebRTC data was ready.""" + self._client_ready = True + pending_speech_events = self._pending_visible_speech_events + self._pending_visible_speech_events = [] + for message in pending_speech_events: + await self._require_runtime().queue_frame( + OutputTransportMessageUrgentFrame(message=message) + ) current_node = ( str(self._manager.current_node) if self._manager and self._manager.current_node @@ -273,11 +295,15 @@ class WorkflowBrain(BaseBrain): } ) - def _agent_config(self, node_id: str) -> NodeConfig: + def _agent_config( + self, + node_id: str, + leading_messages: list[dict[str, str]] | None = None, + ) -> 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 "")) - fixed_reply_messages = ( + fixed_reply_messages: list[dict[str, str]] = ( [{"role": "assistant", "content": entry_speech}] if entry_mode == "fixed_speech" and entry_speech else [] @@ -287,6 +313,16 @@ class WorkflowBrain(BaseBrain): if data.get("contextPolicy") == "fresh" else ContextStrategy.APPEND ) + greeting_messages = ( + [deepcopy(self._greeting_context_message)] + if strategy == ContextStrategy.RESET and self._greeting_context_message + else [] + ) + task_messages = [ + *greeting_messages, + *(leading_messages or []), + *fixed_reply_messages, + ] stage = self._engine.agent_stage_config(node_id) functions: list[FlowsFunctionSchema] = [] for tool_id in stage.tool_ids: @@ -302,7 +338,7 @@ class WorkflowBrain(BaseBrain): # 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, + "task_messages": task_messages, "functions": functions, "context_strategy": ContextStrategyConfig(strategy=strategy), "respond_immediately": entry_mode == "generate", @@ -339,26 +375,32 @@ class WorkflowBrain(BaseBrain): 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(), - "source": source, - **({"nodeId": node_id} if node_id else {}), - } + transcript_message = { + "type": "transcript", + "role": "assistant", + "content": content, + "timestamp": time_now_iso8601(), + "source": source, + **({"nodeId": node_id} if node_id else {}), + } + if self._client_ready: + await runtime.queue_frame( + OutputTransportMessageUrgentFrame(message=transcript_message) ) - ) + else: + self._pending_visible_speech_events.append(transcript_message) await runtime.queue_frame(TTSSpeakFrame(content, append_to_context=False)) - def _passive_node_config(self, node_id: str) -> NodeConfig: + def _passive_node_config( + self, + node_id: str, + task_messages: list[dict[str, str]] | None = None, + ) -> 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()), - "task_messages": [], + "task_messages": list(task_messages or []), "functions": [], "context_strategy": ContextStrategyConfig(strategy=ContextStrategy.APPEND), "respond_immediately": False, @@ -440,20 +482,39 @@ class WorkflowBrain(BaseBrain): ) async def _follow_edge(self, edge: dict) -> NodeConfig: + leading_messages: list[dict[str, str]] = [] speech = self._engine.edge_transition_speech(edge) if speech: - await self._queue_visible_speech(self._store.render(speech)) - return await self._resolve_path(str(edge.get("target") or "")) + content = self._store.render(speech).strip() + if content: + await self._queue_visible_speech( + content, + source="workflow-edge-transition", + node_id=str(edge.get("target") or "") or None, + ) + leading_messages.append( + {"role": "assistant", "content": content} + ) + return await self._resolve_path( + str(edge.get("target") or ""), + leading_messages=leading_messages, + ) - async def _resolve_path(self, node_id: str) -> NodeConfig: + async def _resolve_path( + self, + node_id: str, + *, + leading_messages: list[dict[str, str]] | None = None, + ) -> NodeConfig: + context_messages = list(leading_messages or []) for _ in range(MAX_AUTOMATIC_HOPS): node_type = self._engine.node_type(node_id) if node_type == "agent": await self._apply_agent_stage(node_id) - return self._agent_config(node_id) + return self._agent_config(node_id, context_messages) if node_type == "end": await self._enter_end(node_id) - return self._passive_node_config(node_id) + return self._passive_node_config(node_id, context_messages) if node_type == "action": await self._enter_action(node_id) elif node_type == "handoff": @@ -463,7 +524,7 @@ class WorkflowBrain(BaseBrain): else: raise RuntimeError(f"工作流指向未知节点:{node_id}") if not self._engine.has_outgoing(node_id): - return self._passive_node_config(node_id) + return self._passive_node_config(node_id, context_messages) edge = self._engine.deterministic_edge( node_id, self._store, @@ -473,7 +534,17 @@ class WorkflowBrain(BaseBrain): raise RuntimeError(f"自动节点 {node_id} 没有命中的表达式边或默认边") speech = self._engine.edge_transition_speech(edge) if speech: - await self._queue_visible_speech(self._store.render(speech)) + content = self._store.render(speech).strip() + if content: + target_id = str(edge.get("target") or "") + await self._queue_visible_speech( + content, + source="workflow-edge-transition", + node_id=target_id or None, + ) + context_messages.append( + {"role": "assistant", "content": content} + ) node_id = str(edge.get("target") or "") raise RuntimeError("工作流连续自动跳转超过安全上限") diff --git a/backend/services/pipecat/pipeline_events.py b/backend/services/pipecat/pipeline_events.py index ca7b7c1..074c170 100644 --- a/backend/services/pipecat/pipeline_events.py +++ b/backend/services/pipecat/pipeline_events.py @@ -32,6 +32,7 @@ def bind_cascade_pipeline_events( pending_text_inputs: list[str] = [] greeting_transcript_sent = False + greeting_timestamp = "" async def queue_transcript(role: str, content: str, timestamp: str) -> None: if not content: @@ -122,11 +123,16 @@ def bind_cascade_pipeline_events( nonlocal greeting_transcript_sent if greeting and not greeting_transcript_sent: greeting_transcript_sent = True - await queue_transcript("assistant", greeting, time_now_iso8601()) + await queue_transcript( + "assistant", + greeting, + greeting_timestamp or time_now_iso8601(), + ) await brain.on_client_ready() @transport.event_handler("on_client_connected") async def on_client_connected(_transport, _client): + nonlocal greeting_timestamp if vision_enabled: try: vision_state["client_id"] = get_transport_client_id( @@ -140,8 +146,11 @@ def bind_cascade_pipeline_events( except Exception as exc: # noqa: BLE001 - media availability is optional logger.warning(f"视觉理解摄像头捕获初始化失败: {exc}") if greeting: + # Preserve the actual playback order. The transcript is delivered + # later on client-ready, but the preview sorts by this timestamp. + greeting_timestamp = greeting_timestamp or time_now_iso8601() if brain.spec.owns_context: - context.add_message({"role": "assistant", "content": greeting}) + brain.prepare_greeting_context(greeting, context) await worker.queue_frame( TTSSpeakFrame(greeting, append_to_context=False) ) diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index 507cea6..a0df965 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, + LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMRunFrame, LLMTextFrame, @@ -19,6 +20,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection from schemas import AssistantUpsert, REALTIME_CAPABLE_TYPES from services.brains import BrainRuntime, SPECS, build_brain +from services.brains.base import GREETING_CONTEXT_MARKER from services.brains.dify_llm import ( DifyLLMService, last_user_text, @@ -227,6 +229,19 @@ class PromptBrainTests(unittest.IsolatedAsyncioTestCase): brain = build_brain(cfg) self.assertEqual(brain.system_prompt(cfg), "服务用户 王先生") self.assertEqual(await brain.greeting(cfg), "您好，王先生") + context = LLMContext( + messages=[{"role": "system", "content": brain.system_prompt(cfg)}] + ) + brain.prepare_greeting_context("您好，王先生", context) + self.assertEqual( + [message["role"] for message in context.get_messages()], + ["system", "system"], + ) + self.assertNotEqual(context.get_messages()[0]["role"], "assistant") + self.assertEqual( + context.get_messages()[1]["content"], + f"{GREETING_CONTEXT_MARKER}\n您好，王先生", + ) async def test_end_call_tool_is_owned_by_prompt_brain(self): brain = build_brain( @@ -604,7 +619,10 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): { "id": "start", "type": "start", - "data": {"name": "Start"}, + "data": { + "name": "Start", + "greeting": "欢迎，{{user_name}}", + }, }, { "id": "agent", @@ -710,6 +728,14 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): apply_turn_config=apply_turn_config, ) await brain.setup(cfg, runtime) + greeting = await brain.greeting(cfg) + self.assertEqual(greeting, "欢迎，王先生") + greeting_message = { + "role": "system", + "content": f"{GREETING_CONTEXT_MARKER}\n欢迎，王先生", + } + brain.prepare_greeting_context(greeting, context) + self.assertEqual(context.get_messages(), [greeting_message]) await brain.on_connected() self.assertEqual(brain._manager.current_node, "agent") variable_events = [ @@ -758,7 +784,7 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): agent_config = brain._agent_config("agent") self.assertIn("王先生", agent_config["role_message"]) self.assertIn("工作流路由已在用户一轮输入结束时完成", agent_config["role_message"]) - self.assertEqual(agent_config["task_messages"], []) + self.assertEqual(agent_config["task_messages"], [greeting_message]) self.assertFalse(agent_config["respond_immediately"]) self.assertFalse(any(isinstance(frame, LLMRunFrame) for frame in worker.frames)) self.assertEqual( @@ -785,7 +811,21 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(fixed_config["pre_actions"][0]["text"], "您好，王先生") self.assertEqual( fixed_config["task_messages"], - [{"role": "assistant", "content": "您好，王先生"}], + [ + greeting_message, + {"role": "assistant", "content": "您好，王先生"}, + ], + ) + self.assertEqual( + brain._agent_config( + "agent", + [{"role": "assistant", "content": "正在进入下一阶段"}], + )["task_messages"], + [ + greeting_message, + {"role": "assistant", "content": "正在进入下一阶段"}, + {"role": "assistant", "content": "您好，王先生"}, + ], ) self.assertEqual(fixed_config["pre_actions"][0]["node_id"], "agent") worker.frames.clear() @@ -800,8 +840,19 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): ] self.assertEqual( context_updates[-1].messages, - [{"role": "assistant", "content": "您好，王先生"}], + [ + greeting_message, + {"role": "assistant", "content": "您好，王先生"}, + ], ) + self.assertFalse( + any( + isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("source") == "workflow-fixed-reply" + for frame in queued + ) + ) + await brain.on_client_ready() fixed_reply_events = [ frame.message for frame in queued @@ -833,6 +884,22 @@ class WorkflowBrainTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(call_end.ending) self.assertTrue(call_end.armed) self.assertTrue(any(getattr(frame, "text", "") == "感谢来电" for frame in queued)) + transition_context_frames = [ + frame + for frame in worker.frames + if isinstance(frame, LLMMessagesAppendFrame) + and frame.messages + == [{"role": "assistant", "content": "正在为你结束流程"}] + ] + self.assertTrue(transition_context_frames) + transition_events = [ + frame.message + for frame in queued + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("source") == "workflow-edge-transition" + ] + self.assertEqual(transition_events[0]["content"], "正在为你结束流程") + self.assertEqual(transition_events[0]["nodeId"], "end") assistant_transcripts = [ frame.message.get("content") for frame in queued diff --git a/backend/tests/test_pipeline_events.py b/backend/tests/test_pipeline_events.py new file mode 100644 index 0000000..2a82d08 --- /dev/null +++ b/backend/tests/test_pipeline_events.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from pipecat.frames.frames import OutputTransportMessageUrgentFrame +from services.pipecat.pipeline_events import bind_cascade_pipeline_events + + +class _EventSource: + def __init__(self): + self.handlers = {} + + def event_handler(self, name): + def decorator(handler): + self.handlers[name] = handler + return handler + + return decorator + + +class _Worker: + def __init__(self): + self.frames = [] + + async def queue_frame(self, frame): + self.frames.append(frame) + + +class _Brain: + spec = SimpleNamespace(owns_context=True) + + def __init__(self, worker): + self.worker = worker + self.prepared_greeting = "" + + def prepare_greeting_context(self, greeting, _context): + self.prepared_greeting = greeting + + async def on_connected(self): + pass + + async def on_client_ready(self): + for content, timestamp in ( + ("Start Edge 过渡语", "2026-07-14T10:00:00.200+00:00"), + ("Agent 固定进入语", "2026-07-14T10:00:00.300+00:00"), + ): + await self.worker.queue_frame( + OutputTransportMessageUrgentFrame( + message={ + "type": "transcript", + "role": "assistant", + "content": content, + "timestamp": timestamp, + } + ) + ) + + +class PipelineEventTest(unittest.IsolatedAsyncioTestCase): + async def test_greeting_keeps_playback_timestamp_until_client_ready(self): + transport = _EventSource() + text_input = _EventSource() + user_aggregator = _EventSource() + assistant_aggregator = _EventSource() + worker = _Worker() + brain = _Brain(worker) + + bind_cascade_pipeline_events( + transport=transport, + worker=worker, + brain=brain, + context=SimpleNamespace(), + text_input=text_input, + user_aggregator=user_aggregator, + assistant_aggregator=assistant_aggregator, + greeting="助手开场白", + vision_enabled=False, + vision_state={"client_id": None}, + ) + + greeting_time = "2026-07-14T10:00:00.100+00:00" + with patch( + "services.pipecat.pipeline_events.time_now_iso8601", + return_value=greeting_time, + ) as clock: + await transport.handlers["on_client_connected"](transport, object()) + await text_input.handlers["on_client_ready"](text_input) + + transcripts = [ + frame.message + for frame in worker.frames + if isinstance(frame, OutputTransportMessageUrgentFrame) + and frame.message.get("type") == "transcript" + ] + ordered = sorted(transcripts, key=lambda message: message["timestamp"]) + self.assertEqual( + [message["content"] for message in ordered], + ["助手开场白", "Start Edge 过渡语", "Agent 固定进入语"], + ) + self.assertEqual(transcripts[0]["timestamp"], greeting_time) + self.assertEqual(brain.prepared_greeting, "助手开场白") + clock.assert_called_once_with() + + +if __name__ == "__main__": + unittest.main()