apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-06-22 08:38:13 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat: add chat based testing for voice agent (#308)

Browse source 
* feat: add backend foundations

* feat: add text chat UI

* chore: simplify the reload behaviour

* fix: fix upgrade banner to be triggered after package upload

* feat: simplify TesterPanel design

* chore: fix formatting and generate client

* chore: fix tracing for text chat mode

* fix: fix revert and edit CTA

* refactor: refactor TesterPanel into smaller components

* feat: enable runtime transition of nodes

* fix: fix review comments
This commit is contained in:
Abhishek 2026-05-21 15:20:02 +05:30 • committed by GitHub
parent 67479e98fd
commit d97d1d72cd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
96 changed files with 7630 additions and 1684 deletions
Download patch file Download diff file Expand all files Collapse all files

98
api/services/workflow/pipecat_engine.py
View file

@ -1,4 +1,4 @@
from typing import TYPE_CHECKING, Awaitable, Callable, Dict, Optional, Union
from typing import TYPE_CHECKING, Awaitable, Callable, Dict, Literal, Optional, Union
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import (
@ -7,6 +7,7 @@ from pipecat.frames.frames import (
CancelFrame,
EndFrame,
FunctionCallResultProperties,
LLMContextFrame,
TTSSpeakFrame,
)
from pipecat.pipeline.task import PipelineTask
@ -533,7 +534,7 @@ class PipecatEngine:
)
await self._update_llm_context(system_prompt, functions)
async def set_node(self, node_id: str):
async def set_node(self, node_id: str, emit_transition_event: bool = True):
"""
Simplified set_node implementation according to v2 PRD.
"""
@ -556,7 +557,7 @@ class PipecatEngine:
nodes_visited.append(node.name)
# Send node transition event if callback is provided
if self._node_transition_callback:
if emit_transition_event and self._node_transition_callback:
try:
await self._node_transition_callback(
node_id,
@ -598,8 +599,8 @@ class PipecatEngine:
# Setup LLM context with prompts and functions.
await self._setup_llm_context(node)
def get_start_greeting(self) -> Optional[tuple[str, Optional[str]]]:
"""Return the greeting info for the start node, or None if not configured.
def get_node_greeting(self, node_id: str) -> Optional[tuple[str, Optional[str]]]:
"""Return the greeting info for a node, or None if not configured.
Returns:
A tuple of (greeting_type, value) where:
@ -607,20 +608,93 @@ class PipecatEngine:
- ("audio", recording_id) for pre-recorded audio greetings
Or None if no greeting is configured.
"""
start_node = self.workflow.nodes.get(self.workflow.start_node_id)
if not start_node:
node = self.workflow.nodes.get(node_id)
if not node:
return None
greeting_type = start_node.greeting_type or "text"
greeting_type = node.greeting_type or "text"
if greeting_type == "audio" and start_node.greeting_recording_id:
return ("audio", start_node.greeting_recording_id)
if greeting_type == "audio" and node.greeting_recording_id:
return ("audio", node.greeting_recording_id)
if start_node.greeting:
return ("text", self._format_prompt(start_node.greeting))
if node.greeting:
return ("text", self._format_prompt(node.greeting))
return None
def get_start_greeting(self) -> Optional[tuple[str, Optional[str]]]:
"""Return the greeting info for the start node, or None if not configured."""
return self.get_node_greeting(self.workflow.start_node_id)
async def queue_node_opening(
self,
*,
node_id: str,
previous_node_id: Optional[str] = None,
generate_if_no_greeting: bool = False,
) -> Literal["none", "greeting", "llm"]:
"""Queue the opening behavior for a node.
This is the shared source of truth for how a node begins once the
engine is ready and the node has already been set on the context.
Returns:
"greeting" when a text/audio greeting was queued,
"llm" when an initial LLM generation was queued,
"none" when nothing was queued.
"""
if previous_node_id != node_id:
greeting_info = self.get_node_greeting(node_id)
if greeting_info:
greeting_type, greeting_value = greeting_info
if (
greeting_type == "audio"
and greeting_value
and self._fetch_recording_audio
and self._transport_output is not None
):
logger.debug(f"Playing audio greeting recording: {greeting_value}")
result = await self._fetch_recording_audio(
recording_pk=int(greeting_value)
)
if result:
await play_audio(
result.audio,
sample_rate=self._audio_config.pipeline_sample_rate
if self._audio_config
else 16000,
queue_frame=self._transport_output.queue_frame,
transcript=result.transcript,
append_to_context=True,
)
return "greeting"
logger.warning(
f"Failed to fetch audio greeting {greeting_value}, "
"falling back to LLM generation"
)
elif greeting_value and self.task is not None:
logger.debug("Playing text greeting via TTS")
# append_to_context=True so the assistant aggregator commits
# the greeting to the LLM context once TTS finishes; without
# it the LLM would re-greet on its first generation.
await self.task.queue_frame(
TTSSpeakFrame(greeting_value, append_to_context=True)
)
return "greeting"
if (
generate_if_no_greeting
and self.llm is not None
and self.context is not None
):
logger.debug("Queueing initial LLM generation for node opening")
# Queue after the voicemail detector in the live pipeline so the
# detector can gate initial generations when needed.
await self.llm.queue_frame(LLMContextFrame(self.context))
return "llm"
return "none"
async def _handle_end_node(self, node: Node) -> None:
"""Handle end node execution."""
# Setup LLM context with prompts and functions.

11
api/services/workflow/pipecat_engine_custom_tools.py
View file

@ -511,6 +511,17 @@ class CustomToolManager:
workflow_run = await db_client.get_workflow_run_by_id(
self._engine._workflow_run_id
)
if workflow_run.mode == WorkflowRunMode.TEXTCHAT.value:
textchat_error_result = {
"status": "failed",
"message": "I'm sorry, but call transfers are not available in text chat tests.",
"action": "transfer_failed",
"reason": "textchat_not_supported",
}
await self._handle_transfer_result(
textchat_error_result, function_call_params, properties
)
return
if workflow_run.mode in [
WorkflowRunMode.WEBRTC.value,
WorkflowRunMode.SMALLWEBRTC.value,

39
api/services/workflow/qa/tracing.py
View file

@ -6,7 +6,10 @@ import re
from loguru import logger
from api.db.models import WorkflowRunModel
from api.services.pipecat.tracing_config import get_trace_url
from api.services.pipecat.tracing_config import (
build_remote_parent_context,
get_trace_url,
)
def extract_trace_id(gathered_context: dict) -> str | None:
@ -33,36 +36,12 @@ def setup_langfuse_parent_context(workflow_run: WorkflowRunModel):
Returns the parent context object, or None if tracing is unavailable.
"""
try:
from opentelemetry.trace import (
NonRecordingSpan,
SpanContext,
TraceFlags,
set_span_in_context,
)
from api.services.pipecat.tracing_config import ensure_tracing
if not ensure_tracing():
return None
gathered_context = workflow_run.gathered_context or {}
trace_id = extract_trace_id(gathered_context)
if not trace_id:
logger.debug("No trace_id found, skipping Langfuse tracing")
return None
parent_span_ctx = SpanContext(
trace_id=int(trace_id, 16),
span_id=0x1,
is_remote=True,
trace_flags=TraceFlags(0x01),
)
return set_span_in_context(NonRecordingSpan(parent_span_ctx))
except Exception as e:
logger.warning(f"Failed to set up Langfuse parent context: {e}")
gathered_context = workflow_run.gathered_context or {}
trace_id = extract_trace_id(gathered_context)
if not trace_id:
logger.debug("No trace_id found, skipping Langfuse tracing")
return None
return build_remote_parent_context(trace_id)
def add_qa_span_to_trace(

144
api/services/workflow/text_chat_logs.py Normal file
View file

@ -0,0 +1,144 @@
"""Helpers for projecting text-chat session state into run-log snapshots."""
from typing import Any
from api.services.pipecat.realtime_feedback_events import (
build_bot_text_event,
build_function_call_end_event,
build_function_call_start_event,
build_node_transition_event,
build_pipeline_error_event,
build_user_transcription_event,
realtime_feedback_event_sort_key,
stamp_realtime_feedback_event,
)
def visible_text_chat_turns(session_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Return the active branch of turns for the current text-chat session.
After a rewind, `session_data["turns"]` may still contain future turns until
the next message is sent. Those turns are no longer part of the visible
branch, so callers that synthesize transcript/log views should trim at
`cursor_turn_id`.
"""
turns = list(session_data.get("turns") or [])
cursor_turn_id = session_data.get("cursor_turn_id")
if cursor_turn_id is None:
return turns
for index, turn in enumerate(turns):
if turn.get("id") == cursor_turn_id:
return turns[: index + 1]
return turns
def build_text_chat_realtime_feedback_events(
session_data: dict[str, Any],
) -> list[dict[str, Any]]:
"""Project text-chat session state into `workflow_runs.logs` event format.
`workflow_run_text_sessions` holds the authoritative rewindable conversation
state. Historical run pages and QA helpers read the normalized
`workflow_runs.logs.realtime_feedback_events` schema instead, so this helper
rebuilds that snapshot from the currently visible branch.
"""
events: list[dict[str, Any]] = []
last_emitted_node_id: str | None = None
for turn_index, turn in enumerate(visible_text_chat_turns(session_data)):
turn_events = list(turn.get("events") or [])
for event in turn_events:
payload = dict(event.get("payload") or {})
event_type = event.get("type")
timestamp = event.get("created_at") or turn.get("created_at")
if event_type == "node_transition":
node_id = payload.get("node_id")
if node_id is not None and node_id == last_emitted_node_id:
continue
snapshot_event = stamp_realtime_feedback_event(
build_node_transition_event(
node_id=node_id,
node_name=payload.get("node_name"),
previous_node_id=payload.get("previous_node_id"),
previous_node_name=payload.get("previous_node_name"),
allow_interrupt=bool(payload.get("allow_interrupt", False)),
),
timestamp=timestamp,
turn=turn_index,
node_id=node_id,
node_name=payload.get("node_name"),
)
if node_id is not None:
last_emitted_node_id = node_id
events.append(snapshot_event)
elif event_type == "tool_call_started":
events.append(
stamp_realtime_feedback_event(
build_function_call_start_event(
function_name=payload.get("function_name"),
tool_call_id=payload.get("tool_call_id"),
arguments=payload.get("arguments"),
),
timestamp=timestamp,
turn=turn_index,
)
)
elif event_type == "tool_call_result":
events.append(
stamp_realtime_feedback_event(
build_function_call_end_event(
function_name=payload.get("function_name"),
tool_call_id=payload.get("tool_call_id"),
result=payload.get("result"),
),
timestamp=timestamp,
turn=turn_index,
)
)
elif event_type == "execution_error":
events.append(
stamp_realtime_feedback_event(
build_pipeline_error_event(
error=payload.get("message", "Execution error"),
fatal=True,
),
timestamp=timestamp,
turn=turn_index,
)
)
user_message = turn.get("user_message") or {}
if user_message.get("text"):
message_timestamp = user_message.get("created_at") or turn.get("created_at")
events.append(
stamp_realtime_feedback_event(
build_user_transcription_event(
text=user_message["text"],
final=True,
timestamp=message_timestamp,
),
timestamp=message_timestamp,
turn=turn_index,
)
)
assistant_message = turn.get("assistant_message") or {}
if assistant_message.get("text"):
message_timestamp = assistant_message.get("created_at") or turn.get(
"created_at"
)
events.append(
stamp_realtime_feedback_event(
build_bot_text_event(
text=assistant_message["text"],
timestamp=message_timestamp,
),
timestamp=message_timestamp,
turn=turn_index,
)
)
return sorted(events, key=realtime_feedback_event_sort_key)

649
api/services/workflow/text_chat_runner.py Normal file
View file

@ -0,0 +1,649 @@
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any
from fastapi.encoders import jsonable_encoder
from loguru import logger
from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMAssistantPushAggregationFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
TextFrame,
TTSSpeakFrame,
TTSStoppedFrame,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMAssistantAggregatorParams,
LLMContextAggregatorPair,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.run_context import set_current_org_id
from api.db import db_client
from api.enums import WorkflowRunMode, WorkflowRunState
from api.services.configuration.resolve import resolve_effective_config
from api.services.pipecat.audio_config import create_audio_config
from api.services.pipecat.pipeline_builder import create_pipeline_task
from api.services.pipecat.pipeline_metrics_aggregator import (
PipelineMetricsAggregator,
)
from api.services.pipecat.recording_audio_cache import create_recording_audio_fetcher
from api.services.pipecat.service_factory import create_llm_service
from api.services.pipecat.tracing_config import (
build_remote_parent_context,
get_trace_url,
)
from api.services.workflow.dto import ReactFlowDTO
from api.services.workflow.pipecat_engine import PipecatEngine
from api.services.workflow.workflow_graph import WorkflowGraph
TEXT_CHAT_CHECKPOINT_VERSION = 1
TEXT_CHAT_TURN_TIMEOUT_SECONDS = 60.0
TEXT_CHAT_IDLE_SETTLE_SECONDS = 0.2
TEXT_CHAT_INTERNAL_CANCEL_REASON = "text_chat_turn_complete"
def text_chat_trace_id(workflow_run_id: int) -> str:
"""Deterministic Langfuse trace id for a text-chat session.
Each turn runs in its own short-lived pipeline, so there is no single
long-running task to own the trace the way a voice call does. Deriving the
id from the run id means every turn re-creates the *same* trace id and all
per-turn spans land in one shared trace — without persisting extra state
across the otherwise stateless turn requests.
"""
digest = hashlib.sha256(f"dograh-text-chat:{workflow_run_id}".encode()).hexdigest()
return digest[:32]
def default_text_chat_checkpoint() -> dict[str, Any]:
return {
"version": TEXT_CHAT_CHECKPOINT_VERSION,
"anchor_turn_id": None,
"current_node_id": None,
"messages": [],
"gathered_context": {},
"tool_state": {},
}
def normalize_text_chat_checkpoint(
checkpoint: dict[str, Any] | None,
) -> dict[str, Any]:
normalized = {
**default_text_chat_checkpoint(),
**(checkpoint or {}),
}
normalized["messages"] = list(normalized.get("messages") or [])
normalized["gathered_context"] = dict(normalized.get("gathered_context") or {})
normalized["tool_state"] = dict(normalized.get("tool_state") or {})
return normalized
@dataclass
class TextChatTurnExecutionResult:
assistant_text: str | None
assistant_created_at: str
events: list[dict[str, Any]]
usage: dict[str, Any]
checkpoint: dict[str, Any]
gathered_context: dict[str, Any]
initial_context: dict[str, Any]
state: str
is_completed: bool
@dataclass
class _ResponseWindowState:
active_assistant_segments: int = 0
active_llm_completions: int = 0
pending_context_requests: int = 0
blocking_tool_call_ids: set[str] = field(default_factory=set)
outputs: list[str] = field(default_factory=list)
def note_direct_context_request(self) -> None:
self.pending_context_requests += 1
def note_upstream_context_request(self) -> None:
self.pending_context_requests += 1
def note_llm_start(self) -> None:
if self.pending_context_requests > 0:
self.pending_context_requests -= 1
self.active_llm_completions += 1
def note_llm_end(self) -> None:
if self.active_llm_completions > 0:
self.active_llm_completions -= 1
def note_assistant_turn_started(self) -> None:
self.active_assistant_segments += 1
def note_assistant_turn_stopped(self, content: str) -> None:
if self.active_assistant_segments > 0:
self.active_assistant_segments -= 1
normalized_content = content.strip()
if normalized_content:
self.outputs.append(normalized_content)
def note_function_call_in_progress(self, tool_call_id: str, blocking: bool) -> None:
if blocking:
self.blocking_tool_call_ids.add(tool_call_id)
def note_function_call_result(self, tool_call_id: str) -> None:
self.blocking_tool_call_ids.discard(tool_call_id)
@property
def has_blocking_tool_calls(self) -> bool:
return bool(self.blocking_tool_call_ids)
@property
def frontier_is_idle(self) -> bool:
return (
self.pending_context_requests == 0
and self.active_llm_completions == 0
and self.active_assistant_segments == 0
and not self.has_blocking_tool_calls
)
class _TaskQueueProxy:
def __init__(self, queue_frame):
self.queue_frame = queue_frame
class _TextChatCaptureProcessor(FrameProcessor):
def __init__(self, response_window: _ResponseWindowState) -> None:
super().__init__()
self.last_activity_at = time.monotonic()
self.activity_count = 0
self.events: list[dict[str, Any]] = []
self._response_window = response_window
def _touch(self) -> None:
self.last_activity_at = time.monotonic()
self.activity_count += 1
def _append_event(self, event_type: str, payload: dict[str, Any]) -> None:
self.events.append(
{
"type": event_type,
"created_at": datetime.now(UTC).isoformat(),
"payload": jsonable_encoder(payload),
}
)
async def process_frame(self, frame, direction: FrameDirection):
await super().process_frame(frame, direction)
self._touch()
if isinstance(frame, TTSSpeakFrame):
text_frame = TextFrame(frame.text)
text_frame.append_to_context = (
frame.append_to_context if frame.append_to_context is not None else True
)
await self.push_frame(text_frame, direction)
await self.push_frame(LLMAssistantPushAggregationFrame(), direction)
return
if isinstance(frame, LLMContextFrame) and direction == FrameDirection.UPSTREAM:
self._response_window.note_upstream_context_request()
if isinstance(frame, TTSStoppedFrame):
await self.push_frame(frame, direction)
await self.push_frame(LLMAssistantPushAggregationFrame(), direction)
return
if (
isinstance(frame, LLMFullResponseStartFrame)
and direction == FrameDirection.DOWNSTREAM
):
self._response_window.note_llm_start()
if (
isinstance(frame, LLMFullResponseEndFrame)
and direction is FrameDirection.DOWNSTREAM
):
self._response_window.note_llm_end()
await self.push_frame(frame, direction)
# Text chat has no TTS/output transport, so mixed text+tool responses
# would otherwise leave function calls waiting forever on a
# BotStoppedSpeakingFrame that never arrives.
await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
return
if isinstance(frame, FunctionCallInProgressFrame):
self._response_window.note_function_call_in_progress(
tool_call_id=frame.tool_call_id,
blocking=frame.cancel_on_interruption,
)
self._append_event(
"tool_call_started",
{
"function_name": frame.function_name,
"tool_call_id": frame.tool_call_id,
"arguments": dict(frame.arguments or {}),
},
)
elif isinstance(frame, FunctionCallResultFrame):
self._response_window.note_function_call_result(frame.tool_call_id)
self._append_event(
"tool_call_result",
{
"function_name": frame.function_name,
"tool_call_id": frame.tool_call_id,
"result": frame.result,
},
)
elif isinstance(frame, EndFrame):
self._append_event("session_end", {"reason": frame.reason})
elif isinstance(frame, CancelFrame):
if frame.reason != TEXT_CHAT_INTERNAL_CANCEL_REASON:
self._append_event("session_cancelled", {"reason": frame.reason})
await self.push_frame(frame, direction)
def _merge_usage_info(
existing: dict[str, Any] | None,
delta: dict[str, Any] | None,
) -> dict[str, Any]:
merged = dict(existing or {})
delta = dict(delta or {})
merged_llm = dict(merged.get("llm") or {})
for key, value in (delta.get("llm") or {}).items():
current = dict(merged_llm.get(key) or {})
merged_llm[key] = {
"prompt_tokens": int(current.get("prompt_tokens") or 0)
+ int(value.get("prompt_tokens") or 0),
"completion_tokens": int(current.get("completion_tokens") or 0)
+ int(value.get("completion_tokens") or 0),
"total_tokens": int(current.get("total_tokens") or 0)
+ int(value.get("total_tokens") or 0),
"cache_read_input_tokens": int(current.get("cache_read_input_tokens") or 0)
+ int(value.get("cache_read_input_tokens") or 0),
"cache_creation_input_tokens": int(
current.get("cache_creation_input_tokens") or 0
)
+ int(value.get("cache_creation_input_tokens") or 0),
}
merged["llm"] = merged_llm
for section in ("tts", "stt"):
merged_section = dict(merged.get(section) or {})
for key, value in (delta.get(section) or {}).items():
merged_section[key] = float(merged_section.get(key) or 0) + float(value)
merged[section] = merged_section
merged["call_duration_seconds"] = int(
merged.get("call_duration_seconds") or 0
) + int(delta.get("call_duration_seconds") or 0)
return merged
def merge_text_chat_usage_info(
existing: dict[str, Any] | None,
delta: dict[str, Any] | None,
) -> dict[str, Any]:
return _merge_usage_info(existing, delta)
def _resolve_checkpoint_for_pending_turn(
session_data: dict[str, Any],
checkpoint: dict[str, Any] | None,
) -> dict[str, Any]:
turns = list(session_data.get("turns") or [])
if not turns:
return normalize_text_chat_checkpoint(checkpoint)
pending_turn = turns[-1]
if pending_turn.get("status") != "pending":
return normalize_text_chat_checkpoint(checkpoint)
for turn in reversed(turns[:-1]):
if turn.get("status") != "completed":
continue
stored_checkpoint = turn.get("checkpoint_after_turn")
if stored_checkpoint:
return normalize_text_chat_checkpoint(stored_checkpoint)
break
return normalize_text_chat_checkpoint(checkpoint)
async def _wait_for_quiescence(
*,
capture_processor: _TextChatCaptureProcessor,
response_window: _ResponseWindowState,
runner_task: asyncio.Task,
activity_marker: int,
timeout_seconds: float = TEXT_CHAT_TURN_TIMEOUT_SECONDS,
) -> None:
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout_seconds
while loop.time() < deadline:
if runner_task.done():
await runner_task
return
if (
capture_processor.activity_count <= activity_marker
and response_window.frontier_is_idle
):
await asyncio.sleep(0.05)
continue
if (
response_window.frontier_is_idle
and (time.monotonic() - capture_processor.last_activity_at)
>= TEXT_CHAT_IDLE_SETTLE_SECONDS
):
return
await asyncio.sleep(0.05)
raise TimeoutError(
"Timed out waiting for text chat response window to settle "
f"(pending_context_requests={response_window.pending_context_requests}, "
f"active_llm_completions={response_window.active_llm_completions}, "
f"active_assistant_segments={response_window.active_assistant_segments}, "
f"blocking_tool_calls={sorted(response_window.blocking_tool_call_ids)})"
)
async def execute_text_chat_pending_turn(
*,
workflow_run_id: int,
workflow_id: int,
session_data: dict[str, Any],
checkpoint: dict[str, Any] | None,
) -> TextChatTurnExecutionResult:
turns = list(session_data.get("turns") or [])
if not turns or turns[-1].get("status") != "pending":
raise ValueError("Text chat session has no pending turn to execute")
pending_turn = turns[-1]
pending_user_message = (
((pending_turn.get("user_message") or {}).get("text") or "").strip()
if pending_turn.get("user_message") is not None
else None
)
workflow_run, _ = await db_client.get_workflow_run_with_context(workflow_run_id)
if not workflow_run or workflow_run.workflow_id != workflow_id:
raise ValueError("Workflow run not found for text chat execution")
if workflow_run.definition is None:
raise ValueError("Workflow run is missing a pinned definition")
if workflow_run.workflow is None or workflow_run.workflow.user is None:
raise ValueError("Workflow run is missing workflow context")
workflow = await db_client.get_workflow(
workflow_id, organization_id=workflow_run.workflow.organization_id
)
if workflow is None:
raise ValueError("Workflow not found for text chat execution")
# Stamp the async context so OTEL spans are tagged with this org and routed
# to its Langfuse project (the voice paths do this in run_pipeline /
# webrtc_signaling; the text path previously skipped it, so its spans never
# reached org-specific exporters).
set_current_org_id(workflow.organization_id)
run_definition = workflow_run.definition
run_configs = run_definition.workflow_configurations or {}
user_config = await db_client.get_user_configurations(workflow_run.workflow.user.id)
user_config = resolve_effective_config(
user_config, run_configs.get("model_overrides")
)
if user_config.llm is None:
raise ValueError("Text chat requires an LLM configuration")
llm = create_llm_service(user_config)
inference_llm = llm
runtime_configuration = {
"llm_provider": user_config.llm.provider,
"llm_model": user_config.llm.model,
}
initial_context = {
**(workflow_run.initial_context or {}),
"runtime_configuration": runtime_configuration,
}
workflow_graph = WorkflowGraph(
ReactFlowDTO.model_validate(run_definition.workflow_json)
)
base_checkpoint = _resolve_checkpoint_for_pending_turn(session_data, checkpoint)
response_window = _ResponseWindowState()
capture_processor = _TextChatCaptureProcessor(response_window)
context = LLMContext()
context.set_messages(base_checkpoint["messages"])
node_transition_events = capture_processor.events
async def send_node_transition(
node_id: str,
node_name: str,
previous_node_id: str | None,
previous_node_name: str | None,
allow_interrupt: bool = False,
) -> None:
node_transition_events.append(
{
"type": "node_transition",
"created_at": datetime.now(UTC).isoformat(),
"payload": {
"node_id": node_id,
"node_name": node_name,
"previous_node_id": previous_node_id,
"previous_node_name": previous_node_name,
"allow_interrupt": allow_interrupt,
},
}
)
embeddings_api_key = None
embeddings_model = None
embeddings_base_url = None
if user_config.embeddings:
embeddings_api_key = user_config.embeddings.api_key
embeddings_model = user_config.embeddings.model
embeddings_base_url = getattr(user_config.embeddings, "base_url", None)
has_recordings = await db_client.has_active_recordings(workflow.organization_id)
context_compaction_enabled = (workflow.workflow_configurations or {}).get(
"context_compaction_enabled", False
)
engine = PipecatEngine(
llm=llm,
inference_llm=inference_llm,
context=context,
workflow=workflow_graph,
call_context_vars=initial_context,
workflow_run_id=workflow_run_id,
node_transition_callback=send_node_transition,
embeddings_api_key=embeddings_api_key,
embeddings_model=embeddings_model,
embeddings_base_url=embeddings_base_url,
has_recordings=has_recordings,
context_compaction_enabled=context_compaction_enabled,
)
engine._gathered_context = dict(base_checkpoint["gathered_context"])
assistant_params = LLMAssistantAggregatorParams()
context_aggregator = LLMContextAggregatorPair(
context, assistant_params=assistant_params
)
assistant_context_aggregator = context_aggregator.assistant()
@assistant_context_aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(_aggregator):
response_window.note_assistant_turn_started()
@assistant_context_aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(_aggregator, message):
response_window.note_assistant_turn_stopped(message.content or "")
# Text chat has no wire transport; reuse the neutral 16 kHz config shape
# from the browser pipeline so TTS/recording helpers still have sane defaults.
audio_config = create_audio_config(WorkflowRunMode.SMALLWEBRTC.value)
pipeline_metrics_aggregator = PipelineMetricsAggregator()
# Stitch every per-turn pipeline of this session into one Langfuse trace by
# handing each task the same remote parent context (derived from the run id).
trace_id = text_chat_trace_id(workflow_run_id)
conversation_parent_context = build_remote_parent_context(trace_id)
# The stitched trace has no real root span (each per-turn conversation span
# hangs off a synthetic remote parent), so Langfuse can't infer a name and
# shows "Unnamed trace". Name it explicitly via the conversation span.
trace_span_attributes = {
"langfuse.trace.name": workflow_run.name or f"text-chat-{workflow_run_id}"
}
pipeline = Pipeline(
[
llm,
capture_processor,
assistant_context_aggregator,
pipeline_metrics_aggregator,
]
)
task = create_pipeline_task(
pipeline,
workflow_run_id,
audio_config,
conversation_parent_context=conversation_parent_context,
conversation_type="text",
additional_span_attributes=trace_span_attributes,
)
runner = PipelineRunner(handle_sigint=False, handle_sigterm=False)
runner_task = asyncio.create_task(runner.run(task))
engine.set_task(task)
engine.set_audio_config(audio_config)
engine.set_transport_output(_TaskQueueProxy(task.queue_frame))
engine.set_fetch_recording_audio(
create_recording_audio_fetcher(
organization_id=workflow.organization_id,
pipeline_sample_rate=audio_config.pipeline_sample_rate,
)
)
try:
await asyncio.wait_for(task._pipeline_start_event.wait(), timeout=5.0)
await engine.initialize()
current_node_id = base_checkpoint.get("current_node_id")
target_node_id = current_node_id or workflow_graph.start_node_id
await engine.set_node(
target_node_id,
emit_transition_event=current_node_id is None,
)
opening_marker = capture_processor.activity_count
opening_expects_llm = pending_user_message is None and (
current_node_id == target_node_id
or engine.get_node_greeting(target_node_id) is None
)
if opening_expects_llm:
response_window.note_direct_context_request()
opening_action = await engine.queue_node_opening(
node_id=target_node_id,
previous_node_id=current_node_id,
generate_if_no_greeting=pending_user_message is None,
)
if opening_action != "llm" and opening_expects_llm:
response_window.pending_context_requests = max(
0, response_window.pending_context_requests - 1
)
if opening_action != "none":
await _wait_for_quiescence(
capture_processor=capture_processor,
response_window=response_window,
runner_task=runner_task,
activity_marker=opening_marker,
)
if pending_user_message is not None:
context.add_message({"role": "user", "content": pending_user_message})
generation_marker = capture_processor.activity_count
response_window.note_direct_context_request()
await llm.queue_frame(LLMContextFrame(context))
await _wait_for_quiescence(
capture_processor=capture_processor,
response_window=response_window,
runner_task=runner_task,
activity_marker=generation_marker,
)
finally:
if not task.has_finished():
await task.cancel(reason=TEXT_CHAT_INTERNAL_CANCEL_REASON)
try:
await runner_task
except Exception:
logger.exception(
"Transportless text chat pipeline failed while closing run {}",
workflow_run_id,
)
await engine.cleanup()
raise
await engine.cleanup()
gathered_context = await engine.get_gathered_context()
assistant_text = (
"\n\n".join(part for part in response_window.outputs if part).strip()
if response_window.outputs
else None
)
assistant_created_at = datetime.now(UTC).isoformat()
usage = pipeline_metrics_aggregator.get_all_usage_metrics_serialized()
current_node = getattr(engine, "_current_node", None)
updated_checkpoint = {
"version": TEXT_CHAT_CHECKPOINT_VERSION,
"anchor_turn_id": pending_turn.get("id"),
"current_node_id": current_node.id if current_node else None,
"messages": jsonable_encoder(context.get_messages()),
"gathered_context": jsonable_encoder(gathered_context),
"tool_state": jsonable_encoder(base_checkpoint.get("tool_state") or {}),
}
encoded_gathered_context = jsonable_encoder(gathered_context)
trace_url = get_trace_url(trace_id, org_id=workflow.organization_id)
if trace_url:
encoded_gathered_context = {**encoded_gathered_context, "trace_url": trace_url}
return TextChatTurnExecutionResult(
assistant_text=assistant_text,
assistant_created_at=assistant_created_at,
events=jsonable_encoder(capture_processor.events),
usage=jsonable_encoder(usage),
checkpoint=updated_checkpoint,
gathered_context=encoded_gathered_context,
initial_context=jsonable_encoder(initial_context),
state=(
WorkflowRunState.COMPLETED.value
if engine.is_call_disposed()
else WorkflowRunState.RUNNING.value
),
is_completed=engine.is_call_disposed(),
)

411
api/services/workflow/text_chat_session_service.py Normal file
View file

@ -0,0 +1,411 @@
"""Service helpers for text-chat session lifecycle orchestration."""
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4
from loguru import logger
from api.db import db_client
from api.db.models import WorkflowRunTextSessionModel
from api.db.workflow_run_text_session_client import (
WorkflowRunTextSessionRevisionConflictError,
)
from api.services.pricing.workflow_run_cost import (
apply_usage_delta_to_organization,
build_workflow_run_cost_info,
)
from api.services.workflow.text_chat_logs import (
build_text_chat_realtime_feedback_events,
)
from api.services.workflow.text_chat_runner import (
default_text_chat_checkpoint,
execute_text_chat_pending_turn,
merge_text_chat_usage_info,
normalize_text_chat_checkpoint,
)
TEXT_CHAT_SESSION_VERSION = 1
class TextChatSessionRevisionConflictError(Exception):
def __init__(self, expected_revision: int, actual_revision: int):
self.expected_revision = expected_revision
self.actual_revision = actual_revision
super().__init__(
"Text chat session revision conflict: "
f"expected {expected_revision}, found {actual_revision}"
)
class TextChatSessionExecutionError(Exception):
"""Raised when the assistant turn fails to execute."""
class TextChatPendingTurnLostError(Exception):
"""Raised when the pending turn disappears before persistence completes."""
class TextChatTurnNotFoundError(Exception):
"""Raised when a requested rewind cursor does not exist in the session."""
def default_text_chat_session_data() -> dict[str, Any]:
return {
"version": TEXT_CHAT_SESSION_VERSION,
"status": "idle",
"cursor_turn_id": None,
"turns": [],
"discarded_future": [],
"simulator": {
"enabled": False,
"config": {},
},
}
def normalize_text_chat_session_data(
session_data: dict[str, Any] | None,
) -> dict[str, Any]:
normalized = {
**default_text_chat_session_data(),
**(session_data or {}),
}
normalized["turns"] = list(normalized.get("turns") or [])
normalized["discarded_future"] = list(normalized.get("discarded_future") or [])
simulator = normalized.get("simulator") or {}
normalized["simulator"] = {
"enabled": bool(simulator.get("enabled", False)),
"config": dict(simulator.get("config") or {}),
}
return normalized
async def initialize_text_chat_session(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
) -> WorkflowRunTextSessionModel:
session_data = normalize_text_chat_session_data(text_session.session_data)
checkpoint = normalize_text_chat_checkpoint(text_session.checkpoint)
session_data["turns"] = [build_pending_text_chat_turn(user_text=None)]
session_data["status"] = "pending_assistant_turn"
checkpoint["anchor_turn_id"] = latest_completed_text_chat_turn_id(
session_data["turns"]
)
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
checkpoint=checkpoint,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
return await _reload_text_chat_session(run_id)
async def append_text_chat_user_message(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
user_text: str,
expected_revision: int | None,
) -> WorkflowRunTextSessionModel:
session_data = normalize_text_chat_session_data(text_session.session_data)
checkpoint = normalize_text_chat_checkpoint(text_session.checkpoint)
active_turns, discarded_future = truncate_text_chat_future_turns(session_data)
active_turns.append(build_pending_text_chat_turn(user_text=user_text))
session_data["turns"] = active_turns
session_data["discarded_future"] = discarded_future
session_data["cursor_turn_id"] = None
session_data["status"] = "pending_assistant_turn"
checkpoint["anchor_turn_id"] = latest_completed_text_chat_turn_id(active_turns)
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
checkpoint=checkpoint,
expected_revision=expected_revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
return await _reload_text_chat_session(run_id)
async def rewind_text_chat_session_state(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
cursor_turn_id: str | None,
expected_revision: int | None,
) -> WorkflowRunTextSessionModel:
session_data = normalize_text_chat_session_data(text_session.session_data)
validate_text_chat_turn_cursor(session_data, cursor_turn_id)
session_data["cursor_turn_id"] = cursor_turn_id
session_data["status"] = "rewound" if cursor_turn_id else "idle"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
expected_revision=expected_revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
await db_client.update_workflow_run(
run_id,
logs={
"realtime_feedback_events": build_text_chat_realtime_feedback_events(
session_data
)
},
)
return await _reload_text_chat_session(run_id)
async def execute_pending_text_chat_turn(
*,
workflow_id: int,
run_id: int,
text_session: WorkflowRunTextSessionModel,
) -> WorkflowRunTextSessionModel:
"""Execute the current pending assistant turn and persist its side effects."""
session_data = normalize_text_chat_session_data(text_session.session_data)
checkpoint = normalize_text_chat_checkpoint(text_session.checkpoint)
try:
execution = await execute_text_chat_pending_turn(
workflow_run_id=run_id,
workflow_id=workflow_id,
session_data=session_data,
checkpoint=checkpoint,
)
except Exception as e:
await _mark_pending_turn_failed(
run_id=run_id,
text_session=text_session,
error_message=str(e),
)
raise TextChatSessionExecutionError(
"Failed to execute text chat assistant turn"
) from e
completed_session_data = normalize_text_chat_session_data(text_session.session_data)
completed_turns = list(completed_session_data.get("turns") or [])
if not completed_turns or completed_turns[-1].get("status") != "pending":
raise TextChatPendingTurnLostError(
"Text chat session lost its pending turn before completion"
)
completed_turns[-1]["status"] = "completed"
completed_turns[-1]["assistant_message"] = (
{
"text": execution.assistant_text,
"created_at": execution.assistant_created_at,
}
if execution.assistant_text
else None
)
completed_turns[-1]["events"] = execution.events
completed_turns[-1]["usage"] = execution.usage
completed_turns[-1]["checkpoint_after_turn"] = execution.checkpoint
completed_session_data["turns"] = completed_turns
completed_session_data["status"] = "idle"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=completed_session_data,
checkpoint=execution.checkpoint,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
existing_usage_info = text_session.workflow_run.usage_info or {}
merged_usage_info = merge_text_chat_usage_info(existing_usage_info, execution.usage)
text_chat_logs = {
"realtime_feedback_events": build_text_chat_realtime_feedback_events(
completed_session_data
)
}
await db_client.update_workflow_run(
run_id,
initial_context=execution.initial_context,
usage_info=merged_usage_info,
gathered_context=execution.gathered_context,
logs=text_chat_logs,
state=execution.state,
is_completed=execution.is_completed,
)
workflow_run = await db_client.get_workflow_run_by_id(run_id)
if workflow_run:
try:
# Apply the per-turn delta so org usage tracks cumulative run cost
# without replaying the full session totals on every turn.
await apply_usage_delta_to_organization(workflow_run, execution.usage)
except Exception as e:
logger.error(
f"Failed to update organization usage for text chat run {run_id}: {e}"
)
cost_info = await build_workflow_run_cost_info(workflow_run)
if cost_info is not None:
await db_client.update_workflow_run(run_id, cost_info=cost_info)
return await _reload_text_chat_session(run_id)
def validate_text_chat_turn_cursor(
session_data: dict[str, Any],
cursor_turn_id: str | None,
) -> None:
if cursor_turn_id is None:
return
if not any(turn.get("id") == cursor_turn_id for turn in session_data["turns"]):
raise TextChatTurnNotFoundError("Turn not found in text chat session")
def truncate_text_chat_future_turns(
session_data: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
cursor_turn_id = session_data.get("cursor_turn_id")
turns = list(session_data.get("turns") or [])
discarded_future = list(session_data.get("discarded_future") or [])
if cursor_turn_id is None:
return turns, discarded_future
for index, turn in enumerate(turns):
if turn.get("id") == cursor_turn_id:
active_turns = turns[: index + 1]
future_turns = turns[index + 1 :]
if future_turns:
discarded_future.append(
{
"rewound_from_turn_id": cursor_turn_id,
"discarded_at": datetime.now(UTC).isoformat(),
"turns": future_turns,
}
)
return active_turns, discarded_future
raise TextChatTurnNotFoundError("Turn not found in text chat session")
def latest_completed_text_chat_turn_id(turns: list[dict[str, Any]]) -> str | None:
for turn in reversed(turns):
if turn.get("status") == "completed":
return turn.get("id")
return None
def build_pending_text_chat_turn(*, user_text: str | None) -> dict[str, Any]:
now = datetime.now(UTC).isoformat()
return {
"id": f"turn_{uuid4().hex[:12]}",
"status": "pending",
"created_at": now,
"user_message": (
{
"text": user_text,
"created_at": now,
}
if user_text is not None
else None
),
"assistant_message": None,
"events": [],
"usage": {},
}
async def _mark_pending_turn_failed(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
error_message: str,
) -> None:
failed_session_data = normalize_text_chat_session_data(text_session.session_data)
failed_turns = list(failed_session_data.get("turns") or [])
if not failed_turns or failed_turns[-1].get("status") != "pending":
return
failed_turns[-1]["status"] = "failed"
failed_turns[-1]["events"] = [
*(failed_turns[-1].get("events") or []),
{
"type": "execution_error",
"created_at": datetime.now(UTC).isoformat(),
"payload": {"message": error_message},
},
]
failed_session_data["turns"] = failed_turns
failed_session_data["status"] = "error"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=failed_session_data,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError:
return
async def _reload_text_chat_session(run_id: int) -> WorkflowRunTextSessionModel:
organization_id = await db_client.get_organization_id_by_workflow_run_id(run_id)
if organization_id is None:
raise TextChatSessionExecutionError(
"Workflow run organization not found after update"
)
updated_text_session = await db_client.get_workflow_run_text_session(
run_id,
organization_id=organization_id,
)
if updated_text_session is None:
raise TextChatSessionExecutionError("Text chat session not found after update")
return updated_text_session
__all__ = [
"TEXT_CHAT_SESSION_VERSION",
"TextChatTurnNotFoundError",
"append_text_chat_user_message",
"build_pending_text_chat_turn",
"TextChatPendingTurnLostError",
"TextChatSessionExecutionError",
"TextChatSessionRevisionConflictError",
"default_text_chat_checkpoint",
"default_text_chat_session_data",
"execute_pending_text_chat_turn",
"initialize_text_chat_session",
"latest_completed_text_chat_turn_id",
"normalize_text_chat_checkpoint",
"normalize_text_chat_session_data",
"rewind_text_chat_session_state",
"truncate_text_chat_future_turns",
"validate_text_chat_turn_cursor",
]