diff --git a/python/semantic_kernel/connectors/ai/chat_completion_client_base.py b/python/semantic_kernel/connectors/ai/chat_completion_client_base.py index 974d59af92be..611a464d963b 100644 --- a/python/semantic_kernel/connectors/ai/chat_completion_client_base.py +++ b/python/semantic_kernel/connectors/ai/chat_completion_client_base.py @@ -16,6 +16,7 @@ from semantic_kernel.contents.annotation_content import AnnotationContent from semantic_kernel.contents.file_reference_content import FileReferenceContent from semantic_kernel.contents.function_call_content import FunctionCallContent +from semantic_kernel.contents.text_content import TextContent from semantic_kernel.exceptions.service_exceptions import ServiceInvalidExecutionSettingsError from semantic_kernel.services.ai_service_client_base import AIServiceClientBase from semantic_kernel.utils.telemetry.model_diagnostics.gen_ai_attributes import AVAILABLE_FUNCTIONS @@ -135,16 +136,18 @@ async def get_chat_message_contents( # Auto invoke loop with use_span(self._start_auto_function_invocation_activity(kernel, settings), end_on_exit=True) as _: + function_call_messages: list["ChatMessageContent"] = [] for request_index in range(settings.function_choice_behavior.maximum_auto_invoke_attempts): completions = await self._inner_get_chat_message_contents(chat_history, settings) # Get the function call contents from the chat message. There is only one chat message, # which should be checked in the `_verify_function_choice_settings` method. function_calls = [item for item in completions[0].items if isinstance(item, FunctionCallContent)] if (fc_count := len(function_calls)) == 0: - return completions + return self._combine_auto_invoke_text_responses(completions, function_call_messages) # Since we have a function call, add the assistant's tool call message to the history chat_history.add_message(message=completions[0]) + function_call_messages.append(completions[0]) logger.info(f"processing {fc_count} tool calls in parallel.") @@ -167,11 +170,15 @@ async def get_chat_message_contents( ) if any(result.terminate for result in results if result is not None): - return merge_function_results(chat_history.messages[-len(results) :]) + return self._combine_auto_invoke_text_responses( + merge_function_results(chat_history.messages[-len(results) :]), + function_call_messages, + ) else: # Do a final call, without function calling when the max has been reached. self._reset_function_choice_settings(settings) - return await self._inner_get_chat_message_contents(chat_history, settings) + completions = await self._inner_get_chat_message_contents(chat_history, settings) + return self._combine_auto_invoke_text_responses(completions, function_call_messages) async def get_chat_message_content( self, chat_history: "ChatHistory", settings: "PromptExecutionSettings", **kwargs: Any @@ -377,6 +384,43 @@ def _prepare_chat_history_for_request( if not isinstance(message, (AnnotationContent, FileReferenceContent)) ] + def _combine_auto_invoke_text_responses( + self, + completions: list["ChatMessageContent"], + function_call_messages: list["ChatMessageContent"], + ) -> list["ChatMessageContent"]: + """Prepend text from intermediate function-call responses to the final response.""" + text_parts = [ + item.text + for message in function_call_messages + for item in message.items + if isinstance(item, TextContent) and item.text + ] + if not text_parts: + return completions + + prefix_text = "\n\n".join(text_parts) + combined_completions = copy.deepcopy(completions) + for completion in combined_completions: + text_item = next((item for item in completion.items if isinstance(item, TextContent)), None) + if text_item is None: + completion.items.insert( + 0, + TextContent( + ai_model_id=completion.ai_model_id, + inner_content=completion.inner_content, + metadata=completion.metadata, + text=prefix_text, + encoding=completion.encoding, + ), + ) + continue + + text_item.text = "\n\n".join(part for part in [prefix_text, text_item.text] if part) + text_item.encoding = completion.encoding + + return combined_completions + def _verify_function_choice_settings(self, settings: "PromptExecutionSettings") -> None: """Additional verification to validate settings for function choice behavior. diff --git a/python/tests/unit/connectors/ai/test_chat_completion_client_base.py b/python/tests/unit/connectors/ai/test_chat_completion_client_base.py new file mode 100644 index 000000000000..6115c39f1705 --- /dev/null +++ b/python/tests/unit/connectors/ai/test_chat_completion_client_base.py @@ -0,0 +1,191 @@ +# Copyright (c) Microsoft. All rights reserved. + +from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase +from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings +from semantic_kernel.contents import ( + ChatHistory, + ChatMessageContent, + FunctionCallContent, + FunctionResultContent, + TextContent, +) +from semantic_kernel.contents.utils.author_role import AuthorRole +from semantic_kernel.filters.filter_types import FilterTypes +from semantic_kernel.functions.kernel_arguments import KernelArguments +from semantic_kernel.functions.kernel_function_decorator import kernel_function +from semantic_kernel.kernel import Kernel + + +class StubChatCompletion(ChatCompletionClientBase): + SUPPORTS_FUNCTION_CALLING = True + responses: list[list[ChatMessageContent]] + + async def _inner_get_chat_message_contents( + self, + chat_history: ChatHistory, + settings: PromptExecutionSettings, + ) -> list[ChatMessageContent]: + return self.responses.pop(0) + + +def _tool_call_message(text: str | None, call_id: str = "call_1") -> ChatMessageContent: + items = [] + if text is not None: + items.append(TextContent(text=text)) + items.append(FunctionCallContent(id=call_id, name="test-tool", arguments={})) + return ChatMessageContent(role=AuthorRole.ASSISTANT, items=items) + + +async def test_auto_function_call_preserves_intermediate_text_in_final_response() -> None: + kernel = Kernel() + kernel.add_function("test", kernel_function(lambda: "tool result", name="tool")) + + chat_completion = StubChatCompletion( + ai_model_id="test-model", + responses=[ + [ + ChatMessageContent( + role=AuthorRole.ASSISTANT, + items=[ + TextContent(text="I'll check that."), + FunctionCallContent(id="call_1", name="test-tool", arguments={}), + ], + ) + ], + [ChatMessageContent(role=AuthorRole.ASSISTANT, content="The answer is ready.")], + ], + ) + + result = await chat_completion.get_chat_message_contents( + chat_history=ChatHistory(system_message="Test"), + settings=PromptExecutionSettings(function_choice_behavior=FunctionChoiceBehavior.Auto()), + kernel=kernel, + arguments=KernelArguments(), + ) + + assert result[0].content == "I'll check that.\n\nThe answer is ready." + + +async def test_auto_function_call_preserves_intermediate_text_after_max_attempts() -> None: + kernel = Kernel() + kernel.add_function("test", kernel_function(lambda: "tool result", name="tool")) + + chat_completion = StubChatCompletion( + ai_model_id="test-model", + responses=[ + [_tool_call_message("I'll check that.")], + [ChatMessageContent(role=AuthorRole.ASSISTANT, content="The answer is ready.")], + ], + ) + + result = await chat_completion.get_chat_message_contents( + chat_history=ChatHistory(system_message="Test"), + settings=PromptExecutionSettings( + function_choice_behavior=FunctionChoiceBehavior.Auto(maximum_auto_invoke_attempts=1), + ), + kernel=kernel, + arguments=KernelArguments(), + ) + + assert result[0].content == "I'll check that.\n\nThe answer is ready." + + +async def test_auto_function_call_preserves_text_from_multiple_intermediate_responses() -> None: + kernel = Kernel() + kernel.add_function("test", kernel_function(lambda: "tool result", name="tool")) + + chat_completion = StubChatCompletion( + ai_model_id="test-model", + responses=[ + [_tool_call_message("First tool call.", call_id="call_1")], + [_tool_call_message("Second tool call.", call_id="call_2")], + [ChatMessageContent(role=AuthorRole.ASSISTANT, content="The answer is ready.")], + ], + ) + + result = await chat_completion.get_chat_message_contents( + chat_history=ChatHistory(system_message="Test"), + settings=PromptExecutionSettings(function_choice_behavior=FunctionChoiceBehavior.Auto()), + kernel=kernel, + arguments=KernelArguments(), + ) + + assert result[0].content == "First tool call.\n\nSecond tool call.\n\nThe answer is ready." + + +async def test_auto_function_call_inserts_intermediate_text_when_final_response_has_no_text() -> None: + kernel = Kernel() + kernel.add_function("test", kernel_function(lambda: "tool result", name="tool")) + + chat_completion = StubChatCompletion( + ai_model_id="test-model", + responses=[ + [_tool_call_message("I'll check that.")], + [ChatMessageContent(role=AuthorRole.ASSISTANT, items=[])], + ], + ) + + result = await chat_completion.get_chat_message_contents( + chat_history=ChatHistory(system_message="Test"), + settings=PromptExecutionSettings(function_choice_behavior=FunctionChoiceBehavior.Auto()), + kernel=kernel, + arguments=KernelArguments(), + ) + + assert result[0].content == "I'll check that." + + +async def test_auto_function_call_preserves_intermediate_text_when_filter_terminates() -> None: + kernel = Kernel() + kernel.add_function("test", kernel_function(lambda: "tool result", name="tool")) + + @kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION) + async def auto_invoke_terminate(context, next): + await next(context) + context.terminate = True + + chat_completion = StubChatCompletion( + ai_model_id="test-model", + responses=[ + [_tool_call_message("I'll check that.")], + ], + ) + + result = await chat_completion.get_chat_message_contents( + chat_history=ChatHistory(system_message="Test"), + settings=PromptExecutionSettings(function_choice_behavior=FunctionChoiceBehavior.Auto()), + kernel=kernel, + arguments=KernelArguments(), + ) + + assert result[0].content == "I'll check that." + assert any(isinstance(item, FunctionResultContent) for item in result[0].items) + + +async def test_auto_function_call_preserves_final_response_when_intermediate_has_no_text() -> None: + kernel = Kernel() + kernel.add_function("test", kernel_function(lambda: "tool result", name="tool")) + final_response = ChatMessageContent(role=AuthorRole.ASSISTANT, content="The answer is ready.") + + chat_completion = StubChatCompletion( + ai_model_id="test-model", + responses=[ + [ + ChatMessageContent( + role=AuthorRole.ASSISTANT, + items=[FunctionCallContent(id="call_1", name="test-tool", arguments={})], + ) + ], + [final_response], + ], + ) + + result = await chat_completion.get_chat_message_contents( + chat_history=ChatHistory(system_message="Test"), + settings=PromptExecutionSettings(function_choice_behavior=FunctionChoiceBehavior.Auto()), + kernel=kernel, + arguments=KernelArguments(), + ) + + assert result == [final_response]