improve task creation and cancellation

[aconchillo]

Please describe the changes in your PR. If it is addressing an issue, please reference that as well.

If a FrameProcessor needs to create a task it should use FrameProcessor.create_task() and FrameProcessor.cancel_task(). This gives Pipecat more control over all the tasks that are created in Pipecat.

Both functions internally use the utils module: utils.create_task() and utils.cancel_task() which should also be used outside of FrameProcessors. That is, unless strictly necessary, we should avoid using asyncio.create_task().

[markbackman]

This is really great! I've found two issues so far:

1. This is a regression which causes overlapping speech. To repro, run example 14-function-calling.py, where you set the start_fetch_weather function to:

async def start_fetch_weather(function_name, llm, context):
    """Push a frame to the LLM; this is handy when the LLM response might take a while."""
    await llm.push_frame(
        TextFrame(
            "Let me check on that. One moment while I pull up the weather for you. Ok, got it."
        )
    )
    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")

This will result in the TTS speech overlapping—both the TextFrame and LLM text will be spoken at the same time. This isn’t happening on main.

2. I think this revealed an issue with the GeminiMultimodalLiveLLMService. I ran the simple-chatbot demo using the bot-gemini.py bot file and a user connecting via a javascript client. Upon the user leaving, I got this warning:

2025-01-25 08:59:48.521 | WARNING  | pipecat.pipeline.runner:_print_dangling_tasks:63 - Dangling tasks detected: ['GeminiMultimodalLiveLLMService#0::_handle_transcribe_user_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_user_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_model_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_user_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_model_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_user_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_model_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_model_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_user_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_model_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_model_audio', 'GeminiMultimodalLiveLLMService#0::_handle_transcribe_user_audio']

[markbackman]

Also, in running the twilio-chatbot example, I get some version of this frequently:

2025-01-25 09:32:57.674 | WARNING  | pipecat.pipeline.runner:_print_dangling_tasks:63 - Dangling tasks detected: ['FastAPIWebsocketInputTransport#0::_receive_messages']

Should we be concerned about dangling tasks?