Python: Adding FunctionCallContent and FunctionResultContent while streaming messages

[bbence84]

Hi!
This is a followup of the issue #9408, which has been addressed with a fix. Now the kernel yields both FunctionCallContent and FunctionResultContent. It's possible to combine the streaming message chunks, to get these, but only AFTER the streaming is completed:

    if streamed_chunks:
        streaming_chat_message = reduce(lambda first, second: first + second, streamed_chunks)

But my problem now is that I can't seem to find a way to do this while the streaming is still in progress. The reason why I need this is that my agent is handling user requests that would call multiple functions after each other, the output of some functions is the input of followup function calls. This works pretty well if I emit the function call result in the chat history using a function_invocation filter. But there I am just putting the function call output as an ASSISTANT message, as I couldn't figure out the way to "convert" the context to KernelContent (I suppose I need that to add a TOOL_CALL message to the history).

Anyway, my issue is how to "detect" if a FunctionCallContent and FunctionResultContent stream is "finished", so that I could add them right away to the message history.

Thanks!

[bbence84]

Well, I kind of managed to solve it, though this looks quite hacky. Not sure if there's a "cleaner" way. Basically my concern is that I am assuming that once I get a FunctionResultContent item, there will be a full FunctionCallContent right before. Then when there's a FunctionResultContent, only then I add both to the chat history with add_assistant_message_list and add_tool_message_list. I had some weird exceptions sometimes, but it turned out the problem was caused by parellel function calls enabled by default, so I disabled it. Now it seems to work well.

    answer = await get_ai_response_stream(user_input) 

    streamed_chunks: list[StreamingChatMessageContent] = []
    streamed_chunks_func_call: list[StreamingChatMessageContent] = []
    streamed_chunks_func_result: list[StreamingChatMessageContent] = []    

    chat_history.add_user_message_str(user_input) 

    prev_chat_message = None

    async for message in answer:

        chat_message = message[0]
        items = chat_message.items 
        for i, item in enumerate(items, start=1): 

            if isinstance(item, FunctionCallContent):
                streamed_chunks_func_call.append(chat_message) 
                prev_chat_message = chat_message

            if isinstance(item, FunctionResultContent):
                if len(streamed_chunks_func_call) > 0:
                    streaming_chat_message_func_call = reduce(lambda first, second: first + second, streamed_chunks_func_call)
                    chat_history.add_assistant_message_list(streaming_chat_message_func_call.items)
                
                streamed_chunks_func_result.append(chat_message) 
                if len(streamed_chunks_func_result) > 0:
                    streaming_chat_message_func_result = reduce(lambda first, second: first + second, streamed_chunks_func_result)
                    chat_history.add_tool_message_list(streaming_chat_message_func_result.items)

                streamed_chunks_func_call = []
                streamed_chunks_func_result = []

            if isinstance(item, StreamingChatMessageContent) and chat_message.role == AuthorRole.ASSISTANT:
                streamed_chunks.append(chat_message) 
                prev_chat_message = chat_message

        if token := str(chat_message) or "":
             await msg.stream_token(token)

    if streamed_chunks:
        streaming_chat_message = reduce(lambda first, second: first + second, streamed_chunks)
        chat_history.add_assistant_message_str(str(streaming_chat_message.items[0]))
        

[moonbox3]

If what you have works, then that's great. Another approach, that is slightly different could be to keep local buffers for each type of content and finalize (add to the chat history) only when we detect that a chunk is fully received (when we get the FunctionResultContent we know the FunctionCallContent has ended). This method also avoids having to repeatedly call reduce(...) in different places and tries to make the code flow more self-explanatory.

async def stream_and_accumulate(answer_stream, user_input, chat_history, msg):
    """An example state-based approach to handle function calls/results as they stream."""
    chat_history.add_user_message(user_input)

    # Temporary lists where we accumulate function call chunks and function result chunks
    func_call_chunks = []
    func_result_chunks = []
    assistant_chunks = []

    async for message in answer_stream:
        chat_message = message[0]

        for item in chat_message.items:
            match item:
                case FunctionCallContent():
                    func_call_chunks.append(item)

                case FunctionResultContent():
                    func_result_chunks.append(item)
                    
                    # As soon as we see a function result, that typically means
                    # the previous function call is "done." So finalize them:
                    if func_call_chunks:
                        # Turn the accumulated function-call chunks into a single message
                        all_func_call = _combine_chunks_into_chat_message(
                            func_call_chunks, 
                            role=AuthorRole.ASSISTANT
                        )
                        chat_history.add_assistant_message_list(all_func_call.items)
                        func_call_chunks.clear()

                    # Now also finalize the function-result chunks themselves
                    if func_result_chunks:
                        all_func_result = _combine_chunks_into_chat_message(
                            func_result_chunks, 
                            role=AuthorRole.TOOL
                        )
                        chat_history.add_tool_message_list(all_func_result.items)
                        func_result_chunks.clear()

                case StreamingChatMessageContent() if chat_message.role == AuthorRole.ASSISTANT:
                    assistant_chunks.append(item)

        # In parallel, stream tokens to the user interface (if needed)
        token = str(chat_message)
        if token:
            await msg.stream_token(token)

    # Once the streaming completes, if there's leftover "assistant" text,
    # combine it into a single message and add it to the chat.
    if assistant_chunks:
        combined_assistant_msg = _combine_chunks_into_chat_message(
            assistant_chunks, 
            role=AuthorRole.ASSISTANT
        )
        chat_history.add_assistant_message(str(combined_assistant_msg))
        assistant_chunks.clear()


def _combine_chunks_into_chat_message(chunks, role):
    # Example approach: just merge the text pieces from these chunks.
    # or do the reduce(...) logic here to keep the full messages
    combined_text = "".join(str(chunk) for chunk in chunks)
    return ChatMessageContent(role=role, content=combined_text)

[moonbox3]

Your solution is good, and I provided another. Please ping on the issue if you need further help.