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

feat: unify debug runtime timelines

Browse Source
This commit is contained in:
Xin Wang
2026-08-13 09:30:43 +08:00
parent 7c18cea16d
commit eb111a3890
21 changed files with 1380 additions and 533 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/brains/prompt_brain.py
+26
View File
@@ -231,6 +231,22 @@ class PromptBrain(BaseBrain):
speech,
opening_message,
)
await self._emit_trace(
"message_started",
phase="opening",
hasSpeech=bool(speech),
showsMessage=True,
completionPolicy=MESSAGE_CONFIRMATION,
requiresConfirmation=True,
messageTitle=message_spec.display.title,
messageContent=message_spec.display.message,
messageActions=[
{
"id": "confirmed",
"label": message_spec.display.confirm_label,
}
],
)
message_result = await self._message_stages.run(
message_spec,
speak=self._speak_opening,
@@ -242,10 +258,20 @@ class PromptBrain(BaseBrain):
release_input_on_failure=False,
)
if not message_result.succeeded:
await self._emit_trace(
"message_failed",
phase="opening",
error=message_result.error or "开场消息显示失败",
)
await self._fail_opening(
message_result.error or "开场消息显示失败"
)
return
await self._emit_trace(
"message_completed",
phase="opening",
action=message_result.action,
)
context_message = confirmation_context_message(
message_result,
source="prompt-opening",
backend/services/brains/workflow_brain.py
+14
View File
@@ -1557,6 +1557,20 @@ class WorkflowBrain(BaseBrain):
showsMessage=require_confirmation,
completionPolicy=completion_policy,
requiresConfirmation=require_confirmation,
**(
{
"messageTitle": display.title,
"messageContent": display.message,
"messageActions": [
{
"id": "confirmed",
"label": display.confirm_label,
}
],
}
if display is not None
else {}
),
),
)
if result.succeeded:
backend/services/client_tools.py
+1
View File
@@ -194,6 +194,7 @@ class ClientToolBroker(FrameProcessor):
toolCallId=tool_call_id,
functionName=pending.function_name,
status=status,
internalStage=pending.function_name == "show_message",
**({"userAction": user_action} if user_action else {}),
**(
{"error": str(message.get("message") or "客户端工具执行失败")}
backend/services/conversation_history.py
+2
View File
@@ -120,6 +120,8 @@ class ConversationRecorder:
async def _append_workflow_event(self, message: dict) -> None:
"""Persist bounded control-plane history without creating chat rows."""
if message.get("internalStage") is True:
return
event_id = str(message.get("eventId") or "")
event_name = str(message.get("event") or "")
event_key = f"workflow:{event_id}"
backend/services/pipecat/call_lifecycle.py
+63 -4
View File
@@ -5,6 +5,8 @@ from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any
from uuid import uuid4
from loguru import logger
from pipecat.frames.frames import (
@@ -14,6 +16,38 @@ from pipecat.frames.frames import (
InterruptionFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.time import time_now_iso8601
CALL_END_EVENT_BY_STATE = {
"pending": "conversation_end_requested",
"speaking": "conversation_end_speaking",
"cancelled": "conversation_end_cancelled",
"ended": "conversation_ended",
}
def call_end_trace_event(
state: str,
*,
reason: str = "",
interruptible: bool = False,
voice_interruptible: bool = False,
) -> dict[str, Any]:
"""Build one shared record used by live calls and headless tests."""
return {
"type": "workflow-event",
"eventId": f"wfe_{uuid4().hex[:20]}",
"event": CALL_END_EVENT_BY_STATE.get(
state,
"conversation_end_state_changed",
),
"timestamp": time_now_iso8601(),
**({"reason": reason} if reason else {}),
"interruptible": interruptible,
"voiceInterruptible": voice_interruptible,
}
class SpeechPlaybackCompletion:
@@ -76,6 +110,7 @@ class CallEndCoordinator:
):
self._queue_end = queue_end
self._on_state_change = on_state_change
self._state_tasks: set[asyncio.Task[None]] = set()
self._ending = False
self._interruptible = False
self._armed = False
@@ -86,7 +121,7 @@ class CallEndCoordinator:
self._tracked_speech_completions: set[SpeechPlaybackCompletion] = set()
self._finish_after_tracked_speech = False
self._finished = False
self._reason = "completed"
self._reason = ""
@property
def ending(self) -> bool:
@@ -106,6 +141,12 @@ class CallEndCoordinator:
"""Whether transport output is currently producing bot speech."""
return self._speaking
@property
def reason(self) -> str:
"""Current requested ending reason for client feedback and records."""
return self._reason
async def wait_until_silent(self) -> None:
"""Wait for the transport-owned bot speech boundary."""
await self._speech_stopped.wait()
@@ -124,6 +165,7 @@ class CallEndCoordinator:
"""Cancel one pending ending when its tool explicitly allows it."""
if not self.interruptible:
return False
was_interruptible = self._interruptible
self._ending = False
self._interruptible = False
self._armed = False
@@ -132,12 +174,26 @@ class CallEndCoordinator:
# Transport speech state remains authoritative: if audio is currently
# playing, the interruption path will still emit BotStoppedSpeakingFrame.
logger.info("用户打断结束语,已撤销待挂断状态")
self._schedule_state("cancelled")
self._schedule_state("cancelled", interruptible=was_interruptible)
return True
def _schedule_state(self, state: str) -> None:
def _schedule_state(
self,
state: str,
*,
interruptible: bool | None = None,
) -> None:
if self._on_state_change is not None:
asyncio.create_task(self._on_state_change(state, self.interruptible))
task = asyncio.create_task(
self._on_state_change(
state,
self.interruptible
if interruptible is None
else interruptible,
)
)
self._state_tasks.add(task)
task.add_done_callback(self._state_tasks.discard)
def begin_response(self) -> None:
"""Start tracking speech produced by one LLM response."""
@@ -196,6 +252,9 @@ class CallEndCoordinator:
return
self._finished = True
self._interruptible = False
pending_states = tuple(self._state_tasks)
if pending_states:
await asyncio.gather(*pending_states, return_exceptions=True)
await self._queue_end(self._reason)
async def observe(self, frame) -> None:
backend/services/pipecat/pipeline.py
+56 -6
View File
@@ -17,6 +17,7 @@ from services.conversation_history import ConversationRecorder
from services.pipecat.call_lifecycle import (
CallEndCoordinator,
EndCallAfterSpeechProcessor,
call_end_trace_event,
)
from services.pipecat.service_factory import (
config_with_resource,
@@ -208,6 +209,7 @@ def _call_end_state_frame(
*,
interruptible: bool,
voice_interruptible: bool,
reason: str = "",
) -> OutputTransportMessageUrgentFrame:
"""Build the client event used to render the pending call-end state."""
return OutputTransportMessageUrgentFrame(
@@ -216,10 +218,30 @@ def _call_end_state_frame(
"state": state,
"interruptible": interruptible,
"voice_interruptible": voice_interruptible,
**({"reason": reason} if reason else {}),
}
)
def _call_lifecycle_trace_frame(
state: str,
*,
reason: str = "",
interruptible: bool = False,
voice_interruptible: bool = False,
) -> OutputTransportMessageUrgentFrame:
"""Build one persistent runtime record for the call-ending lifecycle."""
return OutputTransportMessageUrgentFrame(
message=call_end_trace_event(
state,
reason=reason,
interruptible=interruptible,
voice_interruptible=voice_interruptible,
)
)
async def run_pipeline(
transport,
cfg: AssistantConfig,
@@ -307,6 +329,12 @@ async def run_pipeline(
if worker is None:
return
logger.info(f"结束通话: reason={reason}")
await worker.queue_frame(
_call_lifecycle_trace_frame(
"ended",
reason=reason,
)
)
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message={"type": "call-ended", "reason": reason}
@@ -318,13 +346,21 @@ async def run_pipeline(
worker = worker_holder.get("worker")
if worker is None:
return
voice_interruptible = interruptible and current_enable_interrupt
await worker.queue_frame(
_call_end_state_frame(
state,
interruptible=interruptible,
voice_interruptible=(
interruptible and current_enable_interrupt
),
voice_interruptible=voice_interruptible,
reason=call_end.reason,
)
)
await worker.queue_frame(
_call_lifecycle_trace_frame(
state,
reason=call_end.reason,
interruptible=interruptible,
voice_interruptible=voice_interruptible,
)
)
@@ -1043,6 +1079,12 @@ async def run_realtime_pipeline(
if worker is None:
return
logger.info(f"结束 Realtime 通话: reason={reason}")
await worker.queue_frame(
_call_lifecycle_trace_frame(
"ended",
reason=reason,
)
)
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message={"type": "call-ended", "reason": reason}
@@ -1054,13 +1096,21 @@ async def run_realtime_pipeline(
worker = worker_holder.get("worker")
if worker is None:
return
voice_interruptible = interruptible and cfg.enableInterrupt
await worker.queue_frame(
_call_end_state_frame(
state,
interruptible=interruptible,
voice_interruptible=(
interruptible and cfg.enableInterrupt
),
voice_interruptible=voice_interruptible,
reason=call_end.reason,
)
)
await worker.queue_frame(
_call_lifecycle_trace_frame(
state,
reason=call_end.reason,
interruptible=interruptible,
voice_interruptible=voice_interruptible,
)
)
backend/services/test_runs/evaluator.py
+2
View File
@@ -163,6 +163,7 @@ class EvaluationEngine:
user_input=expected_turn.user_input,
assistant_reply="",
tool_calls=[],
runtime_records=[],
)
evaluations = [
await self._evaluate_behavior(behavior, raw)
@@ -175,6 +176,7 @@ class EvaluationEngine:
user_input=raw.user_input,
assistant_reply=raw.assistant_reply,
tool_calls=[item.as_result() for item in raw.tool_calls],
runtime_records=list(raw.runtime_records),
evaluations=evaluations,
)
)
backend/services/test_runs/orchestrator.py
+4
View File
@@ -302,6 +302,7 @@ class TestRunOrchestrator:
"greeting": "",
"assistantOutputs": [],
"autoConfirmedMessages": [],
"runtimeRecords": [],
},
"overallCriteria": [],
}
@@ -324,6 +325,7 @@ class TestRunOrchestrator:
"autoConfirmedMessages": list(
startup.auto_confirmed_messages
),
"runtimeRecords": list(startup.runtime_records),
}
row.result = result
await session.commit()
@@ -346,6 +348,7 @@ class TestRunOrchestrator:
"userInput": turn.user_input,
"assistantReply": turn.assistant_reply,
"toolCalls": [item.as_result() for item in turn.tool_calls],
"runtimeRecords": list(turn.runtime_records),
"evaluations": [],
}
turns = [item for item in turns if item.get("id") != turn.id]
@@ -382,6 +385,7 @@ class TestRunOrchestrator:
"autoConfirmedMessages": list(
startup.auto_confirmed_messages
),
"runtimeRecords": list(startup.runtime_records),
},
"overallCriteria": [
item.model_dump(mode="json", by_alias=True)
backend/services/test_runs/text_runner.py
+98 -2
View File
@@ -26,6 +26,7 @@ from pipecat.frames.frames import (
LLMFullResponseStartFrame,
LLMMessagesAppendFrame,
ManuallySwitchServiceFrame,
OutputTransportMessageUrgentFrame,
TTSSpeakFrame,
)
from pipecat.pipeline.pipeline import Pipeline
@@ -44,6 +45,7 @@ from services.knowledge import search as search_knowledge
from services.pipecat.call_lifecycle import (
CallEndCoordinator,
FixedSpeechPlaybackMarkerFrame,
call_end_trace_event,
)
from services.pipecat.processors import (
KnowledgeRetrievalProcessor,
@@ -102,6 +104,7 @@ class RawTurnResult:
user_input: str
assistant_reply: str
tool_calls: list[RawToolCall]
runtime_records: list[dict[str, Any]] = field(default_factory=list)
@dataclass
@@ -111,6 +114,7 @@ class TextStartupResult:
greeting: str = ""
assistant_outputs: list[str] = field(default_factory=list)
auto_confirmed_messages: list[dict[str, Any]] = field(default_factory=list)
runtime_records: list[dict[str, Any]] = field(default_factory=list)
@dataclass
@@ -148,6 +152,30 @@ class _ResponseWindow:
return self.active_llm == 0 and self.active_assistant == 0 and not self.pending_tools
def _opening_speech_from_message(message: Any) -> str:
"""Extract only the transcript emitted by Prompt's opening Message stage."""
if (
not isinstance(message, dict)
or message.get("type") != "transcript"
or message.get("source") != "prompt-opening-speech"
):
return ""
return str(message.get("content") or "").strip()
def _runtime_record_from_message(message: Any) -> dict[str, Any] | None:
"""Keep the same runtime-record envelope used by live conversation history."""
if (
not isinstance(message, dict)
or message.get("type") != "workflow-event"
or message.get("internalStage") is True
):
return None
return dict(message)
class _TextCaptureProcessor(FrameProcessor):
"""Capture assistant/tool output and expose a reliable response boundary."""
@@ -156,6 +184,8 @@ class _TextCaptureProcessor(FrameProcessor):
self.context = context
self.window = _ResponseWindow()
self.outputs: list[str] = []
self.opening_speech_outputs: list[str] = []
self.runtime_records: list[dict[str, Any]] = []
self.tool_calls: list[RawToolCall] = []
self._tool_by_id: dict[str, RawToolCall] = {}
self.error: TestExecutionError | None = None
@@ -195,6 +225,14 @@ class _TextCaptureProcessor(FrameProcessor):
await self.on_fixed_speech_complete()
return
if isinstance(frame, OutputTransportMessageUrgentFrame):
opening_speech = _opening_speech_from_message(frame.message)
if opening_speech:
self.opening_speech_outputs.append(opening_speech)
runtime_record = _runtime_record_from_message(frame.message)
if runtime_record is not None:
self.runtime_records.append(runtime_record)
if isinstance(frame, FixedSpeechPlaybackMarkerFrame):
await frame.completion.mark_played()
self.window.touch()
@@ -248,6 +286,27 @@ class _TextCaptureProcessor(FrameProcessor):
await self.push_frame(frame, direction)
def _resolve_startup_greeting(
shared_greeting: str,
opening_speech_outputs: list[str],
) -> str:
"""Return the opening speech shown by the headless run.
Prompt confirmation mode deliberately suppresses the shared greeting
because its atomic Message stage owns both speech and the dialog. In text
tests the fixed speech transcript carries a dedicated source marker, so
preserve that exact output instead of guessing from unrelated fixed speech.
"""
greeting = shared_greeting.strip()
if greeting:
return greeting
return next(
(text.strip() for text in opening_speech_outputs if text.strip()),
"",
)
def _context_messages(context_turns: list[ContextTurn]) -> list[dict[str, Any]]:
messages: list[dict[str, Any]] = []
latest_call_by_name: dict[str, str] = {}
@@ -494,9 +553,34 @@ class TextPipelineRunner:
async def queue_call_end(reason: str) -> None:
logger.debug(f"文本测试收到会话结束请求: {reason}")
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message=call_end_trace_event(
"ended",
reason=reason,
)
)
)
await worker.queue_frame(EndFrame(reason=reason))
call_end = CallEndCoordinator(queue_call_end)
async def record_call_end_state(
state: str,
interruptible: bool,
) -> None:
await worker.queue_frame(
OutputTransportMessageUrgentFrame(
message=call_end_trace_event(
state,
reason=call_end.reason,
interruptible=interruptible,
)
)
)
call_end = CallEndCoordinator(
queue_call_end,
on_state_change=record_call_end_state,
)
async def finish_fixed_speech_if_ending() -> None:
if call_end.ending:
@@ -667,6 +751,7 @@ class TextPipelineRunner:
runner_task: asyncio.Task[None] | None = None
try:
startup_runtime_record_marker = len(capture.runtime_records)
await brain.run_preflight()
if executor.unmocked_calls:
raise TestExecutionError(
@@ -693,12 +778,19 @@ class TextPipelineRunner:
runner_task,
marker,
)
startup_greeting = _resolve_startup_greeting(
greeting,
capture.opening_speech_outputs,
)
startup = TextStartupResult(
greeting=greeting,
greeting=startup_greeting,
assistant_outputs=list(capture.outputs),
auto_confirmed_messages=list(
text_client_tools.auto_confirmed_messages
),
runtime_records=list(
capture.runtime_records[startup_runtime_record_marker:]
),
)
if on_startup_complete is not None:
await on_startup_complete(startup)
@@ -775,6 +867,7 @@ class TextPipelineRunner:
activity_marker = capture.window.activity_count
output_marker = len(capture.outputs)
tool_marker = len(capture.tool_calls)
runtime_record_marker = len(capture.runtime_records)
await worker.queue_frame(
LLMMessagesAppendFrame(
messages=[{"role": "user", "content": turn.user_input}],
@@ -795,4 +888,7 @@ class TextPipelineRunner:
user_input=turn.user_input,
assistant_reply="\n\n".join(capture.outputs[output_marker:]).strip(),
tool_calls=list(capture.tool_calls[tool_marker:]),
runtime_records=list(
capture.runtime_records[runtime_record_marker:]
),
)