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

chore: code refactor

Browse source
This commit is contained in:
Abhishek Kumar 2026-02-14 13:43:20 +05:30
parent c0cbc65de3
commit c7812bf189
29 changed files with 538 additions and 800 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -108,13 +108,16 @@ class PipecatEngine:
# Custom tool manager (initialized in initialize())
self._custom_tool_manager: Optional[CustomToolManager] = None
# Tracks whether a call transfer is in progress
self._transferring_call: bool = False
# Embeddings configuration (passed from run_pipeline.py)
self._embeddings_api_key: Optional[str] = embeddings_api_key
self._embeddings_model: Optional[str] = embeddings_model
self._embeddings_base_url: Optional[str] = embeddings_base_url
# Transfer state tracking - prevents auto hang-up during call transfers
self._transfer_in_progress: bool = False
# Audio configuration (set via set_audio_config from _run_pipeline)
self._audio_config = None
async def _get_organization_id(self) -> Optional[int]:
"""Get and cache the organization ID from workflow run."""
@ -248,7 +251,7 @@ class PipecatEngine:
await function_call_params.result_callback(
result, properties=properties
)
except Exception as e:
logger.error(f"Error in transition function {name}: {str(e)}")
error_result = {"status": "error", "error": str(e)}
@ -281,7 +284,7 @@ class PipecatEngine:
async def calculate_func(function_call_params: FunctionCallParams) -> None:
logger.info(f"LLM Function Call EXECUTED: safe_calculator")
logger.info(f"Arguments: {function_call_params.arguments}")
try:
expr = function_call_params.arguments.get("expression", "")
result = safe_calculator(expr)
@ -297,7 +300,7 @@ class PipecatEngine:
) -> None:
logger.info(f"LLM Function Call EXECUTED: get_current_time")
logger.info(f"Arguments: {function_call_params.arguments}")
try:
timezone = function_call_params.arguments.get("timezone", "UTC")
result = get_current_time(timezone)
@ -308,7 +311,7 @@ class PipecatEngine:
async def convert_time_func(function_call_params: FunctionCallParams) -> None:
logger.info(f"LLM Function Call EXECUTED: convert_time")
logger.info(f"Arguments: {function_call_params.arguments}")
try:
result = convert_time(
function_call_params.arguments.get("source_timezone"),
@ -339,7 +342,7 @@ class PipecatEngine:
async def retrieve_kb_func(function_call_params: FunctionCallParams) -> None:
logger.info("LLM Function Call EXECUTED: retrieve_from_knowledge_base")
logger.info(f"Arguments: {function_call_params.arguments}")
try:
query = function_call_params.arguments.get("query", "")
organization_id = await self._get_organization_id()
@ -540,7 +543,9 @@ class PipecatEngine:
self._current_node, run_in_background=False
)
frame_to_push = CancelFrame(reason=reason) if abort_immediately else EndFrame(reason=reason)
frame_to_push = (
CancelFrame(reason=reason) if abort_immediately else EndFrame(reason=reason)
)
# Apply disposition mapping - first try call_disposition if it is,
# extracted from the call conversation then fall back to reason
@ -713,6 +718,10 @@ class PipecatEngine:
f"Stasis connection set for immediate transfers: {connection.channel_id}"
)
def set_audio_config(self, audio_config) -> None:
"""Set the audio configuration for the pipeline."""
self._audio_config = audio_config
def set_mute_pipeline(self, mute: bool) -> None:
"""Set the pipeline mute state.
@ -725,6 +734,15 @@ class PipecatEngine:
logger.debug(f"Setting pipeline mute state to: {mute}")
self._mute_pipeline = mute
def set_transferring_call(self, transferring: bool) -> None:
"""Set the call transfer state.
Args:
transferring: True when a call transfer is in progress, False otherwise
"""
logger.debug(f"Setting transferring call state to: {transferring}")
self._transferring_call = transferring
async def handle_llm_text_frame(self, text: str):
"""Accumulate LLM text frames to build reference text."""
self._current_llm_generation_reference_text += text

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

@ -8,14 +8,17 @@ from __future__ import annotations
import asyncio
import re
import time
import uuid
from typing import TYPE_CHECKING, Any, Optional
import aiohttp
import httpx
from loguru import logger
from api.db import db_client
from api.enums import ToolCategory
from api.services.telephony.call_transfer_manager import get_call_transfer_manager
from api.services.telephony.factory import get_telephony_provider
from api.services.telephony.transfer_event_protocol import TransferContext
from api.services.workflow.disposition_mapper import (
get_organization_id_from_workflow_run,
)
@ -24,25 +27,15 @@ from api.services.workflow.tools.custom_tool import (
execute_http_tool,
tool_to_function_schema,
)
from api.utils.hold_audio import load_hold_audio
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.frames.frames import (
FunctionCallResultProperties,
TTSSpeakFrame,
OutputAudioRawFrame,
TTSSpeakFrame,
)
from pipecat.services.llm_service import FunctionCallParams
from pipecat.utils.enums import EndTaskReason
from pipecat.transports.websocket.fastapi import FastAPIWebsocketClient
from api.utils.hold_audio import load_hold_audio
from api.services.telephony.call_transfer_manager import get_call_transfer_manager
from api.services.telephony.transfer_event_protocol import (
TransferEvent,
TransferContext,
TransferEventType,
)
from dograh.api.utils.common import get_backend_endpoints
if TYPE_CHECKING:
from api.services.workflow.pipecat_engine import PipecatEngine
@ -134,8 +127,15 @@ class CustomToolManager:
function_name = schema["function"]["name"]
# Create and register the handler
handler = self._create_handler(tool, function_name)
self._engine.llm.register_function(function_name, handler)
handler, disable_timeout, cancel_on_interruption = self._create_handler(
tool, function_name
)
self._engine.llm.register_function(
function_name,
handler,
cancel_on_interruption=cancel_on_interruption,
disable_timeout=disable_timeout,
)
logger.debug(
f"Registered custom tool handler: {function_name} "
@ -155,12 +155,21 @@ class CustomToolManager:
Returns:
Async handler function for the tool
"""
if tool.category == ToolCategory.END_CALL.value:
return self._create_end_call_handler(tool, function_name)
elif tool.category == ToolCategory.TRANSFER_CALL.value:
return self._create_transfer_call_handler(tool, function_name)
# Whether to disable function call timeout
disable_timeout = False
cancel_on_interruption = True
return self._create_http_tool_handler(tool, function_name)
if tool.category == ToolCategory.END_CALL.value:
cancel_on_interruption = False
handler = self._create_end_call_handler(tool, function_name)
elif tool.category == ToolCategory.TRANSFER_CALL.value:
disable_timeout = True
cancel_on_interruption = False
handler = self._create_transfer_call_handler(tool, function_name)
else:
handler = self._create_http_tool_handler(tool, function_name)
return handler, disable_timeout, cancel_on_interruption
def _create_http_tool_handler(self, tool: Any, function_name: str):
"""Create a handler function for an HTTP API tool.
@ -314,14 +323,6 @@ class CustomToolManager:
logger.info(f"Playing pre-transfer message: {custom_message}")
await self._engine.task.queue_frame(TTSSpeakFrame(custom_message))
# Get original call information from Pipecat context
from pipecat.utils.run_context import get_current_call_sid
original_call_sid = get_current_call_sid()
caller_number = None # TODO: check if this is redundant now
logger.info(f"Found original call context: call_id={original_call_sid}")
# Get organization ID for provider configuration
organization_id = await self.get_organization_id()
if not organization_id:
@ -336,198 +337,143 @@ class CustomToolManager:
validation_error_result, function_call_params, properties
)
return
#TODO: check if everything in transfer_data is still needed
# Prepare transfer request data
transfer_data = {
"destination": destination,
"organization_id": organization_id, # Required for provider configuration
"tool_call_id": function_call_params.tool_call_id, # Use LLM's tool call ID for pipeline coordination
"tool_uuid": tool.tool_uuid, # Add tool UUID for tracing and validation
"original_call_sid": original_call_sid, # Original caller's call SID
"caller_number": caller_number, # Original caller's phone number
}
# Get telephony provider directly (no HTTP round-trip)
provider = await get_telephony_provider(organization_id)
if not provider.supports_transfers() or not provider.validate_config():
validation_error_result = {
"status": "failed",
"message": "I'm sorry, there's an issue with this call transfer. Please contact support.",
"action": "transfer_failed",
"reason": "provider_does_not_support_transfer",
"end_call": False,
}
await self._handle_transfer_result(
validation_error_result, function_call_params, properties
)
return
import time
# Get original callSID from gathered_context
workflow_run = await db_client.get_workflow_run_by_id(
self._engine._workflow_run_id
)
original_call_sid = workflow_run.gathered_context.get("call_id")
# Get backend endpoint URL
backend_url, _ = await get_backend_endpoints()
# Generate a unique transfer ID for tracking this transfer
transfer_id = str(uuid.uuid4())
# Get transfer coordinator for Redis-based coordination
# Compute conference name from original call SID
conference_name = f"transfer-{original_call_sid}"
# Mark transfer in progress and mute the pipeline
self._engine.set_transferring_call(True)
self._engine.set_mute_pipeline(True)
# Initiate transfer via provider with inline TwiML
transfer_result = await provider.transfer_call(
destination=destination,
transfer_id=transfer_id,
conference_name=conference_name,
timeout=timeout_seconds,
)
call_sid = transfer_result.get("call_sid")
logger.info(f"Transfer call initiated successfully: {call_sid}")
# TODO: Possible race here between saving the transfer context
# and getting a callback response from Twilio? Should we store_transfer_context
# before sending request to Twilio and update the transfer context afterwards?
# Store transfer context in Redis
call_transfer_manager = await get_call_transfer_manager()
transfer_context = TransferContext(
transfer_id=transfer_id,
call_sid=call_sid,
target_number=destination,
tool_uuid=tool.tool_uuid,
original_call_sid=original_call_sid,
conference_name=conference_name,
initiated_at=time.time(),
)
await call_transfer_manager.store_transfer_context(transfer_context)
# Now initiate the transfer call
transfer_url = f"{backend_url}/api/v1/telephony/call-transfer"
# Wait for status callback completion using Redis pub/sub
logger.info(
f"Transfer call initiated for {destination} (transfer_id={transfer_id}), waiting for completion..."
)
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
transfer_url,
json=transfer_data,
headers={"Content-Type": "application/json"},
# Authentication headers added by provider if needed
# Start hold music during transfer waiting period
hold_music_stop_event = asyncio.Event()
hold_music_task = None
try:
# Use audio config for sample rate (set during pipeline setup)
sample_rate = (
self._engine._audio_config.transport_out_sample_rate
if self._engine._audio_config
else 8000
)
if response.status_code == 200:
result_data = response.json()
logger.info(f"Transfer initiated successfully: {result_data}")
logger.info(
f"Starting hold music at {sample_rate}Hz while waiting for transfer"
)
# Wait for webhook completion using standard Pipecat async pattern
logger.info(
f"Transfer call initiated for {destination}, waiting for webhook completion..."
# Start hold music as background task
hold_music_task = asyncio.create_task(
self.play_hold_music_loop(hold_music_stop_event, sample_rate)
)
# Wait for transfer completion using Redis pub/sub
logger.info("Waiting for transfer completion via Redis pub/sub...")
transfer_event = (
await call_transfer_manager.wait_for_transfer_completion(
transfer_id, timeout_seconds
)
)
# Start hold music during transfer waiting period
hold_music_stop_event = asyncio.Event()
hold_music_task = None
except Exception as e:
logger.error(f"Error during transfer wait: {e}")
transfer_event = None
try:
# Mute the pipeline to prevent further LLM generations during transfer
logger.info("Muting pipeline during transfer call")
self._engine.set_mute_pipeline(True)
finally:
# Single cleanup point: stop hold music, unmute pipeline, remove context
logger.info(
"Transfer wait ended, cleaning up hold music, pipeline state, and transfer context"
)
hold_music_stop_event.set()
if hold_music_task:
await hold_music_task
self._engine.set_transferring_call(False)
self._engine.set_mute_pipeline(False)
await call_transfer_manager.remove_transfer_context(transfer_id)
# Determine sample rate from transport (default to 8000Hz for Twilio)
sample_rate = 8000
if hasattr(self._engine.transport, "output") and hasattr(
self._engine.transport.output(), "sample_rate"
):
sample_rate = getattr(
self._engine.transport.output(), "sample_rate", 8000
)
logger.info(
f"Starting hold music at {sample_rate}Hz while waiting for transfer"
)
# Start hold music as background task
hold_music_task = asyncio.create_task(
self.play_hold_music_loop(
hold_music_stop_event, sample_rate
)
)
# Wait for transfer completion using Redis pub/sub
logger.info(
"Waiting for transfer completion via Redis pub/sub..."
)
transfer_event = (
await call_transfer_manager.wait_for_transfer_completion(
transfer_data["tool_call_id"], timeout_seconds
)
)
# Stop hold music and unmute pipeline
logger.info(
"Transfer completed, stopping hold music and unmuting pipeline"
)
hold_music_stop_event.set()
if hold_music_task:
await hold_music_task
self._engine.set_mute_pipeline(False)
if transfer_event:
# Get result from transfer event
final_result = transfer_event.to_result_dict()
# Get transfer context for caller number
transfer_context = (
await call_transfer_manager.get_transfer_context(
transfer_data["tool_call_id"]
)
)
if transfer_context and transfer_context.caller_number:
final_result["caller_number"] = (
transfer_context.caller_number
)
# Handle the transfer result and inform user appropriately
await self._handle_transfer_result(
final_result, function_call_params, properties
)
else:
# Handle timeout case
logger.error(
f"Transfer call timed out after {timeout_seconds} seconds"
)
# Create timeout result and handle it through the same flow
timeout_result = {
"status": "failed",
"message": "I'm sorry, but the call is taking longer than expected to connect. The person might not be available right now. Please try calling back later.",
"action": "transfer_failed",
"reason": "timeout",
"end_call": True,
}
await self._handle_transfer_result(
timeout_result, function_call_params, properties
)
except Exception as e:
logger.error(f"Error during transfer wait: {e}")
# Stop hold music and unmute pipeline on error
logger.info(
"Transfer error, stopping hold music and unmuting pipeline"
)
hold_music_stop_event.set()
if hold_music_task:
await hold_music_task
self._engine.set_mute_pipeline(False)
# Handle error case
error_result = {
"status": "failed",
"message": "I'm sorry, but there was an issue processing the transfer. Please try again.",
"action": "transfer_failed",
"reason": "system_error",
"end_call": True,
}
await self._handle_transfer_result(
error_result, function_call_params, properties
)
else:
error_data = (
response.json()
if response.content
else {"error": "Unknown error"}
)
logger.error(
f"Transfer initiation failed: {response.status_code} - {error_data}"
)
# Handle initiation failure with user-friendly message
initiation_failure_result = {
"status": "failed",
"message": "I'm sorry, but I'm having trouble setting up the call transfer right now. There might be a technical issue. Please try again later or contact support.",
"action": "transfer_failed",
"reason": "initiation_failed",
"end_call": True,
}
await self._handle_transfer_result(
initiation_failure_result, function_call_params, properties
)
except httpx.TimeoutException:
logger.error(f"Transfer call '{function_name}' HTTP request timed out")
# Handle HTTP timeout with user-friendly message
http_timeout_result = {
"status": "failed",
"message": "I'm sorry, but there seems to be a network issue preventing me from setting up the call transfer. Please try again in a moment.",
"action": "transfer_failed",
"reason": "network_timeout",
"end_call": True,
}
await self._handle_transfer_result(
http_timeout_result, function_call_params, properties
)
# Handle result (after cleanup)
if transfer_event:
final_result = transfer_event.to_result_dict()
await self._handle_transfer_result(
final_result, function_call_params, properties
)
else:
logger.error(
f"Transfer call timed out or failed after {timeout_seconds} seconds"
)
timeout_result = {
"status": "failed",
"message": "I'm sorry, but the call is taking longer than expected to connect. The person might not be available right now. Please try calling back later.",
"action": "transfer_failed",
"reason": "timeout",
"end_call": True,
}
await self._handle_transfer_result(
timeout_result, function_call_params, properties
)
except Exception as e:
logger.error(
f"Transfer call tool '{function_name}' execution failed: {e}"
)
self._engine.set_transferring_call(False)
self._engine.set_mute_pipeline(False)
# Handle generic exception with user-friendly message
exception_result = {
@ -550,8 +496,6 @@ class CustomToolManager:
"""Handle different transfer call outcomes and take appropriate action."""
action = result.get("action", "")
status = result.get("status", "")
message = result.get("message", "")
should_end_call = result.get("end_call", False)
logger.info(f"Handling transfer result: action={action}, status={status}")
@ -566,9 +510,7 @@ class CustomToolManager:
)
# Inform LLM of success and end the call with Transfer call reason
response_properties = FunctionCallResultProperties(
run_llm=False
)
response_properties = FunctionCallResultProperties(run_llm=False)
await function_call_params.result_callback(
{
"status": "transfer_success",
@ -585,83 +527,19 @@ class CustomToolManager:
elif action == "transfer_failed":
# Transfer failed - inform user via LLM and then end the call
reason = result.get("reason", "unknown")
logger.info(f"Transfer failed ({reason}), informing user and ending call")
logger.info(f"Transfer failed ({reason}), informing user")
from pipecat.frames.frames import LLMMessagesAppendFrame
# Create system message with clear instructions for transfer failure
failure_instruction = {
"role": "system",
"content": f"IMPORTANT: The transfer call has FAILED. Reason: {reason}. You must inform the customer about this failure using this message: '{message}' Then immediately say goodbye and end the conversation. Do NOT ask if they need anything else or continue the conversation. Do NOT continue with transfer language.",
}
# Push the system message to LLM context
await self._engine.task.queue_frame(
LLMMessagesAppendFrame([failure_instruction], run_llm=True)
)
# Also send the function call result for consistency
response_properties = FunctionCallResultProperties(
run_llm=False
) # LLM will be triggered by system message
await function_call_params.result_callback(
{"status": "transfer_failed", "reason": reason, "message": message},
properties=response_properties,
{
"status": "transfer_failed",
"reason": reason,
"message": "Transfer failed",
}
)
# Set appropriate disposition for analytics
disposition_map = {
"no_answer": "transfer_no_answer",
"busy": "transfer_busy",
"call_failed": "transfer_failed",
"timeout": "transfer_timeout",
"no_destination": "transfer_config_error",
"invalid_destination": "transfer_config_error",
"initiation_failed": "transfer_system_error",
"network_timeout": "transfer_system_error",
"execution_error": "transfer_system_error",
}
disposition = disposition_map.get(reason, "transfer_failed")
logger.info(
f"Setting disposition: {disposition} for transfer failure reason: {reason}"
)
# Give the LLM time to speak the message, then end the call with disposition
# We'll schedule the end call after a brief delay to allow TTS
logger.info("Scheduling call end after LLM delivers failure message")
import asyncio
# Schedule call end after 3 seconds to allow LLM to speak
async def delayed_end_call():
import asyncio
await asyncio.sleep(3)
await self._engine.end_call_with_reason(
f"transfer_failed_{reason}", # Include specific reason in end reason
abort_immediately=False, # Allow any queued speech to complete
)
# Create task to end call asynchronously
asyncio.create_task(delayed_end_call())
elif action == "transfer_completed":
# This should no longer happen since we ignore "completed" status in webhook
# to avoid overriding successful transfers
logger.warning(
"Received unexpected 'transfer_completed' action - this should be ignored by webhook now"
)
logger.warning(
"If you see this message, there might be an issue with the webhook status filtering"
)
# For safety, treat it as a generic result without ending the call
await function_call_params.result_callback(result, properties=properties)
else:
# Unknown action, treat as generic success
logger.warning(f"Unknown transfer action: {action}, treating as success")
await function_call_params.result_callback(result, properties=properties)
await function_call_params.result_callback(result)
async def play_hold_music_loop(
self, stop_event: asyncio.Event, sample_rate: int = 8000