Extend `streaming_generator` to also return the Pipeline result

[anakin87]

Currently, Hayhooks provides streaming_generator and async_streaming_generator utility functions that can be used to stream the pipeline chunks to the client. Info.

These functions work by attaching a streaming_callback to the components in the Pipeline that support streaming, and then streaming the generated chunks.

Even if the Pipeline is executed, the final result is not returned.

In some cases, it would be helpful to have access to both the streamed output and the final result.

Think for example of a RAG pipeline where we also want to show users the original source documents (coming from a Retriever or an AnswerBuilder).

[DaAitch]

I ran into the same problem and studied the async_streaming_generator impl and rolled my own generator that will forward StreamingChunk as well as being aware of callback interceptors where you can also transform pipeline results into events. That is very important for me for convenience for RAG pipelines, where the first token takes some time and I owe the user some information what's happening.

Also, I wasn't too happy with that pipeline discovery of streaming_callback. It's very invasive and implicit, in my opinion, so I thought of a more open solution. While I think it's now "open", it's not yet very composable if you have mutiple streaming_callbacks.

This is my current interface. I thinking about opening a PR about it, but I don't have tests. So sb. helping would be welcome and I think the solution should also solve multiple streaming_callback parameters.

What do you think?

# chat_completion.py
class AsyncStreamingChatCompletion:
    """
    Bring together an async streaming chat completion and pipeline events that
    can be mapped to a list of OpenWebUIEvent items.

    ```
        cc = AsyncStreamingChatCompletion()
        data = {
            "component1": {"question": question},
            "llm": {"streaming_callback": cc.streaming_callback},
        }

        return cc.chat_completion(
            self.pipeline.run_async_generator(data, {"component1", "component2", "llm"}),
            pipeline_callback=pipeline_callback,
            pipeline_error_callback=pipeline_error_callback,
        )
    ```
    """

    async def chat_completion(
        self,
        pipeline: AsyncIterator[Dict[str, Any]],
        *,
        pipeline_callback: Callable[[Dict[str, Any] | None], List[OpenWebUIEvent] | OpenWebUIEvent | None]
        | None = None,
        pipeline_error_callback: Callable[[Exception], List[OpenWebUIEvent] | OpenWebUIEvent | None] | None = None,
    ) -> AsyncGenerator[OpenWebUIEvent | StreamingChunk, None]:
        ...

# pipeline_wrapper.py
def pipeline_callback(result: Dict[str, Any] | None) -> List[OpenWebUIEvent] | OpenWebUIEvent | None:
    # end of the pipeline
    if result is None:
        return create_status_event("🏁 Done", done=True)
    # ..