Files
ai-video-fullstack/backend/services/pipecat/call_lifecycle.py
Xin Wang 00270a5c01 Add Dify integration and enhance workflow node specifications
- Introduce new fields `dify_api_url` and `dify_api_key` in `AssistantConfig` for Dify API integration.
- Update `requirements.txt` to include `dify-client-python` for Dify SDK support.
- Modify `config_resolver` to handle Dify connection information.
- Add a new `globalNode` type in workflow specifications to provide unified settings across workflows.
- Enhance node specifications with additional constraints and default values for better configuration management.
- Update frontend components to support the new `globalNode` type and its properties, improving workflow editor functionality.
2026-07-11 22:26:31 +08:00

61 lines
1.9 KiB
Python

"""Shared call termination timing for prompt tools and workflow end nodes."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from loguru import logger
from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class CallEndCoordinator:
"""End immediately or after the currently armed closing speech finishes."""
def __init__(self, queue_end: Callable[[str], Awaitable[None]]):
self._queue_end = queue_end
self._ending = False
self._armed = False
self._speaking = False
self._finished = False
self._reason = "completed"
@property
def ending(self) -> bool:
return self._ending
def begin(self, reason: str) -> None:
self._ending = True
self._reason = reason or "completed"
def arm_after_speech(self) -> None:
self._armed = True
async def finish(self) -> None:
if self._finished:
return
self._finished = True
await self._queue_end(self._reason)
async def observe(self, frame) -> None:
if isinstance(frame, BotStartedSpeakingFrame) and self._armed:
self._speaking = True
elif (
isinstance(frame, BotStoppedSpeakingFrame)
and self._armed
and self._speaking
):
logger.info("结束语播报完毕,挂断通话")
await self.finish()
class EndCallAfterSpeechProcessor(FrameProcessor):
def __init__(self, coordinator: CallEndCoordinator):
super().__init__()
self._coordinator = coordinator
async def process_frame(self, frame, direction: FrameDirection):
await super().process_frame(frame, direction)
await self.push_frame(frame, direction)
await self._coordinator.observe(frame)