-
Notifications
You must be signed in to change notification settings - Fork 3.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Python: Deprecated retry_mechanism (#9965)
### Motivation and Context <!-- Thank you for your contribution to the semantic-kernel repo! Please help reviewers and future users, providing the following information: 1. Why is this change required? 2. What problem does it solve? 3. What scenario does it contribute to? 4. If it fixes an open issue, please link to the issue here. --> In Python, we have the `retry_mechanism` property in the kernel but it's never used. Address: #9015 ### Description 1. Deprecate the retry_mechanism field. 2. Add a sample showing how to perform retries with the kernel using filter. <!-- Describe your changes, the overall approach, the underlying design. These notes will help understanding how your code works. Thanks! --> ### Contribution Checklist <!-- Before submitting this PR, please make sure: --> - [x] The code builds clean without any errors or warnings - [x] The PR follows the [SK Contribution Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md) and the [pre-submission formatting script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts) raises no violations - [x] All unit tests pass, and I have added new tests where possible - [x] I didn't break anyone 😄
- Loading branch information
1 parent
fbbd444
commit 7531abf
Showing
4 changed files
with
111 additions
and
13 deletions.
There are no files selected for viewing
103 changes: 103 additions & 0 deletions
103
python/samples/concepts/filtering/retry_with_filters.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,103 @@ | ||
# Copyright (c) Microsoft. All rights reserved. | ||
|
||
import asyncio | ||
import logging | ||
from collections.abc import Callable, Coroutine | ||
from typing import Any | ||
|
||
from samples.concepts.setup.chat_completion_services import Services, get_chat_completion_service_and_request_settings | ||
from semantic_kernel import Kernel | ||
from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior | ||
from semantic_kernel.contents import ChatHistory | ||
from semantic_kernel.filters import FunctionInvocationContext | ||
from semantic_kernel.filters.filter_types import FilterTypes | ||
from semantic_kernel.functions import kernel_function | ||
|
||
# This sample shows how to use a filter for retrying a function invocation. | ||
# This sample requires the following components: | ||
# - a ChatCompletionService: This component is responsible for generating responses to user messages. | ||
# - a ChatHistory: This component is responsible for keeping track of the chat history. | ||
# - a Kernel: This component is responsible for managing plugins and filters. | ||
# - a mock plugin: This plugin contains a function that simulates a call to an external service. | ||
# - a filter: This filter retries the function invocation if it fails. | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
# The maximum number of retries for the filter | ||
MAX_RETRIES = 3 | ||
|
||
|
||
class WeatherPlugin: | ||
MAX_FAILURES = 2 | ||
|
||
def __init__(self): | ||
self._invocation_count = 0 | ||
|
||
@kernel_function(name="GetWeather", description="Get the weather of the day at the current location.") | ||
def get_wather(self) -> str: | ||
"""Get the weather of the day at the current location. | ||
Simulates a call to an external service to get the weather. | ||
This function is designed to fail a certain number of times before succeeding. | ||
""" | ||
if self._invocation_count < self.MAX_FAILURES: | ||
self._invocation_count += 1 | ||
print(f"Number of attempts: {self._invocation_count}") | ||
raise Exception("Failed to get the weather") | ||
|
||
return "Sunny" | ||
|
||
|
||
async def retry_filter( | ||
context: FunctionInvocationContext, | ||
next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], | ||
) -> None: | ||
"""A filter that retries the function invocation if it fails. | ||
The filter uses a binary exponential backoff strategy to retry the function invocation. | ||
""" | ||
for i in range(MAX_RETRIES): | ||
try: | ||
await next(context) | ||
return | ||
except Exception as e: | ||
logger.warning(f"Failed to execute the function: {e}") | ||
backoff = 2**i | ||
logger.info(f"Sleeping for {backoff} seconds before retrying") | ||
|
||
|
||
async def main() -> None: | ||
kernel = Kernel() | ||
# Register the plugin to the kernel | ||
kernel.add_plugin(WeatherPlugin(), plugin_name="WeatherPlugin") | ||
# Add the filter to the kernel as a function invocation filter | ||
# A function invocation filter is called during when the kernel executes a function | ||
kernel.add_filter(FilterTypes.FUNCTION_INVOCATION, retry_filter) | ||
|
||
chat_history = ChatHistory() | ||
chat_history.add_user_message("What is the weather today?") | ||
|
||
chat_completion_service, request_settings = get_chat_completion_service_and_request_settings(Services.OPENAI) | ||
# Need to set the function choice behavior to auto such that the | ||
# service will automatically invoke the function in the response. | ||
request_settings.function_choice_behavior = FunctionChoiceBehavior.Auto() | ||
|
||
response = await chat_completion_service.get_chat_message_content( | ||
chat_history=chat_history, | ||
settings=request_settings, | ||
# Need to pass the kernel to the chat completion service so that it has access to the plugins and filters | ||
kernel=kernel, | ||
) | ||
|
||
print(response) | ||
|
||
# Sample output: | ||
# Number of attempts: 1 | ||
# Failed to execute the function: Failed to get the weather | ||
# Number of attempts: 2 | ||
# Failed to execute the function: Failed to get the weather | ||
# The weather today is Sunny | ||
|
||
|
||
if __name__ == "__main__": | ||
asyncio.run(main()) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters